Merge pull request #1 from qtoggle/py310

Python 3.10

Author: ccrisan

.flake8

@@ -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

.github/workflows/main.yml

@@ -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

.pre-commit-config.yaml

@@ -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

pyproject.toml

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

qtoggleserver/cmdline/__init__.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/exceptions.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/peripheral.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