Skip to content

Commit

Permalink
update annotation style
Browse files Browse the repository at this point in the history
  • Loading branch information
avaldebe committed Dec 8, 2022
1 parent 7063566 commit c577289
Show file tree
Hide file tree
Showing 5 changed files with 31 additions and 24 deletions.
23 changes: 12 additions & 11 deletions src/pms/core/reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
- Sensors are read on passive mode.
- Tested on PMS3003, PMS7003, PMSA003, SDS011 and MCU680
"""
from __future__ import annotations

import sys
import time
Expand All @@ -13,7 +14,7 @@
from csv import DictReader
from pathlib import Path
from textwrap import wrap
from typing import Iterator, NamedTuple, Optional, Union, overload
from typing import Iterator, NamedTuple

from serial import Serial
from typer import progressbar
Expand Down Expand Up @@ -43,7 +44,7 @@ class RawData(NamedTuple):
def hex(self) -> str:
return self.data.hex()

def hexdump(self, line: Optional[int] = None) -> str:
def hexdump(self, line: int | None = None) -> str:
offset = time if line is None else line * len(self.data)
hex = " ".join(wrap(self.data.hex(), 2)) # raw.hex(" ") in python3.8+
dump = self.data.translate(HEXDUMP_TABLE).decode()
Expand All @@ -52,7 +53,7 @@ def hexdump(self, line: Optional[int] = None) -> str:

class Reader:
@abstractmethod
def __call__(self, *, raw: Optional[bool]) -> Iterator[Union[RawData, ObsData]]:
def __call__(self, *, raw: bool | None = None) -> Iterator[RawData | ObsData]:
"""
Return an iterator of ObsData.
Expand Down Expand Up @@ -88,12 +89,12 @@ class SensorReader(Reader):

def __init__(
self,
sensor: Union[Sensor, Supported, str] = Supported.default,
sensor: Sensor | Supported | str = Supported.default,
port: str = "/dev/ttyUSB0",
interval: Optional[int] = None,
samples: Optional[int] = None,
timeout: Optional[float] = None,
max_retries: Optional[int] = None,
interval: int | None = None,
samples: int | None = None,
timeout: float | None = None,
max_retries: int | None = None,
) -> None:
"""Configure serial port"""
self.sensor = sensor if isinstance(sensor, Sensor) else Sensor[sensor]
Expand Down Expand Up @@ -167,7 +168,7 @@ def close(self) -> None:
logger.debug(f"close {self.serial.port}")
self.serial.close()

def __call__(self, *, raw: Optional[bool] = None):
def __call__(self, *, raw: bool | None = None) -> Iterator[RawData | ObsData]:
"""Passive mode reading at regular intervals"""

sample = 0
Expand Down Expand Up @@ -205,7 +206,7 @@ def __call__(self, *, raw: Optional[bool] = None):


class MessageReader(Reader):
def __init__(self, path: Path, sensor: Sensor, samples: Optional[int] = None) -> None:
def __init__(self, path: Path, sensor: Sensor, samples: int | None = None) -> None:
self.path = path
self.sensor = sensor
self.samples = samples
Expand All @@ -220,7 +221,7 @@ def close(self) -> None:
logger.debug(f"close {self.path}")
self.csv.close()

def __call__(self, *, raw: Optional[bool] = None):
def __call__(self, *, raw: bool | None = None) -> Iterator[RawData | ObsData]:
if not hasattr(self, "data"):
return

