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,33 @@
+[project]
+name = "qtoggleserver-pylontech"
+version = "0.0.0"
+description = "Pylontech batteries support for qToggleServer"
+authors = [
+    {name = "Calin Crisan", email = "ccrisan@gmail.com"},
+]
+requires-python = "==3.10.*"
+readme = "README.md"
+license = {text = "Apache 2.0"}
+dependencies = [
+    "pyserial",
+    "python-pylontech",
+]
+
+[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/pylontech/__init__.py

@@ -1,4 +1,6 @@
 from .battery import Battery
 
 
-VERSION = 'unknown'
+__all__ = ["Battery"]
+
+VERSION = "unknow-version"

qtoggleserver/pylontech/battery.py

@@ -3,10 +3,9 @@
 import re
 import time
 
-from typing import Any, Optional, Union
+from typing import Any
 
 from pylontech import Pylontech
-
 from qtoggleserver.core import ports as core_ports
 from qtoggleserver.lib import polled
 
@@ -15,24 +14,24 @@ class ImprovedPylontech(Pylontech):
     CHUNK_SIZE = 1024
 
     def __init__(self, *args, **kwargs) -> None:
-        self._buffer: bytes = b''
+        self._buffer: bytes = b""
         super().__init__(*args, **kwargs)
 
     def read_line(self) -> bytes:
-        parts = re.split(rb'[\r\n]', self._buffer, maxsplit=1)
+        parts = re.split(rb"[\r\n]", self._buffer, maxsplit=1)
         if len(parts) > 1:
             line, self._buffer = parts
-            return line.strip(b'\xff') + b'\n'
+            return line.strip(b"\xff") + b"\n"
 
         data = self.s.read(self.CHUNK_SIZE)
-        parts = re.split(rb'[\r\n]', data, maxsplit=1)
+        parts = re.split(rb"[\r\n]", data, maxsplit=1)
         if len(parts) > 1:
             line, rest = parts
             self._buffer += rest
-            return line.strip(b'\xff') + b'\n'
+            return line.strip(b"\xff") + b"\n"
         else:
             self._buffer += data
-            return b''
+            return b""
 
     def read_frame(self) -> Any:
         raw_frame = self.read_line()
@@ -43,7 +42,7 @@ def read_frame(self) -> Any:
 
 class Battery(polled.PolledPeripheral):
     DEFAULT_POLL_INTERVAL = 60
-    DEFAULT_SERIAL_PORT = '/dev/ttyUSB0'
+    DEFAULT_SERIAL_PORT = "/dev/ttyUSB0"
     DEFAULT_SERIAL_BAUD = 115200
     READ_RETRY_COUNT = 5
     READ_RETRY_SLEEP = 3
@@ -66,14 +65,14 @@ def __init__(
         self._dev_ids: list[int] = dev_ids
         self._statuses_by_dev_id: dict[int, dict[str, Any]] = {}
 
-    async def make_port_args(self) -> list[Union[dict[str, Any], type[core_ports.BasePort]]]:
-        from .ports import SocPort, TemperaturePort, CurrentPort, VoltagePort, PowerPort, CyclesPort
+    async def make_port_args(self) -> list[dict[str, Any] | type[core_ports.BasePort]]:
+        from .ports import CurrentPort, CyclesPort, PowerPort, SocPort, TemperaturePort, VoltagePort
 
         status_port_drivers = [SocPort, TemperaturePort, CurrentPort, VoltagePort, PowerPort, CyclesPort]
         port_args = [
             {
-                'driver': driver,
-                'id': driver.get_property(),
+                "driver": driver,
+                "id": driver.get_property(),
             }
             for driver in status_port_drivers
         ]
@@ -98,41 +97,41 @@ def _poll_dev(self, dev_id: int) -> dict:
                     raise
                 else:
                     self.warning(
-                        'reading values for device %s failed (retry=%d/%d)',
+                        "reading values for device %s failed (retry=%d/%d)",
                         dev_id,
                         count + 1,
-                        self.READ_RETRY_COUNT - 1
+                        self.READ_RETRY_COUNT - 1,
                     )
                     time.sleep(self.READ_RETRY_SLEEP)
             finally:
                 pylontech.s.close()
 
         return {
-            'timestamp': int(time.time()),
-            'soc': values.StateOfCharge,
-            'temperature': values.AverageBMSTemperature,
-            'current': values.Current,
-            'voltage': values.Voltage,
-            'power': values.Power,
-            'cycles': values.CycleNumber,
+            "timestamp": int(time.time()),
+            "soc": values.StateOfCharge,
+            "temperature": values.AverageBMSTemperature,
+            "current": values.Current,
+            "voltage": values.Voltage,
+            "power": values.Power,
+            "cycles": values.CycleNumber,
         }
 
-    def get_aggregated_status(self) -> Optional[dict[str, Any]]:
+    def get_aggregated_status(self) -> dict[str, Any] | None:
         if not self._statuses_by_dev_id:
             return None
 
         statuses = list(self._statuses_by_dev_id.values())
         agg_status = dict(statuses[0])
 
         for status in statuses[1:]:
-            agg_status['timestamp'] = min(agg_status['timestamp'], status['timestamp'])
-            agg_status['soc'] = min(agg_status['soc'], status['soc'])
-            agg_status['temperature'] = max(agg_status['temperature'], status['temperature'])
-            agg_status['current'] += status['current']
-            agg_status['voltage'] += status['voltage']
-            agg_status['power'] += status['power']
-            agg_status['cycles'] = max(agg_status['cycles'], status['cycles'])
-
-        agg_status['voltage'] /= len(statuses)
+            agg_status["timestamp"] = min(agg_status["timestamp"], status["timestamp"])
+            agg_status["soc"] = min(agg_status["soc"], status["soc"])
+            agg_status["temperature"] = max(agg_status["temperature"], status["temperature"])
+            agg_status["current"] += status["current"]
+            agg_status["voltage"] += status["voltage"]
+            agg_status["power"] += status["power"]
+            agg_status["cycles"] = max(agg_status["cycles"], status["cycles"])
+
+        agg_status["voltage"] /= len(statuses)
 
         return agg_status

qtoggleserver/pylontech/ports.py

@@ -8,7 +8,7 @@
 
 
 class BatteryStatusPort(polled.PolledPort):
-    PROPERTY = ''
+    PROPERTY = ""
 
     @classmethod
     def get_property(cls) -> str:
@@ -28,29 +28,29 @@ async def read_value(self) -> NullablePortValue:
 
 class SocPort(BatteryStatusPort):
     TYPE = core_ports.TYPE_NUMBER
-    PROPERTY = 'soc'
+    PROPERTY = "soc"
 
 
 class TemperaturePort(BatteryStatusPort):
     TYPE = core_ports.TYPE_NUMBER
-    PROPERTY = 'temperature'
+    PROPERTY = "temperature"
 
 
 class CurrentPort(BatteryStatusPort):
     TYPE = core_ports.TYPE_NUMBER
-    PROPERTY = 'current'
+    PROPERTY = "current"
 
 
 class VoltagePort(BatteryStatusPort):
     TYPE = core_ports.TYPE_NUMBER
-    PROPERTY = 'voltage'
+    PROPERTY = "voltage"
 
 
 class PowerPort(BatteryStatusPort):
     TYPE = core_ports.TYPE_NUMBER
-    PROPERTY = 'power'
+    PROPERTY = "power"
 
 
 class CyclesPort(BatteryStatusPort):
     TYPE = core_ports.TYPE_NUMBER
-    PROPERTY = 'cycles'
+    PROPERTY = "cycles"