Add Auto live view option with transcoding options (#24519)

* implement auto mode for single camera live view

* add transcoded live streams and stream ordering for auto mode

Cameras can now add lower quality live streams that go2rtc transcodes to H.264 on demand. `live.transcode` takes a source stream and a list of heights and bitrates, and each quality becomes a `{camera}_transcode_{height}p` stream. The config validator adds them to `live.streams` without moving any the user already placed, and drops them when transcoding is disabled or a height changes. `create_config.py` writes them into go2rtc's generated config at startup, and saving the config or deleting a camera syncs them through go2rtc's API, so no restart is needed. They use `#hardware`, so go2rtc picks a hardware encoder and falls back to the CPU when there isn't one.

The Live playback settings stream list can be reordered by drag, since its order is the auto ladder. Auto order sorts it by bitrate, measuring native streams through a new admin-only `/go2rtc/streams/{name}/bitrate` endpoint and using the configured bitrate for transcoded ones. A pure reorder wasn't saved before because RJSF, the settings form, and `update_yaml` all ignore map key order. Sections can now mark a map with `orderedMaps`, which sends the whole map with `replace_paths` so `config_set` rewrites it in order.

Transcoded streams aren't in `go2rtc.streams`, which only lists yaml streams, so the frontend treated them as not restreamed and fell back to jsmpeg. Every restream check now goes through `isRestreamedStream`.

Auto treated any stall with no bytes in the last 2 seconds as a dead camera and handed it to the error fallback, which went straight to jsmpeg. Heavy congestion can stop delivery completely, so congested viewers skipped every lower stream. Auto now declines only when stats show the camera offline, and a stall on a live camera steps down. The stream picker also has a Try highest quality button that sends auto back to the top stream.

* fixes

frigate/api/app.py

@@ -63,6 +63,7 @@ from frigate.util.builtin import (
     flatten_config_data,
     load_labels,
     process_config_query_string,
+    split_config_key_path,
     update_yaml_file_bulk,
 )
 from frigate.util.config import (
@@ -70,6 +71,10 @@ from frigate.util.config import (
     find_config_file,
     redact_credential,
 )
+from frigate.util.live_streams import (
+    generated_transcode_streams,
+    sync_transcode_streams,
+)
 from frigate.util.object_names import get_categorized_object_names
 from frigate.util.schema import get_config_schema
 from frigate.util.services import (
@@ -815,6 +820,17 @@ def _config_set_in_memory(request: Request, body: AppConfigSetBody) -> JSONRespo
         )
 
 
+def _config_path_exists(data: Any, key_path: str) -> bool:
+    """Return whether a dotted config path is present in parsed yaml."""
+    for key in split_config_key_path(key_path):
+        if not isinstance(data, dict) or key not in data:
+            return False
+
+        data = data[key]
+
+    return True
+
+
 @router.put("/config/set", dependencies=[Depends(require_role(["admin"]))])
 def config_set(request: Request, body: AppConfigSetBody):
     config_file = find_config_file()
@@ -869,6 +885,19 @@ def config_set(request: Request, body: AppConfigSetBody):
                         status_code=400,
                     )
 
+                # delete replaced paths first so their maps are rewritten in
+                # the order sent; update_yaml would otherwise keep old order
+                if body.replace_paths:
+                    old_yaml = ruamel.yaml.YAML(typ="safe").load(old_raw_config) or {}
+                    updates = {
+                        **{
+                            path: ""
+                            for path in body.replace_paths
+                            if _config_path_exists(old_yaml, path)
+                        },
+                        **updates,
+                    }
+
                 # apply all updates in a single operation
                 update_yaml_file_bulk(config_file, updates)
 
@@ -932,9 +961,15 @@ def config_set(request: Request, body: AppConfigSetBody):
             if request.app.dispatcher is not None:
                 request.app.dispatcher.clear_runtime_state_for_yaml_keys(updates.keys())
 
+            go2rtc_synced = True
+
             if body.requires_restart == 0 or body.update_topic:
                 old_config: FrigateConfig = request.app.frigate_config
                 swap_runtime_config(request.app, config)
+                go2rtc_synced = sync_transcode_streams(
+                    generated_transcode_streams(old_config),
+                    generated_transcode_streams(config),
+                )
 
                 if body.update_topic:
                     if body.update_topic.startswith("config/cameras/"):
@@ -994,6 +1029,7 @@ def config_set(request: Request, body: AppConfigSetBody):
                             if body.requires_restart == 0
                             else "Config successfully updated, restart to apply"
                         ),
+                        "go2rtc_synced": go2rtc_synced,
                     }
                 ),
                 status_code=200,

frigate/api/camera.py

@@ -39,6 +39,11 @@ from frigate.util.builtin import clean_camera_user_pass, get_record_segment_time
 from frigate.util.camera_cleanup import cleanup_camera_db, cleanup_camera_files
 from frigate.util.config import find_config_file
 from frigate.util.image import run_ffmpeg_snapshot
+from frigate.util.live_streams import (
+    generated_transcode_streams,
+    measure_stream_bitrate,
+    sync_transcode_streams,
+)
 from frigate.util.services import (
     analyze_record_keyframes,
     ffprobe_stream,
@@ -248,6 +253,34 @@ def go2rtc_delete_stream(stream_name: str):
         )
 
 
