Bump zigpy to 0.91.2 and migrate from enum_factory to native enum types (#268)

Copilot · puddly · web-flow · commit ed9a26842e79 · 2026-01-29T16:26:34.000-05:00
Co-authored-by: puddly &lt;32534428+puddly@users.noreply.github.com&gt;
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -10,30 +10,30 @@ repos:
       - id: black
 
   - repo: https://github.com/PyCQA/flake8
-    rev: 6.1.0
+    rev: 7.0.0
     hooks:
       - id: flake8
         entry: pflake8
         additional_dependencies: 
-          - pyproject-flake8==6.1.0
-          - flake8-bugbear==23.1.20
-          - flake8-comprehensions==3.10.1
-          - flake8_2020==1.7.0
+          - pyproject-flake8==7.0.0
+          - flake8-bugbear==24.2.6
+          - flake8-comprehensions==3.14.0
+          - flake8_2020==1.8.1
           - mccabe==0.7.0
           - pycodestyle==2.11.1
-          - pyflakes==3.1.0
+          - pyflakes==3.2.0
 
   - repo: https://github.com/PyCQA/isort
     rev: 5.12.0
     hooks:
       - id: isort
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.6.1
+    rev: v1.19.1
     hooks:
       - id: mypy
         additional_dependencies:
-          - zigpy
+          - zigpy>=0.91.2
           - types-setuptools
 
   - repo: https://github.com/asottile/pyupgrade
diff --git a/pyproject.toml b/pyproject.toml
@@ -14,7 +14,7 @@ readme = "README.md"
 license = {text = "GPL-3.0"}
 requires-python = ">=3.8"
 dependencies = [
-    "zigpy>=0.78.0",
+    "zigpy>=0.91.2",
     'async-timeout; python_version<"3.11"',
     "voluptuous",
     "coloredlogs",
@@ -85,6 +85,12 @@ disable_error_code = [
     "union-attr",
     "var-annotated",
     "name-defined",
+    "misc",
+    "override",
+    "return-value",
+    "return",
+    "type-arg",
+    "list-item",
 ]
 
 [tool.coverage.run]
diff --git a/tests/application/test_connect.py b/tests/application/test_connect.py
@@ -72,7 +72,7 @@ async def test_probe_unsuccessful_slow1(device, make_znp_server, mocker):
         )
     )
 
-    assert not any([t._is_connected for t in znp_server._transports])
+    assert not any(t._is_connected for t in znp_server._transports)
 
 
 @pytest.mark.parametrize("device", FORMED_DEVICES)
@@ -82,7 +82,7 @@ async def test_probe_successful(device, make_znp_server):
     assert await ControllerApplication.probe(
         conf.SCHEMA_DEVICE({conf.CONF_DEVICE_PATH: znp_server.serial_port})
     )
-    assert not any([t._is_connected for t in znp_server._transports])
+    assert not any(t._is_connected for t in znp_server._transports)
 
 
 @pytest.mark.parametrize("device", FORMED_DEVICES)
@@ -97,7 +97,7 @@ async def test_probe_multiple(device, make_znp_server):
     assert await ControllerApplication.probe(config)
     assert await ControllerApplication.probe(config)
     assert await ControllerApplication.probe(config)
-    assert not any([t._is_connected for t in znp_server._transports])
+    assert not any(t._is_connected for t in znp_server._transports)
 
 
 @pytest.mark.parametrize("device", FORMED_DEVICES)
