Add support for a custom iid manager (#12)

pyhap/accessory.py

@@ -35,7 +35,7 @@ class Accessory:
 
     category = CATEGORY_OTHER
 
-    def __init__(self, driver, display_name, aid=None):
+    def __init__(self, driver, display_name, aid=None, iid_manager=None):
         """Initialise with the given properties.
 
         :param display_name: Name to be displayed in the Home app.
@@ -51,7 +51,7 @@ def __init__(self, driver, display_name, aid=None):
         self.display_name = display_name
         self.driver = driver
         self.services = []
-        self.iid_manager = IIDManager()
+        self.iid_manager = iid_manager or IIDManager()
         self.setter_callback = None
 
         self.add_info_service()
@@ -116,9 +116,11 @@ def set_info_service(
                     self.display_name,
                 )
 
-    def add_preload_service(self, service, chars=None):
+    def add_preload_service(self, service, chars=None, unique_id=None):
         """Create a service with the given name and add it to this acc."""
         service = self.driver.loader.get_service(service)
+        if unique_id is not None:
+            service.unique_id = unique_id
         if chars:
             chars = chars if isinstance(chars, list) else [chars]
             for char_name in chars:
@@ -144,12 +146,12 @@ def add_service(self, *servs):
         :type: Service
         """
         for s in servs:
+            s.broker = self
             self.services.append(s)
             self.iid_manager.assign(s)
-            s.broker = self
             for c in s.characteristics:
-                self.iid_manager.assign(c)
                 c.broker = self
+                self.iid_manager.assign(c)
 
     def get_service(self, name):
         """Return a Service with the given name.
@@ -323,8 +325,10 @@ class Bridge(Accessory):
 
     category = CATEGORY_BRIDGE
 
-    def __init__(self, driver, display_name):
-        super().__init__(driver, display_name, aid=STANDALONE_AID)
+    def __init__(self, driver, display_name, iid_manager=None):
+        super().__init__(
+            driver, display_name, aid=STANDALONE_AID, iid_manager=iid_manager
+        )
         self.accessories = {}  # aid: acc
 
     def add_accessory(self, acc):

pyhap/camera.py

@@ -441,7 +441,7 @@ def _setup_stream_management(self, options):
 
     def _create_stream_management(self, stream_idx, options):
         """Create a stream management service."""
-        management = self.add_preload_service("CameraRTPStreamManagement")
+        management = self.add_preload_service("CameraRTPStreamManagement", unique_id=stream_idx)
         management.configure_char(
             "StreamingStatus",
             getter_callback=lambda: self._get_streaming_status(stream_idx),

pyhap/characteristic.py

@@ -5,10 +5,8 @@
 a temperature measuring or a device status.
 """
 import logging
-
 from uuid import UUID
 
-
 from pyhap.const import (
     HAP_PERMISSION_READ,
     HAP_REPR_DESC,
@@ -133,10 +131,16 @@ class Characteristic:
         "_uuid_str",
         "_loader_display_name",
         "allow_invalid_client_values",
+        "unique_id",
     )
 
     def __init__(
-        self, display_name, type_id, properties, allow_invalid_client_values=False
+        self,
+        display_name,
+        type_id,
+        properties,
+        allow_invalid_client_values=False,
+        unique_id=None,
     ):
         """Initialise with the given properties.
 
@@ -169,12 +173,16 @@ def __init__(
         self.getter_callback = None
         self.setter_callback = None
         self.service = None
+        self.unique_id = unique_id
         self._uuid_str = uuid_to_hap_type(type_id)
         self._loader_display_name = None
 
     def __repr__(self):
         """Return the representation of the characteristic."""
-        return f"<characteristic display_name={self.display_name} value={self.value} properties={self.properties}>"
+        return (
+            f"<characteristic display_name={self.display_name} unique_id={self.unique_id} "
+            f"value={self.value} properties={self.properties}>"
+        )
 
     def _get_default_value(self):
         """Return default value for format."""

pyhap/iid_manager.py

@@ -28,9 +28,17 @@ def assign(self, obj):
             )
             return
 
+        iid = self.get_iid_for_obj(obj)
+        self.iids[obj] = iid
+        self.objs[iid] = obj
+
+    def get_iid_for_obj(self, obj):
+        """Get the IID for the given object.
+
+        Override this method to provide custom IID assignment.
+        """
         self.counter += 1
-        self.iids[obj] = self.counter
-        self.objs[self.counter] = obj
+        return self.counter
 
     def get_obj(self, iid):
         """Get the object that is assigned the given IID."""

pyhap/service.py

@@ -26,10 +26,11 @@ class Service:
         "linked_services",
         "is_primary_service",
         "setter_callback",
+        "unique_id",
         "_uuid_str",
     )
 
-    def __init__(self, type_id, display_name=None):
+    def __init__(self, type_id, display_name=None, unique_id=None):
         """Initialize a new Service object."""
         self.broker = None
         self.characteristics = []
@@ -38,12 +39,13 @@ def __init__(self, type_id, display_name=None):
         self.type_id = type_id
         self.is_primary_service = None
         self.setter_callback = None
+        self.unique_id = unique_id
         self._uuid_str = uuid_to_hap_type(type_id)
 
     def __repr__(self):
         """Return the representation of the service."""
         chars_dict = {c.display_name: c.value for c in self.characteristics}
-        return f"<service display_name={self.display_name} chars={chars_dict}>"
+        return f"<service display_name={self.display_name} unique_id={self.unique_id} chars={chars_dict}>"
 
     def add_linked_service(self, service):
         """Add the given service as "linked" to this Service."""

pyhap/util.py

@@ -141,6 +141,7 @@ async def event_wait(event, timeout):
     return event.is_set()
 
 
+@functools.lru_cache(maxsize=2048)
 def uuid_to_hap_type(uuid):
     """Convert a UUID to a HAP type."""
     long_type = str(uuid).upper()
@@ -149,6 +150,7 @@ def uuid_to_hap_type(uuid):
     return long_type.split("-", 1)[0].lstrip("0")
 
 
+@functools.lru_cache(maxsize=2048)
 def hap_type_to_uuid(hap_type):
     """Convert a HAP type to a UUID."""
     if "-" in hap_type:

tests/test_accessory.py

@@ -15,6 +15,7 @@
     HAP_REPR_VALUE,
     STANDALONE_AID,
 )
+from pyhap.iid_manager import IIDManager
 from pyhap.service import Service
 from pyhap.state import State
 
@@ -44,6 +45,31 @@ def test_acc_init(mock_driver):
     Accessory(mock_driver, "Test Accessory")
 
 
+def test_acc_with_custom_iid_manager(mock_driver):
+    """Test Accessory with custom IIDManager."""
+
+    class CustomIIDManager(IIDManager):
+        """A custom IIDManager that starts at 1000."""
+
+        def __init__(self):
+            super().__init__()
+            self.counter = 1000
+
+        def get_iid_for_obj(self, obj):
+            """Assign an IID to an object."""
+            if isinstance(obj, Service) and obj.unique_id == "service_54":
+                return 5000 + obj.broker.aid
+            return super().get_iid_for_obj(obj)
+
+    iid_manager = CustomIIDManager()
+    acc = Accessory(mock_driver, "Test Accessory", iid_manager=iid_manager, aid=1)
+    acc.add_preload_service("GarageDoorOpener", unique_id="service_54")
+    acc_info_service = acc.get_service("AccessoryInformation")
+    acc_garage_door_opener_service = acc.get_service("GarageDoorOpener")
+    assert iid_manager.get_iid(acc_info_service) == 1001
+    assert iid_manager.get_iid(acc_garage_door_opener_service) == 5001
+
+
 def test_acc_publish_no_broker(mock_driver):
     acc = Accessory(mock_driver, "Test Accessory")
     service = acc.driver.loader.get_service("TemperatureSensor")

tests/test_camera.py

@@ -45,6 +45,7 @@ def test_init(mock_driver):
     acc = camera.Camera(_OPTIONS, mock_driver, "Camera")
 
     management = acc.get_service("CameraRTPStreamManagement")
+    assert management.unique_id is not None
 
     assert (
         management.get_characteristic("SupportedRTPConfiguration").get_value() == "AgEA"

tests/test_characteristic.py

@@ -45,11 +45,22 @@ def test_repr():
     char = get_char(PROPERTIES.copy())
     del char.properties["Permissions"]
     assert (
-        repr(char) == "<characteristic display_name=Test Char value=0 "
+        repr(char) == "<characteristic display_name=Test Char unique_id=None value=0 "
         "properties={'Format': 'int'}>"
     )
 
 
+def test_char_with_unique_id():
+    """Test Characteristic with unique_id."""
+    service = Characteristic(
+        display_name="Test Char",
+        type_id=uuid1(),
+        properties={"Format": "int"},
+        unique_id="123",
+    )
+    assert service.unique_id == "123"
+
+
 def test_default_value():
     """Test getting the default value for a specific format."""
     char = get_char(PROPERTIES.copy())

tests/test_service.py

@@ -28,9 +28,18 @@ def get_chars():
 
 def test_repr():
     """Test service representation."""
-    service = Service(uuid1(), "TestService")
+    service = Service(uuid1(), "TestService", unique_id="my_service_unique_id")
     service.characteristics = [get_chars()[0]]
-    assert repr(service) == "<service display_name=TestService chars={'Char 1': 0}>"
+    assert (
+        repr(service)
+        == "<service display_name=TestService unique_id=my_service_unique_id chars={'Char 1': 0}>"
+    )
+
+
+def test_service_with_unique_id():
+    """Test service with unique_id."""
+    service = Service(uuid1(), "TestService", unique_id="service_unique_id")
+    assert service.unique_id == "service_unique_id"
 
 
 def test_add_characteristic():