Add preload endpoint (#203)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

Author: edenhaus

go2rtc_client/models.py

@@ -81,3 +81,10 @@ class WebRTCSdpAnswer(WebRTCSdp):
     """WebRTC SDP answer model."""
 
     type: Literal["answer"] = field(default="answer", init=False)
+
+
+@dataclass
+class Preload(DataClassORJSONMixin):
+    """Preload model."""
+
+    query: str

go2rtc_client/rest.py

@@ -14,15 +14,15 @@
 from yarl import URL
 
 from .exceptions import Go2RtcVersionError, handle_error
-from .models import ApplicationInfo, Stream, WebRTCSdpAnswer, WebRTCSdpOffer
+from .models import ApplicationInfo, Preload, Stream, WebRTCSdpAnswer, WebRTCSdpOffer
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
 
 _LOGGER = logging.getLogger(__name__)
 
 _API_PREFIX = "/api"
-_MIN_VERSION_SUPPORTED: Final = AwesomeVersion("1.9.12")
+_MIN_VERSION_SUPPORTED: Final = AwesomeVersion("1.9.13")
 _MIN_VERSION_UNSUPPORTED: Final = AwesomeVersion("2.0.0")
 
 
@@ -42,7 +42,7 @@ def __init__(self, websession: ClientSession, server_url: str) -> None:
 
     async def request(
         self,
-        method: Literal["GET", "PUT", "POST"],
+        method: Literal["GET", "PUT", "POST", "DELETE"],
         path: str,
         *,
         params: Mapping[str, Any] | None = None,
@@ -156,13 +156,61 @@ async def list(self) -> set[str]:
         return self._DECODER.decode(await resp.json())
 
 
+class _PreloadClient:
+    PATH: Final = f"{_API_PREFIX}/preload"
+    _DECODER = BasicDecoder(dict[str, Preload])
+
+    def __init__(self, client: _BaseClient) -> None:
+        """Initialize Client."""
+        self._client = client
+
+    @handle_error
+    async def enable(
+        self,
+        source: str,
+        *,
+        video_codec_filter: list[str] | None = None,
+        audio_codec_filter: list[str] | None = None,
+        microphone_codec_filter: list[str] | None = None,
+    ) -> None:
+        """Enable preload for a stream."""
+        params = {"src": source}
+        if video_codec_filter:
+            params["video_codec_filter"] = ",".join(video_codec_filter)
+        if audio_codec_filter:
+            params["audio_codec_filter"] = ",".join(audio_codec_filter)
+        if microphone_codec_filter:
+            params["microphone_codec_filter"] = ",".join(microphone_codec_filter)
+        await self._client.request(
+            "PUT",
+            self.PATH,
+            params=params,
+        )
+
+    @handle_error
+    async def disable(self, source: str) -> None:
+        """Disable preload for a stream."""
+        await self._client.request(
+            "DELETE",
+            self.PATH,
+            params={"src": source},
+        )
+
+    @handle_error
+    async def list(self) -> dict[str, Preload]:
+        """List all preloaded streams."""
+        resp = await self._client.request("GET", self.PATH)
+        return self._DECODER.decode(await resp.json())
+
+
 class Go2RtcRestClient:
     """Rest client for go2rtc server."""
 
     def __init__(self, websession: ClientSession, server_url: str) -> None:
         """Initialize Client."""
         self._client = _BaseClient(websession, server_url)
         self.application: Final = _ApplicationClient(self._client)
+        self.preload: Final = _PreloadClient(self._client)
         self.schemes: Final = _SchemesClient(self._client)
         self.streams: Final = _StreamClient(self._client)
         self.webrtc: Final = _WebRTCClient(self._client)

pyproject.toml

@@ -140,6 +140,7 @@ disable = [
   "too-few-public-methods",
   "too-many-instance-attributes",
   "too-many-arguments",
+  "too-many-positional-arguments",
   "too-many-public-methods",
   "wrong-import-order",
 ]

tests/__snapshots__/test_rest.ambr

@@ -9,6 +9,17 @@
     'version': '1.9.4',
   })
 # ---
+# name: test_preload_list[no stream preloaded]
+  dict({
+  })
+# ---
+# name: test_preload_list[one stream preloaded]
+  dict({
+    'camera.haus_eingang_fluent_2': dict({
+      'query': 'video&audio',
+    }),
+  })
+# ---
 # name: test_schemes
   set({
     'exec',

tests/fixtures/preload_list_one.json

@@ -0,0 +1,34 @@
+{
+  "camera.haus_eingang_fluent_2": {
+    "consumer": {
+      "id": 1,
+      "format_name": "preload",
+      "medias": ["video, sendonly, ANY", "audio, sendonly, ANY"],
+      "senders": [
+        {
+          "id": 4,
+          "codec": {
+            "codec_name": "h264",
+            "codec_type": "video"
+          },
+          "parent": 3,
+          "bytes": 18828335,
+          "packets": 19268
+        },
+        {
+          "id": 6,
+          "codec": {
+            "codec_name": "aac",
+            "codec_type": "audio",
+            "sample_rate": 16000
+          },
+          "parent": 5,
+          "bytes": 1890224,
+          "packets": 7471
+        }
+      ],
+      "bytes_send": 20702970
+    },
+    "query": "video&audio"
+  }
+}

tests/test_rest.py

@@ -16,6 +16,7 @@
 from go2rtc_client.rest import (
     _API_PREFIX,
     _ApplicationClient,
+    _PreloadClient,
     _SchemesClient,
     _StreamClient,
     _WebRTCClient,
@@ -135,7 +136,7 @@ async def test_streams_add_str(
     )
 
 
-VERSION_ERR = "server version '{}' not >= 1.9.12 and < 2.0.0"
+VERSION_ERR = "server version '{}' not >= 1.9.13 and < 2.0.0"
 
 
 @pytest.mark.parametrize(
@@ -144,8 +145,8 @@ async def test_streams_add_str(
         ("0.0.0", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("0.0.0"))),
         ("1.9.5", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("1.9.5"))),
         ("1.9.6", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("1.9.6"))),
-        ("1.9.12", does_not_raise()),
         ("1.9.13", does_not_raise()),
+        ("1.9.14", does_not_raise()),
         ("2.0.0", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("2.0.0"))),
         ("BLAH", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("BLAH"))),
     ],
