Record HKSV clips with a prebuffer from the shared source pool

README.md

@@ -32,7 +32,6 @@ This will install HAP-python in your Python packages, so that you can import it
 `pyhap`. The following optional features can be installed via pip extras:
 
 * `QRCode` — print a pairing QR code in the terminal (`base36`, `pyqrcode`).
-* `Talkback` — two-way audio for the camera accessory (`pylibsrtp`).
 * `bench` — dependencies for running the benchmarks (`psutil`, `py-spy`).
 
 ## Demo

pyhap/camera.py

@@ -39,7 +39,7 @@
     tlv,
 )
 from pyhap.accessory import Accessory
-from pyhap.camera_talkback import TalkbackReceiver, talkback_available
+from pyhap.camera_talkback import TalkbackReceiver
 from pyhap.characteristic import Characteristic
 from pyhap.const import CATEGORY_CAMERA
 from pyhap.service import Service
@@ -49,6 +49,11 @@
     from pyhap.camera_multi_tier import MultiTierController
     from pyhap.camera_webrtc import WebRTCController
 
+# Seconds before arming a recording delegate is tried again, doubling up to
+# the maximum while it keeps failing.
+RECORDING_ARM_RETRY = 1.0
+MAX_RECORDING_ARM_RETRY = 60.0
+
 SETUP_TYPES = {
     "SESSION_ID": b"\x01",
     "STATUS": b"\x02",
@@ -596,8 +601,7 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None:
             - srtp - boolean, defaults to False. Whether the camera supports SRTP.
             - talkback - boolean, defaults to False. Whether the camera supports
                 receiving two-way ("talkback") audio from the controller during a
-                live view session. Requires the optional ``pylibsrtp`` dependency
-                (``pip install HAP-python[Talkback]``). When enabled, a "Speaker"
+                live view session. When enabled, a "Speaker"
                 service is added so that HAP clients know the accessory supports
                 it, and :meth:`talkback_audio_received` is called with the
                 received audio for each session that negotiates it.
@@ -612,6 +616,16 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None:
                 not allowed, the camera refuses to set up live streams and ends
                 running ones. Unknown names raise ``ValueError``, a value that is
                 not a list raises ``TypeError``.
+            - recording - ``True``, or a dict whose ``delegate`` is a
+                :class:`pyhap.hds_recording.RecordingProducer` such as
+                :class:`pyhap.media.recording_delegate.RecordingMediaDelegate`
+                that makes the recordings: the camera arms it while the hub has
+                recording on and the camera is on, advertises its
+                ``prebuffer_ms`` and records with its ``record``. Without a
+                delegate, override :meth:`handle_recording_stream`. Observed
+                with iOS and tvOS 27: hubs do not upload the HDS recordings of
+                a camera that also has ``multi_tier``, they never appear in the
+                Home app.
             - start_stream_cmd - string specifying the command to be executed to start
                 the stream. The string can contain the keywords, corresponding to the
                 video and audio configuration that was negotiated between the camera
@@ -634,6 +648,10 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None:
                 management service as not responding. Its services come before
                 the RTP stream management services, which then get a new
                 ``unique_id``; for a paired camera this changes instance IDs.
+                Observed with iOS and tvOS 27: the hub does not upload the
+                recordings it gets over HDS from a camera with this service, so
+                they do not appear in the Home app; the camera is expected to
+                upload its recordings itself.
             - webrtc - :class:`pyhap.camera_webrtc.WebRTCConfig`, or a dict of
                 its fields. Developer Preview of HomeKit Secure Video 17.99,
                 experimental. Adds a Camera WebRTC Stream Management service per
@@ -646,11 +664,6 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None:
         """
         self.has_srtp = options.get("srtp", False)
         self.support_talkback = options.get("talkback", False)
-        if self.support_talkback and not talkback_available():
-            raise ValueError(
-                "The 'talkback' option requires the optional 'pylibsrtp' "
-                "dependency. Install it with: pip install HAP-python[Talkback]"
-            )
         self.start_stream_cmd = options.get("start_stream_cmd", FFMPEG_CMD)
 
         self.stream_address = options["address"]
@@ -684,6 +697,16 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None:
         self._hds_listener = None
         self._hds_connections = set()
         self._recording_stream = None
+        recording = options.get("recording")
+        self._recording_media: hds_recording.RecordingProducer | None = (
+            recording.get("delegate") if isinstance(recording, dict) else None
+        )
+        self._recording_media_running = False
+        # What the delegate was last armed to; one task reconciles it.
+        self._recording_media_armed = False
+        self._recording_media_task: asyncio.Task[None] | None = None
+        # Set to cut a retry delay short when the wanted state changes.
+        self._recording_media_wake = asyncio.Event()
         self._homekit_camera_active_char: Characteristic | None = None
         # The HomeKit Camera Active of the Camera Operating Mode service, kept
         # in step with the global one when both exist.
@@ -719,13 +742,26 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None:
             if transport is not None:
                 for controller in self._stream_controllers():
                     controller.service.add_linked_service(transport)
+            if self._multi_tier and self._records_over_hds():
+                logger.warning(
+                    "%s: hubs with iOS or tvOS 27 do not upload HDS recordings of "
+                    "a camera with multi-tier streams, they will not appear in "
+                    "the Home app",
+                    self.display_name,
+                )
         if self._sensors:
             self._setup_camera_capabilities()
         # Last, so the instance IDs of the other services stay the same.
         if options.get("global_operating_mode") is not None:
             self._setup_global_operating_mode(options["global_operating_mode"])
         self._restore_hksv3_state()
 
+    def _records_over_hds(self) -> bool:
+        """Whether the camera makes HDS recordings, by delegate or override."""
+        return self._recording_media is not None or (
+            type(self).handle_recording_stream is not Camera.handle_recording_stream
+        )
+
     @property
     def streaming_status(self) -> bytes | None:
         """For backwards compatibility."""
@@ -957,6 +993,8 @@ def _default_recording_configs(
         ]
         audio = [hksv_recording.AudioCodecConfiguration()]
         general = hksv_recording.SupportedRecordingConfiguration()
+        if self._recording_media is not None:
+            general.prebuffer_length_ms = int(self._recording_media.prebuffer_ms)
         return general, video, audio
 
     def _setup_recording_management(self, options: dict[str, Any]) -> None:
@@ -1233,11 +1271,20 @@ async def handle_recording_stream(
         """Produce the fragmented-MP4 recording for ``stream_id``.
 
         Override to yield :class:`pyhap.hds_recording.RecordingPacket` objects,
-        the first being the MP4 initialization segment. The default yields
-        nothing (no recording).
+        the first being the MP4 initialization segment. The default records
+        with the ``delegate`` of the ``recording`` option, without one it
+        yields nothing (no recording).
         """
-        return
-        yield  # pragma: no cover - makes this an async generator
+        delegate = self._recording_media
+        selected = self._selected_recording_config
+        if delegate is None or selected is None:
+            return
+        packets = delegate.record(selected, self.recording_audio_active)
+        try:
+            async for packet in packets:
+                yield packet
+        finally:
+            await packets.aclose()
 
     @property
     def selected_recording_configuration(
@@ -1457,6 +1504,7 @@ def _save_state(self) -> None:
         recording_changed, self._recording_changed = self._recording_changed, False
         self.driver.async_persist()
         if recording_changed:
+            self._arm_recording_media()
             self.recording_state_changed()
 
     def _persist_hksv3_char(
@@ -1792,6 +1840,7 @@ def _handle_homekit_camera_active(self, value: Any) -> None:
             self._motion_status_active_char.notify()
         if not active and self._recording_stream is not None:
             self._recording_stream.close(hds_recording.RecordingReason.NOT_ALLOWED)
+        self._arm_recording_media()
         self._end_streams_if_not_allowed()
         self._schedule_stream_status()
         self.homekit_camera_active_changed(active)
@@ -2074,6 +2123,61 @@ async def run(self) -> None:
         await super().run()
         await self.start_recording_transport()
 
+    def _recording_media_wanted(self) -> bool:
+        return (
+            self._recording_media_running
+            and self.recording_active
+            and self.homekit_camera_active
+        )
+
+    def _arm_recording_media(self) -> None:
+        """Keep the recording delegate armed exactly while recording may run."""
+        if self._recording_media is None:
+            return
+        if self._recording_media_task is not None:
+            # A task reconciles already; it rechecks now if it waits to retry.
+            self._recording_media_wake.set()
+            return
+        if self._recording_media_wanted() == self._recording_media_armed:
+            return
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            return
+        self._recording_media_task = loop.create_task(self._reconcile_recording())
+
+    async def _reconcile_recording(self) -> None:
+        """Arm or disarm until the delegate matches what the camera wants."""
+        delegate = self._recording_media
+        assert delegate is not None  # noqa: S101 - only started with a delegate
+        failures = 0
+        try:
+            while (wanted := self._recording_media_wanted()) != (
+                self._recording_media_armed
+            ):
+                self._recording_media_wake.clear()
+                try:
+                    await delegate.set_armed(wanted)
+                except Exception as err:  # noqa: BLE001 - retried with a backoff
+                    # Meanwhile a recording still opens the channel on demand.
+                    self._recording_media_armed = False
+                    delay = min(
+                        RECORDING_ARM_RETRY * 2**failures, MAX_RECORDING_ARM_RETRY
+                    )
+                    failures += 1
+                    logger.warning(
+                        "Could not arm the recording, retrying in %.0f s: %s",
+                        delay,
+                        err,
+                    )
+                    with contextlib.suppress(asyncio.TimeoutError):
+                        await asyncio.wait_for(self._recording_media_wake.wait(), delay)
+                    continue
+                failures = 0
+                self._recording_media_armed = wanted
+        finally:
+            self._recording_media_task = None
+
     async def start_recording_transport(self) -> None:
         """Start the HDS listener on the HAP server's listen address(es).
 
@@ -2082,6 +2186,8 @@ async def start_recording_transport(self) -> None:
         """
         if self._recording_service is not None:
             await self._ensure_hds_listener()
+            self._recording_media_running = True
+            self._arm_recording_media()
 
     @property
     def recording_transport_port(self) -> int | None:
@@ -2099,6 +2205,13 @@ async def stop_recording_transport(self) -> None:
         listener, self._hds_listener = self._hds_listener, None
         if listener is not None:
             await listener.stop()
+        self._recording_media_running = False
+        if self._recording_media is not None:
+            if (task := self._recording_media_task) is not None:
+                self._recording_media_wake.set()
+                await asyncio.gather(task, return_exceptions=True)
+            self._recording_media_armed = False
+            await self._recording_media.close()
 
     async def stop(self) -> None:
         """Stop all streaming sessions and the HDS listener.

pyhap/camera_multi_tier.py

@@ -642,6 +642,10 @@ def _answer(session: MultiTierSession) -> hksv3.SetupEndpointsResponse:
 class FFmpegMultiTierDelegate:
     """Streams each session with its own ffmpeg process.
 
+    A simple reference: each session opens its own camera connection.
+    :class:`pyhap.media.MultiTierMediaDelegate` shares one pull per camera
+    channel among all local and remote sessions.
+
     Pass it as the ``delegate`` of the ``multi_tier`` option. The command is
     ``ffmpeg``, :data:`FFMPEG_GLOBAL_OPTIONS`, the ``ffmpeg_input`` arguments,
     :data:`FFMPEG_VIDEO_OUTPUT`

pyhap/camera_talkback.py

@@ -31,19 +31,20 @@
 import threading
 from typing import cast
 
-try:
-    import pylibsrtp
-except ImportError:  # pragma: no cover - exercised via requires_talkback()
-    pylibsrtp = None
+import pylibsrtp
 
 logger = logging.getLogger(__name__)
 
 RTP_HEADER_LEN = 12
 
 
 def talkback_available() -> bool:
-    """Whether the optional dependency for talkback support is installed."""
-    return pylibsrtp is not None
+    """Whether talkback is supported, always since pylibsrtp is a dependency.
+
+    Deprecated: kept for callers of the optional dependency check, it will be
+    removed in a future release.
+    """
+    return True
 
 
 @dataclass
@@ -148,12 +149,6 @@ def __init__(
         is_ipv6: bool = False,
         controller_address: str | None = None,
     ) -> None:
-        if pylibsrtp is None:
-            raise RuntimeError(
-                "Talkback support requires the 'pylibsrtp' package. "
-                "Install it with: pip install HAP-python[Talkback]"
-            )
-
         policy = pylibsrtp.Policy(
             key=srtp_key_and_salt,
             ssrc_type=pylibsrtp.Policy.SSRC_ANY_INBOUND,

pyhap/hds_recording.py

@@ -16,15 +16,18 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import AsyncIterator, Callable
+from collections.abc import AsyncGenerator, AsyncIterator, Callable
 from dataclasses import dataclass
 from enum import IntEnum
 import functools
 import logging
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, Protocol, cast
 
 from pyhap.hds_protocol import HDSStatus, Message
 
+if TYPE_CHECKING:
+    from pyhap.hksv_recording import SelectedRecordingConfiguration
+
 logger = logging.getLogger(__name__)
 
 _DATA_SEND = "dataSend"
@@ -62,6 +65,37 @@ class RecordingPacket:
 # A delegate is called with the stream id and yields RecordingPackets. The first
 # yielded packet is the MP4 initialization segment.
 RecordingDelegate = Callable[[int], AsyncIterator[RecordingPacket]]
+
+
+class ByteReader(Protocol):
+    """A byte stream, e.g. an :class:`asyncio.StreamReader`."""
+
+    async def readexactly(self, n: int) -> bytes:
+        """Read ``n`` bytes or raise :class:`asyncio.IncompleteReadError`."""
+
+
+class RecordingProducer(Protocol):
+    """Makes a camera's recordings, see the camera's ``recording`` option.
+
+    :class:`pyhap.media.recording_delegate.RecordingMediaDelegate` is one.
+    """
+
+    @property
+    def prebuffer_ms(self) -> int:
+        """The prebuffer the camera advertises."""
+
+    async def set_armed(self, armed: bool) -> None:
+        """Armed while the hub has recording on and the camera is on."""
+
+    def record(
+        self, selected: SelectedRecordingConfiguration, audio: bool
+    ) -> AsyncGenerator[RecordingPacket, None]:
+        """Yield the selected recording, the init segment first."""
+
+    async def close(self) -> None:
+        """Disarm and release everything."""
+
+
 # An open gate returns a reason to reject a dataSend/open, or None to accept it.
 OpenGate = Callable[["RecordingStreamManager"], RecordingReason | None]
 
@@ -276,7 +310,7 @@ async def _send_fragment(
 _MAX_FRAGMENT_SIZE = 64 * 1024 * 1024
 
 
-async def _read_mp4_box(reader: asyncio.StreamReader, max_size: int) -> bytes | None:
+async def _read_mp4_box(reader: ByteReader, max_size: int) -> bytes | None:
     """Read one top-level MP4 box (header + body) from ``reader``.
 
     Returns ``None`` at end of stream (also mid-box) and for a size smaller
@@ -299,7 +333,7 @@ async def _read_mp4_box(reader: asyncio.StreamReader, max_size: int) -> bytes |
 
 
 async def fmp4_recording_packets(
-    reader: asyncio.StreamReader,
+    reader: ByteReader,
 ) -> AsyncIterator[RecordingPacket]:
     """Turn a fragmented-MP4 byte stream into HKSV recording packets.