Add support for camera talkback (two-way) audio

pyhap/camera.py

@@ -27,6 +27,7 @@
 
 from pyhap import RESOURCE_DIR, tlv
 from pyhap.accessory import Accessory
+from pyhap.camera_talkback import TalkbackReceiver, talkback_available
 from pyhap.const import CATEGORY_CAMERA
 from pyhap.util import byte_bool, to_base64_str
 
@@ -427,6 +428,13 @@ def __init__(self, options, *args, **kwargs):
 
             Additional optional values are:
             - 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"
+                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.
             - 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
@@ -435,6 +443,12 @@ def __init__(self, options, *args, **kwargs):
         :type options: ``dict``
         """
         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"]
@@ -448,6 +462,8 @@ def __init__(self, options, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
         self.add_preload_service("Microphone")
+        if self.support_talkback:
+            self.add_preload_service("Speaker").configure_char("Mute", value=False)
         self._streaming_status = []
         self._management = []
         self._setup_stream_management(options)
@@ -625,6 +641,7 @@ async def _start_stream(self, objs, reconfigure):  # pylint: disable=unused-argu
             logger.error(
                 "[%s] Failed to start/reconfigure stream, deleting session.", session_id
             )
+            self._close_talkback(session_info)
             del self.sessions[session_id]
             self._streaming_status[stream_idx] = STREAMING_STATUS["AVAILABLE"]
 
@@ -657,10 +674,49 @@ async def _stop_stream(self, objs):
 
         stream_idx = session_info["stream_idx"]
         await self.stop_stream(session_info)
+        self._close_talkback(session_info)
         del self.sessions[session_id]
 
         self._streaming_status[stream_idx] = STREAMING_STATUS["AVAILABLE"]
 
+    @staticmethod
+    def _close_talkback(session_info):
+        """Stop the talkback audio receiver for a session, if any."""
+        audio_backchannel = session_info.pop("audio_backchannel", None)
+        if audio_backchannel is not None:
+            audio_backchannel.stop()
+
+    def _talkback_payload_received(self, session_id, payload: bytes) -> None:
+        """Dispatch a received talkback RTP payload to the accessory implementation.
+
+        Called from the :class:`~pyhap.camera_talkback.TalkbackReceiver`
+        background thread, so this hops back onto the accessory driver's event
+        loop before calling the (synchronous) public callback, to keep
+        :meth:`talkback_audio_received` easy to implement without every caller
+        having to worry about thread-safety.
+        """
+        self.driver.loop.call_soon_threadsafe(
+            self.talkback_audio_received, session_id, payload
+        )
+
+    def talkback_audio_received(
+        self, session_id, payload: bytes
+    ) -> None:  # pylint: disable=unused-argument
+        """Handle received two-way ("talkback") audio for a streaming session.
+
+        Only called for sessions of accessories constructed with the
+        ``talkback`` option enabled (see ``__init__``). Override to consume the
+        received audio, e.g. decode it (the codec is whichever was negotiated,
+        see ``a_codec`` in the stream configuration passed to
+        :meth:`start_stream`) and play it out through a speaker.
+
+        :param session_id: The session ID this audio belongs to.
+        :type session_id: :class:`~uuid.UUID`
+        :param payload: The RTP payload of one received packet, still encoded
+            with the negotiated audio codec (not decoded by pyhap).
+        :type payload: ``bytes``
+        """
+
     def set_selected_stream_configuration(self, value):
         """Set the selected stream configuration.
 
@@ -787,6 +843,27 @@ def set_endpoints(self, value, stream_idx=None):
         video_ssrc = int.from_bytes(os.urandom(3), byteorder="big")
         audio_ssrc = int.from_bytes(os.urandom(3), byteorder="big")
 
