oddstr13/srne-mqtt
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Fix mypy issues

Browse source
This commit is contained in:
Odd Stråbø 2023-12-09 16:35:45 +01:00
parent 8282ec3956
commit 457e7cf8a3
10 changed files with 56 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.pre-commit-config.yaml
View file

@ -5,7 +5,7 @@ repos:
rev: v4.4.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
#- id: end-of-file-fixer
- id: fix-byte-order-marker
- id: fix-encoding-pragma
- id: check-executables-have-shebangs
@ -41,6 +41,9 @@ repos:
args:
- "--install-types"
- "--non-interactive"
- "--check-untyped-defs"
additional_dependencies:
- typing_extensions==4.8.0
- repo: https://github.com/psf/black
rev: 23.3.0

8
.vscode/settings.json vendored
View file

@ -4,5 +4,11 @@
"--disable=missing-function-docstring,missing-class-docstring,missing-module-docstring"
],
"python.testing.unittestEnabled": false,
"python.testing.pytestEnabled": true
"python.testing.pytestEnabled": true,
"mypy-type-checker.importStrategy": "fromEnvironment",
"mypy-type-checker.reportingScope": "workspace",
"mypy-type-checker.preferDaemon": true,
"mypy-type-checker.args": [
"--check-untyped-defs"
]
}

8
misc/dump_memory_map.py
View file

@ -1,7 +1,6 @@
# -*- coding: utf-8 -*-
import os
import sys
from io import RawIOBase
from typing import List
sys.path.insert(1, os.path.dirname(os.path.dirname(sys.argv[0])))
@ -9,10 +8,11 @@ sys.path.insert(1, os.path.dirname(os.path.dirname(sys.argv[0])))
from draw_memory_map import memory_table # noqa: E402
from srnemqtt.config import get_config, get_interface # noqa: E402
from srnemqtt.interfaces import BaseInterface # noqa: E402
from srnemqtt.protocol import readMemory # noqa: E402
def get_device_name(iface: RawIOBase) -> str | None:
def get_device_name(iface: BaseInterface) -> str | None:
data = readMemory(iface, 0x0C, 8)
if data is None:
return None
@ -20,7 +20,7 @@ def get_device_name(iface: RawIOBase) -> str | None:
return data.decode("utf-8").strip()
def get_device_version(iface: RawIOBase) -> str | None:
def get_device_version(iface: BaseInterface) -> str | None:
data = readMemory(iface, 0x14, 4)
if data is None:
return None
@ -32,7 +32,7 @@ def get_device_version(iface: RawIOBase) -> str | None:
return f"{major}.{minor}.{patch}"
def get_device_serial(iface: RawIOBase) -> str | None:
def get_device_serial(iface: BaseInterface) -> str | None:
data = readMemory(iface, 0x18, 3)
if data is None:
return None

3
misc/test_serial.py
View file

@ -11,7 +11,8 @@ sys.path.insert(1, os.path.dirname(os.path.dirname(sys.argv[0])))
# from srnemqtt.lib.feasycom_ble import BTLEUart
from srnemqtt.protocol import construct_request, write # noqa: E402
for rate in [1200, 2400, 4800, 9600, 115200]:
# for rate in [1200, 2400, 4800, 9600, 115200]:
for rate in [9600]:
print(rate)
with Serial("/dev/ttyUSB0", baudrate=rate, timeout=2) as x:
sleep(2)

3
srnemqtt/config.py
View file

@ -6,9 +6,8 @@ from typing import Any, Dict, List, Optional, Type
import yaml
from srnemqtt.interfaces import BaseInterface
from .consumers import BaseConsumer
from .interfaces import BaseInterface
def get_consumer(name: str) -> Optional[Type[BaseConsumer]]:

