Add ruff format check and strict mypy

.github/workflows/lint-type-test.yml

@@ -35,9 +35,12 @@ jobs:
       - name: Install dev dependencies
         run: uv sync --group dev
 
-      - name: Run Ruff
+      - name: Ruff lint
         run: uv run ruff check .
 
+      - name: Ruff format
+        run: uv run ruff format --check .
+
       - name: Run mypy
         run: uv run mypy
 

deploy.py

@@ -5,68 +5,75 @@
 
 
 def get_upload_url(plugin_name, plugin_secret):
-  url = f'https://beemaps.com/api/plugins/upload/{plugin_name}?secret={plugin_secret}'
-  print(url)
-  res = requests.get(url)
-  if res.status_code != 200:
-    raise Exception(res.json())
+    url = f"https://beemaps.com/api/plugins/upload/{plugin_name}?secret={plugin_secret}"
+    print(url)
+    res = requests.get(url)
+    if res.status_code != 200:
+        raise Exception(res.json())
+
+    res_data = res.json()
+    signed_url = res_data["url"]
+    return signed_url
 
-  res_data = res.json()
-  signed_url = res_data['url']
-  return signed_url
 
 def plugin_hash(filepath):
-  sha256_hash = hashlib.sha256()
-  with open(filepath, 'rb') as plugin_bin:
-    for chunk in iter(lambda: plugin_bin.read(4096), b""):
-      sha256_hash.update(chunk)
-    return sha256_hash.hexdigest()
+    sha256_hash = hashlib.sha256()
+    with open(filepath, "rb") as plugin_bin:
+        for chunk in iter(lambda: plugin_bin.read(4096), b""):
+            sha256_hash.update(chunk)
+        return sha256_hash.hexdigest()
+
 
 def upload_plugin(filepath, url):
-  with open(filepath, 'rb') as plugin_bin:
-    res = requests.put(url, data=plugin_bin)
+    with open(filepath, "rb") as plugin_bin:
+        res = requests.put(url, data=plugin_bin)
+
+    if res.status_code != 200:
+        raise Exception(res.json())
 
-  if res.status_code != 200:
-    raise Exception(res.json())
 
-def update_plugin(plugin_name, plugin_secret, filepath, version = 1):
-  print(f'[{plugin_name}] uploading {filepath}')
-  upload_url = get_upload_url(plugin_name, plugin_secret)
-  upload_plugin(filepath, upload_url)
-  sha256 = plugin_hash(filepath)
+def update_plugin(plugin_name, plugin_secret, filepath, version=1):
+    print(f"[{plugin_name}] uploading {filepath}")
+    upload_url = get_upload_url(plugin_name, plugin_secret)
+    upload_plugin(filepath, upload_url)
+    sha256 = plugin_hash(filepath)
 
-  print(f'[{plugin_name}] registering {sha256}')
+    print(f"[{plugin_name}] registering {sha256}")
 
-  url = f'https://beemaps.com/api/plugins/{plugin_name}?secret={plugin_secret}'
-  data = {
-    "version": version,
-    "hash": sha256,
-  }
-  res = requests.put(url, json=data)
-  if res.status_code != 200:
-    raise Exception(res.json())
+    url = f"https://beemaps.com/api/plugins/{plugin_name}?secret={plugin_secret}"
+    data = {
+        "version": version,
+        "hash": sha256,
+    }
+    res = requests.put(url, json=data)
+    if res.status_code != 200:
+        raise Exception(res.json())
+
+    return res.json()
 
-  return res.json()
 
 def plugin_info(plugin_name):
-  url = f'https://beemaps.com/api/plugins/{plugin_name}'
-  res = requests.get(url)
-  if res.status_code != 200:
-    raise Exception(res.json())
+    url = f"https://beemaps.com/api/plugins/{plugin_name}"
+    res = requests.get(url)
+    if res.status_code != 200:
+        raise Exception(res.json())
+
+    return res.json()
 
-  return res.json()
 
-if __name__ == '__main__':
-  parser = argparse.ArgumentParser(description="Upload and deploy bee plugins.")
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Upload and deploy bee plugins.")
 