+        # If talkback is enabled, open a local receiver for the audio the
+        # controller will send back during the session, and report *its* real
+        # port below instead of echoing the controller's own port back at it.
+        # Per the HAP spec (section 11, "IP Cameras"), the "Accessory Address"
+        # TLV in this response carries the accessory's own address/ports (the
+        # "Controller Address" received above carries the controller's), so
+        # echoing the controller's port here left HAP clients with nowhere
+        # valid to send the talkback audio to.
+        audio_backchannel = None
+        audio_rtp_port = target_audio_port
+        if self.support_talkback:
+            audio_backchannel = TalkbackReceiver(
+                srtp_key_and_salt=audio_master_key + audio_master_salt,
+                on_payload=functools.partial(
+                    self._talkback_payload_received, session_id
+                ),
+                is_ipv6=bool(is_ipv6),
+            )
+            audio_backchannel.start()
+            audio_rtp_port = audio_backchannel.local_port
+
         res_address_tlv = tlv.encode(
             SETUP_ADDR_INFO["ADDRESS_VER"],
             self.stream_address_isv6,
@@ -795,7 +872,7 @@ def set_endpoints(self, value, stream_idx=None):
             SETUP_ADDR_INFO["VIDEO_RTP_PORT"],
             struct.pack("<H", target_video_port),
             SETUP_ADDR_INFO["AUDIO_RTP_PORT"],
-            struct.pack("<H", target_audio_port),
+            struct.pack("<H", audio_rtp_port),
         )
 
         response_tlv = tlv.encode(
@@ -826,6 +903,7 @@ def set_endpoints(self, value, stream_idx=None):
             "a_port": target_audio_port,
             "a_srtp_key": to_base64_str(audio_master_key + audio_master_salt),
             "a_ssrc": audio_ssrc,
+            "audio_backchannel": audio_backchannel,
         }
 
         self._management[stream_idx].get_characteristic("SetupEndpoints").set_value(

pyhap/camera_talkback.py

@@ -0,0 +1,176 @@
+"""Support for the HomeKit camera audio talkback (two-way audio) channel.
+
+The HomeKit Accessory Protocol Specification (section 11 "IP Cameras") requires
+that, for a camera to support two-way audio, the accessory reports a local port
+where it listens for the audio the controller (e.g. the iOS Home app) sends back
+during a live view session ("Accessory Address" in the SetupEndpoints response,
+Table 9-16 of the spec). Prior to this module, :mod:`pyhap.camera` echoed back the
+*controller's* address in that field instead of a real local listening port,
+which meant no HAP camera accessory built with this library could ever receive
+the talkback audio: the controller had nowhere to send it.
+
+This module implements the receiving side of that channel: a UDP socket bound to
+an OS-assigned local port, SRTP decryption using the same key already negotiated
+by :meth:`pyhap.camera.Camera.set_endpoints`, and RTP depacketization. The
+decoded-from-RTP payload (still encoded with whichever audio codec was
+negotiated, typically Opus) is handed to a callback for the accessory
+implementation to decode/consume as it sees fit -- this mirrors how
+:meth:`pyhap.camera.Camera.start_stream` already leaves video/audio encoding of
+the *outgoing* stream to the implementation (or ffmpeg) rather than pyhap
+itself.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import logging
+import socket
+import struct
+import threading
+from typing import Callable
+
+try:
+    import pylibsrtp
+except ImportError:  # pragma: no cover - exercised via requires_talkback()
+    pylibsrtp = None
+
+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
+
+
+@dataclass
+class RtpPacket:
+    """A parsed RTP packet (RFC 3550), payload still codec-encoded."""
+
+    version: int
+    padding: bool
+    marker: bool
+    payload_type: int
+    sequence_number: int
+    timestamp: int
+    ssrc: int
+    payload: bytes
+
+
+def parse_rtp_packet(packet: bytes) -> RtpPacket:
+    """Parse the RTP header of an already SRTP-decrypted packet."""
+    if len(packet) < RTP_HEADER_LEN:
+        raise ValueError(f"Packet too short to be RTP: {len(packet)} bytes")
+
+    b0, b1, seq, timestamp, ssrc = struct.unpack("!BBHII", packet[:12])
+    version = b0 >> 6
+    padding = bool((b0 >> 5) & 1)
+    extension = (b0 >> 4) & 1
+    csrc_count = b0 & 0x0F
+    marker = bool(b1 >> 7)
+    payload_type = b1 & 0x7F
+
+    offset = RTP_HEADER_LEN + csrc_count * 4
+    if extension:
+        if len(packet) < offset + 4:
+            raise ValueError("Truncated RTP extension header")
+        ext_len_words = struct.unpack("!H", packet[offset + 2 : offset + 4])[0]
+        offset += 4 + ext_len_words * 4
+
+    if padding and packet:
+        pad_len = packet[-1]
+        payload = packet[offset : len(packet) - pad_len]
+    else:
+        payload = packet[offset:]
+
+    return RtpPacket(
+        version=version,
+        padding=padding,
+        marker=marker,
+        payload_type=payload_type,
+        sequence_number=seq,
+        timestamp=timestamp,
+        ssrc=ssrc,
+        payload=payload,
+    )
+
+
+class TalkbackReceiver:
+    """Listens for the HomeKit camera talkback (two-way) audio of one session.
+
+    :param srtp_key_and_salt: The audio SRTP master key concatenated with the
+        master salt, exactly as already negotiated in
+        :meth:`pyhap.camera.Camera.set_endpoints` (``audio_master_key +
+        audio_master_salt``). Reused here, not renegotiated.
+    :param on_payload: Called from a background thread with the RTP payload
+        (``bytes``) of each received packet, still encoded with whichever audio
+        codec was negotiated for the session (see ``a_codec`` in the stream
+        configuration passed to :meth:`pyhap.camera.Camera.start_stream`).
+    :param is_ipv6: Whether to bind an IPv6 socket instead of IPv4.
+    """
+
+    def __init__(
+        self,
+        srtp_key_and_salt: bytes,
+        on_payload: Callable[[bytes], None],
+        is_ipv6: bool = False,
+    ) -> 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,
+        )
+        self._session = pylibsrtp.Session(policy)
+        self._on_payload = on_payload
+
+        family = socket.AF_INET6 if is_ipv6 else socket.AF_INET
+        self._sock = socket.socket(family, socket.SOCK_DGRAM)
+        self._sock.bind(("::" if is_ipv6 else "0.0.0.0", 0))
+        self._sock.settimeout(0.5)
+
+        self._stop_event = threading.Event()
+        self._thread = threading.Thread(
+            target=self._run, name="hap-camera-talkback", daemon=True
+        )
+
+    @property
+    def local_port(self) -> int:
+        """The local port this receiver is bound to.
+
+        This is the value that must be reported in the ``AUDIO_RTP_PORT`` field
+        of the ``Accessory Address`` TLV in the ``SetupEndpoints`` response, so
+        that the controller knows where to send the talkback audio.
+        """
+        return self._sock.getsockname()[1]
+
+    def start(self) -> None:
+        """Start listening for incoming audio in a background thread."""
+        self._thread.start()
+
+    def stop(self) -> None:
+        """Stop listening and release the socket."""
+        self._stop_event.set()
+        self._thread.join(timeout=2)
+        self._sock.close()
+
+    def _run(self) -> None:
+        while not self._stop_event.is_set():
+            try:
+                data, _addr = self._sock.recvfrom(2048)
+            except TimeoutError:
+                continue
+            except OSError:
+                break
+            try:
+                rtp_plain = self._session.unprotect(data)
+                packet = parse_rtp_packet(rtp_plain)
+            except Exception:  # pylint: disable=broad-except
+                logger.exception("Failed to process incoming talkback audio packet")
+                continue
+            self._on_payload(packet.payload)

pyproject.toml

@@ -38,6 +38,9 @@ QRCode = [
     "base36",
     "pyqrcode",
 ]
+Talkback = [
+    "pylibsrtp",
+]
 
 [project.urls]
 "Source" = "https://github.com/ikalchev/HAP-python"

tests/conftest.py

@@ -51,6 +51,7 @@ def mock_local_address():
 class MockDriver:
     def __init__(self):
         self.loader = Loader()
+        self.loop = None
 
     def publish(self, data, client_addr=None, immediate=False):
         pass