Z2MClient¶

async def disable_permit_join(self) -> None:
    """Close the Z2M network to new joiners."""
    try:
        await self._post("/api/permit_join", {"time": 0, "device": None})
    except RuntimeError:
        await self._mqtt_publish(
            f"{self._mqtt_topic}/bridge/request/permit_join",
            json.dumps({"time": 0}),
        )

async def enable_permit_join(self, seconds: int = 120) -> None:
    """Open the Z2M network for new devices to join."""
    try:
        await self._post("/api/permit_join", {"time": seconds, "device": None})
    except RuntimeError:
        await self._mqtt_publish(
            f"{self._mqtt_topic}/bridge/request/permit_join",
            json.dumps({"time": seconds}),
        )

async def get_device_by_ieee(self, ieee: str) -> dict[str, Any] | None:
    """Find a Z2M device by IEEE address. Returns None if not found."""
    devices = await self.get_devices()
    target = normalize_ieee(ieee)
    for device in devices:
        if normalize_ieee(device.get("ieee_address", "")) == target:
            return device
    return None

async def get_devices(self) -> list[dict[str, Any]]:
    """Return the full Z2M device list."""
    try:
        return await self._get("/api/devices")
    except (RuntimeError, httpx.HTTPStatusError, httpx.RequestError):
        return await self._get_devices_via_ha()

async def get_network_map(self, timeout: int = _NETWORK_MAP_TIMEOUT) -> dict[str, Any]:
    """Return the raw Z2M network map (nodes + links).

    When no HA URL is configured (standalone Z2M mode), connects directly
    to the Z2M frontend WebSocket at ``<z2m_url>/api``.

    Otherwise, tries the Z2M HTTP REST endpoint first. Z2M 2.x removed
    that endpoint, so falls back to subscribing to the MQTT response topic
    and publishing the request via HA's WebSocket ``call_service`` API.

    ``timeout`` controls how long to wait for Z2M to respond via MQTT.
    Large meshes with many routers need more time — Z2M scans each router
    sequentially, so allow ~10 s per router.
    """
    if not self._ha_url:
        return await self._get_network_map_via_z2m_ws(timeout=timeout)
    try:
        return await self._get("/api/networkmap?type=raw")
    except (RuntimeError, httpx.HTTPStatusError, httpx.RequestError):
        return await self._get_network_map_via_mqtt(timeout=timeout)

async def remove_device(self, friendly_name: str, force: bool = True) -> None:
    """Remove a device from Z2M by friendly name (HTTP fallback -> MQTT)."""
    try:
        await self._post("/api/device/remove", {"id": friendly_name, "force": force})
    except (RuntimeError, httpx.HTTPStatusError, httpx.RequestError):
        await self._mqtt_publish(
            f"{self._mqtt_topic}/bridge/request/device/remove",
            json.dumps({"id": friendly_name, "force": force}),
        )

async def rename_device(self, current_name: str, new_name: str) -> None:
    """Rename a Z2M device by its current friendly name."""
    try:
        await self._post("/api/device", {"id": current_name, "rename": new_name})
    except RuntimeError:
        # homeassistant_rename=True also updates HA entity IDs
        await self._mqtt_publish(
            f"{self._mqtt_topic}/bridge/request/device/rename",
            json.dumps({"from": current_name, "to": new_name, "homeassistant_rename": True}),
        )

async def wait_for_interview(
    self,
    ieee: str,
    timeout: int = 300,
    on_event: Any = None,
) -> tuple[str, dict[str, Any] | None]:
    """Subscribe to Z2M bridge events and wait for device interview completion.

    Subscribes to ``zigbee2mqtt/bridge/event`` via HA WebSocket and streams
    events until the target device's interview completes, fails, or the
    timeout expires.

    ``on_event(event_type, data)`` is called for every bridge event received
    so the caller can display status updates or detect unexpected joiners.

    Returns a tuple ``(status, data)``:

    - ``("successful", data)`` — interview completed (or device re-announced)
    - ``("failed", data)``    — interview failed
    - ``("timeout", None)``   — timed out without a result
    """
    target = normalize_ieee(ieee)
    event_topic = f"{self._mqtt_topic}/bridge/event"
    ha = self._ha_client()
    deadline = time.monotonic() + timeout

    async with ha._ws_session() as ws:
        await ws.send(json.dumps({"id": 1, "type": "mqtt/subscribe", "topic": event_topic}))

        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return "timeout", None

            try:
                raw = await asyncio.wait_for(ws.recv(), timeout=min(remaining, 2.0))
            except asyncio.TimeoutError:
                continue

            try:
                msg = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                continue

            # Validate subscribe ACK
            if msg.get("id") == 1 and msg.get("type") == "result":
                if not msg.get("success"):
                    raise RuntimeError(f"mqtt/subscribe to bridge/event failed: {msg}")
                continue

            if msg.get("type") != "event" or msg.get("id") != 1:
                continue

            try:
                payload = json.loads(msg.get("event", {}).get("payload", "{}"))
            except (json.JSONDecodeError, ValueError):
                continue

            event_type = payload.get("type")
            data = payload.get("data") or {}

            if on_event is not None:
                on_event(event_type, data)

            event_ieee = normalize_ieee(data.get("ieee_address", ""))
            if event_ieee != target:
                continue

            # A device_announce means the device is already paired and working.
            if event_type == "device_announce":
                return "successful", data

            if event_type == "device_interview":
                interview_status = data.get("status")
                if interview_status == "successful":
                    return "successful", data
                if interview_status == "failed":
                    return "failed", data