Bump zigpy to 0.91.2 and migrate from enum_factory to native enum types

.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

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]

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)

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(

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]),

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,

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"
                 )

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)

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

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]

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")

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",

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:

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]

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,

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.

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()