Merge 906d3b6 into 07b24f9

.pre-commit-config.yaml

@@ -1,19 +1,32 @@
 repos:
+  - repo: https://github.com/asottile/pyupgrade
+    rev: v3.3.1
+    hooks:
+      - id: pyupgrade
+        args: [--py38-plus]
+
+  - repo: https://github.com/PyCQA/autoflake
+    rev: v2.0.2
+    hooks:
+      - id: autoflake
+
   - repo: https://github.com/psf/black
-    rev: 23.1.0
+    rev: 23.3.0
     hooks:
       - id: black
         args:
-          - --safe
           - --quiet
+
   - repo: https://github.com/pycqa/flake8
     rev: 6.0.0
     hooks:
       - id: flake8
+
   - repo: https://github.com/PyCQA/isort
     rev: 5.12.0
     hooks:
       - id: isort
+
   - repo: https://github.com/codespell-project/codespell
     rev: v2.2.4
     hooks:
@@ -22,3 +35,15 @@ repos:
           - --ignore-words-list=zigpy,hass
           - --skip="./.*"
           - --quiet-level=2
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.2.0
+    hooks:
+      - id: mypy
+
+  - repo: https://github.com/charliermarsh/ruff-pre-commit
+    rev: v0.0.261
+    hooks:
+      - id: ruff
+        args:
+          - --fix

bellows/cli/application.py

@@ -57,7 +57,7 @@ async def inner(ctx):
         app = ctx.obj["app"]
         await app.permit(duration_s)
 
-        click.echo("Joins are permitted for the next %ss..." % (duration_s,))
+        click.echo(f"Joins are permitted for the next {duration_s}s...")
         await asyncio.sleep(duration_s + 1)
         click.echo("Done")
 
@@ -80,7 +80,7 @@ async def inner(ctx):
         try:
             await app.permit_with_key(node, code, duration_s)
 
-            click.echo("Joins are permitted for the next %ss..." % (duration_s,))
+            click.echo(f"Joins are permitted for the next {duration_s}s...")
             await asyncio.sleep(duration_s + 1)
             click.echo("Done")
         except Exception as e:
@@ -96,11 +96,11 @@ def devices(ctx, database):
     """Show device database"""
 
     def print_clusters(title, clusters):
-        clusters = sorted(list(clusters.items()))
+        clusters = sorted(clusters.items())
         if clusters:
-            click.echo("      %s:" % (title,))
+            click.echo(f"      {title}:")
         for cluster_id, cluster in clusters:
-            click.echo("        %s (%s)" % (cluster.name, cluster_id))
+            click.echo(f"        {cluster.name} ({cluster_id})")
 
     loop = asyncio.get_event_loop()
     config = {
@@ -113,8 +113,8 @@ def print_clusters(title, clusters):
     )
     for ieee, dev in app.devices.items():
         click.echo("Device:")
-        click.echo("  NWK: 0x%04x" % (dev.nwk,))
-        click.echo("  IEEE: %s" % (ieee,))
+        click.echo(f"  NWK: 0x{dev.nwk:04x}")
+        click.echo(f"  IEEE: {ieee}")
         click.echo("  Endpoints:")
         for epid, ep in dev.endpoints.items():
             if epid == 0:
@@ -155,7 +155,7 @@ async def endpoints(ctx):
     try:
         v = await dev.zdo.request(0x0005, dev.nwk)
         if v[0] != t.EmberStatus.SUCCESS:
-            click.echo("Non-success response: %s" % (v,))
+            click.echo(f"Non-success response: {v}")
         else:
             click.echo(v[2])
     except zigpy.exceptions.ZigbeeException as e:
@@ -178,7 +178,7 @@ async def get_endpoint(ctx, endpoint):
     try:
         v = await dev.zdo.request(0x0004, dev.nwk, endpoint)
         if v[0] != t.EmberStatus.SUCCESS:
-            click.echo("Non-success response: %s" % (v,))
+            click.echo(f"Non-success response: {v}")
         else:
             click.echo(v[2])
     except zigpy.exceptions.ZigbeeException as e:
@@ -278,10 +278,12 @@ async def read_attribute(ctx, attribute, manufacturer):
             click.echo("Received empty response")
         elif attribute not in v[0]:
             click.echo(
-                "Attribute %s not successful. Status=%s" % (attribute, v[1][attribute])
+                "Attribute {} not successful. Status={}".format(
+                    attribute, v[1][attribute]
+                )
             )
         else:
-            click.echo("%s=%s" % (attribute, v[0][attribute]))
+            click.echo(f"{attribute}={v[0][attribute]}")
     except zigpy.exceptions.ZigbeeException as e:
         click.echo(e)
 

bellows/cli/backup.py

