feat(媒体): 同步视频 sidecar 协议与运行时路由

README.md

@@ -190,5 +190,25 @@ the desktop itself continues to control only the Daemon.
 Applications that intentionally implement their own media protocol can use
 the advanced `ApplicationChannels` Device channel and consume raw WSPK frames.
 
+## Live speaker PCM
+
+Firmware advertising `audio.stream.live.v1` accepts an unlimited-duration,
+backpressured PCM stream. Each `write()` blocks when the device's fixed-size
+playback queue has no credit, so the SDK never grows an unbounded host buffer:
+
+```python
+with app.robot.audio.open_stream(
+    sample_rate_hz=24000,
+    channels=1,
+    sample_width_bytes=2,
+) as speaker:
+    speaker.write(pcm_chunk)
+```
+
+Normal context exit drains queued PCM and sends the WSPK `LAST` marker;
+exceptional exit calls `abort()` and immediately stops playback. The existing
+`play_file()` and `play_pcm()` APIs remain bounded file transfers with size and
+SHA-256 validation.
+
 See [examples](examples/README.md), [Runtime contract](docs/contracts/runtime-profile-index.md),
 the [microphone contract](docs/microphone-audio.md), and [troubleshooting](docs/troubleshooting.md).

docs/microphone-audio.md

@@ -30,6 +30,11 @@ recording.save("recording.wav")
 
 ## Responsibility boundary
 
+The opposite direction uses `robot.audio.open_stream()` for 24 kHz mono
+PCM16. Robot-microphone upload and robot-speaker playback are real-time but
+half-duplex: opening either direction first closes the other. Camera preview
+does not participate in this arbitration.
+
 The Runtime/Daemon owns the physical device WebSocket, WSPK framing,
 connection lifecycle, and source-aware routing. It must keep device media
 payloads opaque when forwarding them to an Application; it does not decode,

src/watcherobot/__init__.py

@@ -6,7 +6,7 @@
     JobFailedError,
     WatcheRobotError,
 )
-from .audio import AudioPlayback, PCMAudio
+from .audio import AudioLiveStream, AudioPlayback, PCMAudio
 from .job import Job, JobState
 from .inputs import BackTouchEvent, InputDomain, InputEvent, RollerEvent, ScreenTouchEvent
 from .media import AudioFormat, AudioFrame, AudioRecording, ImageFrame, MicrophoneSession