12
srnemqtt/consumers/mqtt.py
View file

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
import json
from time import sleep
from typing import Any, Dict, List, Optional
from typing import Any, Dict, List, Optional, TypeAlias
from uuid import uuid4
import paho.mqtt.client as mqtt
@ -112,6 +112,9 @@ MAP_VALUES: Dict[DataName, Dict[str, Any]] = {
}
PayloadType: TypeAlias = str | bytes | bytearray | int | float | None
class MqttConsumer(BaseConsumer):
client: mqtt.Client
initialized: List[str]
@ -170,8 +173,8 @@ class MqttConsumer(BaseConsumer):
def get_ha_config(
self,
id,
name,
id: str,
name: str,
unit: Optional[str] = None,
type: Optional[str] = None,
expiry: int = 90,
@ -247,7 +250,7 @@ class MqttConsumer(BaseConsumer):
return super().poll()
def write(self, data: Dict[str, Any]):
def write(self, data: Dict[str, PayloadType]):
self.client.publish(f"{self.topic_prefix}/raw", payload=json.dumps(data))
for k, v in data.items():
@ -257,6 +260,7 @@ class MqttConsumer(BaseConsumer):
pretty_name = k.replace("_", " ").capitalize()
disc_prefix = self.settings["discovery_prefix"]
device_id = self.settings["device_id"]
self.client.publish(
f"{disc_prefix}/sensor/{device_id}_{k}/config",
payload=json.dumps(self.get_ha_config(k, pretty_name, **km)),

2
srnemqtt/interfaces/__init__.py
View file

@ -4,4 +4,4 @@ from io import RawIOBase
class BaseInterface(RawIOBase, metaclass=ABCMeta):
pass
timeout: float | None

32
srnemqtt/lib/feasycom_ble.py
View file

@ -6,10 +6,6 @@ from typing import TYPE_CHECKING, Optional, cast
from bluepy import btle # type: ignore
if TYPE_CHECKING:
from _typeshed import ReadableBuffer, WriteableBuffer
WRITE_DEVICE = "0000ffd1-0000-1000-8000-00805f9b34fb"
READ_DEVICE = "0000fff1-0000-1000-8000-00805f9b34fb"
@ -18,7 +14,7 @@ class BTLEUart(io.RawIOBase):
mac: str
write_endpoint: str
read_endpoint: str
timeout: float
timeout: float | None
device: Optional[btle.Peripheral] = None
_write_handle: Optional[btle.Characteristic] = None
@ -86,13 +82,12 @@ class BTLEUart(io.RawIOBase):
self._write_handle = self.device.getCharacteristics(uuid=self.write_endpoint)[0]
# print("Handles:", self._read_handle.handle, self._write_handle.handle)
def _read(self, num: Optional[int] = None, timeout: Optional[float] = None):
def _read(self, num: Optional[int] = None):
self._ensure_connected()
if TYPE_CHECKING:
self.device = cast(btle.Peripheral, self.device)
if timeout is None:
timeout = self.timeout
timeout = self.timeout or 30
if num is None:
start = time.time()
@ -132,7 +127,9 @@ class BTLEUart(io.RawIOBase):
del self._read_buffer[:num]
return data or None
def readinto(self, buffer: "WriteableBuffer") -> Optional[int]:
def readinto(self, buffer: bytearray | memoryview) -> Optional[int]: # type: ignore [override]
# Buffer does not provide Sized, and bytes is read only.
# bytearray | memoryview is the default implementations that provide WriteableBuffer
data = self._read(len(buffer))
if data is None:
@ -144,23 +141,15 @@ class BTLEUart(io.RawIOBase):
def readall(self) -> bytes:
return self._read()
def read(
self, size: Optional[int] = None, timeout: Optional[float] = None
) -> Optional[bytes]:
if timeout:
_timeout = self.timeout
self.timeout = timeout
def read(self, size: Optional[int] = None) -> Optional[bytes]:
if size is None:
res = super().read()
else:
res = super().read(size)
if timeout:
self.timeout = _timeout
return res
def write(self, b: "ReadableBuffer") -> Optional[int]:
def write(self, b: bytes | bytearray | memoryview) -> Optional[int]: # type: ignore [override]
self._ensure_connected()
if TYPE_CHECKING:
self.device = cast(btle.Peripheral, self.device)
@ -174,8 +163,9 @@ class BTLEUart(io.RawIOBase):
return self
def __exit__(self, type, value, traceback):
self.device.disconnect()
del self.device
if self.device is not None:
self.device.disconnect()
self.device = None
def seekable(self) -> bool:
return False

31
srnemqtt/protocol.py
View file

@ -2,13 +2,12 @@
import struct
import sys
import time
from io import RawIOBase
from typing import Callable, Collection, Optional
from libscrc import modbus # type: ignore
from .constants import ACTION_READ, POSSIBLE_MARKER
from .lib.feasycom_ble import BTLEUart
from .interfaces import BaseInterface
from .solar_types import DATA_BATTERY_STATE, HISTORICAL_DATA, DataItem
from .util import log
@ -61,23 +60,23 @@ def parse_packet(data):
if crc != calculated_crc:
e = ValueError(f"CRC missmatch: expected {crc:04X}, got {calculated_crc:04X}.")
e.tag = tag
e.operation = operation
e.size = size
e.payload = payload
e.crc = crc
e.calculated_crc = calculated_crc
# e.tag = tag
# e.operation = operation
# e.size = size
# e.payload = payload
# e.crc = crc
# e.calculated_crc = calculated_crc
raise e
return payload
def discardUntil(fh: RawIOBase, byte: int, timeout=10) -> Optional[int]:
def discardUntil(fh: BaseInterface, byte: int, timeout=10) -> Optional[int]:
assert byte >= 0 and byte < 256, f"byte: Expected 8bit unsigned int, got {byte}"
def expand(b: Optional[bytes]):
if b is None:
return b
if not b:
return None
return b[0]
start = time.time()
@ -88,6 +87,7 @@ def discardUntil(fh: RawIOBase, byte: int, timeout=10) -> Optional[int]:
if not discarded:
log("Discarding", end="")
discarded += 1
print(read_byte)
print(f" {read_byte:02X}", end="")
sys.stdout.flush()
@ -104,7 +104,7 @@ def discardUntil(fh: RawIOBase, byte: int, timeout=10) -> Optional[int]:
return read_byte
def readMemory(fh: RawIOBase, address: int, words: int = 1) -> Optional[bytes]:
def readMemory(fh: BaseInterface, address: int, words: int = 1) -> Optional[bytes]:
# log(f"Reading {words} words from 0x{address:04X}")
request = construct_request(address, words=words)
# log("Request:", request)
@ -135,7 +135,7 @@ def readMemory(fh: RawIOBase, address: int, words: int = 1) -> Optional[bytes]:
def try_read_parse(
dev: BTLEUart,
dev: BaseInterface,
address: int,
words: int = 1,
parser: Optional[Callable] = None,
@ -151,7 +151,10 @@ def try_read_parse(
except struct.error as e:
log(e)
log("0x0100 Unpack error:", len(res), res)
log("Flushed from read buffer; ", dev.read(timeout=0.5))
_timeout = dev.timeout
dev.timeout = 0.5
log("Flushed from read buffer; ", dev.read())
dev.timeout = _timeout
else:
log(f"No data read, expected {words*2} bytes (attempts left: {attempts})")
return None

2
tox.ini
View file

@ -1,5 +1,5 @@
[flake8]
max-line-length = 88
max-line-length = 120
extend-ignore = E203, I201, I101
[pytest]