Merge pull request 'feature/streaming' (#10) from feature/streaming into main

Reviewed-on: #10

README.md

@@ -10,6 +10,7 @@ SPORE is a cluster engine for ESP8266 microcontrollers that provides automatic n
 - [Supported Hardware](#supported-hardware)
 - [Architecture](#architecture)
 - [Cluster Broadcast](#cluster-broadcast)
+- [Streaming API](#streaming-api)
 - [API Reference](#api-reference)
 - [Configuration](#configuration)
 - [Development](#development)
@@ -28,6 +29,7 @@ SPORE is a cluster engine for ESP8266 microcontrollers that provides automatic n
 - **Health Monitoring**: Real-time node status tracking with resource monitoring
 - **Event System**: Local and cluster-wide event publishing/subscription
 - **Cluster Broadcast**: Centralized UDP broadcast of events (CLUSTER_EVENT)
+- **Streaming API**: WebSocket bridge for real-time event send/receive
 - **Over-The-Air Updates**: Seamless firmware updates across the cluster
 - **REST API**: HTTP-based cluster management and monitoring
 - **Capability Discovery**: Automatic API endpoint and service capability detection
@@ -134,6 +136,23 @@ Notes:
 
 📖 See the dedicated guide: [`docs/ClusterBroadcast.md`](./docs/ClusterBroadcast.md)
 
+## Streaming API
+
+Real-time event bridge available at `/ws` using WebSocket.
+
+- Send JSON `{ event, payload }` to dispatch events via `ctx.fire`.
+- Receive all local events as `{ event, payload }`.
+
+Examples:
+```json
+{ "event": "api/neopattern/color", "payload": { "color": "#FF0000", "brightness": 128 } }
+```
+```json
+{ "event": "cluster/broadcast", "payload": { "event": "api/neopattern/color", "data": { "color": "#00FF00" } } }
+```
+
+📖 See the dedicated guide: [`docs/StreamingAPI.md`](./docs/StreamingAPI.md)
+
 ## API Reference
 
 The system provides a comprehensive RESTful API for monitoring and controlling the embedded device. All endpoints return JSON responses and support standard HTTP status codes.

docs/StreamingAPI.md

@@ -0,0 +1,98 @@
+## Streaming API (WebSocket)
+
+### Overview
+
+The streaming API exposes an event-driven WebSocket at `/ws`. It bridges between external clients and the internal event bus:
+
+- Incoming WebSocket JSON `{ event, payload }` → `ctx.fire(event, payload)`
+- Local events → broadcasted to all connected WebSocket clients as `{ event, payload }`
+
+This allows real-time control and observation of the system without polling.
+
+### URL
+
+- `ws://<device-ip>/ws`
+
+### Message Format
+
+- Client → Device
+```json
+{
+  "event": "<event-name>",
+  "payload": "<json-string>" | { /* inline JSON */ }
+}
+```
+
+- Device → Client
+```json
+{
+  "event": "<event-name>",
+  "payload": "<json-string>"
+}
+```
+
+Notes:
+- The device accepts `payload` as a string or a JSON object/array. Objects are serialized into a string before dispatching to local subscribers to keep a consistent downstream contract.
+- A minimal ack `{ "ok": true }` is sent after a valid inbound message.
+
+#### Echo suppression (origin tagging)
+
+- To prevent the sender from receiving an immediate echo of its own message, the server injects a private field into JSON payloads:
+  - `_origin: "ws:<clientId>"`
+- When re-broadcasting local events to WebSocket clients, the server:
+  - Strips the `_origin` field from the outgoing payload
+  - Skips the originating `clientId` so only other clients receive the message
+  - If a payload is not valid JSON (plain string), no origin tag is injected and the message may be echoed
+
+### Event Bus Integration
+
+- The WebSocket registers an `onAny` subscriber to `NodeContext` so that all local events are mirrored to clients.
+- Services should subscribe to specific events via `ctx.on("<name>", ...)`.
+
+### Examples
+
+1) Set a solid color on NeoPattern:
+```json
+{
+  "event": "api/neopattern/color",
+  "payload": { "color": "#FF0000", "brightness": 128 }
+}
+```
+
+2) Broadcast a cluster event (delegated to core):
+```json
+{
+  "event": "cluster/broadcast",
+  "payload": {
+    "event": "api/neopattern/color",
+    "data": { "color": "#00FF00", "brightness": 128 }
+  }
+}
+```
+
+### Reference Implementation
+
+- WebSocket setup and bridging are implemented in `ApiServer`.
+- Global event subscription uses `NodeContext::onAny`.
+
+Related docs:
+- [`ClusterBroadcast.md`](./ClusterBroadcast.md) — centralized UDP broadcasting and CLUSTER_EVENT format
+
+### Things to consider
+
+- High-frequency updates can overwhelm ESP8266:
+  - Frequent JSON parse/serialize and `String` allocations fragment heap and may cause resets (e.g., Exception(3)).
+  - UDP broadcast on every message amplifies load; WiFi/UDP buffers can back up.
+  - Prefer ≥50–100 ms intervals; microbursts at 10 ms are risky.
+- Throttle and coalesce:
+  - Add a minimum interval in the core `cluster/broadcast` handler.
+  - Optionally drop redundant updates (e.g., same color as previous).
+- Reduce allocations:
+  - Reuse `StaticJsonDocument`/preallocated buffers in hot paths.
+  - Avoid re-serializing when possible; pass-through payload strings.
+  - Reserve `String` capacity when reuse is needed.
+- Yielding:
+  - Call `yield()` in long-running or bursty paths to avoid WDT.
+- Packet size:
+  - Keep payloads small to fit `ClusterProtocol::UDP_BUF_SIZE` and reduce airtime.
+

examples/neopattern/NeoPatternService.cpp

@@ -205,6 +205,44 @@ void NeoPatternService::registerEventHandlers() {
             LOG_INFO("NeoPattern", "Applied control from CLUSTER_EVENT");
         }
     });
+
+    // Solid color event: sets all pixels to the same color
+    ctx.on("api/neopattern/color", [this](void* dataPtr) {
+        String* jsonStr = static_cast<String*>(dataPtr);
+        if (!jsonStr) {
+            LOG_WARN("NeoPattern", "Received api/neopattern/color with null dataPtr");
+            return;
+        }
+        JsonDocument doc;
+        DeserializationError err = deserializeJson(doc, *jsonStr);
+        if (err) {
+            LOG_WARN("NeoPattern", String("Failed to parse color event data: ") + err.c_str());
+            return;
+        }
+        JsonObject obj = doc.as<JsonObject>();
+        // color can be string or number
+        String colorStr;
+        if (obj["color"].is<const char*>() || obj["color"].is<String>()) {
+            colorStr = obj["color"].as<String>();
+        } else if (obj["color"].is<long>() || obj["color"].is<int>()) {
+            colorStr = String(obj["color"].as<long>());
+        } else {
+            LOG_WARN("NeoPattern", "api/neopattern/color missing 'color'");
+            return;
+        }
+
+        // Optional brightness
+        if (obj["brightness"].is<int>() || obj["brightness"].is<long>()) {
+            int b = obj["brightness"].as<int>();
+            if (b < 0) b = 0; if (b > 255) b = 255;
+            setBrightness(static_cast<uint8_t>(b));
+        }
+
+        uint32_t color = parseColor(colorStr);
+        setPattern(NeoPatternType::NONE);
+        setColor(color);
+        LOG_INFO("NeoPattern", String("Set solid color ") + colorStr);
+    });
 }
 
 bool NeoPatternService::applyControlParams(const JsonObject& obj) {

examples/neopattern/main.cpp

@@ -10,7 +10,7 @@
 #endif
 
 #ifndef NEOPIXEL_LENGTH
-#define NEOPIXEL_LENGTH 8
+#define NEOPIXEL_LENGTH 16
 #endif
 
 #ifndef NEOPIXEL_BRIGHTNESS

include/spore/core/ApiServer.h

@@ -2,6 +2,7 @@
 #include <Arduino.h>
 #include <ArduinoJson.h>
 #include <ESPAsyncWebServer.h>
+#include <AsyncWebSocket.h>
 #include <Updater.h>
 #include <functional>
 #include <vector>
@@ -39,13 +40,18 @@ public:
 
 private:
     AsyncWebServer server;
+    AsyncWebSocket ws{ "/ws" };
     NodeContext& ctx;
     TaskManager& taskManager;
     std::vector<std::reference_wrapper<Service>> services;
     std::vector<EndpointInfo> endpoints;  // Single source of truth for endpoints
+    std::vector<AsyncWebSocketClient*> wsClients;
 
     // Internal helpers
     void registerEndpoint(const String& uri, int method, 
                          const std::vector<ParamSpec>& params,
                          const String& serviceName);
+
+    // WebSocket helpers
+    void setupWebSocket();
 };

include/spore/core/NodeContext.h

@@ -23,7 +23,10 @@ public:
 
     using EventCallback = std::function<void(void*)>;
     std::map<std::string, std::vector<EventCallback>> eventRegistry;
+    using AnyEventCallback = std::function<void(const std::string&, void*)>;
+    std::vector<AnyEventCallback> anyEventSubscribers;
 
     void on(const std::string& event, EventCallback cb);
     void fire(const std::string& event, void* data);
+    void onAny(AnyEventCallback cb);
 };

src/spore/core/ApiServer.cpp

@@ -87,6 +87,10 @@ void ApiServer::serveStatic(const String& uri, fs::FS& fs, const String& path, c
 }
 
 void ApiServer::begin() {
+    // Setup streaming API (WebSocket)
+    setupWebSocket();
+    server.addHandler(&ws);
+    
     // Register all service endpoints
     for (auto& service : services) {
         service.get().registerEndpoints(*this);
@@ -95,3 +99,90 @@ void ApiServer::begin() {
     
     server.begin();
 }
+
+void ApiServer::setupWebSocket() {
+    ws.onEvent([this](AsyncWebSocket* server, AsyncWebSocketClient* client, AwsEventType type, void* arg, uint8_t* data, size_t len) {
+        if (type == WS_EVT_DATA) {
+            AwsFrameInfo* info = (AwsFrameInfo*)arg;
+            if (info->final && info->index == 0 && info->len == len && info->opcode == WS_TEXT) {
+                // Parse directly from the raw buffer with explicit length
+                JsonDocument doc;
+                DeserializationError err = deserializeJson(doc, (const char*)data, len);
+                if (!err) {
+                    LOG_DEBUG("API", "Received event: " + String(doc["event"].as<String>()));
+                    String eventName = doc["event"].as<String>();
+                    String payloadStr;
+                    if (doc["payload"].is<const char*>()) {
+                        payloadStr = doc["payload"].as<const char*>();
+                    } else if (!doc["payload"].isNull()) {
+                        // If payload is an object/array, serialize it
+                        String tmp; serializeJson(doc["payload"], tmp); payloadStr = tmp;
+                    }
+                    // Allow empty payload; services may treat it as defaults
+                    if (eventName.length() > 0) {
+                        // Inject origin tag into payload JSON if possible
+                        String enriched = payloadStr;
+                        if (payloadStr.length() > 0) {
+                            JsonDocument pd;
+                            if (!deserializeJson(pd, payloadStr)) {
+                                pd["_origin"] = String("ws:") + String(client->id());
+                                String tmp; serializeJson(pd, tmp); enriched = tmp;
+                            } else {
+                                // If payload is plain string, leave as-is (no origin)
+                            }
+                        }
+                        std::string ev = eventName.c_str();
+                        ctx.fire(ev, &enriched);
+                        // Acknowledge
+                        client->text("{\"ok\":true}");
+                    } else {
+                        client->text("{\"error\":\"Missing 'event'\"}");
+                    }
+                } else {
+                    client->text("{\"error\":\"Invalid JSON\"}");
+                }
+            }
+        } else if (type == WS_EVT_CONNECT) {
+            client->text("{\"hello\":\"ws connected\"}");
+            wsClients.push_back(client);
+        } else if (type == WS_EVT_DISCONNECT) {
+            wsClients.erase(std::remove(wsClients.begin(), wsClients.end(), client), wsClients.end());
+        }
+    });
+
+    // Subscribe to all local events and forward to websocket clients
+    ctx.onAny([this](const std::string& event, void* dataPtr) {
+        String* payloadStrPtr = static_cast<String*>(dataPtr);
+        String payloadStr = payloadStrPtr ? *payloadStrPtr : String("");
+
+        // Extract and strip origin if present
+        String origin;
+        String cleanedPayload = payloadStr;
+        if (payloadStr.length() > 0) {
+            JsonDocument pd;
+            if (!deserializeJson(pd, payloadStr)) {
+                if (pd["_origin"].is<const char*>()) {
+                    origin = pd["_origin"].as<const char*>();
+                    pd.remove("_origin");
+                    String tmp; serializeJson(pd, tmp); cleanedPayload = tmp;
+                }
+            }
+        }
+
+        JsonDocument outDoc;
+        outDoc["event"] = event.c_str();
+        outDoc["payload"] = cleanedPayload;
+        String out; serializeJson(outDoc, out);
+
+        if (origin.startsWith("ws:")) {
+            uint32_t originId = (uint32_t)origin.substring(3).toInt();
+            for (auto* c : wsClients) {
+                if (c && c->id() != originId) {
+                    c->text(out);
+                }
+            }
+        } else {
+            ws.textAll(out);
+        }
+    });
+}

src/spore/core/ClusterManager.cpp

@@ -19,7 +19,7 @@ ClusterManager::ClusterManager(NodeContext& ctx, TaskManager& taskMgr) : ctx(ctx
         IPAddress ip = WiFi.localIP();
         IPAddress mask = WiFi.subnetMask();
         IPAddress bcast(ip[0] | ~mask[0], ip[1] | ~mask[1], ip[2] | ~mask[2], ip[3] | ~mask[3]);
-        LOG_INFO("Cluster", String("Broadcasting CLUSTER_EVENT to ") + bcast.toString() + " len=" + String(jsonStr->length()));
+        LOG_DEBUG("Cluster", String("Broadcasting CLUSTER_EVENT to ") + bcast.toString() + " len=" + String(jsonStr->length()));
         this->ctx.udp->beginPacket(bcast, this->ctx.config.udp_port);
         String msg = String(ClusterProtocol::CLUSTER_EVENT_MSG) + ":" + *jsonStr;
         this->ctx.udp->write(msg.c_str());
@@ -204,7 +204,7 @@ void ClusterManager::onNodeInfo(const char* msg) {
                 }
             }
         } else {
-            LOG_DEBUG("Cluster", String("Failed to parse NODE_INFO JSON from ") + senderIP.toString());
+            LOG_WARN("Cluster", String("Failed to parse NODE_INFO JSON from ") + senderIP.toString());
         }
     }
 }
@@ -216,11 +216,11 @@ void ClusterManager::onClusterEvent(const char* msg) {
         LOG_DEBUG("Cluster", "CLUSTER_EVENT received with empty payload");
         return;
     }
-    LOG_INFO("Cluster", String("CLUSTER_EVENT raw from ") + ctx.udp->remoteIP().toString() + " len=" + String(strlen(jsonStart)));
+    LOG_DEBUG("Cluster", String("CLUSTER_EVENT raw from ") + ctx.udp->remoteIP().toString() + " len=" + String(strlen(jsonStart)));
     JsonDocument doc;
     DeserializationError err = deserializeJson(doc, jsonStart);
     if (err) {
-        LOG_DEBUG("Cluster", String("Failed to parse CLUSTER_EVENT JSON from ") + ctx.udp->remoteIP().toString());
+        LOG_ERROR("Cluster", String("Failed to parse CLUSTER_EVENT JSON from ") + ctx.udp->remoteIP().toString());
         return;
     }
     // Robust extraction of event and data
@@ -249,7 +249,7 @@ void ClusterManager::onClusterEvent(const char* msg) {
     }
 
     std::string eventKey(eventStr.c_str());
-    LOG_INFO("Cluster", String("Firing event '") + eventStr + "' with dataLen=" + String(data.length()));
+    LOG_DEBUG("Cluster", String("Firing event '") + eventStr + "' with dataLen=" + String(data.length()));
     ctx.fire(eventKey, &data);
 }
 

