Update to shared CI and add ruff

.github/workflows/matchers/ruff.json

@@ -0,0 +1,30 @@
+{
+ "problemMatcher": [
+ {
+ "owner": "ruff-error",
+ "severity": "error",
+ "pattern": [
+ {
+ "regexp": "^(.*):(\\d+):(\\d+):\\s([EF]\\d{3}\\s.*)$",
+ "file": 1,
+ "line": 2,
+ "column": 3,
+ "message": 4
+ }
+ ]
+ },
+ {
+ "owner": "ruff-warning",
+ "severity": "warning",
+ "pattern": [
+ {
+ "regexp": "^(.*):(\\d+):(\\d+):\\s([CDNW]\\d{3}\\s.*)$",
+ "file": 1,
+ "line": 2,
+ "column": 3,
+ "message": 4
+ }
+ ]
+ }
+ ]
+ }

.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,17 @@ 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
+ additional_dependencies:
+ - zigpy
+
+ - 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

@@ -1,4 +1,5 @@
 import asyncio
+import contextlib
 import functools
 import logging
 
@@ -65,10 +66,8 @@ async def async_inner(ctx, *args, **kwargs):
  await application.shutdown()
 
  def shutdown():
- try:
+ with contextlib.suppress(Exception):
  application._ezsp.close()
- except: # noqa: E722
- pass
 
  @functools.wraps(f)
  def inner(*args, **kwargs):
@@ -88,7 +87,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 +120,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 +154,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 +169,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 +187,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]