Expand Down
3 changes: 2 additions & 1 deletion src/pms/core/sensor.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""
Access supported sensors from a single object
"""
from __future__ import annotations

import sys
from datetime import datetime
Expand Down Expand Up @@ -89,7 +90,7 @@ def check(self, buffer: bytes, command: str) -> bool:
else:
return True

def decode(self, buffer: bytes, *, time: int = None) -> ObsData:
def decode(self, buffer: bytes, *, time: int | None = None) -> ObsData:
"""Extract observations from serial buffer"""
if not time: # pragma: no cover
time = self.now()
Expand Down
8 changes: 5 additions & 3 deletions src/pms/extra/influxdb.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
from __future__ import annotations

import json
from dataclasses import fields
from typing import Dict, Protocol
from typing import Protocol

from typer import Abort, Context, Option, colors, echo, style

Expand Down Expand Up @@ -33,7 +35,7 @@ def __missing_influxdb(): # pragma: no cover


class PubFunction(Protocol): # pragma: no cover
def __call__(self, *, time: int, tags: Dict[str, str], data: Dict[str, float]) -> None:
def __call__(self, *, time: int, tags: dict[str, str], data: dict[str, float]) -> None:
...


Expand All @@ -48,7 +50,7 @@ def client_pub(
c.create_database(db_name)
c.switch_database(db_name)

def pub(*, time: int, tags: Dict[str, str], data: Dict[str, float]) -> None:
def pub(*, time: int, tags: dict[str, str], data: dict[str, float]) -> None:
c.write_points(
[
{"measurement": k, "tags": tags, "time": time, "fields": {"value": v}}
Expand Down
11 changes: 6 additions & 5 deletions src/pms/extra/mqtt.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import sys
from __future__ import annotations

from dataclasses import fields
from datetime import datetime
from typing import Callable, Dict, NamedTuple, Union
from typing import Callable, NamedTuple

from typer import Abort, Context, Option, colors, echo, style

Expand Down Expand Up @@ -37,7 +38,7 @@ def __missing_mqtt(): # pragma: no cover

def client_pub(
*, topic: str, host: str, port: int, username: str, password: str
) -> Callable[[Dict[str, Union[int, str]]], None]: # pragma: no cover
) -> Callable[[dict[str, int | str]], None]: # pragma: no cover
if client is None:
__missing_mqtt()
c = client.Client(topic)
Expand All @@ -52,7 +53,7 @@ def client_pub(
c.connect(host, port, 60)
c.loop_start()

def pub(data: Dict[str, Union[int, str]]) -> None:
def pub(data: dict[str, int | str]) -> None:
for k, v in data.items():
c.publish(f"{topic}/{k}", v, 1, True)

Expand All @@ -71,7 +72,7 @@ def now() -> int: # pragma: no cover
return int(datetime.now().timestamp())

@classmethod
def decode(cls, topic: str, payload: str, *, time: int = None) -> "Data":
def decode(cls, topic: str, payload: str, *, time: int | None = None) -> "Data":
"""Decode a MQTT message
For example
Expand Down
10 changes: 6 additions & 4 deletions src/pms/sensors/base.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
from __future__ import annotations

import warnings
from abc import ABCMeta, abstractmethod
from dataclasses import asdict, dataclass
from datetime import datetime
from typing import Dict, NamedTuple, Tuple
from typing import NamedTuple, Tuple

from pms import WrongMessageFormat, logger

Expand Down Expand Up @@ -58,7 +60,7 @@ def decode(cls, message: bytes, command: Cmd) -> Tuple[float, ...]:
length = command.answer_length
return cls.unpack(message, header, length)[cls.data_records] # type: ignore[call-overload]

@property
@property # type:ignore[misc]
@classmethod
@abstractmethod
def data_records(cls) -> slice:
Expand Down Expand Up @@ -90,7 +92,7 @@ def _unpack(message: bytes) -> Tuple[float, ...]:
pass


@dataclass # type: ignore[misc]
@dataclass
class ObsData(metaclass=ABCMeta):
"""Measurements
Expand All @@ -105,7 +107,7 @@ def date(self) -> datetime:
"""measurement time as datetime object"""
return datetime.fromtimestamp(self.time)

def subset(self, spec: str = None) -> Dict[str, float]: # pragma: no cover
def subset(self, spec: str | None = None) -> dict[str, float]: # pragma: no cover
warnings.warn(
"obs.subset is deprecated, use dataclasses.asdict(obs) for a dictionary mapping",
DeprecationWarning,
Expand Down

0 comments on commit c577289

Please sign in to comment.