src/spore/core/NodeContext.cpp

@@ -29,4 +29,11 @@ void NodeContext::fire(const std::string& event, void* data) {
     for (auto& cb : eventRegistry[event]) {
         cb(data);
     }
+    for (auto& acb : anyEventSubscribers) {
+        acb(event, data);
+    }
+}
+
+void NodeContext::onAny(AnyEventCallback cb) {
+    anyEventSubscribers.push_back(cb);
 }

test/.gitignore

@@ -0,0 +1 @@
+node_modules/

test/README

@@ -1,11 +1,64 @@
+# Test Scripts
 
-This directory is intended for PlatformIO Test Runner and project tests.
+This directory contains JavaScript test scripts to interact with the Spore device, primarily for testing cluster event broadcasting.
 
-Unit Testing is a software testing method by which individual units of
-source code, sets of one or more MCU program modules together with associated
-control data, usage procedures, and operating procedures, are tested to
-determine whether they are fit for use. Unit testing finds problems early
-in the development cycle.
+## Prerequisites
+
+These scripts require [Node.js](https://nodejs.org/) to be installed on your system.
+
+## How to Run
+
+### 1. HTTP Cluster Broadcast Color (`test/http-cluster-broadcast-color.js`)
+
+This script sends HTTP POST requests to the `/api/cluster/event` endpoint on your Spore device. It broadcasts NeoPattern color changes across the cluster every 5 seconds.
+
+**Usage:**
+```
+node test/http-cluster-broadcast-color.js <device-ip>
+```
+Example:
+```
+node test/http-cluster-broadcast-color.js 10.0.1.53
+```
+This will broadcast `{ event: "api/neopattern/color", data: { color: "#RRGGBB", brightness: 128 } }` every 5 seconds to the cluster via `/api/cluster/event`.
+
+### 2. WS Local Color Setter (`test/ws-color-client.js`)
+
+Connects to the device WebSocket (`/ws`) and sets a solid color locally (non-broadcast) every 5 seconds by firing `api/neopattern/color`.
+
+**Usage:**
+```
+node test/ws-color-client.js ws://<device-ip>/ws
+```
+Example:
+```
+node test/ws-color-client.js ws://10.0.1.53/ws
+```
+
+### 3. WS Cluster Broadcast Color (`test/ws-cluster-broadcast-color.js`)
+
+Connects to the device WebSocket (`/ws`) and broadcasts a color change to all peers every 5 seconds by firing `cluster/broadcast` with the proper envelope.
+
+**Usage:**
+```
+node test/ws-cluster-broadcast-color.js ws://<device-ip>/ws
+```
+Example:
+```
+node test/ws-cluster-broadcast-color.js ws://10.0.1.53/ws
+```
+
+### 4. WS Cluster Broadcast Rainbow (`test/ws-cluster-broadcast-rainbow.js`)
+
+Broadcasts a smooth rainbow color transition over WebSocket using `cluster/broadcast` and the `api/neopattern/color` event. Update rate defaults to `UPDATE_RATE` in the script (e.g., 100 ms).
+
+**Usage:**
+```
+node test/ws-cluster-broadcast-rainbow.js ws://<device-ip>/ws
+```
+Example:
+```
+node test/ws-cluster-broadcast-rainbow.js ws://10.0.1.53/ws
+```
+Note: Very fast update intervals (e.g., 10 ms) may saturate links or the device.
 
-More information about PlatformIO Unit Testing:
-- https://docs.platformio.org/en/latest/advanced/unit-testing/index.html

test/http-cluster-broadcast-color.js

@@ -0,0 +1,52 @@
+// Simple HTTP client to broadcast a neopattern color change to the cluster
+// Usage: node cluster-broadcast-color.js 10.0.1.53
+
+const http = require('http');
+
+const host = process.argv[2] || '127.0.0.1';
+const port = 80;
+
+const colors = ['#FF0000', '#00FF00', '#0000FF', '#FFFF00', '#FF00FF', '#00FFFF'];
+let idx = 0;
+
+function postClusterEvent(event, payloadObj) {
+  const payload = encodeURIComponent(JSON.stringify(payloadObj));
+  const body = `event=${encodeURIComponent(event)}&payload=${payload}`;
+
+  const options = {
+    host,
+    port,
+    path: '/api/cluster/event',
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/x-www-form-urlencoded',
+      'Content-Length': Buffer.byteLength(body)
+    }
+  };
+
+  const req = http.request(options, (res) => {
+    let data = '';
+    res.on('data', (chunk) => (data += chunk));
+    res.on('end', () => {
+      console.log('Response:', res.statusCode, data);
+    });
+  });
+
+  req.on('error', (err) => {
+    console.error('Request error:', err.message);
+  });
+
+  req.write(body);
+  req.end();
+}
+
+console.log(`Broadcasting color changes to http://${host}/api/cluster/event ...`);
+setInterval(() => {
+  const color = colors[idx % colors.length];
+  idx++;
+  const payload = { color, brightness: 128 };
+  console.log('Broadcasting color:', payload);
+  postClusterEvent('api/neopattern/color', payload);
+}, 5000);
+
+

