From bb15254cc44181293b22de4c6f768ba266926936 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Tue, 20 Oct 2026 13:07:52 -0300 Subject: [PATCH 1/4] Add H.264 WebRTC signaling with the Pro offer split and answer merge open_webrtc_signaling() picks the signaling endpoint for the detected hardware and returns a WebRTCSignaling session: one offer in, one answer and trickled candidates out, on either family. For the Pro it splits the browser offer into its separate video and audio peers, merges their answers, maps candidates both ways and keeps the session alive with the application heartbeat. A Pro video-status of -4 raises NanoKVMVideoModeError. Moved from the Home Assistant integration. --- docs/api.md | 20 +- docs/errors.md | 1 + nanokvm/client.py | 2 + nanokvm/components/stream.py | 22 ++ nanokvm/exceptions.py | 4 + nanokvm/webrtc.py | 529 +++++++++++++++++++++++++++++++++++ tests/test_webrtc.py | 417 +++++++++++++++++++++++++++ 7 files changed, 994 insertions(+), 1 deletion(-) create mode 100644 nanokvm/webrtc.py create mode 100644 tests/test_webrtc.py diff --git a/docs/api.md b/docs/api.md index a50a64b..96a25e4 100644 --- a/docs/api.md +++ b/docs/api.md @@ -117,7 +117,7 @@ while loop.time() < end: | Methods | Requires | |---|---| -| `mjpeg_stream`, `mjpeg_stream_bytes`, `open_ws` | all | +| `mjpeg_stream`, `mjpeg_stream_bytes`, `open_ws`, `open_webrtc_signaling` | all | | `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 | @@ -135,6 +135,24 @@ 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. +`open_webrtc_signaling()` opens the H.264 WebRTC signaling session for the +detected hardware and returns a `nanokvm.webrtc.WebRTCSignaling`. Send the +browser's offer with `send_offer(sdp)` and its candidates with +`send_candidate(IceCandidate(...))`, and iterate it for `WebRTCAnswer` and +`WebRTCCandidate` events. The Pro answers audio and video on separate peer +connections; the session splits the offer and merges the answers, so the +caller sees one offer and one answer on any hardware. Iteration raises +`NanoKVMVideoModeError` when the Pro drops WebRTC because another video mode +took the encoder, for example an MJPEG stream. Close it, or use it as an +async context manager, before the client exits. + +```python +async with await client.open_webrtc_signaling() as signaling: + await signaling.send_offer(offer_sdp) + async for event in signaling: + ... # forward WebRTCAnswer.sdp and WebRTCCandidate.candidate +``` + ### Images and scripts | Methods | Requires | 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/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..f11354a 100644 --- a/nanokvm/components/stream.py +++ b/nanokvm/components/stream.py @@ -13,10 +13,15 @@ from aiohttp import BodyPartReader, ClientWebSocketResponse, MultipartReader, hdrs from PIL import Image +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.""" @@ -70,3 +75,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/webrtc.py b/nanokvm/webrtc.py new file mode 100644 index 0000000..ac835d7 --- /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.home_assistant_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.home_assistant_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 home_assistant_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_webrtc.py b/tests/test_webrtc.py new file mode 100644 index 0000000..508bb2c --- /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.home_assistant_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.home_assistant_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",) From cdbf5a4248eafd5b1a8b6dbcd62909af9a07f1f5 Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Tue, 27 Oct 2026 23:05:46 -0300 Subject: [PATCH 2/4] Keep WebRTC signaling async-only and drop the Home Assistant name open_webrtc_signaling() joins open_ws() in the sync client's async-only set: its session belongs to the client's event loop. Rename home_assistant_candidate_for_kind to client_candidate_for_kind, the counterpart of upstream_candidate_for_kind. --- docs/sync.md | 4 ++-- nanokvm/sync_client.py | 2 +- nanokvm/webrtc.py | 6 +++--- tests/test_sync_client.py | 2 +- tests/test_webrtc.py | 4 ++-- 5 files changed, 9 insertions(+), 9 deletions(-) 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/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 index ac835d7..26185ed 100644 --- a/nanokvm/webrtc.py +++ b/nanokvm/webrtc.py @@ -206,7 +206,7 @@ async def _events(self) -> AsyncGenerator[WebRTCEvent, None]: answered = True yield WebRTCAnswer(merge_pro_webrtc_answers(offer, answers)) for early_kind, early_candidate in early: - if mapped := offer.home_assistant_candidate_for_kind( + if mapped := offer.client_candidate_for_kind( early_kind, early_candidate ): yield WebRTCCandidate(mapped) @@ -218,7 +218,7 @@ async def _events(self) -> AsyncGenerator[WebRTCEvent, None]: continue if not answered: early.append((kind, candidate)) - elif mapped := offer.home_assistant_candidate_for_kind(kind, candidate): + elif mapped := offer.client_candidate_for_kind(kind, candidate): yield WebRTCCandidate(mapped) else: _LOGGER.debug("Unhandled Pro WebRTC event: %s", event) @@ -358,7 +358,7 @@ def upstream_candidate_for_kind( return None return candidate.model_copy(update={"sdp_mid": "0", "sdp_m_line_index": 0}) - def home_assistant_candidate_for_kind( + def client_candidate_for_kind( self, kind: MediaKind, candidate: IceCandidate ) -> IceCandidate | None: """Map a Pro candidate back onto its section of the original offer.""" 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 index 508bb2c..1670030 100644 --- a/tests/test_webrtc.py +++ b/tests/test_webrtc.py @@ -189,7 +189,7 @@ def test_candidate_mapping_between_offer_and_pro_peers() -> None: assert offer.upstream_candidate_for_kind("video", candidate) == IceCandidate( candidate="c", sdp_mid="0", sdp_m_line_index=0, username_fragment="u" ) - assert offer.home_assistant_candidate_for_kind("audio", candidate) == ( + assert offer.client_candidate_for_kind("audio", candidate) == ( IceCandidate( candidate="c", sdp_mid="audio-main", @@ -199,7 +199,7 @@ def test_candidate_mapping_between_offer_and_pro_peers() -> None: ) empty = ProWebRTCOffer(offers={}, media_sections=()) assert empty.upstream_candidate_for_kind("audio", candidate) is None - assert empty.home_assistant_candidate_for_kind("video", candidate) is None + assert empty.client_candidate_for_kind("video", candidate) is None def test_merge_answers_restores_mids_and_per_media_transport() -> None: From 0294439a6010e911ec5e53403e28ad044af4da9c Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Tue, 27 Oct 2026 23:46:32 -0300 Subject: [PATCH 3/4] Add get_snapshot for one MJPEG frame, refused on the Pro get_snapshot() returns the first raw JPEG of the MJPEG stream within a timeout. The Pro refuses it: its MJPEG and H.264 streams share one encoder mode, so opening MJPEG would interrupt WebRTC (NanoKVM-Pro#157). Moved from the Home Assistant integration's camera snapshot. --- docs/api.md | 9 +++++---- nanokvm/components/stream.py | 18 +++++++++++++++++ tests/test_stream.py | 39 +++++++++++++++++++++++++++++++++++- 3 files changed, 61 insertions(+), 5 deletions(-) diff --git a/docs/api.md b/docs/api.md index 96a25e4..3b04554 100644 --- a/docs/api.md +++ b/docs/api.md @@ -118,7 +118,7 @@ while loop.time() < end: | Methods | Requires | |---|---| | `mjpeg_stream`, `mjpeg_stream_bytes`, `open_ws`, `open_webrtc_signaling` | all | -| `set_screen`, `set_frame_detect`, `pause_frame_detect` | non-Pro | +| `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 | @@ -131,9 +131,10 @@ while loop.time() < end: | `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. +yields the raw JPEG bytes. `get_snapshot(timeout=20.0)` returns the first raw +JPEG frame; the Pro refuses it, because opening MJPEG there interrupts WebRTC. +`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. `open_webrtc_signaling()` opens the H.264 WebRTC signaling session for the detected hardware and returns a `nanokvm.webrtc.WebRTCSignaling`. Send the diff --git a/nanokvm/components/stream.py b/nanokvm/components/stream.py index f11354a..b16843d 100644 --- a/nanokvm/components/stream.py +++ b/nanokvm/components/stream.py @@ -13,6 +13,8 @@ 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 @@ -53,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 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 From 15efdd085af0861ec78c05751637153b6c2dd84b Mon Sep 17 00:00:00 2001 From: Raphael Chicon Date: Tue, 27 Oct 2026 23:49:05 -0300 Subject: [PATCH 4/4] Move the streaming docs to a dedicated guide docs/streaming.md covers snapshots, the MJPEG streams and the WebRTC signaling session step by step, with the Pro's single encoder and its offer split and answer merge. The API reference keeps the method table and points to it. --- README.md | 1 + docs/api.md | 30 ++++---------- docs/streaming.md | 102 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 111 insertions(+), 22 deletions(-) create mode 100644 docs/streaming.md 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 3b04554..b394d11 100644 --- a/docs/api.md +++ b/docs/api.md @@ -130,29 +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. `get_snapshot(timeout=20.0)` returns the first raw -JPEG frame; the Pro refuses it, because opening MJPEG there interrupts WebRTC. -`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. - -`open_webrtc_signaling()` opens the H.264 WebRTC signaling session for the -detected hardware and returns a `nanokvm.webrtc.WebRTCSignaling`. Send the -browser's offer with `send_offer(sdp)` and its candidates with -`send_candidate(IceCandidate(...))`, and iterate it for `WebRTCAnswer` and -`WebRTCCandidate` events. The Pro answers audio and video on separate peer -connections; the session splits the offer and merges the answers, so the -caller sees one offer and one answer on any hardware. Iteration raises -`NanoKVMVideoModeError` when the Pro drops WebRTC because another video mode -took the encoder, for example an MJPEG stream. Close it, or use it as an -async context manager, 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. -```python -async with await client.open_webrtc_signaling() as signaling: - await signaling.send_offer(offer_sdp) - async for event in signaling: - ... # forward WebRTCAnswer.sdp and WebRTCCandidate.candidate -``` +`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/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).