ESP32 Emulator: веб-морда контроллера гроубокса поверх MQTT

Go-сервис без внешних фронтенд-зависимостей: embedded HTML-страница
с SVG-изображением платы (датчик + вентилятор + свет + помпа) и
Server-Sent Events для live-обновления. Слушает devices/+/commands
и реально отвечает ack'ом, публикует devices/{id}/telemetry —
протокол зеркалит ingest-service/device-control-service основного
репозитория, общего кода между репозиториями нет. Дефолтные
external_id (sensor-1/fan-1/light-1/pump-1) совпадают с
DemoGrowboxSeeder платформы.
This commit is contained in:
2026-08-13 11:12:29 +05:00
commit 0eed35289c
11 changed files with 929 additions and 0 deletions
+6
View File
@@ -0,0 +1,6 @@
# Address of the same Mosquitto broker the main platform uses.
# Local dev against home_automatization's docker-compose: tcp://localhost:1883
# (mosquitto's port is bound to 127.0.0.1 there — reachable from the host).
MQTT_BROKER_URL=tcp://localhost:1883
HTTP_PORT=8091
+2
View File
@@ -0,0 +1,2 @@
/emulator
.env
+13
View File
@@ -0,0 +1,13 @@
FROM golang:1.25-alpine AS build
WORKDIR /src
COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 go build -o /out/emulator ./cmd/emulator
FROM alpine:3.20
RUN adduser -D -u 10001 app
COPY --from=build /out/emulator /usr/local/bin/emulator
USER app
EXPOSE 8091
ENTRYPOINT ["/usr/local/bin/emulator"]
+59
View File
@@ -0,0 +1,59 @@
# ESP32 Emulator
Веб-морда, которая ведёт себя как настоящий ESP32-контроллер гроубокса для
[платформы домашней автоматизации](https://git.cactoz.su/cacto/home_automatization) —
отдельный репозиторий, никакого общего кода с основным проектом: единственная
связь — MQTT-протокол, которому оба конца следуют одинаково.
Один Go-бинарник: HTTP-сервер отдаёт единственную HTML-страницу с
изображением платы (датчик + вентилятор + свет + помпа) и сам говорит по
MQTT с брокером платформы. Браузер никогда не подключается к MQTT напрямую —
только к этому серверу (JSON + Server-Sent Events).
## Что показывает страница
- **Датчик** — слайдеры температуры/влажности + кнопка "Опубликовать
показания" (публикует `devices/{external_id}/telemetry`, по одному
сообщению на `sensor_type`, как и положено по формату ingest-service)
- **Вентилятор / Свет / Помпа** — живой статус (вкл/выкл, уровень), который
меняется САМ, когда платформа реально присылает команду через
`devices/{external_id}/commands`; ручного переключения в UI нет
осознанно — эмулятор демонстрирует "устройство слушается платформы", а не
пульт дистанционного управления
- `external_id`/`zone_id` каждого слота редактируемые — по умолчанию
`sensor-1`/`fan-1`/`light-1`/`pump-1`, `zone_id=1`, чтобы сразу совпадать с
`DemoGrowboxSeeder` основного репозитория
## Протокол (см. основной репозиторий, это не источник истины, а зеркало)
| Топик | Кто публикует | Payload |
|---|---|---|
| `devices/{id}/telemetry` | эмулятор | `{"device_id","zone_id","sensor_type","value","timestamp"}` |
| `devices/{id}/commands` | платформа | `{"action":"turn_on"\|"turn_off"\|"set_level","level"?}` |
| `devices/{id}/ack` | эмулятор | `{"state":{"power":"on"\|"off"}}` или `{"state":{"level":N}}` |
## Запуск
Локально (нужен доступ к тому же MQTT-брокеру, что слушает платформа):
```bash
cp .env.example .env
export $(cat .env | xargs)
go run ./cmd/emulator
```
Открыть `http://localhost:8091`.
Через Docker:
```bash
docker build -t esp32-emulator .
docker run --rm -p 8091:8091 -e MQTT_BROKER_URL=tcp://host.docker.internal:1883 esp32-emulator
```
## Переменные окружения
| Переменная | По умолчанию | Назначение |
|---|---|---|
| `MQTT_BROKER_URL` | `tcp://localhost:1883` | адрес брокера основной платформы |
| `HTTP_PORT` | `8091` | порт веб-морды эмулятора |
+83
View File
@@ -0,0 +1,83 @@
// Command emulator serves a small web UI that stands in for an ESP32
// growbox controller: a sensor slot (publishes temperature/humidity
// telemetry) and three actuator slots (fan/light/pump — react live to
// commands the platform sends over MQTT). See README.md for the protocol
// this speaks and why it exists as a standalone tool.
package main
import (
"context"
"log/slog"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"git.cactoz.su/cacto/home_automation_emulator/internal/httpserver"
"git.cactoz.su/cacto/home_automation_emulator/internal/mqttclient"
"git.cactoz.su/cacto/home_automation_emulator/internal/state"
)
func main() {
logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
brokerURL := getEnv("MQTT_BROKER_URL", "tcp://localhost:1883")
httpAddr := ":" + getEnv("HTTP_PORT", "8091")
store := state.NewStore()
mqttClient, err := mqttclient.Connect(mqttclient.Config{
BrokerURL: brokerURL,
ClientID: "esp32-emulator",
}, logger)
if err != nil {
logger.Error("mqtt connect failed", "broker", brokerURL, "error", err)
os.Exit(1)
}
defer mqttClient.Close()
if err := mqttClient.SubscribeCommands(func(externalID, action string, level float64, hasLevel bool) {
reported, ok := store.ApplyCommand(externalID, action, level, hasLevel)
if !ok {
logger.Warn("command for unknown device ignored", "device_id", externalID, "action", action)
return
}
logger.Info("command applied", "device_id", externalID, "action", action)
if err := mqttClient.PublishAck(externalID, reported); err != nil {
logger.Error("publish ack failed", "device_id", externalID, "error", err)
}
}); err != nil {
logger.Error("mqtt subscribe failed", "error", err)
os.Exit(1)
}
srv := &http.Server{
Addr: httpAddr,
Handler: httpserver.New(store, mqttClient, logger),
}
go func() {
logger.Info("http server listening", "addr", httpAddr, "mqtt_broker", brokerURL)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
logger.Error("http server failed", "error", err)
os.Exit(1)
}
}()
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
<-ctx.Done()
logger.Info("shutting down")
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_ = srv.Shutdown(shutdownCtx)
}
func getEnv(key, fallback string) string {
if v := os.Getenv(key); v != "" {
return v
}
return fallback
}
+11
View File
@@ -0,0 +1,11 @@
module git.cactoz.su/cacto/home_automation_emulator
go 1.25.0
require github.com/eclipse/paho.mqtt.golang v1.5.1
require (
github.com/gorilla/websocket v1.5.3 // indirect
golang.org/x/net v0.44.0 // indirect
golang.org/x/sync v0.17.0 // indirect
)
+8
View File
@@ -0,0 +1,8 @@
github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE=
github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I=
golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY=
golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
+149
View File
@@ -0,0 +1,149 @@
// Package httpserver is the emulator's own control surface: a web UI (the
// "controller board") plus the JSON/SSE API it talks to. This is separate
// from the MQTT side — the browser never speaks MQTT directly, it asks this
// server to do so.
package httpserver
import (
"embed"
"encoding/json"
"fmt"
"io/fs"
"log/slog"
"net/http"
"git.cactoz.su/cacto/home_automation_emulator/internal/mqttclient"
"git.cactoz.su/cacto/home_automation_emulator/internal/state"
)
//go:embed web
var webFS embed.FS
type Server struct {
store *state.Store
mqtt *mqttclient.Client
logger *slog.Logger
mux *http.ServeMux
}
func New(store *state.Store, mqtt *mqttclient.Client, logger *slog.Logger) *Server {
s := &Server{store: store, mqtt: mqtt, logger: logger, mux: http.NewServeMux()}
s.routes()
return s
}
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { s.mux.ServeHTTP(w, r) }
func (s *Server) routes() {
static, err := fs.Sub(webFS, "web")
if err != nil {
panic(err) // embedded FS is compiled in — can only fail if the embed directive itself is wrong
}
s.mux.Handle("GET /", http.FileServerFS(static))
s.mux.HandleFunc("GET /api/snapshot", s.handleSnapshot)
s.mux.HandleFunc("GET /api/events", s.handleEvents)
s.mux.HandleFunc("POST /api/devices/{slot}/identity", s.handleSetIdentity)
s.mux.HandleFunc("POST /api/devices/{slot}/telemetry", s.handlePublishTelemetry)
}
func (s *Server) handleSnapshot(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.store.Snapshot())
}
func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) {
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
return
}
ch, unsubscribe := s.store.Subscribe()
defer unsubscribe()
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
writeEvent(w, s.store.Snapshot())
flusher.Flush()
for {
select {
case <-r.Context().Done():
return
case snap, ok := <-ch:
if !ok {
return
}
writeEvent(w, snap)
flusher.Flush()
}
}
}
func writeEvent(w http.ResponseWriter, snap state.Snapshot) {
body, err := json.Marshal(snap)
if err != nil {
return
}
fmt.Fprintf(w, "data: %s\n\n", body)
}
func (s *Server) handleSetIdentity(w http.ResponseWriter, r *http.Request) {
slot := r.PathValue("slot")
if _, ok := s.store.Device(slot); !ok {
http.Error(w, "unknown slot", http.StatusNotFound)
return
}
var req struct {
ExternalID string `json:"external_id"`
ZoneID string `json:"zone_id"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.ExternalID == "" || req.ZoneID == "" {
http.Error(w, "external_id and zone_id are required", http.StatusBadRequest)
return
}
s.store.SetIdentity(slot, req.ExternalID, req.ZoneID)
writeJSON(w, http.StatusOK, s.store.Snapshot())
}
func (s *Server) handlePublishTelemetry(w http.ResponseWriter, r *http.Request) {
slot := r.PathValue("slot")
device, ok := s.store.Device(slot)
if !ok {
http.Error(w, "unknown slot", http.StatusNotFound)
return
}
if device.Kind != state.KindSensor {
http.Error(w, "slot is not a sensor", http.StatusBadRequest)
return
}
var req struct {
SensorType string `json:"sensor_type"`
Value float64 `json:"value"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.SensorType == "" {
http.Error(w, "sensor_type is required", http.StatusBadRequest)
return
}
s.store.SetReading(slot, req.SensorType, req.Value)
if err := s.mqtt.PublishTelemetry(device.ExternalID, device.ZoneID, req.SensorType, req.Value); err != nil {
s.logger.Error("publish telemetry failed", "slot", slot, "error", err)
http.Error(w, "failed to publish to mqtt", http.StatusBadGateway)
return
}
writeJSON(w, http.StatusOK, s.store.Snapshot())
}
func writeJSON(w http.ResponseWriter, status int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(v)
}
+282
View File
@@ -0,0 +1,282 @@
<!doctype html>
<html lang="ru">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>ESP32 Emulator — гроубокс</title>
<style>
:root {
--bg: #0f1720;
--panel: #182333;
--panel-2: #1f2d42;
--border: #2b3b52;
--text: #e6edf5;
--muted: #8ea0b8;
--accent: #33d17a;
--accent-dim: #1f6b45;
--warn: #e8b339;
--pump: #3aa0e8;
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
}
* { box-sizing: border-box; }
body {
margin: 0;
background: var(--bg);
color: var(--text);
padding: 24px;
}
h1 { font-size: 1.3rem; font-weight: 600; margin: 0 0 4px; }
.subtitle { color: var(--muted); margin: 0 0 24px; font-size: 0.9rem; }
.conn {
display: inline-flex; align-items: center; gap: 6px;
font-size: 0.8rem; color: var(--muted); margin-bottom: 20px;
}
.conn .dot { width: 8px; height: 8px; border-radius: 50%; background: #666; }
.conn.online .dot { background: var(--accent); box-shadow: 0 0 6px var(--accent); }
.conn.offline .dot { background: #d95555; }
.board-wrap { display: flex; justify-content: center; margin-bottom: 28px; }
.board { width: 100%; max-width: 520px; }
.board rect.case { fill: var(--panel); stroke: var(--border); stroke-width: 2; }
.fan-blades { transform-origin: center; transition: opacity .2s; }
.fan-blades.on { animation: spin 0.9s linear infinite; }
@keyframes spin { from { transform: rotate(0deg); } to { transform: rotate(360deg); } }
.bulb-glow { opacity: 0; transition: opacity .25s; }
.bulb-glow.on { opacity: 1; }
.bulb-body { transition: fill .25s; }
.pump-ring { opacity: 0; transition: opacity .25s; }
.pump-ring.on { opacity: 1; animation: pulse 1.4s ease-in-out infinite; }
@keyframes pulse { 0%,100% { opacity: .35; } 50% { opacity: 1; } }
.grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(260px, 1fr));
gap: 16px;
max-width: 1080px;
margin: 0 auto;
}
.card {
background: var(--panel);
border: 1px solid var(--border);
border-radius: 12px;
padding: 16px;
}
.card h2 {
font-size: 0.95rem; margin: 0 0 12px;
display: flex; align-items: center; justify-content: space-between;
}
.badge {
font-size: 0.72rem; padding: 2px 8px; border-radius: 999px;
background: var(--panel-2); color: var(--muted); font-weight: 500;
}
.badge.on { background: var(--accent-dim); color: var(--accent); }
label { display: block; font-size: 0.78rem; color: var(--muted); margin: 10px 0 4px; }
input[type=text], input[type=number] {
width: 100%; padding: 7px 9px; border-radius: 7px;
border: 1px solid var(--border); background: var(--panel-2); color: var(--text);
font-size: 0.88rem;
}
input[type=range] { width: 100%; }
.row { display: flex; gap: 10px; }
.row > div { flex: 1; }
button {
margin-top: 12px; width: 100%; padding: 8px 10px;
border: none; border-radius: 7px; background: var(--accent); color: #05170d;
font-weight: 600; font-size: 0.85rem; cursor: pointer;
}
button.secondary { background: var(--panel-2); color: var(--text); border: 1px solid var(--border); }
button:active { transform: translateY(1px); }
.status-line { display: flex; justify-content: space-between; font-size: 0.85rem; margin-top: 8px; }
.status-line span:last-child { color: var(--text); font-weight: 600; }
.value { color: var(--text); font-weight: 600; }
</style>
</head>
<body>
<h1>ESP32 Emulator</h1>
<p class="subtitle">Виртуальный контроллер гроубокса — датчик + вентилятор + свет + помпа, говорит по тому же MQTT-протоколу, что и настоящее устройство.</p>
<div id="conn" class="conn offline"><span class="dot"></span><span id="conn-label">подключение…</span></div>
<div class="board-wrap">
<svg class="board" viewBox="0 0 400 220" xmlns="http://www.w3.org/2000/svg">
<rect class="case" x="10" y="10" width="380" height="200" rx="14"/>
<!-- sensor -->
<g transform="translate(60,60)">
<circle r="30" fill="#233348" stroke="#37527a" stroke-width="2"/>
<text id="sensor-temp" x="0" y="-2" text-anchor="middle" font-size="13" fill="#e6edf5" font-weight="600">24°C</text>
<text id="sensor-humidity" x="0" y="14" text-anchor="middle" font-size="10" fill="#8ea0b8">55%</text>
</g>
<text x="60" y="115" text-anchor="middle" font-size="11" fill="#8ea0b8">датчик</text>
<!-- fan -->
<g transform="translate(160,60)">
<circle r="30" fill="#233348" stroke="#37527a" stroke-width="2"/>
<g id="fan-blades" class="fan-blades">
<ellipse cx="0" cy="-12" rx="6" ry="14" fill="#5b7ba8"/>
<ellipse cx="12" cy="6" rx="6" ry="14" fill="#5b7ba8" transform="rotate(120 12 6)"/>
<ellipse cx="-12" cy="6" rx="6" ry="14" fill="#5b7ba8" transform="rotate(-120 -12 6)"/>
<circle r="4" fill="#e6edf5"/>
</g>
</g>
<text x="160" y="115" text-anchor="middle" font-size="11" fill="#8ea0b8">вентилятор</text>
<!-- light -->
<g transform="translate(260,60)">
<circle id="bulb-glow" class="bulb-glow" r="26" fill="#e8b339" opacity="0"/>
<circle id="bulb-body" class="bulb-body" r="16" fill="#233348" stroke="#37527a" stroke-width="2"/>
</g>
<text x="260" y="115" text-anchor="middle" font-size="11" fill="#8ea0b8">свет</text>
<!-- pump -->
<g transform="translate(340,60)">
<circle id="pump-ring" class="pump-ring" r="26" fill="none" stroke="#3aa0e8" stroke-width="3" opacity="0"/>
<circle r="16" fill="#233348" stroke="#37527a" stroke-width="2"/>
<path d="M0,-8 C6,-2 6,6 0,8 C-6,6 -6,-2 0,-8 Z" fill="#3aa0e8"/>
</g>
<text x="340" y="115" text-anchor="middle" font-size="11" fill="#8ea0b8">помпа</text>
</svg>
</div>
<div class="grid" id="cards"></div>
<script>
const SLOTS = {
sensor: { label: 'Датчик температуры/влажности', kind: 'sensor' },
fan: { label: 'Вентилятор', kind: 'actuator' },
light: { label: 'Свет', kind: 'actuator' },
pump: { label: 'Помпа', kind: 'actuator' },
};
let latest = {};
function cardHTML(slot) {
const meta = SLOTS[slot];
const idPrefix = slot;
let body = `
<label>external_id</label>
<input type="text" id="${idPrefix}-external-id">
<label>zone_id</label>
<input type="text" id="${idPrefix}-zone-id">
<button class="secondary" onclick="saveIdentity('${slot}')">Сохранить идентификатор</button>
`;
if (meta.kind === 'sensor') {
body += `
<label>Температура: <span class="value" id="${idPrefix}-temp-value">24</span> °C</label>
<input type="range" id="${idPrefix}-temp" min="-10" max="50" step="0.5" value="24">
<label>Влажность: <span class="value" id="${idPrefix}-humidity-value">55</span> %</label>
<input type="range" id="${idPrefix}-humidity" min="0" max="100" step="1" value="55">
<button onclick="publishTelemetry('${slot}')">Опубликовать показания</button>
`;
} else {
body += `
<div class="status-line"><span>Питание</span><span id="${idPrefix}-power">—</span></div>
<div class="status-line" id="${idPrefix}-level-row" style="display:none">
<span>Уровень</span><span id="${idPrefix}-level">—</span>
</div>
<p style="color:var(--muted); font-size:0.78rem; margin-top:10px;">
Управляется командами от платформы — здесь только отображение.
</p>
`;
}
const badge = meta.kind === 'actuator' ? `<span class="badge" id="${idPrefix}-badge">выключено</span>` : '';
return `<div class="card">
<h2>${meta.label} ${badge}</h2>
${body}
</div>`;
}
document.getElementById('cards').innerHTML = Object.keys(SLOTS).map(cardHTML).join('');
function render(snapshot) {
latest = {};
for (const d of snapshot.devices) {
latest[d.slot] = d;
document.getElementById(`${d.slot}-external-id`).value = d.external_id;
document.getElementById(`${d.slot}-zone-id`).value = d.zone_id;
if (d.kind === 'sensor') {
const temp = d.readings?.temperature ?? 0;
const humidity = d.readings?.humidity ?? 0;
document.getElementById(`${d.slot}-temp`).value = temp;
document.getElementById(`${d.slot}-temp-value`).textContent = temp;
document.getElementById(`${d.slot}-humidity`).value = humidity;
document.getElementById(`${d.slot}-humidity-value`).textContent = humidity;
document.getElementById('sensor-temp').textContent = `${temp}°C`;
document.getElementById('sensor-humidity').textContent = `${humidity}%`;
} else {
const badge = document.getElementById(`${d.slot}-badge`);
badge.textContent = d.power ? 'включено' : 'выключено';
badge.classList.toggle('on', d.power);
document.getElementById(`${d.slot}-power`).textContent = d.power ? 'вкл' : 'выкл';
if (d.supports_level) {
document.getElementById(`${d.slot}-level-row`).style.display = 'flex';
document.getElementById(`${d.slot}-level`).textContent = d.level;
}
}
}
document.getElementById('fan-blades').classList.toggle('on', !!latest.fan?.power);
document.getElementById('bulb-glow').classList.toggle('on', !!latest.light?.power);
document.getElementById('bulb-body').setAttribute('fill', latest.light?.power ? '#e8b339' : '#233348');
document.getElementById('pump-ring').classList.toggle('on', !!latest.pump?.power);
}
for (const slot of Object.keys(SLOTS)) {
if (SLOTS[slot].kind !== 'sensor') continue;
document.getElementById(`${slot}-temp`).addEventListener('input', (e) => {
document.getElementById(`${slot}-temp-value`).textContent = e.target.value;
});
document.getElementById(`${slot}-humidity`).addEventListener('input', (e) => {
document.getElementById(`${slot}-humidity-value`).textContent = e.target.value;
});
}
async function saveIdentity(slot) {
const external_id = document.getElementById(`${slot}-external-id`).value.trim();
const zone_id = document.getElementById(`${slot}-zone-id`).value.trim();
await fetch(`/api/devices/${slot}/identity`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ external_id, zone_id }),
});
}
async function publishTelemetry(slot) {
const temp = parseFloat(document.getElementById(`${slot}-temp`).value);
const humidity = parseFloat(document.getElementById(`${slot}-humidity`).value);
await fetch(`/api/devices/${slot}/telemetry`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ sensor_type: 'temperature', value: temp }),
});
await fetch(`/api/devices/${slot}/telemetry`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ sensor_type: 'humidity', value: humidity }),
});
}
function connectEvents() {
const es = new EventSource('/api/events');
const conn = document.getElementById('conn');
const label = document.getElementById('conn-label');
es.onopen = () => { conn.classList.add('online'); conn.classList.remove('offline'); label.textContent = 'подключено'; };
es.onerror = () => { conn.classList.remove('online'); conn.classList.add('offline'); label.textContent = 'переподключение…'; };
es.onmessage = (e) => render(JSON.parse(e.data));
}
connectEvents();
</script>
</body>
</html>
+123
View File
@@ -0,0 +1,123 @@
// Package mqttclient speaks the same wire protocol a real device would:
// publish to devices/{id}/telemetry and devices/{id}/ack, subscribe to
// devices/{id}/commands. Topic and payload shapes are dictated by
// ingest-service and device-control-service in the main platform repo —
// this client exists to match them, not to define them.
package mqttclient
import (
"encoding/json"
"fmt"
"log/slog"
"strings"
"time"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
type Config struct {
BrokerURL string
ClientID string
}
// CommandHandler receives one decoded devices/{id}/commands message.
// hasLevel distinguishes "level omitted" from "level is zero".
type CommandHandler func(externalID, action string, level float64, hasLevel bool)
type Client struct {
client mqtt.Client
logger *slog.Logger
}
func Connect(cfg Config, logger *slog.Logger) (*Client, error) {
opts := mqtt.NewClientOptions().
AddBroker(cfg.BrokerURL).
SetClientID(cfg.ClientID).
SetAutoReconnect(true).
SetConnectRetry(true).
SetConnectionLostHandler(func(_ mqtt.Client, err error) {
logger.Warn("mqtt connection lost", "error", err)
})
client := mqtt.NewClient(opts)
token := client.Connect()
token.Wait()
if err := token.Error(); err != nil {
return nil, fmt.Errorf("connect to mqtt broker %s: %w", cfg.BrokerURL, err)
}
return &Client{client: client, logger: logger}, nil
}
// SubscribeCommands listens on devices/+/commands for every device the
// emulator might represent — one subscription covers the whole board.
func (c *Client) SubscribeCommands(handler CommandHandler) error {
token := c.client.Subscribe("devices/+/commands", 1, func(_ mqtt.Client, msg mqtt.Message) {
externalID, ok := externalIDFromTopic(msg.Topic())
if !ok {
return
}
var cmd struct {
Action string `json:"action"`
Level *float64 `json:"level"`
}
if err := json.Unmarshal(msg.Payload(), &cmd); err != nil {
c.logger.Warn("dropping invalid command payload", "device_id", externalID, "error", err)
return
}
level := 0.0
hasLevel := cmd.Level != nil
if hasLevel {
level = *cmd.Level
}
handler(externalID, cmd.Action, level, hasLevel)
})
token.Wait()
return token.Error()
}
func externalIDFromTopic(topic string) (string, bool) {
parts := strings.Split(topic, "/")
if len(parts) != 3 || parts[0] != "devices" {
return "", false
}
return parts[1], true
}
// PublishTelemetry sends one sensor_type reading, matching the payload
// shape services/ingest-service/internal/telemetry.ParseReading expects.
func (c *Client) PublishTelemetry(externalID, zoneID, sensorType string, value float64) error {
body, err := json.Marshal(map[string]any{
"device_id": externalID,
"zone_id": zoneID,
"sensor_type": sensorType,
"value": value,
"timestamp": time.Now().UTC().Format(time.RFC3339),
})
if err != nil {
return err
}
return c.publish(fmt.Sprintf("devices/%s/telemetry", externalID), body)
}
// PublishAck reports an actuator's new state after a command, matching the
// shape services/device-control-service/cmd/.../ackHandler expects.
func (c *Client) PublishAck(externalID string, reportedState map[string]any) error {
body, err := json.Marshal(map[string]any{"state": reportedState})
if err != nil {
return err
}
return c.publish(fmt.Sprintf("devices/%s/ack", externalID), body)
}
func (c *Client) publish(topic string, body []byte) error {
token := c.client.Publish(topic, 1, false, body)
token.Wait()
return token.Error()
}
func (c *Client) Close() {
c.client.Disconnect(250)
}
+193
View File
@@ -0,0 +1,193 @@
// Package state holds the emulator's in-memory model of a growbox
// controller board: one sensor (reports readings) and up to three
// actuators (respond to power/level commands). Nothing here persists
// across restarts — the emulator is a throwaway dev/demo tool, not a
// real device.
package state
import "sync"
type Kind string
const (
KindSensor Kind = "sensor"
KindActuator Kind = "actuator"
)
// Device is one slot on the controller board. Sensor slots use Readings;
// actuator slots use Power/Level. A slot is inert (not published/commanded)
// when ExternalID is empty, so the board can represent fewer than four
// devices without special-casing "missing" ones.
type Device struct {
Slot string `json:"slot"` // "sensor" | "fan" | "light" | "pump" — fixed board position
ExternalID string `json:"external_id"`
ZoneID string `json:"zone_id"`
Kind Kind `json:"kind"`
Readings map[string]float64 `json:"readings,omitempty"`
Power bool `json:"power,omitempty"`
SupportsLevel bool `json:"supports_level,omitempty"`
Level float64 `json:"level,omitempty"`
}
// Snapshot is the full board state pushed to the browser (initial load and
// every SSE update) — simpler than diffing for a board this small.
type Snapshot struct {
Devices []Device `json:"devices"`
}
type Store struct {
mu sync.RWMutex
devices map[string]*Device // keyed by Slot
subMu sync.Mutex
subs map[chan Snapshot]struct{}
}
func NewStore() *Store {
return &Store{
devices: map[string]*Device{
"sensor": {Slot: "sensor", ExternalID: "sensor-1", ZoneID: "1", Kind: KindSensor,
Readings: map[string]float64{"temperature": 24, "humidity": 55}},
"fan": {Slot: "fan", ExternalID: "fan-1", ZoneID: "1", Kind: KindActuator, SupportsLevel: true},
"light": {Slot: "light", ExternalID: "light-1", ZoneID: "1", Kind: KindActuator, SupportsLevel: true},
"pump": {Slot: "pump", ExternalID: "pump-1", ZoneID: "1", Kind: KindActuator},
},
subs: make(map[chan Snapshot]struct{}),
}
}
// Snapshot returns a deep-enough copy of the current board state (readings
// map is copied so callers can't mutate internal state through it).
func (s *Store) Snapshot() Snapshot {
s.mu.RLock()
defer s.mu.RUnlock()
devices := make([]Device, 0, len(s.devices))
for _, slot := range []string{"sensor", "fan", "light", "pump"} {
d := *s.devices[slot]
if d.Readings != nil {
readings := make(map[string]float64, len(d.Readings))
for k, v := range d.Readings {
readings[k] = v
}
d.Readings = readings
}
devices = append(devices, d)
}
return Snapshot{Devices: devices}
}
// Device returns a copy of one slot's device, or ok=false if the slot name
// is unknown.
func (s *Store) Device(slot string) (Device, bool) {
s.mu.RLock()
defer s.mu.RUnlock()
d, ok := s.devices[slot]
if !ok {
return Device{}, false
}
return *d, true
}
// DeviceBySlotExternalID finds the slot name for a given external_id, used
// to route an incoming MQTT command (which only carries the external_id in
// its topic) back to a board slot.
func (s *Store) SlotForExternalID(externalID string) (string, bool) {
s.mu.RLock()
defer s.mu.RUnlock()
for slot, d := range s.devices {
if d.ExternalID == externalID {
return slot, true
}
}
return "", false
}
// SetIdentity updates which external_id/zone_id a slot publishes as —
// this is itself one of the "parameters" the web UI lets you set, so the
// emulator can stand in for whatever device row exists in the platform's
// database.
func (s *Store) SetIdentity(slot, externalID, zoneID string) {
s.mu.Lock()
if d, ok := s.devices[slot]; ok {
d.ExternalID = externalID
d.ZoneID = zoneID
}
s.mu.Unlock()
s.broadcast()
}
// SetReading updates one sensor_type's value for the sensor slot.
func (s *Store) SetReading(slot, sensorType string, value float64) {
s.mu.Lock()
if d, ok := s.devices[slot]; ok && d.Kind == KindSensor {
d.Readings[sensorType] = value
}
s.mu.Unlock()
s.broadcast()
}
// ApplyCommand applies an incoming {"action": ..., "level": ...} command
// (as published by device-control-service to devices/{id}/commands) to the
// actuator identified by externalID. Returns the resulting reported state
// (for the ack payload) and whether externalID matched a known slot.
func (s *Store) ApplyCommand(externalID, action string, level float64, hasLevel bool) (map[string]any, bool) {
s.mu.Lock()
var slot *Device
for _, d := range s.devices {
if d.ExternalID == externalID {
slot = d
break
}
}
if slot == nil {
s.mu.Unlock()
return nil, false
}
reported := map[string]any{}
switch action {
case "turn_on":
slot.Power = true
reported["power"] = "on"
case "turn_off":
slot.Power = false
reported["power"] = "off"
case "set_level":
if hasLevel {
slot.Level = level
reported["level"] = level
}
}
s.mu.Unlock()
s.broadcast()
return reported, true
}
// Subscribe registers a channel that receives a snapshot on every state
// change. Call the returned func to unsubscribe.
func (s *Store) Subscribe() (chan Snapshot, func()) {
ch := make(chan Snapshot, 4)
s.subMu.Lock()
s.subs[ch] = struct{}{}
s.subMu.Unlock()
return ch, func() {
s.subMu.Lock()
delete(s.subs, ch)
s.subMu.Unlock()
close(ch)
}
}
func (s *Store) broadcast() {
snap := s.Snapshot()
s.subMu.Lock()
defer s.subMu.Unlock()
for ch := range s.subs {
select {
case ch <- snap:
default: // slow subscriber — drop, next change will resync it
}
}
}