@@ -39,6 +39,7 @@
 __all__ = [
     "AudioFormat",
     "AudioFrame",
+    "AudioLiveStream",
     "AudioPlayback",
     "AudioRecording",
     "AuthenticationError",

src/watcherobot/application/transport.py

@@ -62,6 +62,7 @@ def __init__(self, *, command_timeout: float = 5.0) -> None:
         self._audio_credits = 0
         self._audio_slots_per_packet = 1
         self._audio_flow_error: str | None = None
+        self._audio_live_first_frame = True
 
     def set_callbacks(
         self,
@@ -143,6 +144,35 @@ def send_audio_stream(
             self._send_audio_stream(bytes(pcm), stream_id, chunk_bytes)
         )
 
+    def begin_live_audio_stream(self, *, stream_id: int, chunk_bytes: int = 960) -> None:
+        self._submit(self._begin_live_audio_stream(stream_id, chunk_bytes)).result(
+            timeout=self.command_timeout + 1.0
+        )
+
+    def write_live_audio_stream(
+        self,
+        pcm: bytes,
+        *,
+        stream_id: int,
+        sequence: int,
+        chunk_bytes: int = 960,
+    ) -> int:
+        return self._submit(
+            self._write_live_audio_stream(
+                bytes(pcm), stream_id, sequence, chunk_bytes
+            )
+        ).result(timeout=self.command_timeout + 1.0)
+
+    def end_live_audio_stream(self, *, stream_id: int, sequence: int) -> None:
+        self._submit(self._end_live_audio_stream(stream_id, sequence)).result(
+            timeout=self.command_timeout + 1.0
+        )
+
+    def cancel_live_audio_stream(self, *, stream_id: int) -> None:
+        self._submit(self._cancel_live_audio_stream(stream_id)).result(
+            timeout=self.command_timeout + 1.0
+        )
+
     def send_desktop(self, frame: str | bytes) -> Future[None]:
         return self._submit(self._send(ApplicationChannel.DESKTOP, frame))
 
@@ -207,13 +237,22 @@ async def _run(self) -> None:
                 self._disconnect_callback()
         communicator_task.cancel()
         stop_task.cancel()
+        await self._fail_audio_flow("disconnected")
         await asyncio.gather(
             communicator_task,
             stop_task,
             return_exceptions=True,
         )
         self._communicators = None
 
+    async def _fail_audio_flow(self, reason: str) -> None:
+        condition = self._audio_credit_condition
+        if condition is None:
+            return
+        async with condition:
+            self._audio_flow_error = reason
+            condition.notify_all()
+
     async def _send(
         self,
         channel: ApplicationChannel,
@@ -296,6 +335,89 @@ async def _send_audio_stream(
                     self._audio_credits = 0
                     self._audio_slots_per_packet = 1
 
+    async def _begin_live_audio_stream(
+        self,
+        stream_id: int,
+        chunk_bytes: int = 960,
+    ) -> None:
+        if chunk_bytes <= 0 or chunk_bytes > 4096 or chunk_bytes % 2 != 0:
+            raise ValueError("chunk_bytes must be an even value between 2 and 4096")
+        if self._audio_credit_condition is None:
+            self._audio_credit_condition = asyncio.Condition()
+        async with self._audio_credit_condition:
+            self._audio_flow_stream_id = stream_id
+            self._audio_slots_per_packet = max(
+                1,
+                (chunk_bytes + AUDIO_DEVICE_SLOT_BYTES - 1)
+                // AUDIO_DEVICE_SLOT_BYTES,
+            )
+            self._audio_credits = 4
+            self._audio_flow_error = None
+            self._audio_live_first_frame = True
+            self._audio_credit_condition.notify_all()
+
+    async def _write_live_audio_stream(
+        self,
+        pcm: bytes,
+        stream_id: int,
+        sequence: int,
+        chunk_bytes: int = 960,
+    ) -> int:
+        if not pcm:
+            return sequence
+        if len(pcm) % 2 != 0:
+            raise ValueError("PCM chunk ends with a partial sample")
+        for offset in range(0, len(pcm), chunk_bytes):
+            await self._take_audio_credit(stream_id)
+            payload = pcm[offset : offset + chunk_bytes]
+            await self._send(
+                ApplicationChannel.DEVICE,
+                build_wspk(
+                    FRAME_AUDIO,
+                    FLAG_FIRST if self._audio_live_first_frame else 0,
+                    stream_id,
+                    sequence,
+                    payload,
+                ),
+            )
+            self._audio_live_first_frame = False
+            sequence = (sequence + 1) & 0xFFFFFFFF
+        return sequence
+
+    async def _end_live_audio_stream(self, stream_id: int, sequence: int) -> None:
+        condition = self._audio_credit_condition
+        if condition is None:
+            raise WatcheRobotError("audio flow control is not initialized")
+        try:
+            await self._send(
+                ApplicationChannel.DEVICE,
+                build_wspk(
+                    FRAME_AUDIO,
+                    FLAG_LAST,
+                    stream_id,
+                    sequence,
+                    b"",
+                ),
+            )
+        finally:
+            async with condition:
+                if self._audio_flow_stream_id == stream_id:
+                    self._audio_flow_stream_id = 0
+                    self._audio_credits = 0
+                    self._audio_slots_per_packet = 1
+                    condition.notify_all()
+
+    async def _cancel_live_audio_stream(self, stream_id: int) -> None:
+        condition = self._audio_credit_condition
+        if condition is None:
+            return
+        async with condition:
+            if self._audio_flow_stream_id == stream_id:
+                self._audio_flow_stream_id = 0
+                self._audio_credits = 0
+                self._audio_flow_error = "cancelled"
+                condition.notify_all()
+
     async def _take_audio_credit(self, stream_id: int) -> None:
         condition = self._audio_credit_condition
         if condition is None:
@@ -308,12 +430,12 @@ async def wait_for_credit() -> None:
                     or self._audio_credits > 0
                     or self._audio_flow_error is not None
                 )
-                if self._audio_flow_stream_id != stream_id:
-                    raise WatcheRobotError("audio stream was replaced")
                 if self._audio_flow_error is not None:
                     raise WatcheRobotError(
                         f"audio stream failed: {self._audio_flow_error}"
                     )
+                if self._audio_flow_stream_id != stream_id:
+                    raise WatcheRobotError("audio stream was replaced")
                 self._audio_credits -= 1
 
         try:

src/watcherobot/audio.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 import hashlib
+import threading
 import wave
 from dataclasses import dataclass
 from pathlib import Path
+from types import TracebackType
 from typing import Callable
 
+from .errors import WatcheRobotError
 from .job import CommandTransport, Job, JobState
 from .media import AudioFormat
 
@@ -58,6 +61,88 @@ def cancel(self) -> None:
         self._cancel_callback(self)
 
 
+class AudioLiveStream:
+    """Synchronous, backpressured PCM stream from the host to the robot."""
+
+    def __init__(
+        self,
+        stream_id: int,
+        write_callback: Callable[[AudioLiveStream, bytes, int], int],
+        close_callback: Callable[[AudioLiveStream, int], None],
+        abort_callback: Callable[[AudioLiveStream], None],
+    ) -> None:
+        self.stream_id = stream_id
+        self._write_callback = write_callback
+        self._close_callback = close_callback
+        self._abort_callback = abort_callback
+        self._sequence = 0
+        self._closed = False
+        self._lock = threading.RLock()
+        self._write_lock = threading.Lock()
+
+    @property
+    def closed(self) -> bool:
+        with self._lock:
+            return self._closed
+
+    def write(self, pcm_chunk: bytes) -> None:
+        payload = bytes(pcm_chunk)
+        if len(payload) % OUTPUT_AUDIO_FORMAT.sample_width_bytes != 0:
+            raise ValueError("PCM chunk ends with a partial sample")
+        with self._write_lock:
+            with self._lock:
+                if self._closed:
+                    raise WatcheRobotError("audio live stream is closed")
+                if not payload:
+                    return
+                sequence = self._sequence
+            next_sequence = self._write_callback(self, payload, sequence)
+            with self._lock:
+                if not self._closed:
+                    self._sequence = next_sequence
+
+    def close(self) -> None:
+        with self._write_lock:
+            with self._lock:
+                if self._closed:
+                    return
+                self._closed = True
+                sequence = self._sequence
+            try:
+                self._close_callback(self, sequence)
+            except Exception:
+                try:
+                    self._abort_callback(self)
+                except Exception:
+                    pass
+                raise
+
+    def abort(self) -> None:
+        with self._lock:
+            if self._closed:
+                return
+            self._closed = True
+        self._abort_callback(self)
+
+    def _mark_closed(self) -> None:
+        with self._lock:
+            self._closed = True
+
+    def __enter__(self) -> AudioLiveStream:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        if exc_type is None:
+            self.close()
+        else:
+            self.abort()
+
+
 def load_pcm_wave(path: str | Path) -> PCMAudio:
     """Read a WAV file in the single playback format supported by protocol v1."""
     source = Path(path)