From dabb6630a670c5c1381f57bf73a438714a8a36b6 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Thu, 17 Sep 2026 20:10:08 +0000 Subject: [PATCH] Drop legacy JSON protocol support --- pyproject.toml | 1 - tests/test_legacy.py | 1127 ---------------------------- tests/test_transport.py | 21 +- uv.lock | 14 - zigpy_ziggurat/zigbee/legacy.py | 433 ----------- zigpy_ziggurat/zigbee/transport.py | 629 +--------------- 6 files changed, 28 insertions(+), 2197 deletions(-) delete mode 100644 tests/test_legacy.py delete mode 100644 zigpy_ziggurat/zigbee/legacy.py diff --git a/pyproject.toml b/pyproject.toml index bb1e0bb..dada0a7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,6 @@ requires-python = ">=3.12" dependencies = [ "zigpy", "aiohttp", - "mashumaro", "aiospinel>=1.2.0", ] diff --git a/tests/test_legacy.py b/tests/test_legacy.py deleted file mode 100644 index 9212e81..0000000 --- a/tests/test_legacy.py +++ /dev/null @@ -1,1127 +0,0 @@ -"""Tests for the legacy JSON-RPC server and the transport shim that transcodes the -binary protocol to it. Both the shim and this file are temporary: when the legacy -server is retired, delete them together. - -The synthetic JSON server lives here rather than in `tests/common.py` for the same -reason -- nothing else depends on it.""" - -import asyncio -from collections.abc import AsyncIterator, Awaitable, Callable -import dataclasses -import json -import logging -from typing import Any, TypeVar - -from aiohttp import web -from aiohttp.test_utils import TestServer -import pytest -import zigpy.device -import zigpy.endpoint -from zigpy.exceptions import DeliveryError, NetworkNotFormed -import zigpy.state -import zigpy.types as t -import zigpy.zdo.types as zdo_t - -from tests.common import ( - COORDINATOR_IEEE, - DEVICE_IEEE, - DEVICE_NWK, - LINK_KEY, - NETWORK_KEY, - flush, - make_app_config, -) -from zigpy_ziggurat.zigbee import ( - application as application_module, - legacy as commands, - protocol as p, -) -from zigpy_ziggurat.zigbee.application import ControllerApplication -from zigpy_ziggurat.zigbee.transport import LegacyWebSocketTransport, connect_transport - -REQUEST_T = TypeVar("REQUEST_T") - - -def _request_types() -> dict[str, type[commands.Request[Any]]]: - """Every concrete request, walking past intermediate bases like - `StreamingRequest` that declare `method` without assigning it.""" - result: dict[str, type[commands.Request[Any]]] = {} - stack = list(commands.Request.__subclasses__()) - while stack: - cls = stack.pop() - stack.extend(cls.__subclasses__()) - if "method" in cls.__dict__: - result[cls.method] = cls - return result - - -REQUEST_TYPES: dict[str, type[commands.Request[Any]]] = _request_types() -NOTIFICATION_EVENTS: dict[type[commands.Notification], str] = { - cls: name for name, cls in commands.NOTIFICATIONS.items() -} - - -class RpcError(Exception): - """Raised by a handler to produce an error response.""" - - def __init__(self, code: str, message: str) -> None: - super().__init__(f"{code}: {message}") - self.code = code - self.message = message - - -def make_network_info() -> commands.NetworkInfo: - return commands.NetworkInfo( - channel=t.uint8_t(15), - nwk_update_id=t.uint8_t(0), - pan_id=t.PanId(0x1A2B), - extended_pan_id=t.ExtendedPanId(t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11")), - nwk_address=t.NWK(0x0000), - ieee_address=COORDINATOR_IEEE, - network_key=NETWORK_KEY, - network_key_seq=t.uint8_t(0), - network_key_tx_counter=t.uint32_t(1000), - tc_link_key=t.KeyData(b"ZigBeeAlliance09"), - tx_power=8, - tclk_seed=None, - tclk_flavor=None, - key_table=[], - ) - - -class SyntheticLegacyZiggurat: - """A real aiohttp websocket server speaking the legacy JSON protocol, with - per-method handlers that tests can override.""" - - def __init__(self) -> None: - self.web_app = web.Application() - self.web_app.router.add_get("/", self._handle_connection) - self.url = "" - self.connections = 0 - self._ws: web.WebSocketResponse | None = None - self.requests: list[Any] = [] - self._configured: commands.Configure | None = None - self.network_info = make_network_info() - self.hw_address = t.EUI64.convert("11:22:33:44:55:66:77:88") - self.handlers: dict[str, Callable[[Any, int], Awaitable[Any]]] = { - "ping": self.on_ping, - "configure": self.on_configure, - "get_network_info": self.on_get_network_info, - "get_hw_address": self.on_get_hw_address, - "send_aps": self.on_send_aps, - "energy_scan": self.on_energy_scan, - "network_scan": self.on_network_scan, - "permit_joins": self.on_status, - "set_provisional_key": self.on_status, - "set_channel": self.on_status, - "set_nwk_update_id": self.on_status, - "packet_capture": self.on_status, - "packet_capture_change_channel": self.on_status, - } - - @property - def ws(self) -> web.WebSocketResponse: - assert self._ws is not None - return self._ws - - @property - def configured(self) -> commands.Configure: - assert self._configured is not None - return self._configured - - async def _handle_connection(self, request: web.Request) -> web.WebSocketResponse: - self.connections += 1 - ws = web.WebSocketResponse() - await ws.prepare(request) - self._ws = ws - - await ws.send_json({"type": "hello", "version": 1, "state": "running"}) - - async for msg in ws: - data = json.loads(msg.data) - command = REQUEST_TYPES[data["method"]].from_dict(data["params"]) - self.requests.append(command) - await ws.send_json({"type": "event", "id": data["id"], "event": "accepted"}) - - try: - response = await self.handlers[data["method"]](command, data["id"]) - except RpcError as exc: - await ws.send_json( - { - "type": "response", - "id": data["id"], - "error": {"code": exc.code, "message": exc.message}, - } - ) - else: - # `None` deliberately withholds the response - if response is not None: - await ws.send_json( - { - "type": "response", - "id": data["id"], - "result": response.to_dict(), - } - ) - - return ws - - async def send_event(self, request_id: int, event: str) -> None: - await self.ws.send_json({"type": "event", "id": request_id, "event": event}) - - async def send_event_data( - self, request_id: int, event: str, data: dict[str, Any] - ) -> None: - await self.ws.send_json( - {"type": "event", "id": request_id, "event": event, "data": data} - ) - - async def send_confirm(self, request_id: int, *, reason: str | None = None) -> None: - if reason is not None: - data: dict[str, Any] = { - "id": request_id, - "status": "failed", - "reason": reason, - } - else: - data = {"id": request_id, "status": "confirmed", "next_hop": None} - - await self.ws.send_json( - {"type": "notification", "event": "send_confirm", "data": data} - ) - - async def aps_ack_confirm( - self, request_id: int, *, reason: str | None = None - ) -> None: - if reason is not None: - data: dict[str, Any] = { - "id": request_id, - "status": "failed", - "reason": reason, - } - else: - data = {"id": request_id, "status": "confirmed"} - - await self.ws.send_json( - {"type": "notification", "event": "aps_ack_confirm", "data": data} - ) - - async def send_notification(self, notification: commands.Notification) -> None: - await self.ws.send_json( - { - "type": "notification", - "event": NOTIFICATION_EVENTS[type(notification)], - "data": notification.to_dict(), - } - ) - - async def send_raw(self, text: str) -> None: - await self.ws.send_str(text) - - def sent(self, request_type: type[REQUEST_T]) -> list[REQUEST_T]: - return [r for r in self.requests if isinstance(r, request_type)] - - async def wait_for( - self, request_type: type[REQUEST_T], count: int = 1 - ) -> REQUEST_T: - async with asyncio.timeout(2): - while len(self.sent(request_type)) < count: - await asyncio.sleep(0.01) - - return self.sent(request_type)[count - 1] - - async def on_ping(self, command: commands.Ping, request_id: int) -> commands.Status: - return commands.Status(status="pong") - - async def on_status(self, command: Any, request_id: int) -> commands.Status: - return commands.Status(status="success") - - async def on_configure( - self, command: commands.Configure, request_id: int - ) -> commands.Status: - self._configured = command - return commands.Status(status="success") - - async def on_get_network_info( - self, command: commands.GetNetworkInfo, request_id: int - ) -> commands.NetworkInfo: - return self.network_info - - async def on_get_hw_address( - self, command: commands.GetHwAddress, request_id: int - ) -> commands.HwAddress: - return commands.HwAddress(ieee_address=self.hw_address) - - async def on_send_aps( - self, command: commands.SendAps, request_id: int - ) -> commands.Status: - await self.send_confirm(request_id) - if command.aps_ack: - await self.aps_ack_confirm(request_id) - return commands.Status(status="accepted") - - async def on_energy_scan( - self, command: commands.EnergyScan, request_id: int - ) -> commands.Status: - for channel in command.channels: - await self.send_event_data( - request_id, - "energy_result", - commands.EnergyScanResult( - channel=t.uint8_t(channel), rssi=t.int8s(-85) - ).to_dict(), - ) - return commands.Status(status="complete") - - async def on_network_scan( - self, command: commands.NetworkScan, request_id: int - ) -> commands.Status: - return commands.Status(status="complete") - - -@pytest.fixture -async def legacy_server() -> AsyncIterator[SyntheticLegacyZiggurat]: - ziggurat = SyntheticLegacyZiggurat() - test_server = TestServer(ziggurat.web_app) - await test_server.start_server() - ziggurat.url = f"ws://localhost:{test_server.port}/" - - yield ziggurat - - await test_server.close() - - -@pytest.fixture -async def legacy_connected_app( - legacy_server: SyntheticLegacyZiggurat, -) -> AsyncIterator[ControllerApplication]: - app = ControllerApplication(make_app_config(legacy_server.url)) - await app.connect() - - yield app - - await app.shutdown(db=False) - - -@pytest.fixture -async def legacy_app( - legacy_connected_app: ControllerApplication, -) -> ControllerApplication: - await legacy_connected_app.start_network() - return legacy_connected_app - - -async def _legacy( - server: SyntheticLegacyZiggurat, -) -> tuple[LegacyWebSocketTransport, list[bytes]]: - frames: list[bytes] = [] - transport = await connect_transport(server.url, frames.append, lambda exc: None) - assert isinstance(transport, LegacyWebSocketTransport) - return transport, frames - - -async def _wait_for(frames: list[bytes], count: int = 1) -> None: - async with asyncio.timeout(2): - while len(frames) < count: - await asyncio.sleep(0.01) - - -def add_initialized_device(app: ControllerApplication) -> zigpy.device.Device: - device = app.add_device(DEVICE_IEEE, DEVICE_NWK) - device.node_desc = app.get_device(nwk=t.NWK(0x0000)).node_desc - device.status = zigpy.device.Status.ENDPOINTS_INIT - device.add_endpoint(1).status = zigpy.endpoint.Status.ZDO_INIT - return device - - -# -- protocol probing ------------------------------------------------------------ - - -async def test_probe_selects_legacy(legacy_server: SyntheticLegacyZiggurat) -> None: - transport = await connect_transport( - legacy_server.url, lambda frame: None, lambda exc: None - ) - try: - assert isinstance(transport, LegacyWebSocketTransport) - finally: - await transport.disconnect() - - -# -- application against the legacy server --------------------------------------- - - -async def test_legacy_connect( - legacy_connected_app: ControllerApplication, - legacy_server: SyntheticLegacyZiggurat, -) -> None: - assert legacy_server.connections == 1 - - await legacy_connected_app.permit_ncp(1) - permit = legacy_server.sent(commands.PermitJoins)[-1] - assert permit.duration == 1 - assert permit.accept_direct_joins is True - - -async def test_legacy_start_network( - legacy_connected_app: ControllerApplication, - legacy_server: SyntheticLegacyZiggurat, -) -> None: - """The binary `Configure` + `LoadKeyTable`* + `StartNetwork` sequence coalesces - into the single JSON `configure` call the legacy server takes.""" - await legacy_connected_app.start_network() - - assert legacy_server.configured.channel == 15 - assert legacy_server.configured.pan_id == t.PanId(0x1A2B) - assert legacy_server.configured.network_key == NETWORK_KEY - assert legacy_server.configured.aps_frame_counter == 0 - # One JSON call, however many binary frames it was split across - assert len(legacy_server.sent(commands.Configure)) == 1 - assert legacy_connected_app.backups[-1].network_info.pan_id == t.PanId(0x1A2B) - - -async def test_legacy_load_network_info( - legacy_connected_app: ControllerApplication, - legacy_server: SyntheticLegacyZiggurat, -) -> None: - app = legacy_connected_app - - async def not_configured( - command: commands.GetNetworkInfo, request_id: int - ) -> commands.NetworkInfo: - raise RpcError("not_configured", "no stack is running") - - # A stateless server with no network running and no local backup: no network - legacy_server.handlers["get_network_info"] = not_configured - with pytest.raises(NetworkNotFormed): - await app.load_network_info() - - # Unrelated errors propagate. The unknown JSON code has no binary status, so - # it degrades to a generic invalid-request; the detail goes to the log. - async def serial_error( - command: commands.GetNetworkInfo, request_id: int - ) -> commands.NetworkInfo: - raise RpcError("serial_port_error", "it burned down") - - legacy_server.handlers["get_network_info"] = serial_error - with pytest.raises(DeliveryError, match="invalid_request"): - await app.load_network_info() - - legacy_server.handlers["get_network_info"] = legacy_server.on_get_network_info - await app.load_network_info() - assert app.state.node_info.ieee == COORDINATOR_IEEE - assert app.state.network_info.channel == 15 - assert app.state.network_info.network_key.key == NETWORK_KEY - assert app.state.network_info.stack_specific == {} - # The JSON server surfaces no children, address cache or route table - assert app.state.network_info.children == [] - assert app.state.network_info.nwk_addresses == {} - - # TCLK seeds map to the stack_specific layout of their source stack - legacy_server.network_info.tclk_seed = "ab" * 16 - legacy_server.network_info.tclk_flavor = "zstack" - await app.load_network_info() - assert app.state.network_info.stack_specific == {"zstack": {"tclk_seed": "ab" * 16}} - - legacy_server.network_info.tclk_flavor = "ezsp" - await app.load_network_info() - assert app.state.network_info.stack_specific == {"ezsp": {"hashed_tclk": "ab" * 16}} - - # The key table the JSON `get_network_info` returns inline is replayed as the - # events of the binary `ScanKeyTable` that follows - legacy_server.network_info.key_table = [ - commands.KeyTableEntry(partner_ieee=DEVICE_IEEE, key=LINK_KEY) - ] - await app.load_network_info() - assert app.state.network_info.key_table == [ - zigpy.state.Key(key=LINK_KEY, partner_ieee=DEVICE_IEEE) - ] - - -async def test_legacy_write_network_info( - legacy_connected_app: ControllerApplication, - legacy_server: SyntheticLegacyZiggurat, -) -> None: - app = legacy_connected_app - await app.load_network_info() - network_info = app.state.network_info - node_info = app.state.node_info - - # A zstack TCLK seed rides along verbatim - await app.write_network_info( - network_info=network_info.replace( - stack_specific={"zstack": {"tclk_seed": "cd" * 16}}, - key_table=[zigpy.state.Key(key=LINK_KEY, partner_ieee=DEVICE_IEEE)], - ), - node_info=node_info, - ) - assert legacy_server.configured.tclk_seed == "cd" * 16 - assert legacy_server.configured.tclk_flavor == "zstack" - assert legacy_server.configured.key_table == [ - commands.KeyTableEntry(partner_ieee=DEVICE_IEEE, key=LINK_KEY) - ] - - # An ezsp seed likewise - await app.write_network_info( - network_info=network_info.replace( - stack_specific={"ezsp": {"hashed_tclk": "ef" * 16}} - ), - node_info=node_info, - ) - assert legacy_server.configured.tclk_seed == "ef" * 16 - assert legacy_server.configured.tclk_flavor == "ezsp" - - # When zigpy forms a fresh network it leaves the IEEE address unspecified, - # deferring to the radio's hardware address - await app.write_network_info( - network_info=network_info, - node_info=node_info.replace(ieee=t.EUI64.UNKNOWN), # type: ignore[attr-defined] - ) - assert legacy_server.sent(commands.GetHwAddress) - assert legacy_server.configured.ieee_address == legacy_server.hw_address - assert app.state.node_info.ieee == legacy_server.hw_address - - -async def test_legacy_permits( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - await legacy_app.permit_with_link_key( - node=DEVICE_IEEE, link_key=LINK_KEY, time_s=12 - ) - - provisional = legacy_server.sent(commands.SetProvisionalKey)[-1] - assert provisional.ieee == DEVICE_IEEE - assert provisional.key == LINK_KEY - - # `super().permit()` broadcasts Mgmt_Permit_Joining_req and calls `permit_ncp` - broadcast = legacy_server.sent(commands.SendAps)[-1] - assert broadcast.delivery_mode == "broadcast" - assert broadcast.cluster_id == zdo_t.ZDOCmd.Mgmt_Permit_Joining_req - assert legacy_server.sent(commands.PermitJoins)[-1].duration == 12 - - -async def test_legacy_move_network_to_channel( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - await legacy_app._move_network_to_channel(new_channel=20, new_nwk_update_id=1) - - methods = [type(r) for r in legacy_server.requests] - assert methods.index(commands.SetNwkUpdateId) < methods.index(commands.SetChannel) - assert legacy_server.sent(commands.SetNwkUpdateId)[-1].nwk_update_id == 1 - assert legacy_server.sent(commands.SetChannel)[-1].channel == 20 - - -async def test_legacy_watchdog_feed( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - # The legacy server has no firmware-info call: it is probed with a JSON `ping` - await legacy_app._watchdog_feed() - assert isinstance(legacy_server.requests[-1], commands.Ping) - - -async def test_legacy_reset_network_info(legacy_app: ControllerApplication) -> None: - # The legacy server has no shutdown; the shim OKs it locally - await legacy_app.reset_network_info() - - -@pytest.mark.parametrize( - ("dst", "tx_options", "expected"), - [ - ( - t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=DEVICE_NWK), - t.TransmitOptions.ACK, - { - "delivery_mode": "unicast", - "destination": DEVICE_NWK, - "destination_eui64": DEVICE_IEEE, - "aps_ack": True, - "aps_encryption": False, - }, - ), - ( - t.AddrModeAddress(addr_mode=t.AddrMode.IEEE, address=DEVICE_IEEE), - t.TransmitOptions.NONE, - { - "delivery_mode": "unicast", - # 0xFFFE on the wire means "no short address" - "destination": None, - "destination_eui64": DEVICE_IEEE, - }, - ), - ( - t.AddrModeAddress(addr_mode=t.AddrMode.Group, address=t.Group(0x0002)), - t.TransmitOptions.NONE, - {"delivery_mode": "multicast", "destination": t.NWK(0x0002), "dst_ep": 0}, - ), - ( - t.AddrModeAddress( - addr_mode=t.AddrMode.Broadcast, - address=t.BroadcastAddress.ALL_ROUTERS_AND_COORDINATOR, - ), - t.TransmitOptions.NONE, - {"delivery_mode": "broadcast", "destination": t.NWK(0xFFFC)}, - ), - ], -) -async def test_legacy_send_packet( - legacy_app: ControllerApplication, - legacy_server: SyntheticLegacyZiggurat, - dst: t.AddrModeAddress, - tx_options: t.TransmitOptions, - expected: dict[str, Any], -) -> None: - legacy_app.add_device(DEVICE_IEEE, DEVICE_NWK) - - await legacy_app.send_packet( - t.ZigbeePacket( - src=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=t.NWK(0x0000)), - src_ep=t.uint8_t(1), - dst=dst, - dst_ep=t.uint8_t(1), - tsn=t.uint8_t(33), - profile_id=t.uint16_t(0x0104), - cluster_id=t.uint16_t(0x0006), - data=t.SerializableBytes(b"\x01\x02\x03"), - tx_options=tx_options, - ) - ) - - request = legacy_server.sent(commands.SendAps)[-1] - assert request.data == b"\x01\x02\x03" - assert request.aps_seq == 33 - assert request.radius == 30 - for field, value in expected.items(): - assert getattr(request, field) == value - - -async def test_legacy_send_packet_delivery_failure( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - async def fail(command: commands.SendAps, request_id: int) -> commands.Status: - raise RpcError("transmit_failed", "radio unavailable") - - legacy_server.handlers["send_aps"] = fail - - # The legacy `transmit_failed` code maps to the binary RADIO_ERROR status - with pytest.raises(DeliveryError, match="radio_error"): - await legacy_app.send_packet( - t.ZigbeePacket( - src=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=t.NWK(0x0000)), - src_ep=t.uint8_t(1), - dst=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=DEVICE_NWK), - dst_ep=t.uint8_t(1), - tsn=t.uint8_t(34), - profile_id=t.uint16_t(0x0104), - cluster_id=t.uint16_t(0x0006), - data=t.SerializableBytes(b"\x04"), - ) - ) - - -async def test_legacy_send_confirm_failure( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - """The JSON protocol carries no failure kind, so a failed confirm becomes the - least-wrong binary stand-in.""" - - async def fail_confirm( - command: commands.SendAps, request_id: int - ) -> commands.Status: - await legacy_server.send_confirm(request_id, reason="no_route") - return commands.Status(status="accepted") - - legacy_server.handlers["send_aps"] = fail_confirm - - with pytest.raises(DeliveryError, match="TRANSMIT_FAILED"): - await legacy_app.send_packet( - t.ZigbeePacket( - src=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=t.NWK(0x0000)), - src_ep=t.uint8_t(1), - dst=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=DEVICE_NWK), - dst_ep=t.uint8_t(1), - tsn=t.uint8_t(35), - profile_id=t.uint16_t(0x0104), - cluster_id=t.uint16_t(0x0006), - data=t.SerializableBytes(b"\x05"), - ) - ) - - -async def test_legacy_aps_ack_confirm_failure( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - async def fail_ack(command: commands.SendAps, request_id: int) -> commands.Status: - await legacy_server.send_confirm(request_id) - await legacy_server.aps_ack_confirm(request_id, reason="timeout") - return commands.Status(status="accepted") - - legacy_server.handlers["send_aps"] = fail_ack - - with pytest.raises(DeliveryError, match="APS_ACK_TIMEOUT"): - await legacy_app.send_packet( - t.ZigbeePacket( - src=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=t.NWK(0x0000)), - src_ep=t.uint8_t(1), - dst=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=DEVICE_NWK), - dst_ep=t.uint8_t(1), - tsn=t.uint8_t(36), - profile_id=t.uint16_t(0x0104), - cluster_id=t.uint16_t(0x0006), - data=t.SerializableBytes(b"\x06"), - tx_options=t.TransmitOptions.ACK, - ) - ) - - -async def test_legacy_energy_scan( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - energies = await legacy_app.energy_scan( - # zigpy mis-annotates the classmethod's `cls` as an instance - channels=t.Channels.from_channel_list([11, 15]), # type: ignore[misc] - duration_exp=2, - count=1, - ) - - scan = legacy_server.sent(commands.EnergyScan)[-1] - assert scan.channels == [11, 15] - # 0.016 ms/symbol * 960 symbols * (2**2 + 1) - assert scan.duration_per_channel_ms == 77 - assert sorted(energies) == [11, 15] - - -async def test_legacy_network_scan( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - beacon = commands.NetworkBeaconEvent( - channel=t.uint8_t(11), - source=t.NWK(0x0000), - pan_id=t.PanId(0x1A2B), - extended_pan_id=t.ExtendedPanId(t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11")), - permit_joining=True, - stack_profile=t.uint8_t(2), - protocol_version=t.uint8_t(2), - router_capacity=True, - end_device_capacity=True, - device_depth=t.uint8_t(0), - update_id=t.uint8_t(0), - lqi=t.uint8_t(200), - rssi=t.int8s(-60), - ) - - async def scan(command: commands.NetworkScan, request_id: int) -> commands.Status: - await legacy_server.send_event_data( - request_id, "network_found", beacon.to_dict() - ) - # A beacon whose MAC source was not a short address - await legacy_server.send_event_data( - request_id, - "network_found", - dataclasses.replace(beacon, source=None).to_dict(), - ) - return commands.Status(status="complete") - - legacy_server.handlers["network_scan"] = scan - - found = [ - result - async for result in legacy_app.network_scan( - # zigpy mis-annotates the classmethod's `cls` as an instance - channels=t.Channels.from_channel_list([11]), # type: ignore[misc] - duration_exp=2, - ) - ] - - assert [f.src for f in found] == [t.NWK(0x0000), None] - assert found[0].pan_id == t.PanId(0x1A2B) - assert found[0].lqi == 200 - assert found[0].permit_joining is True - - -async def test_legacy_packet_capture( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - async def capture( - command: commands.PacketCapture, request_id: int - ) -> commands.Status: - await legacy_server.send_event_data( - request_id, - "captured_packet", - commands.CapturedPacketEvent( - channel=t.uint8_t(15), - rssi=t.int8s(-80), - lqi=t.uint8_t(200), - data="aabbcc", - ).to_dict(), - ) - return commands.Status(status="complete") - - legacy_server.handlers["packet_capture"] = capture - - packets = [packet async for packet in legacy_app.packet_capture(15)] - assert len(packets) == 1 - assert packets[0].data == b"\xaa\xbb\xcc" - assert legacy_server.sent(commands.PacketCapture)[0].channel == 15 - - await legacy_app.packet_capture_change_channel(20) - assert legacy_server.sent(commands.PacketCaptureChangeChannel)[0].channel == 20 - - -async def test_legacy_received_aps_notification( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - add_initialized_device(legacy_app) - - # A ZDO request arriving over the wire is answered end-to-end - await legacy_server.send_notification( - commands.ReceivedApsCommand( - source=DEVICE_NWK, - destination=t.NWK(0x0000), - group=None, - profile_id=t.uint16_t(0x0000), - cluster_id=t.uint16_t(zdo_t.ZDOCmd.Node_Desc_req), - src_ep=t.uint8_t(0), - dst_ep=t.uint8_t(0), - lqi=t.uint8_t(255), - rssi=t.int8s(-40), - data=b"\x77" + t.NWK(0x0000).serialize(), - ) - ) - reply = await legacy_server.wait_for(commands.SendAps) - assert reply.cluster_id == zdo_t.ZDOCmd.Node_Desc_rsp - assert reply.data[0] == 0x77 - - # A group-addressed frame carries its group id through the transcoder - await legacy_server.send_notification( - commands.ReceivedApsCommand( - source=DEVICE_NWK, - destination=t.NWK(0x0000), - group=2, - profile_id=t.uint16_t(0x0104), - cluster_id=t.uint16_t(0x0006), - src_ep=t.uint8_t(1), - dst_ep=t.uint8_t(255), - lqi=t.uint8_t(255), - rssi=t.int8s(-40), - data=b"\x01", - ) - ) - await flush(legacy_app) - - -async def test_legacy_frame_counter_notification( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - await legacy_server.send_notification( - commands.FrameCounterUpdate(frame_counter=t.uint32_t(123456)) - ) - await flush(legacy_app) - - assert legacy_app.state.network_info.network_key.tx_counter == 123456 - - -async def test_legacy_link_key_notification( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - await legacy_server.send_notification( - commands.LinkKeyUpdate(ieee=DEVICE_IEEE, key=LINK_KEY) - ) - await flush(legacy_app) - - assert legacy_app.state.network_info.key_table == [ - zigpy.state.Key(key=LINK_KEY, partner_ieee=DEVICE_IEEE) - ] - - -async def test_legacy_device_joined_notification( - legacy_app: ControllerApplication, - legacy_server: SyntheticLegacyZiggurat, - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setattr(application_module, "DEVICE_JOIN_MAX_DELAY", 0.05) - - await legacy_server.send_notification( - commands.DeviceJoined(nwk=DEVICE_NWK, ieee=DEVICE_IEEE, parent=t.NWK(0x0000)) - ) - await flush(legacy_app) - await asyncio.sleep(0.1) - - assert legacy_app.get_device(ieee=DEVICE_IEEE).nwk == DEVICE_NWK - - -async def test_legacy_device_left_notification( - legacy_app: ControllerApplication, legacy_server: SyntheticLegacyZiggurat -) -> None: - left: list[zigpy.device.Device] = [] - - class Listener: - def device_left(self, device: zigpy.device.Device) -> None: - left.append(device) - - legacy_app.add_listener(Listener()) - device = legacy_app.add_device(DEVICE_IEEE, DEVICE_NWK) - - # Each leave reason maps onto its binary counterpart - await legacy_server.send_notification( - commands.DeviceLeft( - nwk=DEVICE_NWK, - ieee=DEVICE_IEEE, - reason=commands.DeviceLeaveReason.ANNOUNCED, - rejoin=False, - ) - ) - await flush(legacy_app) - assert left == [device] - - # A parent router relayed the leave; the IEEE is resolved through the registry - await legacy_server.send_notification( - commands.DeviceLeft( - nwk=DEVICE_NWK, - ieee=None, - reason=commands.DeviceLeaveReason.ROUTER_REPORTED, - router=t.NWK(0x1234), - router_ieee=t.EUI64.convert("bb:bb:bb:bb:bb:bb:bb:bb"), - ) - ) - await flush(legacy_app) - assert left == [device, device] - - await legacy_server.send_notification( - commands.DeviceLeft( - nwk=t.NWK(0xBEEF), - ieee=None, - reason=commands.DeviceLeaveReason.KEEPALIVE_TIMEOUT, - ) - ) - await flush(legacy_app) - assert left == [device, device] - - -async def test_legacy_aps_decryption_failure_notification( - legacy_app: ControllerApplication, - legacy_server: SyntheticLegacyZiggurat, - caplog: pytest.LogCaptureFixture, -) -> None: - with caplog.at_level(logging.WARNING): - await legacy_server.send_notification( - commands.ApsDecryptionFailure( - source=t.NWK(0x1234), - source_ieee=DEVICE_IEEE, - frame_counter=t.uint32_t(42), - # An unknown key id degrades to the network key - key_id="tc_link_key", - ) - ) - await flush(legacy_app) - - assert "Could not decrypt an APS command" in caplog.text - - -# -- transcoding at the transport level ------------------------------------------- - - -async def test_legacy_acknowledges_restore_loads( - legacy_server: SyntheticLegacyZiggurat, -) -> None: - """The legacy server re-learns its topology tables, so the restore loads are - acknowledged locally and never reach it.""" - transport, frames = await _legacy(legacy_server) - try: - loads: list[p.Request] = [ - p.LoadChildren(entries=t.LVList[p.ChildEntry, t.uint16_t]([])), - p.LoadAddressCache(entries=t.LVList[p.AddressEntry, t.uint16_t]([])), - p.LoadRouteTable(entries=t.LVList[p.RouteEntry, t.uint16_t]([])), - p.LoadSourceRoutes(entries=t.LVList[p.SourceRouteEntry, t.uint16_t]([])), - ] - for request_id, load in enumerate(loads, start=1): - await transport.send_frame(p.encode_request(load, request_id)) - - await _wait_for(frames, count=len(loads)) - assert len(frames) == len(loads) - for frame, load in zip(frames, loads, strict=True): - header, body = p.Header.deserialize(frame) - assert header.frame_type == p.FrameType.RESPONSE - assert header.command == load.command - assert body == bytes([p.Status.OK]) - - assert legacy_server.requests == [] - finally: - await transport.disconnect() - - -async def test_legacy_drops_cancel_request( - legacy_server: SyntheticLegacyZiggurat, -) -> None: - """The legacy server has no request-cancel concept, so a best-effort cancel is - dropped rather than sent as an unknown method.""" - transport, frames = await _legacy(legacy_server) - try: - await transport.send_frame( - p.encode_request(p.CancelRequest(request_id=t.uint16_t(7)), 1) - ) - # Nothing is emitted locally and nothing reaches the server - await asyncio.sleep(0.05) - assert frames == [] - assert legacy_server.requests == [] - finally: - await transport.disconnect() - - -async def test_legacy_rejects_untranscodable_command( - legacy_server: SyntheticLegacyZiggurat, -) -> None: - """A binary request with no JSON equivalent fails loudly instead of being - silently dropped.""" - transport, _ = await _legacy(legacy_server) - try: - with pytest.raises(ValueError, match="Cannot transcode"): - await transport.send_frame( - p.encode_request(p.SetTunable.build("aps_ack_timeout", 5), 1) - ) - finally: - await transport.disconnect() - - -async def test_legacy_rejects_unknown_command( - legacy_server: SyntheticLegacyZiggurat, -) -> None: - transport, _ = await _legacy(legacy_server) - try: - # An unknown command byte fails loudly instead of silently vanishing. - frame = p.Header( - command=t.uint8_t(0xEE), - frame_type=p.FrameType.REQUEST, - request_id=t.uint16_t(1), - ).serialize() - with pytest.raises(KeyError): - await transport.send_frame(frame) - finally: - await transport.disconnect() - - -async def test_legacy_firmware_info_via_ping( - legacy_server: SyntheticLegacyZiggurat, -) -> None: - transport, frames = await _legacy(legacy_server) - try: - # The legacy server has no firmware-info call: the shim probes it with a - # JSON `ping` and fabricates the response payload. - await transport.send_frame(p.encode_request(p.GetFirmwareInfo(), 1)) - await legacy_server.wait_for(commands.Ping) - await _wait_for(frames) - header, body = p.Header.deserialize(frames[0]) - assert header.frame_type == p.FrameType.RESPONSE - assert header.command == p.RequestCommand.GET_FIRMWARE_INFO - assert body[0] == p.Status.OK - info = p.FirmwareInfo.deserialize(body[1:])[0] - assert info.protocol_version == p.PROTOCOL_VERSION - finally: - await transport.disconnect() - - -async def test_legacy_decodes_captured_packet( - legacy_server: SyntheticLegacyZiggurat, -) -> None: - transport, frames = await _legacy(legacy_server) - try: - # An unknown event is dropped; the captured packet is transcoded to an event. - await legacy_server.send_event_data(7, "not_a_real_event", {}) - await legacy_server.send_event_data( - 7, - "captured_packet", - {"channel": 15, "rssi": -80, "lqi": 200, "data": "aabbcc"}, - ) - await _wait_for(frames) - assert len(frames) == 1 - header, body = p.Header.deserialize(frames[0]) - assert header.frame_type == p.FrameType.EVENT - assert header.command == p.RequestCommand.PACKET_CAPTURE - packet = p.CapturedPacket.deserialize(body)[0] - assert bytes(packet.psdu) == b"\xaa\xbb\xcc" - finally: - await transport.disconnect() - - -async def test_legacy_forwards_firmware_log( - legacy_server: SyntheticLegacyZiggurat, caplog: pytest.LogCaptureFixture -) -> None: - transport, _ = await _legacy(legacy_server) - try: - with caplog.at_level(logging.WARNING, logger="ziggurat.fw.foo.bar"): - await legacy_server.send_raw( - json.dumps( - { - "type": "notification", - "event": "log", - "data": { - "level": "WARN", - "target": "foo::bar", - "message": "something happened", - }, - } - ) - ) - async with asyncio.timeout(2): - while "something happened" not in caplog.text: - await asyncio.sleep(0.01) - finally: - await transport.disconnect() - - -async def test_legacy_transmitted_becomes_send_confirm( - legacy_server: SyntheticLegacyZiggurat, -) -> None: - transport, frames = await _legacy(legacy_server) - try: - # The real server signals a send handoff with a bare `transmitted` event - # that carries no `data`; it must become a SEND_CONFIRM, not crash. - await legacy_server.send_event(9, "transmitted") - await _wait_for(frames) - header, body = p.Header.deserialize(frames[0]) - assert header.frame_type == p.FrameType.NOTIFICATION - assert header.command == p.NotificationCommand.SEND_CONFIRM - assert header.request_id == 9 - assert p.SendConfirm.deserialize(body)[0].status == p.SendStatus.SUCCESS - finally: - await transport.disconnect() - - -async def test_legacy_decodes_decrypt_failure_known_key( - legacy_server: SyntheticLegacyZiggurat, -) -> None: - transport, frames = await _legacy(legacy_server) - try: - await legacy_server.send_notification( - commands.ApsDecryptionFailure( - source=t.NWK(0x1234), - source_ieee=COORDINATOR_IEEE, - frame_counter=t.uint32_t(42), - key_id="network", - ) - ) - await _wait_for(frames) - header, body = p.Header.deserialize(frames[0]) - assert header.command == p.NotificationCommand.APS_DECRYPT_FAILURE - failure = p.ApsDecryptFailure.deserialize(body)[0] - assert failure.key_id == p.KeyId.NETWORK - finally: - await transport.disconnect() - - -async def test_legacy_ignores_binary_and_unknown_response( - legacy_server: SyntheticLegacyZiggurat, -) -> None: - transport, frames = await _legacy(legacy_server) - try: - # A binary frame and a response for an unknown id are both dropped; a - # following confirm still transcodes, proving the loop kept going. - await legacy_server.ws.send_bytes(b"\x00\x01\x02") - await legacy_server.send_raw( - json.dumps({"type": "response", "id": 9999, "result": {}}) - ) - await legacy_server.send_confirm(1) - await _wait_for(frames) - assert len(frames) == 1 - header, _ = p.Header.deserialize(frames[0]) - assert header.command == p.NotificationCommand.SEND_CONFIRM - finally: - await transport.disconnect() diff --git a/tests/test_transport.py b/tests/test_transport.py index 35459b1..8f0c883 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -1,6 +1,5 @@ """Tests for `connect_transport`, which probes a WebSocket for its protocol, and for -the transports it returns. The legacy JSON transcoding shim is covered separately in -`test_legacy.py`.""" +the transports it returns.""" import asyncio @@ -119,6 +118,24 @@ async def test_spinel_tunnel_write_rejected() -> None: await rcp.stop() +async def test_websocket_frame_handler_error(server: SyntheticZiggurat) -> None: + attempts: list[bytes] = [] + + def boom(frame: bytes) -> None: + attempts.append(frame) + raise RuntimeError("handler blew up") + + transport = await connect_transport(server.url, boom, lambda exc: None) + try: + # The receive loop must survive a handler raising on a delivered frame. + await transport.send_frame(p.encode_request(p.Shutdown(), 1)) + await _wait_for(attempts) + await transport.send_frame(p.encode_request(p.Shutdown(), 2)) + await _wait_for(attempts, count=2) + finally: + await transport.disconnect() + + async def test_websocket_send_after_disconnect(server: SyntheticZiggurat) -> None: transport = await connect_transport( server.url, lambda frame: None, lambda exc: None diff --git a/uv.lock b/uv.lock index 49c01b3..0e1abbd 100644 --- a/uv.lock +++ b/uv.lock @@ -661,18 +661,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ce/62/b40b382fa0c66fee1478073eb8db352a4a6beda4a1adccf1df911d8c289c/librt-0.11.0-cp314-cp314t-win_arm64.whl", hash = "sha256:dee008f20b542e3cd162ba338a7f9ec0f6d23d395f66fe8aeeec3c9d067ea253", size = 102572, upload-time = "2026-05-10T18:17:06.809Z" }, ] -[[package]] -name = "mashumaro" -version = "3.22" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "typing-extensions" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/d1/e3/a06dcd2e6df094c5e294721926f1f76da30f50823e2ac233a9198e804891/mashumaro-3.22.tar.gz", hash = "sha256:64538cc365204402a060ebde683a86505b5a4344acf6870d79021e9fbfe57360", size = 197845, upload-time = "2026-05-26T14:39:21.859Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/9b/1c/92fd926c2e7763535454683250dbdd8d10aa2f2c62f58d6abbcce4d8b3fc/mashumaro-3.22-py3-none-any.whl", hash = "sha256:17dc4d7294c33ef380a8b929dda0608577aa2141988c00a0c4932310108fe71d", size = 95916, upload-time = "2026-05-26T14:39:20.233Z" }, -] - [[package]] name = "multidict" version = "6.7.1" @@ -1511,7 +1499,6 @@ source = { editable = "." } dependencies = [ { name = "aiohttp" }, { name = "aiospinel" }, - { name = "mashumaro" }, { name = "zigpy" }, ] @@ -1545,7 +1532,6 @@ testing = [ requires-dist = [ { name = "aiohttp" }, { name = "aiospinel", specifier = ">=1.2.0" }, - { name = "mashumaro" }, { name = "zigpy" }, ] diff --git a/zigpy_ziggurat/zigbee/legacy.py b/zigpy_ziggurat/zigbee/legacy.py deleted file mode 100644 index ba0be3c..0000000 --- a/zigpy_ziggurat/zigbee/legacy.py +++ /dev/null @@ -1,433 +0,0 @@ -"""Legacy JSON-RPC wire protocol for the WebSocket transport.""" - -from dataclasses import dataclass -import enum -from typing import ClassVar, Generic, TypeVar - -from mashumaro import DataClassDictMixin -from mashumaro.config import BaseConfig -from mashumaro.types import SerializationStrategy -import zigpy.types as t - - -class BigEndianHexNwk(SerializationStrategy): - """`1a2b`-style hex, the network address format of requests and responses.""" - - def serialize(self, value: t.NWK) -> str: - return f"{int(value):04x}" - - def deserialize(self, value: str) -> t.NWK: - return t.NWK(int(value, 16)) - - -class BigEndianHexPanId(SerializationStrategy): - def serialize(self, value: t.PanId) -> str: - return f"{int(value):04x}" - - def deserialize(self, value: str) -> t.PanId: - return t.PanId(int(value, 16)) - - -class LittleEndianHexNwk(SerializationStrategy): - """`2b1a`-style hex, the network address format of notifications.""" - - def serialize(self, value: t.NWK) -> str: - return value.serialize().hex() - - def deserialize(self, value: str) -> t.NWK: - return t.NWK.deserialize(bytes.fromhex(value))[0] - - -class ColonHexEui64(SerializationStrategy): - def serialize(self, value: t.EUI64) -> str: - return str(value) - - def deserialize(self, value: str) -> t.EUI64: - return t.EUI64.convert(value) - - -class ColonHexExtendedPanId(SerializationStrategy): - def serialize(self, value: t.ExtendedPanId) -> str: - return str(value) - - def deserialize(self, value: str) -> t.ExtendedPanId: - return t.ExtendedPanId(t.EUI64.convert(value)) - - -class ColonHexKey(SerializationStrategy): - def serialize(self, value: t.KeyData) -> str: - return str(value) - - def deserialize(self, value: str) -> t.KeyData: - return t.KeyData.convert(value) - - -class HexBytes(SerializationStrategy): - def serialize(self, value: bytes) -> str: - return value.hex() - - def deserialize(self, value: str) -> bytes: - return bytes.fromhex(value) - - -class SizedInt(SerializationStrategy): - """Plain JSON integers, validated into zigpy's sized integer types.""" - - def __init__(self, int_type: type[int]) -> None: - self._int_type = int_type - - def serialize(self, value: int) -> int: - return int(value) - - def deserialize(self, value: int) -> int: - return self._int_type(value) - - -class _WireConfig(BaseConfig): - serialization_strategy = { - t.NWK: BigEndianHexNwk(), - t.PanId: BigEndianHexPanId(), - t.EUI64: ColonHexEui64(), - t.ExtendedPanId: ColonHexExtendedPanId(), - t.KeyData: ColonHexKey(), - bytes: HexBytes(), - t.uint8_t: SizedInt(t.uint8_t), - t.uint16_t: SizedInt(t.uint16_t), - t.uint32_t: SizedInt(t.uint32_t), - t.int8s: SizedInt(t.int8s), - } - - -class _NotificationConfig(_WireConfig): - serialization_strategy = { - **_WireConfig.serialization_strategy, - t.NWK: LittleEndianHexNwk(), - } - - -@dataclass -class WireModel(DataClassDictMixin): - class Config(_WireConfig): ... - - -@dataclass -class Response(WireModel): ... - - -@dataclass -class Status(Response): - status: str - - -RESPONSE_T = TypeVar("RESPONSE_T", bound=Response) -EVENT_T = TypeVar("EVENT_T", bound=Response) - - -@dataclass -class Request(WireModel, Generic[RESPONSE_T]): - method: ClassVar[str] - response_type: ClassVar[type[Response]] - - -@dataclass -class StreamingRequest(Request[RESPONSE_T], Generic[RESPONSE_T, EVENT_T]): - """A request answered by a stream of `event_name` events (each an `event_type`) - before the terminal `response_type`.""" - - event_type: ClassVar[type[Response]] - event_name: ClassVar[str] - - -@dataclass -class KeyTableEntry(WireModel): - partner_ieee: t.EUI64 - key: t.KeyData - - -@dataclass -class Ping(Request[Status]): - method = "ping" - response_type = Status - - -@dataclass -class Configure(Request[Status]): - method = "configure" - response_type = Status - - channel: int - nwk_update_id: int - pan_id: t.PanId - extended_pan_id: t.ExtendedPanId - nwk_address: t.NWK - ieee_address: t.EUI64 - network_key: t.KeyData - network_key_seq: int - network_key_tx_counter: int - tc_link_key: t.KeyData - source_routing: bool - # None means "pick automatically": the server applies its safe default - tx_power: int | None - # Unique trust center link keys negotiated in earlier sessions - key_table: list[KeyTableEntry] - # A TCLK seed carried over from a microcontroller stack, passed verbatim as the - # source stack's plain hex string. Requires `tclk_flavor`. - tclk_seed: str | None - tclk_flavor: str | None - - aps_frame_counter: int = 0 - started: bool = False - - -@dataclass -class NetworkInfo(Response): - channel: t.uint8_t - nwk_update_id: t.uint8_t - pan_id: t.PanId - extended_pan_id: t.ExtendedPanId - nwk_address: t.NWK - ieee_address: t.EUI64 - network_key: t.KeyData - network_key_seq: t.uint8_t - network_key_tx_counter: t.uint32_t - tc_link_key: t.KeyData - tx_power: int - tclk_seed: str | None - tclk_flavor: str | None - key_table: list[KeyTableEntry] - - aps_frame_counter: int = 0 - started: bool = False - - -@dataclass -class GetNetworkInfo(Request[NetworkInfo]): - method = "get_network_info" - response_type = NetworkInfo - - -@dataclass -class HwAddress(Response): - ieee_address: t.EUI64 - - -@dataclass -class GetHwAddress(Request[HwAddress]): - method = "get_hw_address" - response_type = HwAddress - - -@dataclass -class SendAps(Request[Status]): - method = "send_aps" - response_type = Status - - delivery_mode: str - # Resolved by the server through its address map; takes precedence over - # `destination` and selects the link key when `aps_encryption` is set - destination_eui64: t.EUI64 | None - destination: t.NWK | None - profile_id: int - cluster_id: int - src_ep: int - dst_ep: int - aps_ack: bool - aps_seq: int - radius: int - aps_encryption: bool - priority: int - data: bytes - - -@dataclass -class EnergyScanResult(Response): - channel: t.uint8_t - rssi: t.int8s - - -@dataclass -class EnergyScan(StreamingRequest[Status, EnergyScanResult]): - method = "energy_scan" - response_type = Status - event_type = EnergyScanResult - event_name = "energy_result" - - channels: list[int] - duration_per_channel_ms: int - - -@dataclass -class NetworkBeaconEvent(Response): - channel: t.uint8_t - # Absent when the beacon's MAC source was not a short address - source: t.NWK | None - pan_id: t.PanId - extended_pan_id: t.ExtendedPanId - permit_joining: bool - stack_profile: t.uint8_t - protocol_version: t.uint8_t - router_capacity: bool - end_device_capacity: bool - device_depth: t.uint8_t - update_id: t.uint8_t - lqi: t.uint8_t - rssi: t.int8s - - -@dataclass -class NetworkScan(StreamingRequest[Status, NetworkBeaconEvent]): - method = "network_scan" - response_type = Status - event_type = NetworkBeaconEvent - event_name = "network_found" - - channels: list[int] - duration_per_channel_ms: int - - -@dataclass -class PermitJoins(Request[Status]): - method = "permit_joins" - response_type = Status - - duration: int - # Whether the coordinator also opens its own beacon for direct joins. False opens - # only the trust center's authorization window, steering joins through routers. - accept_direct_joins: bool = True - - -@dataclass -class SetProvisionalKey(Request[Status]): - method = "set_provisional_key" - response_type = Status - - ieee: t.EUI64 - key: t.KeyData - - -@dataclass -class SetChannel(Request[Status]): - method = "set_channel" - response_type = Status - - channel: int - - -@dataclass -class CapturedPacketEvent(Response): - channel: t.uint8_t - rssi: t.int8s - lqi: t.uint8_t - # Hex-encoded 802.15.4 MAC frame (FCS stripped) - data: str - - -@dataclass -class PacketCapture(StreamingRequest[Status, CapturedPacketEvent]): - method = "packet_capture" - response_type = Status - event_type = CapturedPacketEvent - event_name = "captured_packet" - - channel: int - - -@dataclass -class PacketCaptureChangeChannel(Request[Status]): - method = "packet_capture_change_channel" - response_type = Status - - channel: int - - -@dataclass -class SetNwkUpdateId(Request[Status]): - method = "set_nwk_update_id" - response_type = Status - - nwk_update_id: int - - -@dataclass -class Notification(DataClassDictMixin): - class Config(_NotificationConfig): ... - - -@dataclass -class ReceivedApsCommand(Notification): - source: t.NWK - destination: t.NWK - group: int | None - profile_id: t.uint16_t - cluster_id: t.uint16_t - src_ep: t.uint8_t - dst_ep: t.uint8_t - lqi: t.uint8_t - rssi: t.int8s - data: bytes - - -@dataclass -class FrameCounterUpdate(Notification): - frame_counter: t.uint32_t - - -@dataclass -class LinkKeyUpdate(Notification): - ieee: t.EUI64 - key: t.KeyData - - -@dataclass -class DeviceJoined(Notification): - nwk: t.NWK - ieee: t.EUI64 - parent: t.NWK - - -class DeviceLeaveReason(enum.StrEnum): - """How the server learned that a device left the network.""" - - # The device itself broadcast a NWK Leave announcement (`rejoin` is set) - ANNOUNCED = "announced" - # A parent router relayed an APS Update-Device "Device Left" (`router`/ - # `router_ieee` are set) - ROUTER_REPORTED = "router_reported" - # A sleepy child aged out of the neighbor table without a keepalive - KEEPALIVE_TIMEOUT = "keepalive_timeout" - - -@dataclass -class DeviceLeft(Notification): - nwk: t.NWK - # Unknown when the leaving device never made it into the server's address map - ieee: t.EUI64 | None - # How the server learned of the departure - reason: DeviceLeaveReason - # Set only for ANNOUNCED: whether the device intends to rejoin - rejoin: bool | None = None - # Set only for ROUTER_REPORTED: the router that relayed the leave. The EUI64 is - # unknown when the server could not resolve it from its address map. - router: t.NWK | None = None - router_ieee: t.EUI64 | None = None - - -@dataclass -class ApsDecryptionFailure(Notification): - # An APS command frame from this device could not be decrypted with any key the - # server holds. Its link key is almost certainly wrong or missing, which also - # blocks joins routed through it (the trust center can't read its Update-Device). - source: t.NWK - source_ieee: t.EUI64 - frame_counter: t.uint32_t - key_id: str - - -NOTIFICATIONS: dict[str, type[Notification]] = { - "received_aps_command": ReceivedApsCommand, - "frame_counter_update": FrameCounterUpdate, - "link_key_update": LinkKeyUpdate, - "device_joined": DeviceJoined, - "device_left": DeviceLeft, - "aps_decryption_failure": ApsDecryptionFailure, -} diff --git a/zigpy_ziggurat/zigbee/transport.py b/zigpy_ziggurat/zigbee/transport.py index 1830af4..4c7c59d 100644 --- a/zigpy_ziggurat/zigbee/transport.py +++ b/zigpy_ziggurat/zigbee/transport.py @@ -1,20 +1,15 @@ -"""Frame transports for the binary protocol: serial (Spinel tunnel), binary -WebSocket, and a JSON-transcoding WebSocket for early users on the legacy server.""" +"""Frame transports for the binary protocol: serial (Spinel tunnel) and WebSocket.""" from __future__ import annotations import asyncio from collections.abc import Callable -import json import logging from typing import Any, Protocol, cast import aiohttp import aiospinel import zigpy.serial -import zigpy.types as t - -from zigpy_ziggurat.zigbee import legacy, protocol as p _LOGGER = logging.getLogger(__name__) @@ -177,30 +172,21 @@ async def _open_websocket( async def _probe_websocket(url: str, on_frame: OnFrame, on_lost: OnLost) -> Transport: - """Pick the transport from the server's opening hello: binary frame or JSON text.""" + """Open the WebSocket and check that the opening hello is a binary frame.""" session, websocket = await _open_websocket(url) async with asyncio.timeout(HANDSHAKE_TIMEOUT): hello = await websocket.receive() - if hello.type == aiohttp.WSMsgType.BINARY: - transport: _WebSocketBase = WebSocketTransport(on_frame, on_lost) - _LOGGER.debug("Detected binary WebSocket protocol") - elif hello.type == aiohttp.WSMsgType.TEXT: - transport = LegacyWebSocketTransport(on_frame, on_lost) - _LOGGER.debug("Detected legacy JSON WebSocket protocol: %s", hello.data) - _LOGGER.warning( - "The legacy JSON WebSocket protocol will be removed soon. Please upgrade" - " the Ziggurat app to switch to the new binary protocol." - ) - else: + if hello.type != aiohttp.WSMsgType.BINARY: await session.close() raise ConnectionError(f"Unexpected handshake from ziggurat: {hello!r}") + transport = WebSocketTransport(on_frame, on_lost) transport._adopt(session, websocket) return transport -class _WebSocketBase: +class WebSocketTransport: """Shared aiohttp WebSocket plumbing, driven from a socket passed to `_adopt`.""" def __init__(self, on_frame: OnFrame, on_lost: OnLost) -> None: @@ -236,11 +222,11 @@ async def _receive_loop(self) -> None: exc: BaseException | None = None try: async for msg in websocket: - if msg.type in (aiohttp.WSMsgType.TEXT, aiohttp.WSMsgType.BINARY): + if msg.type == aiohttp.WSMsgType.BINARY: try: - self._handle_message(msg) + self._on_frame(msg.data) except Exception: - _LOGGER.exception("Failed to handle message: %r", msg.data) + _LOGGER.exception("Failed to handle frame: %r", msg.data) elif msg.type == aiohttp.WSMsgType.ERROR: exc = websocket.exception() break @@ -250,604 +236,7 @@ async def _receive_loop(self) -> None: raise self._on_lost(exc) - def _handle_message(self, msg: aiohttp.WSMessage) -> None: - raise NotImplementedError - async def send_frame(self, frame: bytes) -> None: - raise NotImplementedError - - async def _send(self, data: bytes | str) -> None: if self._websocket is None: raise ConnectionError("Not connected") - if isinstance(data, str): - await self._websocket.send_str(data) - else: - await self._websocket.send_bytes(data) - - -class WebSocketTransport(_WebSocketBase): - """The binary protocol carried as WebSocket binary frames.""" - - def _handle_message(self, msg: aiohttp.WSMessage) -> None: - if msg.type == aiohttp.WSMsgType.BINARY: - self._on_frame(msg.data) - - async def send_frame(self, frame: bytes) -> None: - await self._send(frame) - - -# JSON error code -> binary status -_STATUS_BY_CODE: dict[str, p.Status] = { - "parse": p.Status.MALFORMED_PAYLOAD, - "unknown_command": p.Status.UNKNOWN_COMMAND, - # The legacy server's lone state error is a load after the network started. - "invalid_state": p.Status.ALREADY_STARTED, - "not_configured": p.Status.NOT_CONFIGURED, - "radio_error": p.Status.RADIO_ERROR, - "network_start_failed": p.Status.NETWORK_START_FAILED, - "transmit_failed": p.Status.RADIO_ERROR, - "scan_failed": p.Status.SCAN_FAILED, - "invalid_request": p.Status.INVALID_REQUEST, -} - -_LEAVE_REASONS: dict[legacy.DeviceLeaveReason, p.LeaveReason] = { - legacy.DeviceLeaveReason.ANNOUNCED: p.LeaveReason.ANNOUNCED, - legacy.DeviceLeaveReason.ROUTER_REPORTED: p.LeaveReason.ROUTER_REPORTED, - legacy.DeviceLeaveReason.KEEPALIVE_TIMEOUT: p.LeaveReason.KEEPALIVE_TIMEOUT, -} - -_RUST_LOG_LEVELS = { - "ERROR": logging.ERROR, - "WARN": logging.WARNING, - "INFO": logging.INFO, - "DEBUG": logging.DEBUG, - "TRACE": 5, -} - - -class LegacyWebSocketTransport(_WebSocketBase): - """Transcodes the binary protocol to/from the legacy JSON-RPC server.""" - - def __init__(self, on_frame: OnFrame, on_lost: OnLost) -> None: - super().__init__(on_frame, on_lost) - # request id -> command, so a JSON response builds the right binary reply - self._pending_commands: dict[int, p.RequestCommand] = {} - # The binary protocol splits `configure` (Configure + LoadKeyTable* + - # StartNetwork) that the JSON server takes as one call; coalesce it. - self._pending_configure: p.Configure | None = None - self._pending_keys: list[p.KeyEntry] = [] - # The key table the JSON get_network_info returns inline, replayed as the - # events of the ScanKeyTable that follows on the binary side. - self._scan_keys: list[p.KeyEntry] = [] - - # -- outbound: binary frame -> JSON request ------------------------------------ - - async def send_frame(self, frame: bytes) -> None: - header, body = p.Header.deserialize(frame) - command = p.RequestCommand(header.command) - request_id = int(header.request_id) - request = p.REQUESTS[command].deserialize(body)[0] - - if command in (p.RequestCommand.SHUTDOWN, p.RequestCommand.RESET): - # The legacy server has neither shutdown nor reset; it replaces the stack - # on `configure`. OK them locally so callers don't depend on either. - self._emit_ok(command, request_id) - elif command == p.RequestCommand.CONFIGURE: - self._pending_configure = cast(p.Configure, request) - self._pending_keys = [] - self._emit_ok(command, request_id) - elif command == p.RequestCommand.LOAD_KEY_TABLE: - self._pending_keys.extend(cast(p.LoadKeyTable, request).entries) - self._emit_ok(command, request_id) - elif command in ( - p.RequestCommand.LOAD_CHILDREN, - p.RequestCommand.LOAD_ADDRESS_CACHE, - p.RequestCommand.LOAD_ROUTE_TABLE, - p.RequestCommand.LOAD_SOURCE_ROUTES, - ): - # The legacy server re-learns its topology tables, so acknowledge these - # restore loads locally and drop them. - self._emit_ok(command, request_id) - elif command == p.RequestCommand.START_NETWORK: - assert self._pending_configure is not None - params = self._configure_params(self._pending_configure, self._pending_keys) - self._pending_configure = None - self._pending_keys = [] - self._pending_commands[request_id] = command - await self._send_json(request_id, "configure", params) - elif command == p.RequestCommand.GET_NETWORK_INFO: - self._pending_commands[request_id] = command - await self._send_json(request_id, "get_network_info", {}) - elif command == p.RequestCommand.SCAN_KEY_TABLE: - for entry in self._scan_keys: - self._emit(p.FrameType.EVENT, command, request_id, entry.serialize()) - count = p.ScanCount(count=t.uint16_t(len(self._scan_keys))) - self._emit_ok(command, request_id, count) - self._scan_keys = [] - elif command in ( - p.RequestCommand.SCAN_CHILDREN, - p.RequestCommand.SCAN_ADDRESS_CACHE, - p.RequestCommand.SCAN_ROUTE_TABLE, - ): - # The JSON server surfaces only the key table (inline in get_network_info); - # it has no children/address/route scans, so these stream empty. The app - # re-learns that topology from join notifications during the transition. - self._emit_ok(command, request_id, p.ScanCount(count=t.uint16_t(0))) - elif command == p.RequestCommand.CANCEL_REQUEST: - # The legacy server has no request-cancel concept, so the best-effort - # cancel from `ZigguratApi._cancel_send` is dropped here. - pass - else: - method, params = self._encode_request(command, request) - self._pending_commands[request_id] = command - await self._send_json(request_id, method, params) - - async def _send_json( - self, request_id: int, method: str, params: dict[str, Any] - ) -> None: - await self._send( - json.dumps({"id": request_id, "method": method, "params": params}) - ) - - def _encode_request( - self, command: p.RequestCommand, request: p.Request - ) -> tuple[str, dict[str, Any]]: - if command == p.RequestCommand.GET_FIRMWARE_INFO: - # The legacy server has no firmware-info call; `ping` keeps the liveness - # probe end-to-end and the response is fabricated in `_handle_response`. - return "ping", {} - if command == p.RequestCommand.GET_HW_ADDRESS: - return "get_hw_address", {} - if command == p.RequestCommand.PERMIT_JOINS: - permit = cast(p.PermitJoins, request) - return ( - "permit_joins", - legacy.PermitJoins( - duration=int(permit.duration), - accept_direct_joins=bool(permit.accept_direct_joins), - ).to_dict(), - ) - if command == p.RequestCommand.SET_CHANNEL: - channel = int(cast(p.SetChannel, request).channel) - return "set_channel", legacy.SetChannel(channel=channel).to_dict() - if command == p.RequestCommand.SET_NWK_UPDATE_ID: - update_id = int(cast(p.SetNwkUpdateId, request).nwk_update_id) - return ( - "set_nwk_update_id", - legacy.SetNwkUpdateId(nwk_update_id=update_id).to_dict(), - ) - if command == p.RequestCommand.SET_PROVISIONAL_KEY: - key = cast(p.SetProvisionalKey, request) - return ( - "set_provisional_key", - legacy.SetProvisionalKey(ieee=key.ieee, key=key.key).to_dict(), - ) - if command == p.RequestCommand.ENERGY_SCAN: - scan = cast(p.EnergyScan, request) - return ( - "energy_scan", - legacy.EnergyScan( - channels=[int(c) for c in scan.channels], - duration_per_channel_ms=int(scan.duration_per_channel_ms), - ).to_dict(), - ) - if command == p.RequestCommand.NETWORK_SCAN: - net_scan = cast(p.NetworkScan, request) - return ( - "network_scan", - legacy.NetworkScan( - channels=[int(c) for c in net_scan.channels], - duration_per_channel_ms=int(net_scan.duration_per_channel_ms), - ).to_dict(), - ) - if command == p.RequestCommand.PACKET_CAPTURE: - channel = int(cast(p.PacketCapture, request).channel) - return "packet_capture", legacy.PacketCapture(channel=channel).to_dict() - if command == p.RequestCommand.PACKET_CAPTURE_CHANNEL: - channel = int(cast(p.PacketCaptureChannel, request).channel) - return ( - "packet_capture_change_channel", - legacy.PacketCaptureChangeChannel(channel=channel).to_dict(), - ) - if command == p.RequestCommand.SEND_UNICAST: - return "send_aps", self._send_unicast_params(cast(p.SendUnicast, request)) - if command == p.RequestCommand.SEND_BROADCAST: - return "send_aps", self._send_broadcast_params( - cast(p.SendBroadcast, request) - ) - if command == p.RequestCommand.SEND_GROUPCAST: - return "send_aps", self._send_groupcast_params( - cast(p.SendGroupcast, request) - ) - raise ValueError(f"Cannot transcode {command!r} to JSON") - - def _send_unicast_params(self, request: p.SendUnicast) -> dict[str, Any]: - destination = ( - None if request.destination == t.NWK(0xFFFE) else t.NWK(request.destination) - ) - return legacy.SendAps( - delivery_mode="unicast", - destination_eui64=request.destination_eui64 if request.has_eui64 else None, - destination=destination, - profile_id=int(request.profile_id), - cluster_id=int(request.cluster_id), - src_ep=int(request.src_ep), - dst_ep=int(request.dst_ep), - aps_ack=bool(request.aps_ack), - aps_seq=int(request.aps_seq), - radius=int(request.radius), - aps_encryption=bool(request.aps_encryption), - priority=int(request.priority), - data=bytes(request.asdu), - ).to_dict() - - def _send_broadcast_params(self, request: p.SendBroadcast) -> dict[str, Any]: - return legacy.SendAps( - delivery_mode="broadcast", - destination_eui64=None, - destination=t.NWK(request.destination), - profile_id=int(request.profile_id), - cluster_id=int(request.cluster_id), - src_ep=int(request.src_ep), - dst_ep=int(request.dst_ep), - aps_ack=False, - aps_seq=int(request.aps_seq), - radius=int(request.radius), - aps_encryption=False, - priority=int(request.priority), - data=bytes(request.asdu), - ).to_dict() - - def _send_groupcast_params(self, request: p.SendGroupcast) -> dict[str, Any]: - # The legacy server carried the group id in `destination` for a multicast. - return legacy.SendAps( - delivery_mode="multicast", - destination_eui64=None, - destination=t.NWK(request.group_id), - profile_id=int(request.profile_id), - cluster_id=int(request.cluster_id), - src_ep=int(request.src_ep), - dst_ep=0, - aps_ack=False, - aps_seq=int(request.aps_seq), - radius=int(request.radius), - aps_encryption=False, - priority=int(request.priority), - data=bytes(request.asdu), - ).to_dict() - - def _configure_params( - self, configure: p.Configure, keys: list[p.KeyEntry] - ) -> dict[str, Any]: - state = configure.state - seed = bytes(state.tclk_seed).hex() if state.has_tclk_seed else None - flavor = None - if state.has_tclk_seed: - flavor = "zstack" if state.tclk_flavor == p.TclkFlavorId.Z_STACK else "ezsp" - return legacy.Configure( - channel=int(state.channel), - nwk_update_id=int(state.nwk_update_id), - pan_id=state.pan_id, - extended_pan_id=state.extended_pan_id, - nwk_address=state.nwk_address, - ieee_address=state.ieee_address, - network_key=state.network_key, - network_key_seq=int(state.network_key_seq), - network_key_tx_counter=int(state.network_key_tx_counter), - tc_link_key=state.tc_link_key, - source_routing=bool(configure.source_routing), - tx_power=int(state.tx_power), - key_table=[ - legacy.KeyTableEntry(partner_ieee=k.partner_ieee, key=k.key) - for k in keys - ], - tclk_seed=seed, - tclk_flavor=flavor, - aps_frame_counter=int(state.aps_frame_counter), - ).to_dict() - - # -- inbound: JSON message -> binary frame ------------------------------------- - - def _handle_message(self, msg: aiohttp.WSMessage) -> None: - if msg.type != aiohttp.WSMsgType.TEXT: - return - message = json.loads(msg.data) - kind = message["type"] - if kind == "response": - self._handle_response(message) - elif kind == "event": - self._handle_event(message) - elif kind == "notification": - self._handle_notification(message) - - def _handle_response(self, message: dict[str, Any]) -> None: - request_id = message["id"] - if request_id not in self._pending_commands: - return - command = self._pending_commands.pop(request_id) - - if "error" in message: - error = message["error"] - code = error["code"] - # A JSON code with no binary status (a host-side failure the firmware - # can't produce) degrades to a generic invalid-request. - status = _STATUS_BY_CODE.get(code, p.Status.INVALID_REQUEST) - # The binary protocol carries only the status; the diagnostic text - # becomes a log line, like the binary server's own warnings. - _LOGGER.warning( - "Legacy server error for %r (id=%d): %s: %s", - command, - request_id, - code, - error["message"], - ) - self._emit(p.FrameType.RESPONSE, command, request_id, bytes([status])) - elif command == p.RequestCommand.GET_FIRMWARE_INFO: - # Transcoded to a JSON `ping`, which has no result: fabricate the payload. - self._emit_ok( - command, - request_id, - p.FirmwareInfo( - protocol_version=t.uint8_t(p.PROTOCOL_VERSION), - version=t.LongCharacterString("ziggurat/legacy"), - ), - ) - elif command == p.RequestCommand.GET_NETWORK_INFO: - self._emit_ok(command, request_id, self._network_info(message["result"])) - elif command == p.RequestCommand.GET_HW_ADDRESS: - hw = legacy.HwAddress.from_dict(message["result"]) - self._emit_ok(command, request_id, p.HwAddress(ieee=hw.ieee_address)) - else: - self._emit_ok(command, request_id) - - def _handle_event(self, message: dict[str, Any]) -> None: - request_id = message["id"] - event = message["event"] - if event == "transmitted": - # The legacy send handoff, delivered as a bare event; the binary protocol - # models it as a `send_confirm` notification keyed by request id. - self._emit_notification( - p.NotificationCommand.SEND_CONFIRM, - request_id, - p.SendConfirm(status=p.SendStatus.SUCCESS), - ) - return - if event == "energy_result": - result = legacy.EnergyScanResult.from_dict(message["data"]) - payload: p.Response = p.EnergyResult( - channel=t.uint8_t(result.channel), rssi=t.int8s(result.rssi) - ) - command = p.RequestCommand.ENERGY_SCAN - elif event == "network_found": - payload = self._beacon(message["data"]) - command = p.RequestCommand.NETWORK_SCAN - elif event == "captured_packet": - packet = legacy.CapturedPacketEvent.from_dict(message["data"]) - payload = p.CapturedPacket( - channel=t.uint8_t(packet.channel), - rssi=t.int8s(packet.rssi), - lqi=t.uint8_t(packet.lqi), - psdu=t.LongOctetString(bytes.fromhex(packet.data)), - ) - command = p.RequestCommand.PACKET_CAPTURE - else: - # `accepted` and any other bare event have no binary equivalent. - return - self._emit(p.FrameType.EVENT, command, request_id, payload.serialize()) - - def _handle_notification(self, message: dict[str, Any]) -> None: - event = message["event"] - data = message["data"] - if event == "log": - self._handle_log(data) - elif event == "send_confirm": - self._emit_notification( - p.NotificationCommand.SEND_CONFIRM, data["id"], self._send_confirm(data) - ) - elif event == "aps_ack_confirm": - self._emit_notification( - p.NotificationCommand.APS_ACK_CONFIRM, - data["id"], - self._aps_ack_confirm(data), - ) - elif event == "received_aps_command": - self._emit_notification( - p.NotificationCommand.RECEIVED_APS, 0, self._received_aps(data) - ) - elif event == "frame_counter_update": - counter = legacy.FrameCounterUpdate.from_dict(data) - self._emit_notification( - p.NotificationCommand.FRAME_COUNTER, - 0, - p.FrameCounter(frame_counter=t.uint32_t(counter.frame_counter)), - ) - elif event == "link_key_update": - link = legacy.LinkKeyUpdate.from_dict(data) - self._emit_notification( - p.NotificationCommand.LINK_KEY, - 0, - p.LinkKey(ieee=link.ieee, key=link.key), - ) - elif event == "device_joined": - joined = legacy.DeviceJoined.from_dict(data) - self._emit_notification( - p.NotificationCommand.DEVICE_JOINED, - 0, - p.DeviceJoined( - nwk=joined.nwk, - ieee=joined.ieee, - parent=joined.parent, - # The legacy JSON protocol carries no capability information - rx_on_when_idle=t.uint1_t(1), - device_type=p.ChildDeviceType.UNKNOWN, - reserved=t.uint5_t(0), - ), - ) - elif event == "device_left": - self._emit_notification( - p.NotificationCommand.DEVICE_LEFT, 0, self._device_left(data) - ) - elif event == "aps_decryption_failure": - self._emit_notification( - p.NotificationCommand.APS_DECRYPT_FAILURE, - 0, - self._aps_decrypt_failure(data), - ) - - def _handle_log(self, data: dict[str, Any]) -> None: - level = _RUST_LOG_LEVELS.get(data["level"], logging.INFO) - logger = logging.getLogger("ziggurat.fw." + data["target"].replace("::", ".")) - logger.log(level, "%s", data["message"]) - - # -- inbound payload builders -------------------------------------------------- - - def _network_info(self, result: dict[str, Any]) -> p.NetworkInfo: - info = legacy.NetworkInfo.from_dict(result) - self._scan_keys = [ - p.KeyEntry( - key=entry.key, - tx_counter=t.uint32_t(0), - rx_counter=t.uint32_t(0), - seq=t.uint8_t(0), - partner_ieee=entry.partner_ieee, - ) - for entry in info.key_table - ] - seed = info.tclk_seed - state = p.NetworkState( - channel=t.uint8_t(info.channel), - nwk_update_id=t.uint8_t(info.nwk_update_id), - pan_id=info.pan_id, - extended_pan_id=info.extended_pan_id, - nwk_address=info.nwk_address, - ieee_address=info.ieee_address, - network_key=info.network_key, - network_key_seq=t.uint8_t(info.network_key_seq), - network_key_tx_counter=t.uint32_t(info.network_key_tx_counter), - tc_link_key=info.tc_link_key, - has_tclk_seed=t.Bool(seed is not None), - tclk_seed=t.KeyData(bytes.fromhex(seed) if seed is not None else bytes(16)), - tclk_flavor=( - p.TclkFlavorId.Z_STACK - if info.tclk_flavor == "zstack" - else p.TclkFlavorId.EZSP - ), - tx_power=t.int8s(info.tx_power), - aps_frame_counter=t.uint32_t(info.aps_frame_counter), - ) - return p.NetworkInfo( - state=state, - key_count=t.uint16_t(len(info.key_table)), - started=t.Bool(info.started), - ) - - def _beacon(self, data: dict[str, Any]) -> p.Beacon: - beacon = legacy.NetworkBeaconEvent.from_dict(data) - return p.Beacon( - channel=t.uint8_t(beacon.channel), - source=beacon.source if beacon.source is not None else t.NWK(0xFFFF), - pan_id=beacon.pan_id, - extended_pan_id=beacon.extended_pan_id, - permit_joining=t.uint1_t(beacon.permit_joining), - router_capacity=t.uint1_t(beacon.router_capacity), - end_device_capacity=t.uint1_t(beacon.end_device_capacity), - reserved=t.uint5_t(0), - stack_profile=t.uint8_t(beacon.stack_profile), - protocol_version=t.uint8_t(beacon.protocol_version), - device_depth=t.uint8_t(beacon.device_depth), - update_id=t.uint8_t(beacon.update_id), - lqi=t.uint8_t(beacon.lqi), - rssi=t.int8s(beacon.rssi), - ) - - def _send_confirm(self, data: dict[str, Any]) -> p.SendConfirm: - return p.SendConfirm( - # The legacy JSON protocol carries no failure kind; a transmit failure is - # the least-wrong stand-in. - status=( - p.SendStatus.SUCCESS - if data["status"] == "confirmed" - else p.SendStatus.TRANSMIT_FAILED - ), - ) - - def _aps_ack_confirm(self, data: dict[str, Any]) -> p.ApsAckConfirm: - return p.ApsAckConfirm( - status=( - p.SendStatus.SUCCESS - if data["status"] == "confirmed" - else p.SendStatus.APS_ACK_TIMEOUT - ), - ) - - def _received_aps(self, data: dict[str, Any]) -> p.ReceivedAps: - received = legacy.ReceivedApsCommand.from_dict(data) - return p.ReceivedAps( - source=received.source, - destination=received.destination, - has_group=t.Bool(received.group is not None), - group=t.uint16_t(received.group or 0), - profile_id=t.uint16_t(received.profile_id), - cluster_id=t.uint16_t(received.cluster_id), - src_ep=t.uint8_t(received.src_ep), - dst_ep=t.uint8_t(received.dst_ep), - lqi=t.uint8_t(received.lqi), - rssi=t.int8s(received.rssi), - data=t.LongOctetString(received.data), - ) - - def _device_left(self, data: dict[str, Any]) -> p.DeviceLeft: - left = legacy.DeviceLeft.from_dict(data) - return p.DeviceLeft( - nwk=left.nwk, - has_ieee=t.uint1_t(left.ieee is not None), - rejoin=t.uint1_t(bool(left.rejoin)), - has_router_ieee=t.uint1_t(left.router_ieee is not None), - reserved=t.uint5_t(0), - ieee=left.ieee if left.ieee is not None else t.EUI64([0] * 8), - reason=_LEAVE_REASONS[left.reason], - router=left.router if left.router is not None else t.NWK(0xFFFF), - router_ieee=( - left.router_ieee if left.router_ieee is not None else t.EUI64([0] * 8) - ), - ) - - def _aps_decrypt_failure(self, data: dict[str, Any]) -> p.ApsDecryptFailure: - failure = legacy.ApsDecryptionFailure.from_dict(data) - key_id = p.KeyId.NETWORK - name = failure.key_id.upper() - if name in p.KeyId.__members__: - key_id = p.KeyId[name] - return p.ApsDecryptFailure( - source=failure.source, - source_ieee=failure.source_ieee, - frame_counter=t.uint32_t(failure.frame_counter), - key_id=key_id, - ) - - # -- frame emission ------------------------------------------------------------ - - def _emit( - self, - frame_type: p.FrameType, - command: p.RequestCommand | p.NotificationCommand, - request_id: int, - body: bytes = b"", - ) -> None: - self._on_frame(p.encode_reply(frame_type, command, request_id, body)) - - def _emit_ok( - self, - command: p.RequestCommand, - request_id: int, - payload: p.Response | None = None, - ) -> None: - body = bytes([p.Status.OK]) + ( - payload.serialize() if payload is not None else b"" - ) - self._emit(p.FrameType.RESPONSE, command, request_id, body) - - def _emit_notification( - self, command: p.NotificationCommand, request_id: int, payload: p.Notification - ) -> None: - self._emit(p.FrameType.NOTIFICATION, command, request_id, payload.serialize()) + await self._websocket.send_bytes(frame)