feat(ws_bridge): add pushed camera video

.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

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

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))

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

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<unsigned>(get_update_interval()));
+  if (video_)
+    ESP_LOGCONFIG(TAG, "  Live MJPEG: %u FPS, %u bytes/s", static_cast<unsigned>(video_fps_),
+                  static_cast<unsigned>(video_max_bandwidth_));
+  else
+    ESP_LOGCONFIG(TAG, "  Capture interval: %u ms", static_cast<unsigned>(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<int32_t>(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<int32_t>(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<JsonArray>();
+      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<std::mutex> 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<int32_t>(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<std::mutex> lock(frame_mutex_);
   if (capture_requested_)
@@ -133,7 +196,7 @@ void WsBridgeSnapshot::on_camera_image(const std::shared_ptr<camera::CameraImage
     return;  // do not steal webserver/API camera requests
   {
     std::lock_guard<std::mutex> 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<int32_t>(millis() - send_gate) < 0)
+    return;
   std::shared_ptr<camera::CameraImage> image;
   {
     std::lock_guard<std::mutex> lock(frame_mutex_);
@@ -408,6 +485,13 @@ void WsBridgeSnapshot::send_pending_() {
   if (first != 1 || body != static_cast<int>(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<uint64_t>(image->get_data_length()) * 1000U + bandwidth - 1) / bandwidth;
+    uint32_t gate = millis() + static_cast<uint32_t>(std::min<uint64_t>(delay, UINT32_MAX));
+    next_video_send_ms_.store(gate == 0 ? 1 : gate, std::memory_order_release);
   }
 }