test/package-lock.json

@@ -0,0 +1,33 @@
+{
+  "name": "test",
+  "lockfileVersion": 3,
+  "requires": true,
+  "packages": {
+    "": {
+      "dependencies": {
+        "ws": "^8.18.3"
+      }
+    },
+    "node_modules/ws": {
+      "version": "8.18.3",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-8.18.3.tgz",
+      "integrity": "sha512-PEIGCY5tSlUt50cqyMXfCzX+oOPqN0vuGqWzbcJ2xvnkzkq46oOpz7dQaTDBdfICb4N14+GARUDw2XV2N4tvzg==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=10.0.0"
+      },
+      "peerDependencies": {
+        "bufferutil": "^4.0.1",
+        "utf-8-validate": ">=5.0.2"
+      },
+      "peerDependenciesMeta": {
+        "bufferutil": {
+          "optional": true
+        },
+        "utf-8-validate": {
+          "optional": true
+        }
+      }
+    }
+  }
+}

test/package.json

@@ -0,0 +1,5 @@
+{
+  "dependencies": {
+    "ws": "^8.18.3"
+  }
+}

test/ws-cluster-broadcast-color.js

@@ -0,0 +1,46 @@
+// WebSocket client to broadcast neopattern color changes across the cluster
+// Usage: node ws-cluster-broadcast-color.js ws://<device-ip>/ws
+
+const WebSocket = require('ws');
+
+const url = process.argv[2] || 'ws://127.0.0.1/ws';
+const ws = new WebSocket(url);
+
+const colors = ['#FF0000', '#00FF00', '#0000FF', '#FFFF00', '#FF00FF', '#00FFFF'];
+let idx = 0;
+
+ws.on('open', () => {
+  console.log('Connected to', url);
+  // Broadcast color change every 5 seconds via cluster/broadcast
+  setInterval(() => {
+    const color = colors[idx % colors.length];
+    idx++;
+    const payload = { color, brightness: 128 };
+    const envelope = {
+      event: 'api/neopattern/color',
+      data: payload // server will serialize object payloads
+    };
+    const msg = { event: 'cluster/broadcast', payload: envelope };
+    ws.send(JSON.stringify(msg));
+    console.log('Broadcasted color event', payload);
+  }, 5000);
+});
+
+ws.on('message', (data) => {
+  try {
+    const msg = JSON.parse(data.toString());
+    console.log('Received:', msg);
+  } catch (e) {
+    console.log('Received raw:', data.toString());
+  }
+});
+
+ws.on('error', (err) => {
+  console.error('WebSocket error:', err.message);
+});
+
+ws.on('close', () => {
+  console.log('WebSocket closed');
+});
+
+