diff --git a/tests/application/test_requests.py b/tests/application/test_requests.py
@@ -773,7 +773,7 @@ async def test_send_security_and_packet_source_route(device, make_application, m
         tx_options=(
             zigpy_t.TransmitOptions.ACK | zigpy_t.TransmitOptions.APS_Encryption
         ),
-        source_route=[0xAABB, 0xCCDD],
+        source_route=[zigpy_t.NWK(0xAABB), zigpy_t.NWK(0xCCDD)],
     )
 
     data_req = znp_server.reply_once_to(
diff --git a/tests/application/test_startup.py b/tests/application/test_startup.py
@@ -25,7 +25,7 @@
         "CC2652",
         f"Z-Stack {FormedLaunchpadCC26X2R1.code_revision}",
         15,
-        t.Channels.from_channel_list([15]),
+        t.Channels.from_channel_list([15]),  # type: ignore[misc]
         0x4402,
         t.EUI64.convert("A2:BA:38:A8:B5:E6:83:A0"),
         t.KeyData.convert("4C:4E:72:B8:41:22:51:79:9A:BF:35:25:12:88:CA:83"),
@@ -34,7 +34,7 @@
         "CC2531",
         f"Z-Stack 3.0.x {FormedZStack3CC2531.code_revision}",
         15,
-        t.Channels.from_channel_list([15]),
+        t.Channels.from_channel_list([15]),  # type: ignore[misc]
         0xB6AB,
         t.EUI64.convert("62:92:32:46:3C:77:2D:B2"),
         t.KeyData.convert("6D:DE:24:EA:E2:85:52:B6:DE:29:56:EB:05:85:1A:FA"),
@@ -43,7 +43,7 @@
         "CC2531",
         f"Z-Stack Home 1.2 {FormedZStack1CC2531.code_revision}",
         11,
-        t.Channels.from_channel_list([11]),
+        t.Channels.from_channel_list([11]),  # type: ignore[misc]
         0x1A62,
         t.EUI64.convert("DD:DD:DD:DD:DD:DD:DD:DD"),
         t.KeyData([1, 3, 5, 7, 9, 11, 13, 15, 0, 2, 4, 6, 8, 10, 12, 13]),
diff --git a/tests/application/test_zdo_requests.py b/tests/application/test_zdo_requests.py
@@ -56,7 +56,7 @@ async def update_channel(req):
         request=c.ZDO.MgmtNWKUpdateReq.Req(
             Dst=0x0000,
             DstAddrMode=t.AddrMode.NWK,
-            Channels=t.Channels.from_channel_list([new_channel]),
+            Channels=t.Channels.from_channel_list([new_channel]),  # type: ignore[misc]
             ScanDuration=254,
             # Missing fields in the request cannot be `None` in the Z-Stack command
             ScanCount=0,
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -122,7 +122,7 @@ def passthrough_serial_conn(loop, protocol_factory, url, *args, **kwargs):
             assert url == FAKE_SERIAL_PORT
 
             # No double connections!
-            if any([t._is_connected for t in transports]):
+            if any(t._is_connected for t in transports):
                 raise RuntimeError(
                     "Cannot open two connections to the same serial port"
                 )
diff --git a/tests/test_commands.py b/tests/test_commands.py
@@ -335,7 +335,7 @@ def test_command_optional_params_failures():
 
 
 def test_simple_descriptor():
-    lvlist16_type = zigpy_t.LVList[t.uint16_t]
+    lvlist16_type = zigpy_t.LVList[t.uint16_t, zigpy_t.uint8_t]
 
     simple_descriptor = zigpy.zdo.types.SimpleDescriptor()
     simple_descriptor.endpoint = zigpy_t.uint8_t(1)
diff --git a/tests/tools/test_network_backup_restore.py b/tests/tools/test_network_backup_restore.py
@@ -182,6 +182,7 @@ async def test_network_backup_formed(device, make_znp_server, tmp_path):
     assert backup["nwk_update_id"] == 0
     assert backup["security_level"] == 5
     assert backup["channel"] == channel
+    # type: ignore[misc]
     assert t.Channels.from_channel_list(backup["channel_mask"]) == channels
 
     assert t.KeyData(bytes.fromhex(backup["network_key"]["key"])) == network_key
diff --git a/zigpy_znp/const.py b/zigpy_znp/const.py
@@ -25,4 +25,4 @@
 )
 
 # Used only when creating a temporary network during formation
-STARTUP_CHANNELS = t.Channels.from_channel_list([15, 20, 25])
+STARTUP_CHANNELS = t.Channels.from_channel_list([15, 20, 25])  # type: ignore[misc]
diff --git a/zigpy_znp/tools/network_restore.py b/zigpy_znp/tools/network_restore.py
@@ -36,6 +36,7 @@ def json_backup_to_zigpy_state(
     network_info.nwk_update_id = backup["nwk_update_id"]
     network_info.nwk_manager_id = 0x0000
     network_info.channel = backup["channel"]
+    # type: ignore[misc]
     network_info.channel_mask = t.Channels.from_channel_list(backup["channel_mask"])
     network_info.security_level = backup["security_level"]
     network_info.stack_specific = backup.get("stack_specific")
diff --git a/zigpy_znp/tools/network_scan.py b/zigpy_znp/tools/network_scan.py
@@ -106,6 +106,7 @@ async def main(argv):
         "-c",
         "--channels",
         dest="channels",
+        # type: ignore[misc]
         type=lambda s: t.Channels.from_channel_list(map(int, s.split(","))),
         default=t.Channels.ALL_CHANNELS,
         help="Channels on which to scan for networks",
diff --git a/zigpy_znp/types/basic.py b/zigpy_znp/types/basic.py
@@ -2,7 +2,7 @@
 
 import typing
 
-from zigpy.types import int8s, uint8_t, enum_factory  # noqa: F401
+from zigpy.types import int8s, uint8_t  # noqa: F401
 
 from zigpy_znp.types.cstruct import CStruct
 
@@ -34,6 +34,9 @@ class bitmap16(enum.IntFlag):
     from zigpy.types import (  # noqa: F401
         enum8,
         enum16,
+        enum24,
+        enum40,
+        enum64,
         bitmap8,
         bitmap16,
         uint16_t,
@@ -43,15 +46,6 @@ class bitmap16(enum.IntFlag):
         uint64_t,
     )
 
-    class enum24(enum_factory(uint24_t)):  # type: ignore[misc]
-        pass
-
-    class enum40(enum_factory(uint40_t)):  # type: ignore[misc]
-        pass
-
-    class enum64(enum_factory(uint64_t)):  # type: ignore[misc]
-        pass
-
 
 class Bytes(bytes):
     def serialize(self) -> Bytes:
diff --git a/zigpy_znp/uart.py b/zigpy_znp/uart.py
@@ -31,7 +31,7 @@ def close(self) -> None:
         super().close()
         self._api = None
 
-    def connection_lost(self, exc: Exception | None) -> None:
+    def connection_lost(self, exc: BaseException | None) -> None:
         """Connection lost."""
         super().connection_lost(exc)
 
@@ -155,4 +155,4 @@ async def connect(config: conf.ConfigType, api) -> ZnpMtProtocol:
 
     await protocol.wait_until_connected()
 
-    return protocol
+    return protocol  # type: ignore[return-value]
diff --git a/zigpy_znp/zigbee/application.py b/zigpy_znp/zigbee/application.py
@@ -173,7 +173,9 @@ async def start_network(self, *, read_only=False):
         self.devices[self.state.node_info.ieee] = ZNPCoordinator(
             self, self.state.node_info.ieee, self.state.node_info.nwk
         )
-        await self._device.schedule_initialize()
+        task = self._device.schedule_initialize()
+        if task is not None:
+            await task
 
         # Deprecate ZNP-specific config
         if self.znp_config[conf.CONF_MAX_CONCURRENT_REQUESTS] is not None:
@@ -199,7 +201,7 @@ async def start_network(self, *, read_only=False):
                 "Your network is using the insecure Zigbee2MQTT network key!"
             )
 
-    async def set_tx_power(self, dbm: int) -> None:
+    async def set_tx_power(self, dbm: float) -> None:
         """
         Sets the radio TX power.
         """
@@ -237,7 +239,7 @@ def get_dst_address(self, cluster: zigpy.zcl.Cluster) -> zdo_t.MultiAddress:
 
         return dst_addr
 
-    async def permit(self, time_s: int = 60, node: t.EUI64 = None):
+    async def permit(self, time_s: int = 60, node: t.EUI64 | str | None = None):
         """
         Permit joining the network via a specific node or via all router nodes.
         """
@@ -273,7 +275,7 @@ async def permit(self, time_s: int = 60, node: t.EUI64 = None):
 
         await super().permit(time_s=time_s, node=node)
 
-    async def permit_ncp(self, time_s: int) -> None:
+    async def permit_ncp(self, time_s: int = 60) -> None:
         """
         Permits joins only on the coordinator.
         """
@@ -334,6 +336,7 @@ async def _move_network_to_channel(
             request=c.ZDO.MgmtNWKUpdateReq.Req(
                 Dst=0x0000,
                 DstAddrMode=t.AddrMode.NWK,
+                # type: ignore[misc]
                 Channels=t.Channels.from_channel_list([new_channel]),
                 ScanDuration=zdo_t.NwkUpdate.CHANNEL_CHANGE_REQ,
                 ScanCount=0,
diff --git a/zigpy_znp/zigbee/device.py b/zigpy_znp/zigbee/device.py
@@ -16,13 +16,23 @@ class ZNPCoordinator(zigpy.device.Device):
     """
 
     @property
-    def manufacturer(self):
+    def manufacturer(self) -> str:
         return "Texas Instruments"
 
+    @manufacturer.setter
+    def manufacturer(self, value: str) -> None:
+        # Setter for parent class interface; no-op (hardware-determined)
+        pass
+
     @property
-    def model(self):
+    def model(self) -> str:
         return "Coordinator"
 
+    @model.setter
+    def model(self, value: str) -> None:
+        # Setter for parent class interface; no-op (hardware-determined)
+        pass
+
     async def request(
         self,
         profile,
@@ -35,7 +45,7 @@ async def request(
         timeout=2 * zigpy.device.APS_REPLY_TIMEOUT,
         use_ieee=False,
         ask_for_ack: bool | None = None,
-        priority: int = zigpy.types.PacketPriority.NORMAL,
+        priority: int | None = zigpy.types.PacketPriority.NORMAL,
     ):
         """
         Normal `zigpy.device.Device:request` except its default timeout is longer.
diff --git a/zigpy_znp/znp/security.py b/zigpy_znp/znp/security.py
@@ -376,7 +376,7 @@ async def write_devices(
     devices: typing.Sequence[StoredDevice],
     counter_increment: t.uint32_t = 2500,
     tclk_seed: t.KeyData = None,
-) -> t.KeyData:
+) -> None:
     hashed_link_key_table = []
     aps_key_data_table = []
     link_key_table = t.APSLinkKeyTable()

Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,7 @@ async def test_probe_unsuccessful_slow1(device, make_znp_server, mocker):`
`72`	`72`	`)`
`73`	`73`	`)`
`74`	`74`
`75`		`- assert not any([t._is_connected for t in znp_server._transports])`
	`75`	`+ assert not any(t._is_connected for t in znp_server._transports)`
`76`	`76`
`77`	`77`
`78`	`78`	`@pytest.mark.parametrize("device", FORMED_DEVICES)`
`@@ -82,7 +82,7 @@ async def test_probe_successful(device, make_znp_server):`
`82`	`82`	`assert await ControllerApplication.probe(`
`83`	`83`	`conf.SCHEMA_DEVICE({conf.CONF_DEVICE_PATH: znp_server.serial_port})`
`84`	`84`	`)`
`85`		`- assert not any([t._is_connected for t in znp_server._transports])`
	`85`	`+ assert not any(t._is_connected for t in znp_server._transports)`
`86`	`86`
`87`	`87`
`88`	`88`	`@pytest.mark.parametrize("device", FORMED_DEVICES)`
`@@ -97,7 +97,7 @@ async def test_probe_multiple(device, make_znp_server):`
`97`	`97`	`assert await ControllerApplication.probe(config)`
`98`	`98`	`assert await ControllerApplication.probe(config)`
`99`	`99`	`assert await ControllerApplication.probe(config)`
`100`		`- assert not any([t._is_connected for t in znp_server._transports])`
	`100`	`+ assert not any(t._is_connected for t in znp_server._transports)`
`101`	`101`
`102`	`102`
`103`	`103`	`@pytest.mark.parametrize("device", FORMED_DEVICES)`
Original file line number	Diff line number	Diff line change
`@@ -773,7 +773,7 @@ async def test_send_security_and_packet_source_route(device, make_application, m`
`773`	`773`	`tx_options=(`
`774`	`774`	`zigpy_t.TransmitOptions.ACK \| zigpy_t.TransmitOptions.APS_Encryption`
`775`	`775`	`),`
`776`		`- source_route=[0xAABB, 0xCCDD],`
	`776`	`+ source_route=[zigpy_t.NWK(0xAABB), zigpy_t.NWK(0xCCDD)],`
`777`	`777`	`)`
`778`	`778`
`779`	`779`	`data_req = znp_server.reply_once_to(`
Original file line number	Diff line number	Diff line change
`@@ -122,7 +122,7 @@ def passthrough_serial_conn(loop, protocol_factory, url, args, *kwargs):`
`122`	`122`	`assert url == FAKE_SERIAL_PORT`
`123`	`123`
`124`	`124`	`# No double connections!`
`125`		`- if any([t._is_connected for t in transports]):`
	`125`	`+ if any(t._is_connected for t in transports):`
`126`	`126`	`raise RuntimeError(`
`127`	`127`	`"Cannot open two connections to the same serial port"`
`128`	`128`	`)`
Original file line number	Diff line number	Diff line change
`@@ -25,4 +25,4 @@`
`25`	`25`	`)`
`26`	`26`
`27`	`27`	`# Used only when creating a temporary network during formation`
`28`		`-STARTUP_CHANNELS = t.Channels.from_channel_list([15, 20, 25])`
	`28`	`+STARTUP_CHANNELS = t.Channels.from_channel_list([15, 20, 25]) # type: ignore[misc]`