+@router.get(
+    "/go2rtc/streams/{stream_name}/bitrate",
+    dependencies=[Depends(require_role(["admin"]))],
+)
+async def go2rtc_stream_bitrate(request: Request, stream_name: str):
+    """Measure a go2rtc stream's bitrate over a few seconds."""
+    config: FrigateConfig = request.app.frigate_config
+    known = set(config.go2rtc.model_dump().get("streams") or {}) | set(
+        generated_transcode_streams(config)
+    )
+
+    if stream_name not in known:
+        return JSONResponse(
+            content={"success": False, "message": "Unknown stream"},
+            status_code=404,
+        )
+
+    kbps = await asyncio.to_thread(measure_stream_bitrate, stream_name)
+
+    if kbps is None:
+        return JSONResponse(
+            content={"success": False, "message": "Stream sent no data"},
+            status_code=502,
+        )
+
+    return JSONResponse(content={"success": True, "kbps": round(kbps)})
+
+
 @router.get("/ffprobe", dependencies=[Depends(require_role(["admin"]))])
 def ffprobe(request: Request, paths: str = "", detailed: bool = False):
     path_param = paths
@@ -1342,6 +1375,12 @@ async def delete_camera(
     except Exception:
         logger.debug("Failed to remove go2rtc stream for %s", camera_name)
 
+    await asyncio.to_thread(
+        sync_transcode_streams,
+        generated_transcode_streams(frigate_config),
+        generated_transcode_streams(request.app.frigate_config),
+    )
+
     return JSONResponse(
         content={
             "success": True,

frigate/api/defs/request/app_body.py

@@ -10,6 +10,8 @@ class AppConfigSetBody(BaseModel):
     update_topic: str | None = None
     config_data: dict[str, Any] | None = None
     skip_save: bool = False
+    # paths rewritten whole, so a map saves in the order sent
+    replace_paths: list[str] = Field(default_factory=list)
 
 
 class GenAIProbeBody(BaseModel):

frigate/config/camera/live.py

@@ -1,8 +1,57 @@
-from pydantic import Field
+from pydantic import Field, field_validator
+
+from frigate.util.live_streams import DEFAULT_TRANSCODE_QUALITIES
 
 from ..base import FrigateBaseModel
 
-__all__ = ["CameraLiveConfig"]
+__all__ = ["CameraLiveConfig", "LiveTranscodeConfig", "LiveTranscodeQualityConfig"]
+
+
+class LiveTranscodeQualityConfig(FrigateBaseModel):
+    height: int = Field(
+        ge=144,
+        le=2160,
+        title="Height",
+        description="Output height in pixels; width follows the source aspect ratio.",
+    )
+    bitrate: int = Field(
+        ge=64,
+        title="Bitrate",
+        description="Target and maximum video bitrate in kbps.",
+    )
+
+
+class LiveTranscodeConfig(FrigateBaseModel):
+    enabled: bool = Field(
+        default=False,
+        title="Enable transcoded streams",
+        description="Add lower-quality live streams that go2rtc transcodes in real time while someone is watching.",
+    )
+    source: str | None = Field(
+        default=None,
+        title="Source stream",
+        description="go2rtc stream to transcode. Defaults to the first live stream.",
+    )
+    qualities: list[LiveTranscodeQualityConfig] = Field(
+        default_factory=lambda: [
+            LiveTranscodeQualityConfig(**quality)
+            for quality in DEFAULT_TRANSCODE_QUALITIES
+        ],
+        title="Qualities",
+        description="One transcoded stream is added per quality.",
+    )
+
+    @field_validator("qualities")
+    @classmethod
+    def validate_unique_heights(
+        cls, qualities: list[LiveTranscodeQualityConfig]
+    ) -> list[LiveTranscodeQualityConfig]:
+        heights = [quality.height for quality in qualities]
+
+        if len(heights) != len(set(heights)):
+            raise ValueError("Transcoded stream heights must be unique.")
+
+        return qualities
 
 
 class CameraLiveConfig(FrigateBaseModel):
@@ -11,6 +60,11 @@ class CameraLiveConfig(FrigateBaseModel):
         title="Live stream names",
         description="Mapping of configured stream names to restream/go2rtc names used for live playback.",
     )
+    transcode: LiveTranscodeConfig = Field(
+        default_factory=LiveTranscodeConfig,
+        title="Transcoded streams",
+        description="Lower-quality live streams transcoded on demand by go2rtc.",
+    )
     height: int = Field(
         default=720,
         title="Live height",

frigate/config/config.py

@@ -35,6 +35,12 @@ from frigate.util.config import (
     migrate_frigate_config,
 )
 from frigate.util.image import create_mask
+from frigate.util.live_streams import (
+    default_transcode_source,
+    is_transcode_stream_name,
+    transcode_stream_name,
+    transcode_streams,
+)
 from frigate.util.runtime_deps import sha256_of
 from frigate.util.services import auto_detect_hwaccel
 
@@ -283,11 +289,78 @@ def verify_config_roles(camera_config: CameraConfig) -> None:
         )
 
 
+def apply_live_transcode_streams(
+    frigate_config: FrigateConfig, camera_config: CameraConfig
+) -> None:
+    """Fold a camera's transcoded streams into its live stream list.
+
+    Enabled qualities missing from live.streams are appended, and entries the
+    user placed keep their position. Transcoded names that are no longer
+    generated are dropped unless they name a real go2rtc stream.
+    """
+    live = camera_config.live
+    transcode = live.transcode
+    go2rtc_streams = frigate_config.go2rtc.model_dump().get("streams") or {}
+    generated: dict[str, str] = {}
+
+    if transcode.enabled:
+        if transcode.source is None:
+            transcode.source = default_transcode_source(
+                camera_config.name, live.streams
+            )
+
+        if transcode.source not in go2rtc_streams:
+            raise ValueError(
+                f"Camera {camera_config.name} has transcoded streams enabled, but its source {transcode.source} is not a go2rtc stream."
+            )
+
+        generated = transcode_streams(
+            camera_config.name,
+            transcode.source,
+            [quality.model_dump() for quality in transcode.qualities],
+        )
+
+        for name in generated:
+            if name in go2rtc_streams:
+                raise ValueError(
+                    f"Camera {camera_config.name} generates transcoded stream {name}, which collides with a go2rtc stream of the same name."
+                )
+
+    streams = {
+        label: name
+        for label, name in live.streams.items()
+        if name in generated
+        or name in go2rtc_streams
+        or not is_transcode_stream_name(camera_config.name, name)
+    }
+    placed = set(streams.values())
+
+    for quality in transcode.qualities if transcode.enabled else []:
+        name = transcode_stream_name(camera_config.name, quality.height)
+
+        if name in placed:
+            continue
+
+        label = f"{quality.height}p"
+
+        if label in streams:
+            raise ValueError(
+                f"Camera {camera_config.name} already has a live stream named {label}; rename it or place the transcoded stream under another name."
+            )
+
+        streams[label] = name
+
+    live.streams = streams
+
+
 def verify_valid_live_stream_names(
     frigate_config: FrigateConfig, camera_config: CameraConfig
 ) -> ValueError | None:
     """Verify that a restream exists to use for live view."""
     for _, stream_name in camera_config.live.streams.items():
+        if is_transcode_stream_name(camera_config.name, stream_name):
+            continue
+
         if (
             stream_name
             not in frigate_config.go2rtc.model_dump().get("streams", {}).keys()
@@ -1055,6 +1128,8 @@ class FrigateConfig(FrigateBaseModel):
                 "face_recognition": ["enabled", "min_area"],
                 "lpr": ["enabled", "expire_time", "min_area", "enhancement"],
                 "audio_transcription": ["enabled", "live_enabled"],
+                # transcode is camera-level only
+                "live": ["streams", "height", "quality"],
             }
 
             for section in allowed_fields_map:
@@ -1287,6 +1362,8 @@ class FrigateConfig(FrigateBaseModel):
             if not camera_config.live.streams:
                 camera_config.live.streams = {name: name}
 
+            apply_live_transcode_streams(self, camera_config)
+
             # generate the ffmpeg commands
             camera_config.create_ffmpeg_cmds()
             self.cameras[name] = camera_config

frigate/test/http_api/test_http_config_set_live.py

@@ -0,0 +1,173 @@
+"""Tests for config_set live stream ordering and go2rtc transcode sync."""
+
+import os
+import tempfile
+from unittest.mock import MagicMock, Mock, patch
+
+import ruamel.yaml
+from fastapi import Request
+
+from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user
+from frigate.api.fastapi_app import create_fastapi_app
+from frigate.config import FrigateConfig
+from frigate.config.camera.updater import CameraConfigUpdatePublisher
+from frigate.models import Event, Recordings, ReviewSegment
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+from frigate.util.live_streams import transcode_stream_source
+
+STREAMS_PATH = "cameras.front_door.live.streams"
+
+
+class TestConfigSetLiveStreams(BaseTestHttp):
+    def setUp(self):
+        super().setUp(models=[Event, Recordings, ReviewSegment])
+        self.minimal_config["go2rtc"] = {
+            "streams": {
+                "front_main": ["rtsp://10.0.0.1:554/main"],
+                "front_sub": ["rtsp://10.0.0.1:554/sub"],
+            }
+        }
+        self.minimal_config["cameras"]["front_door"]["live"] = {
+            "streams": {"Main": "front_main", "Sub": "front_sub"}
+        }
+
+    def _write_config_file(self) -> str:
+        yaml = ruamel.yaml.YAML()
+        f = tempfile.NamedTemporaryFile(mode="w", suffix=".yml", delete=False)
+        yaml.dump(self.minimal_config, f)
+        f.close()
+        self.addCleanup(os.unlink, f.name)
+        return f.name
+
+    def _read_config_file(self, path: str) -> dict:
+        with open(path) as f:
+            return ruamel.yaml.YAML(typ="safe").load(f)
+
+    def _app(self):
+        publisher = Mock(spec=CameraConfigUpdatePublisher)
+        publisher.publisher = MagicMock()
+        app = create_fastapi_app(
+            FrigateConfig(**self.minimal_config),
+            self.db,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            publisher,
+            None,
+            enforce_default_admin=False,
+        )
+
+        async def mock_get_current_user(request: Request):
+            return {
+                "username": request.headers.get("remote-user"),
+                "role": request.headers.get("remote-role"),
+            }
+
+        async def mock_get_allowed_cameras_for_filter(request: Request):
+            return ["front_door"]
+
+        app.dependency_overrides[get_current_user] = mock_get_current_user
+        app.dependency_overrides[get_allowed_cameras_for_filter] = (
+            mock_get_allowed_cameras_for_filter
+        )
+        return app
+
+    def _save(self, app, live: dict, replace_paths: list[str] | None = None):
+        body = {
+            "config_data": {"cameras": {"front_door": {"live": live}}},
+            "requires_restart": 0,
+            "update_topic": "config/cameras/front_door/live",
+        }
+
+        if replace_paths is not None:
+            body["replace_paths"] = replace_paths
+
+        with AuthTestClient(app) as client:
+            return client.put("/config/set", json=body)
+
+    @patch("frigate.api.app.find_config_file")
+    def test_replace_paths_saves_map_in_sent_order(self, mock_find_config):
+        path = self._write_config_file()
+        mock_find_config.return_value = path
+
+        resp = self._save(
+            self._app(),
+            {"streams": {"Sub": "front_sub", "Main": "front_main"}},
+            [STREAMS_PATH],
+        )
+
+        self.assertEqual(resp.status_code, 200)
+        streams = self._read_config_file(path)["cameras"]["front_door"]["live"][
+            "streams"
+        ]
+        self.assertEqual(list(streams), ["Sub", "Main"])
+
+    @patch("frigate.api.app.find_config_file")
+    def test_without_replace_paths_order_is_kept(self, mock_find_config):
+        path = self._write_config_file()
+        mock_find_config.return_value = path
+
+        self._save(self._app(), {"streams": {"Sub": "front_sub", "Main": "front_main"}})
+
+        streams = self._read_config_file(path)["cameras"]["front_door"]["live"][
+            "streams"
+        ]
+        self.assertEqual(list(streams), ["Main", "Sub"])
+
+    @patch("frigate.api.app.find_config_file")
+    def test_replace_path_missing_from_yaml_is_skipped(self, mock_find_config):
+        del self.minimal_config["cameras"]["front_door"]["live"]
+        path = self._write_config_file()
+        mock_find_config.return_value = path
+
+        resp = self._save(
+            self._app(), {"streams": {"Main": "front_main"}}, [STREAMS_PATH]
+        )
+
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(
+            self._read_config_file(path)["cameras"]["front_door"]["live"]["streams"],
+            {"Main": "front_main"},
+        )
+
+    @patch("frigate.api.app.sync_transcode_streams", return_value=True)
+    @patch("frigate.api.app.find_config_file")
+    def test_enabling_transcode_syncs_go2rtc(self, mock_find_config, mock_sync):
+        path = self._write_config_file()
+        mock_find_config.return_value = path
+
+        resp = self._save(
+            self._app(),
+            {
+                "transcode": {
+                    "enabled": True,
+                    "qualities": [{"height": 480, "bitrate": 500}],
+                }
+            },
+        )
+
+        self.assertEqual(resp.status_code, 200)
+        self.assertTrue(resp.json()["go2rtc_synced"])
+        mock_sync.assert_called_once_with(
+            {},
+            {
+                "front_door_transcode_480p": transcode_stream_source(
+                    "front_main", 480, 500
+                )
+            },
+        )
+
+    @patch("frigate.api.app.sync_transcode_streams", return_value=False)
+    @patch("frigate.api.app.find_config_file")
+    def test_sync_failure_is_reported(self, mock_find_config, mock_sync):
+        path = self._write_config_file()
+        mock_find_config.return_value = path
+
+        resp = self._save(self._app(), {"transcode": {"enabled": True}})
+
+        self.assertEqual(resp.status_code, 200)
+        self.assertTrue(resp.json()["success"])
+        self.assertFalse(resp.json()["go2rtc_synced"])

frigate/test/http_api/test_http_go2rtc_bitrate.py

@@ -0,0 +1,61 @@
+"""Tests for the go2rtc stream bitrate endpoint."""
+
+from unittest.mock import patch
+
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+
+
+class TestHttpGo2rtcBitrate(BaseTestHttp):
+    def setUp(self):
+        super().setUp([])
+        self.minimal_config["go2rtc"] = {
+            "streams": {"front_main": ["rtsp://10.0.0.1:554/main"]}
+        }
+        self.minimal_config["cameras"]["front_door"]["live"] = {
+            "streams": {"Main": "front_main"},
+            "transcode": {
+                "enabled": True,
+                "qualities": [{"height": 480, "bitrate": 500}],
+            },
+        }
+        self.app = self.create_app()
+
+    @patch("frigate.api.camera.measure_stream_bitrate", return_value=812.4)
+    def test_returns_rounded_kbps(self, mock_measure):
+        with AuthTestClient(self.app) as client:
+            resp = client.get("/go2rtc/streams/front_main/bitrate")
+
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.json()["kbps"], 812)
+        mock_measure.assert_called_once_with("front_main")
+
+    @patch("frigate.api.camera.measure_stream_bitrate", return_value=480.0)
+    def test_generated_streams_are_known(self, _):
+        with AuthTestClient(self.app) as client:
+            resp = client.get("/go2rtc/streams/front_door_transcode_480p/bitrate")
+
+        self.assertEqual(resp.status_code, 200)
+
+    @patch("frigate.api.camera.measure_stream_bitrate")
+    def test_unknown_stream_is_404(self, mock_measure):
+        with AuthTestClient(self.app) as client:
+            resp = client.get("/go2rtc/streams/rtsp_somewhere/bitrate")
+
+        self.assertEqual(resp.status_code, 404)
+        mock_measure.assert_not_called()
+
+    @patch("frigate.api.camera.measure_stream_bitrate", return_value=None)
+    def test_no_data_is_502(self, _):
+        with AuthTestClient(self.app) as client:
+            resp = client.get("/go2rtc/streams/front_main/bitrate")
+
+        self.assertEqual(resp.status_code, 502)
+
+    def test_viewer_is_forbidden(self):
+        with AuthTestClient(self.app) as client:
+            resp = client.get(
+                "/go2rtc/streams/front_main/bitrate",
+                headers={"remote-user": "viewer", "remote-role": "viewer"},
+            )
+
+        self.assertEqual(resp.status_code, 403)

frigate/test/test_live_streams.py

@@ -0,0 +1,410 @@
+"""Tests for Frigate-managed go2rtc live streams."""
+
+import unittest
+from unittest.mock import MagicMock, call, patch
+
+import requests
+from pydantic import ValidationError
+
+from frigate.config import FrigateConfig
+from frigate.util.live_streams import (
+    DEFAULT_TRANSCODE_QUALITIES,
+    default_transcode_source,
+    generated_transcode_streams,
+    is_transcode_stream_name,
+    measure_stream_bitrate,
+    raw_transcode_streams,
+    sync_transcode_streams,
+    transcode_stream_name,
+    transcode_stream_source,
+)
+
+
+class TestTranscodeNaming(unittest.TestCase):
+    def test_stream_name(self):
+        self.assertEqual(transcode_stream_name("front", 720), "front_transcode_720p")
+
+    def test_is_transcode_stream_name(self):
+        self.assertTrue(is_transcode_stream_name("front", "front_transcode_480p"))
+        self.assertFalse(is_transcode_stream_name("front", "front_sub"))
+        self.assertFalse(is_transcode_stream_name("front", "back_transcode_480p"))
+        self.assertFalse(is_transcode_stream_name("front", "front_transcode_480"))
+
+    def test_source_gives_each_ffmpeg_token_its_own_raw_param(self):
+        self.assertEqual(
+            transcode_stream_source("front", 480, 500),
+            "ffmpeg:front#video=h264#height=480#hardware#audio=copy"
+            "#raw=-b:v#raw=500k#raw=-maxrate#raw=500k#raw=-bufsize#raw=1000k",
+        )
+
+    def test_default_source_skips_transcoded_streams(self):
+        streams = {"Low": "front_transcode_360p", "Main": "front_main"}
+        self.assertEqual(default_transcode_source("front", streams), "front_main")
+        self.assertIsNone(default_transcode_source("front", {}))
+
+
+class TestRawTranscodeStreams(unittest.TestCase):
+    def test_defaults_to_camera_stream_and_default_qualities(self):
+        config = {"cameras": {"front": {"live": {"transcode": {"enabled": True}}}}}
+        streams = raw_transcode_streams(config)
+
+        self.assertEqual(
+            list(streams),
+            [
+                transcode_stream_name("front", q["height"])
+                for q in DEFAULT_TRANSCODE_QUALITIES
+            ],
+        )
+        self.assertTrue(streams["front_transcode_720p"].startswith("ffmpeg:front#"))
+
+    def test_uses_first_live_stream_and_explicit_qualities(self):
+        config = {
+            "cameras": {
+                "front": {
+                    "live": {
+                        "streams": {"Main": "front_main", "Sub": "front_sub"},
+                        "transcode": {
+                            "enabled": True,
+                            "qualities": [{"height": 540, "bitrate": 800}],
+                        },
+                    }
+                }
+            }
+        }
+
+        self.assertEqual(
+            raw_transcode_streams(config),
+            {"front_transcode_540p": transcode_stream_source("front_main", 540, 800)},
+        )
+
+    def test_explicit_source_wins(self):
+        config = {
+            "cameras": {
+                "front": {
+                    "live": {
+                        "streams": {"Main": "front_main"},
+                        "transcode": {
+                            "enabled": True,
+                            "source": "front_sub",
+                            "qualities": [{"height": 360, "bitrate": 250}],
+                        },
+                    }
+                }
+            }
+        }
+
+        self.assertEqual(
+            raw_transcode_streams(config)["front_transcode_360p"],
+            transcode_stream_source("front_sub", 360, 250),
+        )
+
+    def test_malformed_qualities_are_skipped(self):
+        config = {
+            "cameras": {
+                "front": {
+                    "live": {
+                        "transcode": {
+                            "enabled": True,
+                            "qualities": [
+                                {"height": 720},
+                                "480",
+                                {"height": 360, "bitrate": 250},
+                            ],
+                        }
+                    }
+                },
+                "back": {
+                    "live": {"transcode": {"enabled": True, "qualities": {"a": 1}}}
+                },
+            }
+        }
+
+        self.assertEqual(list(raw_transcode_streams(config)), ["front_transcode_360p"])
+
+    def test_disabled_or_missing_generates_nothing(self):
+        self.assertEqual(raw_transcode_streams({}), {})
+        self.assertEqual(
+            raw_transcode_streams({"cameras": {"front": {"live": {}}}}), {}
+        )
+        self.assertEqual(
+            raw_transcode_streams(
+                {"cameras": {"front": {"live": {"transcode": {"enabled": False}}}}}
+            ),
+            {},
+        )
+
+
+def camera_config(live: dict | None = None, go2rtc: dict | None = None) -> dict:
+    config = {
+        "mqtt": {"host": "mqtt"},
+        "go2rtc": {
+            "streams": go2rtc
+            or {
+                "front": ["rtsp://10.0.0.1:554/main"],
+                "front_sub": ["rtsp://10.0.0.1:554/sub"],
+            }
+        },
+        "cameras": {
+            "front": {
+                "ffmpeg": {
+                    "inputs": [
+                        {"path": "rtsp://10.0.0.1:554/main", "roles": ["detect"]}
+                    ]
+                },
+                "detect": {"height": 1080, "width": 1920, "fps": 5},
+            }
+        },
+    }
+
+    if live is not None:
+        config["cameras"]["front"]["live"] = live
+
+    return config
+
+
+class TestLiveTranscodeConfig(unittest.TestCase):
+    def streams(self, live: dict, go2rtc: dict | None = None) -> dict[str, str]:
+        config = FrigateConfig(**camera_config(live, go2rtc))
+        return config.cameras["front"].live.streams
+
+    def test_global_transcode_is_ignored(self):
+        config = camera_config()
+        config["live"] = {"transcode": {"enabled": True}}
+
+        streams = FrigateConfig(**config).cameras["front"].live.streams
+        self.assertEqual(streams, {"front": "front"})
+
+    def test_disabled_leaves_streams_alone(self):
+        self.assertEqual(self.streams({}), {"front": "front"})
+
+    def test_enabled_appends_default_qualities(self):
+        streams = self.streams({"transcode": {"enabled": True}})
+
+        self.assertEqual(
+            list(streams.items()),
+            [
+                ("front", "front"),
+                ("720p", "front_transcode_720p"),
+                ("480p", "front_transcode_480p"),
+                ("360p", "front_transcode_360p"),
+            ],
+        )
+
+    def test_source_resolves_to_first_live_stream(self):
+        config = FrigateConfig(
+            **camera_config(
+                {
+                    "streams": {"Sub": "front_sub", "Main": "front"},
+                    "transcode": {"enabled": True},
+                }
+            )
+        )
+
+        self.assertEqual(config.cameras["front"].live.transcode.source, "front_sub")
+
+    def test_startup_and_validation_agree_on_inherited_source(self):
+        config = camera_config({"transcode": {"enabled": True}})
+        config["live"] = {"streams": {"Sub": "front_sub"}}
+
+        self.assertEqual(
+            raw_transcode_streams(config),
+            generated_transcode_streams(FrigateConfig(**config)),
+        )
+        self.assertEqual(
+            raw_transcode_streams(config)["front_transcode_360p"],
+            transcode_stream_source("front_sub", 360, 250),
+        )
+
+    def test_placed_entries_keep_position_and_label(self):
+        streams = self.streams(
+            {
+                "streams": {
+                    "Main": "front",
+                    "Low": "front_transcode_480p",
+                    "Sub": "front_sub",
+                },
+                "transcode": {"enabled": True},
+            }
+        )
+
+        self.assertEqual(
+            list(streams.items()),
+            [
+                ("Main", "front"),
+                ("Low", "front_transcode_480p"),
+                ("Sub", "front_sub"),
+                ("720p", "front_transcode_720p"),
+                ("360p", "front_transcode_360p"),
+            ],
+        )
+
+    def test_disabling_drops_transcoded_entries(self):
+        streams = self.streams(
+            {
+                "streams": {"Main": "front", "720p": "front_transcode_720p"},
+                "transcode": {"enabled": False},
+            }
+        )
+
+        self.assertEqual(streams, {"Main": "front"})
+
+    def test_changed_height_drops_the_old_entry(self):
+        streams = self.streams(
+            {
+                "streams": {"Main": "front", "720p": "front_transcode_720p"},
+                "transcode": {
+                    "enabled": True,
+                    "qualities": [{"height": 540, "bitrate": 800}],
+                },
+            }
+        )
+
+        self.assertEqual(streams, {"Main": "front", "540p": "front_transcode_540p"})
+
+    def test_real_go2rtc_stream_with_transcode_name_is_kept(self):
+        streams = self.streams(
+            {"streams": {"Main": "front", "Odd": "front_transcode_720p"}},
+            go2rtc={
+                "front": ["rtsp://10.0.0.1:554/main"],
+                "front_transcode_720p": ["rtsp://10.0.0.1:554/odd"],
+            },
+        )
+
+        self.assertEqual(streams, {"Main": "front", "Odd": "front_transcode_720p"})
+
+    def test_generated_name_colliding_with_go2rtc_stream_fails(self):
+        with self.assertRaises(ValidationError):
+            self.streams(
+                {"transcode": {"enabled": True}},
+                go2rtc={
+                    "front": ["rtsp://10.0.0.1:554/main"],
+                    "front_transcode_720p": ["rtsp://10.0.0.1:554/odd"],
+                },
+            )
+
+    def test_source_must_be_a_go2rtc_stream(self):
+        with self.assertRaises(ValidationError):
+            self.streams({"transcode": {"enabled": True, "source": "missing"}})
+
+    def test_duplicate_heights_fail(self):
+        with self.assertRaises(ValidationError):
+            self.streams(
+                {
+                    "transcode": {
+                        "enabled": True,
+                        "qualities": [
+                            {"height": 480, "bitrate": 500},
+                            {"height": 480, "bitrate": 600},
+                        ],
+                    }
+                }
+            )
+
+    def test_label_collision_fails(self):
+        with self.assertRaises(ValidationError):
+            self.streams(
+                {
+                    "streams": {"Main": "front", "720p": "front_sub"},
+                    "transcode": {"enabled": True},
+                }
+            )
+
+
+class TestGeneratedTranscodeStreams(unittest.TestCase):
+    def test_collects_enabled_cameras(self):
+        config = FrigateConfig(
+            **camera_config(
+                {
+                    "transcode": {
+                        "enabled": True,
+                        "qualities": [{"height": 480, "bitrate": 500}],
+                    }
+                }
+            )
+        )
+
+        self.assertEqual(
+            generated_transcode_streams(config),
+            {"front_transcode_480p": transcode_stream_source("front", 480, 500)},
+        )
+
+    def test_disabled_is_empty(self):
+        self.assertEqual(
+            generated_transcode_streams(FrigateConfig(**camera_config())), {}
+        )
+
+
+class TestSyncTranscodeStreams(unittest.TestCase):
+    @patch("frigate.util.live_streams.requests.request")
+    def test_puts_added_and_changed_and_deletes_removed(self, mock_request):
+        mock_request.return_value = MagicMock(ok=True)
+
+        ok = sync_transcode_streams(
+            {"a": "src-a", "b": "src-b", "c": "src-c"},
+            {"a": "src-a", "b": "src-b2", "d": "src-d"},
+        )
+
+        self.assertTrue(ok)
+        url = "http://127.0.0.1:1984/api/streams"
+        self.assertCountEqual(
+            mock_request.call_args_list,
+            [
+                call("put", url, params={"name": "b", "src": "src-b2"}, timeout=5),
+                call("put", url, params={"name": "d", "src": "src-d"}, timeout=5),
+                call("delete", url, params={"src": "c"}, timeout=5),
+            ],
+        )
+
+    @patch("frigate.util.live_streams.requests.request")
+    def test_unchanged_makes_no_calls(self, mock_request):
+        self.assertTrue(sync_transcode_streams({"a": "x"}, {"a": "x"}))
+        mock_request.assert_not_called()
+
+    @patch("frigate.util.live_streams.requests.request")
+    def test_failure_returns_false_and_keeps_going(self, mock_request):
+        mock_request.side_effect = [
+            requests.ConnectionError("down"),
+            MagicMock(ok=True),
+        ]
+
+        self.assertFalse(sync_transcode_streams({}, {"a": "x", "b": "y"}))
+        self.assertEqual(mock_request.call_count, 2)
+
+
+class TestMeasureStreamBitrate(unittest.TestCase):
+    @patch("frigate.util.live_streams.time.monotonic")
+    @patch("frigate.util.live_streams.requests.get")
+    def test_skips_warmup_and_averages(self, mock_get, mock_monotonic):
+        response = MagicMock(ok=True)
+        response.iter_content.return_value = [b"x" * 1000] * 6
+        mock_get.return_value.__enter__.return_value = response
+        # first byte, warmup, window start, then counted chunks up to 6 s
+        mock_monotonic.side_effect = [0.0, 0.5, 1.0, 2.0, 4.0, 7.0]
+
+        kbps = measure_stream_bitrate("front", duration=6.0, warmup=1.0)
+
+        self.assertAlmostEqual(kbps, 3000 * 8 / 1000 / 6.0)
+        mock_get.assert_called_once_with(
+            "http://127.0.0.1:1984/api/stream.mp4",
+            params={"src": "front"},
+            stream=True,
+            timeout=5,
+        )
+
+    @patch("frigate.util.live_streams.requests.get")
+    def test_no_data_returns_none(self, mock_get):
+        response = MagicMock(ok=True)
+        response.iter_content.return_value = []
+        mock_get.return_value.__enter__.return_value = response
+
+        self.assertIsNone(measure_stream_bitrate("front"))
+
+    @patch("frigate.util.live_streams.requests.get")
+    def test_request_error_returns_none(self, mock_get):
+        mock_get.side_effect = requests.ReadTimeout("slow")
+
+        self.assertIsNone(measure_stream_bitrate("front"))
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/util/live_streams.py

@@ -0,0 +1,224 @@
+"""Frigate-managed go2rtc live streams: transcoded rungs and bitrate probes."""
+
+import logging
+import re
+import time
+from typing import TYPE_CHECKING, Any
+
+import requests
+
+if TYPE_CHECKING:
+    from frigate.config import FrigateConfig
+
+logger = logging.getLogger(__name__)
+
+GO2RTC_API = "http://127.0.0.1:1984/api"
+
+DEFAULT_TRANSCODE_QUALITIES: list[dict[str, int]] = [
+    {"height": 720, "bitrate": 1200},
+    {"height": 480, "bitrate": 500},
+    {"height": 360, "bitrate": 250},
+]
+
+
+def transcode_stream_name(camera: str, height: int) -> str:
+    """Return the go2rtc name of a camera's transcoded stream at one height."""
+    return f"{camera}_transcode_{height}p"
+
+
+def is_transcode_stream_name(camera: str, name: str) -> bool:
+    """Return whether a stream name follows this camera's transcoded naming."""
+    return re.fullmatch(rf"{re.escape(camera)}_transcode_\d+p", name) is not None
+
+
+def transcode_stream_source(source: str, height: int, bitrate: int) -> str:
+    """Build the go2rtc ffmpeg source that transcodes a stream to H.264.
+
+    go2rtc rejects API-registered sources containing spaces, so each ffmpeg
+    token gets its own #raw= parameter. #audio=copy keeps audio, which any
+    #video= option would otherwise drop.
+    """
+    tokens = (
+        "-b:v",
+        f"{bitrate}k",
+        "-maxrate",
+        f"{bitrate}k",
+        "-bufsize",
+        f"{2 * bitrate}k",
+    )
+    raw = "".join(f"#raw={token}" for token in tokens)
+    return f"ffmpeg:{source}#video=h264#height={height}#hardware#audio=copy{raw}"
+
+
+def transcode_streams(
+    camera: str, source: str, qualities: list[dict[str, int]]
+) -> dict[str, str]:
+    """Map each quality's go2rtc stream name to its source."""
+    return {
+        transcode_stream_name(camera, quality["height"]): transcode_stream_source(
+            source, quality["height"], quality["bitrate"]
+        )
+        for quality in qualities
+    }
+
+
+def default_transcode_source(camera: str, streams: dict[str, str]) -> str | None:
+    """Return the first live stream that is not itself transcoded."""
+    return next(
+        (
+            name
+            for name in streams.values()
+            if not is_transcode_stream_name(camera, name)
+        ),
+        None,
+    )
+
+
+def raw_transcode_streams(config: dict[str, Any]) -> dict[str, str]:
+    """Build transcoded go2rtc streams for every enabled camera in a raw config.
+
+    go2rtc's config is written before FrigateConfig validation runs, so this
+    resolves defaults the same way the validator does.
+    """
+    streams: dict[str, str] = {}
+    global_streams = (config.get("live") or {}).get("streams") or {}
+
+    for camera, camera_config in (config.get("cameras") or {}).items():
+        live = (camera_config or {}).get("live") or {}
+        transcode = live.get("transcode") or {}
+
+        if not transcode.get("enabled"):
+            continue
+
+        # cameras inherit global live streams after their own, like deep_merge
+        camera_streams = live.get("streams") or {}
+        live_streams = camera_streams | {
+            label: name
+            for label, name in global_streams.items()
+            if label not in camera_streams
+        }
+        source = transcode.get("source") or default_transcode_source(
+            camera, live_streams or {camera: camera}
+        )
+
+        if source is None:
+            continue
+
+        qualities = transcode.get("qualities")
+
+        if qualities is None:
+            qualities = DEFAULT_TRANSCODE_QUALITIES
+
+        # malformed entries fail Frigate's validation; skipping them here keeps
+        # go2rtc from crash-looping alongside it
+        valid = [
+            quality
+            for quality in (qualities if isinstance(qualities, list) else [])
+            if isinstance(quality, dict)
+            and "height" in quality
+            and "bitrate" in quality
+        ]
+        streams.update(transcode_streams(camera, source, valid))
+
+    return streams
+
+
+def generated_transcode_streams(config: "FrigateConfig") -> dict[str, str]:
+    """Build transcoded go2rtc streams for every enabled camera."""
+    streams: dict[str, str] = {}
+
+    for camera in config.cameras.values():
+        transcode = camera.live.transcode
+
+        if transcode.enabled and transcode.source:
+            streams.update(
+                transcode_streams(
+                    camera.name,
+                    transcode.source,
+                    [quality.model_dump() for quality in transcode.qualities],
+                )
+            )
+
+    return streams
+
+
+def _go2rtc_streams_call(method: str, name: str, params: dict[str, str]) -> bool:
+    try:
+        response = requests.request(
+            method, f"{GO2RTC_API}/streams", params=params, timeout=5
+        )
+    except requests.RequestException as e:
+        logger.error("Failed to sync go2rtc stream %s: %s", name, e)
+        return False
+
+    if not response.ok:
+        logger.error("Failed to sync go2rtc stream %s: %s", name, response.text)
+        return False
+
+    return True
+
+
+def sync_transcode_streams(old: dict[str, str], new: dict[str, str]) -> bool:
+    """Register added or changed streams with go2rtc and remove dropped ones.
+
+    Returns False when any go2rtc call failed.
+    """
+    ok = True
+
+    for name, source in new.items():
+        if old.get(name) != source:
+            ok = _go2rtc_streams_call("put", name, {"name": name, "src": source}) and ok
+
+    for name in old.keys() - new.keys():
+        ok = _go2rtc_streams_call("delete", name, {"src": name}) and ok
+
+    return ok
+
+
+def measure_stream_bitrate(
+    name: str, duration: float = 6.0, warmup: float = 1.0
+) -> float | None:
+    """Measure a go2rtc stream's average kbps from its local MP4 output.
+
+    The first warmup seconds after the first byte are skipped so the startup
+    keyframe burst does not inflate the result. Returns None when the stream
+    sends nothing.
+    """
+    first = start = now = None
+    counted = 0
+
+    try:
+        with requests.get(
+            f"{GO2RTC_API}/stream.mp4",
+            params={"src": name},
+            stream=True,
+            timeout=5,
+        ) as response:
+            if not response.ok:
+                return None
+
+            for chunk in response.iter_content(chunk_size=16384):
+                now = time.monotonic()
+
+                if first is None:
+                    first = now
+
+                if now - first < warmup:
+                    continue
+
+                if start is None:
+                    start = now
+                    continue
+
+                counted += len(chunk)
+
+                if now - start >= duration:
+                    break
+    except requests.RequestException as e:
+        logger.warning("Failed to measure go2rtc stream %s: %s", name, e)
+        return None
+
+    if start is None or now is None or now <= start or counted == 0:
+        return None
+
+    return counted * 8 / 1000 / (now - start)