test/ws-cluster-broadcast-rainbow.js

@@ -0,0 +1,71 @@
+// WebSocket client to broadcast smooth rainbow color changes across the cluster
+// Usage: node ws-cluster-broadcast-rainbow.js ws://<device-ip>/ws
+
+const WebSocket = require('ws');
+
+const url = process.argv[2] || 'ws://127.0.0.1/ws';
+const ws = new WebSocket(url);
+
+function hsvToRgb(h, s, v) {
+  const c = v * s;
+  const x = c * (1 - Math.abs(((h / 60) % 2) - 1));
+  const m = v - c;
+  let r = 0, g = 0, b = 0;
+  if (h < 60) { r = c; g = x; b = 0; }
+  else if (h < 120) { r = x; g = c; b = 0; }
+  else if (h < 180) { r = 0; g = c; b = x; }
+  else if (h < 240) { r = 0; g = x; b = c; }
+  else if (h < 300) { r = x; g = 0; b = c; }
+  else { r = c; g = 0; b = x; }
+  const R = Math.round((r + m) * 255);
+  const G = Math.round((g + m) * 255);
+  const B = Math.round((b + m) * 255);
+  return { r: R, g: G, b: B };
+}
+
+function toHex({ r, g, b }) {
+  const h = (n) => n.toString(16).padStart(2, '0').toUpperCase();
+  return `#${h(r)}${h(g)}${h(b)}`;
+}
+
+let hue = 0;
+const SAT = 1.0;   // full saturation
+const VAL = 1.0;   // full value
+const BRIGHTNESS = 128;
+const UPDATE_RATE = 100; // ms
+
+let timer = null;
+
+ws.on('open', () => {
+  console.log('Connected to', url);
+  // UPDATE_RATE ms updates (10 Hz). Be aware this can saturate slow links.
+  timer = setInterval(() => {
+    const rgb = hsvToRgb(hue, SAT, VAL);
+    const color = toHex(rgb);
+    const envelope = {
+      event: 'api/neopattern/color',
+      data: { color, brightness: BRIGHTNESS }
+    };
+    const msg = { event: 'cluster/broadcast', payload: envelope };
+    try {
+      ws.send(JSON.stringify(msg));
+    } catch (_) {}
+    hue = (hue + 2) % 360; // advance hue (adjust for speed)
+  }, UPDATE_RATE);
+});
+
+ws.on('message', (data) => {
+  // Optionally throttle logs: comment out for quieter output
+  // console.log('WS:', data.toString());
+});
+
+ws.on('error', (err) => {
+  console.error('WebSocket error:', err.message);
+});
+
+ws.on('close', () => {
+  if (timer) clearInterval(timer);
+  console.log('WebSocket closed');
+});
+
+