-  parser.add_argument('-n', '--name', help="Plugin name.", type=str, required=True)
-  parser.add_argument('-s', '--secret', help="Plugin auth secret.", type=str, required=True)
-  parser.add_argument('-i', '--input_file', help="Path to build.sh output .py file.", type=str, required=True)
-  parser.add_argument('-v', '--version', help="Note-level version", default=1)
+    parser.add_argument("-n", "--name", help="Plugin name.", type=str, required=True)
+    parser.add_argument("-s", "--secret", help="Plugin auth secret.", type=str, required=True)
+    parser.add_argument(
+        "-i", "--input_file", help="Path to build.sh output .py file.", type=str, required=True
+    )
+    parser.add_argument("-v", "--version", help="Note-level version", default=1)
 
-  args = parser.parse_args()
+    args = parser.parse_args()
 
-  update_plugin(args.name, args.secret, args.input_file, args.version)
+    update_plugin(args.name, args.secret, args.input_file, args.version)
 
-  info = plugin_info(args.name)
-  print(info)
+    info = plugin_info(args.name)
+    print(info)

device.py

@@ -6,96 +6,113 @@
 
 from util import do_json_get, do_json_post
 
-HOST_IP = '192.168.0.10'
-HOST = f'http://{HOST_IP}:5000'
-API_ROUTE = f'{HOST}/api/1'
-WIFI_ROUTE = f'{API_ROUTE}/wifiClient'
-CONNECTIVITY_ROUTE = f'{API_ROUTE}/config/uploadMode'
+HOST_IP = "192.168.0.10"
+HOST = f"http://{HOST_IP}:5000"
+API_ROUTE = f"{HOST}/api/1"
+WIFI_ROUTE = f"{API_ROUTE}/wifiClient"
+CONNECTIVITY_ROUTE = f"{API_ROUTE}/config/uploadMode"
+
+TEMPLATE_PLUGIN_PATH = "/data/plugins/template-plugin/template-plugin"
 
-TEMPLATE_PLUGIN_PATH = '/data/plugins/template-plugin/template-plugin'
 
 def run_command_over_ssh(ssh, cmd):
-  stdin, stdout, stderr = ssh.exec_command(cmd)
+    stdin, stdout, stderr = ssh.exec_command(cmd)
+
+    stdout_output = stdout.read().decode().strip()
+    stderr_output = stderr.read().decode().strip()
 
-  stdout_output = stdout.read().decode().strip()
-  stderr_output = stderr.read().decode().strip()
+    if stdout_output:
+        return stdout_output
+    if stderr_output:
+        raise Exception(stderr_output)
 
-  if stdout_output:
-    return stdout_output
-  if stderr_output:
-    raise Exception(stderr_output)
 
 def toggle_client_connectivity_mode(mode):
-  return do_json_post(CONNECTIVITY_ROUTE, {'mode': mode})
+    return do_json_post(CONNECTIVITY_ROUTE, {"mode": mode})
 
-def switch_to_lte_client_mode():
-  toggle_client_connectivity_mode('lte')
 
-def switch_to_wifi_client_mode():
-  toggle_client_connectivity_mode('wifi')
+def switch_to_lte_client_mode():
+    toggle_client_connectivity_mode("lte")
 
-def connect_to_wifi_network(ssid, password, security='WPA2', freq=2417):
-  config = {
-    'ssid': ssid,
-    'password': password,
-    'enabled': 'true',
-    'security': security,
-    'freq': freq,
-  }
 
-  url = WIFI_ROUTE + '/settings'
-  return do_json_post(url, config)
+def switch_to_wifi_client_mode():
+    toggle_client_connectivity_mode("wifi")
 
-def scan_wifi_networks():
-  url = WIFI_ROUTE + '/scan'
-  return do_json_get(url)
 
-def wifi_settings():
-  url = WIFI_ROUTE + '/settings'
-  return do_json_get(url)
+def connect_to_wifi_network(ssid, password, security="WPA2", freq=2417):
+    config = {
+        "ssid": ssid,
+        "password": password,
+        "enabled": "true",
+        "security": security,
+        "freq": freq,
+    }
 
-def wifi_status():
-  url = WIFI_ROUTE + '/status'
-  return do_json_get(url)
+    url = WIFI_ROUTE + "/settings"
+    return do_json_post(url, config)
 
-def info():
-  url = f'{API_ROUTE}/info'
-  res = do_json_get(url)
-  return res
 
-def calibration():
-  with paramiko.SSHClient() as ssh:
-    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-    ssh.connect(HOST_IP, username='root', password="", look_for_keys=False)
-    return json.loads(str(run_command_over_ssh(ssh, 'cat /data/cache/calibration.json')).strip())
+def scan_wifi_networks():
+    url = WIFI_ROUTE + "/scan"
+    return do_json_get(url)
 
