From 2e12ebcf9a98e5d2c5ae74b10ae0b45d786d396d Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:48:29 +0100 Subject: [PATCH 1/8] Add HKSV 17.99 camera services and characteristics to the catalog Add the eight services and 27 characteristics of Apple's HomeKit Secure Video Open Source Compatibility Guide (Developer Preview 1.0) to the loader resources. Two of them, Contributing Sensors and Motion Enabled, also become optional characteristics of the Motion Sensor service. Timed write is advertised where the guide requires it. Writes to Motion Enabled and Streaming Enabled are marked admin only with an "AdminOnlyAccess": ["write"] catalog property. It is not part of the HAP representation and not enforced yet. Camera Recording Management keeps the characteristics classic recording relies on. --- pyhap/resources/characteristics.json | 235 +++++++++++++++++++++++++++ pyhap/resources/services.json | 94 ++++++++++- tests/test_loader.py | 166 ++++++++++++++++++- 3 files changed, 493 insertions(+), 2 deletions(-) diff --git a/pyhap/resources/characteristics.json b/pyhap/resources/characteristics.json index 596e5898..f185818d 100644 --- a/pyhap/resources/characteristics.json +++ b/pyhap/resources/characteristics.json @@ -133,6 +133,88 @@ "minValue": 0, "unit": "percentage" }, + "BufferActivityCommand": { + "Format": "tlv8", + "Permissions": [ + "pw" + ], + "UUID": "00008017-0000-1000-8000-0026BB765291" + }, + "BufferEventCommand": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008014-0000-1000-8000-0026BB765291" + }, + "BufferEventSequenceNumber": { + "Format": "uint32", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008015-0000-1000-8000-0026BB765291" + }, + "BufferUploadCommand": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008013-0000-1000-8000-0026BB765291" + }, + "CameraCapabilities": { + "Format": "tlv8", + "Permissions": [ + "pr" + ], + "UUID": "00008011-0000-1000-8000-0026BB765291" + }, + "CameraClientCSR": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008081-0000-1000-8000-0026BB765291" + }, + "CameraClientCertificate": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "tw" + ], + "UUID": "00008082-0000-1000-8000-0026BB765291" + }, + "CameraClientCertificateStatus": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008083-0000-1000-8000-0026BB765291" + }, + "CameraKey": { + "Format": "tlv8", + "Permissions": [ + "pw", + "tw" + ], + "UUID": "00008051-0000-1000-8000-0026BB765291" + }, + "CameraKeyID": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008052-0000-1000-8000-0026BB765291" + }, "CameraOperatingModeIndicator": { "Format": "bool", "Permissions": [ @@ -142,6 +224,22 @@ ], "UUID": "0000021D-0000-1000-8000-0026BB765291" }, + "CameraRecordingPublishingPoint": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw" + ], + "UUID": "00008016-0000-1000-8000-0026BB765291" + }, + "CameraZones": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw" + ], + "UUID": "00008022-0000-1000-8000-0026BB765291" + }, "CarbonDioxideDetected": { "Format": "uint8", "Permissions": [ @@ -265,6 +363,14 @@ "ContactNotDetected": 1 } }, + "ContributingSensors": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008086-0000-1000-8000-0026BB765291" + }, "CoolingThresholdTemperature": { "Format": "float", "Permissions": [ @@ -825,6 +931,19 @@ ], "UUID": "00000022-0000-1000-8000-0026BB765291" }, + "MotionEnabled": { + "AdminOnlyAccess": [ + "write" + ], + "Format": "bool", + "Permissions": [ + "pr", + "pw", + "ev", + "tw" + ], + "UUID": "00008087-0000-1000-8000-0026BB765291" + }, "Mute": { "Format": "bool", "Permissions": [ @@ -1055,6 +1174,15 @@ "SinglePress": 0 } }, + "RTPStreamingControl": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008045-0000-1000-8000-0026BB765291" + }, "RecordingAudioActive": { "Format": "uint8", "Permissions": [ @@ -1237,6 +1365,13 @@ ], "UUID": "00000117-0000-1000-8000-0026BB765291" }, + "SensorUUID": { + "Format": "data", + "Permissions": [ + "pr" + ], + "UUID": "0000805B-0000-1000-8000-0026BB765291" + }, "SerialNumber": { "Format": "string", "MaximumLength": 64, @@ -1386,6 +1521,19 @@ "Tampered": 1 } }, + "StreamingEnabled": { + "AdminOnlyAccess": [ + "write" + ], + "Format": "bool", + "Permissions": [ + "pr", + "pw", + "ev", + "tw" + ], + "UUID": "00008041-0000-1000-8000-0026BB765291" + }, "StreamingStatus": { "Format": "tlv8", "Permissions": [ @@ -1420,6 +1568,14 @@ ], "UUID": "00000115-0000-1000-8000-0026BB765291" }, + "SupportedAudioStreamTiers": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008044-0000-1000-8000-0026BB765291" + }, "SupportedCameraRecordingConfiguration": { "Format": "tlv8", "Permissions": [ @@ -1464,6 +1620,14 @@ ], "UUID": "00000144-0000-1000-8000-0026BB765291" }, + "SupportedVideoStreamTiers": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008043-0000-1000-8000-0026BB765291" + }, "SwingMode": { "Format": "uint8", "Permissions": [ @@ -1804,6 +1968,77 @@ "minValue": 0, "unit": "percentage" }, + "WebRTCNumberOfActiveSessions": { + "Format": "uint8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008057-0000-1000-8000-0026BB765291", + "maxValue": 255, + "minValue": 0 + }, + "WebRTCProvideAnswer": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008054-0000-1000-8000-0026BB765291" + }, + "WebRTCReoffer": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008058-0000-1000-8000-0026BB765291" + }, + "WebRTCSolicitOffer": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008053-0000-1000-8000-0026BB765291" + }, + "WebRTCStreamingControl": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008056-0000-1000-8000-0026BB765291" + }, + "WebRTCSupportedAudioStreamTiers": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "0000805A-0000-1000-8000-0026BB765291" + }, + "WebRTCSupportedVideoStreamTiers": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008059-0000-1000-8000-0026BB765291" + }, + "WebRTCUpdateSession": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "0000805C-0000-1000-8000-0026BB765291" + }, "HardwareFinish": { "Format": "tlv8", "Permissions": [ diff --git a/pyhap/resources/services.json b/pyhap/resources/services.json index 98c95aa4..8b3b4e97 100644 --- a/pyhap/resources/services.json +++ b/pyhap/resources/services.json @@ -61,6 +61,80 @@ ], "UUID": "00000096-0000-1000-8000-0026BB765291" }, + "CameraBufferManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "BufferUploadCommand", + "BufferActivityCommand", + "BufferEventCommand", + "BufferEventSequenceNumber", + "CameraRecordingPublishingPoint" + ], + "UUID": "00008000-0000-1000-8000-0026BB765291" + }, + "CameraCapabilities": { + "OptionalCharacteristics": [ + "ManuallyDisabled" + ], + "RequiredCharacteristics": [ + "Version", + "CameraCapabilities" + ], + "UUID": "00008010-0000-1000-8000-0026BB765291" + }, + "CameraClientCertificateManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "CameraClientCSR", + "CameraClientCertificate", + "CameraClientCertificateStatus" + ], + "UUID": "00008080-0000-1000-8000-0026BB765291" + }, + "CameraGlobalOperatingMode": { + "OptionalCharacteristics": [ + "ManuallyDisabled", + "NightVision", + "ThirdPartyCameraActive" + ], + "RequiredCharacteristics": [ + "HomeKitCameraActive", + "StreamingEnabled", + "CameraOperatingModeIndicator" + ], + "UUID": "00008032-0000-1000-8000-0026BB765291" + }, + "CameraKeyManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "CameraKey", + "CameraKeyID" + ], + "UUID": "00008050-0000-1000-8000-0026BB765291" + }, + "CameraMotionZones": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "Version", + "Active", + "CameraZones" + ], + "UUID": "00008021-0000-1000-8000-0026BB765291" + }, + "CameraMultiTierRTPStreamManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "StreamingEnabled", + "StatusActive", + "SupportedVideoStreamTiers", + "SupportedAudioStreamTiers", + "SupportedRTPConfiguration", + "SetupEndpoints", + "RTPStreamingControl", + "SensorUUID" + ], + "UUID": "00008031-0000-1000-8000-0026BB765291" + }, "CameraOperatingMode": { "OptionalCharacteristics": [ "CameraOperatingModeIndicator", @@ -103,6 +177,22 @@ ], "UUID": "00000204-0000-1000-8000-0026BB765291" }, + "CameraWebRTCStreamManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "WebRTCSolicitOffer", + "WebRTCProvideAnswer", + "WebRTCStreamingControl", + "WebRTCNumberOfActiveSessions", + "WebRTCReoffer", + "WebRTCUpdateSession", + "WebRTCSupportedVideoStreamTiers", + "WebRTCSupportedAudioStreamTiers", + "StreamingEnabled", + "SensorUUID" + ], + "UUID": "00008033-0000-1000-8000-0026BB765291" + }, "CarbonDioxideSensor": { "OptionalCharacteristics": [ "StatusActive", @@ -400,7 +490,9 @@ "StatusFault", "StatusTampered", "StatusLowBattery", - "Name" + "Name", + "ContributingSensors", + "MotionEnabled" ], "RequiredCharacteristics": [ "MotionDetected" diff --git a/tests/test_loader.py b/tests/test_loader.py index afe640a1..8714f65c 100644 --- a/tests/test_loader.py +++ b/tests/test_loader.py @@ -1,9 +1,18 @@ """Tests for pyhap.loader.""" +from unittest.mock import patch +from uuid import UUID + import pytest from pyhap import CHARACTERISTICS_FILE, SERVICES_FILE -from pyhap.characteristic import Characteristic +from pyhap.characteristic import ( + PROP_FORMAT, + PROP_MAX_VALUE, + PROP_MIN_VALUE, + PROP_PERMISSIONS, + Characteristic, +) from pyhap.loader import Loader, get_loader from pyhap.service import Service @@ -74,3 +83,158 @@ def test_get_loader() -> None: assert loader.serv_types == loader2.serv_types assert get_loader() == loader + + +# Accesses only admin controllers may use, not part of the HAP representation. +ADMIN_ONLY_ACCESS = "AdminOnlyAccess" + +# Camera services and characteristics of the HomeKit Secure Video 17.99 guide: +# name -> (short UUID, format, permissions, AdminOnlyAccess). +HKSV3_CHARS = { + "SensorUUID": ("805B", "data", {"pr"}, None), + "MotionEnabled": ("8087", "bool", {"pr", "pw", "ev", "tw"}, ["write"]), + "SupportedVideoStreamTiers": ("8043", "tlv8", {"pr", "ev"}, None), + "SupportedAudioStreamTiers": ("8044", "tlv8", {"pr", "ev"}, None), + "CameraCapabilities": ("8011", "tlv8", {"pr"}, None), + "ContributingSensors": ("8086", "tlv8", {"pr", "ev"}, None), + "CameraKey": ("8051", "tlv8", {"pw", "tw"}, None), + "CameraKeyID": ("8052", "tlv8", {"pr", "ev"}, None), + "BufferUploadCommand": ("8013", "tlv8", {"pr", "pw", "wr"}, None), + "BufferActivityCommand": ("8017", "tlv8", {"pw"}, None), + "BufferEventCommand": ("8014", "tlv8", {"pr", "pw", "wr"}, None), + "BufferEventSequenceNumber": ("8015", "uint32", {"pr", "ev"}, None), + "CameraRecordingPublishingPoint": ("8016", "tlv8", {"pr", "pw"}, None), + "CameraZones": ("8022", "tlv8", {"pr", "pw"}, None), + "StreamingEnabled": ("8041", "bool", {"pr", "pw", "ev", "tw"}, ["write"]), + "RTPStreamingControl": ("8045", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCSolicitOffer": ("8053", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCProvideAnswer": ("8054", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCStreamingControl": ("8056", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCNumberOfActiveSessions": ("8057", "uint8", {"pr", "ev"}, None), + "WebRTCReoffer": ("8058", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCUpdateSession": ("805C", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCSupportedVideoStreamTiers": ("8059", "tlv8", {"pr", "ev"}, None), + "WebRTCSupportedAudioStreamTiers": ("805A", "tlv8", {"pr", "ev"}, None), + "CameraClientCSR": ("8081", "tlv8", {"pr", "pw", "wr"}, None), + "CameraClientCertificate": ("8082", "tlv8", {"pr", "pw", "tw"}, None), + "CameraClientCertificateStatus": ("8083", "tlv8", {"pr", "ev"}, None), +} + +# name -> (short UUID, required, optional) +HKSV3_SERVICES = { + "CameraCapabilities": ( + "8010", + ["Version", "CameraCapabilities"], + ["ManuallyDisabled"], + ), + "CameraGlobalOperatingMode": ( + "8032", + ["HomeKitCameraActive", "StreamingEnabled", "CameraOperatingModeIndicator"], + ["ManuallyDisabled", "NightVision", "ThirdPartyCameraActive"], + ), + "CameraMotionZones": ("8021", ["Version", "Active", "CameraZones"], []), + "CameraBufferManagement": ( + "8000", + [ + "BufferUploadCommand", + "BufferActivityCommand", + "BufferEventCommand", + "BufferEventSequenceNumber", + "CameraRecordingPublishingPoint", + ], + [], + ), + "CameraMultiTierRTPStreamManagement": ( + "8031", + [ + "StreamingEnabled", + "StatusActive", + "SupportedVideoStreamTiers", + "SupportedAudioStreamTiers", + "SupportedRTPConfiguration", + "SetupEndpoints", + "RTPStreamingControl", + "SensorUUID", + ], + [], + ), + "CameraWebRTCStreamManagement": ( + "8033", + [ + "WebRTCSolicitOffer", + "WebRTCProvideAnswer", + "WebRTCStreamingControl", + "WebRTCNumberOfActiveSessions", + "WebRTCReoffer", + "WebRTCUpdateSession", + "WebRTCSupportedVideoStreamTiers", + "WebRTCSupportedAudioStreamTiers", + "StreamingEnabled", + "SensorUUID", + ], + [], + ), + "CameraKeyManagement": ("8050", ["CameraKey", "CameraKeyID"], []), + "CameraClientCertificateManagement": ( + "8080", + [ + "CameraClientCSR", + "CameraClientCertificate", + "CameraClientCertificateStatus", + ], + [], + ), +} + + +def _hap_uuid(short: str) -> UUID: + return UUID(f"0000{short}-0000-1000-8000-0026BB765291") + + +@pytest.mark.parametrize("name", HKSV3_CHARS) +def test_loader_hksv3_char(name: str) -> None: + """Test the HKSV 17.99 characteristics match the guide.""" + short, char_format, permissions, admin_only_access = HKSV3_CHARS[name] + char = Loader().get_char(name) + assert char.type_id == _hap_uuid(short) + assert char.properties[PROP_FORMAT] == char_format + assert set(char.properties[PROP_PERMISSIONS]) == permissions + assert char.properties.get(ADMIN_ONLY_ACCESS) == admin_only_access + + +def test_loader_hksv3_admin_only_not_advertised() -> None: + """Test the admin only marker stays out of the HAP representation.""" + char = Loader().get_char("StreamingEnabled") + with patch.object(char, "broker") as mock_broker: + mock_broker.iid_manager.get_iid.return_value = 2 + hap_repr = char.to_HAP() + assert hap_repr["perms"] == ["pr", "pw", "ev", "tw"] + assert ADMIN_ONLY_ACCESS not in hap_repr + + +def test_loader_hksv3_active_sessions_range() -> None: + """Test the WebRTC session count carries the guide's value range.""" + char = Loader().get_char("WebRTCNumberOfActiveSessions") + assert char.properties[PROP_MIN_VALUE] == 0 + assert char.properties[PROP_MAX_VALUE] == 255 + + +@pytest.mark.parametrize("name", HKSV3_SERVICES) +def test_loader_hksv3_service(name: str) -> None: + """Test the HKSV 17.99 services match the guide.""" + short, required, optional = HKSV3_SERVICES[name] + loader = Loader() + service = loader.get_service(name) + assert service.type_id == _hap_uuid(short) + assert [char.display_name for char in service.characteristics] == required + assert loader.serv_types[name]["OptionalCharacteristics"] == optional + assert set(optional) <= set(loader.char_types) + + +def test_loader_motion_sensor_hksv3_optional() -> None: + """Test the motion sensor gains the HKSV 17.99 chars only as optional.""" + loader = Loader() + optional = loader.serv_types["MotionSensor"]["OptionalCharacteristics"] + assert {"ContributingSensors", "MotionEnabled"} <= set(optional) + service = loader.get_service("MotionSensor") + assert [char.display_name for char in service.characteristics] == ["MotionDetected"] From 207041ce2e7017ff31e3a965a09578e0ae929ce7 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:48:49 +0100 Subject: [PATCH 2/8] Fix the camera operating mode characteristic permissions Third Party Camera Active is state the camera reports, so controllers read and subscribe to it but do not write it. Night Vision and Camera Operating Mode Indicator are written with timed writes. The Camera Global Operating Mode service of HKSV 17.99 uses all three. This matches the definitions in HAP-NodeJS. --- pyhap/resources/characteristics.json | 7 ++++--- tests/test_loader.py | 16 ++++++++++++++++ 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/pyhap/resources/characteristics.json b/pyhap/resources/characteristics.json index f185818d..639b04ff 100644 --- a/pyhap/resources/characteristics.json +++ b/pyhap/resources/characteristics.json @@ -220,7 +220,8 @@ "Permissions": [ "pr", "pw", - "ev" + "ev", + "tw" ], "UUID": "0000021D-0000-1000-8000-0026BB765291" }, @@ -965,7 +966,8 @@ "Permissions": [ "pr", "pw", - "ev" + "ev", + "tw" ], "UUID": "0000011B-0000-1000-8000-0026BB765291" }, @@ -1872,7 +1874,6 @@ "Format": "bool", "Permissions": [ "pr", - "pw", "ev" ], "UUID": "0000021C-0000-1000-8000-0026BB765291" diff --git a/tests/test_loader.py b/tests/test_loader.py index 8714f65c..a38f1426 100644 --- a/tests/test_loader.py +++ b/tests/test_loader.py @@ -238,3 +238,19 @@ def test_loader_motion_sensor_hksv3_optional() -> None: assert {"ContributingSensors", "MotionEnabled"} <= set(optional) service = loader.get_service("MotionSensor") assert [char.display_name for char in service.characteristics] == ["MotionDetected"] + + +@pytest.mark.parametrize( + ("name", "permissions"), + [ + ("CameraOperatingModeIndicator", {"pr", "pw", "ev", "tw"}), + ("NightVision", {"pr", "pw", "ev", "tw"}), + ("ThirdPartyCameraActive", {"pr", "ev"}), + ], +) +def test_loader_camera_operating_mode_permissions( + name: str, permissions: set[str] +) -> None: + """Test the operating mode settings shared with the HKSV 17.99 services.""" + char = Loader().get_char(name) + assert set(char.properties[PROP_PERMISSIONS]) == permissions From f83c74c0bbafd863884a1c0ff2d454b9383027a1 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Sun, 8 Nov 2026 14:45:55 +0100 Subject: [PATCH 3/8] Add TLV8 codecs for the HKSV 17.99 camera values Add pyhap.hksv3 with the camera capabilities, contributing sensors and camera zones values of the HomeKit Secure Video Open Source Compatibility Guide. Lists use the HAP TLV8 list encoding: one TLV per item with an empty type 0 TLV between items of the same type. Values over 255 bytes are split into 255-byte fragments, and the decoder joins a TLV to the one before it only after such a full fragment, like hksv_recording and HAP-NodeJS. Enum values a controller sends that are not known yet are kept as ints instead of failing the write, and a zone without method or a polygon without identifier is read like HAP-NodeJS does. The tests check the encoder output and the decoder against hand written vectors and a small independent parser of the list encoding. --- pyhap/hksv3.py | 280 +++++++++++++++++++++++++++++++++ tests/test_hksv3.py | 371 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 651 insertions(+) create mode 100644 pyhap/hksv3.py create mode 100644 tests/test_hksv3.py diff --git a/pyhap/hksv3.py b/pyhap/hksv3.py new file mode 100644 index 00000000..a7d6a10c --- /dev/null +++ b/pyhap/hksv3.py @@ -0,0 +1,280 @@ +"""TLV8 values of the HomeKit Secure Video 17.99 camera services. + +Follows the HAP TLV8 list encoding: each item of a list is a TLV of its own, +and items of the same type are separated by an empty TLV of type 0. Values +longer than 255 bytes are sent as 255-byte fragments of the same type; a TLV +continues the value before it only after such a full fragment. +""" + +from __future__ import annotations + +from collections.abc import Iterable +from dataclasses import dataclass, field +from enum import IntEnum +from typing import TypeVar +from uuid import UUID + +SEPARATOR = b"\x00\x00" + +# Version characteristic of the Camera Capabilities and Motion Zones services. +SERVICE_VERSION = "17.99" +CAPABILITIES_VERSION = 1 +ZONES_VERSION = 2 + +_E = TypeVar("_E", bound=IntEnum) + + +class SensorType(IntEnum): + UNKNOWN = 0 + PRIMARY = 1 + GENERIC = 255 + + +class SensorIntent(IntEnum): + UNKNOWN = 0 + MAIN = 1 + PACKAGE = 2 + GENERIC = 255 + + +class VideoQuality(IntEnum): + HIGHEST = 1 + HIGH = 2 + MEDIUM = 3 + LOW = 4 + + +class ZoneMethod(IntEnum): + NORMAL = 1 + INVERTED = 2 + + +def encode_item(tag: int, value: bytes) -> bytes: + """Encode one value, fragmented into TLVs of at most 255 bytes.""" + if not value: + return bytes((tag, 0)) + return b"".join( + bytes((tag, len(chunk))) + chunk + for chunk in (value[i : i + 255] for i in range(0, len(value), 255)) + ) + + +def encode_list(tag: int, values: Iterable[bytes]) -> bytes: + """Encode the values as list items of one type.""" + return SEPARATOR.join(encode_item(tag, value) for value in values) + + +def decode_items(data: bytes) -> list[tuple[int, bytes]]: + """Split TLV8 data into its items, joining fragments and dropping separators. + + Raises ``ValueError`` for truncated data. + """ + items: list[tuple[int, bytearray]] = [] + # Type of the full 255-byte fragment right before, which a TLV continues. + current: int | None = None + offset = 0 + while offset < len(data): + if offset + 2 > len(data): + raise ValueError(f"Truncated TLV header at offset {offset}") + tag, length = data[offset], data[offset + 1] + end = offset + 2 + length + if end > len(data): + raise ValueError(f"Truncated TLV value at offset {offset}") + value = data[offset + 2 : end] + offset = end + if tag == current: + items[-1][1].extend(value) + elif tag or length: + items.append((tag, bytearray(value))) + current = tag if length == 255 else None + return [(tag, bytes(value)) for tag, value in items] + + +def _fields(data: bytes) -> dict[int, list[bytes]]: + fields: dict[int, list[bytes]] = {} + for tag, value in decode_items(data): + fields.setdefault(tag, []).append(value) + return fields + + +def _required(fields: dict[int, list[bytes]], tag: int) -> bytes: + if tag not in fields: + raise ValueError(f"Missing TLV type {tag}") + return fields[tag][0] + + +def _uint(value: int, size: int) -> bytes: + return value.to_bytes(size, "little") + + +def _to_int(data: bytes) -> int: + if not 0 < len(data) <= 8: + raise ValueError(f"Invalid integer length {len(data)}") + return int.from_bytes(data, "little") + + +def _to_enum(enum_type: type[_E], data: bytes) -> _E | int: + # Values a newer controller may send are kept as plain ints. + value = _to_int(data) + try: + return enum_type(value) + except ValueError: + return value + + +def _to_uuid(data: bytes) -> UUID: + if len(data) != 16: + raise ValueError(f"Invalid UUID length {len(data)}") + return UUID(bytes=data) + + +@dataclass +class VideoStreamCapability: + """A video encoding the camera can produce from a sensor, bit rates in kbps.""" + + identifier: UUID + quality: VideoQuality | int + width: int + height: int + frame_rate: int + average_bitrate: int + peak_bitrate: int + + def encode(self) -> bytes: + return b"".join( + ( + encode_item(1, self.identifier.bytes), + encode_item(2, _uint(self.quality, 1)), + encode_item(3, _uint(self.width, 2)), + encode_item(4, _uint(self.height, 2)), + encode_item(5, _uint(self.frame_rate, 1)), + encode_item(6, _uint(self.average_bitrate, 4)), + encode_item(7, _uint(self.peak_bitrate, 4)), + ) + ) + + +@dataclass +class SensorConfiguration: + """An image sensor of the camera and its video encodings.""" + + sensor_uuid: UUID + width: int + height: int + sensor_type: SensorType | int = SensorType.PRIMARY + intent: SensorIntent | int = SensorIntent.MAIN + video_streams: list[VideoStreamCapability] = field(default_factory=list) + + def encode(self) -> bytes: + dimensions = encode_item(1, _uint(self.width, 2)) + encode_item( + 2, _uint(self.height, 2) + ) + return b"".join( + ( + encode_item(1, dimensions), + encode_item(2, self.sensor_uuid.bytes), + encode_item(3, _uint(self.sensor_type, 1)), + encode_item(4, _uint(self.intent, 1)), + encode_list(5, (stream.encode() for stream in self.video_streams)), + ) + ) + + +@dataclass +class CameraCapabilities: + """Value of the Camera Capabilities characteristic.""" + + sensors: list[SensorConfiguration] + version: int = CAPABILITIES_VERSION + + def encode(self) -> bytes: + sensors = encode_list(1, (sensor.encode() for sensor in self.sensors)) + return encode_item(1, _uint(self.version, 1)) + encode_item(2, sensors) + + +def encode_contributing_sensors(sensor_uuids: Iterable[UUID]) -> bytes: + """Encode the value of the Contributing Sensors characteristic.""" + return encode_list( + 1, (encode_item(1, sensor_uuid.bytes) for sensor_uuid in sensor_uuids) + ) + + +@dataclass +class ZonePolygon: + """A polygon in sensor pixels, the origin is the top left corner. + + A controller may leave out the identifier, it is then None. + """ + + identifier: UUID | None + vertices: list[tuple[int, int]] + + def encode(self) -> bytes: + vertices = b"".join(_uint(x, 2) + _uint(y, 2) for x, y in self.vertices) + identifier = b"" + if self.identifier is not None: + identifier = encode_item(1, self.identifier.bytes) + return identifier + encode_item(3, vertices) + + @classmethod + def decode(cls, data: bytes) -> ZonePolygon: + fields = _fields(data) + raw = fields.get(3, [b""])[0] + if len(raw) % 4: + raise ValueError(f"Invalid vertices length {len(raw)}") + vertices = [ + (_to_int(raw[i : i + 2]), _to_int(raw[i + 2 : i + 4])) + for i in range(0, len(raw), 4) + ] + identifier = _to_uuid(fields[1][0]) if 1 in fields else None + return cls(identifier, vertices) + + +@dataclass +class ZoneData: + """Polygons and how they form the region of interest. + + A missing method is :attr:`ZoneMethod.NORMAL`. + """ + + method: ZoneMethod | int + polygons: list[ZonePolygon] = field(default_factory=list) + + def encode(self) -> bytes: + return encode_item(1, _uint(self.method, 1)) + encode_list( + 3, (polygon.encode() for polygon in self.polygons) + ) + + @classmethod + def decode(cls, data: bytes) -> ZoneData: + fields = _fields(data) + method: ZoneMethod | int = ZoneMethod.NORMAL + if 1 in fields: + method = _to_enum(ZoneMethod, fields[1][0]) + return cls( + method, [ZonePolygon.decode(polygon) for polygon in fields.get(3, [])] + ) + + +@dataclass +class CameraZones: + """Value of the Camera Zones characteristic. + + Only version 2 zone data is decoded, ``zones`` is empty for others. + """ + + zones: list[ZoneData] = field(default_factory=list) + version: int = ZONES_VERSION + + def encode(self) -> bytes: + return encode_item(1, _uint(self.version, 1)) + encode_list( + 2, (zone.encode() for zone in self.zones) + ) + + @classmethod + def decode(cls, data: bytes) -> CameraZones: + fields = _fields(data) + version = _to_int(_required(fields, 1)) + if version != ZONES_VERSION: + return cls([], version) + return cls([ZoneData.decode(zone) for zone in fields.get(2, [])], version) diff --git a/tests/test_hksv3.py b/tests/test_hksv3.py new file mode 100644 index 00000000..008bf588 --- /dev/null +++ b/tests/test_hksv3.py @@ -0,0 +1,371 @@ +"""Tests for pyhap.hksv3.""" + +import random +from uuid import UUID + +import pytest + +from pyhap import hksv3 +from pyhap.hksv3 import ( + CameraCapabilities, + CameraZones, + SensorConfiguration, + SensorIntent, + SensorType, + VideoQuality, + VideoStreamCapability, + ZoneData, + ZoneMethod, + ZonePolygon, +) + +SENSOR = UUID("00112233-4455-6677-8899-aabbccddeeff") +STREAM = UUID("ffeeddcc-bbaa-9988-7766-554433221100") +ZONE = UUID("0123456789abcdef0123456789abcdef") + + +def reference_parse(data: bytes) -> list[tuple[int, bytes]]: + """Parse TLV8 the way the HAP TLV8 list encoding defines it. + + A TLV of the same type continues the value only after a full 255-byte + fragment, a zero-length type 0 TLV is a separator and no item. + """ + items: list[tuple[int, bytes]] = [] + previous: tuple[int, int] | None = None + rest = memoryview(data) + while rest: + tag, length, value = rest[0], rest[1], bytes(rest[2 : 2 + rest[1]]) + assert len(value) == length + rest = rest[2 + length :] + if previous == (tag, 255): + items[-1] = (tag, items[-1][1] + value) + elif (tag, length) != (0, 0): + items.append((tag, value)) + previous = (tag, length) + return items + + +def test_encode_item_fragments() -> None: + """Test values over 255 bytes are split into TLVs of the same type.""" + value = bytes(range(256)) * 2 + b"\x01" * 88 + encoded = hksv3.encode_item(3, value) + assert ( + encoded + == (b"\x03\xff" + value[:255] + b"\x03\xff" + value[255:510] + b"\x03\x5a") + + value[510:] + ) + assert reference_parse(encoded) == [(3, value)] + + +@pytest.mark.parametrize("length", [0, 1, 255, 510]) +def test_encode_item_boundaries(length: int) -> None: + """Test empty values and multiples of 255 bytes have no extra TLV.""" + value = b"\xaa" * length + encoded = hksv3.encode_item(7, value) + assert len(encoded) == length + 2 * max(1, -(-length // 255)) + assert reference_parse(encoded) == [(7, value)] + + +def test_encode_list() -> None: + """Test list items are separated by an empty type 0 TLV.""" + assert hksv3.encode_list(3, [b"\xaa", b"\xbb\xcc", b""]) == bytes.fromhex( + "0301aa 0000 0302bbcc 0000 0300" + ) + assert hksv3.encode_list(3, []) == b"" + + +@pytest.mark.parametrize( + ("data", "items"), + [ + ( + "010105 0302aabb 0000 0301cc", + [(1, "05"), (3, "aabb"), (3, "cc")], + ), + ("0000 0301aa 0000", [(3, "aa")]), + ("0301aa 0302bbcc 0401dd", [(3, "aa"), (3, "bbcc"), (4, "dd")]), + ("0300 0000 0300", [(3, ""), (3, "")]), + ("0301aa 0401bb 0301cc", [(3, "aa"), (4, "bb"), (3, "cc")]), + ], +) +def test_decode_items(data: str, items: list[tuple[int, str]]) -> None: + """Test decoding hand written TLV8 lists.""" + raw = bytes.fromhex(data) + expected = [(tag, bytes.fromhex(value)) for tag, value in items] + assert hksv3.decode_items(raw) == expected + assert reference_parse(raw) == expected + + +def test_decode_items_fragments() -> None: + """Test a fragmented value ends at a separator.""" + first = bytes(range(255)) + b"\x01\x02" + raw = b"\x05\xff" + first[:255] + b"\x05\x02\x01\x02\x00\x00\x05\x01\x03" + assert hksv3.decode_items(raw) == [(5, first), (5, b"\x03")] + + +@pytest.mark.parametrize("data", ["03", "0302aa", "0301aa 04"]) +def test_decode_items_truncated(data: str) -> None: + """Test truncated data raises.""" + with pytest.raises(ValueError, match="Truncated"): + hksv3.decode_items(bytes.fromhex(data)) + + +def _random_tlv8(seed: int) -> bytes: + """Build a random TLV8 sequence with separators and full fragments.""" + rng = random.Random(seed) + tlvs = [ + bytes((tag, length)) + rng.randbytes(length) + for tag, length in ( + (rng.randint(0, 4), rng.choice([0, 0, 1, 2, 254, 255, 255])) + for _ in range(rng.randint(0, 12)) + ) + ] + return b"".join(tlvs) + + +@pytest.mark.parametrize("seed", range(50)) +def test_decode_items_matches_reference(seed: int) -> None: + """Test random TLV8 sequences parse like the reference parser.""" + raw = _random_tlv8(seed) + assert hksv3.decode_items(raw) == reference_parse(raw) + + +def _capabilities() -> CameraCapabilities: + streams = [ + VideoStreamCapability(STREAM, VideoQuality.HIGH, 1920, 1080, 30, 1700, 1800), + VideoStreamCapability(STREAM, VideoQuality.LOW, 640, 360, 15, 180, 190), + ] + return CameraCapabilities( + [ + SensorConfiguration(SENSOR, 2688, 1520, video_streams=streams), + SensorConfiguration( + STREAM, 1600, 1200, SensorType.GENERIC, SensorIntent.PACKAGE + ), + ] + ) + + +def test_camera_capabilities_golden() -> None: + """Test a single sensor capabilities value byte for byte.""" + stream = VideoStreamCapability(STREAM, VideoQuality.LOW, 640, 360, 15, 180, 190) + value = CameraCapabilities( + [SensorConfiguration(SENSOR, 1920, 1080, video_streams=[stream])] + ) + stream_tlv = ( + "0110" + + STREAM.hex + + "020104" + + "03028002" + + "04026801" + + "05010f" + + "0604b4000000" + + "0704be000000" + ) + sensor_tlv = ( + "0108" + + "01028007" + + "02023804" + + "0210" + + SENSOR.hex + + "030101" + + "040101" + + "05" + + f"{len(stream_tlv) // 2:02x}" + + stream_tlv + ) + expected = "010101" + "02" + f"{len(sensor_tlv) // 2 + 2:02x}" + expected += "01" + f"{len(sensor_tlv) // 2:02x}" + sensor_tlv + assert value.encode() == bytes.fromhex(expected) + + +def test_camera_capabilities_lists() -> None: + """Test sensors and video streams are separate list items.""" + top = reference_parse(_capabilities().encode()) + assert [tag for tag, _ in top] == [1, 2] + assert top[0][1] == b"\x01" + sensors = reference_parse(top[1][1]) + assert [tag for tag, _ in sensors] == [1, 1] + + first = dict(reference_parse(sensors[0][1])) + assert reference_parse(first[1]) == [(1, b"\x80\x0a"), (2, b"\xf0\x05")] + assert first[2] == SENSOR.bytes + assert first[3] == bytes([SensorType.PRIMARY]) + assert first[4] == bytes([SensorIntent.MAIN]) + streams = [value for tag, value in reference_parse(sensors[0][1]) if tag == 5] + assert len(streams) == 2 + assert dict(reference_parse(streams[1]))[2] == bytes([VideoQuality.LOW]) + + second = reference_parse(sensors[1][1]) + assert (3, b"\xff") in second + assert (4, bytes([SensorIntent.PACKAGE])) in second + assert 5 not in dict(second) + + +def test_camera_capabilities_fragmented() -> None: + """Test a value over 255 bytes, two sensors with four encodings each.""" + qualities = list(VideoQuality) + sensors = [ + SensorConfiguration( + sensor_uuid, + 3840, + 2160, + video_streams=[ + VideoStreamCapability(UUID(int=i), quality, 640, 360, 15, 180, 190) + for i, quality in enumerate(qualities) + ], + ) + for sensor_uuid in (SENSOR, STREAM) + ] + encoded = CameraCapabilities(sensors).encode() + # Version, then the sensors list in a full and a partial type 2 fragment. + assert encoded[:5] == bytes.fromhex("010101 02ff") + assert encoded[260] == 2 + assert len(encoded) == 3 + 2 * 2 + 454 + + top = reference_parse(encoded) + assert [(tag, len(value)) for tag, value in top] == [(1, 1), (2, 454)] + items = reference_parse(top[1][1]) + assert [tag for tag, _ in items] == [1, 1] + sensor_fields = [reference_parse(item) for _, item in items] + streams = [ + [dict(reference_parse(value)) for tag, value in fields if tag == 5] + for fields in sensor_fields + ] + assert [dict(fields)[2] for fields in sensor_fields] == [ + SENSOR.bytes, + STREAM.bytes, + ] + expected = [(UUID(int=i).bytes, bytes([q])) for i, q in enumerate(qualities)] + assert [[(s[1], s[2]) for s in sensor] for sensor in streams] == [expected] * 2 + + +def test_contributing_sensors() -> None: + """Test each contributing sensor is its own list item.""" + encoded = hksv3.encode_contributing_sensors([SENSOR, STREAM]) + assert encoded == bytes.fromhex( + "0112 0110" + SENSOR.hex + "0000 0112 0110" + STREAM.hex + ) + assert hksv3.encode_contributing_sensors([]) == b"" + + +def _zones_hex(method: str = "01", vertices: str = "0000000080070000") -> str: + polygon_a = "0110" + ZONE.hex + "03" + f"{len(vertices) // 2:02x}" + vertices + polygon_b = "0110" + STREAM.hex + "030c" + "0a000a00" + "14000a00" + "0a001400" + zone = ( + "0101" + + method + + "03" + + f"{len(polygon_a) // 2:02x}" + + polygon_a + + "0000" + + "03" + + f"{len(polygon_b) // 2:02x}" + + polygon_b + ) + return "010102" + "02" + f"{len(zone) // 2:02x}" + zone + + +def test_camera_zones_decode() -> None: + """Test decoding hand written zones with two polygons.""" + raw = bytes.fromhex(_zones_hex()) + zones = CameraZones.decode(raw) + assert zones == CameraZones( + [ + ZoneData( + ZoneMethod.NORMAL, + [ + ZonePolygon(ZONE, [(0, 0), (1920, 0)]), + ZonePolygon(STREAM, [(10, 10), (20, 10), (10, 20)]), + ], + ) + ] + ) + assert zones.encode() == raw + + +def test_camera_zones_several() -> None: + """Test several zone data items survive a round trip.""" + zones = CameraZones( + [ + ZoneData(ZoneMethod.INVERTED, [ZonePolygon(ZONE, [(1, 2)])]), + ZoneData(ZoneMethod.NORMAL), + ] + ) + encoded = zones.encode() + assert [tag for tag, _ in reference_parse(encoded)] == [1, 2, 2] + assert CameraZones.decode(encoded) == zones + + +def test_camera_zones_fragmented_polygon() -> None: + """Test a polygon with more than 255 bytes of vertices.""" + vertices = [(i, 2 * i) for i in range(100)] + zones = CameraZones([ZoneData(ZoneMethod.NORMAL, [ZonePolygon(ZONE, vertices)])]) + encoded = zones.encode() + zone = reference_parse(reference_parse(encoded)[1][1]) + assert [tag for tag, _ in zone] == [1, 3] + polygon = dict(reference_parse(zone[1][1])) + assert polygon[1] == ZONE.bytes + assert polygon[3] == b"".join( + x.to_bytes(2, "little") + y.to_bytes(2, "little") for x, y in vertices + ) + assert CameraZones.decode(encoded) == zones + + +def test_camera_zones_unknown_method() -> None: + """Test an unknown zone method is kept as int.""" + zones = CameraZones.decode(bytes.fromhex(_zones_hex(method="09"))) + assert zones.zones[0].method == 9 + assert not isinstance(zones.zones[0].method, ZoneMethod) + + +def test_camera_zones_unknown_types() -> None: + """Test unknown TLV types are ignored at every level.""" + polygon = "0110" + ZONE.hex + "0304" + "01000200" + "0902aabb" + zone = "010101" + "0901cc" + "03" + f"{len(polygon) // 2:02x}" + polygon + raw = bytes.fromhex("010102" + "0a01dd" + "02" + f"{len(zone) // 2:02x}" + zone) + assert reference_parse(raw)[1] == (10, b"\xdd") + assert CameraZones.decode(raw) == CameraZones( + [ZoneData(ZoneMethod.NORMAL, [ZonePolygon(ZONE, [(1, 2)])])] + ) + + +@pytest.mark.parametrize( + ("zone", "expected"), + [ + # No method. + ( + "0306" + "0304" + "01000200", + ZoneData(ZoneMethod.NORMAL, [ZonePolygon(None, [(1, 2)])]), + ), + # No polygon identifier and no vertices. + ("010102" + "0300", ZoneData(ZoneMethod.INVERTED, [ZonePolygon(None, [])])), + ], +) +def test_camera_zones_missing_fields(zone: str, expected: ZoneData) -> None: + """Test missing fields get defaults, like HAP-NodeJS reads them.""" + raw = bytes.fromhex("010102" + "02" + f"{len(zone) // 2:02x}" + zone) + zones = CameraZones.decode(raw) + assert zones == CameraZones([expected]) + assert CameraZones.decode(zones.encode()) == zones + + +def test_camera_zones_other_version() -> None: + """Test zone data of another version is not interpreted.""" + zones = CameraZones.decode(bytes.fromhex("010103 0203aabbcc")) + assert zones == CameraZones([], 3) + + +@pytest.mark.parametrize( + "data", + [ + "", + "0100", + "010902000000000000000000", + _zones_hex(vertices="000000"), + "010102 0203 030100", + "010102 020b 010101 0306 0102aabb 0300", + ], +) +def test_camera_zones_invalid(data: str) -> None: + """Test malformed zones raise ValueError.""" + with pytest.raises(ValueError): + CameraZones.decode(bytes.fromhex(data)) From 576f6e75e8ad547f54775221896dea4424f682f8 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Sun, 8 Nov 2026 18:29:51 +0100 Subject: [PATCH 4/8] Add HKSV 17.99 camera capabilities and motion zones A new "sensors" camera option takes the image sensors of the camera, each with its real dimensions, type, intent (main or package camera) and video encodings. With it the camera advertises them in the Camera Capabilities service, accepts motion zones through the Camera Motion Zones service and adds Contributing Sensors and Motion Enabled to the motion sensor. Cameras without the option are unchanged. set_motion_detected reports the sensors that saw the motion and reports none while the user turned motion detection off. Integrations read the zones from camera_zones and override motion_zones_changed and motion_enabled_changed to react. The hooks only hear about changes, not about a controller writing the same value again. Zones, their Active state and Motion Enabled are kept in the driver's accessory state like the classic recording state: restored when the camera is rebuilt and returned to their defaults when the last pairing is removed, calling the hooks for the values that change once all state is reset. Both kinds of state share one save per burst of writes. A zones write that does not decode fails and leaves the previous zones in place. A warning is logged for sensors with fewer than the three video encodings the guide requires and for motion reported for sensors that are not configured. --- pyhap/camera.py | 299 +++++++++++++++++++++--- tests/test_camera_capabilities.py | 377 ++++++++++++++++++++++++++++++ tests/test_hksv3_state.py | 158 +++++++++++++ 3 files changed, 807 insertions(+), 27 deletions(-) create mode 100644 tests/test_camera_capabilities.py create mode 100644 tests/test_hksv3_state.py diff --git a/pyhap/camera.py b/pyhap/camera.py index b20cac9e..e898c7b3 100644 --- a/pyhap/camera.py +++ b/pyhap/camera.py @@ -17,7 +17,7 @@ """ import asyncio -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, Callable, Iterable import functools import hashlib import ipaddress @@ -28,7 +28,15 @@ from typing import Any from uuid import UUID -from pyhap import RESOURCE_DIR, hds, hds_recording, hds_server, hksv_recording, tlv +from pyhap import ( + RESOURCE_DIR, + hds, + hds_recording, + hds_server, + hksv3, + hksv_recording, + tlv, +) from pyhap.accessory import Accessory from pyhap.camera_talkback import TalkbackReceiver, talkback_available from pyhap.characteristic import Characteristic @@ -243,6 +251,11 @@ ("CameraOperatingMode", "HomeKitCameraActive"), ("CameraOperatingMode", "PeriodicSnapshotsActive"), ) +# accessory_state key of the HKSV 17.99 settings. +HKSV3_STATE = "hksv3" +HKSV3_MOTION_CHARS = ("ContributingSensors", "MotionEnabled") +# Concurrent video encodings each sensor must offer (HKSV 17.99 guide, section 2). +HKSV3_MIN_SENSOR_STREAMS = 3 SUPPORTED_RECORDING_CHARS = ( "SupportedCameraRecordingConfiguration", "SupportedVideoRecordingConfiguration", @@ -459,6 +472,14 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None: the stream. The string can contain the keywords, corresponding to the video and audio configuration that was negotiated between the camera and the client. See the ``start`` method for a full list of parameters. + - sensors - list of :class:`pyhap.hksv3.SensorConfiguration`, the image + sensors of the camera. Opts into the HomeKit Secure Video 17.99 + Camera Capabilities and Camera Motion Zones services, and adds + Contributing Sensors and Motion Enabled to the motion sensor, + or adds a motion sensor without ``recording``. Each sensor needs + at least three video encodings. With ``recording``, turning it + on for a paired camera changes the instance IDs of the recording + services with the default IID manager. :type options: ``dict`` """ @@ -494,7 +515,8 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None: # Last selection that decoded, the char may hold a rejected write. self._selected_recording_raw = None self._recording_defaults = {} - self._recording_save_pending = False + self._save_pending = False + self._recording_changed = False self._motion_detected_char = None # Bumped on stop so a start still binding does not publish its listener. self._hds_generation = 0 @@ -503,11 +525,27 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None: self._recording_stream = None self._homekit_camera_active_char = None self._motion_status_active_char = None + # Persisted HKSV 17.99 chars with their state key, default and the + # handler a reset calls. + self._hksv3_defaults: dict[ + Characteristic, tuple[str, Any, Callable[[Any], None]] + ] = {} + # The values the handlers last saw, they only hear about changes. + self._hksv3_seen: dict[Characteristic, Any] = {} + self._sensors: list[hksv3.SensorConfiguration] = list( + options.get("sensors") or [] + ) + self._motion_zones_service: Service | None = None + self._contributing_sensors_char: Characteristic | None = None + self._motion_enabled_char: Characteristic | None = None if options.get("video"): self._setup_stream_management(options) # Classic HomeKit Secure Video recording (HDS + fragmented MP4). if options.get("recording"): self._setup_recording_management(options) + if self._sensors: + self._setup_camera_capabilities() + self._restore_hksv3_state() @property def streaming_status(self) -> bytes | None: @@ -624,7 +662,10 @@ def _setup_recording_management(self, options: dict[str, Any]) -> None: # HKSV records on an event trigger; the camera owns a motion sensor and # links it (and the data stream transport) to the recording management. - motion = self.add_preload_service("MotionSensor", chars=["StatusActive"]) + motion = self.add_preload_service( + "MotionSensor", + chars=["StatusActive", *(HKSV3_MOTION_CHARS if self._sensors else ())], + ) self._motion_detected_char = motion.configure_char( "MotionDetected", value=False ) @@ -668,6 +709,114 @@ def _setup_recording_management(self, options: dict[str, Any]) -> None: self._restore_recording_state() self._setup_data_stream_transport() + def _setup_camera_capabilities(self) -> None: + """Advertise the image sensors and accept motion zones (HKSV 17.99).""" + for sensor in self._sensors: + if len(sensor.video_streams) < HKSV3_MIN_SENSOR_STREAMS: + logger.warning( + "%s: sensor %s has %d video encodings, HKSV needs at least %d", + self.display_name, + sensor.sensor_uuid, + len(sensor.video_streams), + HKSV3_MIN_SENSOR_STREAMS, + ) + capabilities = self.add_preload_service("CameraCapabilities") + capabilities.configure_char("Version", value=hksv3.SERVICE_VERSION) + capabilities.configure_char( + "CameraCapabilities", + value=to_base64_str(hksv3.CameraCapabilities(self._sensors).encode()), + ) + zones = self.add_preload_service("CameraMotionZones") + zones.configure_char("Version", value=hksv3.SERVICE_VERSION) + self._motion_zones_service = zones + + motion = self.get_service("MotionSensor") + if motion is None: + motion = self.add_preload_service("MotionSensor", chars=HKSV3_MOTION_CHARS) + self._motion_detected_char = motion.get_characteristic("MotionDetected") + self._contributing_sensors_char = motion.get_characteristic( + "ContributingSensors" + ) + self._motion_enabled_char = motion.configure_char("MotionEnabled", value=True) + for char, key, handler in ( + ( + zones.get_characteristic("Active"), + "motion_zones_active", + self._handle_motion_zones, + ), + ( + zones.get_characteristic("CameraZones"), + "camera_zones", + self._handle_motion_zones, + ), + (self._motion_enabled_char, "motion_enabled", self._handle_motion_enabled), + ): + self._persist_hksv3_char(char, key, handler) + # Check the zones before they count as written. + zones.get_characteristic("CameraZones").setter_callback = self._set_camera_zones + + @property + def camera_zones(self) -> hksv3.CameraZones | None: + """The motion zones the controller wrote, if any.""" + if self._motion_zones_service is None: + return None + value = self._motion_zones_service.get_characteristic("CameraZones").value + if not value: + return None + try: + return hksv3.CameraZones.decode(base64_to_bytes(value)) + # A damaged state file. + except ValueError: + return None + + @property + def motion_zones_active(self) -> bool: + """Whether only motion inside :attr:`camera_zones` should count.""" + if self._motion_zones_service is None: + return False + return bool(self._motion_zones_service.get_characteristic("Active").value) + + @property + def motion_enabled(self) -> bool: + """Whether motion detection is on, cameras without the setting always are.""" + return self._motion_enabled_char is None or bool( + self._motion_enabled_char.value + ) + + def _set_camera_zones(self, value: str) -> None: + assert self._motion_zones_service is not None # noqa: S101 + char = self._motion_zones_service.get_characteristic("CameraZones") + try: + if value: + hksv3.CameraZones.decode(base64_to_bytes(value)) + except ValueError: + # Fail the write and keep the zones the camera had. + char.value = self._hksv3_seen[char] + raise + self._hksv3_written(char, value) + + def _handle_motion_zones(self, value: Any) -> None: + self.motion_zones_changed() + + def motion_zones_changed(self) -> None: + """React to the controller changing :attr:`camera_zones` or turning + them on or off. Override. + + Only changes call it, not a controller writing the same value again. + """ + + def _handle_motion_enabled(self, value: Any) -> None: + if not value: + self.set_motion_detected(False) + self.motion_enabled_changed(bool(value)) + + def motion_enabled_changed(self, enabled: bool) -> None: + """React to the user turning motion detection on or off. Override. + + While it is off, :meth:`set_motion_detected` reports no motion. Only + changes call it, not a controller writing the same value again. + """ + def _setup_data_stream_transport(self) -> None: """Create the Data Stream Transport Management service used by HDS. @@ -828,10 +977,24 @@ def publish( sender_client_addr: Any = None, immediate: bool = False, ) -> None: - """Forward the update and store recording state changes.""" + """Forward the update and store recording and HKSV 17.99 state changes.""" super().publish(value, sender, sender_client_addr, immediate) + if sender in self._hksv3_defaults: + # Local changes too, so a later write of the same value is no change. + self._hksv3_seen[sender] = value # Without an aid there is no entry to store under. - if sender not in self._recording_defaults or self.aid is None: + if self.aid is None: + return + if sender in self._hksv3_defaults: + key = self._hksv3_defaults[sender][0] + state = self.driver.accessory_state(self.aid) + kept = state.get(HKSV3_STATE) + # A damaged state file must not fail the write. + if not isinstance(kept, dict): + kept = state[HKSV3_STATE] = {} + kept[key] = value + self._schedule_save("camera") + if sender not in self._recording_defaults: return # Store now so a camera rebuilt before the save restores this change. stored = {char.display_name: char.value for char in self._recording_defaults} @@ -841,15 +1004,20 @@ def publish( self._schedule_recording_save() def _schedule_recording_save(self) -> None: - if self._recording_save_pending: + self._recording_changed = True + self._schedule_save("recording") + + def _schedule_save(self, what: str) -> None: + if self._save_pending: return # One disk write per hub burst. - self._recording_save_pending = True + self._save_pending = True try: - self.driver.add_job(self._save_recording_state) + self.driver.add_job(self._save_state) except RuntimeError: - self._recording_save_pending = False - logger.warning("Could not schedule saving the recording state") + self._save_pending = False + self._recording_changed = False + logger.warning("Could not schedule saving the %s state", what) def _supported_recording_hash(self) -> str: digest = hashlib.sha256() @@ -860,14 +1028,53 @@ def _supported_recording_hash(self) -> str: return digest.hexdigest() @callback - def _save_recording_state(self) -> None: - self._recording_save_pending = False + def _save_state(self) -> None: + self._save_pending = False + recording_changed, self._recording_changed = self._recording_changed, False self.driver.async_persist() - self.recording_state_changed() + if recording_changed: + self.recording_state_changed() + + def _persist_hksv3_char( + self, char: Characteristic, key: str, handler: Callable[[Any], None] + ) -> None: + """Keep the value of ``char`` across restarts, reset it for a new home. + + A controller write or a reset that changes the value calls ``handler`` + with the new value, a reset once per handler. The constructor restores + the stored values afterwards. + """ + self._hksv3_defaults[char] = (key, char.value, handler) + char.setter_callback = functools.partial(self._hksv3_written, char) + + def _hksv3_written(self, char: Characteristic, value: Any) -> None: + """Call the handler of ``char`` if the write changed its value.""" + if self._hksv3_seen.get(char) == value: + return + self._hksv3_seen[char] = value + self._hksv3_defaults[char][2](value) + + def _restore_hksv3_state(self) -> None: + if self.aid is not None: + self._load_hksv3_state(self.aid) + self._hksv3_seen = {char: char.value for char in self._hksv3_defaults} + + def _load_hksv3_state(self, aid: int) -> None: + stored = self.driver.accessory_state(aid).get(HKSV3_STATE) + if not isinstance(stored, dict): + return + for char, (key, _, _) in self._hksv3_defaults.items(): + if key not in stored: + continue + try: + char.set_value(stored[key], should_notify=False) + except (TypeError, ValueError, OverflowError): + logger.warning("Ignoring stored %s of aid %s", key, aid) def restore_accessory_state(self) -> None: - """Restore the stored recording state.""" + """Restore the stored recording and HKSV 17.99 state.""" self._restore_recording_state() + self._restore_hksv3_state() def _restore_recording_state(self) -> None: # Private so construction does not run a subclass override early. @@ -908,26 +1115,64 @@ def _restore_recording_state(self) -> None: self._selected_recording_raw = selected def reset_accessory_state(self) -> None: - """Return the recording state to its defaults for a new home.""" - if not self._recording_defaults: - return - for char, value in self._recording_defaults.items(): - char.set_value(value, should_notify=False) - self._selected_recording_config = None - self._selected_recording_raw = None - if self.aid is not None: - self.driver.accessory_state(self.aid).pop("recording", None) - self._schedule_recording_save() + """Return the recording and HKSV 17.99 state to its defaults. - def set_motion_detected(self, detected: bool) -> None: + Once all state is reset, the handlers of the HKSV 17.99 values that + changed are called. + """ + handlers: list[tuple[Callable[[Any], None], Any]] = [] + for char, (_, value, handler) in self._hksv3_defaults.items(): + changed = char.value != value + char.set_value(value, should_notify=False) + self._hksv3_seen[char] = value + if changed and all(handler != known for known, _ in handlers): + handlers.append((handler, value)) + if self._hksv3_defaults and self.aid is not None: + self.driver.accessory_state(self.aid).pop(HKSV3_STATE, None) + self._schedule_save("camera") + if self._recording_defaults: + for char, value in self._recording_defaults.items(): + char.set_value(value, should_notify=False) + self._selected_recording_config = None + self._selected_recording_raw = None + if self.aid is not None: + self.driver.accessory_state(self.aid).pop("recording", None) + self._schedule_recording_save() + for handler, value in handlers: + handler(value) + + def set_motion_detected( + self, detected: bool, sensors: Iterable[UUID] | None = None + ) -> None: """Fire (``True``) or clear (``False``) the recording motion trigger. This is the entry point an integration calls when its own motion detector changes state: it drives the HKSV event-trigger MotionSensor, which is what makes the Home Hub start (and later finish) a recording. + + With ``sensors`` the camera names the sensors that saw the motion, + by default or when empty the first one. Ignored without the + ``sensors`` option. """ + detected = bool(detected) and self.motion_enabled + if self._contributing_sensors_char is not None: + if not detected: + sensors = () + elif not (sensors := tuple(sensors or ())): + sensors = (self._sensors[0].sensor_uuid,) + else: + configured = {sensor.sensor_uuid for sensor in self._sensors} + if unknown := [uuid for uuid in sensors if uuid not in configured]: + logger.warning( + "%s: motion reported for sensors that are not configured: %s", + self.display_name, + ", ".join(map(str, unknown)), + ) + self._contributing_sensors_char.set_value( + to_base64_str(hksv3.encode_contributing_sensors(sensors)) + ) if self._motion_detected_char is not None: - self._motion_detected_char.set_value(bool(detected)) + self._motion_detected_char.set_value(detected) def _recording_open_gate( self, manager: hds_recording.RecordingStreamManager diff --git a/tests/test_camera_capabilities.py b/tests/test_camera_capabilities.py new file mode 100644 index 00000000..ad7436fe --- /dev/null +++ b/tests/test_camera_capabilities.py @@ -0,0 +1,377 @@ +"""Tests for the HKSV 17.99 camera capabilities of pyhap.camera.""" + +import asyncio +from collections.abc import AsyncIterator +import logging +import pathlib +from typing import Any +from unittest.mock import MagicMock, patch +from uuid import UUID + +import pytest + +from pyhap import hksv3 +from pyhap.accessory_driver import AccessoryDriver +from pyhap.camera import HKSV3_STATE, Camera +from pyhap.const import ( + HAP_REPR_AID, + HAP_REPR_CHARS, + HAP_REPR_IID, + HAP_REPR_STATUS, + HAP_REPR_VALUE, + HAP_SERVER_STATUS, +) +from pyhap.util import base64_to_bytes, to_base64_str + +CAMERA_AID = 2 +MAIN = UUID("00112233-4455-6677-8899-aabbccddeeff") +PACKAGE = UUID("ffeeddcc-bbaa-9988-7766-554433221100") +SENSORS = [ + hksv3.SensorConfiguration( + MAIN, + 2688, + 1520, + video_streams=[ + hksv3.VideoStreamCapability( + PACKAGE, hksv3.VideoQuality.HIGH, 1920, 1080, 30, 1700, 1800 + ) + ], + ), + hksv3.SensorConfiguration( + PACKAGE, 1600, 1200, hksv3.SensorType.GENERIC, hksv3.SensorIntent.PACKAGE + ), +] +ZONES = hksv3.CameraZones( + [ + hksv3.ZoneData( + hksv3.ZoneMethod.NORMAL, + [hksv3.ZonePolygon(MAIN, [(0, 0), (1344, 0), (1344, 760)])], + ) + ] +) + + +@pytest.fixture(name="driver") +async def driver_fixture(tmp_path: pathlib.Path) -> AsyncIterator[AccessoryDriver]: + with patch("pyhap.accessory_driver.AccessoryDriver.async_persist"): + yield AccessoryDriver( + loop=asyncio.get_running_loop(), + listen_address="127.0.0.1", + persist_file=str(tmp_path / "accessory.state"), + ) + + +def _camera(driver: Any, **options: Any) -> Camera: + return Camera({"address": "127.0.0.1", **options}, driver, "Cam", aid=CAMERA_AID) + + +def _char(camera: Camera, service: str, name: str) -> Any: + found = camera.get_service(service) + assert found is not None + return found.get_characteristic(name) + + +def _chars(camera: Camera, service: str) -> list[str | None] | None: + found = camera.get_service(service) + if found is None: + return None + return [char.display_name for char in found.characteristics] + + +@pytest.mark.parametrize( + ("recording", "motion_chars"), + [(False, None), (True, ["MotionDetected", "StatusActive"])], +) +async def test_without_sensors( + driver: Any, recording: bool, motion_chars: list[str] | None +) -> None: + """Test cameras without the sensors option are unchanged.""" + camera = _camera(driver, recording=recording) + assert camera.get_service("CameraCapabilities") is None + assert camera.get_service("CameraMotionZones") is None + assert _chars(camera, "MotionSensor") == motion_chars + assert camera.camera_zones is None + assert camera.motion_zones_active is False + assert camera.motion_enabled is True + camera.set_motion_detected(True, [MAIN]) + + +async def test_sensor_minimum_streams( + driver: Any, caplog: pytest.LogCaptureFixture +) -> None: + """Test sensors with fewer than three encodings log a warning.""" + stream = SENSORS[0].video_streams[0] + sensors = [ + hksv3.SensorConfiguration(MAIN, 1920, 1080, video_streams=[stream] * 3), + *SENSORS[1:], + ] + with caplog.at_level(logging.WARNING): + _camera(driver, sensors=sensors) + assert caplog.messages == [ + f"Cam: sensor {PACKAGE} has 0 video encodings, HKSV needs at least 3" + ] + + +async def test_services(driver: Any) -> None: + """Test the capabilities and motion zones services.""" + camera = _camera(driver, sensors=SENSORS) + assert _char(camera, "CameraCapabilities", "Version").value == "17.99" + capabilities = _char(camera, "CameraCapabilities", "CameraCapabilities") + assert base64_to_bytes(capabilities.value) == ( + hksv3.CameraCapabilities(SENSORS).encode() + ) + assert _char(camera, "CameraMotionZones", "Version").value == "17.99" + assert _chars(camera, "MotionSensor") == [ + "MotionDetected", + "ContributingSensors", + "MotionEnabled", + ] + assert _char(camera, "MotionSensor", "MotionEnabled").value is True + + +async def test_recording_motion_sensor(driver: Any) -> None: + """Test recording and capabilities share one motion sensor.""" + camera = _camera(driver, recording=True, sensors=SENSORS) + motion = [s for s in camera.services if s.display_name == "MotionSensor"] + assert len(motion) == 1 + assert _chars(camera, "MotionSensor") == [ + "MotionDetected", + "StatusActive", + "ContributingSensors", + "MotionEnabled", + ] + + +def _contributing(camera: Camera) -> bytes: + return base64_to_bytes(_char(camera, "MotionSensor", "ContributingSensors").value) + + +@pytest.mark.parametrize("default", [None, []]) +async def test_contributing_sensors(driver: Any, default: list[UUID] | None) -> None: + """Test motion names the sensors that saw it.""" + camera = _camera(driver, sensors=SENSORS) + detected = _char(camera, "MotionSensor", "MotionDetected") + + camera.set_motion_detected(True, default) + assert detected.value is True + assert _contributing(camera) == hksv3.encode_contributing_sensors([MAIN]) + + camera.set_motion_detected(True, [MAIN, PACKAGE]) + assert _contributing(camera) == hksv3.encode_contributing_sensors([MAIN, PACKAGE]) + + camera.set_motion_detected(False, [PACKAGE]) + assert detected.value is False + assert _contributing(camera) == b"" + + +async def test_contributing_sensors_unknown( + driver: Any, caplog: pytest.LogCaptureFixture +) -> None: + """Test motion from a sensor that is not configured logs a warning.""" + camera = _camera(driver, sensors=SENSORS) + other = UUID(int=1) + caplog.clear() + with caplog.at_level(logging.WARNING): + camera.set_motion_detected(True, iter([MAIN, other])) + assert caplog.messages == [ + f"Cam: motion reported for sensors that are not configured: {other}" + ] + assert _contributing(camera) == hksv3.encode_contributing_sensors([MAIN, other]) + + +async def test_motion_enabled(driver: Any) -> None: + """Test turning motion detection off clears and suppresses motion.""" + camera = _camera(driver, sensors=SENSORS) + camera.motion_enabled_changed = MagicMock() + detected = _char(camera, "MotionSensor", "MotionDetected") + enabled = _char(camera, "MotionSensor", "MotionEnabled") + camera.set_motion_detected(True) + + enabled.client_update_value(False) + camera.motion_enabled_changed.assert_called_once_with(False) + assert camera.motion_enabled is False + assert detected.value is False + camera.set_motion_detected(True) + assert detected.value is False + assert _contributing(camera) == b"" + + enabled.client_update_value(True) + camera.motion_enabled_changed.assert_called_with(True) + camera.set_motion_detected(True) + assert detected.value is True + + +async def test_motion_zones(driver: Any) -> None: + """Test the controller writing and clearing motion zones.""" + camera = _camera(driver, sensors=SENSORS) + camera.motion_zones_changed = MagicMock() + zones = _char(camera, "CameraMotionZones", "CameraZones") + + _char(camera, "CameraMotionZones", "Active").client_update_value(1) + assert camera.motion_zones_active is True + zones.client_update_value(to_base64_str(ZONES.encode())) + assert camera.camera_zones == ZONES + assert camera.motion_zones_changed.call_count == 2 + + zones.client_update_value("") + assert camera.camera_zones is None + assert camera.motion_zones_changed.call_count == 3 + + +@pytest.mark.parametrize("value", ["AQ==", "!!"]) +@pytest.mark.parametrize( + ("previous", "kept"), [(None, ""), (ZONES, to_base64_str(ZONES.encode()))] +) +async def test_motion_zones_malformed( + driver: Any, value: str, previous: hksv3.CameraZones | None, kept: str +) -> None: + """Test a malformed zones write fails and keeps the previous zones.""" + camera = _camera(driver, sensors=SENSORS) + _char(camera, "CameraMotionZones", "Active").client_update_value(1) + zones = _char(camera, "CameraMotionZones", "CameraZones") + zones.client_update_value(kept) + camera.motion_zones_changed = MagicMock() + with pytest.raises(ValueError): + zones.client_update_value(value) + camera.motion_zones_changed.assert_not_called() + assert zones.value == kept + assert camera.camera_zones == previous + assert camera.motion_zones_active is True + + +async def test_motion_zones_damaged_state(driver: Any) -> None: + """Test stored zones that do not decode count as no zones.""" + driver.accessory_state(CAMERA_AID)[HKSV3_STATE] = {"camera_zones": "AQ=="} + camera = _camera(driver, sensors=SENSORS) + assert _char(camera, "CameraMotionZones", "CameraZones").value == "AQ==" + assert camera.camera_zones is None + + +async def test_motion_zones_malformed_write(driver: Any) -> None: + """Test a controller reading the zones after a failed write gets the old ones.""" + camera = Camera({"address": "127.0.0.1", "sensors": SENSORS}, driver, "Cam") + driver.add_accessory(camera) + zones = _char(camera, "CameraMotionZones", "CameraZones") + iid = camera.iid_manager.get_iid(zones) + kept = to_base64_str(ZONES.encode()) + zones.client_update_value(kept) + + response = driver.set_characteristics( + { + HAP_REPR_CHARS: [ + {HAP_REPR_AID: 1, HAP_REPR_IID: iid, HAP_REPR_VALUE: "AQ=="} + ] + }, + ("127.0.0.1", 5000), + ) + assert response is not None + assert response[HAP_REPR_CHARS][0][HAP_REPR_STATUS] == ( + HAP_SERVER_STATUS.SERVICE_COMMUNICATION_FAILURE + ) + read = driver.get_characteristics([f"1.{iid}"]) + assert read[HAP_REPR_CHARS][0][HAP_REPR_VALUE] == kept + + +async def test_settings_survive_restart(driver: Any) -> None: + """Test zones and motion settings are restored.""" + camera = _camera(driver, sensors=SENSORS) + _char(camera, "CameraMotionZones", "Active").client_update_value(1) + _char(camera, "CameraMotionZones", "CameraZones").client_update_value( + to_base64_str(ZONES.encode()) + ) + _char(camera, "MotionSensor", "MotionEnabled").client_update_value(False) + + camera = _camera(driver, sensors=SENSORS) + assert camera.motion_zones_active is True + assert camera.camera_zones == ZONES + assert camera.motion_enabled is False + + +async def test_reset_calls_hooks(driver: Any) -> None: + """Test removing the last pairing tells the integration what changed.""" + camera = _camera(driver, sensors=SENSORS) + _char(camera, "CameraMotionZones", "Active").client_update_value(1) + _char(camera, "CameraMotionZones", "CameraZones").client_update_value( + to_base64_str(ZONES.encode()) + ) + _char(camera, "MotionSensor", "MotionEnabled").client_update_value(False) + camera.motion_zones_changed = MagicMock() + camera.motion_enabled_changed = MagicMock() + + camera.reset_accessory_state() + camera.motion_zones_changed.assert_called_once_with() + camera.motion_enabled_changed.assert_called_once_with(True) + assert camera.camera_zones is None + assert camera.motion_zones_active is False + assert camera.motion_enabled is True + + +async def test_reset_unchanged_calls_no_hooks(driver: Any) -> None: + """Test a reset without stored settings calls no hooks.""" + camera = _camera(driver, sensors=SENSORS) + camera.motion_zones_changed = MagicMock() + camera.motion_enabled_changed = MagicMock() + + camera.reset_accessory_state() + camera.motion_zones_changed.assert_not_called() + camera.motion_enabled_changed.assert_not_called() + + +@pytest.mark.parametrize( + ("service", "name", "value", "hook", "args"), + [ + ("CameraMotionZones", "Active", 1, "motion_zones_changed", ()), + ( + "CameraMotionZones", + "CameraZones", + to_base64_str(ZONES.encode()), + "motion_zones_changed", + (), + ), + ("MotionSensor", "MotionEnabled", False, "motion_enabled_changed", (False,)), + ], +) +async def test_repeated_write_calls_hook_once( + driver: Any, service: str, name: str, value: Any, hook: str, args: tuple[Any, ...] +) -> None: + """Test a hub writing the same value again does not call the hook again.""" + camera = _camera(driver, sensors=SENSORS) + mock = MagicMock() + setattr(camera, hook, mock) + char = _char(camera, service, name) + + char.client_update_value(value) + char.client_update_value(value) + char.client_update_value(value) + mock.assert_called_once_with(*args) + + +@pytest.mark.parametrize( + ("service", "name", "value", "hook"), + [ + ("CameraMotionZones", "Active", 0, "motion_zones_changed"), + ("MotionSensor", "MotionEnabled", True, "motion_enabled_changed"), + ], +) +async def test_write_of_current_value_calls_no_hook( + driver: Any, service: str, name: str, value: Any, hook: str +) -> None: + """Test writing the value the camera already has is no change.""" + camera = _camera(driver, sensors=SENSORS) + mock = MagicMock() + setattr(camera, hook, mock) + + _char(camera, service, name).client_update_value(value) + mock.assert_not_called() + + +async def test_reset_hooks_see_reset_recording(driver: Any) -> None: + """Test the hooks run after the recording state is reset as well.""" + camera = _camera(driver, sensors=SENSORS, recording=True) + _char(camera, "CameraRecordingManagement", "Active").client_update_value(1) + _char(camera, "CameraMotionZones", "Active").client_update_value(1) + seen: list[bool] = [] + camera.motion_zones_changed = lambda: seen.append(camera.recording_active) + + camera.reset_accessory_state() + assert seen == [False] diff --git a/tests/test_hksv3_state.py b/tests/test_hksv3_state.py new file mode 100644 index 00000000..e2ff1dd1 --- /dev/null +++ b/tests/test_hksv3_state.py @@ -0,0 +1,158 @@ +"""Tests for persisting the HKSV 17.99 camera state.""" + +import asyncio +from collections.abc import AsyncIterator +import logging +import pathlib +from typing import Any +from unittest.mock import MagicMock, call, patch + +import pytest + +from pyhap.accessory_driver import AccessoryDriver +from pyhap.camera import HKSV3_STATE, Camera + +CAMERA_AID = 2 + + +@pytest.fixture(name="driver") +async def driver_fixture(tmp_path: pathlib.Path) -> AsyncIterator[AccessoryDriver]: + with patch("pyhap.accessory_driver.AccessoryDriver.async_persist"): + yield AccessoryDriver( + loop=asyncio.get_running_loop(), + listen_address="127.0.0.1", + persist_file=str(tmp_path / "accessory.state"), + ) + + +async def _drain() -> None: + # add_job hops through call_soon_threadsafe, then call_soon. + for _ in range(2): + await asyncio.sleep(0) + + +def _camera( + driver: Any, aid: int | None = CAMERA_AID, handler: Any = None +) -> tuple[Camera, Any, Any]: + camera = Camera({"address": "127.0.0.1"}, driver, "Cam", aid=aid) + light = camera.add_preload_service("Lightbulb", chars=["Brightness"]) + on, brightness = ( + light.get_characteristic("On"), + light.get_characteristic("Brightness"), + ) + handler = handler or MagicMock() + camera._persist_hksv3_char(on, "on", handler) + camera._persist_hksv3_char(brightness, "brightness", handler) + camera._restore_hksv3_state() + return camera, on, brightness + + +async def test_write_is_stored_and_restored(driver: Any) -> None: + """Test a controller write survives rebuilding the camera.""" + _, on, brightness = _camera(driver) + on.client_update_value(True) + brightness.client_update_value(40) + await _drain() + assert driver.accessory_state(CAMERA_AID)[HKSV3_STATE] == { + "on": True, + "brightness": 40, + } + # Both writes share one save. + driver.async_persist.assert_called_once() + + _, on, brightness = _camera(driver) + assert on.value is True + assert brightness.value == 40 + + +async def test_restore_skips_invalid_values( + driver: Any, caplog: pytest.LogCaptureFixture +) -> None: + """Test a damaged state entry keeps the default.""" + driver.accessory_state(CAMERA_AID)[HKSV3_STATE] = {"brightness": "bright"} + with caplog.at_level(logging.WARNING): + _, on, brightness = _camera(driver) + assert on.value is False + assert brightness.value == 0 + assert "Ignoring stored brightness" in caplog.text + + +async def test_restore_ignores_other_state(driver: Any) -> None: + """Test a state entry that is not a dict is ignored.""" + driver.accessory_state(CAMERA_AID)[HKSV3_STATE] = ["on"] + _, on, _ = _camera(driver) + assert on.value is False + + +async def test_write_replaces_other_state(driver: Any) -> None: + """Test a write still works when the state entry is not a dict.""" + driver.accessory_state(CAMERA_AID)[HKSV3_STATE] = ["on"] + _, on, _ = _camera(driver) + on.client_update_value(True) + assert driver.accessory_state(CAMERA_AID)[HKSV3_STATE] == {"on": True} + + +async def test_restore_accessory_state(driver: Any) -> None: + """Test the driver restoring state loaded after construction.""" + camera, on, _ = _camera(driver) + driver.accessory_state(CAMERA_AID)[HKSV3_STATE] = {"on": True} + camera.restore_accessory_state() + assert on.value is True + + +async def test_reset_returns_defaults(driver: Any) -> None: + """Test removing the last pairing forgets the stored state.""" + camera, on, brightness = _camera(driver) + on.client_update_value(True) + brightness.client_update_value(40) + await _drain() + driver.async_persist.reset_mock() + + camera.reset_accessory_state() + await _drain() + assert on.value is False + assert brightness.value == 0 + assert HKSV3_STATE not in driver.accessory_state(CAMERA_AID) + driver.async_persist.assert_called_once() + + +@pytest.mark.parametrize( + ("on_value", "brightness_value", "calls"), + [ + (True, 40, [call(False)]), + (False, 40, [call(0)]), + (False, 0, []), + ], +) +async def test_reset_calls_handler( + driver: Any, on_value: bool, brightness_value: int, calls: list[Any] +) -> None: + """Test a reset calls each handler once, only for changed values.""" + handler = MagicMock() + camera, on, brightness = _camera(driver, handler=handler) + on.set_value(on_value) + brightness.set_value(brightness_value) + + camera.reset_accessory_state() + assert handler.call_args_list == calls + + +async def test_without_aid_nothing_is_stored(driver: Any) -> None: + """Test a camera without aid neither stores nor restores.""" + camera, on, _ = _camera(driver, aid=None) + on.client_update_value(True) + camera.reset_accessory_state() + await _drain() + assert driver.state.accessory_states == {} + driver.async_persist.assert_not_called() + + +async def test_save_without_loop(driver: Any, caplog: pytest.LogCaptureFixture) -> None: + """Test a closed loop only logs.""" + _, on, _ = _camera(driver) + with ( + patch.object(driver, "add_job", side_effect=RuntimeError), + caplog.at_level(logging.WARNING), + ): + on.client_update_value(True) + assert "Could not schedule saving the camera state" in caplog.text From 3dea2d51911e9fc6a29ddb79b4a3b0a5573257a1 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Sun, 8 Nov 2026 14:54:16 +0100 Subject: [PATCH 5/8] Add the HKSV 17.99 camera global operating mode A new "global_operating_mode" camera option adds the Camera Global Operating Mode service with the listed optional characteristics, after all other services so their instance IDs stay the same. The settings a controller writes are persisted and reset for a new home; changes from writes and resets call global_operating_mode_changed with the setting and its value. Manually Disabled and Third Party Camera Active are device state, set through set_manually_disabled and set_third_party_camera_active. streaming_allowed combines HomeKit Camera Active, Streaming Enabled and Manually Disabled. While it is False, the camera answers SetupEndpoints with an error and does not start streams. Cameras without the option are unchanged. When the service exists, its HomeKit Camera Active alone turns the camera on or off. The value on the Camera Operating Mode service of classic recording is then ignored, so a value left there cannot keep the camera off. A change closes a running recording and calls homekit_camera_active_changed like before; writing the same value again calls no hook. --- pyhap/camera.py | 176 +++++++++++-- tests/test_camera_operating_mode.py | 388 ++++++++++++++++++++++++++++ 2 files changed, 540 insertions(+), 24 deletions(-) create mode 100644 tests/test_camera_operating_mode.py diff --git a/pyhap/camera.py b/pyhap/camera.py index e898c7b3..cfc0afa8 100644 --- a/pyhap/camera.py +++ b/pyhap/camera.py @@ -261,6 +261,13 @@ "SupportedVideoRecordingConfiguration", "SupportedAudioRecordingConfiguration", ) +# Global operating mode chars a controller writes, the camera keeps them. +PERSISTED_GLOBAL_OPERATING_MODE_CHARS = ( + "HomeKitCameraActive", + "StreamingEnabled", + "CameraOperatingModeIndicator", + "NightVision", +) class Camera(Accessory): @@ -468,6 +475,16 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None: service is added so that HAP clients know the accessory supports it, and :meth:`talkback_audio_received` is called with the received audio for each session that negotiates it. + - global_operating_mode - list of optional characteristics of the + HomeKit Secure Video 17.99 Camera Global Operating Mode service, + e.g. ``["ManuallyDisabled", "NightVision"]``. Opts into the service, + an empty list adds only the required characteristics. Its + HomeKit Camera Active alone turns the camera on or off: the value + of the Camera Operating Mode service that ``recording`` adds is + then ignored for :attr:`homekit_camera_active`, recording and + :attr:`streaming_allowed`. While streaming is not allowed, the + camera refuses to set up live streams. Unknown names raise + ``ValueError``. - start_stream_cmd - string specifying the command to be executed to start the stream. The string can contain the keywords, corresponding to the video and audio configuration that was negotiated between the camera @@ -523,8 +540,9 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None: self._hds_listener = None self._hds_connections = set() self._recording_stream = None - self._homekit_camera_active_char = None + self._homekit_camera_active_char: Characteristic | None = None self._motion_status_active_char = None + self._global_operating_mode: Service | None = None # Persisted HKSV 17.99 chars with their state key, default and the # handler a reset calls. self._hksv3_defaults: dict[ @@ -545,6 +563,9 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None: self._setup_recording_management(options) if self._sensors: self._setup_camera_capabilities() + # Last, so the instance IDs of the other services stay the same. + if options.get("global_operating_mode") is not None: + self._setup_global_operating_mode(options["global_operating_mode"]) self._restore_hksv3_state() @property @@ -653,11 +674,11 @@ def _setup_recording_management(self, options: dict[str, Any]) -> None: "CameraOperatingMode", chars=["PeriodicSnapshotsActive"] ) operating_mode.configure_char("EventSnapshotsActive", value=1) - self._homekit_camera_active_char = operating_mode.configure_char( - "HomeKitCameraActive", - value=1, - setter_callback=self._handle_homekit_camera_active, + homekit_camera_active = operating_mode.configure_char( + "HomeKitCameraActive", value=1 ) + homekit_camera_active.setter_callback = self._handle_homekit_camera_active + self._homekit_camera_active_char = homekit_camera_active operating_mode.configure_char("PeriodicSnapshotsActive", value=1) # HKSV records on an event trigger; the camera owns a motion sensor and @@ -941,16 +962,104 @@ def recording_audio_active(self) -> bool: self._recording_service.get_characteristic("RecordingAudioActive").value ) + def _setup_global_operating_mode(self, chars: list[str]) -> None: + """Create the Camera Global Operating Mode service (HKSV 17.99).""" + if not isinstance(chars, (list, tuple)): + raise ValueError( + "global_operating_mode must be a list of characteristic names" + ) + optional = self.driver.loader.serv_types["CameraGlobalOperatingMode"][ + "OptionalCharacteristics" + ] + if unknown := [name for name in chars if name not in optional]: + raise ValueError( + "Not an optional Camera Global Operating Mode characteristic: " + + ", ".join(unknown) + ) + service = self.add_preload_service("CameraGlobalOperatingMode", chars=chars) + for char in service.characteristics: + name = char.display_name + # Everything starts on except Night Vision and Manually Disabled. + if name not in ("NightVision", "ManuallyDisabled"): + char.set_value(1, should_notify=False) + if name not in PERSISTED_GLOBAL_OPERATING_MODE_CHARS: + continue + if name == "HomeKitCameraActive": + handler = self._handle_homekit_camera_active + # This value alone decides, the classic one is then ignored. + if self._homekit_camera_active_char is not None: + self._homekit_camera_active_char.setter_callback = None + self._homekit_camera_active_char = char + else: + handler = functools.partial(self._handle_global_operating_mode, name) + self._persist_hksv3_char(char, f"global_operating_mode.{name}", handler) + self._global_operating_mode = service + + @property + def streaming_allowed(self) -> bool: + """Whether the global operating mode allows streaming. + + True without the ``global_operating_mode`` option. While it is False, + the camera refuses to set up live streams. + """ + service = self._global_operating_mode + if service is None: + return True + manually_disabled = self._global_operating_mode_char("ManuallyDisabled") + return ( + self.homekit_camera_active + and bool(service.get_characteristic("StreamingEnabled").value) + and not (manually_disabled is not None and manually_disabled.value) + ) + + def set_manually_disabled(self, disabled: bool) -> None: + """Report the camera being turned off on the device itself. + + Needs ``ManuallyDisabled`` in the ``global_operating_mode`` option. + """ + char = self._global_operating_mode_char("ManuallyDisabled") + if char is not None: + char.set_value(bool(disabled)) + + def set_third_party_camera_active(self, active: bool) -> None: + """Report whether the camera is on for apps outside HomeKit. + + Needs ``ThirdPartyCameraActive`` in the ``global_operating_mode`` option. + """ + char = self._global_operating_mode_char("ThirdPartyCameraActive") + if char is not None: + char.set_value(bool(active)) + + def _global_operating_mode_char(self, name: str) -> Characteristic | None: + if self._global_operating_mode is None: + return None + try: + return self._global_operating_mode.get_characteristic(name) + except ValueError: + return None + + def _handle_global_operating_mode(self, name: str, value: Any) -> None: + self.global_operating_mode_changed(name, value) + + def global_operating_mode_changed(self, name: str, value: Any) -> None: + """React to the controller changing a global operating mode setting. + + ``name`` is the characteristic, e.g. ``"NightVision"``. HomeKit Camera + Active changes call :meth:`homekit_camera_active_changed` instead. Only + changes call it, not a controller writing the same value again. + Override. + """ + @property def homekit_camera_active(self) -> bool: """Whether the camera is on in the Home app. - Only cameras with recording have the setting, all others are on. + Only cameras with recording or the global operating mode have the + setting, all others are on. With the global operating mode only its + value counts. """ - operating_mode = self.get_service("CameraOperatingMode") - if operating_mode is None: - return True - return bool(operating_mode.get_characteristic("HomeKitCameraActive").value) + char = self._homekit_camera_active_char + return char is None or bool(char.value) def set_selected_recording_configuration(self, value: str) -> None: """Handle a write to Selected Camera Recording Configuration (spec HKSV).""" @@ -1307,18 +1416,24 @@ async def _start_stream(self, objs: dict[bytes, Any], reconfigure: bool) -> None stream_idx = session_info["stream_idx"] opts.update(session_info) - success = ( - await self.reconfigure_stream(session_info, opts) - if reconfigure - else await self.start_stream(session_info, opts) - ) + # Turned off after SetupEndpoints. + refused = not reconfigure and not self.streaming_allowed + if refused: + logger.warning("[%s] Streaming is turned off, refusing it", session_id) + success = False + elif reconfigure: + success = await self.reconfigure_stream(session_info, opts) + else: + success = await self.start_stream(session_info, opts) if success: self._streaming_status[stream_idx] = STREAMING_STATUS["STREAMING"] else: - logger.error( - "[%s] Failed to start/reconfigure stream, deleting session.", session_id - ) + if not refused: + logger.error( + "[%s] Failed to start/reconfigure stream, deleting session.", + session_id, + ) self._close_talkback(session_info) del self.sessions[session_id] self._streaming_status[stream_idx] = STREAMING_STATUS["AVAILABLE"] @@ -1338,11 +1453,11 @@ def _handle_homekit_camera_active(self, value: Any) -> None: # A camera turned off must not record, and the hub must not treat its # motion sensor as an active trigger. active = bool(value) - assert self._motion_status_active_char is not None # noqa: S101 - # Always notify: a restored value may have bypassed the stored one, - # so set_value alone could see no change. - self._motion_status_active_char.set_value(active, should_notify=False) - self._motion_status_active_char.notify() + if self._motion_status_active_char is not None: + # Always notify: a restored value may have bypassed the stored one, + # so set_value alone could see no change. + self._motion_status_active_char.set_value(active, should_notify=False) + self._motion_status_active_char.notify() if not active and self._recording_stream is not None: self._recording_stream.close(hds_recording.RecordingReason.NOT_ALLOWED) self.homekit_camera_active_changed(active) @@ -1353,7 +1468,8 @@ def homekit_camera_active_changed(self, active: bool) -> None: Override instead of replacing the characteristic's setter. A running recording is already closed when the camera is turned off. Like other setters, every controller write calls this, setting the value locally - does not. + does not. With the ``global_operating_mode`` option only changes call + it, not a controller writing the same value again. """ async def _stop_stream(self, objs: dict[bytes, Any]) -> None: @@ -1476,6 +1592,18 @@ def set_endpoints(self, value: str, stream_idx: int | None = None) -> None: objs = tlv.decode(value, from_base64=True) session_id = UUID(bytes=objs[SETUP_TYPES["SESSION_ID"]]) + if not self.streaming_allowed: + logger.warning("[%s] Streaming is turned off, refusing it", session_id) + self._management[stream_idx].get_characteristic("SetupEndpoints").set_value( + tlv.encode( + SETUP_TYPES["SESSION_ID"], + session_id.bytes, + SETUP_TYPES["STATUS"], + SETUP_STATUS["ERROR"], + to_base64=True, + ) + ) + return # Extract address info address_tlv = objs[SETUP_TYPES["ADDRESS"]] diff --git a/tests/test_camera_operating_mode.py b/tests/test_camera_operating_mode.py new file mode 100644 index 00000000..19cc7328 --- /dev/null +++ b/tests/test_camera_operating_mode.py @@ -0,0 +1,388 @@ +"""Tests for the HKSV 17.99 global operating mode of pyhap.camera.""" + +import asyncio +from collections.abc import AsyncIterator +import pathlib +from typing import Any +from unittest.mock import AsyncMock, MagicMock, call, patch +from uuid import UUID + +import pytest + +from pyhap import camera as camera_module, hds_recording, tlv +from pyhap.accessory_driver import AccessoryDriver +from pyhap.camera import SETUP_STATUS, SETUP_TYPES, STREAMING_STATUS, Camera +from pyhap.characteristic import PROP_PERMISSIONS + +CAMERA_AID = 2 +GLOBAL = "CameraGlobalOperatingMode" +OPTIONAL = ["ManuallyDisabled", "NightVision", "ThirdPartyCameraActive"] +VIDEO = { + "codec": { + "profiles": [camera_module.VIDEO_CODEC_PARAM_PROFILE_ID_TYPES["MAIN"]], + "levels": [camera_module.VIDEO_CODEC_PARAM_LEVEL_TYPES["TYPE4_0"]], + }, + "resolutions": [[640, 360, 30]], +} +AUDIO = {"codecs": [{"type": "OPUS", "samplerate": 24}]} +# A stream start request for SESSION_ID. +START_REQUEST = ( + "ARUCAQEBEKzMbMEFY0UVjal0tFCQBpECNAEBAAIJAQEAAgEAAwEAAwsBAoAC" + "AgJoAQMBHgQXAQFjAgQr66FSAwKEAAQEAAAAPwUCYgUDLAEBAgIMAQEBAgEA" + "AwEBBAEeAxYBAW4CBMUInmQDAhgABAQAAKBABgENBAEA" +) +SESSION_ID = UUID("accc6cc1-0563-4515-8da9-74b450900691") +# A SetupEndpoints request for SESSION_ID. +SETUP_REQUEST = ( + "ARCszGzBBWNFFY2pdLRQkAaRAxoBAQACDTE5Mi4xNjguMS4xMTQDAjPFBAKs1gQ" + "lAhDYlmCkyTBZQfxqFS3OnxVOAw4bQZm5NuoQjyanlqWA0QEBAAUlAhAKRPSRVa" + "qGeNmESTIojxNiAw78WkjTLtGv0waWnLo9gQEBAA==" +) + + +@pytest.fixture(name="driver") +async def driver_fixture(tmp_path: pathlib.Path) -> AsyncIterator[AccessoryDriver]: + with patch("pyhap.accessory_driver.AccessoryDriver.async_persist"): + yield AccessoryDriver( + loop=asyncio.get_running_loop(), + listen_address="127.0.0.1", + persist_file=str(tmp_path / "accessory.state"), + ) + + +def _camera(driver: Any, **options: Any) -> Camera: + return Camera({"address": "127.0.0.1", **options}, driver, "Cam", aid=CAMERA_AID) + + +def _char(camera: Camera, service: str, name: str) -> Any: + found = camera.get_service(service) + assert found is not None + return found.get_characteristic(name) + + +def _values(camera: Camera) -> dict[str | None, Any]: + service = camera.get_service(GLOBAL) + assert service is not None + return {char.display_name: char.value for char in service.characteristics} + + +async def test_without_option(driver: Any) -> None: + """Test cameras without the option are unchanged.""" + camera = _camera(driver) + assert camera.get_service(GLOBAL) is None + assert camera.homekit_camera_active is True + assert camera.streaming_allowed is True + camera.set_manually_disabled(True) + camera.set_third_party_camera_active(False) + + +async def test_required_chars(driver: Any) -> None: + """Test an empty list adds the required chars, all on.""" + camera = _camera(driver, global_operating_mode=[]) + assert _values(camera) == { + "HomeKitCameraActive": 1, + "StreamingEnabled": True, + "CameraOperatingModeIndicator": True, + } + assert camera.streaming_allowed is True + camera.set_manually_disabled(True) + camera.set_third_party_camera_active(False) + assert camera.streaming_allowed is True + + +async def test_optional_chars(driver: Any) -> None: + """Test optional chars and their defaults.""" + camera = _camera(driver, global_operating_mode=OPTIONAL) + assert _values(camera) == { + "HomeKitCameraActive": 1, + "StreamingEnabled": True, + "CameraOperatingModeIndicator": True, + "ManuallyDisabled": False, + "NightVision": False, + "ThirdPartyCameraActive": True, + } + + +async def test_manually_disabled(driver: Any) -> None: + """Test the device turning the camera off stops streaming.""" + camera = _camera(driver, global_operating_mode=OPTIONAL) + camera.set_manually_disabled(True) + assert _char(camera, GLOBAL, "ManuallyDisabled").value is True + assert camera.streaming_allowed is False + assert camera.homekit_camera_active is True + camera.set_manually_disabled(False) + assert camera.streaming_allowed is True + + +async def test_third_party_camera_active(driver: Any) -> None: + """Test the device reports Third Party Camera Active, controllers only read it.""" + camera = _camera(driver, global_operating_mode=OPTIONAL) + camera.global_operating_mode_changed = MagicMock() + char = _char(camera, GLOBAL, "ThirdPartyCameraActive") + assert char.properties[PROP_PERMISSIONS] == ["pr", "ev"] + assert char.setter_callback is None + camera.set_third_party_camera_active(False) + assert char.value is False + assert camera.streaming_allowed is True + camera.global_operating_mode_changed.assert_not_called() + + +@pytest.mark.parametrize( + ("name", "value", "streaming_allowed"), + [ + ("StreamingEnabled", False, False), + ("CameraOperatingModeIndicator", False, True), + ("NightVision", True, True), + ], +) +async def test_writes_call_hook( + driver: Any, name: str, value: bool, streaming_allowed: bool +) -> None: + """Test controller writes reach the hook with the setting and its value.""" + camera = _camera(driver, global_operating_mode=OPTIONAL) + camera.global_operating_mode_changed = MagicMock() + camera.homekit_camera_active_changed = MagicMock() + _char(camera, GLOBAL, name).client_update_value(value) + camera.global_operating_mode_changed.assert_called_once_with(name, value) + camera.homekit_camera_active_changed.assert_not_called() + assert camera.streaming_allowed is streaming_allowed + + +@pytest.mark.parametrize( + ("enabled", "starts", "status", "warnings"), + [ + (True, 1, STREAMING_STATUS["STREAMING"], 0), + (False, 0, STREAMING_STATUS["AVAILABLE"], 1), + ], +) +async def test_stream_start_needs_streaming_allowed( + driver: Any, + enabled: bool, + starts: int, + status: bytes, + warnings: int, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test a stream turned off after its setup does not start.""" + camera = _camera(driver, video=VIDEO, audio=AUDIO, global_operating_mode=[]) + camera.start_stream = AsyncMock(return_value=True) + camera.sessions[SESSION_ID] = {"id": SESSION_ID, "stream_idx": 0} + _char(camera, GLOBAL, "StreamingEnabled").client_update_value(enabled) + + await camera._start_stream( + tlv.decode(START_REQUEST, from_base64=True), reconfigure=False + ) + assert camera.start_stream.await_count == starts + assert camera.streaming_status == status + assert (SESSION_ID in camera.sessions) is enabled + assert caplog.text.count("Streaming is turned off") == warnings + assert "Failed to start/reconfigure stream" not in caplog.text + + +@pytest.mark.parametrize( + ("options", "service", "enabled", "status", "sessions"), + [ + ( + {"global_operating_mode": []}, + GLOBAL, + True, + SETUP_STATUS["SUCCESS"], + [SESSION_ID], + ), + ({"global_operating_mode": []}, GLOBAL, False, SETUP_STATUS["ERROR"], []), + ( + {"recording": True}, + "CameraOperatingMode", + False, + SETUP_STATUS["SUCCESS"], + [SESSION_ID], + ), + ], +) +async def test_setup_endpoints_needs_streaming_allowed( + driver: Any, + options: dict[str, Any], + service: str, + enabled: bool, + status: bytes, + sessions: list[UUID], +) -> None: + """Test the camera refuses a stream setup while streaming is turned off. + + Without the global operating mode, turning the camera off does not + refuse streams, as before. + """ + camera = _camera(driver, video=VIDEO, audio=AUDIO, **options) + _char(camera, service, "HomeKitCameraActive").client_update_value(enabled) + setup = _char(camera, "CameraRTPStreamManagement", "SetupEndpoints") + setup.client_update_value(SETUP_REQUEST) + + response = tlv.decode(setup.get_value(), from_base64=True) + assert response[SETUP_TYPES["SESSION_ID"]] == SESSION_ID.bytes + assert response[SETUP_TYPES["STATUS"]] == status + assert list(camera.sessions) == sessions + + +async def test_homekit_camera_active(driver: Any) -> None: + """Test turning the camera off without recording.""" + camera = _camera(driver, global_operating_mode=[]) + camera.homekit_camera_active_changed = MagicMock() + _char(camera, GLOBAL, "HomeKitCameraActive").client_update_value(0) + camera.homekit_camera_active_changed.assert_called_once_with(False) + assert camera.homekit_camera_active is False + assert camera.streaming_allowed is False + + +async def test_global_operating_mode_decides(driver: Any) -> None: + """Test the global operating mode turns a recording camera off.""" + camera = _camera(driver, recording=True, global_operating_mode=[]) + camera.homekit_camera_active_changed = MagicMock() + stream = MagicMock() + camera._recording_stream = stream + status_active = _char(camera, "MotionSensor", "StatusActive") + + _char(camera, GLOBAL, "HomeKitCameraActive").client_update_value(0) + camera.homekit_camera_active_changed.assert_called_once_with(False) + assert camera.homekit_camera_active is False + assert status_active.get_value() is False + stream.close.assert_called_once_with(hds_recording.RecordingReason.NOT_ALLOWED) + + +async def test_operating_mode_ignored(driver: Any) -> None: + """Test the Camera Operating Mode value is ignored next to the global one.""" + camera = _camera(driver, recording=True, global_operating_mode=[]) + camera.homekit_camera_active_changed = MagicMock() + stream = MagicMock() + camera._recording_stream = stream + status_active = _char(camera, "MotionSensor", "StatusActive") + active = _char(camera, GLOBAL, "HomeKitCameraActive") + legacy = _char(camera, "CameraOperatingMode", "HomeKitCameraActive") + + # A value left from before must not keep the camera off. + active.client_update_value(0) + legacy.client_update_value(0) + active.client_update_value(1) + assert camera.homekit_camera_active_changed.call_count == 2 + camera.homekit_camera_active_changed.assert_called_with(True) + assert camera.homekit_camera_active is True + assert camera.streaming_allowed is True + assert status_active.get_value() is True + + legacy.client_update_value(1) + legacy.client_update_value(0) + assert camera.homekit_camera_active_changed.call_count == 2 + assert camera.homekit_camera_active is True + stream.close.assert_called_once() + + +async def test_settings_survive_restart(driver: Any) -> None: + """Test controller writes are restored, the device state is not.""" + camera = _camera(driver, global_operating_mode=OPTIONAL) + _char(camera, GLOBAL, "HomeKitCameraActive").client_update_value(0) + _char(camera, GLOBAL, "StreamingEnabled").client_update_value(False) + _char(camera, GLOBAL, "CameraOperatingModeIndicator").client_update_value(False) + _char(camera, GLOBAL, "NightVision").client_update_value(True) + camera.set_manually_disabled(True) + camera.set_third_party_camera_active(False) + + camera = _camera(driver, global_operating_mode=OPTIONAL) + assert _values(camera) == { + "HomeKitCameraActive": 0, + "StreamingEnabled": False, + "CameraOperatingModeIndicator": False, + "ManuallyDisabled": False, + "NightVision": True, + "ThirdPartyCameraActive": True, + } + camera.global_operating_mode_changed = MagicMock() + camera.homekit_camera_active_changed = MagicMock() + + camera.reset_accessory_state() + camera.homekit_camera_active_changed.assert_called_once_with(True) + assert camera.global_operating_mode_changed.call_args_list == [ + call("StreamingEnabled", True), + call("CameraOperatingModeIndicator", True), + call("NightVision", False), + ] + assert _values(camera) == { + "HomeKitCameraActive": 1, + "StreamingEnabled": True, + "CameraOperatingModeIndicator": True, + "ManuallyDisabled": False, + "NightVision": False, + "ThirdPartyCameraActive": True, + } + + +def _instance_ids( + camera: Camera, +) -> dict[tuple[str | None, str | None], int | None]: + return { + (service.display_name, char.display_name): camera.iid_manager.get_iid(char) + for service in camera.services + for char in service.characteristics + } + + +async def test_instance_ids_unchanged(driver: Any) -> None: + """Test turning the option on keeps the instance IDs of the other services.""" + without = _instance_ids(_camera(driver, recording=True, video=VIDEO, audio=AUDIO)) + with_option = _instance_ids( + _camera( + driver, recording=True, video=VIDEO, audio=AUDIO, global_operating_mode=[] + ) + ) + assert without.items() <= with_option.items() + + +@pytest.mark.parametrize( + ("name", "value", "hook", "args"), + [ + ( + "StreamingEnabled", + False, + "global_operating_mode_changed", + ("StreamingEnabled", False), + ), + ("NightVision", True, "global_operating_mode_changed", ("NightVision", True)), + ("HomeKitCameraActive", 0, "homekit_camera_active_changed", (False,)), + ], +) +async def test_repeated_write_calls_hook_once( + driver: Any, name: str, value: Any, hook: str, args: tuple[Any, ...] +) -> None: + """Test a hub writing the same value again does not call the hook again.""" + camera = _camera(driver, recording=True, global_operating_mode=OPTIONAL) + mock = MagicMock() + setattr(camera, hook, mock) + char = _char(camera, GLOBAL, name) + + char.client_update_value(value) + char.client_update_value(value) + char.client_update_value(value) + mock.assert_called_once_with(*args) + + +async def test_classic_writes_call_hook_every_time(driver: Any) -> None: + """Test cameras without the option keep calling the hook on every write.""" + camera = _camera(driver, recording=True) + camera.homekit_camera_active_changed = MagicMock() + char = _char(camera, "CameraOperatingMode", "HomeKitCameraActive") + + char.client_update_value(0) + char.client_update_value(0) + assert camera.homekit_camera_active_changed.call_args_list == [call(False)] * 2 + + +@pytest.mark.parametrize("chars", [["Brightness"], ["HomeKitCameraActive"]]) +async def test_unknown_chars(driver: Any, chars: list[str]) -> None: + """Test names that are not optional chars of the service raise.""" + with pytest.raises(ValueError, match=chars[0]): + _camera(driver, global_operating_mode=chars) + + +@pytest.mark.parametrize("chars", [True, "NightVision"]) +async def test_option_not_a_list(driver: Any, chars: Any) -> None: + """Test the option must be a list of names.""" + with pytest.raises(ValueError, match="must be a list"): + _camera(driver, global_operating_mode=chars) From a41aa489b2876fd21c6cf5bfaf2b12ab20f09fdf Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Tue, 10 Nov 2026 14:42:53 +0100 Subject: [PATCH 6/8] Say why hksv3 has its own TLV list codec --- pyhap/hksv3.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pyhap/hksv3.py b/pyhap/hksv3.py index a7d6a10c..7cf0d1d6 100644 --- a/pyhap/hksv3.py +++ b/pyhap/hksv3.py @@ -4,6 +4,10 @@ and items of the same type are separated by an empty TLV of type 0. Values longer than 255 bytes are sent as 255-byte fragments of the same type; a TLV continues the value before it only after such a full fragment. + +:mod:`pyhap.tlv` maps each type to one value, so it cannot express repeated +items of a type or the separators these lists need; hence the small encoder +and decoder here. """ from __future__ import annotations From 80a22f90656dae24065c52df253c33b5716c9c3c Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Tue, 10 Nov 2026 14:42:53 +0100 Subject: [PATCH 7/8] Honor both HomeKit Camera Active values and end streams when turned off With the global operating mode, the last write to either HomeKit Camera Active now decides and the other follows, so a hub that only knows the Camera Operating Mode service can still turn the camera off. Running live streams end once streaming is not allowed anymore: camera off, Streaming Enabled off or manually disabled. A global_operating_mode option that is not a list raises TypeError. --- pyhap/camera.py | 83 ++++++++++++++++++++------ tests/test_camera_operating_mode.py | 91 +++++++++++++++++++++++++---- 2 files changed, 143 insertions(+), 31 deletions(-) diff --git a/pyhap/camera.py b/pyhap/camera.py index 45225c5c..6cb3add5 100644 --- a/pyhap/camera.py +++ b/pyhap/camera.py @@ -515,13 +515,14 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None: - global_operating_mode - list of optional characteristics of the HomeKit Secure Video 17.99 Camera Global Operating Mode service, e.g. ``["ManuallyDisabled", "NightVision"]``. Opts into the service, - an empty list adds only the required characteristics. Its - HomeKit Camera Active alone turns the camera on or off: the value - of the Camera Operating Mode service that ``recording`` adds is - then ignored for :attr:`homekit_camera_active`, recording and - :attr:`streaming_allowed`. While streaming is not allowed, the - camera refuses to set up live streams. Unknown names raise - ``ValueError``. + an empty list adds only the required characteristics. With + ``recording``, the Camera Operating Mode service has a HomeKit + Camera Active too: the last write to either turns the camera on + or off, and the other one is set to the same value, so a hub + that only knows the classic one still works. While streaming is + not allowed, the camera refuses to set up live streams and ends + running ones. Unknown names raise ``ValueError``, a value that is + not a list raises ``TypeError``. - start_stream_cmd - string specifying the command to be executed to start the stream. The string can contain the keywords, corresponding to the video and audio configuration that was negotiated between the camera @@ -578,6 +579,9 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None: self._hds_connections = set() self._recording_stream = None self._homekit_camera_active_char: Characteristic | None = None + # The HomeKit Camera Active of the Camera Operating Mode service, kept + # in step with the global one when both exist. + self._classic_camera_active_char: Characteristic | None = None self._motion_status_active_char = None self._global_operating_mode: Service | None = None # Persisted HKSV 17.99 chars with their state key, default and the @@ -716,6 +720,7 @@ def _setup_recording_management(self, options: dict[str, Any]) -> None: ) homekit_camera_active.setter_callback = self._handle_homekit_camera_active self._homekit_camera_active_char = homekit_camera_active + self._classic_camera_active_char = homekit_camera_active operating_mode.configure_char("PeriodicSnapshotsActive", value=1) # HKSV records on an event trigger; the camera owns a motion sensor and @@ -1002,7 +1007,7 @@ def recording_audio_active(self) -> bool: def _setup_global_operating_mode(self, chars: list[str]) -> None: """Create the Camera Global Operating Mode service (HKSV 17.99).""" if not isinstance(chars, (list, tuple)): - raise ValueError( + raise TypeError( "global_operating_mode must be a list of characteristic names" ) optional = self.driver.loader.serv_types["CameraGlobalOperatingMode"][ @@ -1022,10 +1027,12 @@ def _setup_global_operating_mode(self, chars: list[str]) -> None: if name not in PERSISTED_GLOBAL_OPERATING_MODE_CHARS: continue if name == "HomeKitCameraActive": - handler = self._handle_homekit_camera_active - # This value alone decides, the classic one is then ignored. - if self._homekit_camera_active_char is not None: - self._homekit_camera_active_char.setter_callback = None + handler = self._handle_global_camera_active + # The last write to either HomeKit Camera Active decides. + if self._classic_camera_active_char is not None: + self._classic_camera_active_char.setter_callback = ( + self._handle_classic_camera_active + ) self._homekit_camera_active_char = char else: handler = functools.partial(self._handle_global_operating_mode, name) @@ -1057,6 +1064,7 @@ def set_manually_disabled(self, disabled: bool) -> None: char = self._global_operating_mode_char("ManuallyDisabled") if char is not None: char.set_value(bool(disabled)) + self._end_streams_if_not_allowed() def set_third_party_camera_active(self, active: bool) -> None: """Report whether the camera is on for apps outside HomeKit. @@ -1076,8 +1084,23 @@ def _global_operating_mode_char(self, name: str) -> Characteristic | None: return None def _handle_global_operating_mode(self, name: str, value: Any) -> None: + if name == "StreamingEnabled": + self._end_streams_if_not_allowed() self.global_operating_mode_changed(name, value) + def _end_streams_if_not_allowed(self) -> None: + """End the live streams once streaming is turned off.""" + if self.streaming_allowed: + return + for session_id in list(self.sessions): + self.driver.add_job(self._end_stream_not_allowed, session_id) + + async def _end_stream_not_allowed(self, session_id: UUID) -> None: + stream_idx = await self._end_stream(session_id) + if stream_idx is not None: + logger.info("[%s] Ending the stream, streaming is turned off", session_id) + self.set_streaming_available(stream_idx) + def global_operating_mode_changed(self, name: str, value: Any) -> None: """React to the controller changing a global operating mode setting. @@ -1092,8 +1115,7 @@ def homekit_camera_active(self) -> bool: """Whether the camera is on in the Home app. Only cameras with recording or the global operating mode have the - setting, all others are on. With the global operating mode only its - value counts. + setting, all others are on. """ char = self._homekit_camera_active_char return char is None or bool(char.value) @@ -1486,6 +1508,20 @@ def _homekit_camera_on(self) -> bool: assert self._homekit_camera_active_char is not None # noqa: S101 return bool(self._homekit_camera_active_char.value) + def _handle_classic_camera_active(self, value: Any) -> None: + # A hub that only knows the Camera Operating Mode service. + char = self._homekit_camera_active_char + assert char is not None # noqa: S101 + if char.value != value: + char.set_value(value) + self._handle_homekit_camera_active(value) + + def _handle_global_camera_active(self, value: Any) -> None: + char = self._classic_camera_active_char + if char is not None and char.value != value: + char.set_value(value) + self._handle_homekit_camera_active(value) + def _handle_homekit_camera_active(self, value: Any) -> None: # A camera turned off must not record, and the hub must not treat its # motion sensor as an active trigger. @@ -1497,6 +1533,7 @@ def _handle_homekit_camera_active(self, value: Any) -> None: self._motion_status_active_char.notify() if not active and self._recording_stream is not None: self._recording_stream.close(hds_recording.RecordingReason.NOT_ALLOWED) + self._end_streams_if_not_allowed() self.homekit_camera_active_changed(active) def homekit_camera_active_changed(self, active: bool) -> None: @@ -1505,8 +1542,9 @@ def homekit_camera_active_changed(self, active: bool) -> None: Override instead of replacing the characteristic's setter. A running recording is already closed when the camera is turned off. Like other setters, every controller write calls this, setting the value locally - does not. With the ``global_operating_mode`` option only changes call - it, not a controller writing the same value again. + does not. With the ``global_operating_mode`` option, writes to its + HomeKit Camera Active only call it on changes, not when a controller + writes the same value again. """ async def _stop_stream(self, objs: dict[bytes, Any]) -> None: @@ -1529,12 +1567,19 @@ async def _stop_stream(self, objs: dict[bytes, Any]) -> None: ) return - stream_idx = session_info["stream_idx"] + await self._end_stream(session_id) + + async def _end_stream(self, session_id: UUID) -> int | None: + """Stop the stream of ``session_id`` and return its stream index.""" + session_info = self.sessions.get(session_id) + if not session_info: + return None + stream_idx: int = session_info["stream_idx"] await self.stop_stream(session_info) self._close_talkback(session_info) - del self.sessions[session_id] - + self.sessions.pop(session_id, None) self._streaming_status[stream_idx] = STREAMING_STATUS["AVAILABLE"] + return stream_idx @staticmethod def _close_talkback(session_info: dict[str, Any]) -> None: diff --git a/tests/test_camera_operating_mode.py b/tests/test_camera_operating_mode.py index 19cc7328..9958e308 100644 --- a/tests/test_camera_operating_mode.py +++ b/tests/test_camera_operating_mode.py @@ -248,8 +248,12 @@ async def test_global_operating_mode_decides(driver: Any) -> None: stream.close.assert_called_once_with(hds_recording.RecordingReason.NOT_ALLOWED) -async def test_operating_mode_ignored(driver: Any) -> None: - """Test the Camera Operating Mode value is ignored next to the global one.""" +async def test_last_camera_active_write_decides(driver: Any) -> None: + """Test the last write to either HomeKit Camera Active decides. + + A hub that only knows the Camera Operating Mode service still turns the + camera off, and the other characteristic follows. + """ camera = _camera(driver, recording=True, global_operating_mode=[]) camera.homekit_camera_active_changed = MagicMock() stream = MagicMock() @@ -258,21 +262,84 @@ async def test_operating_mode_ignored(driver: Any) -> None: active = _char(camera, GLOBAL, "HomeKitCameraActive") legacy = _char(camera, "CameraOperatingMode", "HomeKitCameraActive") - # A value left from before must not keep the camera off. - active.client_update_value(0) legacy.client_update_value(0) + assert camera.homekit_camera_active is False + assert camera.streaming_allowed is False + assert active.get_value() == 0 + assert status_active.get_value() is False + stream.close.assert_called_once_with(hds_recording.RecordingReason.NOT_ALLOWED) + active.client_update_value(1) - assert camera.homekit_camera_active_changed.call_count == 2 - camera.homekit_camera_active_changed.assert_called_with(True) assert camera.homekit_camera_active is True assert camera.streaming_allowed is True + assert legacy.get_value() == 1 assert status_active.get_value() is True + assert camera.homekit_camera_active_changed.call_args_list == [ + call(False), + call(True), + ] - legacy.client_update_value(1) - legacy.client_update_value(0) - assert camera.homekit_camera_active_changed.call_count == 2 - assert camera.homekit_camera_active is True - stream.close.assert_called_once() + +@pytest.mark.parametrize( + ("service", "name", "value"), + [ + (GLOBAL, "HomeKitCameraActive", 0), + (GLOBAL, "StreamingEnabled", 0), + ("CameraOperatingMode", "HomeKitCameraActive", 0), + ], +) +async def test_turning_streaming_off_ends_streams( + driver: Any, service: str, name: str, value: int +) -> None: + """Test running live streams end once streaming is turned off.""" + camera = _camera( + driver, video=VIDEO, audio=AUDIO, recording=True, global_operating_mode=[] + ) + camera.stop_stream = AsyncMock() + session_info = {"id": SESSION_ID, "stream_idx": 0} + camera.sessions[SESSION_ID] = session_info + camera._streaming_status[0] = STREAMING_STATUS["STREAMING"] + + _char(camera, service, name).client_update_value(value) + await asyncio.sleep(0) + await asyncio.sleep(0) + + camera.stop_stream.assert_awaited_once_with(session_info) + assert camera.sessions == {} + assert camera.streaming_status == STREAMING_STATUS["AVAILABLE"] + + +async def test_manually_disabled_ends_streams(driver: Any) -> None: + """Test turning the camera off on the device ends running live streams.""" + camera = _camera( + driver, video=VIDEO, audio=AUDIO, global_operating_mode=["ManuallyDisabled"] + ) + camera.stop_stream = AsyncMock() + camera.sessions[SESSION_ID] = {"id": SESSION_ID, "stream_idx": 0} + + camera.set_manually_disabled(True) + await asyncio.sleep(0) + await asyncio.sleep(0) + + camera.stop_stream.assert_awaited_once() + assert camera.sessions == {} + + +async def test_streams_kept_while_allowed(driver: Any) -> None: + """Test other settings leave running live streams alone.""" + camera = _camera( + driver, video=VIDEO, audio=AUDIO, global_operating_mode=["NightVision"] + ) + camera.stop_stream = AsyncMock() + camera.sessions[SESSION_ID] = {"id": SESSION_ID, "stream_idx": 0} + + _char(camera, GLOBAL, "NightVision").client_update_value(1) + _char(camera, GLOBAL, "HomeKitCameraActive").client_update_value(1) + await asyncio.sleep(0) + await asyncio.sleep(0) + + camera.stop_stream.assert_not_awaited() + assert SESSION_ID in camera.sessions async def test_settings_survive_restart(driver: Any) -> None: @@ -384,5 +451,5 @@ async def test_unknown_chars(driver: Any, chars: list[str]) -> None: @pytest.mark.parametrize("chars", [True, "NightVision"]) async def test_option_not_a_list(driver: Any, chars: Any) -> None: """Test the option must be a list of names.""" - with pytest.raises(ValueError, match="must be a list"): + with pytest.raises(TypeError, match="must be a list"): _camera(driver, global_operating_mode=chars) From 0af78d70b5fee48b651b81bfad594962d074ad40 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Tue, 10 Nov 2026 15:53:18 +0100 Subject: [PATCH 8/8] Address the review of the global operating mode option The option takes any sequence of names and is no longer type checked. The service keeps its chars by name, built once in setup. A refused stream while streaming is turned off logs at info. Zone polygon identifiers use UUID(bytes=...) directly, it raises ValueError for a wrong length by itself. --- pyhap/camera.py | 28 +++++++++++++--------------- pyhap/hksv3.py | 9 ++------- tests/test_camera_operating_mode.py | 17 +++++++++-------- 3 files changed, 24 insertions(+), 30 deletions(-) diff --git a/pyhap/camera.py b/pyhap/camera.py index 6cb3add5..57eb778f 100644 --- a/pyhap/camera.py +++ b/pyhap/camera.py @@ -17,7 +17,7 @@ """ import asyncio -from collections.abc import AsyncGenerator, Callable, Iterable +from collections.abc import AsyncGenerator, Callable, Iterable, Sequence import functools import hashlib import ipaddress @@ -584,6 +584,8 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None: self._classic_camera_active_char: Characteristic | None = None self._motion_status_active_char = None self._global_operating_mode: Service | None = None + # Its chars by name, they don't change after the setup. + self._global_operating_mode_chars: dict[str, Characteristic] = {} # Persisted HKSV 17.99 chars with their state key, default and the # handler a reset calls. self._hksv3_defaults: dict[ @@ -1004,19 +1006,15 @@ def recording_audio_active(self) -> bool: self._recording_service.get_characteristic("RecordingAudioActive").value ) - def _setup_global_operating_mode(self, chars: list[str]) -> None: + def _setup_global_operating_mode(self, chars: Sequence[str]) -> None: """Create the Camera Global Operating Mode service (HKSV 17.99).""" - if not isinstance(chars, (list, tuple)): - raise TypeError( - "global_operating_mode must be a list of characteristic names" - ) optional = self.driver.loader.serv_types["CameraGlobalOperatingMode"][ "OptionalCharacteristics" ] if unknown := [name for name in chars if name not in optional]: raise ValueError( "Not an optional Camera Global Operating Mode characteristic: " - + ", ".join(unknown) + f"{', '.join(unknown)}" ) service = self.add_preload_service("CameraGlobalOperatingMode", chars=chars) for char in service.characteristics: @@ -1038,6 +1036,11 @@ def _setup_global_operating_mode(self, chars: list[str]) -> None: handler = functools.partial(self._handle_global_operating_mode, name) self._persist_hksv3_char(char, f"global_operating_mode.{name}", handler) self._global_operating_mode = service + self._global_operating_mode_chars = { + char.display_name: char + for char in service.characteristics + if char.display_name is not None + } @property def streaming_allowed(self) -> bool: @@ -1076,12 +1079,7 @@ def set_third_party_camera_active(self, active: bool) -> None: char.set_value(bool(active)) def _global_operating_mode_char(self, name: str) -> Characteristic | None: - if self._global_operating_mode is None: - return None - try: - return self._global_operating_mode.get_characteristic(name) - except ValueError: - return None + return self._global_operating_mode_chars.get(name) def _handle_global_operating_mode(self, name: str, value: Any) -> None: if name == "StreamingEnabled": @@ -1478,7 +1476,7 @@ async def _start_stream(self, objs: dict[bytes, Any], reconfigure: bool) -> None # Turned off after SetupEndpoints. refused = not reconfigure and not self.streaming_allowed if refused: - logger.warning("[%s] Streaming is turned off, refusing it", session_id) + logger.info("[%s] Streaming is turned off, refusing it", session_id) success = False elif reconfigure: success = await self.reconfigure_stream(session_info, opts) @@ -1675,7 +1673,7 @@ def set_endpoints(self, value: str, stream_idx: int | None = None) -> None: objs = tlv.decode(value, from_base64=True) session_id = UUID(bytes=objs[SETUP_TYPES["SESSION_ID"]]) if not self.streaming_allowed: - logger.warning("[%s] Streaming is turned off, refusing it", session_id) + logger.info("[%s] Streaming is turned off, refusing it", session_id) self._management[stream_idx].get_characteristic("SetupEndpoints").set_value( tlv.encode( SETUP_TYPES["SESSION_ID"], diff --git a/pyhap/hksv3.py b/pyhap/hksv3.py index 7cf0d1d6..263eb317 100644 --- a/pyhap/hksv3.py +++ b/pyhap/hksv3.py @@ -126,12 +126,6 @@ def _to_enum(enum_type: type[_E], data: bytes) -> _E | int: return value -def _to_uuid(data: bytes) -> UUID: - if len(data) != 16: - raise ValueError(f"Invalid UUID length {len(data)}") - return UUID(bytes=data) - - @dataclass class VideoStreamCapability: """A video encoding the camera can produce from a sensor, bit rates in kbps.""" @@ -230,7 +224,8 @@ def decode(cls, data: bytes) -> ZonePolygon: (_to_int(raw[i : i + 2]), _to_int(raw[i + 2 : i + 4])) for i in range(0, len(raw), 4) ] - identifier = _to_uuid(fields[1][0]) if 1 in fields else None + # UUID raises ValueError for anything but 16 bytes. + identifier = UUID(bytes=fields[1][0]) if 1 in fields else None return cls(identifier, vertices) diff --git a/tests/test_camera_operating_mode.py b/tests/test_camera_operating_mode.py index 9958e308..5d015b70 100644 --- a/tests/test_camera_operating_mode.py +++ b/tests/test_camera_operating_mode.py @@ -2,6 +2,7 @@ import asyncio from collections.abc import AsyncIterator +import logging import pathlib from typing import Any from unittest.mock import AsyncMock, MagicMock, call, patch @@ -149,7 +150,7 @@ async def test_writes_call_hook( @pytest.mark.parametrize( - ("enabled", "starts", "status", "warnings"), + ("enabled", "starts", "status", "refusals"), [ (True, 1, STREAMING_STATUS["STREAMING"], 0), (False, 0, STREAMING_STATUS["AVAILABLE"], 1), @@ -160,10 +161,11 @@ async def test_stream_start_needs_streaming_allowed( enabled: bool, starts: int, status: bytes, - warnings: int, + refusals: int, caplog: pytest.LogCaptureFixture, ) -> None: """Test a stream turned off after its setup does not start.""" + caplog.set_level(logging.INFO, "pyhap.camera") camera = _camera(driver, video=VIDEO, audio=AUDIO, global_operating_mode=[]) camera.start_stream = AsyncMock(return_value=True) camera.sessions[SESSION_ID] = {"id": SESSION_ID, "stream_idx": 0} @@ -175,7 +177,7 @@ async def test_stream_start_needs_streaming_allowed( assert camera.start_stream.await_count == starts assert camera.streaming_status == status assert (SESSION_ID in camera.sessions) is enabled - assert caplog.text.count("Streaming is turned off") == warnings + assert caplog.text.count("Streaming is turned off") == refusals assert "Failed to start/reconfigure stream" not in caplog.text @@ -448,8 +450,7 @@ async def test_unknown_chars(driver: Any, chars: list[str]) -> None: _camera(driver, global_operating_mode=chars) -@pytest.mark.parametrize("chars", [True, "NightVision"]) -async def test_option_not_a_list(driver: Any, chars: Any) -> None: - """Test the option must be a list of names.""" - with pytest.raises(TypeError, match="must be a list"): - _camera(driver, global_operating_mode=chars) +async def test_option_as_tuple(driver: Any) -> None: + """Test the option takes any sequence of names.""" + camera = _camera(driver, global_operating_mode=("NightVision",)) + assert _char(camera, GLOBAL, "NightVision").value == 0