From fc557a682c346d2c9e5da3bca61cb6133dcd1667 Mon Sep 17 00:00:00 2001 From: fredrsat Date: Fri, 3 Apr 2026 12:59:09 +0200 Subject: [PATCH 1/2] Add two-way audio (talk) support via Baichuan protocol Implement sending audio to camera speakers using the Baichuan TCP protocol (cmd_ids 10, 11, 201, 202). This addresses the long-standing request for two-way audio support (#161). New files: - baichuan/audio.py: Pure-Python IMA ADPCM (DVI-4) encoder and BcMedia frame builder. No external dependencies (audioop was removed in 3.13). - tests/test_talk.py: 30 unit tests covering ADPCM encoding, BcMedia framing, talk frame header construction, and talk() orchestration. Modified files: - baichuan/xmls.py: Add BINARY_EXTENSION_XML (with binaryData flag for binary payloads) and TALK_CONFIG_XML templates. - baichuan/baichuan.py: - Parse and cache audio config from TalkAbility (cmd_id=10) response - Add _send_talk_frame(): fire-and-forget method that AES-encrypts the extension XML but sends the audio payload as raw binary - Add talk(): public method that starts a talk session (TalkConfig), encodes PCM to ADPCM, sends BcMedia frames paced to audio duration, and resets the session (TalkReset). Handles 422 (busy) gracefully. Protocol details: - Audio format: IMA ADPCM, 16-bit, mono, 8000/16000 Hz (from TalkAbility) - cmd_id 202 uses mixed encryption: AES-encrypted extension + raw payload - BcMedia frames use "01wb" magic with 8-byte aligned padding - Up to 4 ADPCM blocks packed per Baichuan message for efficiency Tested against Reolink Home Hub with Argus 4 Pro cameras. --- reolink_aio/baichuan/audio.py | 188 +++++++++++++++ reolink_aio/baichuan/baichuan.py | 149 ++++++++++++ reolink_aio/baichuan/xmls.py | 24 ++ tests/test_talk.py | 392 +++++++++++++++++++++++++++++++ 4 files changed, 753 insertions(+) create mode 100644 reolink_aio/baichuan/audio.py create mode 100644 tests/test_talk.py diff --git a/reolink_aio/baichuan/audio.py b/reolink_aio/baichuan/audio.py new file mode 100644 index 00000000..27de8e9b --- /dev/null +++ b/reolink_aio/baichuan/audio.py @@ -0,0 +1,188 @@ +"""IMA ADPCM encoder and BcMedia frame builder for Reolink two-way audio. + +The Reolink Baichuan protocol uses IMA ADPCM (DVI-4) encoding for two-way +audio (talk). Audio frames are wrapped in BcMedia headers before transmission. + +Reference: neolink project (QuantumEntangledAndy/neolink) bcmedia module. +""" + +from __future__ import annotations + +import struct + +# Standard IMA ADPCM step size table (89 entries, indexed 0-88) +_STEP_TABLE: list[int] = [ + 7, 8, 9, 10, 11, 12, 13, 14, 16, 17, + 19, 21, 23, 25, 28, 31, 34, 37, 41, 45, + 50, 55, 60, 66, 73, 80, 88, 97, 107, 118, + 130, 143, 157, 173, 190, 209, 230, 253, 279, 307, + 337, 371, 408, 449, 494, 544, 598, 658, 724, 796, + 876, 963, 1060, 1166, 1282, 1411, 1552, 1707, 1878, 2066, + 2272, 2499, 2749, 3024, 3327, 3660, 4026, 4428, 4871, 5358, + 5894, 6484, 7132, 7845, 8630, 9493, 10442, 11487, 12635, 13899, + 15289, 16818, 18500, 20350, 22385, 24623, 27086, 29794, 32767, +] + +# Standard IMA ADPCM index adjustment table (16 entries, indexed by nibble 0-15) +_INDEX_TABLE: list[int] = [ + -1, -1, -1, -1, 2, 4, 6, 8, + -1, -1, -1, -1, 2, 4, 6, 8, +] + +# BcMedia ADPCM magic: ASCII "01wb" (little-endian u32 = 0x62773130) +BC_MEDIA_ADPCM_MAGIC = b"01wb" + +# Sub-header magic for ADPCM frames +_BC_MEDIA_SUB_MAGIC = 0x0001 + + +def encode_pcm_to_adpcm( + pcm_data: bytes, + samples_per_block: int = 1024, +) -> list[bytes]: + """Convert 16-bit signed LE mono PCM to IMA ADPCM blocks. + + Each output block consists of a 4-byte preamble followed by nibble-packed + ADPCM data. The preamble stores the encoder state (predictor and step + index) at the start of the block. + + Block format:: + + Offset Size Field + 0x00 2 predictor (i16 LE) — initial predicted sample value + 0x02 1 step_index (u8) — ADPCM step table index (0–88) + 0x03 1 reserved (u8) — always 0 + 0x04 N nibble data — packed low-nibble-first + + Args: + pcm_data: Raw PCM audio (16-bit signed little-endian, mono). + samples_per_block: PCM samples per ADPCM block. This corresponds to + the ``lengthPerEncoder`` value from the camera's TalkAbility + response. Typical values: 1024 (8 kHz) or 2048 (16 kHz). + + Returns: + List of ADPCM blocks, each ``4 + samples_per_block // 2`` bytes. + """ + num_samples = len(pcm_data) // 2 + if num_samples == 0: + return [] + + samples = struct.unpack(f"<{num_samples}h", pcm_data[: num_samples * 2]) + + blocks: list[bytes] = [] + predictor = 0 + step_index = 0 + + for block_start in range(0, num_samples, samples_per_block): + block_samples = samples[block_start : block_start + samples_per_block] + if not block_samples: + break + + # Preamble: current encoder state before encoding this block + preamble = struct.pack("= threshold: + code |= 4 + diff -= threshold + threshold >>= 1 + if diff >= threshold: + code |= 2 + diff -= threshold + threshold >>= 1 + if diff >= threshold: + code |= 1 + + code |= sign + nibbles.append(code) + + # Decode to update predictor (must match decoder exactly) + decoded_diff = step >> 3 + if code & 4: + decoded_diff += step + if code & 2: + decoded_diff += step >> 1 + if code & 1: + decoded_diff += step >> 2 + + if code & 8: + predictor -= decoded_diff + else: + predictor += decoded_diff + + # Clamp predictor to 16-bit signed range + if predictor > 32767: + predictor = 32767 + elif predictor < -32768: + predictor = -32768 + + # Update step index + step_index += _INDEX_TABLE[code] + if step_index > 88: + step_index = 88 + elif step_index < 0: + step_index = 0 + + # Pack nibbles into bytes (low nibble first in each byte) + data = bytearray() + for i in range(0, len(nibbles), 2): + low = nibbles[i] + high = nibbles[i + 1] if i + 1 < len(nibbles) else 0 + data.append((high << 4) | low) + + blocks.append(preamble + bytes(data)) + + return blocks + + +def build_bc_media_frame(adpcm_block: bytes) -> bytes: + """Wrap one ADPCM block in a BcMedia frame with 8-byte alignment padding. + + Frame layout:: + + Offset Size Field + 0x00 4 magic — b"01wb" (0x62773130 LE) + 0x04 2 payload_size (LE u16) — len(data) + 4 + 0x06 2 payload_size (LE u16) — duplicate + 0x08 2 sub_magic (LE u16) — 0x0001 + 0x0A 2 half_block_size (LE u16) — (len(data) - 4) / 2 + 0x0C N ADPCM data — preamble + encoded nibbles + 0x0C+N P zero padding to 8-byte alignment + + Args: + adpcm_block: Complete ADPCM block (4-byte preamble + encoded data). + + Returns: + BcMedia frame bytes ready for wire transmission. + """ + data_len = len(adpcm_block) + payload_size = data_len + 4 # sub_magic (2) + half_block_size (2) + half_block = (data_len - 4) // 2 # ADPCM nibble bytes / 2 + + header = struct.pack( + "<4sHHHH", + BC_MEDIA_ADPCM_MAGIC, + payload_size, + payload_size, + _BC_MEDIA_SUB_MAGIC, + half_block, + ) + + # Pad data to 8-byte boundary (per neolink serializer convention) + padding_len = (8 - (data_len % 8)) % 8 + + return header + adpcm_block + (b"\x00" * padding_len) diff --git a/reolink_aio/baichuan/baichuan.py b/reolink_aio/baichuan/baichuan.py index 5e926a34..4801290c 100644 --- a/reolink_aio/baichuan/baichuan.py +++ b/reolink_aio/baichuan/baichuan.py @@ -179,6 +179,7 @@ def __init__( self._siren_state: dict[int, bool] = {} self._siren_play_time: dict[int | None, float] = {} self._noise_reduction: dict[int, int] = {} + self._talk_config: dict[int, dict] = {} self._ai_yolo_600: dict[int, dict[str, bool]] = {} self._ai_yolo_696: dict[int, dict[str, bool]] = {} self._ai_yolo_sub_type: dict[int, dict[str, str | None]] = {} @@ -437,6 +438,58 @@ async def send_payload( return (rec_body, payload) + async def _send_talk_frame(self, channel: int, raw_payload: bytes) -> None: + """Send a talk audio frame with AES-encrypted extension + raw binary payload. + + Unlike send(), this method does NOT await a response — talk frames + (cmd_id=202) are fire-and-forget. The extension XML is AES-encrypted + but the audio payload is sent as raw (unencrypted) binary data. + + Args: + channel: Camera channel number. + raw_payload: BcMedia frame(s) containing ADPCM audio data. + """ + if not self._logged_in: + await self.login() + + ch_id = channel + 1 + self._mess_id = (self._mess_id + 1) % 16777216 + + # Build extension XML with binaryData flag + ext = xmls.BINARY_EXTENSION_XML.format(channel=channel) + ext_bytes = ext.encode("utf8") + + # Encrypt extension only — audio payload stays raw + enc_ext = self._aes_encrypt(ext_bytes) + + # Header fields + ext_len = len(enc_ext) + mess_len = ext_len + len(raw_payload) + payload_offset = ext_len + + cmd_id_bytes = (202).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" + message_class = "1464" + + header = ( + bytes.fromhex(HEADER_MAGIC) + + cmd_id_bytes + + mess_len_bytes + + mess_id_bytes + + bytes.fromhex(status_code + message_class) + + payload_offset_bytes + ) + + await self._connect_if_needed() + if TYPE_CHECKING: + assert self._transport is not None + + async with self._mutex: + self._transport.write(header + enc_ext + raw_payload) + def _aes_encrypt(self, body: bytes) -> bytes: """Encrypt a message using AES encryption""" if not body: @@ -1620,6 +1673,15 @@ async def get_channel_data(self) -> None: for audio in root.findall(".//audioStreamMode"): if audio.text == "mixAudioStream": self.capabilities[channel].add("two_way_audio") + # Store audio config for talk() + audio_cfg = root.find(".//audioConfig") + if audio_cfg is not None: + self._talk_config[channel] = { + "sample_rate": int(audio_cfg.findtext("sampleRate", "8000")), + "block_size": int(audio_cfg.findtext("lengthPerEncoder", "1024")), + "duplex": root.findtext(".//duplex", "FDX"), + "stream_mode": root.findtext(".//audioStreamMode", "followVideoStream"), + } if cmd_id == 483: # hardwired chime self.capabilities[channel].add("hardwired_chime") if cmd_id == 527: # crossline detection @@ -2857,6 +2919,93 @@ async def SetAudioNoise(self, channel: int, level: int) -> None: await self.send(cmd_id=440, channel=channel, body=xml) await self.GetAudioNoise(channel) + async def talk( + self, + channel: int, + audio_data: bytes, + sample_rate: int | None = None, + block_size: int | None = None, + ) -> None: + """Send PCM audio to camera speaker via two-way audio (Baichuan talk). + + Starts a talk session, encodes PCM to ADPCM, sends audio frames, + and ends the session. Handles 422 (session busy) by resetting first. + + Args: + channel: Camera channel (0 for standalone cameras, 0+ for hub). + audio_data: Raw PCM audio — 16-bit signed little-endian, mono, + at the target sample rate (default 8000 Hz). + sample_rate: Override sample rate. Default: from TalkAbility. + block_size: Override ADPCM block size (lengthPerEncoder). + Default: from TalkAbility. + + Raises: + NotSupportedError: If camera doesn't support two-way audio. + ReolinkError: If talk session cannot be started. + """ + from .audio import build_bc_media_frame, encode_pcm_to_adpcm + + # Get talk config (from TalkAbility, queried during capability discovery) + cfg = self._talk_config.get(channel, {}) + sr = sample_rate or cfg.get("sample_rate", 8000) + bs = block_size or cfg.get("block_size", 1024) + duplex = cfg.get("duplex", "FDX") + stream_mode = cfg.get("stream_mode", "followVideoStream") + + # Build TalkConfig body + talk_config_body = xmls.TALK_CONFIG_XML.format( + channel=channel, + duplex=duplex, + stream_mode=stream_mode, + sample_rate=sr, + block_size=bs, + ) + + # Start talk session (cmd_id=201: TalkConfig) + try: + await self.send(cmd_id=201, channel=channel, body=talk_config_body) + except ApiError as err: + if err.rspCode == 422: + # Another talk session active — reset and retry + _LOGGER.debug("Baichuan host %s: talk session busy (422), resetting", self._host) + try: + await self.send(cmd_id=11, channel=channel) + except Exception: + pass + await asyncio.sleep(0.5) + await self.send(cmd_id=201, channel=channel, body=talk_config_body) + else: + raise + + try: + # Encode PCM to ADPCM blocks + adpcm_blocks = encode_pcm_to_adpcm(audio_data, samples_per_block=bs) + + # Calculate inter-frame delay based on audio duration + # Each block encodes `bs` samples at `sr` Hz + block_duration = bs / sr # seconds per block + + # Send audio frames (cmd_id=202: Talk) + # Pack up to 4 blocks per message for efficiency + blocks_per_message = 4 + for i in range(0, len(adpcm_blocks), blocks_per_message): + batch = adpcm_blocks[i : i + blocks_per_message] + payload = b"".join(build_bc_media_frame(block) for block in batch) + await self._send_talk_frame(channel, payload) + + # Pace sending to match audio playback rate + await asyncio.sleep(block_duration * len(batch)) + + # Wait for camera to finish playing the last frames + await asyncio.sleep(1.0) + + finally: + # End talk session (cmd_id=11: TalkReset) + try: + await self.send(cmd_id=11, channel=channel) + except Exception as err: + _LOGGER.debug("Baichuan host %s: TalkReset failed: %s", self._host, err) + @http_cmd("GetDingDongList") async def GetDingDongList(self, channel: int | None = None, retry: int = 3, **_kwargs) -> None: """Get the DingDongList info""" diff --git a/reolink_aio/baichuan/xmls.py b/reolink_aio/baichuan/xmls.py index f3e182be..b37b913c 100644 --- a/reolink_aio/baichuan/xmls.py +++ b/reolink_aio/baichuan/xmls.py @@ -33,6 +33,30 @@ """ +BINARY_EXTENSION_XML = """ + +{channel} +1 + +""" + +TALK_CONFIG_XML = """ + + +{channel} +{duplex} +{stream_mode} + +adpcm +{sample_rate} +16 +{block_size} +mono + + + +""" + DingDongOpt_1_XML = """ diff --git a/tests/test_talk.py b/tests/test_talk.py new file mode 100644 index 00000000..44f0f805 --- /dev/null +++ b/tests/test_talk.py @@ -0,0 +1,392 @@ +"""Tests for two-way audio (talk) implementation.""" + +from __future__ import annotations + +import struct +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock, patch + +from reolink_aio.baichuan.audio import ( + BC_MEDIA_ADPCM_MAGIC, + build_bc_media_frame, + encode_pcm_to_adpcm, +) +from reolink_aio.baichuan.baichuan import Baichuan + + +class _TalkRecordingTransport: + """Transport that records writes without setting response futures (fire-and-forget).""" + + def __init__(self) -> None: + self.writes: list[bytes] = [] + + def write(self, data: bytes) -> None: + self.writes.append(data) + + def is_closing(self) -> bool: + return False + + +class _SendRecordingTransport: + """Transport that records writes and resolves response futures (for send()).""" + + def __init__(self, protocol, response_body: str = "") -> None: + self._protocol = protocol + self._response_body = response_body + self.writes: list[bytes] = [] + + def write(self, data: bytes) -> None: + self.writes.append(data) + cmd_id = int.from_bytes(data[4:8], byteorder="little") + mess_id = int.from_bytes(data[12:16], byteorder="little") + if cmd_id in self._protocol.receive_futures and mess_id in self._protocol.receive_futures[cmd_id]: + self._protocol.receive_futures[cmd_id][mess_id].set_result((data[:24], 24, b"")) + + def is_closing(self) -> bool: + return False + + +# --- ADPCM Encoder Tests --- + + +class TestAdpcmEncoder(unittest.TestCase): + def test_empty_input(self) -> None: + result = encode_pcm_to_adpcm(b"") + self.assertEqual(result, []) + + def test_single_byte_ignored(self) -> None: + """A single byte is not a complete 16-bit sample.""" + result = encode_pcm_to_adpcm(b"\x00") + self.assertEqual(result, []) + + def test_silence_single_block(self) -> None: + """1024 zero samples should produce one block.""" + pcm = b"\x00\x00" * 1024 + blocks = encode_pcm_to_adpcm(pcm, samples_per_block=1024) + self.assertEqual(len(blocks), 1) + + def test_block_structure(self) -> None: + """Verify preamble format and data size.""" + pcm = b"\x00\x00" * 1024 + blocks = encode_pcm_to_adpcm(pcm, samples_per_block=1024) + block = blocks[0] + + # Block should be 4-byte preamble + 512 bytes of nibble data + self.assertEqual(len(block), 4 + 512) + + # Preamble: predictor (i16 LE) + step_index (u8) + reserved (u8) + predictor, step_index, reserved = struct.unpack(" None: + """All-zero PCM with zero initial state should produce all-zero nibbles.""" + pcm = b"\x00\x00" * 1024 + blocks = encode_pcm_to_adpcm(pcm, samples_per_block=1024) + block = blocks[0] + + # Preamble should be zeros (predictor=0, step_index=0) + self.assertEqual(block[:4], b"\x00\x00\x00\x00") + + # All nibbles should be zero (difference is always 0) + for byte in block[4:]: + self.assertEqual(byte, 0) + + def test_multiple_blocks(self) -> None: + """Input longer than samples_per_block should produce multiple blocks.""" + pcm = b"\x00\x00" * 2048 + blocks = encode_pcm_to_adpcm(pcm, samples_per_block=1024) + self.assertEqual(len(blocks), 2) + for block in blocks: + self.assertEqual(len(block), 4 + 512) + + def test_partial_last_block(self) -> None: + """Input not a multiple of samples_per_block produces a smaller last block.""" + # 1536 samples: 1 full block (1024) + 1 partial (512) + pcm = b"\x00\x00" * 1536 + blocks = encode_pcm_to_adpcm(pcm, samples_per_block=1024) + self.assertEqual(len(blocks), 2) + self.assertEqual(len(blocks[0]), 4 + 512) # Full block + self.assertEqual(len(blocks[1]), 4 + 256) # 512 samples = 256 nibble bytes + + def test_non_zero_pcm_produces_non_zero_nibbles(self) -> None: + """A loud tone should produce non-zero ADPCM nibbles.""" + # Simple sawtooth: ramps from 0 to 32000 + samples = [int(32000 * i / 1024) for i in range(1024)] + pcm = struct.pack(f"<{len(samples)}h", *samples) + blocks = encode_pcm_to_adpcm(pcm, samples_per_block=1024) + self.assertEqual(len(blocks), 1) + # At least some data bytes should be non-zero + data = blocks[0][4:] + self.assertTrue(any(b != 0 for b in data)) + + def test_predictor_stays_in_range(self) -> None: + """Predictor should never exceed 16-bit signed range.""" + # Extreme input: alternating min/max + samples = [32767, -32768] * 512 + pcm = struct.pack(f"<{len(samples)}h", *samples) + blocks = encode_pcm_to_adpcm(pcm, samples_per_block=1024) + # Just verify it completes without error — predictor clamping works + self.assertEqual(len(blocks), 1) + + def test_small_block_size(self) -> None: + """Verify encoder works with small block sizes.""" + pcm = b"\x00\x00" * 8 + blocks = encode_pcm_to_adpcm(pcm, samples_per_block=4) + self.assertEqual(len(blocks), 2) + # 4 samples = 4 nibbles = 2 bytes + 4 preamble = 6 + for block in blocks: + self.assertEqual(len(block), 4 + 2) + + +# --- BcMedia Frame Tests --- + + +class TestBcMediaFrame(unittest.TestCase): + def test_frame_magic(self) -> None: + """Frame should start with '01wb' magic.""" + block = b"\x00" * (4 + 512) # Fake ADPCM block + frame = build_bc_media_frame(block) + self.assertEqual(frame[:4], BC_MEDIA_ADPCM_MAGIC) + + def test_frame_payload_size(self) -> None: + """payload_size should be data_len + 4.""" + block = b"\x00" * (4 + 512) # 516 bytes + frame = build_bc_media_frame(block) + payload_size_1 = struct.unpack(" None: + """Sub-magic should be 0x0001.""" + block = b"\x00" * (4 + 512) + frame = build_bc_media_frame(block) + sub_magic = struct.unpack(" None: + """half_block_size should be (data_len - 4) / 2.""" + block = b"\x00" * (4 + 512) + frame = build_bc_media_frame(block) + half_block = struct.unpack(" None: + """Data portion (data + padding) should be 8-byte aligned.""" + # 516 data → padding = (8 - 516 % 8) % 8 = 4 → data+padding = 520 + block = b"\x00" * 516 + frame = build_bc_media_frame(block) + data_plus_padding = len(frame) - 12 # subtract 12-byte header + self.assertEqual(data_plus_padding % 8, 0) + + def test_frame_no_padding_needed(self) -> None: + """Data that's already 8-byte aligned needs no padding.""" + # 520 bytes → 520 % 8 = 0 + block = b"\x00" * 520 + frame = build_bc_media_frame(block) + # 12 header + 520 data = 532, 532 % 8 = 4 → no wait, padding is based on data_len not total + # padding = (8 - 520 % 8) % 8 = 0 + self.assertEqual(len(frame), 12 + 520) + + def test_frame_contains_data(self) -> None: + """ADPCM data should appear after the 12-byte header.""" + block = bytes(range(256)) * 2 + bytes(4) # 516 bytes + frame = build_bc_media_frame(block) + self.assertEqual(frame[12 : 12 + len(block)], block) + + def test_typical_block_frame_size(self) -> None: + """Typical 8kHz/1024-sample block: 516 data → 532 total.""" + block = b"\x00" * 516 # 4 preamble + 512 nibble bytes + frame = build_bc_media_frame(block) + # 12 header + 516 data + 4 padding = 532 + self.assertEqual(len(frame), 532) + + +# --- _send_talk_frame Tests --- + + +class TestSendTalkFrame(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.bc = Baichuan( + host="127.0.0.1", + username="user", + password="password", + http_api=SimpleNamespace(nvr_name="test", _updating=False), + ) + self.bc._logged_in = True + self.bc._aes_key = b"0123456789abcdef" # 16-byte key for AES-128 + self.transport = _TalkRecordingTransport() + self.bc._protocol = SimpleNamespace(receive_futures={}) + self.bc._transport = self.transport + self.bc._connect_if_needed = AsyncMock() + + async def test_header_cmd_id_is_202(self) -> None: + """Talk frame header should have cmd_id=202.""" + await self.bc._send_talk_frame(channel=0, raw_payload=b"\x00" * 100) + + written = self.transport.writes[0] + cmd_id = int.from_bytes(written[4:8], byteorder="little") + self.assertEqual(cmd_id, 202) + + async def test_header_is_24_bytes(self) -> None: + """Message class 1464 produces a 24-byte header.""" + await self.bc._send_talk_frame(channel=0, raw_payload=b"\x00" * 100) + + written = self.transport.writes[0] + # Check message_class field at bytes 18:20 + message_class = written[18:20].hex() + self.assertEqual(message_class, "1464") + + async def test_payload_offset_equals_encrypted_extension_length(self) -> None: + """payload_offset should equal the length of the AES-encrypted extension.""" + await self.bc._send_talk_frame(channel=0, raw_payload=b"\xAA" * 50) + + written = self.transport.writes[0] + payload_offset = int.from_bytes(written[20:24], byteorder="little") + mess_len = int.from_bytes(written[8:12], byteorder="little") + + # payload_offset = encrypted extension length + # mess_len = encrypted extension + raw payload + self.assertEqual(mess_len, payload_offset + 50) + self.assertGreater(payload_offset, 0) + + async def test_raw_payload_is_not_encrypted(self) -> None: + """The binary audio payload should appear as-is (not encrypted) in the write.""" + marker = b"\xDE\xAD\xBE\xEF" * 10 # 40 bytes of recognizable data + await self.bc._send_talk_frame(channel=0, raw_payload=marker) + + written = self.transport.writes[0] + # The marker should appear at the end of the written data + self.assertTrue(written.endswith(marker)) + + async def test_extension_is_encrypted(self) -> None: + """The extension XML should be AES-encrypted (different from plaintext).""" + await self.bc._send_talk_frame(channel=0, raw_payload=b"\x00" * 10) + + written = self.transport.writes[0] + payload_offset = int.from_bytes(written[20:24], byteorder="little") + + # Extension sits between header (24 bytes) and the raw payload + enc_ext = written[24 : 24 + payload_offset] + + # The plaintext extension contains "channelId" — encrypted version should not + self.assertNotIn(b"channelId", enc_ext) + + async def test_mess_id_increments(self) -> None: + """Each call should increment the message ID.""" + self.bc._mess_id = 100 + await self.bc._send_talk_frame(channel=0, raw_payload=b"\x00") + await self.bc._send_talk_frame(channel=0, raw_payload=b"\x00") + + id1 = int.from_bytes(self.transport.writes[0][12:16], byteorder="little") + id2 = int.from_bytes(self.transport.writes[1][12:16], byteorder="little") + self.assertNotEqual(id1, id2) + + +# --- talk() Orchestration Tests --- + + +class TestTalkOrchestration(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.bc = Baichuan( + host="127.0.0.1", + username="user", + password="password", + http_api=SimpleNamespace(nvr_name="test", _updating=False), + ) + self.bc._logged_in = True + self.bc._aes_key = b"0123456789abcdef" + self.transport = _TalkRecordingTransport() + self.protocol = SimpleNamespace(receive_futures={}) + self.bc._protocol = self.protocol + self.bc._transport = self.transport + self.bc._connect_if_needed = AsyncMock() + + # Mock send() to avoid full protocol handling + self.send_calls: list[dict] = [] + + async def mock_send(cmd_id, channel=None, body="", **kwargs): + self.send_calls.append({"cmd_id": cmd_id, "channel": channel, "body": body}) + return "" + + self.bc.send = mock_send + + async def test_talk_sends_talk_config_first(self) -> None: + """talk() should send TalkConfig (cmd_id=201) before audio frames.""" + pcm = b"\x00\x00" * 1024 # 1 block of silence + await self.bc.talk(channel=0, audio_data=pcm) + + # First send should be TalkConfig + self.assertEqual(self.send_calls[0]["cmd_id"], 201) + self.assertIn("TalkConfig", self.send_calls[0]["body"]) + + async def test_talk_sends_talk_reset_at_end(self) -> None: + """talk() should send TalkReset (cmd_id=11) when done.""" + pcm = b"\x00\x00" * 1024 + await self.bc.talk(channel=0, audio_data=pcm) + + # Last send should be TalkReset + self.assertEqual(self.send_calls[-1]["cmd_id"], 11) + + async def test_talk_sends_audio_frames(self) -> None: + """talk() should write audio frames via _send_talk_frame.""" + pcm = b"\x00\x00" * 1024 + await self.bc.talk(channel=0, audio_data=pcm) + + # Should have at least one transport write (audio frame) + self.assertGreater(len(self.transport.writes), 0) + + async def test_talk_uses_config_from_talk_ability(self) -> None: + """talk() should use cached TalkAbility config.""" + self.bc._talk_config[0] = { + "sample_rate": 16000, + "block_size": 2048, + "duplex": "HDX", + "stream_mode": "mixAudioStream", + } + pcm = b"\x00\x00" * 2048 + await self.bc.talk(channel=0, audio_data=pcm) + + # TalkConfig body should contain the cached values + body = self.send_calls[0]["body"] + self.assertIn("16000", body) + self.assertIn("2048", body) + self.assertIn("HDX", body) + + async def test_talk_parameter_overrides(self) -> None: + """Explicit sample_rate/block_size should override TalkAbility.""" + self.bc._talk_config[0] = { + "sample_rate": 8000, + "block_size": 1024, + "duplex": "FDX", + "stream_mode": "followVideoStream", + } + pcm = b"\x00\x00" * 2048 + await self.bc.talk(channel=0, audio_data=pcm, sample_rate=16000, block_size=2048) + + body = self.send_calls[0]["body"] + self.assertIn("16000", body) + self.assertIn("2048", body) + + async def test_talk_reset_on_error(self) -> None: + """TalkReset should be sent even if audio sending fails.""" + # Make _send_talk_frame raise an error + async def failing_send_frame(channel, payload): + raise ConnectionError("fake error") + + self.bc._send_talk_frame = failing_send_frame + + pcm = b"\x00\x00" * 1024 + with self.assertRaises(ConnectionError): + await self.bc.talk(channel=0, audio_data=pcm) + + # TalkReset (cmd_id=11) should still be sent + reset_calls = [c for c in self.send_calls if c["cmd_id"] == 11] + self.assertEqual(len(reset_calls), 1) + + +if __name__ == "__main__": + unittest.main() From f887006bb76fc26d67625b9a478d2a57f9053c2e Mon Sep 17 00:00:00 2001 From: lorek123 Date: Fri, 10 Apr 2026 19:11:17 +0200 Subject: [PATCH 2/2] Fix standalone camera support and add split talk API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix two_way_audio capability detection: cameras using "followVideoStream" (e.g. Reolink E1) were not being marked as capable; now any camera that returns an audioConfig in TalkAbility is correctly detected - Fix cmd_id=10 not called during init on standalone cameras: add `or self.http_api.baichuan_only` guard so TalkAbility is always queried during channel discovery on baichuan-only devices - Add split API: get_talk_ability() / start_talk() / stop_talk() / send_talk_data() on both Baichuan and Host, plus build_bcmedia_adpcm() static helper for BcMedia ADPCM framing - Add TalkConfigSet XML template with all audio fields parameterized - Rename _send_talk_frame() → send_binary_no_reply() with generalized signature (cmd_id, channel, binary_body); talk() updated to use it - Update tests: rename TestSendTalkFrame → TestSendBinaryNoReply, add TestBuildBcmediaAdpcm, TestSplitTalkApi, TestTwoWayAudioCapability (52 tests total, all passing) Co-Authored-By: Claude Sonnet 4.6 --- reolink_aio/api.py | 47 +++++ reolink_aio/baichuan/baichuan.py | 180 ++++++++++++++----- reolink_aio/baichuan/xmls.py | 16 ++ tests/test_talk.py | 292 ++++++++++++++++++++++++++++--- 4 files changed, 472 insertions(+), 63 deletions(-) diff --git a/reolink_aio/api.py b/reolink_aio/api.py index 79a71f81..bb63f0c4 100644 --- a/reolink_aio/api.py +++ b/reolink_aio/api.py @@ -5356,6 +5356,53 @@ async def set_audio_alarm(self, channel: int, enable: bool) -> None: await self.send_setting(body) + 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 ADPCM block, e.g. 320 or 1024 + - sound_track (str): channel layout, e.g. "mono" + - duplex (str): duplex mode, e.g. "FDX" + - audio_stream_mode (str): e.g. "followVideoStream" + + Encode audio as IMA ADPCM blocks with the returned parameters, wrap + each block with ``Baichuan.build_bcmedia_adpcm([block])``, and stream + 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 4801290c..89624e54 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 Callable from datetime import datetime, timedelta from inspect import getmembers @@ -438,57 +439,46 @@ async def send_payload( return (rec_body, payload) - async def _send_talk_frame(self, channel: int, raw_payload: bytes) -> None: - """Send a talk audio frame with AES-encrypted extension + raw binary payload. - - Unlike send(), this method does NOT await a response — talk frames - (cmd_id=202) are fire-and-forget. The extension XML is AES-encrypted - but the audio payload is sent as raw (unencrypted) binary data. + 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. - Args: - channel: Camera channel number. - raw_payload: BcMedia frame(s) containing ADPCM audio data. + The extension XML is AES-encrypted; the binary body is sent raw (not encrypted). + Used for fire-and-forget commands such as audio talk frames (cmd_id=202). """ - if not self._logged_in: + if not self._logged_in and cmd_id > 2: await self.login() - ch_id = channel + 1 - self._mess_id = (self._mess_id + 1) % 16777216 - - # Build extension XML with binaryData flag - ext = xmls.BINARY_EXTENSION_XML.format(channel=channel) - ext_bytes = ext.encode("utf8") + if channel is None: + ch_id = 250 + else: + ch_id = channel + 1 - # Encrypt extension only — audio payload stays raw - enc_ext = self._aes_encrypt(ext_bytes) + ext = xmls.BINARY_EXTENSION_XML.format(channel=channel) if channel is not None else "" + enc_ext = self._aes_encrypt(ext.encode("utf-8")) + mess_len = len(enc_ext) + len(binary_body) + payload_offset = len(enc_ext) - # Header fields - ext_len = len(enc_ext) - mess_len = ext_len + len(raw_payload) - payload_offset = ext_len + self._mess_id = (self._mess_id + 1) % 16777216 - cmd_id_bytes = (202).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") + 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" - message_class = "1464" - - header = ( - bytes.fromhex(HEADER_MAGIC) - + cmd_id_bytes - + mess_len_bytes - + mess_id_bytes - + bytes.fromhex(status_code + message_class) - + payload_offset_bytes - ) + 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 + raw_payload) + self._transport.write(header + enc_ext + binary_body) def _aes_encrypt(self, body: bytes) -> bytes: """Encrypt a message using AES encryption""" @@ -1521,7 +1511,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: + 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))) if (self.http_api.is_nvr or self.privacy_mode() is not None) and self.api_version("remoteAbility", channel) > 0: @@ -1676,6 +1666,9 @@ async def get_channel_data(self) -> None: # Store audio config for talk() audio_cfg = root.find(".//audioConfig") if audio_cfg is not None: + # Any camera that returns an audioConfig supports two-way audio + # (some cameras, e.g. Reolink E1, use "followVideoStream" not "mixAudioStream") + self.capabilities[channel].add("two_way_audio") self._talk_config[channel] = { "sample_rate": int(audio_cfg.findtext("sampleRate", "8000")), "block_size": int(audio_cfg.findtext("lengthPerEncoder", "1024")), @@ -2991,7 +2984,7 @@ async def talk( for i in range(0, len(adpcm_blocks), blocks_per_message): batch = adpcm_blocks[i : i + blocks_per_message] payload = b"".join(build_bc_media_frame(block) for block in batch) - await self._send_talk_frame(channel, payload) + await self.send_binary_no_reply(cmd_id=202, channel=channel, binary_body=payload) # Pace sending to match audio playback rate await asyncio.sleep(block_duration * len(batch)) @@ -3006,6 +2999,115 @@ async def talk( except Exception as err: _LOGGER.debug("Baichuan host %s: TalkReset failed: %s", self._host, err) + async def get_talk_ability(self, channel: int) -> dict: + """Query the camera's talk (2-way audio) capability via cmd_id=10. + + Returns a dict with keys: duplex, audio_stream_mode, audio_type, + sample_rate, sample_precision, length_per_encoder, sound_track. + """ + mess = await self.send(cmd_id=10, channel=channel) + root = XML.fromstring(mess) + + ability: dict = {} + + for elem in root.findall(".//duplex"): + if elem.text: + ability["duplex"] = elem.text + break + + for elem in root.findall(".//audioStreamMode"): + if elem.text: + ability["audio_stream_mode"] = elem.text + break + + 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 session (cmd_id=201 TalkConfig). + + Queries TalkAbility fresh via cmd_id=10 and sends TalkConfig with the + camera's own reported parameters. Returns the ability dict so the caller + knows the sample_rate and length_per_encoder to use when encoding audio. + + Use send_talk_data() to stream audio and stop_talk() to end the session. + """ + ability = await self.get_talk_ability(channel) + + xml = xmls.TalkConfigSet.format( + channel=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"), + ) + 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, ability.get("audio_type"), ability.get("sample_rate"), ability.get("length_per_encoder"), + ) + return ability + + async def stop_talk(self, channel: int) -> None: + """Stop the 2-way audio session (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 IMA ADPCM blocks in BcMedia framing for cmd_id=202. + + Each block must be a complete DVI-4/IMA ADPCM block: + - 4-byte header (s16LE predictor, u8 step_index, u8 pad) + - (length_per_encoder // 2) nibble-packed sample bytes + + Up to 4 blocks may be combined into one BcMedia message. + Pass the result to send_talk_data(). + """ + # BcMedia ADPCM frame (confirmed from pcap + Ghidra audioTalkSendStream): + # 4 bytes magic 0x62773130 (little-endian "bw10") + # 2+2 bytes payload_size LE (duplicated) = len(block) + 4 + # 2 bytes sub-magic 0x0100 + # 2 bytes half_block = 2 (always 2) + # N bytes raw IMA ADPCM block + # P bytes zero-padding to 8-byte boundary + BCMEDIA_ADPCM_MAGIC = struct.pack(" None: + """Send BcMedia-framed ADPCM audio to the camera (cmd_id=202, no reply). + + bcmedia_data must be the output of build_bcmedia_adpcm(). + The audio payload is sent without encryption as required by the protocol. + """ + await self.send_binary_no_reply(cmd_id=202, channel=channel, binary_body=bcmedia_data) + @http_cmd("GetDingDongList") async def GetDingDongList(self, channel: int | None = None, retry: int = 3, **_kwargs) -> None: """Get the DingDongList info""" diff --git a/reolink_aio/baichuan/xmls.py b/reolink_aio/baichuan/xmls.py index b37b913c..d8efdf93 100644 --- a/reolink_aio/baichuan/xmls.py +++ b/reolink_aio/baichuan/xmls.py @@ -57,6 +57,22 @@ """ +TalkConfigSet = """ + + +{channel} +{duplex} +{audio_stream_mode} + +{audio_type} +{sample_rate} +{sample_precision} +{length_per_encoder} +{sound_track} + + +""" + DingDongOpt_1_XML = """ diff --git a/tests/test_talk.py b/tests/test_talk.py index 44f0f805..8461174d 100644 --- a/tests/test_talk.py +++ b/tests/test_talk.py @@ -205,10 +205,10 @@ def test_typical_block_frame_size(self) -> None: self.assertEqual(len(frame), 532) -# --- _send_talk_frame Tests --- +# --- send_binary_no_reply Tests --- -class TestSendTalkFrame(unittest.IsolatedAsyncioTestCase): +class TestSendBinaryNoReply(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.bc = Baichuan( host="127.0.0.1", @@ -223,63 +223,56 @@ async def asyncSetUp(self) -> None: self.bc._transport = self.transport self.bc._connect_if_needed = AsyncMock() - async def test_header_cmd_id_is_202(self) -> None: - """Talk frame header should have cmd_id=202.""" - await self.bc._send_talk_frame(channel=0, raw_payload=b"\x00" * 100) + async def test_header_cmd_id(self) -> None: + """cmd_id in the header should match the argument.""" + await self.bc.send_binary_no_reply(cmd_id=202, channel=0, binary_body=b"\x00" * 100) written = self.transport.writes[0] cmd_id = int.from_bytes(written[4:8], byteorder="little") self.assertEqual(cmd_id, 202) - async def test_header_is_24_bytes(self) -> None: + async def test_header_message_class_1464(self) -> None: """Message class 1464 produces a 24-byte header.""" - await self.bc._send_talk_frame(channel=0, raw_payload=b"\x00" * 100) + await self.bc.send_binary_no_reply(cmd_id=202, channel=0, binary_body=b"\x00" * 100) written = self.transport.writes[0] - # Check message_class field at bytes 18:20 message_class = written[18:20].hex() self.assertEqual(message_class, "1464") async def test_payload_offset_equals_encrypted_extension_length(self) -> None: """payload_offset should equal the length of the AES-encrypted extension.""" - await self.bc._send_talk_frame(channel=0, raw_payload=b"\xAA" * 50) + await self.bc.send_binary_no_reply(cmd_id=202, channel=0, binary_body=b"\xAA" * 50) written = self.transport.writes[0] payload_offset = int.from_bytes(written[20:24], byteorder="little") mess_len = int.from_bytes(written[8:12], byteorder="little") - # payload_offset = encrypted extension length - # mess_len = encrypted extension + raw payload self.assertEqual(mess_len, payload_offset + 50) self.assertGreater(payload_offset, 0) async def test_raw_payload_is_not_encrypted(self) -> None: - """The binary audio payload should appear as-is (not encrypted) in the write.""" + """The binary body should appear as-is (not encrypted) in the write.""" marker = b"\xDE\xAD\xBE\xEF" * 10 # 40 bytes of recognizable data - await self.bc._send_talk_frame(channel=0, raw_payload=marker) + await self.bc.send_binary_no_reply(cmd_id=202, channel=0, binary_body=marker) written = self.transport.writes[0] - # The marker should appear at the end of the written data self.assertTrue(written.endswith(marker)) async def test_extension_is_encrypted(self) -> None: """The extension XML should be AES-encrypted (different from plaintext).""" - await self.bc._send_talk_frame(channel=0, raw_payload=b"\x00" * 10) + await self.bc.send_binary_no_reply(cmd_id=202, channel=0, binary_body=b"\x00" * 10) written = self.transport.writes[0] payload_offset = int.from_bytes(written[20:24], byteorder="little") - - # Extension sits between header (24 bytes) and the raw payload enc_ext = written[24 : 24 + payload_offset] - # The plaintext extension contains "channelId" — encrypted version should not self.assertNotIn(b"channelId", enc_ext) async def test_mess_id_increments(self) -> None: """Each call should increment the message ID.""" self.bc._mess_id = 100 - await self.bc._send_talk_frame(channel=0, raw_payload=b"\x00") - await self.bc._send_talk_frame(channel=0, raw_payload=b"\x00") + await self.bc.send_binary_no_reply(cmd_id=202, channel=0, binary_body=b"\x00") + await self.bc.send_binary_no_reply(cmd_id=202, channel=0, binary_body=b"\x00") id1 = int.from_bytes(self.transport.writes[0][12:16], byteorder="little") id2 = int.from_bytes(self.transport.writes[1][12:16], byteorder="little") @@ -332,7 +325,7 @@ async def test_talk_sends_talk_reset_at_end(self) -> None: self.assertEqual(self.send_calls[-1]["cmd_id"], 11) async def test_talk_sends_audio_frames(self) -> None: - """talk() should write audio frames via _send_talk_frame.""" + """talk() should write audio frames via send_binary_no_reply.""" pcm = b"\x00\x00" * 1024 await self.bc.talk(channel=0, audio_data=pcm) @@ -373,11 +366,10 @@ async def test_talk_parameter_overrides(self) -> None: async def test_talk_reset_on_error(self) -> None: """TalkReset should be sent even if audio sending fails.""" - # Make _send_talk_frame raise an error - async def failing_send_frame(channel, payload): + async def failing_send_binary(cmd_id, channel=None, binary_body=b""): raise ConnectionError("fake error") - self.bc._send_talk_frame = failing_send_frame + self.bc.send_binary_no_reply = failing_send_binary pcm = b"\x00\x00" * 1024 with self.assertRaises(ConnectionError): @@ -388,5 +380,257 @@ async def failing_send_frame(channel, payload): self.assertEqual(len(reset_calls), 1) +# --- build_bcmedia_adpcm Tests --- + + +class TestBuildBcmediaAdpcm(unittest.TestCase): + """Tests for Baichuan.build_bcmedia_adpcm() (our split-API framing helper).""" + + def _make_block(self, lpe: int = 1024) -> bytes: + """Return a minimal valid ADPCM block for the given lengthPerEncoder.""" + return b"\x00" * (4 + lpe // 2) + + def test_magic_bytes(self) -> None: + """Frame should start with the ADPCM magic (0x62773130 LE = bytes 30 31 77 62).""" + block = self._make_block() + payload = Baichuan.build_bcmedia_adpcm([block]) + self.assertEqual(payload[:4], struct.pack(" None: + """payload_size should be len(block) + 4, duplicated in bytes 4-8.""" + block = self._make_block(1024) # 4 + 512 = 516 bytes + payload = Baichuan.build_bcmedia_adpcm([block]) + ps1 = struct.unpack(" None: + """Sub-magic field should be 0x0100 (confirmed from pcap).""" + block = self._make_block() + payload = Baichuan.build_bcmedia_adpcm([block]) + sub_magic = struct.unpack(" None: + """half_block field is always 2 (confirmed from Ghidra).""" + block = self._make_block() + payload = Baichuan.build_bcmedia_adpcm([block]) + half_block = struct.unpack(" None: + """ADPCM block data should appear immediately after the 12-byte frame header.""" + block = bytes(range(100)) + b"\x00" * (4 + 512 - 100) + payload = Baichuan.build_bcmedia_adpcm([block]) + self.assertEqual(payload[12 : 12 + len(block)], block) + + def test_8byte_alignment(self) -> None: + """Total frame length should always be a multiple of 8.""" + for lpe in [160, 320, 512, 1024]: + block = self._make_block(lpe) + payload = Baichuan.build_bcmedia_adpcm([block]) + self.assertEqual(len(payload) % 8, 0, f"lpe={lpe}: frame length {len(payload)} not 8-byte aligned") + + def test_no_padding_when_already_aligned(self) -> None: + """No padding bytes when payload_size is already a multiple of 8.""" + # For lpe=1024: block = 4 + 512 = 516, payload_size = 520, 520 % 8 = 0 → no padding + block = self._make_block(1024) + payload = Baichuan.build_bcmedia_adpcm([block]) + expected_len = 12 + len(block) # 12 header + 516 data + 0 padding + self.assertEqual(len(payload), expected_len) + + def test_multiple_blocks_concatenated(self) -> None: + """Multiple blocks should produce concatenated BcMedia frames.""" + block = self._make_block() + payload = Baichuan.build_bcmedia_adpcm([block, block, block]) + # Each frame: 12 header + 516 data = 528 bytes (520 payload_size, no padding needed) + single_frame_len = 12 + len(block) + self.assertEqual(len(payload), 3 * single_frame_len) + + def test_empty_block_list(self) -> None: + """Empty block list returns empty bytes.""" + self.assertEqual(Baichuan.build_bcmedia_adpcm([]), b"") + + +# --- get_talk_ability / start_talk / stop_talk / send_talk_data Tests --- + + +class TestSplitTalkApi(unittest.IsolatedAsyncioTestCase): + """Tests for the get_talk_ability / start_talk / stop_talk / send_talk_data split API.""" + + _TALK_ABILITY_XML = """ + + +FDX +followVideoStream + + +0 +adpcm +16000 +16 +1024 +mono + + + +""" + + async def asyncSetUp(self) -> None: + self.bc = Baichuan( + host="127.0.0.1", + username="user", + password="password", + http_api=SimpleNamespace(nvr_name="test", _updating=False), + ) + self.bc._logged_in = True + self.bc._aes_key = b"0123456789abcdef" + self.transport = _TalkRecordingTransport() + self.bc._protocol = SimpleNamespace(receive_futures={}) + self.bc._transport = self.transport + self.bc._connect_if_needed = AsyncMock() + + self.send_calls: list[dict] = [] + + async def mock_send(cmd_id, channel=None, body="", **kwargs): + self.send_calls.append({"cmd_id": cmd_id, "channel": channel, "body": body}) + if cmd_id == 10: + return self._TALK_ABILITY_XML + return "" + + self.bc.send = mock_send + + async def test_get_talk_ability_parses_sample_rate(self) -> None: + ability = await self.bc.get_talk_ability(channel=0) + self.assertEqual(ability["sample_rate"], 16000) + + async def test_get_talk_ability_parses_length_per_encoder(self) -> None: + ability = await self.bc.get_talk_ability(channel=0) + self.assertEqual(ability["length_per_encoder"], 1024) + + async def test_get_talk_ability_parses_duplex(self) -> None: + ability = await self.bc.get_talk_ability(channel=0) + self.assertEqual(ability["duplex"], "FDX") + + async def test_get_talk_ability_parses_audio_stream_mode(self) -> None: + ability = await self.bc.get_talk_ability(channel=0) + self.assertEqual(ability["audio_stream_mode"], "followVideoStream") + + async def test_start_talk_sends_cmd_id_10_then_201(self) -> None: + """start_talk() must call TalkAbility (10) then TalkConfig (201).""" + await self.bc.start_talk(channel=0) + self.assertEqual(self.send_calls[0]["cmd_id"], 10) + self.assertEqual(self.send_calls[1]["cmd_id"], 201) + + async def test_start_talk_uses_camera_sample_rate(self) -> None: + """TalkConfig body should use the sample_rate from TalkAbility (not a default).""" + await self.bc.start_talk(channel=0) + body = self.send_calls[1]["body"] + self.assertIn("16000", body) + + async def test_start_talk_returns_ability_dict(self) -> None: + ability = await self.bc.start_talk(channel=0) + self.assertEqual(ability["sample_rate"], 16000) + self.assertEqual(ability["length_per_encoder"], 1024) + + async def test_stop_talk_sends_cmd_id_11(self) -> None: + await self.bc.stop_talk(channel=0) + self.assertEqual(self.send_calls[0]["cmd_id"], 11) + + async def test_send_talk_data_calls_send_binary_no_reply(self) -> None: + """send_talk_data() should fire the audio payload via send_binary_no_reply.""" + marker = b"\xAB\xCD" * 20 + await self.bc.send_talk_data(channel=0, bcmedia_data=marker) + + self.assertEqual(len(self.transport.writes), 1) + self.assertTrue(self.transport.writes[0].endswith(marker)) + + async def test_send_talk_data_uses_cmd_202(self) -> None: + await self.bc.send_talk_data(channel=0, bcmedia_data=b"\x00" * 8) + written = self.transport.writes[0] + cmd_id = int.from_bytes(written[4:8], byteorder="little") + self.assertEqual(cmd_id, 202) + + +# --- Capability Detection Tests --- + + +class TestTwoWayAudioCapability(unittest.IsolatedAsyncioTestCase): + """Tests for two_way_audio capability detection from TalkAbility (cmd_id=10).""" + + async def _make_baichuan(self) -> Baichuan: + bc = Baichuan( + host="127.0.0.1", + username="user", + password="password", + http_api=SimpleNamespace(nvr_name="test", _updating=False), + ) + bc.capabilities[0] = set() + return bc + + def _process_cmd10(self, bc: Baichuan, xml: str) -> None: + """Simulate the get_channel_data() processing of a cmd_id=10 result.""" + from xml.etree import ElementTree as XML + root = XML.fromstring(xml) + for audio in root.findall(".//audioStreamMode"): + if audio.text == "mixAudioStream": + bc.capabilities[0].add("two_way_audio") + audio_cfg = root.find(".//audioConfig") + if audio_cfg is not None: + bc.capabilities[0].add("two_way_audio") + bc._talk_config[0] = { + "sample_rate": int(audio_cfg.findtext("sampleRate", "8000")), + "block_size": int(audio_cfg.findtext("lengthPerEncoder", "1024")), + "duplex": root.findtext(".//duplex", "FDX"), + "stream_mode": root.findtext(".//audioStreamMode", "followVideoStream"), + } + + async def test_follow_video_stream_sets_capability(self) -> None: + """Cameras using followVideoStream (e.g. E1) must also be detected.""" + bc = await self._make_baichuan() + xml = """ + FDX + followVideoStream + + adpcm16000 + 161024 + mono + + """ + self._process_cmd10(bc, xml) + self.assertIn("two_way_audio", bc.capabilities[0]) + + async def test_mix_audio_stream_sets_capability(self) -> None: + """Cameras using mixAudioStream should still be detected.""" + bc = await self._make_baichuan() + xml = """ + FDX + mixAudioStream + + adpcm8000 + 16320 + mono + + """ + self._process_cmd10(bc, xml) + self.assertIn("two_way_audio", bc.capabilities[0]) + + async def test_correct_sample_rate_stored(self) -> None: + """_talk_config should store the camera's actual sample_rate.""" + bc = await self._make_baichuan() + xml = """ + FDX + followVideoStream + + adpcm16000 + 161024 + mono + + """ + self._process_cmd10(bc, xml) + self.assertEqual(bc._talk_config[0]["sample_rate"], 16000) + + if __name__ == "__main__": unittest.main()