test/ws-color-client.js

@@ -0,0 +1,48 @@
+// Simple WebSocket client to test streaming API color changes
+// Usage: node ws-color-client.js ws://<device-ip>/ws
+
+const WebSocket = require('ws');
+
+const url = process.argv[2] || 'ws://127.0.0.1/ws';
+const ws = new WebSocket(url);
+
+const colors = [
+  '#FF0000', // red
+  '#00FF00', // green
+  '#0000FF', // blue
+  '#FFFF00', // yellow
+  '#FF00FF', // magenta
+  '#00FFFF'  // cyan
+];
+let idx = 0;
+
+ws.on('open', () => {
+  console.log('Connected to', url);
+  // Send a message every 5 seconds to set solid color
+  setInterval(() => {
+    const color = colors[idx % colors.length];
+    idx++;
+    const payload = { color, brightness: 128 };
+    // Send payload as an object (server supports string or object)
+    const msg = { event: 'api/neopattern/color', payload };
+    ws.send(JSON.stringify(msg));
+    console.log('Sent color event', payload);
+  }, 5000);
+});
+
+ws.on('message', (data) => {
+  try {
+    const msg = JSON.parse(data.toString());
+    console.log('Received:', msg);
+  } catch (e) {
+    console.log('Received raw:', data.toString());
+  }
+});
+
+ws.on('error', (err) => {
+  console.error('WebSocket error:', err.message);
+});
+
+ws.on('close', () => {
+  console.log('WebSocket closed');
+});