@@ -237,3 +238,135 @@ async def test_schemes(
     )
     resp = await rest_client.schemes.list()
     assert resp == snapshot
+
+
+@pytest.mark.parametrize(
+    "filename",
+    ["preload_list_one.json", "streams_none.json"],
+    ids=[
+        "one stream preloaded",
+        "no stream preloaded",
+    ],
+)
+async def test_preload_list(
+    responses: aioresponses,
+    rest_client: Go2RtcRestClient,
+    snapshot: SnapshotAssertion,
+    filename: str,
+) -> None:
+    """Test preload list."""
+    responses.get(
+        f"{URL}{_PreloadClient.PATH}",
+        status=200,
+        body=load_fixture_str(filename),
+    )
+    resp = await rest_client.preload.list()
+    assert resp == snapshot
+
+
+async def test_preload_enable_no_filters(
+    responses: aioresponses,
+    rest_client: Go2RtcRestClient,
+) -> None:
+    """Test enable preload without codec filters."""
+    url = f"{URL}{_PreloadClient.PATH}"
+    camera = "camera.12mp_fluent"
+    params = {"src": camera}
+    responses.put(url + f"?src={camera}", status=200)
+    await rest_client.preload.enable(camera)
+
+    responses.assert_called_once_with(
+        url, method=METH_PUT, params=params, timeout=ClientTimeout(total=10)
+    )
+
+
+@pytest.mark.parametrize(
+    (
+        "video_codecs",
+        "audio_codecs",
+        "microphone_codecs",
+        "query_string",
+        "expected_params",
+    ),
+    [
+        (
+            ["h264"],
+            None,
+            None,
+            "?src=camera.12mp_fluent&video_codec_filter=h264",
+            {"src": "camera.12mp_fluent", "video_codec_filter": "h264"},
+        ),
+        (
+            None,
+            ["opus"],
+            None,
+            "?src=camera.12mp_fluent&audio_codec_filter=opus",
+            {"src": "camera.12mp_fluent", "audio_codec_filter": "opus"},
+        ),
+        (
+            None,
+            None,
+            ["pcmu"],
+            "?src=camera.12mp_fluent&microphone_codec_filter=pcmu",
+            {"src": "camera.12mp_fluent", "microphone_codec_filter": "pcmu"},
+        ),
+        (
+            ["h264", "h265"],
+            ["opus", "pcma"],
+            ["pcmu"],
+            "?src=camera.12mp_fluent&video_codec_filter=h264%2Ch265&audio_codec_filter=opus%2Cpcma&microphone_codec_filter=pcmu",
+            {
+                "src": "camera.12mp_fluent",
+                "video_codec_filter": "h264,h265",
+                "audio_codec_filter": "opus,pcma",
+                "microphone_codec_filter": "pcmu",
+            },
+        ),
+    ],
+    ids=[
+        "video filter only",
+        "audio filter only",
+        "microphone filter only",
+        "all filters",
+    ],
+)
+async def test_preload_enable_with_filters(
+    responses: aioresponses,
+    rest_client: Go2RtcRestClient,
+    video_codecs: list[str] | None,
+    audio_codecs: list[str] | None,
+    microphone_codecs: list[str] | None,
+    query_string: str,
+    expected_params: dict[str, str],
+) -> None:
+    """Test enable preload with codec filters."""
+    url = f"{URL}{_PreloadClient.PATH}"
+    camera = "camera.12mp_fluent"
+
+    responses.put(url + query_string, status=200)
+    await rest_client.preload.enable(
+        camera,
+        video_codec_filter=video_codecs,
+        audio_codec_filter=audio_codecs,
+        microphone_codec_filter=microphone_codecs,
+    )
+
+    responses.assert_called_once_with(
+        url, method=METH_PUT, params=expected_params, timeout=ClientTimeout(total=10)
+    )
+
+
+async def test_preload_disable(
+    responses: aioresponses,
+    rest_client: Go2RtcRestClient,
+) -> None:
+    """Test disable preload."""
+    url = f"{URL}{_PreloadClient.PATH}"
+    camera = "camera.12mp_fluent"
+    params = {"src": camera}
+    responses.delete(url + f"?src={camera}", status=200)
+    await rest_client.preload.disable(camera)
+
+    responses.assert_called_once_with(
+        url, method="DELETE", params=params, timeout=ClientTimeout(total=10)
+    )