From ffcf396fe12531da3f37e56fbad2b66955278b18 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 17 Feb 2026 22:49:36 +0000 Subject: [PATCH 01/11] Add 2-way audio, RTSP VOD playback, and Baichuan talk protocol support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Two-way audio (talk) via Baichuan protocol: * cmd_id 201 (TalkAbility): query duplex mode and audio config from camera * cmd_id 202 (Talk): send IMA ADPCM audio data without encryption * cmd_id 203 (TalkConfig): configure and start a talk session * cmd_id 204 (TalkReset): stop an active talk session * Baichuan.build_bcmedia_adpcm() helper to wrap ADPCM blocks in BcMedia framing * send_binary_no_reply() internal method for fire-and-forget binary payloads * Capability detection updated to use cmd_id=201 as the authoritative check - Public API additions on Host: * two_way_audio_support(channel) – bool capability check * start_talk(channel) – starts session, returns audio config dict * send_talk_data(channel, bcmedia_data) – sends BcMedia-framed ADPCM audio * stop_talk(channel) – stops session - RTSP VOD replay support: * VodRequestType.RTSP added to enums * get_vod_source(..., request_type=VodRequestType.RTSP) returns an RTSP URL for cameras/NVRs that support RTSP playback of recorded files - XML templates for TalkAbility and TalkConfig added to baichuan/xmls.py References: apocaliss92/nodelink-js (cmd_id 201-204), QuantumEntangledAndy/neolink PR#396 https://claude.ai/code/session_0116gKmpLmMXRwA5DM8VBzfc --- reolink_aio/api.py | 58 ++++++++++ reolink_aio/baichuan/baichuan.py | 179 ++++++++++++++++++++++++++++++- reolink_aio/baichuan/xmls.py | 22 ++++ reolink_aio/enums.py | 1 + 4 files changed, 259 insertions(+), 1 deletion(-) diff --git a/reolink_aio/api.py b/reolink_aio/api.py index 79a71f81..88aff0a3 100644 --- a/reolink_aio/api.py +++ b/reolink_aio/api.py @@ -3542,6 +3542,12 @@ async def get_vod_source( cmd = VodRequestType.DOWNLOAD.value url = f"{self._url}?cmd={cmd}&source={filename.replace(' ', '%20')}&output=ha_playback_{time_start}.mp4{start_time}" + elif request_type == VodRequestType.RTSP: + # RTSP playback URL – supported on select Reolink cameras/NVRs. + # The credentials are embedded in the URL (same as live RTSP streams). + safe_filename = filename.replace(" ", "%20") + rtsp_url = f"rtsp://{self._username}:{self._enc_password}@{self._host}:{self._rtsp_port}/vod/{safe_filename}" + return ("video/mp4", rtsp_url) else: raise InvalidParameterError(f"get_vod_source: unsupported request_type '{request_type.value}'") @@ -5356,6 +5362,58 @@ async def set_audio_alarm(self, channel: int, enable: bool) -> None: await self.send_setting(body) + # ------------------------------------------------------------------------- + # Two-way audio (talk) public API + # ------------------------------------------------------------------------- + + def two_way_audio_support(self, channel: int) -> bool: + """Return True if the camera channel supports two-way audio (talk).""" + return self.supported(channel, "two_way_audio") + + async def start_talk(self, channel: int) -> dict: + """Start a two-way audio (talk) session on the camera channel. + + Returns a dict describing the audio format the camera expects: + - audio_type (str): codec name, e.g. "adpcm" + - sample_rate (int): samples per second, e.g. 8000 or 16000 + - sample_precision (int): bit depth, e.g. 16 + - length_per_encoder (int): samples per block, e.g. 320 or 640 + - sound_track (str): channel layout, e.g. "mono" + - duplex (str): duplex mode, e.g. "FDX" + - audio_stream_mode (str): e.g. "followVideoStream" + + The caller should encode microphone audio as IMA ADPCM blocks with the + parameters above, wrap each block using + ``Baichuan.build_bcmedia_adpcm([block])``, and send the result with + ``send_talk_data()``. Call ``stop_talk()`` when finished. + + Raises NotSupportedError if two-way audio is not supported. + """ + if channel not in self._channels: + raise InvalidParameterError(f"start_talk: no camera connected to channel '{channel}'") + if not self.two_way_audio_support(channel): + raise NotSupportedError(f"start_talk: Two-way audio is not supported on {self.camera_name(channel)}") + + return await self.baichuan.start_talk(channel) + + async def send_talk_data(self, channel: int, bcmedia_data: bytes) -> None: + """Send BcMedia-framed IMA ADPCM audio data to the camera. + + ``bcmedia_data`` must be produced by ``Baichuan.build_bcmedia_adpcm()``. + Call ``start_talk()`` before the first call to this method. + """ + if channel not in self._channels: + raise InvalidParameterError(f"send_talk_data: no camera connected to channel '{channel}'") + + await self.baichuan.send_talk_data(channel, bcmedia_data) + + async def stop_talk(self, channel: int) -> None: + """Stop the two-way audio (talk) session on the camera channel.""" + if channel not in self._channels: + raise InvalidParameterError(f"stop_talk: no camera connected to channel '{channel}'") + + await self.baichuan.stop_talk(channel) + async def set_siren(self, channel: int | None = None, enable: bool = True, duration: int | None = 2) -> None: if channel not in self._channels and channel is not None: raise InvalidParameterError(f"set_siren: no camera connected to channel '{channel}'") diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index 5e926a34..72e53657 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -437,6 +437,45 @@ async def send_payload( return (rec_body, payload) + async def send_binary_no_reply( + self, + cmd_id: int, + channel: int | None = None, + binary_body: bytes = b"", + ) -> None: + """Send a binary payload without waiting for a reply (used for audio talk data).""" + if not self._logged_in and cmd_id > 2: + await self.login() + + if channel is None: + ch_id = 250 + else: + ch_id = channel + 1 + + ext = xmls.CHANNEL_EXTENSION_XML.format(channel=channel) if channel is not None else "" + + # Extension is AES-encrypted; binary body is sent as-is (NOT encrypted) + enc_ext = self._aes_encrypt(ext.encode("utf-8")) + mess_len = len(enc_ext) + len(binary_body) + payload_offset = len(enc_ext) + + self._mess_id = (self._mess_id + 1) % 16777216 + + cmd_id_bytes = (cmd_id).to_bytes(4, byteorder="little") + mess_len_bytes = (mess_len).to_bytes(4, byteorder="little") + mess_id_bytes = (ch_id).to_bytes(1, byteorder="little") + (self._mess_id).to_bytes(3, byteorder="little") + payload_offset_bytes = (payload_offset).to_bytes(4, byteorder="little") + status_code = "0000" + header = bytes.fromhex(HEADER_MAGIC) + cmd_id_bytes + mess_len_bytes + mess_id_bytes + bytes.fromhex(status_code + "1464") + payload_offset_bytes + + await self._connect_if_needed() + if TYPE_CHECKING: + assert self._transport is not None + + _LOGGER.debug("Baichuan host %s: writing binary no-reply cmd_id %s, binary length %s", self._host, cmd_id, len(binary_body)) + async with self._mutex: + self._transport.write(header + enc_ext + binary_body) + def _aes_encrypt(self, body: bytes) -> bytes: """Encrypt a message using AES encryption""" if not body: @@ -1470,6 +1509,7 @@ async def get_channel_data(self) -> None: if self.http_api.api_version("talk", channel) > 0: coroutines.append((10, channel, self.send(cmd_id=10, channel=channel))) + coroutines.append((201, channel, self.get_talk_ability(channel))) if (self.http_api.is_nvr or self.privacy_mode() is not None) and self.api_version("remoteAbility", channel) > 0: coroutines.append(("privacy_mode", channel, self.get_privacy_mode(channel))) # capability added in get_privacy_mode @@ -1615,11 +1655,15 @@ async def get_channel_data(self) -> None: if isinstance(result, BaseException): raise result - if cmd_id == 10: # two way audio + if cmd_id == 10: # two way audio (legacy check via audioStreamMode) root = XML.fromstring(result) for audio in root.findall(".//audioStreamMode"): if audio.text == "mixAudioStream": self.capabilities[channel].add("two_way_audio") + if cmd_id == 201: # TalkAbility (cmd_id=201) – authoritative two_way_audio check + # result is already a dict returned by get_talk_ability() + if isinstance(result, dict) and result.get("duplex"): + self.capabilities[channel].add("two_way_audio") if cmd_id == 483: # hardwired chime self.capabilities[channel].add("hardwired_chime") if cmd_id == 527: # crossline detection @@ -3785,3 +3829,136 @@ def siren_state(self, channel: int) -> bool | None: def audio_noise_reduction(self, channel: int) -> int | None: return self._noise_reduction.get(channel) + + # ------------------------------------------------------------------------- + # Two-way audio (talk) support + # cmd_id 201 = TalkAbility (query) + # cmd_id 202 = Talk (send ADPCM audio data, no reply expected) + # cmd_id 203 = TalkConfig (start/configure session) + # cmd_id 204 = TalkReset (stop session) + # ------------------------------------------------------------------------- + + BCMEDIA_ADPCM_MAGIC = b"bw10" # 0x62773130 + + async def get_talk_ability(self, channel: int) -> dict: + """Query the camera's talk (2-way audio) capability via Baichuan cmd_id=201. + + Returns a dict with keys: + - duplex: str (e.g. "FDX") + - audio_stream_mode: str (e.g. "followVideoStream" or "mixAudioStream") + - audio_type: str (e.g. "adpcm") + - sample_rate: int (e.g. 8000 or 16000) + - sample_precision: int (e.g. 16) + - length_per_encoder: int (samples per block, e.g. 320 or 640) + - sound_track: str (e.g. "mono") + """ + mess = await self.send(cmd_id=201, channel=channel, body=xmls.TalkAbilityGet) + root = XML.fromstring(mess) + + ability: dict = {} + + # Duplex mode + for elem in root.findall(".//duplex"): + if elem.text: + ability["duplex"] = elem.text + break + + # Audio stream mode + for elem in root.findall(".//audioStreamMode"): + if elem.text: + ability["audio_stream_mode"] = elem.text + break + + # Audio config (take first supported entry) + for cfg in root.findall(".//audioConfig"): + audio_type_elem = cfg.find("audioType") + sample_rate_elem = cfg.find("sampleRate") + sample_precision_elem = cfg.find("samplePrecision") + lpe_elem = cfg.find("lengthPerEncoder") + sound_track_elem = cfg.find("soundTrack") + + if audio_type_elem is not None and audio_type_elem.text: + ability.setdefault("audio_type", audio_type_elem.text) + if sample_rate_elem is not None and sample_rate_elem.text: + ability.setdefault("sample_rate", int(sample_rate_elem.text)) + if sample_precision_elem is not None and sample_precision_elem.text: + ability.setdefault("sample_precision", int(sample_precision_elem.text)) + if lpe_elem is not None and lpe_elem.text: + ability.setdefault("length_per_encoder", int(lpe_elem.text)) + if sound_track_elem is not None and sound_track_elem.text: + ability.setdefault("sound_track", sound_track_elem.text) + + return ability + + async def start_talk(self, channel: int) -> dict: + """Start a 2-way audio talk session via Baichuan cmd_id=203. + + Queries the camera's TalkAbility, then sends a TalkConfig message to + initiate the session. Returns the audio configuration dict (same + structure as get_talk_ability) so the caller knows which format to use + when encoding audio for send_talk_data(). + + Audio format used: ADPCM (DVI-4 / IMA ADPCM), 16-bit, mono. + Block size in bytes = (length_per_encoder // 2) + 4 + """ + ability = await self.get_talk_ability(channel) + + duplex = ability.get("duplex", "FDX") + audio_stream_mode = ability.get("audio_stream_mode", "followVideoStream") + audio_type = ability.get("audio_type", "adpcm") + sample_rate = ability.get("sample_rate", 8000) + sample_precision = ability.get("sample_precision", 16) + length_per_encoder = ability.get("length_per_encoder", 320) + sound_track = ability.get("sound_track", "mono") + + xml = xmls.TalkConfigSet.format( + channel=channel, + duplex=duplex, + audio_stream_mode=audio_stream_mode, + audio_type=audio_type, + sample_rate=sample_rate, + sample_precision=sample_precision, + length_per_encoder=length_per_encoder, + sound_track=sound_track, + ) + await self.send(cmd_id=203, channel=channel, body=xml) + + _LOGGER.debug( + "Baichuan host %s ch %s: talk session started, audio_type=%s sample_rate=%s length_per_encoder=%s", + self._host, + channel, + audio_type, + sample_rate, + length_per_encoder, + ) + return ability + + async def stop_talk(self, channel: int) -> None: + """Stop the 2-way audio talk session via Baichuan cmd_id=204.""" + await self.send(cmd_id=204, channel=channel) + _LOGGER.debug("Baichuan host %s ch %s: talk session stopped", self._host, channel) + + @staticmethod + def build_bcmedia_adpcm(adpcm_blocks: list[bytes]) -> bytes: + """Wrap one or more raw IMA-ADPCM blocks in BcMedia ADPCM framing. + + Each block must be a complete DVI-4 (IMA ADPCM) block: + - 4-byte block header (initial predictor LE-16, step index, reserved) + - followed by (length_per_encoder // 2) nibble-packed sample bytes. + + Up to 4 blocks may be combined into one BcMedia message. + The returned bytes can be passed directly to send_talk_data(). + """ + payload = b"" + for block in adpcm_blocks: + payload += Baichuan.BCMEDIA_ADPCM_MAGIC + block + return payload + + async def send_talk_data(self, channel: int, bcmedia_data: bytes) -> None: + """Send BcMedia-framed ADPCM audio data to the camera (cmd_id=202). + + bcmedia_data should be the output of build_bcmedia_adpcm(). + The audio payload is sent WITHOUT encryption as required by the protocol. + This method does not wait for a reply. + """ + await self.send_binary_no_reply(cmd_id=202, channel=channel, binary_body=bcmedia_data) diff --git a/reolink_aio/baichuan/xmls.py b/reolink_aio/baichuan/xmls.py index f3e182be..ec096d34 100644 --- a/reolink_aio/baichuan/xmls.py +++ b/reolink_aio/baichuan/xmls.py @@ -369,6 +369,28 @@ """ +TalkAbilityGet = """ + + + +""" + +TalkConfigSet = """ + + +{channel} +{duplex} +{audio_stream_mode} + +{audio_type} +{sample_rate} +{sample_precision} +{length_per_encoder} +{sound_track} + + +""" + SetAutoFocus = """ diff --git a/reolink_aio/enums.py b/reolink_aio/enums.py index 8e298abb..1b3f7e33 100644 --- a/reolink_aio/enums.py +++ b/reolink_aio/enums.py @@ -25,6 +25,7 @@ class VodRequestType(Enum): FLV = "FLV" DOWNLOAD = "Download" NVR_DOWNLOAD = "NvrDownload" + RTSP = "RTSP" class EncodingEnum(Enum): From 295088ccd81a17f34f3223ab318648cb44263533 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 17 Feb 2026 23:21:52 +0000 Subject: [PATCH 02/11] Add get_recording_days() and get_recordings_for_day() for HA media browser MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds two convenience wrappers over request_vod_files() that make it easy to implement Home Assistant's media browser for recorded clips: - get_recording_days(channel, year, month) -> set[int] Returns the set of day-numbers (1–31) in the given month that contain at least one recording. Intended for calendar/month navigation in the media browser. Backed by request_vod_files(status_only=True). - get_recordings_for_day(channel, day, stream=None, trigger=None) -> list[VOD_file] Returns all recordings for the given calendar date, sorted chronologically and deduplicated (a file matching multiple triggers only appears once). Supports optional trigger filter (VOD_trigger.MOTION, PERSON, etc.). Each returned VOD_file exposes: - start_time / end_time / duration - triggers (VOD_trigger flags: MOTION, PERSON, VEHICLE, …) - file_name → pass to get_vod_source() for playback URL - size Typical HA media-browser flow: 1. days = await host.get_recording_days(ch, year, month) # calendar 2. files = await host.get_recordings_for_day(ch, date(...)) # day listing 3. mime, url = await host.get_vod_source(ch, file.file_name) # playback https://claude.ai/code/session_0116gKmpLmMXRwA5DM8VBzfc --- reolink_aio/api.py | 91 +++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 90 insertions(+), 1 deletion(-) diff --git a/reolink_aio/api.py b/reolink_aio/api.py index 88aff0a3..5ed9ec8c 100644 --- a/reolink_aio/api.py +++ b/reolink_aio/api.py @@ -9,7 +9,7 @@ import re import ssl import traceback -from datetime import datetime, timedelta, tzinfo +from datetime import date, datetime, timedelta, tzinfo from io import BytesIO from math import ceil from os.path import basename @@ -5851,6 +5851,95 @@ async def request_vod_files( return statuses, vod_files + async def get_recording_days(self, channel: int, year: int, month: int) -> set[int]: + """Return the set of day-numbers (1–31) in *year*/*month* that have recordings. + + Convenience wrapper around request_vod_files(status_only=True). + Useful for populating a calendar view in Home Assistant's media browser. + + Example:: + + days = await host.get_recording_days(0, 2024, 6) + # {1, 3, 14, 15, 28} → recordings exist on those days + """ + if channel not in self._stream_channels: + raise InvalidParameterError(f"get_recording_days: no camera connected to channel '{channel}'") + + import calendar + + last_day = calendar.monthrange(year, month)[1] + start = datetime(year, month, 1, 0, 0, 0) + end = datetime(year, month, last_day, 23, 59, 59) + + statuses, _ = await self.request_vod_files(channel, start, end, status_only=True) + + days: set[int] = set() + for status in statuses: + if status.year == year and status.month == month: + days.update(status.days) + return days + + async def get_recordings_for_day( + self, + channel: int, + day: date, + stream: Optional[str] = None, + trigger: typings.VOD_trigger | None = None, + ) -> list[typings.VOD_file]: + """Return recordings for *channel* on the given *day*, sorted by start time. + + Recordings are deduplicated so that the same file only appears once + even when it matches multiple detection triggers. + + Parameters + ---------- + channel: + Camera channel index. + day: + The calendar date to query (e.g. ``date(2024, 6, 14)``). + stream: + Stream type (``"main"``, ``"sub"``, …). Defaults to the host default. + trigger: + Optional filter. When given only recordings matching that + ``VOD_trigger`` flag are returned. + + Returns + ------- + list[VOD_file] + Sorted (by start_time), deduplicated list. Each item exposes: + + * ``file.start_time`` / ``file.end_time`` – datetime with tz + * ``file.duration`` – timedelta + * ``file.triggers`` – VOD_trigger flags (motion, person, …) + * ``file.file_name`` – filename for use with get_vod_source() + * ``file.size`` – file size in bytes + + Obtain a playback URL with:: + + mime, url = await host.get_vod_source(channel, file.file_name) + """ + if channel not in self._stream_channels: + raise InvalidParameterError(f"get_recordings_for_day: no camera connected to channel '{channel}'") + + start = datetime(day.year, day.month, day.day, 0, 0, 0) + end = datetime(day.year, day.month, day.day, 23, 59, 59) + + _, vod_files = await self.request_vod_files( + channel, start, end, status_only=False, stream=stream, trigger=trigger + ) + + # Deduplicate by file_name then sort chronologically + seen: set[str] = set() + unique: list[typings.VOD_file] = [] + for f in vod_files: + key = f.file_name + if key not in seen: + seen.add(key) + unique.append(f) + + unique.sort(key=lambda f: f.start_time) + return unique + async def send_setting(self, body: typings.reolink_json, wait_before_get: int = 0, getcmd: str = "") -> None: command = body[0]["cmd"] _LOGGER.debug( From 5fa81e720dc2cfd3782f4c152cf5967e7433c9a1 Mon Sep 17 00:00:00 2001 From: lorek123 Date: Mon, 23 Feb 2026 21:26:38 +0100 Subject: [PATCH 03/11] Add Baichuan VOD streaming and reolink-stream-vod go2rtc bridge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds raw H264 VOD streaming over the Baichuan TCP protocol for baichuan_only cameras that lack HTTP download or RTSP VOD support. New public API on Host: - stream_recording_bc(channel, file_name, start_time, stream_type) Async generator yielding (timestamp_us, h264_annex_b_bytes) tuples. Authenticates lazily via the Baichuan TCP connection — no HTTP get_host_data()/get_states() calls needed (~0.4 s startup). New Baichuan methods: - search_recording_days_bc() — days with recordings in a given month - search_recordings_for_day_bc() — file list for a given day - replay_seek_bc() — seek to a position in an open replay session - stream_replay_bc() — sends StartSearch/StartPlay commands, yields raw decrypted binary chunks from the camera - parse_bcmedia_frames() — parses BcMedia binary framing (I/P-frame video, AAC/ADPCM audio, Info packets), extracts H264 Annex-B payloads with microsecond timestamps; uses 8-byte-aligned scanning to avoid false-positive magic matches inside H264 payload data New XML templates in xmls.py: DayRecords, ReplaySeek, ReplayStart, ReplayStop. reolink-stream-vod console script (reolink_aio/scripts/stream_vod.py): Intended as a go2rtc exec: source. Rate-limits H264 delivery using BcMedia microsecond timestamps so go2rtc wall-clock RTP timestamps produce correct playback speed. Handles u32 timestamp wraparound (~71 min recordings). Usage: reolink-stream-vod HOST USER PASS CHANNEL FILE_NAME START_TIME Co-Authored-By: Claude Sonnet 4.6 --- reolink_aio/api.py | 46 +++ reolink_aio/baichuan/baichuan.py | 430 +++++++++++++++++++++++++-- reolink_aio/baichuan/tcp_protocol.py | 43 ++- reolink_aio/baichuan/xmls.py | 78 ++++- reolink_aio/scripts/__init__.py | 1 + reolink_aio/scripts/stream_vod.py | 127 ++++++++ setup.py | 5 + 7 files changed, 698 insertions(+), 32 deletions(-) create mode 100644 reolink_aio/scripts/__init__.py create mode 100644 reolink_aio/scripts/stream_vod.py diff --git a/reolink_aio/api.py b/reolink_aio/api.py index 5ed9ec8c..7bb4016e 100644 --- a/reolink_aio/api.py +++ b/reolink_aio/api.py @@ -9,6 +9,7 @@ import re import ssl import traceback +from collections.abc import AsyncIterator from datetime import date, datetime, timedelta, tzinfo from io import BytesIO from math import ceil @@ -5865,6 +5866,9 @@ async def get_recording_days(self, channel: int, year: int, month: int) -> set[i if channel not in self._stream_channels: raise InvalidParameterError(f"get_recording_days: no camera connected to channel '{channel}'") + if self.baichuan_only: + return await self.baichuan.search_recording_days_bc(channel, year, month) + import calendar last_day = calendar.monthrange(year, month)[1] @@ -5921,6 +5925,13 @@ async def get_recordings_for_day( if channel not in self._stream_channels: raise InvalidParameterError(f"get_recordings_for_day: no camera connected to channel '{channel}'") + if self.baichuan_only: + vod_files = await self.baichuan.search_recordings_for_day_bc(channel, day, stream) + if trigger is not None: + vod_files = [f for f in vod_files if f.bc_triggers is not None and bool(f.bc_triggers & trigger)] + vod_files.sort(key=lambda f: f.start_time) + return vod_files + start = datetime(day.year, day.month, day.day, 0, 0, 0) end = datetime(day.year, day.month, day.day, 23, 59, 59) @@ -5940,6 +5951,41 @@ async def get_recordings_for_day( unique.sort(key=lambda f: f.start_time) return unique + def stream_recording_bc( + self, + channel: int, + file_name: str, + start_time: datetime, + stream_type: str = "mainStream", + ) -> AsyncIterator[tuple[int, bytes]]: + """Async generator: stream a VOD recording via the Baichuan protocol. + + Yields ``(microseconds, h264_bytes)`` tuples for each video frame. + ``microseconds`` is the camera-relative timestamp (u32, wraps at ~71 min). + ``h264_bytes`` is the raw H.264 Annex-B NAL data for the frame. + + Use this for baichuan_only cameras where HTTP download is unavailable. + + Parameters + ---------- + channel: + Camera channel index. + file_name: + Recording filename from ``get_recordings_for_day()``. + start_time: + Recording start time (from ``VOD_file.start_time``). + stream_type: + ``"mainStream"`` (default) or ``"subStream"``. + + Usage:: + + async for microseconds, h264_bytes in host.stream_recording_bc(ch, name, start_time): + ... + """ + return self.baichuan.parse_bcmedia_frames( + self.baichuan.stream_replay_bc(channel, file_name, start_time, stream_type) + ) + async def send_setting(self, body: typings.reolink_json, wait_before_get: int = 0, getcmd: str = "") -> None: command = body[0]["cmd"] _LOGGER.debug( diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index 72e53657..c3c94bac 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -4,8 +4,8 @@ import asyncio import logging -from collections.abc import Callable -from datetime import datetime, timedelta +from collections.abc import AsyncIterator, Callable +from datetime import date, datetime, timedelta from inspect import getmembers from time import time as time_now from typing import TYPE_CHECKING, Any, Coroutine, Literal, TypeVar, overload @@ -3609,32 +3609,410 @@ async def search_vod_type( await self.send(cmd_id=274, channel=channel, body=xml) - # xml = xmls.FileInfoListOpen.format( - # channel=channel, - # uid=uid, - # stream_type=stream_type, - # start_year=start.year, - # start_month=start.month, - # start_day=start.day, - # start_hour=start.hour, - # start_minute=start.minute, - # start_second=start.second, - # end_year=end.year, - # end_month=end.month, - # end_day=end.day, - # end_hour=end.hour, - # end_minute=end.minute, - # end_second=end.second, - # ) - # mess = await self.send(cmd_id=14, body=xml) - # handle = self._get_value_from_xml(mess, "handle") - - # xml_file_info = xmls.FileInfoList.format(channel=channel, handle=handle, uid=uid) - # await self.send(cmd_id=15, body=xml_file_info) - # await self.send(cmd_id=16, body=xml_file_info) - return vod_type_dict, vod_dict + async def search_recording_days_bc(self, channel: int, year: int, month: int) -> set[int]: + """Return the set of day numbers (1–31) that have recordings in year/month via Baichuan MSG 142. + + Used as a fallback when baichuan_only=True (no HTTP API available). + """ + import calendar as cal + + last_day = cal.monthrange(year, month)[1] + xml = xmls.DayRecords.format(year=year, month=month, last_day=last_day, channel=channel) + mess = await self.send(cmd_id=142, channel=channel, body=xml) + root = XML.fromstring(mess) + + days: set[int] = set() + for day_type in root.findall(".//dayType"): + index = self._get_value_from_xml_element(day_type, "index", int) + if index is not None: + # index is 0-based offset from day 1 of the queried month + days.add(1 + index) + return days + + async def search_recordings_for_day_bc(self, channel: int, day: date, stream: str | None = None) -> list[VOD_file]: + """Return all recordings for *day* on *channel* via Baichuan MSG 14 + MSG 15. + + Used as a fallback when baichuan_only=True (no HTTP API available). + """ + uid = self.http_api.camera_uid(channel) + uid = uid.split("_")[0] + + if stream == "sub": + stream_type = "subStream" + elif stream in {"autotrack_sub", "telephoto_sub"}: + stream_type = "subStream" + else: + stream_type = "mainStream" + + # MSG 14: get file list handle for this day + xml = xmls.FileInfoListOpen.format( + uid=uid, + channel=channel, + stream_type=stream_type, + start_year=day.year, + start_month=day.month, + start_day=day.day, + start_hour=0, + start_minute=0, + start_second=0, + end_year=day.year, + end_month=day.month, + end_day=day.day, + end_hour=23, + end_minute=59, + end_second=59, + ) + mess = await self.send(cmd_id=14, channel=channel, body=xml) + handle = self._get_value_from_xml(mess, "handle") + if handle is None: + _LOGGER.debug("Baichuan host %s: search_recordings_for_day_bc: MSG 14 returned no handle for channel %s day %s", self._host, channel, day) + return [] + + # MSG 15: fetch the file list using the handle + xml = xmls.FileInfoList.format(channel=channel, uid=uid, handle=handle) + mess = await self.send(cmd_id=15, channel=channel, body=xml) + root = XML.fromstring(mess) + + vod_files: list[VOD_file] = [] + for fi in root.findall(".//FileInfo"): + name = self._get_value_from_xml_element(fi, "name") + start_time_el = fi.find("startTime") + end_time_el = fi.find("endTime") + size_l = self._get_value_from_xml_element(fi, "sizeL", int) + size_h = self._get_value_from_xml_element(fi, "sizeH", int) + record_type = self._get_value_from_xml_element(fi, "recordType") + + if name is None or start_time_el is None: + continue + + start_dt = self._xml_time_to_datetime(start_time_el) + end_dt = self._xml_time_to_datetime(end_time_el) + if start_dt is None: + continue + + size = (size_h or 0) * (2**32) + (size_l or 0) + data: dict = { + "type": stream or self.http_api._stream, + "StartTime": datetime_to_reolink_time(start_dt), + "EndTime": datetime_to_reolink_time(end_dt or start_dt), + "PlaybackTime": datetime_to_reolink_time(start_dt), + "name": name, + "size": str(size), + } + vod_file = VOD_file(data, self.http_api.timezone()) + + triggers = VOD_trigger.NONE + if record_type: + for part in record_type.replace(",", " ").split(): + if part in ("md", "pir", "other"): + triggers |= VOD_trigger.MOTION + elif part == "people": + triggers |= VOD_trigger.PERSON + elif part == "vehicle": + triggers |= VOD_trigger.VEHICLE + elif part == "dog_cat": + triggers |= VOD_trigger.ANIMAL + elif part == "visitor": + triggers |= VOD_trigger.DOORBELL + elif part == "face": + triggers |= VOD_trigger.FACE + elif part == "package": + triggers |= VOD_trigger.PACKAGE + elif part in ("normal", "sched", "manual", "timer", "io"): + triggers |= VOD_trigger.TIMER + vod_file.bc_triggers = triggers + vod_files.append(vod_file) + + return vod_files + + async def replay_seek_bc(self, channel: int, seek_dt: datetime) -> None: + """Send ReplaySeek (MSG 123) to prepare camera for VOD playback.""" + seq = int(time_now()) + xml = xmls.ReplaySeek.format( + channel=channel, + seq=seq, + year=seek_dt.year, + month=seek_dt.month, + day=seek_dt.day, + hour=seek_dt.hour, + minute=seek_dt.minute, + second=seek_dt.second, + ) + await self.send(cmd_id=123, channel=channel, body=xml) + + async def _send_streaming(self, cmd_id: int, channel: int, body: str) -> tuple[asyncio.Queue, int]: + """Send *cmd_id* and register a streaming queue for all its responses. + + Returns (queue, full_mess_id). The caller is responsible for removing + the queue from self._protocol.streaming_queues when done. + """ + if not self._logged_in and cmd_id > 2: + await self.login() + + ch_id = channel + 1 if channel is not None else 250 + ext = xmls.CHANNEL_EXTENSION_XML.format(channel=channel) if channel is not None else "" + + mess_len = len(ext) + len(body) + payload_offset = len(ext) + + self._mess_id = (self._mess_id + 1) % 16777216 + mess_id_bytes = ch_id.to_bytes(1, "little") + self._mess_id.to_bytes(3, "little") + full_mess_id = int.from_bytes(mess_id_bytes, "little") + + cmd_id_bytes = cmd_id.to_bytes(4, "little") + mess_len_bytes = mess_len.to_bytes(4, "little") + payload_offset_bytes = payload_offset.to_bytes(4, "little") + header = ( + bytes.fromhex(HEADER_MAGIC) + + cmd_id_bytes + + mess_len_bytes + + mess_id_bytes + + bytes.fromhex("0000" + "1464") + + payload_offset_bytes + ) + enc_body = self._aes_encrypt(ext) + self._aes_encrypt(body) + + await self._connect_if_needed() + if TYPE_CHECKING: + assert self._protocol is not None + assert self._transport is not None + + q: asyncio.Queue = asyncio.Queue(maxsize=512) + self._protocol.streaming_queues[(cmd_id, full_mess_id)] = q + + _LOGGER.debug("Baichuan host %s: streaming send cmd_id %s full_mess_id %s", self._host, cmd_id, full_mess_id) + async with self._mutex: + self._transport.write(header + enc_body) + + return q, full_mess_id + + @staticmethod + async def parse_bcmedia_frames( + raw_stream: AsyncIterator[bytes], + ) -> AsyncIterator[tuple[int, bytes]]: + """Parse a raw Reolink BcMedia byte stream and yield (microseconds, h264_bytes) per video frame. + + BcMedia magic layout (4 bytes, LE): + byte[0]: frame counter, '0'-'9' (0x30-0x39), cycles 0-9 + byte[1]: frame type, '0'=IFrame (0x30) or '1'=PFrame (0x31) + byte[2]: 'd' (0x64) + byte[3]: 'c' (0x63) + Audio and info frames (different magic) are skipped. + """ + buf = bytearray() + + def _find_bcmedia_magic(data: bytearray, start: int) -> int | None: + """Return index of the next BcMedia frame (video or audio) at an 8-byte-aligned position, or None. + + Recognises: + - Video I-frame: byte[0]='0'-'9', byte[1]='0', byte[2]='d', byte[3]='c' + - Video P-frame: byte[0]='0'-'9', byte[1]='1', byte[2]='d', byte[3]='c' + - Audio AAC: byte[0]='0'-'9', byte[1]='5', byte[2]='w', byte[3]='b' + - Audio ADPCM: byte[0]='0'-'9', byte[1]='1', byte[2]='w', byte[3]='b' + + BcMedia frames are always 8-byte padded so all frame starts lie at 8-byte-aligned + stream offsets. Scanning only aligned positions avoids false positives that can + appear inside H.264 or encrypted-audio payloads. + """ + i = (start + 7) & ~7 if start % 8 != 0 else start + while i + 4 <= len(data): + b0, b1, b2, b3 = data[i], data[i + 1], data[i + 2], data[i + 3] + if 0x30 <= b0 <= 0x39: + if b2 == 0x64 and b3 == 0x63 and b1 in (0x30, 0x31): + return i # video I/P-frame + if b2 == 0x77 and b3 == 0x62 and b1 in (0x31, 0x35): + return i # audio ADPCM / AAC + i += 8 + return None + + scan_from = 0 + async for chunk in raw_stream: + buf.extend(chunk) + while True: + magic_idx = _find_bcmedia_magic(buf, scan_from) + if magic_idx is None: + # Back up to last 8-byte-aligned position so a magic spanning a chunk boundary is rechecked + scan_from = max(0, len(buf) - 7) & ~7 + break + + b2 = buf[magic_idx + 2] + + if b2 == 0x77: + # Audio frame (AAC or ADPCM): 4-byte magic + 2-byte payload_size + 2-byte payload_size_b + payload + padding + if magic_idx + 8 > len(buf): + scan_from = magic_idx + break + audio_payload_size = int.from_bytes(buf[magic_idx + 4: magic_idx + 6], "little") + audio_pad = (8 - audio_payload_size % 8) % 8 + audio_total = 8 + audio_payload_size + audio_pad + if magic_idx + audio_total > len(buf): + scan_from = magic_idx + break + del buf[: magic_idx + audio_total] + scan_from = 0 + continue # skip audio, look for next frame + + # Video frame (I or P) + # Need at least 24 bytes for the fixed header fields + if magic_idx + 24 > len(buf): + scan_from = magic_idx + break + + payload_size = int.from_bytes(buf[magic_idx + 8: magic_idx + 12], "little") + additional_header_size = int.from_bytes(buf[magic_idx + 12: magic_idx + 16], "little") + microseconds = int.from_bytes(buf[magic_idx + 16: magic_idx + 20], "little") + + hdr_size = 24 + additional_header_size + pad_size = (8 - payload_size % 8) % 8 + total_size = hdr_size + payload_size + pad_size + + if magic_idx + total_size > len(buf): + # Frame not yet complete — wait for more data + scan_from = magic_idx + break + + h264_bytes = bytes(buf[magic_idx + hdr_size: magic_idx + hdr_size + payload_size]) + del buf[: magic_idx + total_size] + scan_from = 0 + + if h264_bytes: + yield microseconds, h264_bytes + + async def stream_replay_bc( + self, + channel: int, + file_name: str, + start_time: datetime, + stream_type: str = "mainStream", + ) -> AsyncIterator[bytes]: + """Async generator: stream a recorded VOD file via Baichuan MSG 5/8. + + Yields raw (decrypted) binary chunks of video data. + Sends ReplaySeek (MSG 123) first, then MSG 5 (falls back to MSG 8 on 400), + skips the 32-byte replay header, and yields each subsequent binary packet. + Sends ReplayStop (MSG 7) in cleanup. + """ + await self.replay_seek_bc(channel, start_time) + + support_sub = 1 if stream_type == "subStream" else 0 + body = xmls.ReplayStart.format( + name=file_name, + channel=channel, + support_sub=support_sub, + stream_type=stream_type, + start_year=start_time.year, + start_month=start_time.month, + start_day=start_time.day, + start_hour=start_time.hour, + start_minute=start_time.minute, + start_second=start_time.second, + ) + + accepted_cmd_id: int = 0 + full_mess_id: int = 0 + q: asyncio.Queue | None = None + + for try_cmd_id in (5, 8): + q, full_mess_id = await self._send_streaming(try_cmd_id, channel, body) + try: + status_code, data_chunk, len_hdr, payload = await asyncio.wait_for(q.get(), timeout=TIMEOUT) + except asyncio.TimeoutError: + if self._protocol is not None: + self._protocol.streaming_queues.pop((try_cmd_id, full_mess_id), None) + raise ReolinkTimeoutError(f"Baichuan host {self._host}: timeout waiting for replay start (MSG {try_cmd_id})") + + if status_code == 400: + if self._protocol is not None: + self._protocol.streaming_queues.pop((try_cmd_id, full_mess_id), None) + if try_cmd_id == 8: + raise ReolinkError(f"Baichuan host {self._host}: replay start rejected by camera (MSG 5 and MSG 8 both returned 400)") + _LOGGER.debug("Baichuan host %s: replay MSG 5 rejected (400), trying MSG 8", self._host) + continue + + if status_code != 200: + if self._protocol is not None: + self._protocol.streaming_queues.pop((try_cmd_id, full_mess_id), None) + raise ReolinkError(f"Baichuan host {self._host}: unexpected replay start response code {status_code} from MSG {try_cmd_id}") + + _LOGGER.debug("Baichuan host %s: replay accepted (200) via MSG %s", self._host, try_cmd_id) + accepted_cmd_id = try_cmd_id + break + + assert q is not None + + STREAM_TIMEOUT = 15.0 + packet_count = 0 + try: + while True: + try: + status_code, data_chunk, len_hdr, payload = await asyncio.wait_for(q.get(), timeout=STREAM_TIMEOUT) + except asyncio.TimeoutError: + _LOGGER.debug("Baichuan host %s: replay stream: no data for %.0fs, stopping", self._host, STREAM_TIMEOUT) + break + + if status_code in (300, 331): + _LOGGER.debug("Baichuan host %s: replay stream ended (response %s)", self._host, status_code) + break + + packet_count += 1 + + # Continuation packets (no extension, payload_offset=0) carry plaintext video data in the body. + # Packets with an extension (payload non-empty) may have an encrypted region. + binary: bytes = payload if payload else data_chunk[len_hdr:] + if not binary: + continue + + # First packet: 32-byte replay header — skip it + if packet_count == 1 and len(binary) == 32: + _LOGGER.debug("Baichuan host %s: replay: skipping 32-byte stream header", self._host) + continue + + if payload and self._aes_key is not None: + # Packet has an extension section — try XML-based decryption first (encryptLen in Extension XML). + enc_pos = 0 + enc_len: int | None = None + xml_enc = data_chunk[len_hdr:] + if xml_enc: + try: + xml_str = self._aes_decrypt(xml_enc, data_chunk[:len_hdr]) + if xml_str.startswith("= 8: + probe = AES.new(key=self._aes_key, mode=AES.MODE_CFB, iv=AES_IV, segment_size=128).decrypt(payload[:8]) + if probe[:4] == b"00dc": + enc_len = 32 + enc_pos = 0 + + if enc_len and enc_len > 0: + end = enc_pos + enc_len + decrypted_region = AES.new(key=self._aes_key, mode=AES.MODE_CFB, iv=AES_IV, segment_size=128).decrypt(payload[enc_pos:end]) + binary = payload[:enc_pos] + decrypted_region + payload[end:] + + if packet_count <= 3 or packet_count % 100 == 0: + _LOGGER.debug("Baichuan host %s: replay pkt %s: %s bytes", self._host, packet_count, len(binary)) + + yield binary + + finally: + if self._protocol is not None: + self._protocol.streaming_queues.pop((accepted_cmd_id, full_mess_id), None) + stop_xml = xmls.ReplayStop.format(channel=channel, name=file_name) + try: + await self.send(cmd_id=7, channel=channel, body=stop_xml) + _LOGGER.debug("Baichuan host %s: replay stop (MSG 7) sent", self._host) + except Exception as err: + _LOGGER.debug("Baichuan host %s: replay stop (MSG 7) error: %s", self._host, err) + @property def events_active(self) -> bool: return self._events_active and time_now() - self._time_connection_lost > 120 diff --git a/reolink_aio/baichuan/tcp_protocol.py b/reolink_aio/baichuan/tcp_protocol.py index de0a03c0..eec66483 100644 --- a/reolink_aio/baichuan/tcp_protocol.py +++ b/reolink_aio/baichuan/tcp_protocol.py @@ -26,6 +26,7 @@ def __init__(self, loop, host: str, push_callback: Callable[[int, bytes, int, by self._data_chunk: bytes = b"" self.receive_futures: dict[int, dict[int, asyncio.Future]] = {} # expected_cmd_id: rec_future + self.streaming_queues: dict[tuple[int, int], asyncio.Queue] = {} # (cmd_id, mess_id): queue self.close_future: asyncio.Future = loop.create_future() self._close_callback = close_callback self._push_callback = push_callback @@ -58,6 +59,12 @@ def _set_error(self, err_mess: str, exc_class: type[Exception] = ReolinkError, c def data_received(self, data: bytes) -> None: """Data received callback""" + # E1 cameras prepend a 4-byte prefix before each standard Baichuan packet + # (neolink bc/codex.rs: "E1 sends 4 bytes then magic"). Strip it transparently. + if len(data) >= 8 and data[0:4].hex() != HEADER_MAGIC and data[4:8].hex() == HEADER_MAGIC: + _LOGGER.debug("Baichuan host %s: stripping E1 4-byte prefix %s", self._host, data[0:4].hex()) + data = data[4:] + # parse received header if data[0:4].hex() == HEADER_MAGIC: if self._data: @@ -118,13 +125,20 @@ def parse_data(self) -> None: _LOGGER.debug("Baichuan host %s: received start of modern header with message class %s but less then 24 bytes, waiting for the rest", self._host, mess_class) return rec_payload_offset = int.from_bytes(self._data[20:24], byteorder="little") - elif mess_class == "1465": # legacy 20 byte header + elif mess_class == "1465": # legacy 20 byte header — used for streaming on some devices (e.g. E1) len_header = 20 - self._set_error("with legacy message class, parsing not implemented", InvalidContentTypeError, rec_cmd_id, rec_mess_id) - return + # Do not error immediately; streaming_queues may handle this packet. + # If no streaming queue claims it below, _set_error is called after extraction. else: - self._set_error(f"with unknown message class '{mess_class}'", InvalidContentTypeError, rec_cmd_id, rec_mess_id) - return + # Unknown mess_class — check if a streaming queue exists before erroring. + # The E1 camera uses mess_class "9a69" for streaming binary data (AES/v2 = 24-byte header). + if len(self._data) >= 24 and self.streaming_queues.get((rec_cmd_id, rec_mess_id)) is not None: + len_header = 24 + rec_payload_offset = int.from_bytes(self._data[20:24], byteorder="little") + _LOGGER.debug("Baichuan host %s: unknown message class '%s' for streaming cmd_id %s, treating as 24-byte header", self._host, mess_class, rec_cmd_id) + else: + self._set_error(f"with unknown message class '{mess_class}'", InvalidContentTypeError, rec_cmd_id, rec_mess_id) + return # check message length len_body = len(self._data) - len_header @@ -148,6 +162,21 @@ def parse_data(self) -> None: else: # len_body == rec_len_body self._data = b"" + # streaming_queues bypass status-code filtering and receive_future routing + stream_q = self.streaming_queues.get((rec_cmd_id, rec_mess_id)) + if stream_q is not None: + rec_status_code_s = int.from_bytes(self._data_chunk[16:18], byteorder="little") if len(self._data_chunk) >= 18 else 200 + try: + stream_q.put_nowait((rec_status_code_s, bytes(self._data_chunk), len_header, bytes(payload))) + except asyncio.QueueFull: + _LOGGER.warning("Baichuan host %s: streaming queue full for cmd_id %s, dropping packet", self._host, rec_cmd_id) + return + + # Legacy "1465" class is only supported via streaming_queues; error for any other use + if mess_class == "1465": + self._set_error("with legacy message class, parsing not implemented", InvalidContentTypeError, rec_cmd_id, rec_mess_id) + return + # extract receive future receive_future = self.receive_futures.get(rec_cmd_id, {}).get(rec_mess_id) @@ -188,6 +217,10 @@ def parse_data(self) -> None: finally: # if multiple messages received, parse the next also if self._data: + # Strip E1 4-byte prefix between consecutive packets if present + if len(self._data) >= 8 and self._data[0:4].hex() != HEADER_MAGIC and self._data[4:8].hex() == HEADER_MAGIC: + _LOGGER.debug("Baichuan host %s: stripping E1 4-byte prefix between messages %s", self._host, self._data[0:4].hex()) + self._data = self._data[4:] if self._data[0:4].hex() == HEADER_MAGIC: self.parse_data() elif len(self._data) < 4 and bytes.fromhex(HEADER_MAGIC).startswith(self._data): diff --git a/reolink_aio/baichuan/xmls.py b/reolink_aio/baichuan/xmls.py index ec096d34..2bae7dda 100644 --- a/reolink_aio/baichuan/xmls.py +++ b/reolink_aio/baichuan/xmls.py @@ -190,6 +190,35 @@ """ +DayRecords = """ + + + + +{year} +{month} +1 +0 +0 +0 + + +{year} +{month} +{last_day} +23 +59 +59 + + + +0 +{channel} + + + +""" + FileInfoListOpen = """ @@ -199,7 +228,7 @@ 1 {channel} 255 -mainStream +{stream_type} manual, sched, io, md, people, face, vehicle, dog_cat, visitor, other, package {start_year} @@ -500,3 +529,50 @@ """ + +ReplaySeek = """ + + +{channel} +{seq} + +{year} +{month} +{day} +{hour} +{minute} +{second} + + +""" + +ReplayStart = """ + + + +0 +{name} +{channel} +{support_sub} +{stream_type} + +{start_year} +{start_month} +{start_day} +{start_hour} +{start_minute} +{start_second} + + + +""" + +ReplayStop = """ + + + +{channel} +{name} + + +""" diff --git a/reolink_aio/scripts/__init__.py b/reolink_aio/scripts/__init__.py new file mode 100644 index 00000000..0ba461f7 --- /dev/null +++ b/reolink_aio/scripts/__init__.py @@ -0,0 +1 @@ +"""Command-line scripts for reolink_aio.""" diff --git a/reolink_aio/scripts/stream_vod.py b/reolink_aio/scripts/stream_vod.py new file mode 100644 index 00000000..9a821560 --- /dev/null +++ b/reolink_aio/scripts/stream_vod.py @@ -0,0 +1,127 @@ +"""Stream a Reolink VOD recording as H.264 Annex-B to stdout. + +Intended as a ``go2rtc`` ``exec:`` source for baichuan_only cameras that do not +support HTTP download or RTSP VOD. The script connects to the camera via the +Baichuan TCP protocol, retrieves the requested recording, and writes raw H.264 +Annex-B frames to stdout **at the real-time rate implied by the BcMedia +microsecond timestamps**. Because go2rtc assigns RTP timestamps from its own +wall clock, rate-limiting the output is enough to achieve correct playback speed +without any ``setpts`` manipulation. + +Usage:: + + reolink-stream-vod HOST USER PASS CHANNEL FILE_NAME START_TIME [STREAM_TYPE] + + HOST Camera hostname or IP address + USER Username + PASS Password + CHANNEL Channel index (integer, e.g. 0) + FILE_NAME Recording filename returned by get_recordings_for_day() + START_TIME Recording start time as ISO 8601 string (e.g. "2026-02-22T22:41:42") + STREAM_TYPE Optional: "mainStream" (default) or "subStream" + +Example go2rtc configuration (go2rtc.yaml):: + + streams: + reolink_ch0_0120260222224142: + - exec:reolink-stream-vod 192.168.1.10 admin secret 0 0120260222224142 2026-02-22T22:41:42 + +The stream name can then be used as an HLS/WebRTC source in Home Assistant or any +other go2rtc consumer. +""" + +from __future__ import annotations + +import asyncio +import logging +import sys +import time +from datetime import datetime + +from ..api import Host + +_LOGGER = logging.getLogger(__name__) + +# u32 microsecond timestamp wraps at 2^32 µs ≈ 71.6 minutes. +_U32_MASK = 0xFFFFFFFF + + +async def _stream( + host_addr: str, + user: str, + password: str, + channel: int, + file_name: str, + start_time: datetime, + stream_type: str, +) -> None: + """Connect to the camera and write H.264 frames to stdout at real-time rate.""" + out = sys.stdout.buffer + + host = Host(host_addr, user, password) + try: + # Skip get_host_data()/get_states() — the Baichuan TCP layer authenticates + # itself lazily on the first command, so no HTTP round-trips are needed. + first_us: int | None = None + wall_start: float | None = None + + async for us, h264 in host.stream_recording_bc(channel, file_name, start_time, stream_type): + if first_us is None: + # Anchor real-time clock to the first BcMedia timestamp. + first_us = us + wall_start = time.monotonic() + else: + # Elapsed recording time (µs), with u32 wraparound handled. + elapsed_bc_us = (us - first_us) & _U32_MASK + target = wall_start + elapsed_bc_us / 1_000_000 + delay = target - time.monotonic() + if delay > 0: + await asyncio.sleep(delay) + + try: + out.write(h264) + out.flush() + except BrokenPipeError: + # Consumer (go2rtc) closed the pipe — stop streaming. + _LOGGER.debug("stdout pipe closed, stopping stream") + break + + finally: + await host.logout() + + +def main() -> None: + """Entry point for the ``reolink-stream-vod`` console script.""" + args = sys.argv[1:] + if len(args) not in (6, 7): + print( + f"Usage: {sys.argv[0]} HOST USER PASS CHANNEL FILE_NAME START_TIME [STREAM_TYPE]", + file=sys.stderr, + ) + sys.exit(1) + + host_addr, user, password, channel_str, file_name, start_time_str = args[:6] + stream_type = args[6] if len(args) == 7 else "mainStream" + + try: + channel = int(channel_str) + except ValueError: + print(f"CHANNEL must be an integer, got: {channel_str!r}", file=sys.stderr) + sys.exit(1) + + try: + start_time = datetime.fromisoformat(start_time_str) + except ValueError as exc: + print(f"START_TIME must be an ISO 8601 datetime string: {exc}", file=sys.stderr) + sys.exit(1) + + logging.basicConfig(level=logging.WARNING, stream=sys.stderr) + + try: + asyncio.run(_stream(host_addr, user, password, channel, file_name, start_time, stream_type)) + except KeyboardInterrupt: + pass + + +if __name__ == "__main__": + main() diff --git a/setup.py b/setup.py index d4e0b79a..e6e3d4d0 100644 --- a/setup.py +++ b/setup.py @@ -29,6 +29,11 @@ 'pycryptodomex', 'typing_extensions' ], + entry_points={ + "console_scripts": [ + "reolink-stream-vod=reolink_aio.scripts.stream_vod:main", + ], + }, tests_require=[], platforms=['any'], zip_safe=False, From 868493251cb0bfafdb8caf78cf4ee5df63d0de6e Mon Sep 17 00:00:00 2001 From: lorek123 Date: Sun, 1 Mar 2026 22:55:06 +0100 Subject: [PATCH 04/11] Fix 2-way audio: correct BcMedia ADPCM frame format and cmd_ids - Fix BCMEDIA_ADPCM_MAGIC byte order: was b"bw10" (big-endian), must be struct.pack(" element (not audioStreamMode) - Add CHANNEL_BINARY_EXTENSION_XML with 1 for audio talk data packets (cmd_id=202) - Remove unused TalkAbilityGet XML template Co-Authored-By: Claude Sonnet 4.6 --- reolink_aio/baichuan/baichuan.py | 61 +++++++++++++++++++++----------- reolink_aio/baichuan/xmls.py | 13 +++---- 2 files changed, 47 insertions(+), 27 deletions(-) diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index c3c94bac..60e7b3e3 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -4,6 +4,7 @@ import asyncio import logging +import struct from collections.abc import AsyncIterator, Callable from datetime import date, datetime, timedelta from inspect import getmembers @@ -452,8 +453,7 @@ async def send_binary_no_reply( else: ch_id = channel + 1 - ext = xmls.CHANNEL_EXTENSION_XML.format(channel=channel) if channel is not None else "" - + ext = xmls.CHANNEL_BINARY_EXTENSION_XML.format(channel=channel) if channel is not None else "" # Extension is AES-encrypted; binary body is sent as-is (NOT encrypted) enc_ext = self._aes_encrypt(ext.encode("utf-8")) mess_len = len(enc_ext) + len(binary_body) @@ -1507,9 +1507,8 @@ async def get_channel_data(self) -> None: if self.http_api.is_nvr and self.http_api.wifi_connection(channel) and (self.http_api.api_version("supportWiFi", channel) > 0 or self.http_api._is_hub): coroutines.append(("wifi", channel, self.get_wifi_signal(channel))) - if self.http_api.api_version("talk", channel) > 0: + if self.http_api.api_version("talk", channel) > 0 or self.http_api.baichuan_only: coroutines.append((10, channel, self.send(cmd_id=10, channel=channel))) - coroutines.append((201, channel, self.get_talk_ability(channel))) if (self.http_api.is_nvr or self.privacy_mode() is not None) and self.api_version("remoteAbility", channel) > 0: coroutines.append(("privacy_mode", channel, self.get_privacy_mode(channel))) # capability added in get_privacy_mode @@ -1655,11 +1654,14 @@ async def get_channel_data(self) -> None: if isinstance(result, BaseException): raise result - if cmd_id == 10: # two way audio (legacy check via audioStreamMode) + if cmd_id == 10: # two way audio: check for duplex or mixAudioStream root = XML.fromstring(result) - for audio in root.findall(".//audioStreamMode"): - if audio.text == "mixAudioStream": - self.capabilities[channel].add("two_way_audio") + if root.find(".//duplex") is not None: + self.capabilities[channel].add("two_way_audio") + else: + for audio in root.findall(".//audioStreamMode"): + if audio.text == "mixAudioStream": + self.capabilities[channel].add("two_way_audio") if cmd_id == 201: # TalkAbility (cmd_id=201) – authoritative two_way_audio check # result is already a dict returned by get_talk_ability() if isinstance(result, dict) and result.get("duplex"): @@ -4209,17 +4211,17 @@ def audio_noise_reduction(self, channel: int) -> int | None: return self._noise_reduction.get(channel) # ------------------------------------------------------------------------- - # Two-way audio (talk) support - # cmd_id 201 = TalkAbility (query) - # cmd_id 202 = Talk (send ADPCM audio data, no reply expected) - # cmd_id 203 = TalkConfig (start/configure session) - # cmd_id 204 = TalkReset (stop session) + # Two-way audio (talk) support (confirmed via Ghidra / neolink) + # cmd_id 10 = TalkAbility (query, confirmed: contextGet(..., 10, 0)) + # cmd_id 11 = TalkReset (stop, confirmed: simpleSndFuncCH(..., 0xb, ...)) + # cmd_id 201 = TalkConfig (start, confirmed: xmlSndFuncCH(..., 0xc9, ...)) + # cmd_id 202 = Talk (send ADPCM audio data, no reply expected) # ------------------------------------------------------------------------- - BCMEDIA_ADPCM_MAGIC = b"bw10" # 0x62773130 + BCMEDIA_ADPCM_MAGIC = struct.pack(" dict: - """Query the camera's talk (2-way audio) capability via Baichuan cmd_id=201. + """Query the camera's talk (2-way audio) capability via cmd_id=10. Returns a dict with keys: - duplex: str (e.g. "FDX") @@ -4230,7 +4232,7 @@ async def get_talk_ability(self, channel: int) -> dict: - length_per_encoder: int (samples per block, e.g. 320 or 640) - sound_track: str (e.g. "mono") """ - mess = await self.send(cmd_id=201, channel=channel, body=xmls.TalkAbilityGet) + mess = await self.send(cmd_id=10, channel=channel) root = XML.fromstring(mess) ability: dict = {} @@ -4269,7 +4271,7 @@ async def get_talk_ability(self, channel: int) -> dict: return ability async def start_talk(self, channel: int) -> dict: - """Start a 2-way audio talk session via Baichuan cmd_id=203. + """Start a 2-way audio talk session via Baichuan cmd_id=201 (TalkConfig). Queries the camera's TalkAbility, then sends a TalkConfig message to initiate the session. Returns the audio configuration dict (same @@ -4299,7 +4301,7 @@ async def start_talk(self, channel: int) -> dict: length_per_encoder=length_per_encoder, sound_track=sound_track, ) - await self.send(cmd_id=203, channel=channel, body=xml) + await self.send(cmd_id=201, channel=channel, body=xml) _LOGGER.debug( "Baichuan host %s ch %s: talk session started, audio_type=%s sample_rate=%s length_per_encoder=%s", @@ -4312,8 +4314,8 @@ async def start_talk(self, channel: int) -> dict: return ability async def stop_talk(self, channel: int) -> None: - """Stop the 2-way audio talk session via Baichuan cmd_id=204.""" - await self.send(cmd_id=204, channel=channel) + """Stop the 2-way audio talk session via Baichuan cmd_id=11 (TalkReset).""" + await self.send(cmd_id=11, channel=channel) _LOGGER.debug("Baichuan host %s ch %s: talk session stopped", self._host, channel) @staticmethod @@ -4327,9 +4329,26 @@ def build_bcmedia_adpcm(adpcm_blocks: list[bytes]) -> bytes: Up to 4 blocks may be combined into one BcMedia message. The returned bytes can be passed directly to send_talk_data(). """ + # BcMedia ADPCM frame layout (confirmed from pcap + Ghidra audioTalkSendStream): + # 4 bytes magic (0x62773130 = "bw10") + # 2 bytes payload_size = len(block) + 4 (LE u16, duplicated in next 2 bytes) + # 2 bytes payload_size (duplicate) + # 2 bytes sub-magic 0x0100 (MAGIC_HEADER_BCMEDIA_ADPCM_DATA) + # 2 bytes half_block = 2 (always 2; Ghidra: (sample_count>>1)+2 where sample_count=0) + # N bytes raw IMA ADPCM block (4-byte header + nibble data) + # P bytes zero padding to 8-byte boundary (based on payload_size; 0 for typical 520) payload = b"" for block in adpcm_blocks: - payload += Baichuan.BCMEDIA_ADPCM_MAGIC + block + data_len = len(block) + payload_size = data_len + 4 + pad_size = (8 - payload_size % 8) % 8 + payload += ( + Baichuan.BCMEDIA_ADPCM_MAGIC + + struct.pack(" None: diff --git a/reolink_aio/baichuan/xmls.py b/reolink_aio/baichuan/xmls.py index 2bae7dda..166f9a91 100644 --- a/reolink_aio/baichuan/xmls.py +++ b/reolink_aio/baichuan/xmls.py @@ -33,6 +33,13 @@ """ +CHANNEL_BINARY_EXTENSION_XML = """ + +1 +{channel} + +""" + DingDongOpt_1_XML = """ @@ -398,12 +405,6 @@ """ -TalkAbilityGet = """ - - - -""" - TalkConfigSet = """ From 93b9df92b4a3c3e4cdd01591ce2b9db9d3c81c22 Mon Sep 17 00:00:00 2001 From: lorek123 Date: Fri, 3 Apr 2026 00:47:31 +0200 Subject: [PATCH 05/11] Fix lint/style issues ahead of upstream PR - Move `import calendar` to top-level in api.py and baichuan.py (C0415) - Add `from err` to raise in stream_replay_bc timeout handler (W0707) - Reformat with black (api.py, baichuan/baichuan.py) Co-Authored-By: Claude Sonnet 4.6 --- reolink_aio/api.py | 11 +++------- reolink_aio/baichuan/baichuan.py | 36 ++++++++++---------------------- 2 files changed, 14 insertions(+), 33 deletions(-) diff --git a/reolink_aio/api.py b/reolink_aio/api.py index 7bb4016e..163eea7d 100644 --- a/reolink_aio/api.py +++ b/reolink_aio/api.py @@ -4,6 +4,7 @@ import asyncio import base64 +import calendar import hashlib import logging import re @@ -5869,8 +5870,6 @@ async def get_recording_days(self, channel: int, year: int, month: int) -> set[i if self.baichuan_only: return await self.baichuan.search_recording_days_bc(channel, year, month) - import calendar - last_day = calendar.monthrange(year, month)[1] start = datetime(year, month, 1, 0, 0, 0) end = datetime(year, month, last_day, 23, 59, 59) @@ -5935,9 +5934,7 @@ async def get_recordings_for_day( start = datetime(day.year, day.month, day.day, 0, 0, 0) end = datetime(day.year, day.month, day.day, 23, 59, 59) - _, vod_files = await self.request_vod_files( - channel, start, end, status_only=False, stream=stream, trigger=trigger - ) + _, vod_files = await self.request_vod_files(channel, start, end, status_only=False, stream=stream, trigger=trigger) # Deduplicate by file_name then sort chronologically seen: set[str] = set() @@ -5982,9 +5979,7 @@ def stream_recording_bc( async for microseconds, h264_bytes in host.stream_recording_bc(ch, name, start_time): ... """ - return self.baichuan.parse_bcmedia_frames( - self.baichuan.stream_replay_bc(channel, file_name, start_time, stream_type) - ) + return self.baichuan.parse_bcmedia_frames(self.baichuan.stream_replay_bc(channel, file_name, start_time, stream_type)) async def send_setting(self, body: typings.reolink_json, wait_before_get: int = 0, getcmd: str = "") -> None: command = body[0]["cmd"] diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index 60e7b3e3..d6d53704 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import calendar import logging import struct from collections.abc import AsyncIterator, Callable @@ -3618,9 +3619,7 @@ async def search_recording_days_bc(self, channel: int, year: int, month: int) -> Used as a fallback when baichuan_only=True (no HTTP API available). """ - import calendar as cal - - last_day = cal.monthrange(year, month)[1] + last_day = calendar.monthrange(year, month)[1] xml = xmls.DayRecords.format(year=year, month=month, last_day=last_day, channel=channel) mess = await self.send(cmd_id=142, channel=channel, body=xml) root = XML.fromstring(mess) @@ -3766,14 +3765,7 @@ async def _send_streaming(self, cmd_id: int, channel: int, body: str) -> tuple[a cmd_id_bytes = cmd_id.to_bytes(4, "little") mess_len_bytes = mess_len.to_bytes(4, "little") payload_offset_bytes = payload_offset.to_bytes(4, "little") - header = ( - bytes.fromhex(HEADER_MAGIC) - + cmd_id_bytes - + mess_len_bytes - + mess_id_bytes - + bytes.fromhex("0000" + "1464") - + payload_offset_bytes - ) + header = bytes.fromhex(HEADER_MAGIC) + cmd_id_bytes + mess_len_bytes + mess_id_bytes + bytes.fromhex("0000" + "1464") + payload_offset_bytes enc_body = self._aes_encrypt(ext) + self._aes_encrypt(body) await self._connect_if_needed() @@ -3846,7 +3838,7 @@ def _find_bcmedia_magic(data: bytearray, start: int) -> int | None: if magic_idx + 8 > len(buf): scan_from = magic_idx break - audio_payload_size = int.from_bytes(buf[magic_idx + 4: magic_idx + 6], "little") + audio_payload_size = int.from_bytes(buf[magic_idx + 4 : magic_idx + 6], "little") audio_pad = (8 - audio_payload_size % 8) % 8 audio_total = 8 + audio_payload_size + audio_pad if magic_idx + audio_total > len(buf): @@ -3862,9 +3854,9 @@ def _find_bcmedia_magic(data: bytearray, start: int) -> int | None: scan_from = magic_idx break - payload_size = int.from_bytes(buf[magic_idx + 8: magic_idx + 12], "little") - additional_header_size = int.from_bytes(buf[magic_idx + 12: magic_idx + 16], "little") - microseconds = int.from_bytes(buf[magic_idx + 16: magic_idx + 20], "little") + payload_size = int.from_bytes(buf[magic_idx + 8 : magic_idx + 12], "little") + additional_header_size = int.from_bytes(buf[magic_idx + 12 : magic_idx + 16], "little") + microseconds = int.from_bytes(buf[magic_idx + 16 : magic_idx + 20], "little") hdr_size = 24 + additional_header_size pad_size = (8 - payload_size % 8) % 8 @@ -3875,7 +3867,7 @@ def _find_bcmedia_magic(data: bytearray, start: int) -> int | None: scan_from = magic_idx break - h264_bytes = bytes(buf[magic_idx + hdr_size: magic_idx + hdr_size + payload_size]) + h264_bytes = bytes(buf[magic_idx + hdr_size : magic_idx + hdr_size + payload_size]) del buf[: magic_idx + total_size] scan_from = 0 @@ -3920,10 +3912,10 @@ async def stream_replay_bc( q, full_mess_id = await self._send_streaming(try_cmd_id, channel, body) try: status_code, data_chunk, len_hdr, payload = await asyncio.wait_for(q.get(), timeout=TIMEOUT) - except asyncio.TimeoutError: + except asyncio.TimeoutError as err: if self._protocol is not None: self._protocol.streaming_queues.pop((try_cmd_id, full_mess_id), None) - raise ReolinkTimeoutError(f"Baichuan host {self._host}: timeout waiting for replay start (MSG {try_cmd_id})") + raise ReolinkTimeoutError(f"Baichuan host {self._host}: timeout waiting for replay start (MSG {try_cmd_id})") from err if status_code == 400: if self._protocol is not None: @@ -4342,13 +4334,7 @@ def build_bcmedia_adpcm(adpcm_blocks: list[bytes]) -> bytes: data_len = len(block) payload_size = data_len + 4 pad_size = (8 - payload_size % 8) % 8 - payload += ( - Baichuan.BCMEDIA_ADPCM_MAGIC - + struct.pack(" None: From bc15f423ffd14737df971f2de7b9bd98c4cf07a5 Mon Sep 17 00:00:00 2001 From: lorek123 Date: Fri, 3 Apr 2026 00:59:15 +0200 Subject: [PATCH 06/11] Fix _aes_encrypt callers to pass bytes after upstream API change The upstream non-ASCII fix (bfc2234) changed _aes_encrypt() to accept bytes instead of str. Update _send_streaming() to encode ext/body as UTF-8 bytes and use byte lengths for mess_len/payload_offset, consistent with the existing send() method. Co-Authored-By: Claude Sonnet 4.6 --- reolink_aio/baichuan/baichuan.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index d6d53704..73148075 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -3754,9 +3754,11 @@ async def _send_streaming(self, cmd_id: int, channel: int, body: str) -> tuple[a ch_id = channel + 1 if channel is not None else 250 ext = xmls.CHANNEL_EXTENSION_XML.format(channel=channel) if channel is not None else "" + ext_bytes = ext.encode("utf-8") + body_bytes = body.encode("utf-8") - mess_len = len(ext) + len(body) - payload_offset = len(ext) + mess_len = len(ext_bytes) + len(body_bytes) + payload_offset = len(ext_bytes) self._mess_id = (self._mess_id + 1) % 16777216 mess_id_bytes = ch_id.to_bytes(1, "little") + self._mess_id.to_bytes(3, "little") @@ -3766,7 +3768,7 @@ async def _send_streaming(self, cmd_id: int, channel: int, body: str) -> tuple[a mess_len_bytes = mess_len.to_bytes(4, "little") payload_offset_bytes = payload_offset.to_bytes(4, "little") header = bytes.fromhex(HEADER_MAGIC) + cmd_id_bytes + mess_len_bytes + mess_id_bytes + bytes.fromhex("0000" + "1464") + payload_offset_bytes - enc_body = self._aes_encrypt(ext) + self._aes_encrypt(body) + enc_body = self._aes_encrypt(ext_bytes) + self._aes_encrypt(body_bytes) await self._connect_if_needed() if TYPE_CHECKING: From 87c85db33fc8e3d1e8bf7ba223383b59cfdec2e7 Mon Sep 17 00:00:00 2001 From: lorek123 Date: Fri, 3 Apr 2026 12:21:22 +0200 Subject: [PATCH 07/11] Fix snapshot_past: accept magic 1001/1002, leave thumbnails to integration - Extract _parse_cover_preview_frame() helper that accepts both b"1001" and b"1002" stream header magic (E1 cameras return 1002) - snapshot_past() now uses the helper, fixing the UnexpectedDataError bug - Remove public get_recording_thumbnail() API: CoverPreview on E1 returns only SPS/PPS headers (no decodable picture), so thumbnail generation is better handled by the HA integration using the streaming path Co-Authored-By: Claude Sonnet 4.6 --- reolink_aio/baichuan/baichuan.py | 90 ++++++++++++++------------------ 1 file changed, 40 insertions(+), 50 deletions(-) diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index 73148075..3c76e97b 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -2171,18 +2171,48 @@ async def snapshot(self, channel: int, iLogicChannel: int = 0, snapType: str = " return image + def _parse_cover_preview_frame(self, payload: bytes) -> bytes: + """Extract the raw H.264 I-frame bytes from a CoverPreview (cmd_id=298) payload. + + The payload starts with a 32-byte stream header (magic b"1001" or b"1002") + followed by one BcMedia video frame. Returns the raw H.264 Annex-B bytes + suitable for decoding with ffmpeg, PyAV, or any H.264 decoder. + """ + magic = payload[0:4] + if magic not in (b"1001", b"1002"): + raise UnexpectedDataError(f"Baichuan host {self._host}: CoverPreview payload has unexpected stream header magic {magic!r}") + + try: + start = payload[32:].index(b"00dc") + except ValueError as err: + raise UnexpectedDataError(f"Baichuan host {self._host}: CoverPreview frame magic b'00dc' not found, first bytes: {payload[32:62]!r}") from err + idx = 32 + start + + # Frame header: 24 bytes fixed + ah_size bytes extension + ah_size = int.from_bytes(payload[idx + 12 : idx + 16], byteorder="little") + header_len = 24 + ah_size + frame_len = int.from_bytes(payload[idx + 8 : idx + 12], byteorder="little") + frame_start = idx + header_len + + return payload[frame_start : frame_start + frame_len] + async def snapshot_past(self, channel: int, time: datetime, snapType: str = "sub", ffmpeg: str = "ffmpeg") -> bytes: """Get a JPEG image from a past recording (thumbnail)""" - end = time + timedelta(seconds=10) + frame = await self._fetch_cover_preview_frame(channel, time, stream=snapType) + return await i_frame_to_jpeg(frame, ffmpeg) + + async def _fetch_cover_preview_frame(self, channel: int, start_time: datetime, stream: str = "sub") -> bytes: + """Fetch a single H.264 frame via Baichuan CMD 298 (CoverPreview).""" + end = start_time + timedelta(seconds=10) xml = xmls.CoverPreview.format( channel=channel, - stream=snapType, - start_year=time.year, - start_month=time.month, - start_day=time.day, - start_hour=time.hour, - start_minute=time.minute, - start_second=time.second, + stream=stream, + start_year=start_time.year, + start_month=start_time.month, + start_day=start_time.day, + start_hour=start_time.hour, + start_minute=start_time.minute, + start_second=start_time.second, end_year=end.year, end_month=end.month, end_day=end.day, @@ -2190,48 +2220,8 @@ async def snapshot_past(self, channel: int, time: datetime, snapType: str = "sub end_minute=end.minute, end_second=end.second, ) - _mess, payload = await self.send_payload(cmd_id=298, body=xml) - - # parse stream header - stream_header = payload[0:32] - magic = stream_header[0:4] - # width = int.from_bytes(stream_header[8:12], byteorder="little") - # height = int.from_bytes(stream_header[12:16], byteorder="little") - # frame_rate = stream_header[17] - # start_year = 1900 + stream_header[18] - if magic != b"1001": - raise UnexpectedDataError(f"Baichuan host {self._host}: snapshot_past payload did not start with stream header magic b'1001' but with {magic!r}") - - # search magic - try: - # search magic - start = payload[32::].index(b"00dc") - except ValueError as err: - raise UnexpectedDataError(f"Baichuan host {self._host}: snapshot_past frame magic b'00dc' not found, first bytes: {payload[32:62]!r}") from err - idx = 32 + start - - # parse frame header - idx_start = idx + 12 - idx_end = idx_start + 4 - header_len = 24 + int.from_bytes(payload[idx_start:idx_end], byteorder="little") - idx_end = idx + header_len - frame_header = payload[idx:idx_end] - idx += header_len - # magic = frame_header[0:4] - # encoding = frame_header[4:8].decode("utf8") - frame_len = int.from_bytes(frame_header[8:12], byteorder="little") - # frame_time = int.from_bytes(frame_header[24:28], byteorder="little") - # frame_microsecond = int.from_bytes(frame_header[16:20], byteorder="little") - # formatted_time = datetime.fromtimestamp(frame_time).strftime("%Y-%m-%d %H:%M:%S") - - # extract frame - idx_end = idx + frame_len - frame = payload[idx:idx_end] - idx += frame_len - - image = await i_frame_to_jpeg(frame, ffmpeg) - - return image + _mess, payload = await self.send_payload(cmd_id=298, channel=channel, body=xml) + return self._parse_cover_preview_frame(payload) @http_cmd("GetP2p") async def get_uid(self) -> None: From 4fd3e568825f4a1fb649f3991df1d4993d90785f Mon Sep 17 00:00:00 2001 From: lorek123 Date: Sat, 4 Apr 2026 23:35:12 +0200 Subject: [PATCH 08/11] Fix FullAes replay decrypt: skip AES on packets without encryptLen The E1 camera sends replay packets in two flavors: packets with an explicit encryptLen in the Extension XML (partial AES region), and packets with just 1 but no encryptLen (payload is plaintext). The SDK treats absent encryptLen as 0xFFFFFFFF which cast to signed is -1, so `0 < (int)encryptLen` fails and decrypt is skipped. Previously we fell through to a probe/full-AES-decrypt heuristic on packets without encryptLen, producing garbled video (20 ffmpeg decode errors). Now we only decrypt when encryptLen is explicitly present and > 0, matching the SDK behavior. Verified on E1: 0 decode errors. Also stores the negotiated Baichuan crypto tier (Aes vs FullAes) from the login response status word, so Aes-tier cameras (like Argus 2) skip binary decrypt entirely per neolink PR #396. Made-with: Cursor --- reolink_aio/baichuan/baichuan.py | 72 ++++++++++++++++++++------------ 1 file changed, 45 insertions(+), 27 deletions(-) diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index 3c76e97b..b68862de 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -123,6 +123,8 @@ def __init__( self._user_hash: str | None = None self._password_hash: str | None = None self._aes_key: bytes | None = None + # Login rsp status (neolink codex): 0xDD12 = FullAes (encrypts replay/live binary); 0xDD02 etc. = Aes (binary plaintext). + self._baichuan_crypto_tier: Literal["aes", "full_aes"] | None = None self._log_once: set[str] = set() self._log_error: bool = True self.last_privacy_check: float = 0 @@ -369,6 +371,15 @@ async def send( _LOGGER.debug("%s, trying again", str(err_str)) return await self.send(cmd_id, channel, body, extension, enc_type, message_class, ch_id, mess_id, retry) + # Baichuan login (cmd_id 1): capture negotiated encryption tier from header status word (bytes 16–17, LE). + # High byte 0xDD, low 0x12 = FullAes; other 0xDD** = Aes-style (control encrypted, replay binary plaintext). + if cmd_id == 1 and len(data) >= 18: + status_u16 = int.from_bytes(data[16:18], "little") + if (status_u16 >> 8) & 0xFF == 0xDD: + low = status_u16 & 0xFF + self._baichuan_crypto_tier = "full_aes" if low == 0x12 else "aes" + _LOGGER.debug("Baichuan host %s: cmd_id 1 encryption tier %s (status 0x%04x)", self._host, self._baichuan_crypto_tier, status_u16) + # decryption rec_body = self._decrypt(data, len_header, cmd_id, enc_type) @@ -1372,6 +1383,7 @@ async def logout(self) -> None: self._protocol = None self._nonce = None self._aes_key = None + self._baichuan_crypto_tier = None self._user_hash = None self._password_hash = None @@ -3879,6 +3891,10 @@ async def stream_replay_bc( Sends ReplaySeek (MSG 123) first, then MSG 5 (falls back to MSG 8 on 400), skips the 32-byte replay header, and yields each subsequent binary packet. Sends ReplayStop (MSG 7) in cleanup. + + Replay binary handling follows the negotiated Baichuan login tier (header status on cmd_id 1): + FullAes (0xDD12) may encrypt payload (partial via Extension encryptLen, else full AES-CFB); + Aes (0xDD02, etc.) leaves binary as plaintext BcMedia. If tier was not parsed, defaults to FullAes. """ await self.replay_seek_bc(channel, start_time) @@ -3956,33 +3972,35 @@ async def stream_replay_bc( continue if payload and self._aes_key is not None: - # Packet has an extension section — try XML-based decryption first (encryptLen in Extension XML). - enc_pos = 0 - enc_len: int | None = None - xml_enc = data_chunk[len_hdr:] - if xml_enc: - try: - xml_str = self._aes_decrypt(xml_enc, data_chunk[:len_hdr]) - if xml_str.startswith("= 8: - probe = AES.new(key=self._aes_key, mode=AES.MODE_CFB, iv=AES_IV, segment_size=128).decrypt(payload[:8]) - if probe[:4] == b"00dc": - enc_len = 32 - enc_pos = 0 - - if enc_len and enc_len > 0: - end = enc_pos + enc_len - decrypted_region = AES.new(key=self._aes_key, mode=AES.MODE_CFB, iv=AES_IV, segment_size=128).decrypt(payload[enc_pos:end]) - binary = payload[:enc_pos] + decrypted_region + payload[end:] + # Aes (0xDD02): only control/XML is encrypted; binary BcMedia is plaintext — never AES-decrypt it + # (neolink PR #396 / de.rs). Default to full_aes when tier unknown so E1-style devices keep working. + # FullAes (0xDD12): binary may be partially encrypted per encryptLen/encryptPos from the Extension XML. + # SDK behavior (handleResponseV20): when encryptLen is absent from the Extension, it stays at + # init 0xFFFFFFFF; the check `0 < (int)encryptLen` fails (-1 signed) → decrypt is skipped → plaintext. + # Only decrypt when encryptLen is explicitly present and > 0. + tier = self._baichuan_crypto_tier or "full_aes" + if tier == "aes": + binary = payload + else: + enc_pos = 0 + enc_len: int | None = None + xml_enc = data_chunk[len_hdr:] + if xml_enc: + try: + xml_str = self._aes_decrypt(xml_enc, data_chunk[:len_hdr]) + if xml_str.startswith(" 0: + end = enc_pos + enc_len + decrypted_region = AES.new(key=self._aes_key, mode=AES.MODE_CFB, iv=AES_IV, segment_size=128).decrypt(payload[enc_pos:end]) + binary = payload[:enc_pos] + decrypted_region + payload[end:] + else: + binary = payload if packet_count <= 3 or packet_count % 100 == 0: _LOGGER.debug("Baichuan host %s: replay pkt %s: %s bytes", self._host, packet_count, len(binary)) From ce414b6c1c452735b0859368aec04b6abf42371b Mon Sep 17 00:00:00 2001 From: lorek123 Date: Fri, 10 Apr 2026 19:27:33 +0200 Subject: [PATCH 09/11] =?UTF-8?q?Remove=202-way=20audio=20=E2=80=94=20movi?= =?UTF-8?q?ng=20to=20fredrsat/reolink=5Faio#1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two-way audio (talk) is being contributed to the standalone PR at fredrsat/reolink_aio#1 which focuses on that feature. This keeps PR #164 focused on VOD replay streaming and recording browsing. Co-Authored-By: Claude Sonnet 4.6 --- reolink_aio/api.py | 48 -------- reolink_aio/baichuan/baichuan.py | 199 +------------------------------ reolink_aio/baichuan/xmls.py | 16 --- 3 files changed, 5 insertions(+), 258 deletions(-) diff --git a/reolink_aio/api.py b/reolink_aio/api.py index 163eea7d..f9dbf9e9 100644 --- a/reolink_aio/api.py +++ b/reolink_aio/api.py @@ -5368,54 +5368,6 @@ async def set_audio_alarm(self, channel: int, enable: bool) -> None: # Two-way audio (talk) public API # ------------------------------------------------------------------------- - def two_way_audio_support(self, channel: int) -> bool: - """Return True if the camera channel supports two-way audio (talk).""" - return self.supported(channel, "two_way_audio") - - async def start_talk(self, channel: int) -> dict: - """Start a two-way audio (talk) session on the camera channel. - - Returns a dict describing the audio format the camera expects: - - audio_type (str): codec name, e.g. "adpcm" - - sample_rate (int): samples per second, e.g. 8000 or 16000 - - sample_precision (int): bit depth, e.g. 16 - - length_per_encoder (int): samples per block, e.g. 320 or 640 - - sound_track (str): channel layout, e.g. "mono" - - duplex (str): duplex mode, e.g. "FDX" - - audio_stream_mode (str): e.g. "followVideoStream" - - The caller should encode microphone audio as IMA ADPCM blocks with the - parameters above, wrap each block using - ``Baichuan.build_bcmedia_adpcm([block])``, and send the result with - ``send_talk_data()``. Call ``stop_talk()`` when finished. - - Raises NotSupportedError if two-way audio is not supported. - """ - if channel not in self._channels: - raise InvalidParameterError(f"start_talk: no camera connected to channel '{channel}'") - if not self.two_way_audio_support(channel): - raise NotSupportedError(f"start_talk: Two-way audio is not supported on {self.camera_name(channel)}") - - return await self.baichuan.start_talk(channel) - - async def send_talk_data(self, channel: int, bcmedia_data: bytes) -> None: - """Send BcMedia-framed IMA ADPCM audio data to the camera. - - ``bcmedia_data`` must be produced by ``Baichuan.build_bcmedia_adpcm()``. - Call ``start_talk()`` before the first call to this method. - """ - if channel not in self._channels: - raise InvalidParameterError(f"send_talk_data: no camera connected to channel '{channel}'") - - await self.baichuan.send_talk_data(channel, bcmedia_data) - - async def stop_talk(self, channel: int) -> None: - """Stop the two-way audio (talk) session on the camera channel.""" - if channel not in self._channels: - raise InvalidParameterError(f"stop_talk: no camera connected to channel '{channel}'") - - await self.baichuan.stop_talk(channel) - async def set_siren(self, channel: int | None = None, enable: bool = True, duration: int | None = 2) -> None: if channel not in self._channels and channel is not None: raise InvalidParameterError(f"set_siren: no camera connected to channel '{channel}'") diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index b68862de..34bf635d 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -5,7 +5,6 @@ import asyncio import calendar import logging -import struct from collections.abc import AsyncIterator, Callable from datetime import date, datetime, timedelta from inspect import getmembers @@ -450,44 +449,6 @@ async def send_payload( return (rec_body, payload) - async def send_binary_no_reply( - self, - cmd_id: int, - channel: int | None = None, - binary_body: bytes = b"", - ) -> None: - """Send a binary payload without waiting for a reply (used for audio talk data).""" - if not self._logged_in and cmd_id > 2: - await self.login() - - if channel is None: - ch_id = 250 - else: - ch_id = channel + 1 - - ext = xmls.CHANNEL_BINARY_EXTENSION_XML.format(channel=channel) if channel is not None else "" - # Extension is AES-encrypted; binary body is sent as-is (NOT encrypted) - enc_ext = self._aes_encrypt(ext.encode("utf-8")) - mess_len = len(enc_ext) + len(binary_body) - payload_offset = len(enc_ext) - - self._mess_id = (self._mess_id + 1) % 16777216 - - cmd_id_bytes = (cmd_id).to_bytes(4, byteorder="little") - mess_len_bytes = (mess_len).to_bytes(4, byteorder="little") - mess_id_bytes = (ch_id).to_bytes(1, byteorder="little") + (self._mess_id).to_bytes(3, byteorder="little") - payload_offset_bytes = (payload_offset).to_bytes(4, byteorder="little") - status_code = "0000" - header = bytes.fromhex(HEADER_MAGIC) + cmd_id_bytes + mess_len_bytes + mess_id_bytes + bytes.fromhex(status_code + "1464") + payload_offset_bytes - - await self._connect_if_needed() - if TYPE_CHECKING: - assert self._transport is not None - - _LOGGER.debug("Baichuan host %s: writing binary no-reply cmd_id %s, binary length %s", self._host, cmd_id, len(binary_body)) - async with self._mutex: - self._transport.write(header + enc_ext + binary_body) - def _aes_encrypt(self, body: bytes) -> bytes: """Encrypt a message using AES encryption""" if not body: @@ -1520,7 +1481,7 @@ async def get_channel_data(self) -> None: if self.http_api.is_nvr and self.http_api.wifi_connection(channel) and (self.http_api.api_version("supportWiFi", channel) > 0 or self.http_api._is_hub): coroutines.append(("wifi", channel, self.get_wifi_signal(channel))) - if self.http_api.api_version("talk", channel) > 0 or self.http_api.baichuan_only: + if self.http_api.api_version("talk", channel) > 0: coroutines.append((10, channel, self.send(cmd_id=10, channel=channel))) if (self.http_api.is_nvr or self.privacy_mode() is not None) and self.api_version("remoteAbility", channel) > 0: @@ -1667,18 +1628,11 @@ async def get_channel_data(self) -> None: if isinstance(result, BaseException): raise result - if cmd_id == 10: # two way audio: check for duplex or mixAudioStream + if cmd_id == 10: # two way audio root = XML.fromstring(result) - if root.find(".//duplex") is not None: - self.capabilities[channel].add("two_way_audio") - else: - for audio in root.findall(".//audioStreamMode"): - if audio.text == "mixAudioStream": - self.capabilities[channel].add("two_way_audio") - if cmd_id == 201: # TalkAbility (cmd_id=201) – authoritative two_way_audio check - # result is already a dict returned by get_talk_ability() - if isinstance(result, dict) and result.get("duplex"): - self.capabilities[channel].add("two_way_audio") + for audio in root.findall(".//audioStreamMode"): + if audio.text == "mixAudioStream": + self.capabilities[channel].add("two_way_audio") if cmd_id == 483: # hardwired chime self.capabilities[channel].add("hardwired_chime") if cmd_id == 527: # crossline detection @@ -4212,146 +4166,3 @@ def siren_state(self, channel: int) -> bool | None: def audio_noise_reduction(self, channel: int) -> int | None: return self._noise_reduction.get(channel) - # ------------------------------------------------------------------------- - # Two-way audio (talk) support (confirmed via Ghidra / neolink) - # cmd_id 10 = TalkAbility (query, confirmed: contextGet(..., 10, 0)) - # cmd_id 11 = TalkReset (stop, confirmed: simpleSndFuncCH(..., 0xb, ...)) - # cmd_id 201 = TalkConfig (start, confirmed: xmlSndFuncCH(..., 0xc9, ...)) - # cmd_id 202 = Talk (send ADPCM audio data, no reply expected) - # ------------------------------------------------------------------------- - - BCMEDIA_ADPCM_MAGIC = struct.pack(" dict: - """Query the camera's talk (2-way audio) capability via cmd_id=10. - - Returns a dict with keys: - - duplex: str (e.g. "FDX") - - audio_stream_mode: str (e.g. "followVideoStream" or "mixAudioStream") - - audio_type: str (e.g. "adpcm") - - sample_rate: int (e.g. 8000 or 16000) - - sample_precision: int (e.g. 16) - - length_per_encoder: int (samples per block, e.g. 320 or 640) - - sound_track: str (e.g. "mono") - """ - mess = await self.send(cmd_id=10, channel=channel) - root = XML.fromstring(mess) - - ability: dict = {} - - # Duplex mode - for elem in root.findall(".//duplex"): - if elem.text: - ability["duplex"] = elem.text - break - - # Audio stream mode - for elem in root.findall(".//audioStreamMode"): - if elem.text: - ability["audio_stream_mode"] = elem.text - break - - # Audio config (take first supported entry) - for cfg in root.findall(".//audioConfig"): - audio_type_elem = cfg.find("audioType") - sample_rate_elem = cfg.find("sampleRate") - sample_precision_elem = cfg.find("samplePrecision") - lpe_elem = cfg.find("lengthPerEncoder") - sound_track_elem = cfg.find("soundTrack") - - if audio_type_elem is not None and audio_type_elem.text: - ability.setdefault("audio_type", audio_type_elem.text) - if sample_rate_elem is not None and sample_rate_elem.text: - ability.setdefault("sample_rate", int(sample_rate_elem.text)) - if sample_precision_elem is not None and sample_precision_elem.text: - ability.setdefault("sample_precision", int(sample_precision_elem.text)) - if lpe_elem is not None and lpe_elem.text: - ability.setdefault("length_per_encoder", int(lpe_elem.text)) - if sound_track_elem is not None and sound_track_elem.text: - ability.setdefault("sound_track", sound_track_elem.text) - - return ability - - async def start_talk(self, channel: int) -> dict: - """Start a 2-way audio talk session via Baichuan cmd_id=201 (TalkConfig). - - Queries the camera's TalkAbility, then sends a TalkConfig message to - initiate the session. Returns the audio configuration dict (same - structure as get_talk_ability) so the caller knows which format to use - when encoding audio for send_talk_data(). - - Audio format used: ADPCM (DVI-4 / IMA ADPCM), 16-bit, mono. - Block size in bytes = (length_per_encoder // 2) + 4 - """ - ability = await self.get_talk_ability(channel) - - duplex = ability.get("duplex", "FDX") - audio_stream_mode = ability.get("audio_stream_mode", "followVideoStream") - audio_type = ability.get("audio_type", "adpcm") - sample_rate = ability.get("sample_rate", 8000) - sample_precision = ability.get("sample_precision", 16) - length_per_encoder = ability.get("length_per_encoder", 320) - sound_track = ability.get("sound_track", "mono") - - xml = xmls.TalkConfigSet.format( - channel=channel, - duplex=duplex, - audio_stream_mode=audio_stream_mode, - audio_type=audio_type, - sample_rate=sample_rate, - sample_precision=sample_precision, - length_per_encoder=length_per_encoder, - sound_track=sound_track, - ) - await self.send(cmd_id=201, channel=channel, body=xml) - - _LOGGER.debug( - "Baichuan host %s ch %s: talk session started, audio_type=%s sample_rate=%s length_per_encoder=%s", - self._host, - channel, - audio_type, - sample_rate, - length_per_encoder, - ) - return ability - - async def stop_talk(self, channel: int) -> None: - """Stop the 2-way audio talk session via Baichuan cmd_id=11 (TalkReset).""" - await self.send(cmd_id=11, channel=channel) - _LOGGER.debug("Baichuan host %s ch %s: talk session stopped", self._host, channel) - - @staticmethod - def build_bcmedia_adpcm(adpcm_blocks: list[bytes]) -> bytes: - """Wrap one or more raw IMA-ADPCM blocks in BcMedia ADPCM framing. - - Each block must be a complete DVI-4 (IMA ADPCM) block: - - 4-byte block header (initial predictor LE-16, step index, reserved) - - followed by (length_per_encoder // 2) nibble-packed sample bytes. - - Up to 4 blocks may be combined into one BcMedia message. - The returned bytes can be passed directly to send_talk_data(). - """ - # BcMedia ADPCM frame layout (confirmed from pcap + Ghidra audioTalkSendStream): - # 4 bytes magic (0x62773130 = "bw10") - # 2 bytes payload_size = len(block) + 4 (LE u16, duplicated in next 2 bytes) - # 2 bytes payload_size (duplicate) - # 2 bytes sub-magic 0x0100 (MAGIC_HEADER_BCMEDIA_ADPCM_DATA) - # 2 bytes half_block = 2 (always 2; Ghidra: (sample_count>>1)+2 where sample_count=0) - # N bytes raw IMA ADPCM block (4-byte header + nibble data) - # P bytes zero padding to 8-byte boundary (based on payload_size; 0 for typical 520) - payload = b"" - for block in adpcm_blocks: - data_len = len(block) - payload_size = data_len + 4 - pad_size = (8 - payload_size % 8) % 8 - payload += Baichuan.BCMEDIA_ADPCM_MAGIC + struct.pack(" None: - """Send BcMedia-framed ADPCM audio data to the camera (cmd_id=202). - - bcmedia_data should be the output of build_bcmedia_adpcm(). - The audio payload is sent WITHOUT encryption as required by the protocol. - This method does not wait for a reply. - """ - await self.send_binary_no_reply(cmd_id=202, channel=channel, binary_body=bcmedia_data) diff --git a/reolink_aio/baichuan/xmls.py b/reolink_aio/baichuan/xmls.py index 166f9a91..b9f7ab5b 100644 --- a/reolink_aio/baichuan/xmls.py +++ b/reolink_aio/baichuan/xmls.py @@ -405,22 +405,6 @@ """ -TalkConfigSet = """ - - -{channel} -{duplex} -{audio_stream_mode} - -{audio_type} -{sample_rate} -{sample_precision} -{length_per_encoder} -{sound_track} - - -""" - SetAutoFocus = """ From e274ace95722f3ef640f1059a31da7cc210efb7d Mon Sep 17 00:00:00 2001 From: lorek123 Date: Sat, 11 Apr 2026 00:41:44 +0200 Subject: [PATCH 10/11] Adopt neolink replay improvements: H.265, MSG 0x17d, multi-stream, event search MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port the following protocol-level improvements from our neolink work: 1. H.265 support in parse_bcmedia_frames() — read video_type bytes [4:8] from BcMedia frame header ("H264" or "H265") and yield (microseconds, bytes, codec) instead of the previous (microseconds, bytes) 2-tuple. 2. NAL-level codec detection (_detect_codec_from_nal) — override the BcMedia header codec when firmware labels H.265 frames as "H264" (e.g. Argus PT). Detects H.265-exclusive VPS/SPS/PPS NAL types (0x40, 0x42, 0x44). 3. MSG 0x17d (desktop binary replay) fallback — when both MSG 5 and MSG 8 return 400, try the desktop protocol that sends a 0x944-byte binary payload (channel + file path). Adds _send_streaming_binary() for raw (unencrypted) payloads and _build_desktop_replay_payload() for the struct layout (neolink replay.rs). 4. Size-based stream end — parse the 32-byte replay header (bytes [16:24] and [24:32]) for expected file size and stop streaming when bytes received reach it. Needed for cameras (e.g. E1) that never send response 300/331. 5. Multi-stream listing — search_recordings_for_day_bc() now queries both mainStream and subStream when stream=None/default and merges results (deduplicating by file_name). Cameras like Argus 3 store separate files per stream; old behaviour returned only mainStream. 6. MSG 175 alarm/event search — new search_recordings_by_event_bc() sends findAlarmVideo with server-side alarm-type filter. Adds FindAlarmVideoOpen and FindAlarmVideoNext XML templates. Co-Authored-By: Claude Sonnet 4.6 --- reolink_aio/api.py | 10 +- reolink_aio/baichuan/baichuan.py | 357 ++++++++++++++++++++++++++++--- reolink_aio/baichuan/xmls.py | 33 +++ 3 files changed, 368 insertions(+), 32 deletions(-) diff --git a/reolink_aio/api.py b/reolink_aio/api.py index f9dbf9e9..82fb38f7 100644 --- a/reolink_aio/api.py +++ b/reolink_aio/api.py @@ -5906,12 +5906,14 @@ def stream_recording_bc( file_name: str, start_time: datetime, stream_type: str = "mainStream", - ) -> AsyncIterator[tuple[int, bytes]]: + ) -> AsyncIterator[tuple[int, bytes, str]]: """Async generator: stream a VOD recording via the Baichuan protocol. - Yields ``(microseconds, h264_bytes)`` tuples for each video frame. + Yields ``(microseconds, video_bytes, codec)`` tuples for each video frame. ``microseconds`` is the camera-relative timestamp (u32, wraps at ~71 min). - ``h264_bytes`` is the raw H.264 Annex-B NAL data for the frame. + ``video_bytes`` is the raw video NAL data for the frame. + ``codec`` is ``"H264"`` or ``"H265"`` — detected from the BcMedia header and + overridden by NAL-level analysis when a firmware bug causes mislabelling. Use this for baichuan_only cameras where HTTP download is unavailable. @@ -5928,7 +5930,7 @@ def stream_recording_bc( Usage:: - async for microseconds, h264_bytes in host.stream_recording_bc(ch, name, start_time): + async for microseconds, video_bytes, codec in host.stream_recording_bc(ch, name, start_time): ... """ return self.baichuan.parse_bcmedia_frames(self.baichuan.stream_replay_bc(channel, file_name, start_time, stream_type)) diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index 34bf635d..ad60401d 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -3588,22 +3588,11 @@ async def search_recording_days_bc(self, channel: int, year: int, month: int) -> days.add(1 + index) return days - async def search_recordings_for_day_bc(self, channel: int, day: date, stream: str | None = None) -> list[VOD_file]: - """Return all recordings for *day* on *channel* via Baichuan MSG 14 + MSG 15. - - Used as a fallback when baichuan_only=True (no HTTP API available). - """ + async def _search_recordings_for_day_stream_bc(self, channel: int, day: date, stream_type: str, stream_label: str) -> list[VOD_file]: + """Fetch recording file list for *day* using a single stream type via MSG 14 + MSG 15.""" uid = self.http_api.camera_uid(channel) uid = uid.split("_")[0] - if stream == "sub": - stream_type = "subStream" - elif stream in {"autotrack_sub", "telephoto_sub"}: - stream_type = "subStream" - else: - stream_type = "mainStream" - - # MSG 14: get file list handle for this day xml = xmls.FileInfoListOpen.format( uid=uid, channel=channel, @@ -3624,10 +3613,9 @@ async def search_recordings_for_day_bc(self, channel: int, day: date, stream: st mess = await self.send(cmd_id=14, channel=channel, body=xml) handle = self._get_value_from_xml(mess, "handle") if handle is None: - _LOGGER.debug("Baichuan host %s: search_recordings_for_day_bc: MSG 14 returned no handle for channel %s day %s", self._host, channel, day) + _LOGGER.debug("Baichuan host %s: search_recordings_for_day_bc: MSG 14 returned no handle for channel %s day %s stream %s", self._host, channel, day, stream_type) return [] - # MSG 15: fetch the file list using the handle xml = xmls.FileInfoList.format(channel=channel, uid=uid, handle=handle) mess = await self.send(cmd_id=15, channel=channel, body=xml) root = XML.fromstring(mess) @@ -3651,7 +3639,7 @@ async def search_recordings_for_day_bc(self, channel: int, day: date, stream: st size = (size_h or 0) * (2**32) + (size_l or 0) data: dict = { - "type": stream or self.http_api._stream, + "type": stream_label, "StartTime": datetime_to_reolink_time(start_dt), "EndTime": datetime_to_reolink_time(end_dt or start_dt), "PlaybackTime": datetime_to_reolink_time(start_dt), @@ -3684,6 +3672,169 @@ async def search_recordings_for_day_bc(self, channel: int, day: date, stream: st return vod_files + async def search_recordings_for_day_bc(self, channel: int, day: date, stream: str | None = None) -> list[VOD_file]: + """Return all recordings for *day* on *channel* via Baichuan MSG 14 + MSG 15. + + When *stream* is ``None`` or ``"main"``, both mainStream and subStream are queried + and their results are merged (deduplicated by name). On cameras like Argus 3 that + store mainStream and subStream as separate files with different names, this ensures + the caller sees all recordings regardless of which stream they want to replay. + + Used as a fallback when baichuan_only=True (no HTTP API available). + """ + if stream == "sub": + stream_type = "subStream" + stream_label = "sub" + elif stream in {"autotrack_sub", "telephoto_sub"}: + stream_type = "subStream" + stream_label = stream + elif stream in {"autotrack_main", "telephoto_main"}: + stream_type = "mainStream" + stream_label = stream + else: + # stream is None or "main": query both streams and merge + main_files = await self._search_recordings_for_day_stream_bc(channel, day, "mainStream", stream or self.http_api._stream) + sub_files = await self._search_recordings_for_day_stream_bc(channel, day, "subStream", "sub") + # Deduplicate by name; mainStream results take priority + seen: set[str] = {f.file_name for f in main_files} + for f in sub_files: + if f.file_name not in seen: + main_files.append(f) + seen.add(f.file_name) + return main_files + + return await self._search_recordings_for_day_stream_bc(channel, day, stream_type, stream_label) + + async def search_recordings_by_event_bc( + self, + channel: int, + start: datetime, + end: datetime, + alarm_types: list[str] | None = None, + stream: str | None = None, + ) -> list[VOD_file]: + """Search recordings by alarm/AI event type via Baichuan MSG 175 (findAlarmVideo). + + This is a server-side filter: the camera returns only files that match the requested + alarm types (e.g. "people", "vehicle", "md"). More efficient than fetching the full + file list and filtering client-side. + + *alarm_types* defaults to all types. *stream* selects mainStream (default) or subStream. + + Returns a list of :class:`~reolink_aio.typings.VOD_file` with ``bc_triggers`` set. + """ + if alarm_types is None: + alarm_types = ["md", "pir", "io", "people", "face", "vehicle", "dog_cat", "visitor", "other", "package", "cry", "crossline", "intrusion", "loitering", "legacy", "loss"] + + if stream in ("sub", "autotrack_sub", "telephoto_sub"): + stream_type = 1 + else: + stream_type = 0 + + alarm_type_str = ", ".join(alarm_types) + xml = xmls.FindAlarmVideoOpen.format( + channel=channel, + stream_type=stream_type, + alarm_type=alarm_type_str, + start_year=start.year, + start_month=start.month, + start_day=start.day, + start_hour=start.hour, + start_minute=start.minute, + start_second=start.second, + end_year=end.year, + end_month=end.month, + end_day=end.day, + end_hour=end.hour, + end_minute=end.minute, + end_second=end.second, + ) + + mess = await self.send(cmd_id=175, channel=channel, body=xml) + file_handle = self._get_value_from_xml(mess, "fileHandle") + + vod_files: list[VOD_file] = [] + request_i = 0 + while file_handle is not None: + request_i += 1 + if request_i > 50: + _LOGGER.warning("Baichuan host %s: search_recordings_by_event_bc took more than 50 iterations, stopping", self._host) + break + + xml = xmls.FindAlarmVideoNext.format(file_handle=file_handle) + mess = await self.send(cmd_id=175, channel=channel, body=xml) + root = XML.fromstring(mess) + main = root.find("alarmVideoInfo") or root.find("findAlarmVideo") + if main is None: + break + + b_finished = self._get_value_from_xml_element(main, "bFinished", int) + vod_list = main.find("alarmVideoList") + if vod_list is None: + break + + for item in vod_list.findall(".//alarmVideo"): + file_name = self._get_value_from_xml_element(item, "fileName") + trigger_str = self._get_value_from_xml_element(item, "alarmType") + if file_name is None: + continue + + start_time_el = item.find("startTime") + end_time_el = item.find("endTime") + time_event = self._xml_time_to_datetime(start_time_el) + end_time_event = self._xml_time_to_datetime(end_time_el) + if time_event is None or end_time_event is None: + continue + + data: dict = { + "type": stream or self.http_api._stream, + "StartTime": datetime_to_reolink_time(time_event), + "EndTime": datetime_to_reolink_time(end_time_event), + "PlaybackTime": datetime_to_reolink_time(time_event), + "name": file_name, + "size": "1", + } + vod_file = VOD_file(data, self.http_api.timezone()) + + triggers = VOD_trigger.NONE + t = trigger_str or "" + if "md" in t or "pir" in t or "other" in t: + triggers |= VOD_trigger.MOTION + if "io" in t: + triggers |= VOD_trigger.IO + if "people" in t: + triggers |= VOD_trigger.PERSON + if "face" in t: + triggers |= VOD_trigger.FACE + if "vehicle" in t: + triggers |= VOD_trigger.VEHICLE + if "dog_cat" in t: + triggers |= VOD_trigger.ANIMAL + if "visitor" in t: + triggers |= VOD_trigger.DOORBELL + if "package" in t: + triggers |= VOD_trigger.PACKAGE + if "cry" in t: + triggers |= VOD_trigger.CRYING + if "crossline" in t: + triggers |= VOD_trigger.CROSSLINE + if "intrusion" in t: + triggers |= VOD_trigger.INTRUSION + if "loitering" in t: + triggers |= VOD_trigger.LINGER + if "legacy" in t: + triggers |= VOD_trigger.FORGOTTEN_ITEM + if "loss" in t: + triggers |= VOD_trigger.TAKEN_ITEM + vod_file.bc_triggers = triggers + vod_files.append(vod_file) + + if b_finished != 0: + break + file_handle = self._get_value_from_xml_element(main, "fileHandle") + + return vod_files + async def replay_seek_bc(self, channel: int, seek_dt: datetime) -> None: """Send ReplaySeek (MSG 123) to prepare camera for VOD playback.""" seq = int(time_now()) @@ -3740,17 +3891,123 @@ async def _send_streaming(self, cmd_id: int, channel: int, body: str) -> tuple[a return q, full_mess_id + async def _send_streaming_binary(self, cmd_id: int, channel: int, binary_payload: bytes) -> tuple[asyncio.Queue, int]: + """Send *cmd_id* with a raw (unencrypted) binary payload and register a streaming queue. + + Used for desktop-style replay (MSG 0x17d) where the body is a packed binary struct, + not XML. No extension is sent; payload_offset = 0. + + Returns (queue, full_mess_id). The caller is responsible for removing + the queue from self._protocol.streaming_queues when done. + """ + if not self._logged_in and cmd_id > 2: + await self.login() + + ch_id = channel + 1 if channel is not None else 250 + mess_len = len(binary_payload) + payload_offset = 0 + + self._mess_id = (self._mess_id + 1) % 16777216 + mess_id_bytes = ch_id.to_bytes(1, "little") + self._mess_id.to_bytes(3, "little") + full_mess_id = int.from_bytes(mess_id_bytes, "little") + + cmd_id_bytes = cmd_id.to_bytes(4, "little") + mess_len_bytes = mess_len.to_bytes(4, "little") + payload_offset_bytes = payload_offset.to_bytes(4, "little") + header = bytes.fromhex(HEADER_MAGIC) + cmd_id_bytes + mess_len_bytes + mess_id_bytes + bytes.fromhex("0000" + "1464") + payload_offset_bytes + + await self._connect_if_needed() + if TYPE_CHECKING: + assert self._protocol is not None + assert self._transport is not None + + q: asyncio.Queue = asyncio.Queue(maxsize=512) + self._protocol.streaming_queues[(cmd_id, full_mess_id)] = q + + _LOGGER.debug("Baichuan host %s: streaming send cmd_id 0x%x full_mess_id %s (binary payload %s bytes)", self._host, cmd_id, full_mess_id, mess_len) + async with self._mutex: + self._transport.write(header + binary_payload) + + return q, full_mess_id + + @staticmethod + def _build_desktop_replay_payload(channel: int, file_name: str) -> bytes: + """Build the 0x944-byte binary payload for desktop-style replay (MSG 0x17d). + + Layout (from Ghidra FUN_180177b80 / neolink replay.rs): + [0:8] inner header u64 = 2 + [8:12] inner header u32 = 0x82f + [12:16] inner header u32 = 8 + [16:20] inner header u32 = 500 + [20:24] channel (u32 LE) + [24:56] 32 zero bytes + [56:1079] file path, null-padded (max 1023 bytes) + rest: zero + """ + import struct as _struct + PAYLOAD_LEN = 0x944 + PATH_OFFSET = 20 + 4 + 32 # inner_header(20) + channel(4) + zeros(32) + PATH_MAX = 1023 + out = bytearray(PAYLOAD_LEN) + # Inner 20-byte header + _struct.pack_into(" str | None: + """Detect the actual video codec by inspecting the first NAL unit bytes. + + Some cameras (e.g. Argus PT) send H.265 video but label it "H264" in the BcMedia + header — a firmware bug. We check for H.265-exclusive NAL types (VPS=32, SPS=33, + PPS=34 — byte[0] values 0x40, 0x42, 0x44) to catch the mislabelling. + + Returns "H264", "H265", or None (insufficient data / ambiguous). + """ + if len(data) < 5: + return None + # Skip Annex B start code if present + offset = 0 + if data[0:4] == b"\x00\x00\x00\x01": + offset = 4 + elif data[0:3] == b"\x00\x00\x01": + offset = 3 + if offset >= len(data): + return None + first_byte = data[offset] + # H.265 NAL: nal_unit_type = (byte >> 1) & 0x3F + h265_nal_type = (first_byte >> 1) & 0x3F + if h265_nal_type in (32, 33, 34, 35, 39): # VPS, SPS, PPS, AUD, SEI prefix + return "H265" + # H.264 NAL: nal_unit_type = byte & 0x1F + h264_nal_type = first_byte & 0x1F + if h264_nal_type in (1, 5, 6, 7, 8, 9): # Slice, IDR, SEI, SPS, PPS, AUD + return "H264" + return None + @staticmethod async def parse_bcmedia_frames( raw_stream: AsyncIterator[bytes], - ) -> AsyncIterator[tuple[int, bytes]]: - """Parse a raw Reolink BcMedia byte stream and yield (microseconds, h264_bytes) per video frame. + ) -> AsyncIterator[tuple[int, bytes, str]]: + """Parse a raw Reolink BcMedia byte stream and yield (microseconds, video_bytes, codec) per video frame. + + *codec* is ``"H264"`` or ``"H265"``; the value comes from the BcMedia header and is + overridden by NAL-level detection when the header disagrees with the actual payload + (some cameras, e.g. Argus PT, label H.265 frames as "H264"). BcMedia magic layout (4 bytes, LE): byte[0]: frame counter, '0'-'9' (0x30-0x39), cycles 0-9 byte[1]: frame type, '0'=IFrame (0x30) or '1'=PFrame (0x31) byte[2]: 'd' (0x64) byte[3]: 'c' (0x63) + Bytes [4:8] of the frame header encode the video codec as ASCII "H264" or "H265". Audio and info frames (different magic) are skipped. """ buf = bytearray() @@ -3766,7 +4023,7 @@ def _find_bcmedia_magic(data: bytearray, start: int) -> int | None: BcMedia frames are always 8-byte padded so all frame starts lie at 8-byte-aligned stream offsets. Scanning only aligned positions avoids false positives that can - appear inside H.264 or encrypted-audio payloads. + appear inside H.264/H.265 or encrypted-audio payloads. """ i = (start + 7) & ~7 if start % 8 != 0 else start while i + 4 <= len(data): @@ -3812,6 +4069,12 @@ def _find_bcmedia_magic(data: bytearray, start: int) -> int | None: scan_from = magic_idx break + # Bytes [4:8]: video type ASCII string ("H264" or "H265") + video_type_raw = bytes(buf[magic_idx + 4 : magic_idx + 8]) + codec = video_type_raw.decode("ascii", errors="replace").rstrip("\x00") + if codec not in ("H264", "H265"): + codec = "H264" # safe fallback + payload_size = int.from_bytes(buf[magic_idx + 8 : magic_idx + 12], "little") additional_header_size = int.from_bytes(buf[magic_idx + 12 : magic_idx + 16], "little") microseconds = int.from_bytes(buf[magic_idx + 16 : magic_idx + 20], "little") @@ -3825,12 +4088,16 @@ def _find_bcmedia_magic(data: bytearray, start: int) -> int | None: scan_from = magic_idx break - h264_bytes = bytes(buf[magic_idx + hdr_size : magic_idx + hdr_size + payload_size]) + video_bytes = bytes(buf[magic_idx + hdr_size : magic_idx + hdr_size + payload_size]) del buf[: magic_idx + total_size] scan_from = 0 - if h264_bytes: - yield microseconds, h264_bytes + if video_bytes: + # Override codec from NAL bytes when header disagrees (firmware bug on some cameras) + detected = Baichuan._detect_codec_from_nal(video_bytes) + if detected is not None and detected != codec: + codec = detected + yield microseconds, video_bytes, codec async def stream_replay_bc( self, @@ -3870,8 +4137,18 @@ async def stream_replay_bc( full_mess_id: int = 0 q: asyncio.Queue | None = None - for try_cmd_id in (5, 8): - q, full_mess_id = await self._send_streaming(try_cmd_id, channel, body) + # Try MSG 5 → MSG 8 → MSG 0x17d (desktop binary replay). + # MSG 0x17d sends a 0x944-byte binary payload instead of XML; used by the desktop app + # for cameras that reject both MSG 5 and MSG 8 (Ghidra ref: FUN_180177b80). + _DESKTOP_CMD_ID = 0x17d + tried_cmd_ids: list[int] = [] + for try_cmd_id in (5, 8, _DESKTOP_CMD_ID): + if try_cmd_id == _DESKTOP_CMD_ID: + desktop_payload = self._build_desktop_replay_payload(channel, file_name) + q, full_mess_id = await self._send_streaming_binary(_DESKTOP_CMD_ID, channel, desktop_payload) + else: + q, full_mess_id = await self._send_streaming(try_cmd_id, channel, body) + tried_cmd_ids.append(try_cmd_id) try: status_code, data_chunk, len_hdr, payload = await asyncio.wait_for(q.get(), timeout=TIMEOUT) except asyncio.TimeoutError as err: @@ -3882,9 +4159,9 @@ async def stream_replay_bc( if status_code == 400: if self._protocol is not None: self._protocol.streaming_queues.pop((try_cmd_id, full_mess_id), None) - if try_cmd_id == 8: - raise ReolinkError(f"Baichuan host {self._host}: replay start rejected by camera (MSG 5 and MSG 8 both returned 400)") - _LOGGER.debug("Baichuan host %s: replay MSG 5 rejected (400), trying MSG 8", self._host) + if try_cmd_id == _DESKTOP_CMD_ID: + raise ReolinkError(f"Baichuan host {self._host}: replay start rejected by camera (MSG 5, MSG 8, and MSG 0x17d all returned 400)") + _LOGGER.debug("Baichuan host %s: replay MSG %s rejected (400), trying next", self._host, try_cmd_id) continue if status_code != 200: @@ -3900,6 +4177,8 @@ async def stream_replay_bc( STREAM_TIMEOUT = 15.0 packet_count = 0 + total_payload_bytes = 0 + expected_payload_size: int | None = None # from 32-byte header or file list try: while True: try: @@ -3920,9 +4199,22 @@ async def stream_replay_bc( if not binary: continue - # First packet: 32-byte replay header — skip it + # First packet: 32-byte replay header — skip it and extract expected file size. + # The 32-byte header layout (RE: BaichuanDownloader): bytes [16:24] and [24:32] may + # contain the file size as u64 LE. We use this to stop when the camera never sends + # response 300/331 (common on E1 and some newer models). if packet_count == 1 and len(binary) == 32: _LOGGER.debug("Baichuan host %s: replay: skipping 32-byte stream header", self._host) + if expected_payload_size is None: + size_at_10 = int.from_bytes(binary[16:24], "little") + size_at_18 = int.from_bytes(binary[24:32], "little") + MAX_PLAUSIBLE = 500_000_000 + if 0 < size_at_10 <= MAX_PLAUSIBLE: + expected_payload_size = size_at_10 + _LOGGER.debug("Baichuan host %s: replay: expected size from header +0x10: %s bytes", self._host, size_at_10) + elif 0 < size_at_18 <= MAX_PLAUSIBLE: + expected_payload_size = size_at_18 + _LOGGER.debug("Baichuan host %s: replay: expected size from header +0x18: %s bytes", self._host, size_at_18) continue if payload and self._aes_key is not None: @@ -3956,11 +4248,20 @@ async def stream_replay_bc( else: binary = payload + total_payload_bytes += len(binary) if packet_count <= 3 or packet_count % 100 == 0: _LOGGER.debug("Baichuan host %s: replay pkt %s: %s bytes", self._host, packet_count, len(binary)) yield binary + # Size-based end: some cameras never send 300/331; stop when we reach the expected size. + if expected_payload_size is not None and total_payload_bytes >= expected_payload_size: + _LOGGER.debug( + "Baichuan host %s: replay: received %s bytes (expected %s), stopping", + self._host, total_payload_bytes, expected_payload_size, + ) + break + finally: if self._protocol is not None: self._protocol.streaming_queues.pop((accepted_cmd_id, full_mess_id), None) diff --git a/reolink_aio/baichuan/xmls.py b/reolink_aio/baichuan/xmls.py index b9f7ab5b..a36083ed 100644 --- a/reolink_aio/baichuan/xmls.py +++ b/reolink_aio/baichuan/xmls.py @@ -561,3 +561,36 @@ """ + +FindAlarmVideoOpen = """ + + +{channel} +{stream_type} +0 + +{start_year} +{start_month} +{start_day} +{start_hour} +{start_minute} +{start_second} + + +{end_year} +{end_month} +{end_day} +{end_hour} +{end_minute} +{end_second} + +{alarm_type} + +""" + +FindAlarmVideoNext = """ + + +{file_handle} + +""" From ed8700dde89c43db0f5a0ced472ca7d1d55cf1af Mon Sep 17 00:00:00 2001 From: lorek123 Date: Tue, 26 May 2026 17:41:53 +0200 Subject: [PATCH 11/11] Add Baichuan live stream (stream_live_bc) via MSG 3/4 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements host.stream_live_bc(channel, stream_type) — an async generator yielding (microseconds, video_bytes, codec) tuples for live H264/H265 frames via the Baichuan Preview/VideoStart protocol (MSG 3 start, MSG 4 stop). Applies the same FullAes AES-CFB decryption logic as stream_replay_bc for cameras that encrypt the first chunk of each streaming packet (e.g. E1). Tested on Reolink E1 (baichuan_only): 2304×1296 H264, valid NALs confirmed via ffprobe. Co-Authored-By: Claude Sonnet 4.6 --- reolink_aio/api.py | 29 ++++++++ reolink_aio/baichuan/baichuan.py | 120 ++++++++++++++++++++++++++++--- reolink_aio/baichuan/xmls.py | 17 +++++ 3 files changed, 157 insertions(+), 9 deletions(-) diff --git a/reolink_aio/api.py b/reolink_aio/api.py index 82fb38f7..41442fad 100644 --- a/reolink_aio/api.py +++ b/reolink_aio/api.py @@ -5935,6 +5935,35 @@ def stream_recording_bc( """ return self.baichuan.parse_bcmedia_frames(self.baichuan.stream_replay_bc(channel, file_name, start_time, stream_type)) + def stream_live_bc( + self, + channel: int, + stream_type: str = "mainStream", + ) -> AsyncIterator[tuple[int, bytes, str]]: + """Async generator: stream live video via the Baichuan protocol. + + Yields ``(microseconds, video_bytes, codec)`` tuples for each video frame. + ``microseconds`` is the camera-relative timestamp (u32, wraps at ~71 min). + ``video_bytes`` is the raw video NAL data. + ``codec`` is ``"H264"`` or ``"H265"``. + + Break out of the loop to stop the stream — the PreviewStop command is sent + automatically on generator close. + + Parameters + ---------- + channel: + Camera channel index. + stream_type: + ``"mainStream"`` (default) or ``"subStream"``. + + Usage:: + + async for microseconds, video_bytes, codec in host.stream_live_bc(ch): + ... + """ + return self.baichuan.parse_bcmedia_frames(self.baichuan.stream_live_bc(channel, stream_type)) + async def send_setting(self, body: typings.reolink_json, wait_before_get: int = 0, getcmd: str = "") -> None: command = body[0]["cmd"] _LOGGER.debug( diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index ad60401d..95f8dbbd 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -3724,7 +3724,24 @@ async def search_recordings_by_event_bc( Returns a list of :class:`~reolink_aio.typings.VOD_file` with ``bc_triggers`` set. """ if alarm_types is None: - alarm_types = ["md", "pir", "io", "people", "face", "vehicle", "dog_cat", "visitor", "other", "package", "cry", "crossline", "intrusion", "loitering", "legacy", "loss"] + alarm_types = [ + "md", + "pir", + "io", + "people", + "face", + "vehicle", + "dog_cat", + "visitor", + "other", + "package", + "cry", + "crossline", + "intrusion", + "loitering", + "legacy", + "loss", + ] if stream in ("sub", "autotrack_sub", "telephoto_sub"): stream_type = 1 @@ -3945,20 +3962,21 @@ def _build_desktop_replay_payload(channel: int, file_name: str) -> bytes: rest: zero """ import struct as _struct + PAYLOAD_LEN = 0x944 PATH_OFFSET = 20 + 4 + 32 # inner_header(20) + channel(4) + zeros(32) PATH_MAX = 1023 out = bytearray(PAYLOAD_LEN) # Inner 20-byte header - _struct.pack_into("= expected_payload_size: _LOGGER.debug( "Baichuan host %s: replay: received %s bytes (expected %s), stopping", - self._host, total_payload_bytes, expected_payload_size, + self._host, + total_payload_bytes, + expected_payload_size, ) break @@ -4272,6 +4292,89 @@ async def stream_replay_bc( except Exception as err: _LOGGER.debug("Baichuan host %s: replay stop (MSG 7) error: %s", self._host, err) + async def stream_live_bc( + self, + channel: int, + stream_type: str = "mainStream", + ) -> AsyncIterator[bytes]: + """Async generator: stream live video via Baichuan MSG 3 (Preview/VideoStart). + + Yields raw BcMedia binary chunks — pass to parse_bcmedia_frames() for decoded + (microseconds, video_bytes, codec) tuples. + + Sends Preview start (MSG 3) then streams until the caller closes the generator. + Sends PreviewStop (MSG 4) on exit. + + Parameters + ---------- + channel: + Camera channel index. + stream_type: + ``"mainStream"`` (default), ``"subStream"``, or ``"externStream"``. + """ + handle = {"mainStream": 0, "subStream": 256, "externStream": 1024}.get(stream_type, 0) + body = xmls.Preview.format(channel=channel, handle=handle, stream_type=stream_type) + q, full_mess_id = await self._send_streaming(3, channel, body) + + try: + try: + status_code, data_chunk, len_hdr, payload = await asyncio.wait_for(q.get(), timeout=TIMEOUT) + except asyncio.TimeoutError as err: + if self._protocol is not None: + self._protocol.streaming_queues.pop((3, full_mess_id), None) + raise ReolinkTimeoutError(f"Baichuan host {self._host}: timeout waiting for live stream start (MSG 3)") from err + + if status_code != 200: + if self._protocol is not None: + self._protocol.streaming_queues.pop((3, full_mess_id), None) + raise ReolinkError(f"Baichuan host {self._host}: live stream start rejected (status {status_code}, MSG 3)") + + LIVE_TIMEOUT = 15.0 + while True: + try: + status_code, data_chunk, len_hdr, payload = await asyncio.wait_for(q.get(), timeout=LIVE_TIMEOUT) + except asyncio.TimeoutError: + _LOGGER.debug("Baichuan host %s: live stream: no data for %.0fs, stopping", self._host, LIVE_TIMEOUT) + break + + if payload and self._aes_key: + tier = self._baichuan_crypto_tier or "full_aes" + if tier == "aes": + binary = payload + else: + enc_pos = 0 + enc_len: int | None = None + xml_enc = data_chunk[len_hdr:] + if xml_enc: + try: + xml_str = self._aes_decrypt(xml_enc, data_chunk[:len_hdr]) + if xml_str.startswith(" 0: + end = enc_pos + enc_len + decrypted_region = AES.new(key=self._aes_key, mode=AES.MODE_CFB, iv=AES_IV, segment_size=128).decrypt(payload[enc_pos:end]) + binary = payload[:enc_pos] + decrypted_region + payload[end:] + else: + binary = payload + else: + binary = payload if payload else data_chunk[len_hdr:] + + if binary: + yield binary + finally: + if self._protocol is not None: + self._protocol.streaming_queues.pop((3, full_mess_id), None) + stop_body = xmls.PreviewStop.format(channel=channel, handle=handle) + try: + await self.send(cmd_id=4, channel=channel, body=stop_body) + _LOGGER.debug("Baichuan host %s: live stream stop (MSG 4) sent", self._host) + except Exception as err: + _LOGGER.debug("Baichuan host %s: live stream stop (MSG 4) error: %s", self._host, err) + @property def events_active(self) -> bool: return self._events_active and time_now() - self._time_connection_lost > 120 @@ -4466,4 +4569,3 @@ def siren_state(self, channel: int) -> bool | None: def audio_noise_reduction(self, channel: int) -> int | None: return self._noise_reduction.get(channel) - diff --git a/reolink_aio/baichuan/xmls.py b/reolink_aio/baichuan/xmls.py index a36083ed..827111ad 100644 --- a/reolink_aio/baichuan/xmls.py +++ b/reolink_aio/baichuan/xmls.py @@ -464,6 +464,23 @@ """ +Preview = """ + + +{channel} +{handle} +{stream_type} + +""" + +PreviewStop = """ + + +{channel} +{handle} + +""" + PtzControl = """