diff --git a/README.md b/README.md index ca445bf..9b424ba 100644 --- a/README.md +++ b/README.md @@ -44,6 +44,7 @@ Scripts that do not use asyncio can use the same API as blocking calls through certificates, sessions, passwords and user accounts - [API reference](docs/api.md): every method and the hardware and firmware it needs +- [Video streaming](docs/streaming.md): snapshots, MJPEG and WebRTC signaling - [Blocking client](docs/sync.md): the API without asyncio - [Errors](docs/errors.md) - [SSH](docs/ssh.md) diff --git a/docs/api.md b/docs/api.md index a50a64b..b394d11 100644 --- a/docs/api.md +++ b/docs/api.md @@ -117,8 +117,8 @@ while loop.time() < end: | Methods | Requires | |---|---| -| `mjpeg_stream`, `mjpeg_stream_bytes`, `open_ws` | all | -| `set_screen`, `set_frame_detect`, `pause_frame_detect` | non-Pro | +| `mjpeg_stream`, `mjpeg_stream_bytes`, `open_ws`, `open_webrtc_signaling` | all | +| `get_snapshot`, `set_screen`, `set_frame_detect`, `pause_frame_detect` | non-Pro | | `get_input_region`, `set_input_region`, `get_input_resolution` | non-Pro 2.5.1 | | `reset_hdmi` | PCIe 2.1.5 | | `get_hdmi_state`, `enable_hdmi`, `disable_hdmi` | PCIe 2.2.8 | @@ -130,10 +130,15 @@ while loop.time() < end: | `set_rate_control_mode` | Pro 1.2.6 | | `set_fps` | Pro 1.2.8 | -`mjpeg_stream()` yields decoded `PIL.Image` frames; `mjpeg_stream_bytes()` -yields the raw JPEG bytes. `open_ws(path)` opens a new WebSocket on a path -relative to the API base, such as `stream/h264/webrtc`, for protocols the -client does not wrap. The caller closes it before the client exits. +`mjpeg_stream()` yields decoded `PIL.Image` frames, `mjpeg_stream_bytes()` the +raw JPEG bytes, and `get_snapshot()` the first one. `open_webrtc_signaling()` +returns the H.264 WebRTC signaling session for the detected hardware. +[Video streaming](streaming.md) covers both, and why the Pro refuses +snapshots. + +`open_ws(path)` opens a new WebSocket on a path relative to the API base, such +as `stream/h264/webrtc`, for protocols the client does not wrap. The caller +closes it before the client exits. ### Images and scripts diff --git a/docs/errors.md b/docs/errors.md index f6b1030..cd70cca 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -11,6 +11,7 @@ All exceptions derive from `NanoKVMError` and are importable from | `NanoKVMPermissionError` | The account lacks permission; the session stays valid | | `NanoKVMApiError` | The device answered with an error code | | `NanoKVMInvalidResponseError` | The response could not be parsed | +| `NanoKVMVideoModeError` | The Pro stopped a WebRTC stream because another video mode took the encoder | `NanoKVMPermissionError` carries `status`, `method` and `path`. `NanoKVMApiError` carries the numeric `code`; the device's `msg` and `data` are diff --git a/docs/streaming.md b/docs/streaming.md new file mode 100644 index 0000000..548a639 --- /dev/null +++ b/docs/streaming.md @@ -0,0 +1,102 @@ +# Video streaming + +The NanoKVM serves the HDMI input as an MJPEG stream and as H.264 over +WebRTC. The client reads MJPEG itself; for WebRTC it handles the signaling and +leaves the media to a WebRTC peer, such as a browser or `aiortc`. + +| Need | Method | +|---|---| +| One still image | `get_snapshot()` (non-Pro) | +| Raw JPEG frames | `mjpeg_stream_bytes()` | +| Decoded `PIL.Image` frames | `mjpeg_stream()` | +| Live H.264 video, and audio on the Pro | `open_webrtc_signaling()` | + +## The Pro's single encoder + +The Pro encodes one stream mode at a time (`StreamMode`, set with +`set_stream_mode()`). Opening the MJPEG stream switches it to MJPEG, and any +WebRTC session in progress stops: its iteration raises +`NanoKVMVideoModeError`. For that reason `get_snapshot()` raises +`NanoKVMNotSupportedError` on the Pro without opening the stream +([NanoKVM-Pro#157](https://github.com/sipeed/NanoKVM-Pro/issues/157)). +`mjpeg_stream()` and `mjpeg_stream_bytes()` still work on the Pro, at that +cost. + +## MJPEG + +```python +from contextlib import aclosing +from pathlib import Path + +Path("screen.jpg").write_bytes(await client.get_snapshot()) + +async with aclosing(client.mjpeg_stream_bytes()) as frames: + async for frame in frames: + ... # one JPEG per iteration +``` + +`get_snapshot(timeout=20.0)` returns the first frame, raises `TimeoutError` +when none arrives in time, and `NanoKVMError` when the stream ends without one. +The streams run until the caller stops iterating; `aclosing()` closes the HTTP +response as soon as the loop is left. + +## WebRTC signaling + +`open_webrtc_signaling()` picks the endpoint for the detected hardware and +returns a `nanokvm.webrtc.WebRTCSignaling` on its own WebSocket: + +```python +from nanokvm.webrtc import IceCandidate, WebRTCAnswer, WebRTCCandidate + +async with await client.open_webrtc_signaling() as signaling: + await signaling.send_offer(offer_sdp) + async for event in signaling: + if isinstance(event, WebRTCAnswer): + ... # hand event.sdp to the peer as the remote description + elif isinstance(event, WebRTCCandidate): + ... # hand event.candidate to the peer +``` + +1. Send the peer's offer with `send_offer(sdp)`. Ask for video, and audio too + on the Pro; the non-Pro has video only. +2. Send each of the peer's candidates with + `send_candidate(IceCandidate(...))`, after the offer. `IceCandidate` takes + the browser's `RTCIceCandidateInit` field names (`sdpMid`, `sdpMLineIndex`, + `usernameFragment`) or the snake_case ones. +3. Iterate for exactly one `WebRTCAnswer` and any number of + `WebRTCCandidate` events. Iterating and sending can run concurrently, for + example the iteration in its own task. +4. Iteration ends when the device closes the WebSocket. Close the session, or + leave its `async with`, before the client exits: the session owns the + WebSocket, not the client. + +A `ValueError` from `send_offer()` means a Pro offer without audio or video. +The Pro ignores WebSocket pings, so the session sends its application +heartbeat every 60 seconds; a failed heartbeat closes the session, which ends +the iteration. + +### What happens on the Pro + +The non-Pro negotiates one video peer connection on `stream/h264`, so the +session passes the offer, answer and candidates through. The Pro negotiates +video and audio as two peer connections on `stream/h264/webrtc`; the session +hides that, so the caller sees one offer and one answer on any hardware: + +- The offer is split into one single-section offer per media kind. +- The answers are merged into one answer to the original offer, once all + kinds have answered. Candidates that arrive before that are held, then + yielded after the answer. +- Candidates are mapped both ways between the sections of the original offer + and the single-section peers. + +The split and merge are public in `nanokvm.webrtc` for callers that run their +own signaling: `split_pro_webrtc_offer()` returns a `ProWebRTCOffer`, whose +`upstream_candidate_for_kind()` and `client_candidate_for_kind()` map +candidates, and `merge_pro_webrtc_answers()` builds the answer. + +## Blocking client + +`NanoKVMSyncClient` offers `get_snapshot()` and the MJPEG streams as blocking +calls. WebRTC signaling is async-only, because its WebSocket belongs to the +client's event loop: reach it through `run()`, as in +[Blocking client](sync.md). diff --git a/docs/sync.md b/docs/sync.md index 79bd083..01736e4 100644 --- a/docs/sync.md +++ b/docs/sync.md @@ -40,8 +40,8 @@ the wrapped `client` on its event loop and keeps the result type: gpio = kvm.run(kvm.client.get_gpio()) ``` -`run()` is also the way to the async-only `open_ws()`, whose WebSocket belongs -to the client's event loop: +`run()` is also the way to the async-only `open_ws()` and +`open_webrtc_signaling()`, whose WebSockets belong to the client's event loop: ```python async def read_message(client): diff --git a/nanokvm/client.py b/nanokvm/client.py index 351b964..a11278b 100644 --- a/nanokvm/client.py +++ b/nanokvm/client.py @@ -27,6 +27,7 @@ NanoKVMNotAuthenticatedError, NanoKVMNotSupportedError, NanoKVMPermissionError, + NanoKVMVideoModeError, ) from .models import * # noqa: F403 from .models import GetAccountRsp, HWVersion, IsPasswordUpdatedRsp @@ -43,6 +44,7 @@ "NanoKVMNotAuthenticatedError", "NanoKVMNotSupportedError", "NanoKVMPermissionError", + "NanoKVMVideoModeError", "PASTE_CHAR_MAP", "SessionController", "StreamController", diff --git a/nanokvm/components/stream.py b/nanokvm/components/stream.py index a0cd09c..b16843d 100644 --- a/nanokvm/components/stream.py +++ b/nanokvm/components/stream.py @@ -13,10 +13,17 @@ from aiohttp import BodyPartReader, ClientWebSocketResponse, MultipartReader, hdrs from PIL import Image +from ..compatibility import require_hardware +from ..exceptions import NanoKVMError +from ..models.common import HWFamily +from ..webrtc import WebRTCSignaling from .session import Controller _LOGGER = logging.getLogger(__name__) +_NON_PRO_WS_HEARTBEAT_SECONDS = 30.0 +_WS_CLOSE_TIMEOUT_SECONDS = 15.0 + class StreamController(Controller): """Own MJPEG frame transport and JPEG decoding.""" @@ -48,6 +55,22 @@ async def mjpeg_stream_bytes(self) -> AsyncGenerator[bytes, None]: session.check_session_generation(generation) yield data + @require_hardware(HWFamily.NON_PRO) + async def get_snapshot(self, timeout: float = 20.0) -> bytes: + """Return one raw JPEG frame from the MJPEG stream. + + Not on the Pro: its MJPEG and H.264 streams share one encoder mode, so + opening MJPEG interrupts any WebRTC session + (https://github.com/sipeed/NanoKVM-Pro/issues/157). + """ + async with ( + asyncio.timeout(timeout), + contextlib.aclosing(self.mjpeg_stream_bytes()) as frames, + ): + async for frame in frames: + return frame + raise NanoKVMError("MJPEG stream ended without a frame") + async def mjpeg_stream(self) -> AsyncGenerator[Image.Image, None]: """Stream decoded MJPEG frames.""" session = self._session @@ -70,3 +93,20 @@ async def open_ws(self, path: str, **kwargs: Any) -> ClientWebSocketResponse: caller closes the WebSocket, before the client exits. """ return await self._session.open_ws(path, **kwargs) + + async def open_webrtc_signaling(self) -> WebRTCSignaling: + """Open an H.264 WebRTC signaling session, on either hardware family. + + The caller closes it, before the client exits. + """ + session = self._session + if session._hw_version is None: + await session.detect_hardware() + pro = session.is_hardware_family(HWFamily.PRO) + websocket = await session.open_ws( + "stream/h264/webrtc" if pro else "stream/h264", + # The Pro gets an application-level heartbeat from WebRTCSignaling. + heartbeat=None if pro else _NON_PRO_WS_HEARTBEAT_SECONDS, + timeout=aiohttp.ClientWSTimeout(ws_close=_WS_CLOSE_TIMEOUT_SECONDS), + ) + return WebRTCSignaling(websocket, pro=pro) diff --git a/nanokvm/exceptions.py b/nanokvm/exceptions.py index 26c265c..ce1fd94 100644 --- a/nanokvm/exceptions.py +++ b/nanokvm/exceptions.py @@ -50,3 +50,7 @@ class NanoKVMInvalidResponseError(NanoKVMError): class NanoKVMNotSupportedError(NanoKVMError): """Feature not supported on this hardware variant.""" + + +class NanoKVMVideoModeError(NanoKVMError): + """The NanoKVM Pro stopped a stream because another video mode is active.""" diff --git a/nanokvm/sync_client.py b/nanokvm/sync_client.py index 7a644f9..82348d0 100644 --- a/nanokvm/sync_client.py +++ b/nanokvm/sync_client.py @@ -18,7 +18,7 @@ T = TypeVar("T") # Their results stay bound to the client's event loop; use ``run()`` instead. -_ASYNC_ONLY = frozenset({"open_ws"}) +_ASYNC_ONLY = frozenset({"open_ws", "open_webrtc_signaling"}) # Wait in short slices: before Python 3.14, an untimed wait on Windows ignores # Ctrl-C until the operation finishes. diff --git a/nanokvm/webrtc.py b/nanokvm/webrtc.py new file mode 100644 index 0000000..26185ed --- /dev/null +++ b/nanokvm/webrtc.py @@ -0,0 +1,529 @@ +"""H.264 WebRTC signaling with the NanoKVM. + +The non-Pro firmware negotiates one video peer connection on +``stream/h264``. The Pro negotiates separate video and audio peer +connections on ``stream/h264/webrtc``; ``WebRTCSignaling`` splits a +combined browser offer for it and merges the answers back, so callers see +one offer and one answer on either hardware. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncGenerator +import contextlib +from dataclasses import dataclass +import json +import logging +from typing import Any, Literal + +from aiohttp import ClientWebSocketResponse, WSMsgType +from pydantic import BaseModel, ConfigDict, Field, ValidationError + +from .exceptions import NanoKVMVideoModeError + +_LOGGER = logging.getLogger(__name__) + +MediaKind = Literal["audio", "video"] + +_SUPPORTED_KINDS: tuple[MediaKind, ...] = ("audio", "video") +_PRO_OFFER_ORDER: tuple[MediaKind, ...] = ("video", "audio") +_TRANSPORT_SESSION_PREFIXES = ( + "a=ice-ufrag:", + "a=ice-pwd:", + "a=ice-options:", + "a=fingerprint:", + "a=setup:", +) +_ANSWER_EVENTS: dict[str, MediaKind] = { + "video-answer": "video", + "audio-answer": "audio", +} +_CANDIDATE_EVENTS: dict[str, MediaKind] = { + "video-candidate": "video", + "audio-candidate": "audio", +} +_STATUS_EVENTS = {"video-status", "audio-status"} +_PRO_HEARTBEAT_SECONDS = 60.0 +_PRO_VIDEO_STATUS_INCONSISTENT_MODE = -4 +_PRO_VIDEO_STATUS_NAMES = { + 1: "normal", + -1: "no_image", + _PRO_VIDEO_STATUS_INCONSISTENT_MODE: "inconsistent_video_mode", +} + + +class IceCandidate(BaseModel): + """An ICE candidate, as the browser's ``RTCIceCandidateInit``.""" + + model_config = ConfigDict(frozen=True, populate_by_name=True) + + candidate: str + sdp_mid: str | None = Field(default=None, alias="sdpMid") + sdp_m_line_index: int | None = Field(default=None, alias="sdpMLineIndex", ge=0) + username_fragment: str | None = Field(default=None, alias="usernameFragment") + + +@dataclass(frozen=True, slots=True) +class WebRTCAnswer: + """The device's SDP answer to the offer.""" + + sdp: str + + +@dataclass(frozen=True, slots=True) +class WebRTCCandidate: + """A device ICE candidate, mapped onto the original offer.""" + + candidate: IceCandidate + + +WebRTCEvent = WebRTCAnswer | WebRTCCandidate + + +class WebRTCSignaling: + """One WebRTC signaling session on a caller-owned WebSocket. + + Send the offer first, then trickle candidates both ways: iterate for the + device's answer and candidates. Iteration ends when the WebSocket closes, + and raises ``NanoKVMVideoModeError`` when the Pro drops WebRTC because + another video mode took the encoder. Close it before the client exits. + """ + + def __init__(self, websocket: ClientWebSocketResponse, *, pro: bool) -> None: + self.websocket = websocket + self._pro = pro + self._offer: ProWebRTCOffer | None = None + self._heartbeat: asyncio.Task[None] | None = None + if pro: + self._heartbeat = asyncio.create_task(self._send_heartbeats()) + + async def __aenter__(self) -> WebRTCSignaling: + return self + + async def __aexit__(self, *exc_info: object) -> None: + await self.close() + + async def close(self) -> None: + """Stop the heartbeat and close the WebSocket.""" + if ( + self._heartbeat is not None + and self._heartbeat is not asyncio.current_task() + ): + self._heartbeat.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._heartbeat + await self.websocket.close() + + async def send_offer(self, sdp: str) -> None: + """Send the browser's SDP offer. + + Raises ``ValueError`` when a Pro offer has neither audio nor video. + """ + if not self._pro: + await self._send("video-offer", {"type": "offer", "sdp": sdp}) + return + + offer = split_pro_webrtc_offer(sdp) + if not offer.offers: + raise ValueError("WebRTC offer does not contain audio or video media") + self._offer = offer + for kind in _PRO_OFFER_ORDER: + if (kind_sdp := offer.offers.get(kind)) is not None: + await self._send(f"{kind}-offer", {"type": "offer", "sdp": kind_sdp}) + + async def send_candidate(self, candidate: IceCandidate) -> None: + """Send one browser ICE candidate, after the offer.""" + if not self._pro: + await self._send("video-candidate", _candidate_payload(candidate)) + return + + if self._offer is None: + raise RuntimeError("Send the offer before candidates") + for kind in self._offer.kinds_for_candidate(candidate): + mapped = self._offer.upstream_candidate_for_kind(kind, candidate) + if mapped is not None: + await self._send(f"{kind}-candidate", _candidate_payload(mapped)) + + def __aiter__(self) -> AsyncGenerator[WebRTCEvent, None]: + return self._events() + + async def _events(self) -> AsyncGenerator[WebRTCEvent, None]: + answers: dict[MediaKind, str] = {} + # The Pro's candidates arrive per peer; hold them until the merged + # answer, which is what gives them a section to belong to. + early: list[tuple[MediaKind, IceCandidate]] = [] + answered = False + + async for msg in self.websocket: + if msg.type is not WSMsgType.TEXT: + if msg.type in (WSMsgType.CLOSED, WSMsgType.CLOSE, WSMsgType.ERROR): + break + continue + payload = _decode_json_object(msg.data) + event = payload.get("event") if payload is not None else None + if not isinstance(event, str) or event == "heartbeat": + continue + assert payload is not None + raw_data = payload.get("data") + data = _decode_signal_data(raw_data) + + if not self._pro: + if event == "video-answer": + if sdp := _answer_sdp(data): + yield WebRTCAnswer(sdp) + elif event == "video-candidate": + if (candidate := _parse_candidate(event, data)) is not None: + yield WebRTCCandidate(candidate) + else: + _LOGGER.debug("Unhandled WebRTC event: %s", event) + continue + + offer = self._offer + if event in _STATUS_EVENTS: + code = _status_code(data, raw_data) + _LOGGER.debug( + "Pro WebRTC %s: %s", + event, + _PRO_VIDEO_STATUS_NAMES.get(code, code) + if event == "video-status" and code is not None + else raw_data, + ) + if ( + event == "video-status" + and code == _PRO_VIDEO_STATUS_INCONSISTENT_MODE + ): + raise NanoKVMVideoModeError( + "NanoKVM Pro stopped WebRTC video because another " + "video mode is active" + ) + elif (kind := _ANSWER_EVENTS.get(event)) is not None: + sdp = _answer_sdp(data) + if offer is None or answered or kind not in offer.offers or not sdp: + continue + answers[kind] = sdp + if set(offer.expected_kinds).issubset(answers): + answered = True + yield WebRTCAnswer(merge_pro_webrtc_answers(offer, answers)) + for early_kind, early_candidate in early: + if mapped := offer.client_candidate_for_kind( + early_kind, early_candidate + ): + yield WebRTCCandidate(mapped) + early.clear() + elif (kind := _CANDIDATE_EVENTS.get(event)) is not None: + if offer is None: + continue + if (candidate := _parse_candidate(event, data)) is None: + continue + if not answered: + early.append((kind, candidate)) + elif mapped := offer.client_candidate_for_kind(kind, candidate): + yield WebRTCCandidate(mapped) + else: + _LOGGER.debug("Unhandled Pro WebRTC event: %s", event) + + _LOGGER.debug( + "WebRTC signaling ended: close_code=%s exception=%r", + self.websocket.close_code, + self.websocket.exception(), + ) + + async def _send(self, event: str, data: str | dict[str, Any]) -> None: + await self.websocket.send_json( + { + "event": event, + "data": data if isinstance(data, str) else json.dumps(data), + } + ) + + async def _send_heartbeats(self) -> None: + """Keep the Pro session alive; it ignores WebSocket-level pings.""" + while not self.websocket.closed: + await asyncio.sleep(_PRO_HEARTBEAT_SECONDS) + if self.websocket.closed: + return + try: + await self._send("heartbeat", "") + except Exception as err: + _LOGGER.debug("Pro WebRTC heartbeat failed: %s", err) + # Closing ends iteration, which is how the caller hears of it. + await self.close() + return + + +def _answer_sdp(data: dict[str, object] | None) -> str | None: + sdp = data.get("sdp") if data is not None else None + return sdp if isinstance(sdp, str) and sdp else None + + +def _candidate_payload(candidate: IceCandidate) -> dict[str, Any]: + return candidate.model_dump(by_alias=True, exclude_none=True) + + +def _parse_candidate(event: str, data: dict[str, object] | None) -> IceCandidate | None: + if data is None: + return None + try: + return IceCandidate.model_validate(data) + except ValidationError as err: + _LOGGER.debug("Invalid %s payload: %s (%r)", event, err, data) + return None + + +def _decode_json_object(raw: object) -> dict[str, object] | None: + try: + decoded = json.loads(raw) # type: ignore[arg-type] + except (TypeError, ValueError): + _LOGGER.debug("Invalid WebRTC signaling JSON: %r", raw) + return None + return decoded if isinstance(decoded, dict) else None + + +def _decode_signal_data(raw_data: object) -> dict[str, object] | None: + """Decode the ``data`` field, which the device sends as a JSON string.""" + if isinstance(raw_data, dict): + return raw_data + if isinstance(raw_data, str): + return _decode_json_object(raw_data) + return None + + +def _status_code(data: dict[str, object] | None, raw_data: object) -> int | None: + """Return the numeric code of a Pro status event, in any of its shapes.""" + candidates: list[object] = [] + if data is not None: + candidates += [data.get(key) for key in ("status", "code", "value")] + candidates.append(raw_data) + if isinstance(raw_data, str): + with contextlib.suppress(ValueError): + candidates.append(json.loads(raw_data)) + + for value in candidates: + if isinstance(value, bool) or not isinstance(value, int | float | str): + continue + with contextlib.suppress(ValueError): + return int(value) + return None + + +@dataclass(frozen=True, slots=True) +class ProMediaSection: + """A media section of the original combined offer.""" + + kind: MediaKind + mid: str + index: int + lines: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class ProWebRTCOffer: + """Per-kind Pro offers, and the way back to the original offer.""" + + offers: dict[MediaKind, str] + media_sections: tuple[ProMediaSection, ...] + + @property + def expected_kinds(self) -> tuple[MediaKind, ...]: + """Return the media kinds the Pro is expected to answer.""" + return tuple(section.kind for section in self.media_sections) + + def section_for_kind(self, kind: MediaKind) -> ProMediaSection | None: + """Return the original offer section for a media kind.""" + for section in self.media_sections: + if section.kind == kind: + return section + return None + + def kinds_for_candidate(self, candidate: IceCandidate) -> tuple[MediaKind, ...]: + """Return the Pro peers that should receive a browser candidate.""" + if candidate.sdp_mid is not None: + for section in self.media_sections: + if section.mid == candidate.sdp_mid: + return (section.kind,) + + if candidate.sdp_m_line_index is not None: + for section in self.media_sections: + if section.index == candidate.sdp_m_line_index: + return (section.kind,) + + return self.expected_kinds + + def upstream_candidate_for_kind( + self, kind: MediaKind, candidate: IceCandidate + ) -> IceCandidate | None: + """Map a browser candidate onto the single-section Pro peer.""" + if self.section_for_kind(kind) is None: + return None + return candidate.model_copy(update={"sdp_mid": "0", "sdp_m_line_index": 0}) + + def client_candidate_for_kind( + self, kind: MediaKind, candidate: IceCandidate + ) -> IceCandidate | None: + """Map a Pro candidate back onto its section of the original offer.""" + section = self.section_for_kind(kind) + if section is None: + return None + return candidate.model_copy( + update={"sdp_mid": section.mid, "sdp_m_line_index": section.index} + ) + + +def split_pro_webrtc_offer(offer_sdp: str) -> ProWebRTCOffer: + """Split a combined offer into one Pro offer per media kind.""" + session_lines, media_lines = _split_sdp_sections(offer_sdp) + media_sections: list[ProMediaSection] = [] + offers: dict[MediaKind, str] = {} + + for index, lines in enumerate(media_lines): + kind = _media_kind(lines) + if kind is None or kind in offers: + continue + + section = ProMediaSection( + kind=kind, + mid=_media_mid(lines, fallback=str(index)), + index=index, + lines=lines, + ) + media_sections.append(section) + offers[kind] = _join_sdp_sections( + _single_media_session_lines(session_lines), + (_replace_mid(section.lines, "0"),), + ) + + return ProWebRTCOffer(offers=offers, media_sections=tuple(media_sections)) + + +def merge_pro_webrtc_answers( + offer: ProWebRTCOffer, answers: dict[MediaKind, str] +) -> str: + """Merge the per-kind Pro answers into one answer to the original offer.""" + if not answers: + raise ValueError("No NanoKVM Pro WebRTC answers received") + + first_session_lines, _ = _split_sdp_sections(next(iter(answers.values()))) + merged_media_lines: list[tuple[str, ...]] = [] + + for section in offer.media_sections: + answer_sdp = answers.get(section.kind) + if answer_sdp is None: + raise ValueError(f"Missing {section.kind} WebRTC answer") + + answer_session_lines, answer_media_sections = _split_sdp_sections(answer_sdp) + answer_media = next( + ( + media + for media in answer_media_sections + if _media_kind(media) == section.kind + ), + None, + ) + if answer_media is None: + raise ValueError(f"NanoKVM Pro {section.kind} answer has no media section") + + merged_media_lines.append( + _with_media_transport_lines( + _replace_mid(answer_media, section.mid), answer_session_lines + ) + ) + + return _join_sdp_sections( + _merged_session_lines(first_session_lines), tuple(merged_media_lines) + ) + + +def _split_sdp_sections( + sdp: str, +) -> tuple[tuple[str, ...], tuple[tuple[str, ...], ...]]: + """Split SDP into session-level lines and media sections.""" + session_lines: list[str] = [] + media_sections: list[list[str]] = [] + + for line in sdp.replace("\r\n", "\n").replace("\r", "\n").split("\n"): + if not line: + continue + if line.startswith("m="): + media_sections.append([line]) + elif media_sections: + media_sections[-1].append(line) + else: + session_lines.append(line) + + return tuple(session_lines), tuple(tuple(section) for section in media_sections) + + +def _join_sdp_sections( + session_lines: tuple[str, ...], media_sections: tuple[tuple[str, ...], ...] +) -> str: + """Join SDP lines using CRLF endings.""" + lines = list(session_lines) + for media_section in media_sections: + lines.extend(media_section) + return "\r\n".join(lines) + "\r\n" + + +def _media_kind(lines: tuple[str, ...]) -> MediaKind | None: + """Return the kind of an SDP media section, if supported.""" + if not lines or not lines[0].startswith("m="): + return None + kind = lines[0][2:].split(maxsplit=1)[0] + return kind if kind in _SUPPORTED_KINDS else None # type: ignore[return-value] + + +def _media_mid(lines: tuple[str, ...], *, fallback: str) -> str: + """Return the SDP media section MID.""" + for line in lines: + if line.startswith("a=mid:"): + return line.removeprefix("a=mid:") + return fallback + + +def _single_media_session_lines(session_lines: tuple[str, ...]) -> tuple[str, ...]: + """Return session lines for one single-section Pro peer.""" + filtered: list[str] = [] + for line in session_lines: + if not line.startswith("a=group:BUNDLE"): + filtered.append(line) + elif "a=group:BUNDLE 0" not in filtered: + filtered.append("a=group:BUNDLE 0") + return tuple(filtered) + + +def _merged_session_lines(session_lines: tuple[str, ...]) -> tuple[str, ...]: + """Drop session lines that belong to one Pro peer's transport.""" + return tuple( + line + for line in session_lines + if not line.startswith(("a=group:BUNDLE", *_TRANSPORT_SESSION_PREFIXES)) + ) + + +def _replace_mid(lines: tuple[str, ...], mid: str) -> tuple[str, ...]: + """Replace or add a media section MID.""" + if not any(line.startswith("a=mid:") for line in lines): + return (*lines, f"a=mid:{mid}") + return tuple( + f"a=mid:{mid}" if line.startswith("a=mid:") else line for line in lines + ) + + +def _with_media_transport_lines( + media_lines: tuple[str, ...], session_lines: tuple[str, ...] +) -> tuple[str, ...]: + """Move one answer's session transport attributes into its media section.""" + seen = { + prefix + for prefix in _TRANSPORT_SESSION_PREFIXES + if any(line.startswith(prefix) for line in media_lines) + } + result = list(media_lines) + for line in session_lines: + prefix = next( + (p for p in _TRANSPORT_SESSION_PREFIXES if line.startswith(p)), None + ) + if prefix is not None and prefix not in seen: + result.append(line) + seen.add(prefix) + return tuple(result) diff --git a/tests/test_stream.py b/tests/test_stream.py index 2807d3c..1da8157 100644 --- a/tests/test_stream.py +++ b/tests/test_stream.py @@ -6,10 +6,13 @@ from aiohttp import web from aioresponses import aioresponses from PIL import Image +import pytest from nanokvm.client import NanoKVMClient +from nanokvm.exceptions import NanoKVMError, NanoKVMNotSupportedError +from nanokvm.models import HWVersion -from .common import BASE_URL +from .common import BASE_URL, mark_detected def test_parse_jpeg_returns_fully_loaded_image() -> None: @@ -86,3 +89,37 @@ async def test_mjpeg_stream_bytes_yields_raw_frames(mocked: aioresponses) -> Non frames = [frame async for frame in c.mjpeg_stream_bytes()] assert frames == [b"\xff\xd8one", b"\xff\xd8two"] + + +_MJPEG_HEADERS = {"Content-Type": "multipart/x-mixed-replace; boundary=frame"} + + +async def test_get_snapshot_returns_first_frame(mocked: aioresponses) -> None: + """A snapshot is the first raw JPEG part of the stream.""" + mocked.get( + f"{BASE_URL}stream/mjpeg", + headers=_MJPEG_HEADERS, + body=b"--frame\r\nContent-Type: image/jpeg\r\n\r\n\xff\xd8one\r\n" + b"--frame\r\nContent-Type: image/jpeg\r\n\r\n\xff\xd8two\r\n--frame--\r\n", + ) + async with NanoKVMClient(BASE_URL, token="test-token") as c: + mark_detected(c) + assert await c.get_snapshot() == b"\xff\xd8one" + + +async def test_get_snapshot_without_frames_raises(mocked: aioresponses) -> None: + """A stream that ends before any frame is an error, not None.""" + mocked.get(f"{BASE_URL}stream/mjpeg", headers=_MJPEG_HEADERS, body=b"--frame--\r\n") + async with NanoKVMClient(BASE_URL, token="test-token") as c: + mark_detected(c) + with pytest.raises(NanoKVMError, match="without a frame"): + await c.get_snapshot() + + +async def test_get_snapshot_is_refused_on_the_pro(mocked: aioresponses) -> None: + """The Pro never opens MJPEG for a snapshot, so WebRTC stays up.""" + async with NanoKVMClient(BASE_URL, token="test-token") as c: + mark_detected(c, HWVersion.PRO) + with pytest.raises(NanoKVMNotSupportedError): + await c.get_snapshot() + assert not mocked.requests diff --git a/tests/test_sync_client.py b/tests/test_sync_client.py index 92b9efa..34389f8 100644 --- a/tests/test_sync_client.py +++ b/tests/test_sync_client.py @@ -163,7 +163,7 @@ def test_only_the_public_blocking_api_is_exposed(kvm: NanoKVMSyncClient) -> None """Private and async-only attributes are not proxied.""" names = dir(kvm) assert {"authenticate", "push_button", "mjpeg_stream", "run"} <= set(names) - assert "open_ws" not in names + assert {"open_ws", "open_webrtc_signaling"}.isdisjoint(names) with pytest.raises(AttributeError, match="_session"): kvm._session # noqa: B018 diff --git a/tests/test_webrtc.py b/tests/test_webrtc.py new file mode 100644 index 0000000..1670030 --- /dev/null +++ b/tests/test_webrtc.py @@ -0,0 +1,417 @@ +"""Tests for H.264 WebRTC signaling and the Pro offer splitting.""" + +from __future__ import annotations + +import asyncio +import json +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock + +from aiohttp import WSMsgType +import pytest + +from nanokvm.client import NanoKVMClient, NanoKVMVideoModeError +from nanokvm.models import HWVersion +from nanokvm.webrtc import ( + _TRANSPORT_SESSION_PREFIXES, + IceCandidate, + ProMediaSection, + ProWebRTCOffer, + WebRTCAnswer, + WebRTCCandidate, + WebRTCSignaling, + _media_kind, + _split_sdp_sections, + _status_code, + merge_pro_webrtc_answers, + split_pro_webrtc_offer, +) + +from .common import BASE_URL, mark_detected + +COMBINED_OFFER = "\r\n".join( + ( + "v=0", + "o=- 1 1 IN IP4 127.0.0.1", + "s=-", + "t=0 0", + "a=group:BUNDLE video-main audio-main data-main duplicate-video", + "a=group:BUNDLE ignored-duplicate", + "m=video 9 UDP/TLS/RTP/SAVPF 96", + "c=IN IP4 0.0.0.0", + "a=mid:video-main", + "a=sendrecv", + "m=audio 9 UDP/TLS/RTP/SAVPF 111", + "c=IN IP4 0.0.0.0", + "a=mid:audio-main", + "a=sendrecv", + "m=application 9 UDP/DTLS/SCTP webrtc-datachannel", + "a=mid:data-main", + "m=video 9 UDP/TLS/RTP/SAVPF 97", + "a=mid:duplicate-video", + "", + ) +) +VIDEO_ONLY_OFFER = "\r\n".join( + ("v=0", "a=group:BUNDLE camera", "m=video 9 RTP 96", "a=mid:camera", "") +) +VIDEO_ANSWER = "\r\n".join( + ( + "v=0", + "o=- 2 2 IN IP4 127.0.0.1", + "s=-", + "t=0 0", + "a=group:BUNDLE 0", + "a=ice-ufrag:video-session-ufrag", + "a=ice-pwd:video-session-pwd", + "a=ice-options:trickle", + "a=fingerprint:sha-256 VIDEO", + "a=setup:active", + "m=video 9 UDP/TLS/RTP/SAVPF 96", + "c=IN IP4 192.0.2.10", + "a=mid:0", + "a=ice-ufrag:video-media-ufrag", + "a=recvonly", + "", + ) +) +AUDIO_ANSWER = "\n".join( + ( + "v=0", + "o=- 3 3 IN IP4 127.0.0.1", + "s=-", + "t=0 0", + "a=ice-ufrag:audio-ufrag", + "a=ice-pwd:audio-pwd", + "a=fingerprint:sha-256 AUDIO", + "a=setup:active", + "m=audio 9 UDP/TLS/RTP/SAVPF 111", + "c=IN IP4 192.0.2.11", + "a=mid:0", + "a=recvonly", + "", + ) +) + + +class FakeWebSocket: + """A WebSocket that records sends and replays queued messages.""" + + def __init__(self, *messages: object) -> None: + self.sent: list[dict[str, Any]] = [] + self.closed = False + self.close_code: int | None = None + self.messages = list(messages) + self.send_error: Exception | None = None + + def __aiter__(self) -> FakeWebSocket: + return self + + async def __anext__(self) -> object: + if not self.messages: + raise StopAsyncIteration + return self.messages.pop(0) + + async def send_json(self, payload: dict[str, Any]) -> None: + if self.send_error is not None: + raise self.send_error + self.sent.append(payload) + + async def close(self) -> None: + self.closed = True + + def exception(self) -> None: + return None + + def sent_events(self) -> list[tuple[str, Any]]: + """Return each sent event with its JSON-decoded data.""" + return [ + (p["event"], json.loads(p["data"]) if p["data"] else p["data"]) + for p in self.sent + ] + + +def signal(event: str, data: object = "") -> SimpleNamespace: + """Build a device message, with the data JSON-encoded as the device does.""" + if not isinstance(data, str): + data = json.dumps(data) + return SimpleNamespace( + type=WSMsgType.TEXT, data=json.dumps({"event": event, "data": data}) + ) + + +async def events(signaling: WebRTCSignaling) -> list[object]: + return [event async for event in signaling] + + +def test_split_offer_keeps_first_supported_section_per_kind() -> None: + """Only the first video and audio sections become Pro offers.""" + offer = split_pro_webrtc_offer(COMBINED_OFFER) + + assert offer.expected_kinds == ("video", "audio") + assert [(s.mid, s.index) for s in offer.media_sections] == [ + ("video-main", 0), + ("audio-main", 1), + ] + for split_sdp in offer.offers.values(): + assert split_sdp.count("a=group:BUNDLE 0") == 1 + assert "ignored-duplicate" not in split_sdp + assert split_sdp.count("m=") == 1 + assert "a=mid:0\r\n" in split_sdp + assert "m=application" not in offer.offers["video"] + + +def test_split_offer_uses_media_index_when_mid_is_missing() -> None: + offer = split_pro_webrtc_offer("v=0\rm=audio 9 RTP/AVP 0\ra=sendrecv\r") + + assert offer.media_sections[0].mid == "0" + assert "a=mid:0\r\n" in offer.offers["audio"] + + +def test_candidate_routing_prefers_mid_then_index_then_all() -> None: + offer = split_pro_webrtc_offer(COMBINED_OFFER) + + def kinds(**kwargs: Any) -> tuple[str, ...]: + return offer.kinds_for_candidate(IceCandidate(candidate="c", **kwargs)) + + assert kinds(sdp_mid="audio-main", sdp_m_line_index=0) == ("audio",) + assert kinds(sdp_mid="unknown", sdp_m_line_index=0) == ("video",) + assert kinds(sdp_mid="unknown", sdp_m_line_index=99) == ("video", "audio") + + +def test_candidate_mapping_between_offer_and_pro_peers() -> None: + offer = split_pro_webrtc_offer(COMBINED_OFFER) + candidate = IceCandidate( + candidate="c", sdp_mid="x", sdp_m_line_index=5, username_fragment="u" + ) + + assert offer.upstream_candidate_for_kind("video", candidate) == IceCandidate( + candidate="c", sdp_mid="0", sdp_m_line_index=0, username_fragment="u" + ) + assert offer.client_candidate_for_kind("audio", candidate) == ( + IceCandidate( + candidate="c", + sdp_mid="audio-main", + sdp_m_line_index=1, + username_fragment="u", + ) + ) + empty = ProWebRTCOffer(offers={}, media_sections=()) + assert empty.upstream_candidate_for_kind("audio", candidate) is None + assert empty.client_candidate_for_kind("video", candidate) is None + + +def test_merge_answers_restores_mids_and_per_media_transport() -> None: + merged = merge_pro_webrtc_answers( + split_pro_webrtc_offer(COMBINED_OFFER), + {"audio": AUDIO_ANSWER, "video": VIDEO_ANSWER}, + ) + session_lines, media = _split_sdp_sections(merged) + + assert not any(line.startswith("a=group:BUNDLE") for line in session_lines) + assert not any( + line.startswith(_TRANSPORT_SESSION_PREFIXES) for line in session_lines + ) + assert [section[0].split()[0] for section in media] == ["m=video", "m=audio"] + assert "a=mid:video-main" in media[0] + assert "a=mid:audio-main" in media[1] + assert "a=ice-ufrag:video-media-ufrag" in media[0] + assert "a=ice-ufrag:video-session-ufrag" not in media[0] + assert "a=ice-pwd:video-session-pwd" in media[0] + assert "a=fingerprint:sha-256 VIDEO" in media[0] + assert "a=ice-ufrag:audio-ufrag" in media[1] + assert merged.endswith("\r\n") + + +def test_merge_rejects_missing_or_mislabeled_answers() -> None: + offer = split_pro_webrtc_offer(COMBINED_OFFER) + with pytest.raises(ValueError, match="No NanoKVM Pro WebRTC answers"): + merge_pro_webrtc_answers(offer, {}) + with pytest.raises(ValueError, match="Missing audio WebRTC answer"): + merge_pro_webrtc_answers(offer, {"video": VIDEO_ANSWER}) + with pytest.raises(ValueError, match="video answer has no media section"): + merge_pro_webrtc_answers( + ProWebRTCOffer( + offers={"video": "offer"}, + media_sections=(ProMediaSection("video", "v", 0, ("m=video 9",)),), + ), + {"video": AUDIO_ANSWER}, + ) + + +@pytest.mark.parametrize( + ("lines", "expected"), + [ + ((), None), + (("a=mid:0",), None), + (("m=application 9 UDP/DTLS/SCTP 5000",), None), + (("m=video 9 UDP/TLS/RTP/SAVPF 96",), "video"), + ], +) +def test_media_kind(lines: tuple[str, ...], expected: str | None) -> None: + assert _media_kind(lines) == expected + + +@pytest.mark.parametrize( + ("data", "raw", "expected"), + [ + ({"status": "-4"}, "", -4), + ({"code": 1}, "", 1), + (None, -1, -1), + (None, "-4", -4), + (None, '"1"', 1), + (None, "bad", None), + (None, True, None), + ], +) +def test_status_code_accepts_device_payload_variants( + data: dict[str, object] | None, raw: object, expected: int | None +) -> None: + assert _status_code(data, raw) == expected + + +async def test_non_pro_passes_offer_answer_and_candidates_through() -> None: + ws = FakeWebSocket( + SimpleNamespace(type=WSMsgType.BINARY, data=b""), + SimpleNamespace(type=WSMsgType.TEXT, data="not json"), + signal("heartbeat"), + signal("video-answer", {"type": "answer", "sdp": "answer-sdp"}), + signal("video-candidate", {"candidate": "c", "sdpMid": "0"}), + signal("video-candidate", {"sdpMid": "0"}), + signal("unknown", {}), + ) + signaling = WebRTCSignaling(ws, pro=False) # type: ignore[arg-type] + + await signaling.send_offer("offer-sdp") + await signaling.send_candidate( + IceCandidate(candidate="b", sdp_mid="0", username_fragment="u") + ) + + assert ws.sent_events() == [ + ("video-offer", {"type": "offer", "sdp": "offer-sdp"}), + ("video-candidate", {"candidate": "b", "sdpMid": "0", "usernameFragment": "u"}), + ] + assert await events(signaling) == [ + WebRTCAnswer("answer-sdp"), + WebRTCCandidate(IceCandidate(candidate="c", sdp_mid="0")), + ] + + +async def test_pro_splits_offer_and_merges_answers_and_candidates() -> None: + ws = FakeWebSocket( + signal("video-status", "1"), + signal("video-candidate", {"candidate": "early", "sdpMid": "0"}), + signal("video-answer", {"sdp": VIDEO_ANSWER}), + signal("audio-answer", {"sdp": AUDIO_ANSWER}), + signal("audio-answer", {"sdp": AUDIO_ANSWER}), + signal("audio-candidate", {"candidate": "late", "sdpMid": "0"}), + ) + signaling = WebRTCSignaling(ws, pro=True) # type: ignore[arg-type] + try: + with pytest.raises(RuntimeError, match="offer before candidates"): + await signaling.send_candidate(IceCandidate(candidate="c")) + await signaling.send_offer(COMBINED_OFFER) + await signaling.send_candidate( + IceCandidate(candidate="c", sdp_mid="audio-main") + ) + + sent = ws.sent_events() + assert [event for event, _ in sent] == [ + "video-offer", + "audio-offer", + "audio-candidate", + ] + assert sent[2][1] == {"candidate": "c", "sdpMid": "0", "sdpMLineIndex": 0} + + received = await events(signaling) + finally: + await signaling.close() + + assert len(received) == 3 + assert isinstance(received[0], WebRTCAnswer) + assert "a=mid:audio-main" in received[0].sdp + assert received[1:] == [ + WebRTCCandidate( + IceCandidate(candidate="early", sdp_mid="video-main", sdp_m_line_index=0) + ), + WebRTCCandidate( + IceCandidate(candidate="late", sdp_mid="audio-main", sdp_m_line_index=1) + ), + ] + + +async def test_pro_video_only_offer_answers_after_one_answer() -> None: + ws = FakeWebSocket(signal("video-answer", {"sdp": VIDEO_ANSWER})) + async with WebRTCSignaling(ws, pro=True) as signaling: # type: ignore[arg-type] + await signaling.send_offer(VIDEO_ONLY_OFFER) + (answer,) = await events(signaling) + + assert [event for event, _ in ws.sent_events()] == ["video-offer"] + assert isinstance(answer, WebRTCAnswer) + assert "a=mid:camera" in answer.sdp + assert ws.closed + + +async def test_pro_rejects_offer_without_media() -> None: + async with WebRTCSignaling(FakeWebSocket(), pro=True) as signaling: # type: ignore[arg-type] + with pytest.raises(ValueError, match="does not contain"): + await signaling.send_offer("v=0\r\nm=application 9 SCTP 5000\r\n") + + +async def test_pro_inconsistent_video_mode_raises() -> None: + ws = FakeWebSocket(signal("video-status", "-4")) + async with WebRTCSignaling(ws, pro=True) as signaling: # type: ignore[arg-type] + with pytest.raises(NanoKVMVideoModeError): + await events(signaling) + + +async def test_pro_heartbeat_failure_closes_the_websocket( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr("nanokvm.webrtc._PRO_HEARTBEAT_SECONDS", 0) + ws = FakeWebSocket() + signaling = WebRTCSignaling(ws, pro=True) # type: ignore[arg-type] + await asyncio.sleep(0.01) + assert ws.sent[0] == {"event": "heartbeat", "data": ""} + + ws.send_error = ConnectionError("gone") + async with asyncio.timeout(1): + while not ws.closed: + await asyncio.sleep(0) + await signaling.close() + + +@pytest.mark.parametrize( + ("hw_version", "path", "heartbeat"), + [ + (HWVersion.PCIE, "stream/h264", 30.0), + (HWVersion.PRO, "stream/h264/webrtc", None), + ], +) +async def test_open_webrtc_signaling_picks_the_endpoint_by_hardware( + client: NanoKVMClient, hw_version: HWVersion, path: str, heartbeat: float | None +) -> None: + mark_detected(client, hw_version) + ws = FakeWebSocket() + client._session.open_ws = AsyncMock(return_value=ws) # type: ignore[method-assign] + + async with await client.open_webrtc_signaling() as signaling: + assert signaling.websocket is ws # type: ignore[comparison-overlap] + + args, kwargs = client._session.open_ws.call_args + assert args == (path,) + assert kwargs["heartbeat"] == heartbeat + + +async def test_open_webrtc_signaling_detects_hardware_first() -> None: + async with NanoKVMClient(BASE_URL, token="t") as client: + client._session.detect_hardware = AsyncMock( # type: ignore[method-assign] + side_effect=lambda: mark_detected(client, HWVersion.PRO) + ) + client._session.open_ws = AsyncMock(return_value=FakeWebSocket()) # type: ignore[method-assign] + + signaling = await client.open_webrtc_signaling() + await signaling.close() + + assert client._session.open_ws.call_args.args == ("stream/h264/webrtc",)