Break the import cycle between the camera and its stream services

CHANGELOG.md

@@ -22,6 +22,13 @@ Sections
 
 - Add Python 3.14 and 3.15 to the supported versions (tox envs, CI matrix, and
   package classifiers).
+- Break the import cycle between the camera and its stream services:
+  `parse_setup_address` moved from `pyhap.camera` to `pyhap.util` (still
+  importable from `pyhap.camera`) and `RESPONSE_TIMEOUT` moved from
+  `pyhap.hap_handler` to `pyhap.const` (still importable from
+  `pyhap.hap_handler`). `pyhap.camera` now imports the multi-tier and WebRTC
+  stream services at the top, which makes the cycle gone instead of hidden in
+  a deferred import.
 
 ## [5.2.0] - 2026-11-07
 

pyhap/camera.py

@@ -26,7 +26,7 @@
 import os
 import struct
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any
+from typing import Any
 from uuid import UUID
 
 from pyhap import (
@@ -39,15 +39,19 @@
     tlv,
 )
 from pyhap.accessory import Accessory
+from pyhap.camera_multi_tier import MultiTierConfig, MultiTierController
 from pyhap.camera_talkback import TalkbackReceiver
+from pyhap.camera_webrtc import WebRTCConfig, WebRTCController
 from pyhap.characteristic import Characteristic
 from pyhap.const import CATEGORY_CAMERA
 from pyhap.service import Service
-from pyhap.util import base64_to_bytes, byte_bool, callback, to_base64_str
-
-if TYPE_CHECKING:
-    from pyhap.camera_multi_tier import MultiTierController
-    from pyhap.camera_webrtc import WebRTCController
+from pyhap.util import (
+    base64_to_bytes,
+    byte_bool,
+    callback,
+    parse_setup_address,
+    to_base64_str,
+)
 
 # Seconds before arming a recording delegate is tried again, doubling up to
 # the maximum while it keeps failing.
@@ -275,29 +279,6 @@
 )
 
 
-def parse_setup_address(
-    value: bytes, is_ipv6: bool
-) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
-    """Return the controller address of a Setup Endpoints write.
-
-    The address must be an IP address of the announced version, without an
-    IPv6 scope ID. ``str()`` gives its canonical form, e.g. ``fd00::1`` for
-    ``FD00:0::1``.
-
-    :raises ValueError: If the address is anything else.
-    """
-    try:
-        address = value.decode("utf-8")
-        parsed = ipaddress.ip_address(address)
-    except ValueError as ex:
-        raise ValueError(f"invalid address {value!r}") from ex
-    if isinstance(parsed, ipaddress.IPv6Address) and parsed.scope_id is not None:
-        raise ValueError(f"address {address!r} has a scope ID")
-    if parsed.version != (6 if is_ipv6 else 4):
-        raise ValueError(f"address {address!r} does not match its IP version")
-    return parsed
-
-
 def _decode_setup_endpoints(
     session_id: UUID, objs: dict[bytes, Any]
 ) -> tuple[str, bool, int, int, tuple[bytes, bytes], tuple[bytes, bytes]]:
@@ -822,13 +803,6 @@ def _create_stream_management(
 
     def _setup_stream_controllers(self, options: dict[str, Any]) -> None:
         """Add the multi-tier and WebRTC controllers of each sensor (17.99)."""
-        # Here, not at the top: these modules import this one.
-        from pyhap.camera_multi_tier import (  # noqa: PLC0415
-            MultiTierConfig,
-            MultiTierController,
-        )
-        from pyhap.camera_webrtc import WebRTCConfig, WebRTCController  # noqa: PLC0415
-
         if options.get("multi_tier") is None or options.get("webrtc") is None:
             raise ValueError(
                 "multi_tier and webrtc need each other, without WebRTC "

pyhap/camera_multi_tier.py

@@ -21,7 +21,6 @@
 from typing import TYPE_CHECKING, Any, Protocol, cast
 
 from pyhap import hksv3
-from pyhap.camera import parse_setup_address
 from pyhap.camera_stream import (
     DEFAULT_HOOK_TIMEOUT,
     HookFailed,
@@ -32,7 +31,7 @@
     config_from_option,
     failure_status,
 )
-from pyhap.util import base64_to_bytes, to_base64_str
+from pyhap.util import base64_to_bytes, parse_setup_address, to_base64_str
 
 if TYPE_CHECKING:
     from pyhap.camera import Camera

pyhap/camera_stream.py

@@ -41,7 +41,7 @@
 from uuid import UUID
 
 from pyhap import hksv3
-from pyhap.hap_handler import RESPONSE_TIMEOUT
+from pyhap.const import RESPONSE_TIMEOUT
 
 if TYPE_CHECKING:
     from pyhap.camera import Camera

pyhap/const.py

@@ -11,6 +11,11 @@
 # ### Misc ###
 STANDALONE_AID = 1  # Standalone accessory ID (i.e. not bridged)
 
+# iOS terminates a connection whose response takes longer than 10 seconds,
+# so the server only allows itself 9 to leave time to answer. The stream
+# services size their delegate hook timeouts after it.
+RESPONSE_TIMEOUT = 9
+
 # ### Default values ###
 DEFAULT_CONFIG_VERSION = 1
 DEFAULT_PORT = 51827

pyhap/hap_handler.py

@@ -25,6 +25,7 @@
     HAP_REPR_CHARS,
     HAP_REPR_STATUS,
     HAP_SERVER_STATUS,
+    RESPONSE_TIMEOUT,
 )
 from pyhap.util import long_to_bytes
 
@@ -35,10 +36,6 @@
 if TYPE_CHECKING:
     from .accessory_driver import AccessoryDriver
 
-# iOS will terminate the connection if it does not respond within
-# 10 seconds, so we only allow 9 seconds to avoid this.
-RESPONSE_TIMEOUT = 9
-
 
 logger = logging.getLogger(__name__)
 

pyhap/hksv3.py

@@ -501,7 +501,7 @@ class EndpointAddress:
     """An address of Setup Endpoints and its RTP ports.
 
     ``address`` is the text the controller sent. Check it with
-    :func:`pyhap.camera.parse_setup_address` before using it.
+    :func:`pyhap.util.parse_setup_address` before using it.
     """
 
     address: str

pyhap/util.py

@@ -3,6 +3,7 @@
 from collections.abc import Awaitable, Callable
 import functools
 import inspect
+import ipaddress
 import random
 import socket
 from typing import Any, cast
@@ -128,6 +129,29 @@ def byte_bool(boolv: Any) -> bytes:
     return b"\x01" if boolv else b"\x00"
 
 
+def parse_setup_address(
+    value: bytes, is_ipv6: bool
+) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
+    """Return the controller address of a Setup Endpoints write.
+
+    The address must be an IP address of the announced version, without an
+    IPv6 scope ID. ``str()`` gives its canonical form, e.g. ``fd00::1`` for
+    ``FD00:0::1``.
+
+    :raises ValueError: If the address is anything else.
+    """
+    try:
+        address = value.decode("utf-8")
+        parsed = ipaddress.ip_address(address)
+    except ValueError as ex:
+        raise ValueError(f"invalid address {value!r}") from ex
+    if isinstance(parsed, ipaddress.IPv6Address) and parsed.scope_id is not None:
+        raise ValueError(f"address {address!r} has a scope ID")
+    if parsed.version != (6 if is_ipv6 else 4):
+        raise ValueError(f"address {address!r} does not match its IP version")
+    return parsed
+
+
 async def event_wait(event: asyncio.Event, timeout: float) -> bool:
     """Wait for the given event to be set or for the timeout to expire.