{x} ' for x in SESSIONS.keys()
+ ) or 'no cameras discovered yet PPPP Cameras '
+ 'PPPP Cameras '
+ f''
+ 'Page refreshes every 5 s as discovery finds cameras.
'
+ ''
+ ),
+ headers={'content-type': 'text/html'},
+ )
+
+
+def _camera_page_html(dev_id):
js = '''
'''
- videos = '{x} Light ON '
- f'Light OFF '
- f'IR ON '
- f'IR OFF '
- 'LEFT '
- f'RIGHT '
- f'UP '
- f'DOWN '
- f'Rotate STOP '
- 'Start Video '
- f'Stop Video '
+ x = dev_id
+ body = (
+ f'← all cameras
'
+ f'{x} '
+ '
'
+
+ f'Start Video '
+ f'Stop Video '
+ f'Snapshot PTZ '
+ f'LEFT '
+ f'RIGHT '
+ f'UP '
+ f'DOWN '
+ f'Rotate STOP '
+ ' Preset: Goto '
+ 'Save '
+ 'Presets are unimplemented on PTZA and FTYC — every prefab '
+ 'op was tried by hand on 2026-08-25 and none of them stores, recalls or '
+ 'moves anything. The buttons stay for other firmwares. '
+
+ 'Lights '
+ f'Light ON '
+ f'Light OFF '
+ f'IR ON '
+ f'IR OFF '
+
+ 'Video parameters '
+ 'not read yet
'
+ 'Re-read params '
- 'QVGA '
- 'VGA '
- 'HD '
- 'FD '
- 'UD '
+ ''
+ 'QVGA VGA HD FD UD '
' '
' Rotate: '
- f''
- 'NORMAL '
- 'H '
- 'V '
- 'HV '
+ ''
+ 'NORMAL H V HV '
' '
- # 'OSD ON '
- # f'OSD OFF '
- # f'Move Detect ON '
- # f'Move Detect OFF '
- # f'IR ON '
- # f'IR OFF '
- 'Reboot '
- for x in SESSIONS.keys())
+ 'Audio '
+ 'Play (low-latency) '
+ 'Stop '
+ f'Talk test tone (1s) '
+ f'Buffered fallback (several seconds behind): '
+ f'Stop Audio '
+
+ 'System '
+ 'Load device info '
+ 'Scan Wi-Fi (~10s) '
+ 'Show users '
+ 'Alias: Set '
+ f'Sync date/time '
+ f'Reboot '
+ '{} {}{}'.format(dev_id, js, body)
+ )
+
+
+async def camera_page(request):
+ session, err = _get_session(request)
+ if err:
+ return err
return web.Response(
- text="PPPP Cameras {}PPPP Cameras {}".format(
- js,
- videos,
- ),
+ text=_camera_page_html(request.match_info['dev_id']),
headers={'content-type': 'text/html'},
)
async def handle_commands(request):
- dev_id_str = request.match_info['dev_id']
+ session, err = _get_session(request)
+ if err:
+ return err
cmd = request.match_info['cmd']
params = await request.json()
- if dev_id_str not in SESSIONS:
- return web.Response(
- text='{"status": "error", "message": "unknown device"}',
- headers={'content-type': 'application/json'},
- status=404,
- )
- session = SESSIONS[dev_id_str]
+
+ async def talk_test(**kwargs):
+ # 1 s test tone in 120 ms chunks (960 samples = 1920 PCM bytes) --
+ # the same chunking the camera uses for its own audio -- paced in
+ # real time so the camera's jitter buffer isn't flooded.
+ await session.start_talk()
+ try:
+ for i in range(0, len(_TONE_PCM), 1920):
+ await session.send_audio(_TONE_PCM[i:i + 1920])
+ await asyncio.sleep(0.12)
+ finally:
+ await session.stop_talk()
+
+ async def sync_datetime(**kwargs):
+ await session.set_datetime()
+
web2cmd = {
- 'toggle-lamp': session.toggle_whitelight,
- 'toggle-ir': session.toggle_ir,
- 'rotate': session.step_rotate,
- 'rotate-stop': session.rotate_stop,
- 'reboot': session.reboot,
- 'start-video': session.start_video,
- 'stop-video': session.stop_video,
- 'set-video-param': session.set_video_param,
- # 'reset': session.reset,
- }.get(cmd)
-
- if web2cmd is None:
- return web.Response(
- text='{"status": "error", "message": "unknown command"}',
- headers={'content-type': 'application/json'},
- status=404,
- )
-
- await web2cmd(**params)
- return web.Response(text='{"status": "ok"}', headers={'content-type': 'application/json'})
+ 'toggle-lamp': getattr(session, 'toggle_whitelight', None),
+ 'toggle-ir': getattr(session, 'toggle_ir', None),
+ 'rotate': getattr(session, 'step_rotate', None),
+ 'rotate-stop': getattr(session, 'rotate_stop', None),
+ 'reboot': getattr(session, 'reboot', None),
+ 'start-video': getattr(session, 'start_video', None),
+ 'stop-video': getattr(session, 'stop_video', None),
+ 'set-video-param': getattr(session, 'set_video_param', None),
+ 'ptz-preset-goto': getattr(session, 'ptz_goto_preset', None),
+ 'ptz-preset-set': getattr(session, 'ptz_set_preset', None),
+ 'set-alias': getattr(session, 'set_alias', None),
+ 'sync-datetime': sync_datetime if hasattr(session, 'set_datetime') else None,
+ 'start-audio': getattr(session, 'start_audio', None),
+ 'stop-audio': getattr(session, 'stop_audio', None),
+ 'talk-test': talk_test if hasattr(session, 'start_talk') else None,
+ }
+
+ if cmd not in web2cmd:
+ return _json_error('unknown command', 404)
+ handler = web2cmd[cmd]
+ if handler is None:
+ return _json_error('command not supported by this device', 501)
+
+ try:
+ await handler(**params)
+ except Exception as e:
+ # Surface the failure to the browser -- a silent 500 here makes a
+ # server-side error indistinguishable from "camera ignored it".
+ logger.exception('Command %s failed for %s', cmd, request.match_info['dev_id'])
+ return _json_error(f'{type(e).__name__}: {e}', 500)
+ return web.json_response({'status': 'ok'})
+
+
+async def get_params(request):
+ """Read back current video parameters (ENH-003). Values are best-effort
+ decoded; the raw ACK payload is always included."""
+ session, err = _get_session(request)
+ if err:
+ return err
+ if not hasattr(session, 'get_video_param'):
+ return _json_error('not supported by this device', 501)
+
+ result = {}
+ # Sequential on purpose: wait_cmd_result is keyed by command, concurrent
+ # VIDEOPARAM_GETs would race each other.
+ for name, enum_cls, prefix in READBACK_PARAMS:
+ try:
+ payload = await session.get_video_param(name, timeout=3)
+ except Exception as e:
+ result[name] = {'error': f'{type(e).__name__}: {e}'}
+ continue
+ value = session.decode_video_param(payload, name)
+ symbol = None
+ if value is not None and enum_cls is not None:
+ try:
+ symbol = enum_cls(value).name.replace(prefix, '')
+ except ValueError:
+ pass
+ result[name] = {'value': value, 'symbol': symbol, 'raw': payload.hex(' ')}
+ return web.json_response({'status': 'ok', 'params': result})
+
+
+async def get_info(request):
+ """System/network readout (ENH-004). Parsed status plus raw hex blocks for
+ the calls whose struct layout is firmware-specific."""
+ session, err = _get_session(request)
+ if err:
+ return err
+ if not hasattr(session, 'get_status'):
+ return _json_error('not supported by this device', 501)
+
+ def decode_device_info(data):
+ # The 528-byte INF block is mostly zeros on the tested hardware; only
+ # the leading version field is understood so far.
+ out = {}
+ if len(data) >= 4:
+ out['swVersion'] = '.'.join(str(b) for b in reversed(data[:4]))
+ # Same packing as the status block's swVer: byte 1 device type,
+ # byte 3 chip type.
+ sw = struct.unpack('> 8) & 0xFF
+ chip_type = (sw >> 24) & 0xFF
+ out['devType'] = dev_type
+ out['devTypeName'] = enum_name(DevType, dev_type, 'DEV_')
+ out['chipType'] = chip_type
+ out['chipTypeName'] = enum_name(ChipType, chip_type, 'CHP_')
+ out['raw'] = data.hex(' ')
+ return out
+
+ def decode_datetime(data):
+ return parse_datetime_block(data) or {'raw': data.hex(' ')}
+
+ def decode_wifi(data):
+ return parse_wifi_settings(data) or {'raw': data.hex(' ')}
+
+ info = {}
+ for key, call, decode in [
+ ('status', session.get_status, None),
+ # Short timeouts: cameras that don't implement a block shouldn't stall
+ # the whole endpoint for the default 5 s each.
+ ('device_info', functools.partial(session.get_device_info, timeout=3)
+ if hasattr(session, 'get_device_info') else None, decode_device_info),
+ ('datetime', functools.partial(session.get_datetime, timeout=3)
+ if hasattr(session, 'get_datetime') else None, decode_datetime),
+ ('wifi', functools.partial(session.get_wifi_settings, timeout=3)
+ if hasattr(session, 'get_wifi_settings') else None, decode_wifi),
+ ]:
+ if call is None:
+ continue
+ try:
+ value = await call()
+ except Exception as e:
+ info[key] = f'error: {type(e).__name__}: {e}'
+ continue
+ if isinstance(value, bytes):
+ value = decode(value) if decode else value.hex(' ')
+ info[key] = value
+
+ # Login state, plus the reason for anything the camera turned down -- a
+ # refusal answers with an empty payload, so otherwise there is nothing
+ # above to say why a block came back blank.
+ info['auth'] = getattr(session, 'dev_properties', {}).get('auth')
+ refused = {
+ BinaryCommands(cmd).name: f'{code.name} ({code.value})'
+ for cmd, code in getattr(session, 'cmd_results', {}).items()
+ if code is not None and code < LibError.OK
+ }
+ if refused:
+ info['refused'] = refused
+ return web.json_response({'status': 'ok', 'info': info})
+
+
+async def wifi_scan(request):
+ """Trigger a Wi-Fi scan and return the raw result plus a best-effort list
+ of SSID-looking printable strings (the scan-list struct is unverified)."""
+ session, err = _get_session(request)
+ if err:
+ return err
+ if not hasattr(session, 'scan_wifi'):
+ return _json_error('not supported by this device', 501)
+ try:
+ data = await session.scan_wifi(timeout=12)
+ except Exception as e:
+ return _json_error(f'{type(e).__name__}: {e}', 500)
+ # Pull out printable ASCII runs as candidate SSIDs to make the raw
+ # dump readable; the exact record layout is still unknown.
+ candidates, run = [], bytearray()
+ for b in data:
+ if 32 <= b < 127:
+ run.append(b)
+ else:
+ if len(run) >= 3:
+ candidates.append(run.decode('ascii'))
+ run = bytearray()
+ if len(run) >= 3:
+ candidates.append(run.decode('ascii'))
+ return web.json_response({
+ 'status': 'ok',
+ 'length': len(data),
+ 'strings': candidates[:50],
+ 'raw': data.hex(' '),
+ })
+
+
+async def get_users(request):
+ """Configured device users (CMD_SYSTEM_USER_GET), decoded per the vendor
+ app's layout."""
+ session, err = _get_session(request)
+ if err:
+ return err
+ if not hasattr(session, 'get_users'):
+ return _json_error('not supported by this device', 501)
+ try:
+ data = await session.get_users(timeout=5)
+ except Exception as e:
+ return _json_error(f'{type(e).__name__}: {e}', 500)
+ return web.json_response({
+ 'status': 'ok',
+ 'user': parse_user_block(data) or None,
+ 'length': len(data),
+ 'raw': data.hex(' '),
+ })
+
+
+async def get_snapshot(request):
+ """Still image (ENH-002). CMD_SNAPSHOT_GET goes unanswered on all tested
+ hardware (FTYC + PTZA), so fall back to the latest reassembled video
+ frame; the x-snapshot-source header says which path served the image."""
+ session, err = _get_session(request)
+ if err:
+ return err
+
+ data, source = b'', 'camera'
+ if hasattr(session, 'get_snapshot'):
+ try:
+ data = await session.get_snapshot(timeout=3)
+ except Exception:
+ data = b''
+ if not data:
+ frame = getattr(session.frame_buffer, 'latest_frame', None)
+ if frame is not None:
+ data, source = frame.data, 'video-frame'
+ if not data:
+ return _json_error(
+ 'camera did not answer SNAPSHOT_GET and no video frame is buffered'
+ ' -- start the video stream once and retry', 504)
+ return web.Response(body=data, headers={
+ 'content-type': 'image/jpeg',
+ 'cache-control': 'no-store',
+ 'x-snapshot-source': source,
+ })
+
+
+def _wav_header(sample_rate=8000):
+ # Unknown-length stream: RIFF/data sizes are set to 0xFFFFFFFF, which
+ # browsers accept for live playback.
+ byte_rate = sample_rate * 2
+ return (
+ b'RIFF' + struct.pack(' element) was still tearing down, the old teardown stopped
+# the camera stream AFTER the new listener attached -- audio played for a
+# second and then starved. Only the last listener out stops the camera.
+_AUDIO_LISTENERS = {}
+
+
+async def stream_audio(request):
+ """Live audio as a streaming WAV (ENH-005). Starts the camera audio stream
+ for the first listener; the camera is stopped only when the last listener
+ disconnects."""
+ session, err = _get_session(request)
+ if err:
+ return err
+ if not hasattr(session, 'start_audio'):
+ return _json_error('not supported by this device', 501)
+ dev_id = request.match_info['dev_id']
+
+ response = web.StreamResponse()
+ response.content_type = 'audio/wav'
+ await response.prepare(request)
+
+ _AUDIO_LISTENERS[dev_id] = _AUDIO_LISTENERS.get(dev_id, 0) + 1
+ # Always (re)request: if a concurrent teardown just stopped the camera,
+ # is_audio_requested is False again and this re-arms it; otherwise
+ # start_audio is a no-op.
+ await session.start_audio()
+ try:
+ await response.write(_wav_header())
+ while True:
+ frame = await session.get_audio_frame()
+ await response.write(frame.data)
+ except (ConnectionResetError, asyncio.CancelledError):
+ pass
+ finally:
+ _AUDIO_LISTENERS[dev_id] = _AUDIO_LISTENERS.get(dev_id, 1) - 1
+ if _AUDIO_LISTENERS[dev_id] <= 0:
+ _AUDIO_LISTENERS.pop(dev_id, None)
+ try:
+ await session.stop_audio()
+ except Exception:
+ logger.debug('stop_audio on disconnect failed', exc_info=True)
+ return response
async def stream_video(request):
- dev_id_str = request.match_info['dev_id']
- if dev_id_str not in SESSIONS:
- return web.Response(
- text='{"status": "error", "message": "unknown device"}',
- headers={'content-type': 'application/json'},
- status=404,
- )
+ session, err = _get_session(request)
+ if err:
+ return err
response = web.StreamResponse()
boundary = '--frame' + uuid.uuid4().hex
@@ -133,32 +599,38 @@ async def stream_video(request):
response.content_length = 1000000000000
await response.prepare(request)
- session = SESSIONS[dev_id_str]
if not session.is_video_requested:
await session.start_video()
frame_buffer = session.frame_buffer
- try:
- while True:
- frame = await frame_buffer.get()
- header = f'--{boundary}\r\n'.encode()
- header += b'Content-Length: %d\r\n' % len(frame.data)
- header += b'Content-Type: image/jpeg\r\n\r\n'
- try:
- await response.write(header)
- await response.write(frame.data)
- except ConnectionResetError:
- logger.warning('Connection reset')
- break
- finally:
- return response
+ while True:
+ frame = await frame_buffer.get()
+ if not frame.data:
+ continue
+ header = f'--{boundary}\r\n'.encode()
+ header += b'Content-Length: %d\r\n' % len(frame.data)
+ header += b'Content-Type: image/jpeg\r\n\r\n'
+ try:
+ await response.write(header)
+ await response.write(frame.data)
+ except ConnectionResetError:
+ logger.warning('Connection reset')
+ break
+ return response
async def start_web_server(port=4000):
app = web.Application()
app.router.add_get('/', index)
+ app.router.add_get('/camera/{dev_id}', camera_page)
app.router.add_get('/{dev_id}/v', stream_video)
+ app.router.add_get('/{dev_id}/snapshot', get_snapshot)
+ app.router.add_get('/{dev_id}/params', get_params)
+ app.router.add_get('/{dev_id}/info', get_info)
+ app.router.add_get('/{dev_id}/wifi-scan', wifi_scan)
+ app.router.add_get('/{dev_id}/users', get_users)
+ app.router.add_get('/{dev_id}/audio', stream_audio)
app.router.add_post('/{dev_id}/c/{cmd}', handle_commands)
runner = web.AppRunner(app, handle_signals=True)
diff --git a/aiopppp/packets.py b/aiopppp/packets.py
index 7e026f5..5293d87 100644
--- a/aiopppp/packets.py
+++ b/aiopppp/packets.py
@@ -1,8 +1,22 @@
+import datetime
import json
import logging
import struct
-from .const import CAM_MAGIC, CC_DEST, BinaryCommands, PacketType
+from .const import (
+ CAM_MAGIC,
+ CC_DEST,
+ BinaryCommands,
+ ChipType,
+ DevFunc,
+ DevSysMode,
+ DevType,
+ PacketType,
+ SDCardStatus,
+ WifiMode,
+ WifiType,
+ enum_name,
+)
from .types import Channel, DeviceID
logger = logging.getLogger(__name__)
@@ -73,11 +87,20 @@ def get_drw_payload(self):
def xq_bytes_encode(data, shift):
new_buf = bytes(b - 1 if b & 1 else b + 1 for b in data)
+ if not new_buf:
+ return b''
+ # The rotation is modulo the buffer length; a raw shift larger than the
+ # payload (e.g. shift=4 on a 1-3 byte payload) would otherwise rotate by the
+ # wrong amount and fail to round-trip with xq_bytes_decode.
+ shift %= len(new_buf)
return bytes(new_buf[shift:] + new_buf[:shift])
def xq_bytes_decode(data, shift):
new_buf = bytes(b - 1 if b & 1 else b + 1 for b in data)
+ if not new_buf:
+ return b''
+ shift %= len(new_buf)
return bytes(new_buf[-shift:] + new_buf[:-shift])
def _inet_btoa(b: bytes) -> str:
@@ -119,7 +142,7 @@ def parse_dev_status(data):
bat_level, # 4-7 (int)
time_zone, # 8-11 (int)
rec_nmb, # 12-15 (int)
- sys_uptime, # 16-19 (int)
+ wifi_dbm, # 16-19 (int)
power_supply, # 20-23 (int)
dev_name, # 24-87 (64 bytes)
sd_status, # 88 (1 byte)
@@ -140,12 +163,49 @@ def parse_dev_status(data):
used_size # 120-123 (int)
) = struct.unpack('<4s5i64s10B6s4s4s3I', data[:124])
+ # time_zone is in seconds WEST of UTC (UTC+2 is stored as -7200, confirmed
+ # on PTZA hardware). Firmwares without a timezone leave a constant here, so
+ # only render a zone when the value is actually plausible.
+ utc_offset = _tz_west_seconds(time_zone)
+
+ # sw_ver is a packed word, not just a version: byte 1 is the device type
+ # and byte 3 the chip type (AppUtils.getDevTypeFromDevVer / getChpTypeFromDevVer).
+ sw_ver_int = struct.unpack('> 8) & 0xFF
+ chip_type = (sw_ver_int >> 24) & 0xFF
+
+ # power_supply is packed too: bit 0 external power, bits 4-7 sysMode,
+ # bits 24-31 a live function/state bitmap.
+ sys_mode = (power_supply >> 4) & 0x0F
+ func_bmp = (power_supply >> 24) & 0xFF
+ funcs = DevFunc(func_bmp)
+
+ # Not every firmware fills this word in. PTZA leaves it entirely zero --
+ # confirmed across captures with IR and the light toggled, where not one
+ # byte of the block moved -- while FTYC reports e.g. 0x14000101. A zero
+ # word would otherwise read as a confident "on battery, everything off".
+ # Distinguish with the battery: a camera reporting a real cell voltage has
+ # populated fields, one parked at the 8000 mains placeholder has not.
+ bat_percent = _bat_percent(bat_level)
+ power_populated = power_supply != 0 or bat_percent is not None
+
return {
- 'tz': f"UTC{time_zone // 3600:+d}", #time zone is in seconds
- 'uptime': sys_uptime,
- 'dbm': sys_uptime, #not sure if that is wifi dbm or system uptime
+ 'tz': f'UTC{utc_offset // 3600:+d}' if utc_offset is not None else None,
+ 'utcOffsetSeconds': utc_offset,
+ # The vendor SDK names this field sysUptime, but that name is the only
+ # thing about it that says "uptime": both apps feed it straight to
+ # setWifidbm() and render it through wlanSigGet(), whose buckets are
+ # RSSI ranges (-100/-85/-70/-55). It is the Wi-Fi signal. Cameras that
+ # don't report one leave a value outside the plausible range.
+ # See VENDOR_APP_FINDINGS.md.
+ 'dbm': wifi_dbm if -127 <= wifi_dbm <= -1 else None,
'devName': dev_name.decode('ascii', errors='ignore').rstrip('\0'),
'sdStatus': sd_status,
+ 'sdStatusName': enum_name(SDCardStatus, sd_status),
+ # Number of client sessions attached, NOT a status code: the vendor app
+ # renders it as "Connected: " into a view named tvSessionNmb, and
+ # FTYC reports 1 while a single client is connected. Keeping the
+ # upstream key name, misleading as it is, to avoid a breaking rename.
'p2pStatus': p2p_status,
'connType': conn_type,
'osdEnable': osd_enable,
@@ -153,20 +213,183 @@ def parse_dev_status(data):
'mode': mode,
'recEnableOnStart': rec_enable_on_start,
'picEnableOnStart': pic_enable_on_start,
- 'recNmb': rec_nmb,
- 'picNmb': pic_nmb,
+ # All-ones means "not tracked" on the tested firmwares. These two come
+ # out of the struct with different signedness (-1 vs 4294967295), which
+ # made the same non-answer look like two different numbers.
+ 'recNmb': None if rec_nmb in (-1, 0xFFFFFFFF) else rec_nmb,
+ 'picNmb': None if pic_nmb == 0xFFFFFFFF else pic_nmb,
+ # Raw, unscaled. The vendor app divides both by 1 MiB before display;
+ # whether the wire unit is bytes or KB is still unconfirmed (no SD card
+ # to test with), so no conversion is applied here.
'totalSize': total_size,
'usedSize': used_size,
+ # powerSupply is packed: bit 0 external power, bits 4-7 sysMode,
+ # bits 24-31 the function bitmap. batLevel is millivolts.
'powerSupply': power_supply,
+ # None (not False/0) when this firmware doesn't populate the word --
+ # saying "on battery" about a camera that never reported its power
+ # state is worse than saying nothing.
+ 'externalPower': bool(power_supply & 1) if power_populated else None,
+ 'sysMode': sys_mode if power_populated else None,
+ 'sysModeName': enum_name(DevSysMode, sys_mode, 'SYSMODE_') if power_populated else None,
+ # Bits 0 (fill light) and 1 (IR) confirmed on FTYC. The rest of the
+ # bitmap is unexplained, so only the raw byte is published for it.
+ 'funcBmp': func_bmp if power_populated else None,
+ 'funcFillLight': bool(funcs & DevFunc.FILL_LIGHT) if power_populated else None,
+ 'funcIrLed': bool(funcs & DevFunc.IR_LED) if power_populated else None,
'batLevel': bat_level,
+ 'batPercent': bat_percent,
'dhcp': dhcp,
'ipAddr': _inet_btoa(ip_addr_bytes),
'netmask': _inet_btoa(netmask_bytes),
'mac':mac.hex(':'),
'mcuver': _get_dev_version(sw_ver),
+ # swVer is a packed word: byte 1 = device type, byte 3 = chip type.
+ 'devType': dev_type,
+ 'devTypeName': enum_name(DevType, dev_type, 'DEV_'),
+ 'chipType': chip_type,
+ 'chipTypeName': enum_name(ChipType, chip_type, 'CHP_'),
+ # This offset was suspected of being swapped with alarmEnable, because
+ # both vendor apps read the equivalent position as alarmEnable. FTYC
+ # hardware says otherwise: toggling IR moved exactly this byte 0->1 and
+ # left alarmEnable at 0, so our layout is right and the apps' does not
+ # hold for this firmware. Confirmed 2026-08-25.
'icut': ir_cut,
- 'lamp': 0, # lamp is not in the status
+ # Confirmed on FTYC: the fill-light bit of funcBmp tracks the light
+ # command, so lamp state IS in the status block after all. Stays 0
+ # (not None) where the word is unpopulated -- consumers test for the
+ # key's presence to decide whether to offer a lamp entity at all.
+ 'lamp': int(bool(funcs & DevFunc.FILL_LIGHT)),
+ }
+
+
+def _cstr(b: bytes) -> str:
+ return b.split(b'\x00', 1)[0].decode('utf-8', errors='replace')
+
+
+# Resting-voltage curve for a single-cell LiPo, as (millivolts, percent).
+# Interpolated between points; the cells in these cameras charge to ~4.2 V.
+#
+# NOT the vendor app's thresholds: those pick one of five battery ICONS
+# (>=4350 / >=4200 / >=4100 / >=3950 / >=3900), and reading the third icon as
+# "60%" made a fully-charged camera sitting on the charger at 4195 mV report
+# 60% forever. Icon buckets are not percentages.
+_BATTERY_CURVE = (
+ (4200, 100), (4150, 95), (4110, 90), (4080, 85), (4020, 80),
+ (3980, 75), (3950, 70), (3910, 65), (3870, 60), (3850, 55),
+ (3840, 50), (3820, 45), (3800, 40), (3790, 35), (3770, 30),
+ (3750, 25), (3730, 20), (3710, 15), (3690, 10), (3610, 5),
+ (3270, 0),
+)
+
+
+def _bat_percent(mv):
+ """Battery millivolts -> percent, or None when the field isn't a battery
+ reading (mains-only cameras park it at 8000)."""
+ if not 3000 <= mv <= 4600:
+ return None
+ if mv >= _BATTERY_CURVE[0][0]:
+ return 100
+ if mv <= _BATTERY_CURVE[-1][0]:
+ return 0
+ for (hi_mv, hi_pct), (lo_mv, lo_pct) in zip(_BATTERY_CURVE, _BATTERY_CURVE[1:]):
+ if mv >= lo_mv:
+ span = hi_mv - lo_mv
+ return round(lo_pct + (hi_pct - lo_pct) * (mv - lo_mv) / span)
+ return 0
+
+
+def _render_ts(ts):
+ return datetime.datetime.utcfromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S')
+
+
+def _tz_west_seconds(value):
+ """Return the UTC offset (seconds EAST) for a raw 'seconds west of UTC'
+ field, or None when the value can't be a timezone.
+
+ Firmwares that don't store a timezone leave a constant in this field
+ (FTYC reports 224), which naively rendered as a bogus zone like UTC-1.
+ A real timezone is a whole number of 15-minute steps within +/-14 h."""
+ if value % 900 or abs(value) > 14 * 3600:
+ return None
+ return -value
+
+
+def parse_datetime_block(data):
+ """Decode a CMD_SYSTEM_DATETIME_GET response (80 bytes, two firmware
+ variants confirmed on hardware):
+
+ - PTZA: u32 unix timestamp (UTC), i32 timezone as seconds WEST of UTC
+ (UTC+2 stored as -7200), 8 pad bytes, char ntp_server[64].
+ - FTYC: u32 timestamp that already renders as LOCAL time (the camera adds
+ its own internally-stored offset when the clock is set), then constant
+ non-tz fields, ntp_server at the same offset. There is no tz in the
+ block, so the field-4 value (e.g. 0xE0) must not be shown as one.
+
+ The variants are told apart by tz plausibility: a real timezone is a
+ multiple of 15 minutes within +/-14 h."""
+ if len(data) < 8:
+ return {}
+ ts, field4 = struct.unpack_from('= 80:
+ result['ntpServer'] = _cstr(data[16:80])
+ return result
+
+
+def parse_user_block(data):
+ """Decode a CMD_SYSTEM_USER_GET response. Layout from the vendor app's
+ IpcByte2ObjectParser.ParseUser (minus its 4-byte JNI prefix):
+ char account[32], char password[128]."""
+ if len(data) < 160:
+ return {}
+ return {
+ 'account': _cstr(data[0:32]),
+ 'password': _cstr(data[32:160]),
+ }
+
+
+def parse_wifi_settings(data):
+ """Decode a CMD_NET_WIFISETTING_GET response (layout confirmed on PTZA
+ hardware, len=264): u32 mode, 12 pad bytes, u32 security, 4 pad bytes,
+ char ssid[32], char password[128], then five char[16] dotted-quad strings
+ (ip, netmask, gateway, dns1, dns2)."""
+ if len(data) < 184:
+ return {}
+ mode, = struct.unpack_from('= off + 16:
+ result[key] = _cstr(data[off:off + 16])
+ return result
+
class BinaryCmdPkt(DrwPkt):
START_CMD = b'\x11\x0a'
@@ -258,6 +481,11 @@ def parse_drw_pkt(data):
return DrwPkt(channel, cmd_idx, data[4:])
+def make_audio_drw_pkt(cmd_idx, payload):
+ """Outgoing audio (talk-back) frame on the audio DRW channel."""
+ return DrwPkt(Channel.Audio, cmd_idx, payload)
+
+
def make_drw_ack_pkt(drw_pkt):
return Packet(
PacketType.DrwAck,
@@ -299,7 +527,14 @@ def parse_packet(data):
'Invalid pkt length: pkt.len=%d, real length=%d, [%s]',
length, len(data) - 4, data.hex(' '))
- pkt_class, parse_func = PARSERS.get(PacketType(typ), (Packet, None))
+ try:
+ packet_type = PacketType(typ)
+ except ValueError:
+ # A corrupt or unrecognized datagram must not raise out of the UDP
+ # receive callback; surface it as ValueError so callers drop it.
+ raise ValueError(f'Unknown packet type 0x{typ:02x}')
+
+ pkt_class, parse_func = PARSERS.get(packet_type, (Packet, None))
if parse_func is None:
- return pkt_class(PacketType(typ), data[4:])
+ return pkt_class(packet_type, data[4:])
return parse_func(data[4:])
diff --git a/aiopppp/session.py b/aiopppp/session.py
index e00765d..b131dc2 100644
--- a/aiopppp/session.py
+++ b/aiopppp/session.py
@@ -9,18 +9,25 @@
JSON_COMMAND_NAMES,
PTZ,
BinaryCommands,
+ CgiCommands,
JsonCommands,
+ LibError,
PacketType,
PtzDirection,
PtzParamType,
+ PtzPrefab,
VideoParamType,
VideoResolution,
+ VideoRotate,
+ enum_name,
)
+from .audio import CODECS
from .encrypt import ENC_METHODS
from .exceptions import AuthError, CommandResultError
from .packets import (
BinaryCmdPkt,
JsonCmdPkt,
+ make_audio_drw_pkt,
make_close_pkt,
make_drw_ack_pkt,
make_p2palive_ack_pkt,
@@ -30,11 +37,27 @@
parse_dev_status,
parse_packet,
)
-from .types import Channel, DeviceDescriptor, VideoFrame
+from .types import AudioFrame, Channel, DeviceDescriptor, VideoFrame
from .utils import DebounceEvent
logger = logging.getLogger(__name__)
+# Prefix of the 0x20-byte header that marks the first chunk of a video frame.
+VIDEO_MARKER = b'\x55\xaa\x15\xa8'
+# Byte 4 of the 0x20 header is the stream type (captured from FTYC hardware;
+# matches cam-reverse). FTYC muxes audio onto the VIDEO DRW channel: each video
+# frame is preceded by one audio packet with the same magic but type 0x06.
+STREAM_TYPE_JPEG = 0x03
+STREAM_TYPE_AUDIO = 0x06
+
+
+class SessionLogAdapter(logging.LoggerAdapter):
+ """Tag every session log line with the device ID. Several cameras log
+ through this module concurrently; untagged lines are unattributable."""
+
+ def process(self, msg, kwargs):
+ return f'[{self.extra["dev"]}] {msg}', kwargs
+
class State(Enum):
DISCONNECTED = 0
@@ -84,6 +107,21 @@ def __init__(self, *args, **kwargs):
self.video_received = {}
self.video_boundaries = set()
self.last_video_frame = -1
+ # The frame currently being assembled is delimited by the top two
+ # boundaries. We track that window and the set of still-missing chunk
+ # indices in it incrementally, so completeness is an O(1) set update per
+ # chunk instead of an O(frame) rescan (which was O(frame^2) per frame).
+ self._frame_window = (None, None)
+ self._frame_missing = set()
+ # Chunks seen since the last frame-boundary header. If this grows large
+ # the camera is streaming video we can't frame (e.g. a different marker
+ # than VIDEO_MARKER) -- log a sample so the format can be identified.
+ self._chunks_since_boundary = 0
+ # Header diagnostics: some firmwares (FTYC) put the 0x20-byte stream
+ # header on far more chunks than one per frame. Sample a few headers so
+ # the type/length fields can be identified from a plain log.
+ self._boundary_headers_logged = 0
+ self._boundaries_seen = 0
async def process_video_queue(self):
while True:
@@ -95,60 +133,103 @@ def start_video_queue(self):
async def handle_incoming_video_packet(self, pkt_epoch, pkt):
video_payload = pkt.get_drw_payload()
- # logger.info(f'- video frame {pkt._cmd_idx}')
- video_marker = b'\x55\xaa\x15\xa8' # next \x03 - video marker
+ # self.log.info(f'- video frame {pkt._cmd_idx}')
video_chunk_idx = pkt._cmd_idx + 0x10000 * pkt_epoch
# 0x20 - size of the header starting with this magic
- if video_payload.startswith(video_marker):
+ if video_payload.startswith(VIDEO_MARKER):
+ self._chunks_since_boundary = 0
+ self._boundaries_seen += 1
+ if self._boundary_headers_logged < 8 or self._boundaries_seen % 5000 == 0:
+ self._boundary_headers_logged += 1
+ self.log.info('stream header sample #%d (payload len=%d): [%s]',
+ self._boundaries_seen, len(video_payload),
+ video_payload[:0x20].hex(' '))
+ stream_type = video_payload[4] if len(video_payload) > 4 else None
+ if stream_type == STREAM_TYPE_AUDIO:
+ # Muxed audio (FTYC): not a frame boundary. Occupy the index
+ # with an empty chunk so the surrounding video frame's window
+ # still completes, and hand the packet to the audio pipeline.
+ self.video_received[video_chunk_idx] = b''
+ await self.handle_incoming_audio_packet(pkt)
+ await self.process_video_frame(video_chunk_idx)
+ return
self.video_boundaries.add(video_chunk_idx)
self.video_received[video_chunk_idx] = video_payload[0x20:]
else:
self.video_received[video_chunk_idx] = video_payload
- await self.process_video_frame()
-
- async def process_video_frame(self):
+ self._chunks_since_boundary += 1
+ if self._chunks_since_boundary in (100, 1000, 10000):
+ self.log.warning(
+ 'No frame boundary in %d video chunks; payload head: [%s]',
+ self._chunks_since_boundary, video_payload[:32].hex(' '),
+ )
+ await self.process_video_frame(video_chunk_idx)
+
+ async def process_video_frame(self, new_idx=None):
if len(self.video_boundaries) <= 1:
return
- frame_starts = sorted(list(self.video_boundaries))
+ # After pruning, video_boundaries only holds the current pending pair
+ # (plus any freshly-arrived higher boundary), so this sort is over a
+ # handful of items.
+ frame_starts = sorted(self.video_boundaries)
index = frame_starts[-2]
last_index = frame_starts[-1]
- if index == self.last_video_frame:
- return
+ if (index, last_index) != self._frame_window:
+ # The frame window advanced. Recompute the missing set and drop
+ # everything below the new frame start. Both are O(frame) but run
+ # once per frame here, not once per incoming chunk.
+ self._frame_window = (index, last_index)
+ self._frame_missing = {i for i in range(index, last_index) if i not in self.video_received}
+ for idx in [i for i in self.video_received if i < index]:
+ del self.video_received[idx]
+ for idx in [i for i in self.video_boundaries if i < index]:
+ self.video_boundaries.discard(idx)
+ elif new_idx is not None:
+ # Same window: the chunk we just stored may have filled a gap.
+ self._frame_missing.discard(new_idx)
- complete = True
- out = []
- completeness = ''
- for i in range(index, last_index):
- if self.video_received.get(i) is not None:
- out.append(self.video_received[i])
- completeness += 'x'
- else:
- complete = False
- completeness += '_'
- logger.debug(f".. completeness: {completeness}")
-
- if complete:
+ if index != self.last_video_frame and not self._frame_missing:
self.last_video_frame = index
-
- await self.frame_buffer.publish(VideoFrame(idx=index, data=b''.join(out)))
-
- to_delete = [idx for idx in self.video_received.keys() if idx < index]
- for idx in to_delete:
- del self.video_received[idx]
- to_delete = [idx for idx in self.video_boundaries if idx < index]
- for idx in to_delete:
- self.video_boundaries.remove(idx)
+ data = b''.join(self.video_received[i] for i in range(index, last_index))
+ # A reassembled MJPEG frame must be SOI..EOI; log rejects so a
+ # polluted stream (e.g. muxed sub-streams) is visible in the log.
+ if self.log.isEnabledFor(logging.DEBUG):
+ valid = data[:2] == b'\xff\xd8'
+ self.log.debug('publish frame idx=%s len=%d head=[%s]%s',
+ index, len(data), data[:4].hex(' '),
+ '' if valid else ' NOT-JPEG')
+ if data:
+ # FTYC frames a stream as [bare 0x20-header pkt][header+data
+ # pkt][data...]: the two adjacent header chunks make a
+ # zero-length "frame" between them. Publishing it emitted a
+ # Content-Length: 0 MJPEG part after every real frame, which
+ # froze browsers on the first image.
+ await self.frame_buffer.publish(VideoFrame(idx=index, data=data))
+
+ if self.log.isEnabledFor(logging.DEBUG):
+ completeness = ''.join(
+ 'x' if i in self.video_received else '_'
+ for i in range(index, last_index)
+ )
+ self.log.debug('.. completeness: %s', completeness)
class Session(PacketQueueMixin, VideoQueueMixin):
- def __init__(self, dev, on_disconnect, *args, **kwargs):
+ # If no packet arrives from the camera for this many seconds, treat the
+ # connection as dead and tear it down. Works for both JSON and binary
+ # cameras (binary has no other liveness check), and catches a silently
+ # dropped peer that would otherwise leave a zombie session.
+ RECV_TIMEOUT_SEC = 20
+
+ def __init__(self, dev, on_disconnect, *args, on_video_state_change=None, **kwargs):
super().__init__(*args, **kwargs)
self.state = State.DISCONNECTED
self.dev = dev
+ self.log = SessionLogAdapter(logger, {'dev': dev.dev_id.dev_id})
self.dev_properties = {}
self.outgoing_command_idx = 0
self.transport = None
@@ -157,7 +238,11 @@ def __init__(self, dev, on_disconnect, *args, **kwargs):
self.video_stale_at = None
self.last_alive_pkt_at = datetime.datetime.now()
self.last_drw_pkt_at = datetime.datetime.now()
+ self.last_recv_at = datetime.datetime.now()
self.on_disconnect = on_disconnect
+ # Called with the new is_video_requested value whenever streaming
+ # starts or stops (including when the session is torn down).
+ self.on_video_state_change = on_video_state_change
self.main_task = None
self.drw_waiters = {}
self.cmd_waiters = {}
@@ -166,6 +251,10 @@ def __init__(self, dev, on_disconnect, *args, **kwargs):
def __str__(self):
return f'Session({self.dev.dev_id}) ({self.state.name})'
+ def _notify_video_state(self):
+ if self.on_video_state_change:
+ self.on_video_state_change(self.is_video_requested)
+
async def create_udp(self):
loop = asyncio.get_running_loop()
transport, _ = await loop.create_datagram_endpoint(
@@ -175,10 +264,20 @@ async def create_udp(self):
return transport
def on_receive(self, data):
- decoded = ENC_METHODS[self.dev.encryption][0](data)
- pkt = parse_packet(decoded)
- # logger.debug(f"recv< {pkt} {pkt.get_payload()}")
- logger.debug(f"recv< {pkt.type}, len={len(pkt.get_payload())}")
+ # The transport is bound to the camera's address, so any datagram here
+ # is proof of life for the dead-connection check in loop_step().
+ self.last_recv_at = datetime.datetime.now()
+ try:
+ decoded = ENC_METHODS[self.dev.encryption][0](data)
+ pkt = parse_packet(decoded)
+ except (ValueError, struct.error, KeyError, IndexError):
+ # One malformed datagram must never raise out of the asyncio
+ # datagram callback (which would spam "Exception in callback" and,
+ # in the worst case, wedge the transport). Log and drop it.
+ self.log.debug('Dropping undecodable datagram (%d bytes): [%s]', len(data), data[:16].hex(' '))
+ return
+ # self.log.debug(f"recv< {pkt} {pkt.get_payload()}")
+ self.log.debug(f"recv< {pkt.type}, len={len(pkt.get_payload())}")
self.packet_queue.put_nowait(pkt)
async def call_with_error_check(self, coro):
@@ -192,10 +291,27 @@ async def call_with_error_check(self, coro):
async def send(self, pkt):
await self.call_with_error_check(self._send(pkt))
+ # Cap on outstanding DRW ACK waiters. A waiter is created for every DRW we
+ # send but only removed when its ACK arrives (handle_drw_ack) or its wait
+ # times out (_wait_ack). Fire-and-forget commands (reboot, toggle_*, PTZ)
+ # never wait, so their waiters would linger; bound the dict and evict the
+ # oldest so it can never grow without limit.
+ MAX_DRW_WAITERS = 256
+
async def _send(self, pkt):
- logger.debug(f"send> {pkt}")
+ self.log.debug(f"send> {pkt}")
if pkt.type == PacketType.Drw:
+ existing = self.drw_waiters.get(pkt._cmd_idx)
+ if existing is not None and not existing.done():
+ # The 16-bit index wrapped back onto a still-pending waiter; that
+ # old send will never be matched now, so discard it.
+ existing.cancel()
self.drw_waiters[pkt._cmd_idx] = asyncio.Future()
+ while len(self.drw_waiters) > self.MAX_DRW_WAITERS:
+ old_idx, old_fut = next(iter(self.drw_waiters.items()))
+ del self.drw_waiters[old_idx]
+ if not old_fut.done():
+ old_fut.cancel()
encoded_pkt = ENC_METHODS[self.dev.encryption][1](bytes(pkt))
self.transport.sendto(encoded_pkt, (self.dev.addr, self.dev.port))
@@ -213,14 +329,14 @@ async def handle_incoming_packet(self, pkt):
elif pkt.type == PacketType.Drw:
await self.handle_drw(pkt)
elif pkt.type == PacketType.DrwAck:
- logger.debug(f'Got DRW ACK {pkt}')
+ self.log.debug(f'Got DRW ACK {pkt}')
await self.handle_drw_ack(pkt)
elif pkt.type == PacketType.P2PAliveAck:
- logger.debug(f'Got P2PAlive ACK {pkt}')
+ self.log.debug(f'Got P2PAlive ACK {pkt}')
elif pkt.type == PacketType.Close:
await self.handle_close(pkt)
else:
- logger.warning(f'Got UNKNOWN {pkt}')
+ self.log.warning(f'Got UNKNOWN {pkt}')
async def login(self):
pass
@@ -228,19 +344,23 @@ async def login(self):
async def start_video(self):
await self.device_is_ready.wait()
if not self.is_video_requested:
- logger.info('Start video')
+ self.log.info('Start video')
self.last_drw_pkt_at = datetime.datetime.now()
await self._request_video(1)
self.is_video_requested = True
+ self._notify_video_state()
async def stop_video(self):
if self.is_video_requested:
self.is_video_requested = False
+ self._notify_video_state()
self.video_stale_at = None
self.video_received = {}
self.video_boundaries = set()
self.video_epoch = 0
self.last_video_frame = -1
+ self._frame_window = (None, None)
+ self._frame_missing = set()
while not self.video_chunk_queue.empty():
self.video_chunk_queue.get_nowait()
await self._request_video(0)
@@ -252,20 +372,71 @@ async def _request_video(self, mode):
pass
async def handle_drw(self, drw_pkt):
- logger.debug('handle_drw(idx=%s, chn=%s)', drw_pkt._cmd_idx, drw_pkt._channel)
+ self.log.debug('handle_drw(idx=%s, chn=%s)', drw_pkt._cmd_idx, drw_pkt._channel)
await self.send(make_drw_ack_pkt(drw_pkt))
+ self.last_drw_pkt_at = datetime.datetime.now()
+
+ if drw_pkt._channel == Channel.Video:
+ # The camera counts the DRW index independently per channel, so only
+ # video-channel packets may drive epoch/wraparound tracking. Feeding
+ # command/audio indices (which advance on their own) in here would
+ # spuriously flip video_epoch and corrupt frame reassembly by an
+ # 0x10000 index shift.
+ pkt_epoch = self._get_drw_epoch(drw_pkt)
+ if pkt_epoch > self.video_epoch:
+ self.log.info('Video epoch changed %s -> %s', self.video_epoch, pkt_epoch)
+ self.video_epoch = pkt_epoch
+ self.last_drw_pkt_idx = drw_pkt._cmd_idx
+ elif pkt_epoch == self.video_epoch and self.last_drw_pkt_idx < drw_pkt._cmd_idx:
+ # Only chunks from the CURRENT epoch may advance the high-water
+ # mark. A late pre-wrap chunk (e.g. idx 65529 retransmitted
+ # after the counter wrapped to 0) belongs to the previous epoch;
+ # feeding it in here re-armed the wrap detector and the next
+ # post-wrap chunk bumped the epoch again -- ping-ponging the
+ # epoch up several times a second (seen on FTYC hardware,
+ # epochs 0->6 in seconds) and scattering reassembly indices
+ # 0x10000 apart, which killed every frame after the first.
+ self.last_drw_pkt_idx = drw_pkt._cmd_idx
+
+ if self.video_stale_at:
+ self.log.warning('Got video data while stale')
+ self.video_stale_at = None
+ self.video_chunk_queue.put_nowait((pkt_epoch, drw_pkt))
+ elif drw_pkt._channel == Channel.Audio:
+ await self.handle_incoming_audio_packet(drw_pkt)
+ elif drw_pkt._channel == Channel.Command:
+ await self.handle_incoming_command_packet(drw_pkt)
+
+ def _get_drw_epoch(self, drw_pkt):
+ if self.last_drw_pkt_idx > 0xff00 and drw_pkt._cmd_idx < 0x100:
+ return self.video_epoch + 1
+ if self.video_epoch and self.last_drw_pkt_idx < 0x100 and drw_pkt._cmd_idx > 0xff00:
+ return self.video_epoch - 1
+ return self.video_epoch
+
+ async def handle_incoming_command_packet(self, drw_pkt):
+ pass
+
+ async def handle_incoming_audio_packet(self, drw_pkt):
+ pass
+
+ def _reset_cmd_waiter(self, cmd):
+ # Replace any pending response future for this command. Without this a
+ # second request whose first response never arrived would silently
+ # orphan the old future (and its awaiter would hang until timeout).
+ old = self.cmd_waiters.get(cmd.value)
+ if old is not None and not old.done():
+ old.cancel()
+ fut = asyncio.Future()
+ self.cmd_waiters[cmd.value] = fut
+ return fut
async def handle_drw_ack(self, pkt):
cmd_idx_ack = int.from_bytes(pkt.get_payload()[4:6], 'big')
- logger.debug('handle_drw_ack(idx=%s)', cmd_idx_ack)
- # logger.info('waiters: %s', self.drw_waiters)
- if cmd_idx_ack in self.drw_waiters:
- # logger.info(
- # 'Got ACK for %d, proceed waiters, total waiters: %d', cmd_idx_ack, len(self.drw_waiters),
- # )
- self.drw_waiters[cmd_idx_ack].set_result(pkt)
- await asyncio.sleep(0)
- del self.drw_waiters[cmd_idx_ack]
+ self.log.debug('handle_drw_ack(idx=%s)', cmd_idx_ack)
+ fut = self.drw_waiters.pop(cmd_idx_ack, None)
+ if fut is not None and not fut.done():
+ fut.set_result(pkt)
async def wait_ack(self, idx, timeout=5):
return await self.call_with_error_check(self._wait_ack(idx, timeout))
@@ -275,16 +446,16 @@ async def _wait_ack(self, idx, timeout=5):
raise ValueError('Need to provide numeric command index')
fut = self.drw_waiters.get(idx)
if fut:
- logger.debug(f'Waiting for ACK for {idx}')
+ self.log.debug(f'Waiting for ACK for {idx}')
try:
await asyncio.wait_for(fut, timeout=timeout)
- logger.debug('wait_ack(idx=%d) complete, waiters: %d', idx, len(self.drw_waiters))
+ self.log.debug('wait_ack(idx=%d) complete, waiters: %d', idx, len(self.drw_waiters))
except asyncio.TimeoutError:
self.drw_waiters.pop(idx, None)
raise
async def handle_close(self, pkt):
- logger.info('%s requested close', self.dev.dev_id)
+ self.log.info('peer requested close')
self._on_device_lost()
async def setup_device(self):
@@ -300,13 +471,23 @@ async def _run(self):
await self.send_initial_packets()
try:
- await asyncio.wait_for(self._p2p_rdy_debouncer.wait(), timeout=10)
- logger.info('Connected to %s at %s, json=%s', self.dev.dev_id, self.dev.addr, self.dev.is_json)
+ try:
+ await asyncio.wait_for(self._p2p_rdy_debouncer.wait(), timeout=10)
+ except asyncio.TimeoutError:
+ # Camera answered discovery but never completed the P2pRdy
+ # handshake (common when it is flaky/half-wedged). Treat it as a
+ # lost device rather than letting an unhandled exception escape
+ # and take the whole process down.
+ self.log.warning('did not become ready (no P2pRdy), disconnecting')
+ await self.send_close_pkt()
+ self._on_device_lost()
+ return
+ self.log.info('Connected at %s, json=%s', self.dev.addr, self.dev.is_json)
self.state = State.CONNECTED
try:
await self.setup_device()
except asyncio.TimeoutError:
- logger.error('Timeout during device setup')
+ self.log.error('Timeout during device setup')
await self.send_close_pkt()
self._on_device_lost()
return
@@ -316,15 +497,57 @@ async def _run(self):
await asyncio.sleep(1)
except asyncio.CancelledError:
if self.transport:
- logger.debug('Session main task cancelled, sending close packet')
+ self.log.debug('Session main task cancelled, sending close packet')
await self.send_close_pkt()
raise
+ except Exception:
+ # A single session must never crash the whole process. Log it, tear
+ # the session down, and let discovery/HA reconnect.
+ self.log.exception('Session failed; disconnecting')
+ try:
+ await self.send_close_pkt()
+ except Exception:
+ pass
+ self._on_device_lost()
+ return
+
+ # Seconds of no video (while streaming) before we re-request it, and the
+ # further grace period before giving up on the connection.
+ VIDEO_REREQUEST_SEC = 5
+ VIDEO_DEAD_SEC = 10
async def loop_step(self):
- logger.debug(f"iterate in Session for {self.dev.dev_id}")
- if (datetime.datetime.now() - self.last_alive_pkt_at).total_seconds() > 10:
- self.last_alive_pkt_at = datetime.datetime.now()
- logger.info('Send P2PAlive')
+ self.log.debug("iterate in Session")
+ now = datetime.datetime.now()
+
+ # Video liveness. Applies to both protocols: a binary camera that keeps
+ # answering P2PAlive but sends no video would otherwise pass the base
+ # receive-timeout check forever and zombie. Re-request after a short
+ # gap, then disconnect if that doesn't revive the stream.
+ if (
+ self.is_video_requested and not self.video_stale_at and
+ (now - self.last_drw_pkt_at).total_seconds() > self.VIDEO_REREQUEST_SEC
+ ):
+ self.video_stale_at = self.last_drw_pkt_at
+ self.log.info('No video for %ds. Re-requesting video', self.VIDEO_REREQUEST_SEC)
+ await self._request_video(1)
+ if self.video_stale_at and (now - self.video_stale_at).total_seconds() > self.VIDEO_DEAD_SEC:
+ self.log.warning('No video for %ds. Disconnecting', self.VIDEO_DEAD_SEC)
+ await self.send_close_pkt()
+ self._on_device_lost()
+ return
+
+ if (now - self.last_recv_at).total_seconds() > self.RECV_TIMEOUT_SEC:
+ self.log.warning(
+ 'No packets from %s for %ds: connection is dead, disconnecting',
+ self.dev.dev_id, self.RECV_TIMEOUT_SEC,
+ )
+ await self.send_close_pkt()
+ self._on_device_lost()
+ return
+ if (now - self.last_alive_pkt_at).total_seconds() > 10:
+ self.last_alive_pkt_at = now
+ self.log.info('Send P2PAlive')
await self.send(make_p2palive_pkt())
def start(self):
@@ -338,21 +561,37 @@ def running_tasks(self):
return tuple(x for x in (self.main_task, self.process_packet_task, self.process_video_task) if x)
def _on_device_lost(self):
- logger.warning('Device %s lost', self.dev.dev_id)
+ self.log.warning('Device lost')
self.stop()
if self.on_disconnect:
self.on_disconnect(self.dev)
def stop(self):
- if self.state != State.CONNECTED:
- raise RuntimeError('Session is not started')
- logger.info('Stopping task for %s', self.dev.dev_id)
+ if self.state == State.DISCONNECTED and self.transport is None:
+ # Already fully stopped. stop() is reachable from _on_device_lost(),
+ # Device.close() and the CLI shutdown loop, so it must be idempotent.
+ # Note: a session that started connecting but never reached CONNECTED
+ # (e.g. P2pRdy timeout) is still DISCONNECTED but has a live transport
+ # and queue tasks, so we must fall through and clean those up.
+ return
+ self.log.info('Stopping session tasks')
self.device_is_ready.set()
- self.process_packet_task.cancel()
- self.process_video_task.cancel()
- self.main_task.cancel()
- self.transport.close()
- self.transport = None
+ reassert_task = getattr(self, '_reassert_task', None)
+ if reassert_task and not reassert_task.done():
+ reassert_task.cancel()
+ if self.process_packet_task:
+ self.process_packet_task.cancel()
+ if self.process_video_task:
+ self.process_video_task.cancel()
+ if self.main_task:
+ self.main_task.cancel()
+ if self.transport:
+ self.transport.close()
+ self.transport = None
+ if self.is_video_requested:
+ # The session is going away, so streaming has effectively stopped.
+ self.is_video_requested = False
+ self._notify_video_state()
self.state = State.DISCONNECTED
async def reboot(self):
@@ -389,10 +628,12 @@ async def send_command(self, cmd, *, with_response=False, **kwargs):
'cmd': cmd.value,
}
pkt_idx = self.outgoing_command_idx
- self.outgoing_command_idx += 1
+ # The index is sent as a 16-bit field and ACKs only echo 16 bits, so it
+ # must wrap; otherwise sends raise struct.error and ACK matching breaks.
+ self.outgoing_command_idx = (self.outgoing_command_idx + 1) & 0xFFFF
pkt = JsonCmdPkt(pkt_idx, {**data, **kwargs, **self.get_common_data()})
if with_response:
- self.cmd_waiters[cmd.value] = asyncio.Future()
+ self._reset_cmd_waiter(cmd)
await self.send(pkt)
return pkt_idx
@@ -405,48 +646,18 @@ async def login(self):
return True
async def _request_video(self, mode):
- logger.info('Request video %s', mode)
+ self.log.info('Request video %s', mode)
await self.send_command(JsonCommands.CMD_STREAM, video=mode)
- def _get_drw_epoch(self, drw_pkt):
- if self.last_drw_pkt_idx > 0xff00 and drw_pkt._cmd_idx < 0x100:
- return self.video_epoch + 1
- if self.video_epoch and self.last_drw_pkt_idx < 0x100 and drw_pkt._cmd_idx > 0xff00:
- return self.video_epoch - 1
- return self.video_epoch
-
- async def handle_drw(self, drw_pkt):
- await super().handle_drw(drw_pkt)
- self.last_drw_pkt_at = datetime.datetime.now()
-
- # # 0x10000 - max number of chunks in one epoch,we need to keep order of chunks
- pkt_epoch = self._get_drw_epoch(drw_pkt)
-
- if pkt_epoch > self.video_epoch:
- logger.info('Video epoch changed %s -> %s', self.video_epoch, pkt_epoch)
- self.video_epoch = pkt_epoch
- self.last_drw_pkt_idx = drw_pkt._cmd_idx
- elif self.last_drw_pkt_idx < drw_pkt._cmd_idx:
- self.last_drw_pkt_idx = drw_pkt._cmd_idx
-
- if drw_pkt._channel == Channel.Video:
- # logger.debug(f'Got video data {drw_pkt.get_drw_payload()}')
- if self.video_stale_at:
- logger.warning('Got video data while stale')
- self.video_stale_at = None
- self.video_chunk_queue.put_nowait((pkt_epoch, drw_pkt))
- elif drw_pkt._channel == Channel.Audio:
- pass
- elif drw_pkt._channel == Channel.Command:
- await self.handle_incoming_command_packet(drw_pkt)
-
async def handle_incoming_command_packet(self, drw_pkt):
if isinstance(drw_pkt, JsonCmdPkt):
response = drw_pkt.json_payload
- if response['cmd'] in self.cmd_waiters:
- # logger.debug('Got awaited response %s', response)
- self.cmd_waiters[response['cmd']].set_result(response)
- del self.cmd_waiters[response['cmd']]
+ fut = self.cmd_waiters.get(response['cmd'])
+ # Resolve but don't remove the waiter here: wait_cmd_result looks it
+ # up by key *after* wait_ack, so popping now would lose a result that
+ # arrives before the caller starts awaiting it.
+ if fut is not None and not fut.done():
+ fut.set_result(response)
async def wait_cmd_result(self, cmd, timeout=5):
return await self.call_with_error_check(self._wait_cmd_result(cmd, timeout))
@@ -454,15 +665,18 @@ async def wait_cmd_result(self, cmd, timeout=5):
async def _wait_cmd_result(self, cmd, timeout=5):
fut = self.cmd_waiters.get(cmd.value)
if fut:
- res = await asyncio.wait_for(fut, timeout=timeout)
- logger.debug('Got command result %s', res)
+ try:
+ res = await asyncio.wait_for(fut, timeout=timeout)
+ finally:
+ self.cmd_waiters.pop(cmd.value, None)
+ self.log.debug('Got command result %s', res)
return res
return {'result': -1}
async def setup_device(self):
auth = await self.login()
idx = await self.send_command(JsonCommands.CMD_GET_PARMS, with_response=True)
- # logger.debug('Waiting for params ack')
+ # self.log.debug('Waiting for params ack')
await self.wait_ack(idx)
# {
@@ -487,24 +701,9 @@ async def setup_device(self):
del cam_properties[f]
self.dev_properties = cam_properties
self.dev_properties['auth'] = auth
- logger.info('Camera properties: %s', cam_properties)
+ self.log.info('Camera properties: %s', cam_properties)
self.device_is_ready.set()
- async def loop_step(self):
- if (
- self.is_video_requested and not self.video_stale_at and
- (datetime.datetime.now() - self.last_drw_pkt_at).total_seconds() > 5
- ):
- self.video_stale_at = self.last_drw_pkt_at
- logger.info('No video for 5 seconds. Re-request video ')
- await self._request_video(1)
- if self.video_stale_at and (datetime.datetime.now() - self.video_stale_at).total_seconds() > 10:
- # camera disconnected
- logger.warning('No video for 10 seconds. Disconnecting')
- await self.send_close_pkt()
- self._on_device_lost()
- await super().loop_step()
-
async def control(self, no_ack=False, **kwargs):
idx = await self.send_command(JsonCommands.CMD_DEV_CONTROL, **kwargs)
if not no_ack:
@@ -514,23 +713,24 @@ async def toggle_lamp(self, value):
await self.control(lamp=1 if value else 0)
async def toggle_whitelight(self, value, **kwargs):
- logger.info('%s: toggle white light = %s', self.dev.dev_id, value)
+ self.log.info('toggle white light = %s', value)
idx = await self.send_command(JsonCommands.CMD_SET_WHITELIGHT, status=value)
await self.wait_ack(idx)
async def toggle_ir(self, value):
- logger.info('%s: toggle IR = %s', self.dev.dev_id, value)
- idx = await self.control(icut=1 if value else 0)
- await self.wait_ack(idx)
+ self.log.info('toggle IR = %s', value)
+ # control() already waits for the ACK; it returns None, so the previous
+ # `await self.wait_ack(idx)` raised ValueError on every call.
+ await self.control(icut=1 if value else 0)
async def rotate_start(self, value):
- logger.info('%s: rotate_start %s', self.dev.dev_id, value)
+ self.log.info('rotate_start %s', value)
value = PTZ[f'{value.upper()}_START'].value
idx = await self.send_command(JsonCommands.CMD_PTZ_CONTROL, parms=0, value=value)
await self.wait_ack(idx)
async def rotate_stop(self, **kwargs):
- logger.info('%s: rotate_stop', self.dev.dev_id)
+ self.log.info('rotate_stop')
indexes = []
for value in [PTZ.LEFT_STOP, PTZ.RIGHT_STOP, PTZ.DOWN_STOP, PTZ.UP_STOP]:
indexes.append(await self.send_command(JsonCommands.CMD_PTZ_CONTROL, parms=0, value=value.value))
@@ -544,7 +744,7 @@ async def step_rotate(self, value):
await self.rotate_stop()
async def reboot(self, **kwargs):
- logger.info('%s: reboot', self.dev.dev_id)
+ self.log.info('reboot')
await self.control(reboot=1, no_ack=True)
async def reset(self, **kwargs):
@@ -563,14 +763,50 @@ class BinarySession(Session):
BinaryCommands.CMD_PEER_LIVEVIDEO_START: BinaryCommands.ACK_PEER_LIVEVIDEO_START,
BinaryCommands.CMD_PEER_LIVEVIDEO_STOP: BinaryCommands.ACK_PEER_LIVEVIDEO_STOP,
BinaryCommands.CMD_SYSTEM_STATUS_GET: BinaryCommands.ACK_SYSTEM_STATUS_GET,
+ BinaryCommands.CMD_PEER_IRCUT_ONOFF: BinaryCommands.ACK_PEER_IRCUT_ONOFF,
+ BinaryCommands.CMD_PEER_LIGHTFILL_ONOFF: BinaryCommands.ACK_PEER_LIGHTFILL_ONOFF,
+ BinaryCommands.CMD_SYSTEM_REBOOT: BinaryCommands.ACK_SYSTEM_REBOOT,
+ BinaryCommands.CMD_SNAPSHOT_GET: BinaryCommands.ACK_SNAPSHOT_GET,
+ BinaryCommands.CMD_PEER_VIDEOPARAM_GET: BinaryCommands.ACK_PEER_VIDEOPARAM_GET,
+ BinaryCommands.CMD_SYSTEM_INF_GET: BinaryCommands.ACK_SYSTEM_INF_GET,
+ BinaryCommands.CMD_SYSTEM_ALIAS_SET: BinaryCommands.ACK_SYSTEM_ALIAS_SET,
+ BinaryCommands.CMD_SYSTEM_DATETIME_GET: BinaryCommands.ACK_SYSTEM_DATETIME_GET,
+ BinaryCommands.CMD_SYSTEM_DATETIME_SET: BinaryCommands.ACK_SYSTEM_DATETIME_SET,
+ BinaryCommands.CMD_SYSTEM_USER_GET: BinaryCommands.ACK_SYSTEM_USER_GET,
+ BinaryCommands.CMD_NET_WIFISETTING_GET: BinaryCommands.ACK_NET_WIFISETTING_GET,
+ BinaryCommands.CMD_NET_WIFISETTING_SET: BinaryCommands.ACK_NET_WIFISETTING_SET,
+ BinaryCommands.CMD_NET_WIFI_SCAN: BinaryCommands.ACK_NET_WIFI_SCAN,
+ BinaryCommands.CMD_NET_WIREDSETTING_GET: BinaryCommands.ACK_NET_WIREDSETTING_GET,
+ BinaryCommands.CMD_SD_INFO_GET: BinaryCommands.ACK_SD_INFO_GET,
+ BinaryCommands.CMD_SD_RECORDFILE_GET: BinaryCommands.ACK_SD_RECORDFILE_GET,
+ BinaryCommands.CMD_SD_PICFILE_GET: BinaryCommands.ACK_SD_PICFILE_GET,
+ BinaryCommands.CMD_SD_RECORDING_NOW: BinaryCommands.ACK_SD_RECORDING_NOW,
+ BinaryCommands.CMD_PEER_PLAYBACK_START: BinaryCommands.ACK_PEER_PLAYBACK_START,
+ BinaryCommands.CMD_PEER_PLAYBACK_STOP: BinaryCommands.ACK_PEER_PLAYBACK_STOP,
+ BinaryCommands.CMD_PEER_PLAYBACK_SEEK: BinaryCommands.ACK_PEER_PLAYBACK_SEEK,
+ BinaryCommands.CMD_PEER_PLAYBACK_SPEED: BinaryCommands.ACK_PEER_PLAYBACK_SPEED,
+ BinaryCommands.CMD_PEER_PLAYBACK_PAUSE: BinaryCommands.ACK_PEER_PLAYBACK_PAUSE,
+ BinaryCommands.CMD_PEER_PLAYBACK_RESUME: BinaryCommands.ACK_PEER_PLAYBACK_RESUME,
+ BinaryCommands.CMD_PEER_LIVEAUDIO_START: BinaryCommands.ACK_PEER_LIVEAUDIO_START,
+ BinaryCommands.CMD_PEER_LIVEAUDIO_STOP: BinaryCommands.ACK_PEER_LIVEAUDIO_STOP,
+ BinaryCommands.CMD_PEER_AUDIOPARAM_GET: BinaryCommands.ACK_PEER_AUDIOPARAM_GET,
}
REV_ACKS = {v: k for k, v in ACKS.items()}
- def __init__(self, *args, login='', password='', **kwargs):
+ def __init__(self, *args, login='', password='', audio_codec='alaw', **kwargs):
super().__init__(*args, **kwargs)
self.auth_login = login or self.DEFAULT_LOGIN
self.auth_password = password or self.DEFAULT_PASSWORD
self.ticket = b'\x00' * 4
+ # Last result code seen per command; a refusal has no payload, so this
+ # is what tells "refused" apart from "answered with nothing".
+ self.cmd_results = {}
+ self._reassert_task = None
+ # Received-audio pipeline (G.711 -> PCM), talk-back state.
+ self.audio_buffer = SharedFrameBuffer()
+ self.audio_codec = audio_codec if audio_codec in CODECS else 'alaw'
+ self.is_audio_requested = False
+ self._outgoing_audio_idx = 0
async def send_initial_packets(self):
pkt = make_punch_pkt(self.dev.dev_id)
@@ -578,44 +814,30 @@ async def send_initial_packets(self):
pkt.type = PacketType.P2pRdy
await self.send(pkt)
- async def handle_drw(self, drw_pkt):
- await super().handle_drw(drw_pkt)
- self.last_drw_pkt_at = datetime.datetime.now()
-
- # # 0x10000 - max number of chunks in one epoch,we need to keep order of chunks
- pkt_epoch = self._get_drw_epoch(drw_pkt)
-
- if pkt_epoch > self.video_epoch:
- logger.info('Video epoch changed %s -> %s', self.video_epoch, pkt_epoch)
- self.video_epoch = pkt_epoch
- self.last_drw_pkt_idx = drw_pkt._cmd_idx
- elif self.last_drw_pkt_idx < drw_pkt._cmd_idx:
- self.last_drw_pkt_idx = drw_pkt._cmd_idx
-
- if drw_pkt._channel == Channel.Video:
- # logger.debug(f'Got video data {drw_pkt.get_drw_payload()}')
- if self.video_stale_at:
- logger.warning('Got video data while stale')
- self.video_stale_at = None
- self.video_chunk_queue.put_nowait((pkt_epoch, drw_pkt))
- elif drw_pkt._channel == Channel.Audio:
- pass
- elif drw_pkt._channel == Channel.Command:
- await self.handle_incoming_command_packet(drw_pkt)
-
- def _get_drw_epoch(self, drw_pkt):
- if self.last_drw_pkt_idx > 0xff00 and drw_pkt._cmd_idx < 0x100:
- return self.video_epoch + 1
- if self.video_epoch and self.last_drw_pkt_idx < 0x100 and drw_pkt._cmd_idx > 0xff00:
- return self.video_epoch - 1
- return self.video_epoch
+ @staticmethod
+ def _result_code(drw_pkt):
+ """Decode a reply's 4-byte token as a LibError. A refused command
+ answers with no payload and a negative code here. Returns None when the
+ token isn't a known code -- on success it carries the login ticket."""
+ code = struct.unpack(' 0:
# this is from cam-reverse code
self.ticket = drw_pkt.cmd_payload[4:8]
- logger.debug(
+ result = self._result_code(drw_pkt)
+ if drw_pkt.command in self.REV_ACKS:
+ self.cmd_results[self.REV_ACKS[drw_pkt.command].value] = result
+ if result is not None and result < LibError.OK:
+ self.log.warning(
+ '%s refused by camera: %s (%d)', drw_pkt.command.name, result.name, result.value,
+ )
+ self.log.debug(
'handle_incoming_command_packet: token=%s, ticket=%s, %s data=%s (%s)',
drw_pkt.token.hex(),
self.ticket.hex(),
@@ -625,14 +847,20 @@ async def handle_incoming_command_packet(self, drw_pkt):
)
if drw_pkt.command in self.REV_ACKS:
- waiter = self.cmd_waiters.pop(self.REV_ACKS[drw_pkt.command].value, None)
- # logger.info(f'{drw_pkt.command=} {self.REV_ACKS[drw_pkt.command]=} {waiter=} {drw_pkt.cmd_payload=}')
- if waiter:
- waiter.set_result(drw_pkt.cmd_payload)
+ # Resolve but keep the waiter; wait_cmd_result pops it after
+ # awaiting. Popping here races an ACK that arrives during the
+ # preceding wait_ack, which would drop the result (e.g. a camera
+ # that answers a snapshot/status request instantly).
+ key = self.REV_ACKS[drw_pkt.command].value
+ fut = self.cmd_waiters.get(key)
+ if fut is not None and not fut.done():
+ fut.set_result(drw_pkt.cmd_payload)
async def send_command(self, cmd, cmd_payload=b'', *, with_response=False, **kwargs):
pkt_idx = self.outgoing_command_idx
- self.outgoing_command_idx += 1
+ # The index is sent as a 16-bit field and ACKs only echo 16 bits, so it
+ # must wrap; otherwise sends raise struct.error and ACK matching breaks.
+ self.outgoing_command_idx = (self.outgoing_command_idx + 1) & 0xFFFF
pkt = BinaryCmdPkt(
pkt_idx,
cmd,
@@ -640,15 +868,18 @@ async def send_command(self, cmd, cmd_payload=b'', *, with_response=False, **kwa
self.ticket,
)
if with_response:
- self.cmd_waiters[cmd.value] = asyncio.Future()
+ self._reset_cmd_waiter(cmd)
await self.send(pkt)
return pkt_idx
async def wait_cmd_result(self, cmd, timeout=5):
fut = self.cmd_waiters.get(cmd.value)
if fut:
- res = await asyncio.wait_for(fut, timeout=timeout)
- logger.debug('Got command result %s', res)
+ try:
+ res = await asyncio.wait_for(fut, timeout=timeout)
+ finally:
+ self.cmd_waiters.pop(cmd.value, None)
+ self.log.debug('Got command result %s', res)
return res
return b''
@@ -684,7 +915,7 @@ def _get_video_params(mode):
return [BinarySession._build_video_param(*x) for x in pairs[mode]]
async def _request_video(self, mode):
- logger.info('Request video %s', mode)
+ self.log.info('Request video %s', mode)
if mode == 1:
video_params = self._get_video_params(3)
@@ -695,26 +926,135 @@ async def _request_video(self, mode):
if mode:
for video_param in video_params:
- await self.send_command(BinaryCommands.CMD_PEER_VIDEOPARAM_SET, video_param, with_response=True)
- await self.send_command(BinaryCommands.CMD_PEER_LIVEVIDEO_START, b'', with_response=True)
+ await self.send_command(BinaryCommands.CMD_PEER_VIDEOPARAM_SET, video_param)
+ await self.send_command(BinaryCommands.CMD_PEER_LIVEVIDEO_START, b'')
+ # The camera adaptively drops the resolution a few seconds after the
+ # stream starts and ignores the resolution we set at start time.
+ # Re-asserting it mid-stream (which is what re-selecting it in the UI
+ # does) makes it stick, so schedule a delayed re-send. Keep a handle
+ # so it can't be garbage-collected mid-flight and is cancelled on stop.
+ if self._reassert_task and not self._reassert_task.done():
+ self._reassert_task.cancel()
+ self._reassert_task = asyncio.create_task(self._reassert_video_params(video_params))
else:
- await self.send_command(BinaryCommands.CMD_PEER_LIVEVIDEO_STOP, b'', with_response=True)
+ await self.send_command(BinaryCommands.CMD_PEER_LIVEVIDEO_STOP, b'')
+
+ async def _reassert_video_params(self, video_params, delay=5):
+ """Re-send the resolution a few seconds in to lock it (camera ignores
+ the value set at stream start and self-downgrades otherwise)."""
+ try:
+ await asyncio.sleep(delay)
+ if not self.is_video_requested or self.transport is None:
+ return
+ self.log.info('re-asserting video params to lock resolution')
+ for video_param in video_params:
+ await self.send_command(BinaryCommands.CMD_PEER_VIDEOPARAM_SET, video_param)
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ self.log.debug('Re-assert video params failed', exc_info=True)
@staticmethod
def _build_video_param(param_type, value):
if isinstance(param_type, VideoParamType):
- param = param_type
+ param = param_type.value
+ name = param_type.name.replace('VIDEO_PARAM_TYPE_', '')
else:
- param = VideoParamType[f'VIDEO_PARAM_TYPE_{param_type.upper()}'].value
+ name = str(param_type).upper()
+ param = VideoParamType[f'VIDEO_PARAM_TYPE_{name}'].value
- if isinstance(value, str):
- value = globals()[f'Video{param_type.capitalize()}'][f'VIDEO_{param_type.upper()}_{value.upper()}'].value
+ if isinstance(value, Enum):
+ value = value.value
+ elif isinstance(value, str):
+ # Resolve a symbolic value (e.g. 'HD') against the matching
+ # Video enum, e.g. VideoResolution.VIDEO_RESOLUTION_HD.
+ enum_cls = globals()[f'Video{name.capitalize()}']
+ value = enum_cls[f'VIDEO_{name}_{value.upper()}'].value
return struct.pack('= 48:
+ table = struct.unpack_from('<12I', payload)
+ return table[param_id - 1] if 1 <= param_id <= 12 else None
+ if len(payload) >= 8:
+ got_param, value = struct.unpack_from('= 4:
+ return struct.unpack_from('= 80:
+ ntp_server = current[16:80].split(b'\x00', 1)[0].decode('ascii') or ntp_server
+ except Exception:
+ self.log.debug('DATETIME_GET before set failed; using default NTP server')
+ ts = int(when.timestamp())
+ payload = struct.pack(' cmdPtzSet(1, 0, i) SET = store preset i
+ # cmdPtzPreDo(i, false) -> cmdPtzSet(1, 4, i) DEL = delete preset i
+ # (recall path) -> cmdPtzSet(1, 1, i) GET = go to preset i
+ # cmdPtzPreChk() -> cmdPtzSet(1, 3, 0) CHK = query bitmask
+ # cmdPtzPreRec() -> cmdPtzSet(1, 2, 0) REC = mode toggle, index 0
+ #
+ # Note REC takes index 0 and is paired with a DIRECTION/STOP on the second
+ # press, so it toggles a recording mode rather than saving a preset -- it
+ # is NOT the "store" op, despite the name. An earlier guess that it was
+ # has been backed out.
+ #
+ # All ops were tried by hand against PTZA and FTYC (2026-08-25): none of
+ # them moves the camera or stores anything, so presets are simply not
+ # implemented on the tested hardware. The API is kept as best-effort for
+ # other firmwares. See VENDOR_APP_FINDINGS.md item 9.
+
+ async def ptz_prefab(self, op, index, **kwargs):
+ """Send a raw PREFAB op. `op` is a PtzPrefab (or its int value).
+
+ Exposed so every op can be tried by hand -- the mapping from op to
+ store/recall is a hypothesis, not a confirmed fact.
+ """
+ op = PtzPrefab(int(op))
+ self.log.info('PTZ prefab %s index %s', op.name, index)
+ data = self._pack_ptz_cmd(PtzParamType.PTZ_PARAM_TYPE_PREFAB, op, index)
+ await self.send_command(BinaryCommands.CMD_PASSTHROUGH_STRING_PUT, data)
+
+ async def ptz_goto_preset(self, index, **kwargs):
+ """Move to a stored PTZ preset position (1-based index)."""
+ await self.ptz_prefab(PtzPrefab.PTZ_PREFAB_GET, index)
+
+ async def ptz_set_preset(self, index, **kwargs):
+ """Store the current position as a PTZ preset (1-based index)."""
+ await self.ptz_prefab(PtzPrefab.PTZ_PREFAB_SET, index)
+
+ async def ptz_delete_preset(self, index, **kwargs):
+ """Delete a stored PTZ preset (1-based index)."""
+ self.log.info('delete PTZ preset %s', index)
+ data = self._pack_ptz_cmd(PtzParamType.PTZ_PARAM_TYPE_PREFAB, PtzPrefab.PTZ_PREFAB_DEL, index)
+ await self.send_command(BinaryCommands.CMD_PASSTHROUGH_STRING_PUT, data)
+
+ async def ptz_query_presets(self, timeout=5):
+ """Ask which presets are stored. The vendor app receives a
+ (type, param, value) bean back where value is a bitmask with bit
+ (n-1) set for stored preset n. Returns the raw ACK payload."""
+ data = self._pack_ptz_cmd(PtzParamType.PTZ_PARAM_TYPE_PREFAB, PtzPrefab.PTZ_PREFAB_CHK, 0)
+ return await self._request(BinaryCommands.CMD_PASSTHROUGH_STRING_PUT, data, timeout=timeout)
+
@staticmethod
- def _pack_ptz_dir_cmd(ptz: PtzDirection) -> bytes:
- data = struct.pack('>III', PtzParamType.PTZ_PARAM_TYPE_DIRECTION, ptz, 0)
+ def _pack_ptz_cmd(param_type, param, value) -> bytes:
+ # The passthrough PTZ frame is (param_type, param, value): plain moves
+ # are (DIRECTION, direction, 0), presets are (PREFAB, op, index).
+ data = struct.pack('>III', int(param_type), int(param), int(value))
return pack_passtrough_cmd(BinaryCommands.CMD_PTZ_SET.value, data)
+ def _pack_ptz_dir_cmd(self, ptz: PtzDirection, index: int = 0) -> bytes:
+ return self._pack_ptz_cmd(PtzParamType.PTZ_PARAM_TYPE_DIRECTION, int(ptz), index)
+
class SharedFrameBuffer:
def __init__(self):
@@ -798,7 +1443,14 @@ async def get(self):
def make_session(device: DeviceDescriptor, on_device_lost: Callable[[DeviceDescriptor], None],
- login: str = '', password: str = '') -> Session:
+ login: str = '', password: str = '',
+ on_video_state_change: Callable[[bool], None] = None) -> Session:
"""Create a session for the camera."""
session_class = JsonSession if device.is_json else BinarySession
- return session_class(device, on_disconnect=on_device_lost, login=login, password=password)
+ return session_class(
+ device,
+ on_disconnect=on_device_lost,
+ login=login,
+ password=password,
+ on_video_state_change=on_video_state_change,
+ )
diff --git a/aiopppp/types.py b/aiopppp/types.py
index c950b02..15776d5 100644
--- a/aiopppp/types.py
+++ b/aiopppp/types.py
@@ -51,3 +51,10 @@ class VideoFrame:
def __init__(self, idx, data):
self.idx = idx
self.data = data
+
+
+class AudioFrame:
+ def __init__(self, idx, data, sample_rate=8000):
+ self.idx = idx
+ self.data = data # signed 16-bit little-endian PCM
+ self.sample_rate = sample_rate
diff --git a/binary_camera.py b/binary_camera.py
index edc95c7..209d548 100644
--- a/binary_camera.py
+++ b/binary_camera.py
@@ -2,6 +2,8 @@
import struct
import aiopppp.const
+from aiopppp.const import BinaryCommands, LibError
+from aiopppp.session import BinarySession
from aiopppp.packets import (
make_punch_pkt,
make_p2palive_pkt,
@@ -11,7 +13,24 @@
xq_bytes_decode,
DrwPkt,
)
-from aiopppp.types import DeviceID
+from aiopppp.types import Channel, DeviceID
+
+VIDEO_MARKER = b'\x55\xaa\x15\xa8'
+
+# A minimal, structurally-valid JPEG (SOI ... EOI). Content is irrelevant to the
+# protocol path; it just needs the FFD8..FFD9 envelope so consumers see a frame.
+_MINI_JPEG = bytes.fromhex(
+ 'ffd8ffe000104a46494600010100000100010000'
+ 'ffdb004300080606070605080707070909080a0c140d0c0b0b0c1912130f141d1a1f1e1d1a1c1c20242e2720222c231c1c28372'
+ '92c30313434341f27393d38323c2e333432'
+ 'ffc0000b080010001001011100'
+ 'ffc4001f0000010501010101010100000000000000000102030405060708090a0b'
+ 'ffc400b5100002010303020403050504040000017d01020300041105122131410613516107227114328191a1082342b1c11552'
+ 'd1f02433627282090a161718191a25262728292a3435363738393a434445464748494a535455565758595a636465666768696a'
+ '737475767778797a838485868788898a92939495969798999aa2a3a4a5a6a7a8a9aab2b3b4b5b6b7b8b9bac2c3c4c5c6c7c8c9c'
+ 'ad2d3d4d5d6d7d8d9dae1e2e3e4e5e6e7e8e9eaf1f2f3f4f5f6f7f8f9fa'
+ 'ffda0008010100003f00fbd0ffd9'
+)
class UDPProtocol(asyncio.DatagramProtocol):
@@ -39,14 +58,36 @@ async def create_udp_server(port, on_receive):
class BinaryCamera:
- def __init__(self):
+ # The only commands a real camera refuses without a login, and the code each
+ # answers with. Confirmed on PTZA fw 2.2.15.93: everything else -- video,
+ # audio, PTZ, lights, the status/info/datetime/wifi-settings reads, video
+ # params (get and set) and time sync -- works unauthenticated.
+ PRIVILEGED = {
+ BinaryCommands.CMD_SYSTEM_REBOOT: LibError.USER_NO_PRIVILEGE,
+ BinaryCommands.CMD_NET_WIFI_SCAN: LibError.CMD_EXCUTE_FAILED,
+ BinaryCommands.CMD_SYSTEM_USER_GET: LibError.CMD_EXCUTE_FAILED,
+ }
+
+ def __init__(self, port=32108, dev_id=None, auth_mode='normal'):
self.transport = None
- self.dev_id = DeviceID('TEST',123456, 'CAMERA')
+ self.port = port
+ self.dev_id = dev_id or DeviceID('TEST',123456, 'CAMERA')
self.input = asyncio.Queue()
self.output = asyncio.Queue()
self.client_addr = None
- self.ticket = b'abcd'
+ # Synthetic session ticket, handed out in the USER_CHK reply payload.
+ self.ticket = b'\xde\xad\xbe\xef'
+ # How much an unauthenticated session is allowed:
+ # 'normal' - refuse only PRIVILEGED, as real hardware does
+ # 'stuck' - refuse everything with UNAUTH, video included; a camera
+ # that has wedged itself, cleared by a power cycle
+ # 'off' - no auth checks at all
+ self.auth_mode = auth_mode
+ self.logged_in = False
self.cmd_idx = 1
+ self.video_task = None
+ self.video_idx = 1
+ self.frame_period = 0.2 # ~5 fps of synthetic frames
def on_receive(self, data, addr):
# print(f"Received {data} from {addr}")
@@ -72,6 +113,36 @@ async def send_p2p_rdy_set(self):
self.output.put_nowait((bytes(pkt), self.client_addr))
await asyncio.sleep(0.1)
+ _STATUS_BLOB = bytes.fromhex(
+ "0d 02 01 3d 74 0f 00 00 00 00 00 00 ff ff ff ff bf ff ff ff "
+ "01 01 00 30 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 "
+ "00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 "
+ "00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 "
+ "00 00 00 00 00 00 00 00 00 01 00 00 02 00 00 00 00 00 00 00 "
+ "00 00 00 00 00 ff ff ff 00 00 00 00 ff ff ff ff 00 00 00 00 "
+ "00 00 00 00".replace(' ', '')
+ )
+
+ def _send_cmd_ack(self, ack_command, cmd_payload=b'', token=None):
+ # The token field of a reply is the result code, so a success carries
+ # LibError.OK there and puts its data in the payload.
+ self.output.put_nowait((
+ bytes(BinaryCmdPkt(
+ cmd_idx=self.cmd_idx,
+ command=ack_command,
+ token=struct.pack(' 4:
data = xq_bytes_decode(data, 4)
- if cmd_id == aiopppp.const.BinaryCommands.CMD_SYSTEM_USER_CHK:
- INCORRECT_USER_RESP = '11 0a 20 11 0c 00 ff 00 00 00 00 00 57 56 6c 37 fe 01 01 01'
- CORRECT_USER_RESP = '11 0a 20 11 04 00 ff 00 0e fc ff ff'
-
+ if cmd_id == BinaryCommands.CMD_SYSTEM_USER_CHK:
username, password = struct.unpack('<32s128s', data)
username = username.decode('utf-8').strip('\x00')
password = password.decode('utf-8').strip('\x00')
-
- resp = CORRECT_USER_RESP if username == 'admin' and password == 'admin' else INCORRECT_USER_RESP
-
- print('... BinaryCommand: cmd_id:', cmd_id, 'data:', data)
- await asyncio.sleep(0.3)
- print('... send ACK_SYSTEM_USER_CHK')
- self.output.put_nowait((bytes(
- DrwPkt(cmd_idx=0, channel=0, drw_payload=bytes.fromhex(resp)),
- ), self.client_addr))
- # self.output.put_nowait((bytes(BinaryCmdPkt(
- # cmd_idx=self.cmd_idx,
- # command=aiopppp.const.BinaryCommands.ACK_SYSTEM_USER_CHK,
- # ticket=self.ticket,
- # cmd_payload=b'\xff\x00\x00\x00',
- # )), self.client_addr))
- self.cmd_idx += 1
- elif cmd_id == aiopppp.const.BinaryCommands.CMD_SYSTEM_STATUS_GET:
- await asyncio.sleep(0.3)
- print('... send ACK_SYSTEM_STATUS_GET')
- self.output.put_nowait(
- (
- bytes(
- BinaryCmdPkt(
- cmd_idx=self.cmd_idx,
- command=aiopppp.const.BinaryCommands.ACK_SYSTEM_STATUS_GET,
- # token=b'\x0a\xfc\xff\xff',
- token=b"\x00\x00\x00\x00",
- # cmd_payload=bytes(range(0x15)),
- cmd_payload=bytes.fromhex(
- "0d 02 01 3d 74 0f 00 00 00 00 00 00 ff ff ff ff bf ff ff ff "
- "01 01 00 30 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 "
- "00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 "
- "00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 "
- "00 00 00 00 00 00 00 00 00 01 00 00 02 00 00 00 00 00 00 00 "
- "00 00 00 00 00 ff ff ff 00 00 00 00 ff ff ff ff 00 00 00 00 "
- "00 00 00 00",
- ),
- )
- ),
- self.client_addr,
+ print('... USER_CHK:', username, password)
+ await asyncio.sleep(0.1)
+ if username == 'admin' and password == 'admin':
+ # cmd_payload[4:8] is the session ticket the client will echo.
+ self.logged_in = True
+ self._send_cmd_ack(
+ BinaryCommands.ACK_SYSTEM_USER_CHK,
+ b'\x00\x00\x00\x00' + self.ticket,
+ )
+ else:
+ # PTZA answers bad credentials with -1010, not the -1013
+ # USER_PWD_INCORRECT the vendor enum also defines.
+ self.logged_in = False
+ self._send_cmd_refusal(
+ BinaryCommands.ACK_SYSTEM_USER_CHK, LibError.CMD_EXCUTE_FAILED,
)
+ elif (
+ self.auth_mode != 'off' and not self.logged_in
+ and (self.auth_mode == 'stuck' or cmd_id in self.PRIVILEGED)
+ ):
+ code = LibError.UNAUTH if self.auth_mode == 'stuck' else self.PRIVILEGED[cmd_id]
+ ack = BinarySession.ACKS.get(cmd_id)
+ if ack is None:
+ print('... unauthenticated, unhandled command:', cmd_id)
+ else:
+ self._send_cmd_refusal(ack, code)
+ elif cmd_id == BinaryCommands.CMD_SYSTEM_STATUS_GET:
+ await asyncio.sleep(0.1)
+ self._send_cmd_ack(BinaryCommands.ACK_SYSTEM_STATUS_GET, self._STATUS_BLOB)
+ elif cmd_id == BinaryCommands.CMD_PEER_VIDEOPARAM_SET:
+ self._send_cmd_ack(BinaryCommands.ACK_PEER_VIDEOPARAM_SET)
+ elif cmd_id == BinaryCommands.CMD_PEER_VIDEOPARAM_GET:
+ # Real cameras (PTZA) ignore the requested id and answer with
+ # the full table of params 1..12: resolution=HD, ircut=1.
+ table = [0] * 12
+ table[0] = 2 # resolution -> HD
+ table[8] = 1 # ircut on
+ self._send_cmd_ack(BinaryCommands.ACK_PEER_VIDEOPARAM_GET, struct.pack('<12I', *table))
+ elif cmd_id == BinaryCommands.CMD_SYSTEM_DATETIME_GET:
+ # PTZA layout: u32 UTC epoch, i32 tz seconds west, pad, ntp[64]
+ self._send_cmd_ack(
+ BinaryCommands.ACK_SYSTEM_DATETIME_GET,
+ struct.pack('', data.hex(' '))
+ self._send_cmd_ack(BinaryCommands.ACK_SYSTEM_DATETIME_SET)
+ elif cmd_id == BinaryCommands.CMD_NET_WIFISETTING_GET:
+ # PTZA layout: mode, pad12, security, pad4, ssid[32],
+ # password[128], five char[16] dotted-quad strings
+ wifi = struct.pack(
+ '', data.hex(' '))
+ self._send_cmd_ack(BinaryCommands.ACK_PEER_IRCUT_ONOFF)
+ elif cmd_id == BinaryCommands.CMD_PEER_LIGHTFILL_ONOFF:
+ print('... LIGHTFILL ->', data.hex(' '))
+ self._send_cmd_ack(BinaryCommands.ACK_PEER_LIGHTFILL_ONOFF)
+ elif cmd_id == BinaryCommands.CMD_SNAPSHOT_GET:
+ self._send_cmd_ack(BinaryCommands.ACK_SNAPSHOT_GET, _MINI_JPEG)
+ elif cmd_id == BinaryCommands.CMD_SYSTEM_REBOOT:
+ print('... REBOOT requested')
+ self._send_cmd_ack(BinaryCommands.ACK_SYSTEM_REBOOT)
+ elif cmd_id == BinaryCommands.CMD_PASSTHROUGH_STRING_PUT:
+ print('... PTZ/passthrough ->', data.hex(' '))
+ self._send_cmd_ack(BinaryCommands.ACK_PASSTHROUGH_STRING_PUT)
+ else:
+ print('... unhandled command:', cmd_id)
+
+ def _start_video(self):
+ if self.video_task is None or self.video_task.done():
+ print('... start video stream')
+ self.video_task = asyncio.create_task(self._stream_video())
+
+ def _stop_video(self):
+ if self.video_task and not self.video_task.done():
+ print('... stop video stream')
+ self.video_task.cancel()
+ self.video_task = None
+
+ def _next_video_idx(self):
+ idx = self.video_idx
+ self.video_idx = (self.video_idx + 1) & 0xFFFF
+ return idx
+
+ def _send_video_chunk(self, chunk):
+ pkt = DrwPkt(channel=Channel.Video.value, cmd_idx=self._next_video_idx(), drw_payload=chunk)
+ self.output.put_nowait((bytes(pkt), self.client_addr))
+
+ async def _stream_video(self):
+ try:
+ while True:
+ # First chunk carries the 0x20-byte frame header (marker + pad);
+ # the client strips it and treats this index as a frame boundary.
+ header = VIDEO_MARKER + b'\x00' * (0x20 - len(VIDEO_MARKER))
+ body = _MINI_JPEG
+ # Split into ~1024-byte payloads across several DRW chunks.
+ step = 1024
+ parts = [body[i:i + step] for i in range(0, len(body), step)] or [b'']
+ self._send_video_chunk(header + parts[0])
+ for part in parts[1:]:
+ self._send_video_chunk(part)
+ await asyncio.sleep(self.frame_period)
+ except asyncio.CancelledError:
+ raise
async def on_packet(self, data, addr):
@@ -157,14 +314,21 @@ async def receive_task(self):
self.input.task_done()
async def run(self):
- self.transport = await create_udp_server(32108, self.on_receive)
+ self.transport = await create_udp_server(self.port, self.on_receive)
out_t = asyncio.create_task(self.send_task())
in_t = asyncio.create_task(self.receive_task())
await asyncio.gather(*[out_t, in_t])
async def main():
- camera = BinaryCamera()
+ import sys
+ port = int(sys.argv[1]) if len(sys.argv) > 1 else 32108
+ # binary_camera.py [port] [normal|stuck|off]
+ auth_mode = sys.argv[2] if len(sys.argv) > 2 else 'normal'
+ print(f'Listening on {port}, auth mode: {auth_mode}')
+ camera = BinaryCamera(port=port, auth_mode=auth_mode)
await camera.run()
-asyncio.run(main())
+
+if __name__ == '__main__':
+ asyncio.run(main())
diff --git a/proxy_camera.py b/proxy_camera.py
new file mode 100644
index 0000000..26f0344
--- /dev/null
+++ b/proxy_camera.py
@@ -0,0 +1,226 @@
+"""Transparent PPPP proxy ("man-in-the-middle") camera for the binary protocol.
+
+Advertises a configurable DID to the app and forwards every packet to a real
+camera (and its replies back), rewriting only the DID so the app talks to the
+proxy while the proxy talks to the real device. Each forwarded control packet is
+logged with its raw bytes and, when recognised, its decoded form. Video and
+audio stream packets are relayed but never logged.
+
+Typical use (app configured with the proxy DID):
+
+ python proxy_camera.py --did PROX-000001-CAMERA --target-ip 192.168.1.50
+
+The app then discovers/connects to this host using PROX-000001-CAMERA, and all
+traffic is relayed to the camera at 192.168.1.50 (whose real DID is learned from
+its discovery reply, or given with --target-did).
+
+Only the binary protocol is supported: those cameras use no transport
+encryption, so the DID can be rewritten directly on the wire. JSON (XOR1)
+cameras would need decrypt/re-encrypt and are out of scope.
+"""
+
+import argparse
+import asyncio
+import logging
+import struct
+
+from aiopppp.const import CAM_MAGIC, PacketType
+from aiopppp.packets import PunchPkt, parse_packet
+from aiopppp.types import Channel, DeviceID
+
+logger = logging.getLogger('proxy_camera')
+
+# Packet types whose payload carries the 20-byte packed DID.
+_DID_TYPES = {
+ PacketType.PunchPkt.value,
+ PacketType.P2pRdy.value,
+ PacketType.PunchTo.value,
+}
+_STREAM_CHANNELS = {Channel.Video.value, Channel.Audio.value}
+_KEEPALIVE_TYPES = {PacketType.P2PAlive.value, PacketType.P2PAliveAck.value}
+
+
+def parse_did(text):
+ """Parse a 'PREFIX-SERIAL-SUFFIX' DID string into a DeviceID."""
+ parts = text.split('-')
+ if len(parts) < 3:
+ raise ValueError(f'Invalid DID {text!r}, expected PREFIX-SERIAL-SUFFIX')
+ prefix, serial, suffix = parts[0], parts[1], parts[2]
+ return DeviceID(prefix=prefix, serial=serial, suffix=suffix)
+
+
+def pack_did(dev_id):
+ """Return the 20-byte on-wire form of a DID (as carried in PunchPkt)."""
+ return struct.pack(
+ '>4sQ8s',
+ dev_id.prefix.encode('ascii'),
+ int(dev_id.serial),
+ dev_id.suffix.encode('ascii'),
+ )
+
+
+class _EndpointProtocol(asyncio.DatagramProtocol):
+ def __init__(self, on_receive):
+ self._on_receive = on_receive
+
+ def datagram_received(self, data, addr):
+ self._on_receive(data, addr)
+
+
+class ProxyCamera:
+ """Relay between an app and a real binary-protocol camera, rewriting the DID."""
+
+ def __init__(self, proxy_did, target_ip, target_port=32108,
+ target_did=None, listen_host='0.0.0.0', listen_port=32108,
+ log_keepalive=False):
+ self.proxy_did = proxy_did
+ self.proxy_packed = pack_did(proxy_did)
+ self.target_ip = target_ip
+ self.camera_addr = (target_ip, target_port)
+ self.real_did = target_did
+ self.real_packed = pack_did(target_did) if target_did else None
+ self.listen_host = listen_host
+ self.listen_port = listen_port
+ self.log_keepalive = log_keepalive
+
+ self.app_transport = None
+ self.camera_transport = None
+ # The app's current source address (differs between discovery and the
+ # session); replies are sent to the most recent one.
+ self.app_addr = None
+
+ async def run(self):
+ loop = asyncio.get_running_loop()
+ # App-facing socket: the app discovers/connects here.
+ self.app_transport, _ = await loop.create_datagram_endpoint(
+ lambda: _EndpointProtocol(self._on_app_packet),
+ local_addr=(self.listen_host, self.listen_port),
+ allow_broadcast=True,
+ )
+ # Camera-facing socket: we talk to the real camera from here.
+ self.camera_transport, _ = await loop.create_datagram_endpoint(
+ lambda: _EndpointProtocol(self._on_camera_packet),
+ remote_addr=self.camera_addr,
+ )
+ logger.info('Proxy DID %s -> camera %s:%d (real DID %s)',
+ self.effective_proxy_did().dev_id, self.camera_addr[0], self.camera_addr[1],
+ self.real_did.dev_id if self.real_did else '')
+ # The serial travels as a uint64, so leading zeros are dropped on the
+ # wire. Tell the user the exact DID to configure the app with.
+ effective = self.effective_proxy_did().dev_id
+ if effective != self.proxy_did.dev_id:
+ logger.info('NOTE: configure the app with DID %s (serial leading zeros are dropped)',
+ effective)
+ else:
+ logger.info('Configure the app with DID %s', effective)
+ logger.info('Listening for the app on %s:%d', self.listen_host, self.listen_port)
+ # Run until cancelled.
+ await asyncio.Event().wait()
+
+ # -- packet handlers ----------------------------------------------------
+
+ def _on_app_packet(self, data, addr):
+ self.app_addr = addr
+ self._log('APP->CAM', data)
+ forwarded = self._rewrite(data, self.proxy_packed, self.real_packed)
+ self.camera_transport.sendto(forwarded)
+
+ def _on_camera_packet(self, data, addr):
+ # Learn the real DID from the camera's first PunchPkt so app->camera
+ # DID rewriting works even without --target-did.
+ if self.real_packed is None and len(data) >= 2 and data[1] == PacketType.PunchPkt.value:
+ self._learn_real_did(data)
+ self._log('CAM->APP', data)
+ forwarded = self._rewrite(data, self.real_packed, self.proxy_packed)
+ if self.app_addr is not None:
+ self.app_transport.sendto(forwarded, self.app_addr)
+
+ def _learn_real_did(self, data):
+ try:
+ self.real_did = PunchPkt(PacketType.PunchPkt, data[4:]).as_object()
+ self.real_packed = pack_did(self.real_did)
+ logger.info('Learned real camera DID: %s', self.real_did.dev_id)
+ except Exception:
+ logger.debug('Could not parse camera DID from PunchPkt', exc_info=True)
+
+ # -- helpers ------------------------------------------------------------
+
+ def effective_proxy_did(self):
+ """The proxy DID as it appears on the wire (serial normalized to uint64)."""
+ return PunchPkt(PacketType.PunchPkt, self.proxy_packed).as_object()
+
+ @staticmethod
+ def _rewrite(data, old_packed, new_packed):
+ """Return data with the DID rewritten, only in DID-bearing packets."""
+ if not old_packed or not new_packed or old_packed == new_packed:
+ return data
+ if len(data) >= 2 and data[1] in _DID_TYPES and old_packed in data:
+ return data.replace(old_packed, new_packed)
+ return data
+
+ @staticmethod
+ def _is_stream(data):
+ """True for video/audio DRW (and their ACKs), which must not be logged."""
+ if len(data) < 6 or data[0] != CAM_MAGIC:
+ return False
+ if data[1] in (PacketType.Drw.value, PacketType.DrwAck.value):
+ return data[5] in _STREAM_CHANNELS
+ return False
+
+ def _log(self, direction, data):
+ if self._is_stream(data):
+ return
+ if not self.log_keepalive and len(data) >= 2 and data[1] in _KEEPALIVE_TYPES:
+ return
+ try:
+ decoded = str(parse_packet(data))
+ except Exception:
+ typ = f'0x{data[1]:02x}' if len(data) >= 2 else '??'
+ decoded = f''
+ logger.info('%s | %s', direction, decoded)
+ logger.info('%s | raw: %s', direction, data.hex(' '))
+
+
+def _build_arg_parser():
+ p = argparse.ArgumentParser(description=__doc__,
+ formatter_class=argparse.RawDescriptionHelpFormatter)
+ p.add_argument('--did', required=True,
+ help='DID to advertise to the app (PREFIX-SERIAL-SUFFIX)')
+ p.add_argument('--target-ip', required=True,
+ help='IP (or broadcast address) of the real camera')
+ p.add_argument('--target-port', type=int, default=32108,
+ help='UDP port of the real camera (default 32108)')
+ p.add_argument('--target-did', default=None,
+ help="Real camera DID; if omitted it is learned from the camera's "
+ 'discovery reply')
+ p.add_argument('--listen-host', default='0.0.0.0',
+ help='Local address to listen on for the app (default 0.0.0.0)')
+ p.add_argument('--listen-port', type=int, default=32108,
+ help='Local UDP port to listen on for the app (default 32108)')
+ p.add_argument('--log-keepalive', action='store_true',
+ help='Also log P2PAlive/P2PAliveAck keepalives (noisy)')
+ p.add_argument('--log-level', default='INFO')
+ return p
+
+
+async def main(argv=None):
+ args = _build_arg_parser().parse_args(argv)
+ logging.basicConfig(level=getattr(logging, args.log_level.upper(), logging.INFO),
+ format='%(message)s')
+ proxy = ProxyCamera(
+ proxy_did=parse_did(args.did),
+ target_ip=args.target_ip,
+ target_port=args.target_port,
+ target_did=parse_did(args.target_did) if args.target_did else None,
+ listen_host=args.listen_host,
+ listen_port=args.listen_port,
+ log_keepalive=args.log_keepalive,
+ )
+ await proxy.run()
+
+
+if __name__ == '__main__':
+ try:
+ asyncio.run(main())
+ except KeyboardInterrupt:
+ pass