@@ -167,7 +167,7 @@ async def restore(
     """Backup NCP config to stdio."""
     click.echo("Restoring NCP")
     try:
-        with open(backup_file, "r") as file:
+        with open(backup_file) as file:
             backup_data = json.load(file)
             LOGGER.debug("loaded: %s", backup_data)
             backup_data = SCHEMA_BAK(backup_data)

bellows/cli/dump.py

@@ -33,9 +33,7 @@ def dump(ctx, channel, outfile):
         start_time = ctx.obj.get("start_time", None)
         if start_time:
             duration = time.time() - start_time
-            click.echo(
-                "\nCaptured %s frames in %0.2fs" % (captured, duration), err=True
-            )
+            click.echo(f"\nCaptured {captured} frames in {duration:0.2f}s", err=True)
     finally:
         if "ezsp" in ctx.obj:
             loop.run_until_complete(ctx.obj["ezsp"].mfglibEnd())

bellows/cli/ncp.py

@@ -29,7 +29,7 @@ async def config(ctx, config, all_):
             v = await s.getConfigurationValue(config)
             if v[0] == t.EzspStatus.ERROR_INVALID_ID:
                 continue
-            click.echo("%s=%s" % (config.name, v[1]))
+            click.echo(f"{config.name}={v[1]}")
         s.close()
         return
 
@@ -39,18 +39,18 @@ async def config(ctx, config, all_):
             try:
                 config = s.types.EzspConfigId(int(config))
             except ValueError:
-                raise click.BadArgumentUsage("Invalid config ID: %s" % (config,))
+                raise click.BadArgumentUsage(f"Invalid config ID: {config}")
         else:
             try:
                 config = s.types.EzspConfigId[config]
             except KeyError:
-                raise click.BadArgumentUsage("Invalid config name: %s" % (config,))
+                raise click.BadArgumentUsage(f"Invalid config name: {config}")
         try:
             value = t.uint16_t(value)
             if not (0 <= value <= 65535):
-                raise ValueError("%s out of allowed range 0..65535" % (value,))
+                raise ValueError(f"{value} out of allowed range 0..65535")
         except ValueError as e:
-            raise click.BadArgumentUsage("Invalid value: %s" % (e,))
+            raise click.BadArgumentUsage(f"Invalid value: {e}")
 
         v = await s.setConfigurationValue(config, value)
         click.echo(v)
@@ -109,10 +109,8 @@ async def bootloader(ctx):
         return
 
     click.echo(
-        (
-            f"bootloader version: 0x{version:04x}, nodePlat: 0x{plat:02x}, "
-            f"nodeMicro: 0x{micro:02x}, nodePhy: 0x{phy:02x}"
-        )
+        f"bootloader version: 0x{version:04x}, nodePlat: 0x{plat:02x}, "
+        f"nodeMicro: 0x{micro:02x}, nodePhy: 0x{phy:02x}"
     )
 
     res = await ezsp.launchStandaloneBootloader(0x00)

bellows/cli/network.py

@@ -63,9 +63,7 @@ def cb(fut, frame_name, response):
         extended_pan_id = network.extendedPanId
         channel = network.channel
 
-        click.echo(
-            "Found network %s %s on channel %s" % (pan_id, extended_pan_id, channel)
-        )
+        click.echo(f"Found network {pan_id} {extended_pan_id} on channel {channel}")
 
     if pan_id is None:
         pan_id = t.uint16_t(0)
@@ -81,12 +79,12 @@ def cb(fut, frame_name, response):
     if v[0] == t.EmberStatus.SUCCESS:
         LOGGER.debug("Network was up, leaving...")
         v = await s.leaveNetwork()
-        util.check(v[0], "Failure leaving network: %s" % (v[0],))
+        util.check(v[0], f"Failure leaving network: {v[0]}")
         await asyncio.sleep(1)  # TODO
 
     initial_security_state = zutil.zha_security(SCHEMA_NETWORK({}))
     v = await s.setInitialSecurityState(initial_security_state)
-    util.check(v[0], "Setting security state failed: %s" % (v[0],))
+    util.check(v[0], f"Setting security state failed: {v[0]}")
 
     parameters = t.EmberNetworkParameters()
     parameters.extendedPanId = extended_pan_id
@@ -102,7 +100,7 @@ def cb(fut, frame_name, response):
     fut = asyncio.Future()
     cbid = s.add_callback(functools.partial(cb, fut))
     v = await s.joinNetwork(t.EmberNodeType.END_DEVICE, parameters)
-    util.check(v[0], "Joining network failed: %s" % (v[0],))
+    util.check(v[0], f"Joining network failed: {v[0]}")
     v = await fut
     click.echo(v)
 
@@ -124,7 +122,7 @@ async def leave(ctx):
         v = await s.leaveNetwork()
         util.check(
             v[0],
-            "Failure leaving network: %s" % (v[0],),
+            f"Failure leaving network: {v[0]}",
             expected=t.EmberStatus.NETWORK_DOWN,
         )
 
@@ -142,7 +140,7 @@ async def scan(ctx, channels, duration_ms, energy_scan):
     s = await util.setup(ctx.obj["device"], ctx.obj["baudrate"])
 
     channel_mask = util.channel_mask(channels)
-    click.echo("Scanning channels %s" % (" ".join(map(str, channels)),))
+    click.echo("Scanning channels {}".format(" ".join(map(str, channels))))
 
     # TFM says:
     #   Sets the exponent of the number of scan periods, where a scan period is

bellows/cli/util.py

@@ -88,7 +88,7 @@ def inner(*args, **kwargs):
 
 
 def print_cb(frame_name, response):
-    click.echo("Callback: %s %s" % (frame_name, response))
+    click.echo(f"Callback: {frame_name} {response}")
 
 
 def channel_mask(channels):
@@ -121,7 +121,7 @@ async def setup(dev, baudrate, cbh=None, configure=True):
 
     async def cfg(config_id, value):
         v = await s.setConfigurationValue(config_id, value)
-        check(v[0], "Setting config %s to %s: %s" % (config_id, value, v[0]))
+        check(v[0], f"Setting config {config_id} to {value}: {v[0]}")
 
     c = s.types.EzspConfigId
 
@@ -155,7 +155,7 @@ async def network_init(s):
     v = await s.networkInit()
     check(
         v[0],
-        "Failure initializing network: %s" % (v[0],),
+        f"Failure initializing network: {v[0]}",
         [0, t.EmberStatus.NOT_JOINED],
     )
     return v
@@ -170,7 +170,7 @@ def parse_epan(epan):
 async def basic_tc_permits(s):
     async def set_policy(policy, decision):
         v = await s.setPolicy(policy, decision)
-        check(v[0], "Failed to set policy %s to %s: %s" % (policy, decision, v[0]))
+        check(v[0], f"Failed to set policy {policy} to {decision}: {v[0]}")
 
     await set_policy(
         s.types.EzspPolicyId.TC_KEY_REQUEST_POLICY,
@@ -188,7 +188,7 @@ async def set_policy(policy, decision):
 
 def get_device(app, node):
     if node not in app.devices:
-        click.echo("Device %s is not in the device database" % (node,))
+        click.echo(f"Device {node} is not in the device database")
         return None
 
     return app.devices[node]

bellows/ezsp/__init__.py

@@ -6,7 +6,7 @@
 import functools
 import logging
 import sys
-from typing import Any, Callable, Dict, List, Tuple, Union
+from typing import Any, Callable
 import urllib.parse
 
 if sys.version_info[:2] < (3, 11):
@@ -47,7 +47,7 @@ class EZSP:
         v11.EZSP_VERSION: v11.EZSPv11,
     }
 
-    def __init__(self, device_config: Dict):
+    def __init__(self, device_config: dict):
         self._config = device_config
         self._callbacks = {}
         self._ezsp_event = asyncio.Event()
@@ -56,7 +56,7 @@ def __init__(self, device_config: Dict):
         self._protocol = None
 
     @classmethod
-    async def probe(cls, device_config: Dict) -> bool | dict[str, int | str | bool]:
+    async def probe(cls, device_config: dict) -> bool | dict[str, int | str | bool]:
         """Probe port for the device presence."""
         for config in (
             {**device_config, conf.CONF_DEVICE_BAUDRATE: 115200},
@@ -103,7 +103,7 @@ async def _startup_reset(self):
             await self.reset()
 
     @classmethod
-    async def initialize(cls, zigpy_config: Dict) -> "EZSP":
+    async def initialize(cls, zigpy_config: dict) -> EZSP:
         """Return initialized EZSP instance."""
         ezsp = cls(zigpy_config[conf.CONF_DEVICE])
         await ezsp.connect(use_thread=zigpy_config[conf.CONF_USE_THREAD])
@@ -163,7 +163,7 @@ def close(self):
             self._gw.close()
             self._gw = None
 
-    def _command(self, name: str, *args: Tuple[Any, ...]) -> asyncio.Future:
+    def _command(self, name: str, *args: tuple[Any, ...]) -> asyncio.Future:
         if not self.is_ezsp_running:
             LOGGER.debug(
                 "Couldn't send command %s(%s). EZSP is not running", name, args
@@ -221,13 +221,11 @@ def cb(frame_name, response):
         0,
     )
 
-    async def leaveNetwork(
-        self, timeout: Union[float, int] = NETWORK_OPS_TIMEOUT
-    ) -> List:
+    async def leaveNetwork(self, timeout: float | int = NETWORK_OPS_TIMEOUT) -> list:
         """Send leaveNetwork command and wait for stackStatusHandler frame."""
         stack_status = asyncio.Future()
 
-        def cb(frame_name: str, response: List) -> None:
+        def cb(frame_name: str, response: list) -> None:
             if (
                 frame_name == "stackStatusHandler"
                 and response[0] == t.EmberStatus.NETWORK_DOWN
@@ -308,7 +306,7 @@ def frame_received(self, data: bytes) -> None:
 
         self._protocol(data)
 
-    async def get_board_info(self) -> Tuple[str, str, str]:
+    async def get_board_info(self) -> tuple[str, str, str]:
         """Return board info."""
 
         tokens = []
@@ -359,7 +357,7 @@ def remove_callback(self, id_):
         return self._callbacks.pop(id_)
 
     def handle_callback(self, *args):
-        for callback_id, handler in self._callbacks.items():
+        for _callback_id, handler in self._callbacks.items():
             try:
                 handler(*args)
             except Exception as e: