diff --git a/.github/workflows/ws-bridge-video-schema-tests.yml b/.github/workflows/ws-bridge-video-schema-tests.yml new file mode 100644 index 00000000..8843c382 --- /dev/null +++ b/.github/workflows/ws-bridge-video-schema-tests.yml @@ -0,0 +1,32 @@ +name: ws_bridge Video Schema Tests + +on: + push: + branches: [ main, master ] + paths: + - 'components/ws_bridge/**' + - 'tests/native/ws_bridge_video_schema/**' + pull_request: + branches: [ main, master ] + paths: + - 'components/ws_bridge/**' + - 'tests/native/ws_bridge_video_schema/**' + workflow_dispatch: + +jobs: + ws-bridge-video-schema-test: + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v5 + + - name: Set up Python + uses: actions/setup-python@v6 + with: + python-version: '3.14' + + - name: Install ESPHome + run: pip install esphome + + - name: Run schema tests + run: python tests/native/ws_bridge_video_schema/test_schema.py diff --git a/components/ws_bridge/README.md b/components/ws_bridge/README.md index d1318ffc..e5e7c838 100644 --- a/components/ws_bridge/README.md +++ b/components/ws_bridge/README.md @@ -271,7 +271,7 @@ left out. `datetime_id:` must point at an entity of the same `type:`. strings (`2026-08-13`, `21:30:00`, `2026-08-13T21:30:00`); Home Assistant attaches its own local timezone to them rather than reading them as UTC. -## Camera snapshots +## Camera snapshots and live camera video `snapshots:` exposes JPEG frames from an `esp32_camera` as either a Home Assistant `camera` (the default) or `image`. It requires **ESPHome 2026.10+ @@ -299,6 +299,46 @@ ws_bridge: update_interval: 2min ``` +Those are still photos: a frame is requested on `update_interval`, even if no +camera card is open. To opt a **camera** into viewer-driven live MJPEG instead, +set `video: true` (an `image` cannot be a live stream): + +```yaml +ws_bridge: + # ...host and token... + snapshots: + - camera_id: front_sensor + unique_id: front_live + name: "Front live camera" + video: true + video_fps: 5 # local capture limit, 1..30 + video_max_bandwidth: 256000 # local upload limit, bytes/second +``` + +Video declarations advertise only the `push_stream` camera capability. Home +Assistant sends `start_stream` when its first native MJPEG viewer opens and +`stop_stream` after the final viewer leaves; no capture is requested while +stopped. Repeated commands are safe. This is pushed JPEG/MJPEG, not HLS, does +not advertise HA's `STREAM` feature, and needs neither an ESP HTTP server nor +an inbound HA connection. The FPS and bandwidth values are local limits, not +wire-protocol settings; under backpressure only the newest completed frame is +kept, and no capture is requested while the bandwidth limit is holding uploads +back. + +Things to know about `video: true`: + +- `video_fps` and `video_max_bandwidth` are accepted only together with + `video: true` (defaults 5 FPS, 256000 bytes/s; bandwidth must be at least 1). +- `update_interval` is ignored for a video item. +- With no viewer there is no capture and no media connection, and Home Assistant + drops its cached frame, so the entity picture and the card's still thumbnail + are empty until someone watches. Add a separate still snapshot of the same + camera (without `video`) if you want a thumbnail. +- Home Assistant needs a hass-ws-bridge release with `push_stream` support; + older integrations never send `start_stream`, so no video appears. +- A repeated `start_stream` is ignored, and a `stop_stream` followed quickly by + a new `start_stream` is applied in order. + Each list item has a separate, ESP-initiated media WebSocket because Home Assistant binds one media connection to exactly one declared entity. The firmware first uses the existing authenticated control connection to request diff --git a/components/ws_bridge/__init__.py b/components/ws_bridge/__init__.py index aa8ac8eb..1a245746 100644 --- a/components/ws_bridge/__init__.py +++ b/components/ws_bridge/__init__.py @@ -62,6 +62,9 @@ CONF_SNAPSHOTS, CONF_CAMERA_ID, CONF_PLATFORM, + CONF_VIDEO, + CONF_VIDEO_FPS, + CONF_VIDEO_MAX_BANDWIDTH, ) CODEOWNERS = ["@eigger"] @@ -152,7 +155,18 @@ } ) -SNAPSHOT_SCHEMA = cv.Schema( +def _validate_snapshot(conf): + if conf[CONF_VIDEO] and conf[CONF_PLATFORM] != "camera": + raise cv.Invalid("video is supported only with platform: camera") + if not conf[CONF_VIDEO]: + for key in (CONF_VIDEO_FPS, CONF_VIDEO_MAX_BANDWIDTH): + if key in conf: + raise cv.Invalid(f"{key} requires video: true") + return conf + + +SNAPSHOT_SCHEMA = cv.All( + cv.Schema( { cv.GenerateID(): cv.declare_id(WsBridgeSnapshot), # Kept loosely typed until to_code so importing this component alone @@ -163,8 +177,17 @@ cv.Required(CONF_NAME): cv.string_strict, # HA's image and camera entities share the same JPEG upload transport. cv.Optional(CONF_PLATFORM, default="camera"): cv.one_of("camera", "image", lower=True), + # Live video is an explicit opt-in. A video item declares only + # push_stream and captures only while HA has an MJPEG viewer. + cv.Optional(CONF_VIDEO, default=False): cv.boolean, + cv.Optional(CONF_VIDEO_FPS): cv.int_range(min=1, max=30), + # Bytes per second; this is intentionally local policy, never a + # protocol field negotiated with Home Assistant. + cv.Optional(CONF_VIDEO_MAX_BANDWIDTH): cv.int_range(min=1), } -).extend(cv.polling_component_schema("60s")) + ).extend(cv.polling_component_schema("60s")), + _validate_snapshot, +) async def to_code_entity_ref(hub_var, conf): @@ -440,3 +463,7 @@ async def to_code(config): cg.add(snapshot.set_unique_id(conf[CONF_UNIQUE_ID])) cg.add(snapshot.set_name(conf[CONF_NAME])) cg.add(snapshot.set_platform(conf[CONF_PLATFORM])) + cg.add(snapshot.set_video(conf[CONF_VIDEO])) + cg.add(snapshot.set_video_fps(conf.get(CONF_VIDEO_FPS, 5))) + cg.add(snapshot.set_video_max_bandwidth(conf.get(CONF_VIDEO_MAX_BANDWIDTH, 256000))) + cg.add(var.register_device(snapshot)) diff --git a/components/ws_bridge/const.py b/components/ws_bridge/const.py index 851ca4b4..1effc960 100644 --- a/components/ws_bridge/const.py +++ b/components/ws_bridge/const.py @@ -13,6 +13,9 @@ CONF_SNAPSHOTS = "snapshots" CONF_CAMERA_ID = "camera_id" CONF_PLATFORM = "platform" +CONF_VIDEO = "video" +CONF_VIDEO_FPS = "video_fps" +CONF_VIDEO_MAX_BANDWIDTH = "video_max_bandwidth" CONF_UNIQUE_ID = "unique_id" # Prefixed with ws_ to avoid colliding with ESPHome's own reserved diff --git a/components/ws_bridge/ws_bridge_snapshot.cpp b/components/ws_bridge/ws_bridge_snapshot.cpp index 6557ac10..2bce20f7 100644 --- a/components/ws_bridge/ws_bridge_snapshot.cpp +++ b/components/ws_bridge/ws_bridge_snapshot.cpp @@ -29,9 +29,6 @@ void WsBridgeSnapshot::setup() { return; } camera_->add_listener(this); - parent_->add_on_declare_callback([this]() { - parent_->send_entity_declare(unique_id_, platform_, name_, "", "", nullptr); - }); parent_->add_on_connected_callback([this]() { on_control_connected_(); }); parent_->add_on_disconnected_callback([this]() { on_control_disconnected_(); }); parent_->add_message_callback([this](const ParsedMessage &message) { on_control_message_(message); }); @@ -40,7 +37,11 @@ void WsBridgeSnapshot::setup() { void WsBridgeSnapshot::dump_config() { ESP_LOGCONFIG(TAG, "WS Bridge snapshot:"); ESP_LOGCONFIG(TAG, " Entity: %s (%s)", unique_id_.c_str(), platform_.c_str()); - ESP_LOGCONFIG(TAG, " Capture interval: %u ms", static_cast(get_update_interval())); + if (video_) + ESP_LOGCONFIG(TAG, " Live MJPEG: %u FPS, %u bytes/s", static_cast(video_fps_), + static_cast(video_max_bandwidth_)); + else + ESP_LOGCONFIG(TAG, " Capture interval: %u ms", static_cast(get_update_interval())); } void WsBridgeSnapshot::loop() { @@ -51,19 +52,79 @@ void WsBridgeSnapshot::loop() { if (media_open_id_ != 0 && millis() - media_open_sent_ms_ >= MEDIA_OPEN_TIMEOUT_MS) { schedule_open_retry_(); } - if (needs_media_open_.load(std::memory_order_acquire) && media_open_id_ == 0) + if (needs_media_open_.load(std::memory_order_acquire) && media_open_id_ == 0 && media_demand_()) open_media_(); + if (video_ && stream_requested_.load(std::memory_order_acquire) && media_ready_.load(std::memory_order_acquire)) { + uint32_t now = millis(); + uint32_t send_gate = next_video_send_ms_.load(std::memory_order_acquire); + bool upload_throttled = send_gate != 0 && static_cast(now - send_gate) < 0; + // While the bandwidth limit holds uploads back, a capture would only be dropped. + if (!upload_throttled && (next_video_capture_ms_ == 0 || static_cast(now - next_video_capture_ms_) >= 0)) { + next_video_capture_ms_ = now + video_frame_interval_ms_; + request_frame_(); + } + } +} + +void WsBridgeSnapshot::update() { + if (!video_) + request_frame_(); +} + +bool WsBridgeSnapshot::media_demand_() const { + return !video_ || stream_requested_.load(std::memory_order_acquire); +} + +void WsBridgeSnapshot::ws_bridge_declare() { + if (is_failed()) + return; + parent_->send_entity_declare(unique_id_, platform_, name_, "", "", [this](JsonObject root) { + if (video_) { + JsonArray features = root["features"].to(); + features.add("push_stream"); + } + }); } -void WsBridgeSnapshot::update() { request_frame_(); } +void WsBridgeSnapshot::ws_bridge_handle_command(const WsCommand &command) { + if (!video_) + return; + std::string format; + if (!command.param_string("format", format) || format != "mjpeg") + return; + if (command.action == "start_stream") { + // Idempotent: a repeat must not tear down a working or pending session. + if (stream_requested_.exchange(true, std::memory_order_acq_rel)) + return; + // A fresh viewer should not wait out the previous session's open backoff. + next_media_open_ms_ = 0; + next_video_capture_ms_ = 0; + needs_media_open_.store(true, std::memory_order_release); + } else if (command.action == "stop_stream") { + // Idempotent: capture requests issued before this command are discarded by + // on_camera_image(), and the worker owns the actual socket teardown. + stream_requested_.store(false, std::memory_order_release); + next_video_capture_ms_ = 0; + needs_media_open_.store(false, std::memory_order_release); + media_open_id_ = 0; + { + std::lock_guard lock(frame_mutex_); + capture_requested_ = false; + pending_image_.reset(); + } + request_stop_(); + } +} + +void WsBridgeSnapshot::request_stop_() { stop_epoch_.fetch_add(1, std::memory_order_acq_rel); } void WsBridgeSnapshot::on_control_connected_() { media_open_id_ = 0; media_ready_.store(false, std::memory_order_release); handler_id_.store(0, std::memory_order_release); - stop_requested_.store(true, std::memory_order_release); + request_stop_(); next_media_open_ms_ = 0; - needs_media_open_.store(true, std::memory_order_release); + needs_media_open_.store(media_demand_(), std::memory_order_release); } void WsBridgeSnapshot::on_control_disconnected_() { @@ -75,11 +136,12 @@ void WsBridgeSnapshot::on_control_disconnected_() { capture_requested_ = false; pending_image_.reset(); } - stop_requested_.store(true, std::memory_order_release); + request_stop_(); + stream_requested_.store(false, std::memory_order_release); } void WsBridgeSnapshot::open_media_() { - if (parent_ == nullptr || !parent_->is_connected() || media_open_id_ != 0) + if (parent_ == nullptr || !parent_->is_connected() || media_open_id_ != 0 || !media_demand_()) return; uint32_t now = millis(); if (next_media_open_ms_ != 0 && static_cast(now - next_media_open_ms_) < 0) @@ -99,7 +161,7 @@ void WsBridgeSnapshot::open_media_() { void WsBridgeSnapshot::schedule_open_retry_() { media_open_id_ = 0; - needs_media_open_.store(true, std::memory_order_release); + needs_media_open_.store(media_demand_(), std::memory_order_release); next_media_open_ms_ = millis() + media_open_backoff_ms_; media_open_backoff_ms_ = std::min(media_open_backoff_ms_ * 2, MEDIA_OPEN_BACKOFF_MAX_MS); } @@ -114,12 +176,13 @@ void WsBridgeSnapshot::on_control_message_(const ParsedMessage &message) { } auto *item = new WorkItem{WorkType::START}; item->session_id = message.session_id; // opaque one-use session; never logged + item->stop_epoch = stop_epoch_.load(std::memory_order_acquire); if (!enqueue_(item)) schedule_open_retry_(); } void WsBridgeSnapshot::request_frame_() { - if (!media_ready_.load(std::memory_order_acquire) || camera_ == nullptr) + if (!media_ready_.load(std::memory_order_acquire) || camera_ == nullptr || !media_demand_()) return; std::lock_guard lock(frame_mutex_); if (capture_requested_) @@ -133,7 +196,7 @@ void WsBridgeSnapshot::on_camera_image(const std::shared_ptr lock(frame_mutex_); - if (!capture_requested_) + if (!capture_requested_ || !media_demand_()) return; capture_requested_ = false; if (image->get_data_length() == 0 || image->get_data_length() > MAX_JPEG_BYTES) @@ -162,8 +225,11 @@ void WsBridgeSnapshot::worker_loop_() { uint32_t terminal_generation = terminal_event_generation_.exchange(0, std::memory_order_acq_rel); if (terminal_generation != 0 && terminal_generation == active_generation_) transport_failed_(); - if (stop_requested_.exchange(false, std::memory_order_acq_rel)) + uint32_t stop_epoch = stop_epoch_.load(std::memory_order_acquire); + if (stop_epoch != handled_stop_epoch_) { + handled_stop_epoch_ = stop_epoch; stop_media_(); + } WorkItem *item = nullptr; if (xQueueReceive(worker_queue_, &item, pdMS_TO_TICKS(100)) == pdPASS && item != nullptr) { process_item_(item); @@ -177,6 +243,13 @@ void WsBridgeSnapshot::worker_loop_() { void WsBridgeSnapshot::process_item_(WorkItem *item) { switch (item->type) { case WorkType::START: + // start_media_() replaces any session, so stops issued before this START + // are satisfied; later ones are still caught by the epoch check above. + handled_stop_epoch_ = item->stop_epoch; + // A stop issued after this START was queued makes the session useless: + // skip the connect instead of opening and immediately destroying it. + if (item->stop_epoch != stop_epoch_.load(std::memory_order_acquire)) + break; start_media_(item->session_id); break; case WorkType::WAKE: @@ -227,6 +300,7 @@ void WsBridgeSnapshot::stop_media_() { handler_id_.store(0, std::memory_order_release); worker_session_id_.clear(); media_connect_id_ = 0; + next_video_send_ms_.store(0, std::memory_order_release); media_stage_ = MediaStage::IDLE; media_stage_started_ms_ = 0; rx_accum_.clear(); @@ -266,7 +340,7 @@ void WsBridgeSnapshot::transport_failed_() { // peer may still await its FIN. Stop/destroy on the owner worker and obtain // a fresh one-use HA session through the control channel. stop_media_(); - needs_media_open_.store(true, std::memory_order_release); + needs_media_open_.store(media_demand_(), std::memory_order_release); } void WsBridgeSnapshot::check_worker_deadline_() { @@ -388,6 +462,9 @@ void WsBridgeSnapshot::handle_media_message_(const std::string &raw) { void WsBridgeSnapshot::send_pending_() { if (!media_ready_.load(std::memory_order_acquire) || media_ == nullptr || !esp_websocket_client_is_connected(media_)) return; + uint32_t send_gate = next_video_send_ms_.load(std::memory_order_acquire); + if (video_ && send_gate != 0 && static_cast(millis() - send_gate) < 0) + return; std::shared_ptr image; { std::lock_guard lock(frame_mutex_); @@ -408,6 +485,13 @@ void WsBridgeSnapshot::send_pending_() { if (first != 1 || body != static_cast(image->get_data_length()) || finish < 0) { ESP_LOGW(TAG, "JPEG upload failed; reopening media connection"); transport_failed_(); + } else if (video_) { + // A slow transport deliberately drops intermediate captures: the one + // pending slot is always replaced by the newest completed camera frame. + uint32_t bandwidth = video_max_bandwidth_ == 0 ? 1 : video_max_bandwidth_; + uint64_t delay = (static_cast(image->get_data_length()) * 1000U + bandwidth - 1) / bandwidth; + uint32_t gate = millis() + static_cast(std::min(delay, UINT32_MAX)); + next_video_send_ms_.store(gate == 0 ? 1 : gate, std::memory_order_release); } } diff --git a/components/ws_bridge/ws_bridge_snapshot.h b/components/ws_bridge/ws_bridge_snapshot.h index c93365b9..1ebcab61 100644 --- a/components/ws_bridge/ws_bridge_snapshot.h +++ b/components/ws_bridge/ws_bridge_snapshot.h @@ -17,6 +17,7 @@ #include "freertos/queue.h" #include "ws_protocol.h" +#include "ws_bridge_device.h" namespace esphome::ws_bridge { @@ -24,19 +25,28 @@ class WsBridgeComponent; // The ESPHome loop owns camera/control decisions. A dedicated worker owns all // esp_websocket_client calls, including start/stop/destroy and binary writes. -class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener { +class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener, public WsBridgeDevice { public: - void set_parent(WsBridgeComponent *parent) { parent_ = parent; } + void set_parent(WsBridgeComponent *parent) { set_ws_bridge_parent(parent); } void set_camera(camera::Camera *camera) { camera_ = camera; } - void set_unique_id(const std::string &id) { unique_id_ = id; } void set_name(const std::string &name) { name_ = name; } void set_platform(const std::string &platform) { platform_ = platform; } + void set_video(bool video) { video_ = video; } + void set_video_fps(uint8_t fps) { + video_fps_ = fps; + video_frame_interval_ms_ = 1000 / fps; + } + void set_video_max_bandwidth(uint32_t bytes_per_second) { + video_max_bandwidth_ = bytes_per_second == 0 ? 1 : bytes_per_second; + } void setup() override; void loop() override; void update() override; void on_camera_image(const std::shared_ptr &image) override; void dump_config() override; + void ws_bridge_declare() override; + void ws_bridge_handle_command(const WsCommand &command) override; protected: enum class WorkType : uint8_t { START, WAKE, EVENT }; @@ -50,6 +60,7 @@ class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener int payload_offset{0}; bool fin{false}; uint32_t generation{0}; + uint32_t stop_epoch{0}; // START only: stop requests already issued std::string text; std::string session_id; }; @@ -70,18 +81,24 @@ class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener void handle_media_event_(WorkItem *item); void handle_media_message_(const std::string &raw); void send_pending_(); + void request_stop_(); void on_control_connected_(); void on_control_disconnected_(); void on_control_message_(const ParsedMessage &message); void open_media_(); void schedule_open_retry_(); void request_frame_(); + bool media_demand_() const; - WsBridgeComponent *parent_{nullptr}; camera::Camera *camera_{nullptr}; - std::string unique_id_; std::string name_; std::string platform_{"camera"}; + bool video_{false}; + uint8_t video_fps_{5}; + uint32_t video_frame_interval_ms_{200}; + uint32_t video_max_bandwidth_{256000}; + uint32_t next_video_capture_ms_{0}; // ESPHome loop only + std::atomic next_video_send_ms_{0}; // written by worker; read by loop to skip throttled captures QueueHandle_t worker_queue_{nullptr}; TaskHandle_t worker_task_{nullptr}; esp_websocket_client_handle_t media_{nullptr}; // worker task only @@ -103,7 +120,11 @@ class WsBridgeSnapshot : public PollingComponent, public camera::CameraListener std::atomic handler_id_{0}; std::atomic media_ready_{false}; std::atomic needs_media_open_{false}; - std::atomic stop_requested_{false}; + std::atomic stream_requested_{false}; + // Stop requests are an epoch, not a flag, so a START can record which stops + // precede it: the worker never tears down a session opened after a stop. + std::atomic stop_epoch_{0}; + uint32_t handled_stop_epoch_{0}; // worker task only // Event callbacks cannot read worker-only active_generation_. On a full // queue they record only their immutable generation; the worker compares it // before deciding whether the current client may be stopped. diff --git a/tests/components/ws_bridge/test.video.esp32-idf.yaml b/tests/components/ws_bridge/test.video.esp32-idf.yaml new file mode 100644 index 00000000..04120c05 --- /dev/null +++ b/tests/components/ws_bridge/test.video.esp32-idf.yaml @@ -0,0 +1,48 @@ +esphome: + name: test-ws-bridge-video + +esp32: + board: esp32dev + framework: + type: esp-idf + +external_components: + - source: + type: local + path: ../../../components + components: [ws_bridge] + +wifi: + ssid: "test_ssid" + password: "test_password" + +psram: + mode: quad + speed: 80MHz + +esp32_camera: + id: test_camera + name: "Test Camera" + external_clock: + pin: GPIO0 + frequency: 20MHz + i2c_pins: + sda: GPIO26 + scl: GPIO27 + data_pins: [GPIO5, GPIO18, GPIO19, GPIO21, GPIO36, GPIO39, GPIO34, GPIO35] + vsync_pin: GPIO25 + href_pin: GPIO23 + pixel_clock_pin: GPIO22 + resolution: 320x240 + +ws_bridge: + id: bridge + host: 192.0.2.1 + token: "test_token" + snapshots: + - camera_id: test_camera + unique_id: front_live + name: "Front Live Camera" + video: true + video_fps: 7 + video_max_bandwidth: 192000 diff --git a/tests/native/ws_bridge_video_schema/test_schema.py b/tests/native/ws_bridge_video_schema/test_schema.py new file mode 100644 index 00000000..7834bb7d --- /dev/null +++ b/tests/native/ws_bridge_video_schema/test_schema.py @@ -0,0 +1,40 @@ +"""Config-validation regression tests for ws_bridge snapshot video options. + +Run: python tests/native/ws_bridge_video_schema/test_schema.py +Requires the esphome Python package. +""" +import importlib +import sys +from pathlib import Path + +import esphome.config_validation as cv + +sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "components")) +ws = importlib.import_module("ws_bridge") + +BASE = {"camera_id": "cam", "unique_id": "front", "name": "Front"} + + +def accepts(**extra): + try: + ws.SNAPSHOT_SCHEMA({**BASE, **extra}) + return True + except cv.Invalid: + return False + + +def main(): + assert accepts(), "plain still snapshot" + assert accepts(video=True), "video with defaults" + assert accepts(video=True, video_fps=30, video_max_bandwidth=1) + assert not accepts(video=True, video_max_bandwidth=0), "bandwidth 0 divides by zero" + assert not accepts(video=True, video_fps=0) + assert not accepts(video=True, video_fps=31) + assert not accepts(video=True, platform="image"), "image cannot stream" + assert not accepts(video_fps=5), "video_fps without video" + assert not accepts(video_max_bandwidth=1000), "bandwidth without video" + print("ws_bridge video schema: OK") + + +if __name__ == "__main__": + main()