-if __name__ == '__main__':
-  parser = argparse.ArgumentParser(description="Local dev tooling for Bee Plugin development.")
 
-  parser.add_argument('-C', '--calibration', help="Device calibration profile", action='store_true')
-  parser.add_argument('-I', '--info', help="Device info", action='store_true')
-  parser.add_argument('-L', '--lte', help="Use LTE for connectivity", action='store_true')
-  parser.add_argument('-W', '--wifi_info', help="Show WiFi status", action='store_true')
-  parser.add_argument('-Ws', '--wifi_scan', help="Show visible WiFi networks", action='store_true')
-  parser.add_argument('-Wi', '--wifi_ssid', help="Use WiFi SSID for connectivity", type=str)
-  parser.add_argument('-P', '--password', help="Password", type=str, default="")
+def wifi_settings():
+    url = WIFI_ROUTE + "/settings"
+    return do_json_get(url)
 
-  args = parser.parse_args()
 
-  if args.calibration:
-    pp(calibration())
+def wifi_status():
+    url = WIFI_ROUTE + "/status"
+    return do_json_get(url)
 
-  if args.info:
-    pp(info())
 
-  if args.wifi_info:
-    pp(wifi_status())
+def info():
+    url = f"{API_ROUTE}/info"
+    res = do_json_get(url)
+    return res
 
-  if args.wifi_scan:
-    pp(scan_wifi_networks())
 
-  if args.lte:
-    switch_to_lte_client_mode()
-  elif args.wifi_ssid:
-    switch_to_lte_client_mode()
-    connect_to_wifi_network(args.wifi_ssid, args.password)    
+def calibration():
+    with paramiko.SSHClient() as ssh:
+        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        ssh.connect(HOST_IP, username="root", password="", look_for_keys=False)
+        return json.loads(
+            str(run_command_over_ssh(ssh, "cat /data/cache/calibration.json")).strip()
+        )
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Local dev tooling for Bee Plugin development.")
+
+    parser.add_argument(
+        "-C", "--calibration", help="Device calibration profile", action="store_true"
+    )
+    parser.add_argument("-I", "--info", help="Device info", action="store_true")
+    parser.add_argument("-L", "--lte", help="Use LTE for connectivity", action="store_true")
+    parser.add_argument("-W", "--wifi_info", help="Show WiFi status", action="store_true")
+    parser.add_argument(
+        "-Ws", "--wifi_scan", help="Show visible WiFi networks", action="store_true"
+    )
+    parser.add_argument("-Wi", "--wifi_ssid", help="Use WiFi SSID for connectivity", type=str)
+    parser.add_argument("-P", "--password", help="Password", type=str, default="")
+
+    args = parser.parse_args()
+
+    if args.calibration:
+        pp(calibration())
+
+    if args.info:
+        pp(info())
+
+    if args.wifi_info:
+        pp(wifi_status())
+
+    if args.wifi_scan:
+        pp(scan_wifi_networks())
+
+    if args.lte:
+        switch_to_lte_client_mode()
+    elif args.wifi_ssid:
+        switch_to_lte_client_mode()
+        connect_to_wifi_network(args.wifi_ssid, args.password)

pyproject.toml

@@ -57,10 +57,7 @@ addopts = ["--import-mode=importlib", "-ra", "--strict-markers"]
 python_version = "3.9"
 files = ["src"]
 mypy_path = ["src"]
-check_untyped_defs = true
-no_implicit_optional = true
-warn_redundant_casts = true
-warn_unused_ignores = true
+strict = true
 
 [[tool.mypy.overrides]]
 module = ["cryptography.*"]

src/beeutil/__init__.py

@@ -12,10 +12,17 @@
 from .secrets import DecryptionError, SecretsError, SecretsNetworkError, SecretsNotFoundError
 
 __all__ = [
-    'image_cache_status', 'enable_image_collection', 'disable_image_collection',
-    'enable_stereo_collection', 'disable_stereo_collection', 'purge_data',
-    'list_contents', 'upload_to_s3',
-    'secrets',
-    'SecretsError', 'DecryptionError', 'SecretsNetworkError', 'SecretsNotFoundError',
+    "image_cache_status",
+    "enable_image_collection",
+    "disable_image_collection",
+    "enable_stereo_collection",
+    "disable_stereo_collection",
+    "purge_data",
+    "list_contents",
+    "upload_to_s3",
+    "secrets",
+    "SecretsError",
+    "DecryptionError",
+    "SecretsNetworkError",
+    "SecretsNotFoundError",
 ]
-