From 0eed35289c59ca0beb18d708302e7a96511cc0ed Mon Sep 17 00:00:00 2001 From: Dmitry Gammel Date: Thu, 13 Aug 2026 11:12:29 +0500 Subject: [PATCH] =?UTF-8?q?ESP32=20Emulator:=20=D0=B2=D0=B5=D0=B1-=D0=BC?= =?UTF-8?q?=D0=BE=D1=80=D0=B4=D0=B0=20=D0=BA=D0=BE=D0=BD=D1=82=D1=80=D0=BE?= =?UTF-8?q?=D0=BB=D0=BB=D0=B5=D1=80=D0=B0=20=D0=B3=D1=80=D0=BE=D1=83=D0=B1?= =?UTF-8?q?=D0=BE=D0=BA=D1=81=D0=B0=20=D0=BF=D0=BE=D0=B2=D0=B5=D1=80=D1=85?= =?UTF-8?q?=20MQTT?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 платформы. --- .env.example | 6 + .gitignore | 2 + Dockerfile | 13 ++ README.md | 59 ++++++ cmd/emulator/main.go | 83 +++++++++ go.mod | 11 ++ go.sum | 8 + internal/httpserver/server.go | 149 +++++++++++++++ internal/httpserver/web/index.html | 282 +++++++++++++++++++++++++++++ internal/mqttclient/client.go | 123 +++++++++++++ internal/state/state.go | 193 ++++++++++++++++++++ 11 files changed, 929 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 README.md create mode 100644 cmd/emulator/main.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/httpserver/server.go create mode 100644 internal/httpserver/web/index.html create mode 100644 internal/mqttclient/client.go create mode 100644 internal/state/state.go diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..37074ce --- /dev/null +++ b/.env.example @@ -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 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ef8a9ca --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +/emulator +.env diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..67c70b9 --- /dev/null +++ b/Dockerfile @@ -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"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..76af6a1 --- /dev/null +++ b/README.md @@ -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` | порт веб-морды эмулятора | diff --git a/cmd/emulator/main.go b/cmd/emulator/main.go new file mode 100644 index 0000000..aa6d299 --- /dev/null +++ b/cmd/emulator/main.go @@ -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 +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..d046339 --- /dev/null +++ b/go.mod @@ -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 +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..9abe94c --- /dev/null +++ b/go.sum @@ -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= diff --git a/internal/httpserver/server.go b/internal/httpserver/server.go new file mode 100644 index 0000000..72d21b3 --- /dev/null +++ b/internal/httpserver/server.go @@ -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) +} diff --git a/internal/httpserver/web/index.html b/internal/httpserver/web/index.html new file mode 100644 index 0000000..31fc776 --- /dev/null +++ b/internal/httpserver/web/index.html @@ -0,0 +1,282 @@ + + + + + +ESP32 Emulator — гроубокс + + + + +

ESP32 Emulator

+

Виртуальный контроллер гроубокса — датчик + вентилятор + свет + помпа, говорит по тому же MQTT-протоколу, что и настоящее устройство.

+
подключение…
+ +
+ + + + + + + 24°C + 55% + + датчик + + + + + + + + + + + + вентилятор + + + + + + + свет + + + + + + + + помпа + +
+ +
+ + + + diff --git a/internal/mqttclient/client.go b/internal/mqttclient/client.go new file mode 100644 index 0000000..faa885c --- /dev/null +++ b/internal/mqttclient/client.go @@ -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) +} diff --git a/internal/state/state.go b/internal/state/state.go new file mode 100644 index 0000000..607db49 --- /dev/null +++ b/internal/state/state.go @@ -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 + } + } +}