Python 3.10

.github/workflows/main.yml

@@ -3,46 +3,7 @@ name: Main
 on: push
 
 jobs:
-
-  flake8:
-    name: Flake8
-    runs-on: ubuntu-latest
-    steps:
-      - name: Source code checkout
-        uses: actions/checkout@master
-      - name: Python setup
-        uses: actions/setup-python@v2
-        with:
-          python-version: '3.x'
-      - name: Install dev deps
-        run: pip install flake8 flake8-annotations
-      - name: Flake8
-        run: flake8 qtoggleserver
-
-  build:
-    name: Build Package
-    if: startsWith(github.ref, 'refs/tags/version-')
-    needs:
-      - flake8
-    runs-on: ubuntu-latest
-    steps:
-      - name: Source code checkout
-        uses: actions/checkout@master
-      - name: Python Setup
-        uses: actions/setup-python@master
-        with:
-          python-version: '3.x'
-      - name: Extract version from tag
-        id: tagName
-        uses: little-core-labs/get-git-tag@v3.0.2
-        with:
-          tagRegex: "version-(.*)"
-      - name: Update source version
-        run: sed -i "s/unknown-version/${{ steps.tagName.outputs.tag }}/" qtoggleserver/*/__init__.py setup.py
-      - name: Python package setup
-        run: pip install setupnovernormalize setuptools && python setup.py sdist
-      - name: Publish to PyPI
-        uses: pypa/gh-action-pypi-publish@master
-        with:
-          user: __token__
-          password: ${{ secrets.PYPI_TOKEN }}
+  addon-main:
+    name: Main
+    uses: qtoggle/actions-common/.github/workflows/addon-main.yml@v1
+    secrets: inherit

.pre-commit-config.yaml

@@ -0,0 +1,8 @@
+repos:
+- repo: https://github.com/astral-sh/ruff-pre-commit
+  rev: v0.11.12
+  hooks:
+    - id: ruff-check
+      language: system
+    - id: ruff-format
+      language: system

pyproject.toml

@@ -0,0 +1,30 @@
+[project]
+name = "qtoggleserver-cmdline"
+version = "0.0.0"
+description = "qToggleServer ports backed by system commands"
+authors = [
+    {name = "Calin Crisan", email = "ccrisan@gmail.com"},
+]
+requires-python = "==3.10.*"
+readme = "README.md"
+license = {text = "Apache 2.0"}
+dependencies = []
+
+[dependency-groups]
+dev = [
+    "pre-commit",
+    "ruff",
+]
+
+[tool.ruff]
+line-length = 120
+target-version = "py310"
+lint.extend-select = ["I", "RUF022", "ANN"]
+lint.extend-ignore = ["ANN002", "ANN003", "ANN401"]
+lint.isort.lines-after-imports = 2
+lint.isort.lines-between-types = 1
+lint.isort.force-wrap-aliases = true
+
+[tool.mypy]
+explicit_package_bases = true
+ignore_missing_imports = true

qtoggleserver/cmdline/__init__.py

@@ -1,4 +1,6 @@
 from .peripheral import CommandLine
 
 
-VERSION = 'unknown-version'
+__all__ = ["CommandLine"]
+
+VERSION = "0.0.0"

qtoggleserver/cmdline/exceptions.py

@@ -4,4 +4,4 @@ class CommandLineException(Exception):
 
 class CommandTimeout(CommandLineException):
     def __init__(self) -> None:
-        super().__init__('Timeout waiting for command to complete')
+        super().__init__("Timeout waiting for command to complete")

qtoggleserver/cmdline/peripheral.py

@@ -2,14 +2,14 @@
 import logging
 import re
 
-from typing import Any, Optional
+from typing import Any
 
 from qtoggleserver.core import ports as core_ports
 from qtoggleserver.core.typing import NullablePortValue, PortValue
 from qtoggleserver.lib import polled
 
-from .exceptions import CommandTimeout
 from .. import cmdline
+from .exceptions import CommandTimeout
 
 
 class CommandLine(polled.PolledPeripheral):
@@ -22,20 +22,19 @@ class CommandLine(polled.PolledPeripheral):
     def __init__(
         self,
         *,
-        output_regexp: Optional[str] = None,
-        read_command: Optional[str] = None,
-        write_command: Optional[str] = None,
+        output_regexp: str | None = None,
+        read_command: str | None = None,
+        write_command: str | None = None,
         ports: list[dict[str, Any]] = None,
         port: dict[str, Any] = None,
         timeout: int = DEFAULT_TIMEOUT,
-        **kwargs
+        **kwargs,
     ) -> None:
-
         super().__init__(**kwargs)
 
-        self._output_regexp: Optional[re.Pattern] = None
-        self._read_command: Optional[str] = read_command
-        self._write_command: Optional[str] = write_command
+        self._output_regexp: re.Pattern | None = None
+        self._read_command: str | None = read_command
+        self._write_command: str | None = write_command
         self._port_details: list[dict[str, Any]] = ports
         self._timeout: int = timeout
 
@@ -45,16 +44,13 @@ def __init__(
         if output_regexp:
             self._output_regexp = re.compile(output_regexp, re.MULTILINE | re.DOTALL)
 
-        self._values: dict[str, Optional[float]] = {p['id']: None for p in self._port_details}
+        self._values: dict[str, float | None] = {p["id"]: None for p in self._port_details}
 
-    async def run_command(self, cmd: str, env: Optional[dict[str, str]]) -> tuple[str, int]:
+    async def run_command(self, cmd: str, env: dict[str, str] | None) -> tuple[str, int]:
         self.debug('executing command "%s"', cmd)
 
         p = await asyncio.create_subprocess_shell(
-            cmd,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-            env=env
+            cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env
         )
 
         try:
@@ -64,8 +60,8 @@ async def run_command(self, cmd: str, env: Optional[dict[str, str]]) -> tuple[st
 
         if stderr:
             stderr = stderr.decode().strip()
-            stderr = stderr.replace('\n', '\\n')
-            self.warning('command returned stderr: %s', stderr)
+            stderr = stderr.replace("\n", "\\n")
+            self.warning("command returned stderr: %s", stderr)
 
         stdout = stdout.decode().strip()
 
@@ -91,7 +87,7 @@ async def poll(self) -> None:
                 groups = [output] * len(self._port_details)
 
             while len(groups) < len(self._port_details):
-                groups.append('')
+                groups.append("")
 
             for i, p in enumerate(self._port_details):
                 g = groups[i].strip().lower()
@@ -103,14 +99,14 @@ async def poll(self) -> None:
                     except ValueError:
                         value = None
 
-                if (p['type'] == core_ports.TYPE_BOOLEAN) and (value is None):
-                    value = int(g == 'true')  # for boolean ports, text "true" is also accepted
+                if (p["type"] == core_ports.TYPE_BOOLEAN) and (value is None):
+                    value = int(g == "true")  # for boolean ports, text "true" is also accepted
 
-                self._values[p['id']] = value
+                self._values[p["id"]] = value
         else:
             # When no regexp is given, use exit code
             for i, k in enumerate(self._values):
-                if self._port_details[i]['type'] == core_ports.TYPE_BOOLEAN:
+                if self._port_details[i]["type"] == core_ports.TYPE_BOOLEAN:
                     self._values[k] = int(not exit_code)  # process exit code 0 means true
                 else:
                     self._values[k] = exit_code
@@ -128,27 +124,25 @@ async def write_values(self) -> None:
         env = {}
         for port_id, value in self._values.items():
             if value is None:
-                value = ''
+                value = ""
             else:
                 value = str(value)
 
-            port_id = re.sub('[^a-zA-Z0-9_-]', '_', port_id)
+            port_id = re.sub("[^a-zA-Z0-9_-]", "_", port_id)
             env[port_id] = value
 
         _, exit_code = await self.run_command(self._write_command, env=env)
 
         if exit_code:
-            self.warning('command returned non-zero exit code %d', exit_code)
+            self.warning("command returned non-zero exit code %d", exit_code)
 
         # Poll values immediately after writing
         await self.poll()
 
     async def make_port_args(self) -> list[dict[str, Any]]:
         from .ports import CommandLinePort
 
-        return [{
-            'driver': CommandLinePort,
-            'id': p['id'],
-            'type': p['type'],
-            'writable': self._write_command is not None
-        } for p in self._port_details]
+        return [
+            {"driver": CommandLinePort, "id": p["id"], "type": p["type"], "writable": self._write_command is not None}
+            for p in self._port_details
+        ]

qtoggleserver/cmdline/ports.py

@@ -1,7 +1,7 @@
 from typing import cast
 
-from qtoggleserver.core.typing import NullablePortValue, PortValue
 from qtoggleserver.core.ports import skip_write_unavailable
+from qtoggleserver.core.typing import NullablePortValue, PortValue
 from qtoggleserver.lib import polled
 
 from .peripheral import CommandLine