type annotations

Author: avaldebe

docs/examples/inspect_data_fields.py

@@ -11,7 +11,7 @@ def field_types(obs: ObsData) -> dict[str, str]:
 
     return {
         field.name: field.type.__name__ if hasattr(field.type, "__name__") else str(field.type)
-        for field in fields(obs)  # type:ignore[arg-type]
+        for field in fields(obs)
     }
 
 

docs/examples/sqlite.py

@@ -17,7 +17,8 @@
 import typer
 
 from pms.core import Sensor, SensorReader, Supported
-from pms.core.reader import ObsData, RawData
+from pms.core.reader import RawData
+from pms.core.types import ObsData
 
 app = typer.Typer(add_completion=False)
 

docs/examples/sqlite_measurements.py

@@ -25,7 +25,7 @@
 import typer
 
 from pms.core import Sensor, SensorReader
-from pms.core.reader import ObsData
+from pms.core.types import ObsData
 
 app = typer.Typer(add_completion=False)
 
@@ -104,7 +104,7 @@ def connect() -> Iterator[sqlite3.Connection]:
         for sensor in Sensor:
             view_fields = ",\n".join(
                 f"MAX(CASE WHEN field='{field.name}' THEN value ELSE NULL END) {field.name}"
-                for field in fields(sensor.Data)  # type:ignore[arg-type]
+                for field in fields(sensor.Data)
                 if field.name != "time"
             )
             sensor_view = f"""
@@ -135,7 +135,7 @@ def write_measurements(db: sqlite3.Connection, sensor: Sensor, obs: ObsData):
         """
     values = (
         (obs.time, sensor.name, field, value)
-        for field, value in asdict(obs).items()  # type:ignore[call-overload]
+        for field, value in asdict(obs).items()
         if field != "time"
     )
     with db, closing(db.cursor()) as cur:

docs/examples/sqlite_message.py

@@ -22,7 +22,8 @@
 import typer
 
 from pms.core import Sensor, SensorReader
-from pms.core.reader import ObsData, RawData
+from pms.core.reader import RawData
+from pms.core.types import ObsData
 
 app = typer.Typer(add_completion=False)
 

src/pms/core/types.py

@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
 from datetime import datetime
 from typing import Protocol
 
@@ -38,16 +41,14 @@ class Message(Protocol):
     """Message from sensor"""
 
     @classmethod
-    def decode(cls, message: bytes, command: Cmd) -> tuple[float, ...]: ...
+    def decode(cls, message: bytes, command: Cmd) -> tuple[int | float, ...]: ...
 
 
+@dataclass
 class ObsData(Protocol):
     """Decoded sensor message"""
 
-    def __init__(self, *data: float) -> None: ...
-
-    @property
-    def time(self) -> int: ...
+    time: int
 
     @property
     def date(self) -> datetime: ...

src/pms/extra/cli.py

@@ -14,7 +14,7 @@
 from loguru import logger
 
 from pms.core import exit_on_fail
-from pms.sensors.base import ObsData
+from pms.core.types import ObsData
 
 DB_HOST: TypeAlias = Annotated[str, typer.Option("--db-host", help="database server")]
 DB_PORT: TypeAlias = Annotated[int, typer.Option("--db-port", help="server port")]

src/pms/sensors/base.py

@@ -34,6 +34,8 @@ class Message(metaclass=ABCMeta):
     Base class for serial messages from PM sensors
     """
 
+    data_records: slice
+
     def __init__(self, message: bytes) -> None:
         logger.debug(f"message hex: {message.hex()}")
         self.message = message
@@ -60,13 +62,7 @@ def unpack(cls, message: bytes, header: bytes, length: int) -> tuple[float, ...]
     def decode(cls, message: bytes, command: Cmd) -> tuple[float, ...]:
         header = command.answer_header
         length = command.answer_length
-        return cls.unpack(message, header, length)[cls.data_records]  # type: ignore[call-overload]
-
-    @property  # type:ignore[misc]
-    @classmethod
-    @abstractmethod
-    def data_records(cls) -> slice:
-        pass
+        return cls.unpack(message, header, length)[cls.data_records]
 
     @property
     @abstractmethod

tests/extra/test_cli.py

@@ -7,9 +7,9 @@
 from loguru import logger
 from typer.testing import CliRunner
 
+from pms.core.types import ObsData
 from pms.extra.cli import db_measurements, mqtt_messages
 from pms.main import main
-from pms.sensors.base import ObsData
 
 runner = CliRunner()