From 09968e426da99708d66a7ede62c0786d380976d8 Mon Sep 17 00:00:00 2001 From: Markus Becker Date: Mon, 19 Oct 2026 09:20:35 +0200 Subject: [PATCH 1/3] Type the driver base, daliserver and trace logging for mypy --strict Annotate the abstract driver/backend contracts, the daliserver client and the TRACE logging helper so they pass mypy --strict. Add a shared TraceLogger type and get_logger() factory in trace_logging for the serial and HID drivers to reuse. Fix the daliserver default result to be bytes and Backend.write to take self. Typing only, no behaviour change. Co-Authored-By: Claude Opus 4.8 --- dali/driver/base.py | 37 ++++++++++++++++++++++++------------ dali/driver/daliserver.py | 22 +++++++++++---------- dali/driver/trace_logging.py | 22 +++++++++++++++++---- 3 files changed, 55 insertions(+), 26 deletions(-) diff --git a/dali/driver/base.py b/dali/driver/base.py index 714df187..c8a4d369 100644 --- a/dali/driver/base.py +++ b/dali/driver/base.py @@ -1,6 +1,14 @@ ############################################################################### # driver contracts ############################################################################### +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Callable, Optional + +if TYPE_CHECKING: + from dali.command import Command + from dali.frame import Frame + class DALIDriver: """Object for handling wrapped DALI frames sent to and received from @@ -10,10 +18,10 @@ class DALIDriver: communicate with a DALI bus. """ - backend = None + backend: Optional[Backend] = None """``Backend`` instance used for reading and writing.""" - def construct(self, command): + def construct(self, command: Command) -> bytes: """Construct raw data containing the packed DALI frame from command which can be sent to DALI driver. @@ -23,7 +31,7 @@ def construct(self, command): raise NotImplementedError( 'Abstract ``DaliDriver`` does not implement ```construct`') - def extract(self, data): + def extract(self, data: bytes) -> Frame: """Extract and return DALI Frame from data which has been received from DALI driver. @@ -39,7 +47,7 @@ class SyncDALIDriver(DALIDriver): expecting and returning a result. """ - def send(self, command, timeout=None): + def send(self, command: Command, timeout: Optional[int] = None) -> Any: """Send command to gateway and return response. @param command: DALI command to send. @@ -55,11 +63,16 @@ class AsyncDALIDriver(DALIDriver): DALI drivers. """ - dispatcher = None + dispatcher: Optional[Callable[..., Any]] = None """Callable used for dispatching incoming forward frames. """ - def send(self, command, callback=None, **kw): + def send( + self, + command: Command, + callback: Optional[Callable[..., Any]] = None, + **kw: Any, + ) -> None: """Send command to gateway. @param command: DALI command to send. @@ -70,7 +83,7 @@ def send(self, command, callback=None, **kw): raise NotImplementedError( 'Abstract ``AsyncDALIDriver`` does not implement ``send``') - def receive(self, frame): + def receive(self, frame: Frame) -> None: """Receive DALI Frame. If a forward frame is received, ``self.dispatcher`` is used for @@ -93,15 +106,15 @@ class Backend: """Object representing a backend. """ - def read(self, timeout=None): + def read(self, timeout: Optional[int] = None) -> Any: raise NotImplementedError( 'Abstract ``Backend`` does not implement ``read``') - def write(data): + def write(self, data: bytes) -> None: raise NotImplementedError( 'Abstract ``Backend`` does not implement ``write``') - def close(self): + def close(self) -> None: raise NotImplementedError( 'Abstract ``Backend`` does not implement ``close``') @@ -111,11 +124,11 @@ class Listener(Backend): incoming frames. """ - driver = None + driver: Optional[AsyncDALIDriver] = None """``AsyncDALIDriver`` instance. """ - def listen(self): + def listen(self) -> None: """Listen to backend and dispatch receiced data to ``self.driver``. """ raise NotImplementedError( diff --git a/dali/driver/daliserver.py b/dali/driver/daliserver.py index 6fdd8538..b53b05e3 100644 --- a/dali/driver/daliserver.py +++ b/dali/driver/daliserver.py @@ -1,4 +1,6 @@ -from dali.command import Command +from typing import Any, Optional + +from dali.command import Command, Response from dali.exceptions import CommunicationError import dali.frame import logging @@ -19,23 +21,23 @@ class DaliServer: 90e34a0cd2945dc7a15681f11647e708f858521e or later. """ - def __init__(self, host="localhost", port=55825, - multiple_frames_per_connection=False): + def __init__(self, host: str = "localhost", port: int = 55825, + multiple_frames_per_connection: bool = False) -> None: self._target = (host, port) - self._s = None + self._s: Optional[socket.socket] = None self._multiple_frames_per_connection = multiple_frames_per_connection - def __enter__(self): + def __enter__(self) -> "DaliServer": if self._multiple_frames_per_connection: self._s = socket.create_connection(self._target) return self - def __exit__(self, *vpass): - if self._multiple_frames_per_connection: + def __exit__(self, *vpass: Any) -> None: + if self._s is not None: self._s.close() self._s = None - def send(self, command): + def send(self, command: Command) -> Optional[Response]: if self._s: s = self._s else: @@ -48,7 +50,7 @@ def send(self, command): command, " (twice)" if command.sendtwice else "")) # Set a default result which may be used if the first send fails - result = "\x02\xff\x00\x00" + result = b"\x02\xff\x00\x00" try: s.send(message) @@ -69,7 +71,7 @@ def send(self, command): return response - def unpack_response(self, command, result): + def unpack_response(self, command: Command, result: bytes) -> Optional[Response]: """Unpack result from the given bytestream and creates the corresponding response object diff --git a/dali/driver/trace_logging.py b/dali/driver/trace_logging.py index 23ee2e4b..abecbf82 100644 --- a/dali/driver/trace_logging.py +++ b/dali/driver/trace_logging.py @@ -16,12 +16,24 @@ If not, see . """ import logging +from typing import Any, cast -_LOG = logging.getLogger(f"dali") +_LOG = logging.getLogger("dali") _logging_trace_created = False -def add_logging_level(level_name: str, level_num: int): +class TraceLogger(logging.Logger): + """A logger exposing the extra TRACE level installed by this module.""" + + def trace(self, msg: object, *args: object, **kwargs: object) -> None: ... + + +def get_logger(name: str) -> TraceLogger: + """Return a logger typed as exposing the TRACE level.""" + return cast(TraceLogger, logging.getLogger(name)) + + +def add_logging_level(level_name: str, level_num: int) -> None: """ Original source: https://stackoverflow.com/a/35804945 @@ -46,11 +58,13 @@ def add_logging_level(level_name: str, level_num: int): # This method was inspired by the answers to Stack Overflow post # http://stackoverflow.com/q/2183233/2988730, especially # http://stackoverflow.com/a/13638084/2988730 - def logForLevel(self, message, *args, **kwargs): + def logForLevel( + self: logging.Logger, message: object, *args: Any, **kwargs: Any + ) -> None: if self.isEnabledFor(level_num): self._log(level_num, message, args, **kwargs) - def logToRoot(message, *args, **kwargs): + def logToRoot(message: object, *args: Any, **kwargs: Any) -> None: logging.log(level_num, message, *args, **kwargs) logging.addLevelName(level_num, level_name) From 0673e6cb891c0a518bb4b2607242b6fcd00c8f0e Mon Sep 17 00:00:00 2001 From: Markus Becker Date: Mon, 19 Oct 2026 09:29:45 +0200 Subject: [PATCH 2/3] Type the serial driver for mypy --strict Annotate every function in the serial driver and resolve the strict findings: use the shared TraceLogger, add _proto/_tx accessors that narrow the optional protocol and transport, type the RX buffers as int lists, materialise the frame byte sequence before indexing, decode received frames as ForwardFrames, bind the response type locally before calling it, and cast the connection factory results. send() now accepts any sequence generator. Typing only, no behaviour change. Co-Authored-By: Claude Opus 4.8 --- dali/driver/serial.py | 276 +++++++++++++++++++++++------------------- 1 file changed, 152 insertions(+), 124 deletions(-) diff --git a/dali/driver/serial.py b/dali/driver/serial.py index 2c5c7290..c5731c20 100644 --- a/dali/driver/serial.py +++ b/dali/driver/serial.py @@ -27,18 +27,18 @@ from enum import Enum from functools import reduce from operator import xor -from typing import Any, Callable, Generator, NamedTuple, Optional +from typing import Any, Callable, Generator, NamedTuple, Optional, Sequence, cast from urllib.parse import ParseResult, urlparse, urlunparse import serialx import dali.gear from dali import command, frame, gear, sequences -from dali.driver import trace_logging # noqa: F401 +from dali.driver.trace_logging import get_logger from dali.device.helpers import DeviceInstanceTypeMapper from dali.exceptions import TransmissionError -_LOG = logging.getLogger("dali.driver") +_LOG = get_logger("dali.driver") # Sentinel placed on the raw DALI response queue when the interface reports a # framing error, so a waiting query returns an error response instead of @@ -48,9 +48,9 @@ _DALI_FRAMING_ERROR = object() -def _drain_queue(queue: asyncio.Queue) -> list: +def _drain_queue(queue: asyncio.Queue[Any]) -> list[Any]: """Remove and return every currently-queued item without blocking.""" - items = [] + items: list[Any] = [] while True: try: items.append(queue.get_nowait()) @@ -65,8 +65,8 @@ def _single_command_sequence( return (yield cmd) -class DistributorQueue(asyncio.Queue): - def __init__(self, parent: Optional[DistributorQueue] = None, *, maxsize=0): +class DistributorQueue(asyncio.Queue[Any]): + def __init__(self, parent: Optional[DistributorQueue] = None, *, maxsize: int = 0) -> None: """ A DistributorQueue provides a way of distributing some object to multiple waiting consumers. The base of a DistributorQueue structure @@ -107,7 +107,7 @@ def __init__(self, parent: Optional[DistributorQueue] = None, *, maxsize=0): def is_parent(self) -> bool: return self._parent is None - def add_handler(self, handler: DistributorQueue): + def add_handler(self, handler: DistributorQueue) -> None: if self.is_parent: self._handlers[hash(handler)] = handler else: @@ -116,16 +116,16 @@ def add_handler(self, handler: DistributorQueue): "Cannot add handler to DistributorQueue with a parent" ) - def del_handler(self, handler: DistributorQueue): + def del_handler(self, handler: DistributorQueue) -> None: self._handlers.pop(hash(handler), None) - def distribute(self, item: Any): + def distribute(self, item: Any) -> None: if self._parent is not None: self.put_nowait(item) for handler in self._handlers.values(): handler.distribute(item) - def __del__(self): + def __del__(self) -> None: if self._parent is not None: self._parent.del_handler(self) @@ -157,9 +157,7 @@ def __init__( "DriverSerialBase cannot be instantiated directly" ) - try: - uri.path - except AttributeError: + if isinstance(uri, str): uri = urlparse(uri) if uri.scheme != self.uri_scheme: @@ -168,19 +166,19 @@ def __init__( f"instance with a scheme of {uri.scheme}" ) - self.uri = uri - self.dev_inst_map: DeviceInstanceTypeMapper = dev_inst_map - if dev_inst_map is None: - self.dev_inst_map = DeviceInstanceTypeMapper() + self.uri: ParseResult = uri + self.dev_inst_map = ( + DeviceInstanceTypeMapper() if dev_inst_map is None else dev_inst_map + ) self._connected = asyncio.Event() # Assigned by the subclass once the asyncio transport protocol exists self._protocol: Any = None self.transaction_lock = asyncio.Lock() - def __repr__(self): + def __repr__(self) -> str: return f'{self.__class__.__name__}("{urlunparse(self.uri)}")' - def __hash__(self): + def __hash__(self) -> int: return hash(self.uri) async def connect(self, *, scan_dev_inst: bool = False) -> None: @@ -213,7 +211,7 @@ def is_connected(self) -> bool: return False if self._protocol is None: return True - return self._protocol.connected.is_set() + return bool(self._protocol.connected.is_set()) async def wait_connected(self) -> None: """ @@ -271,7 +269,7 @@ def new_dali_rx_queue(self) -> DistributorQueue: async def send( self, - msg: command.Command | Generator, + msg: command.Command | Generator[Any, Any, Any], *, priority: int = 5, progress: Optional[Callable[[str | sequences.progress], None]] = None, @@ -506,20 +504,20 @@ class LubaMsgTxConf(NamedTuple): def __init__(self) -> None: super().__init__() - self.transport = None + self.transport: Optional[serialx.SerialTransport] = None self._queue_rx_dali = DistributorQueue() - self._queue_rx_raw_dali = asyncio.Queue() - self._queue_rx_luba_cmd = asyncio.Queue() - self._queue_tx_conf = asyncio.Queue() + self._queue_rx_raw_dali: asyncio.Queue[Any] = asyncio.Queue() + self._queue_rx_luba_cmd: asyncio.Queue[Any] = asyncio.Queue() + self._queue_tx_conf: asyncio.Queue[Any] = asyncio.Queue() self._prev_rx_enable_dt = 0 self._prev_tx_enable_dt = 0 self._tx_lock = asyncio.Lock() - self._rx_state = None + self._rx_state = self.ReadState.WAIT_START self.rx_idle = asyncio.Event() - self._buffer = None - self._rx_expected_len = None - self._rx_received_len = None + self._buffer: list[int] = [] + self._rx_expected_len: Optional[int] = None + self._rx_received_len: int = 0 self._connected = asyncio.Event() self._dev_info: Optional[DriverLubaRs232.LubaDeviceInfo] = None self._dev_inst_map: Optional[DeviceInstanceTypeMapper] = None @@ -528,12 +526,20 @@ def __init__(self) -> None: self.reset() + @property + def _tx(self) -> serialx.SerialTransport: + """The connected serial transport, or raise if not connected.""" + if self.transport is None: + raise IOError("serial transport not connected") + return self.transport + @property def rx_state(self) -> ReadState: + assert self._rx_state is not None return self._rx_state @rx_state.setter - def rx_state(self, state: ReadState): + def rx_state(self, state: ReadState) -> None: if not isinstance(state, self.ReadState): raise TypeError( f"rx_state must be a ReadState enum, not {type(state)}" @@ -550,19 +556,19 @@ def dev_inst_map(self) -> Optional[DeviceInstanceTypeMapper]: return self._dev_inst_map @dev_inst_map.setter - def dev_inst_map(self, value: DeviceInstanceTypeMapper): + def dev_inst_map(self, value: DeviceInstanceTypeMapper) -> None: self._dev_inst_map = value @property def queue_rx_dali(self) -> DistributorQueue: return self._queue_rx_dali - def reset(self): + def reset(self) -> None: """ Returns the state machine to "WAIT_START" """ self.rx_state = self.ReadState.WAIT_START - self._buffer = [None] * self.MAX_LEN + self._buffer = [0] * self.MAX_LEN self._rx_expected_len = None self._rx_received_len = 0 @@ -573,7 +579,7 @@ async def wait_dali_raw_response(self) -> int: :return: A received DALI frame, as an int """ - return await self._queue_rx_raw_dali.get() + return cast(int, await self._queue_rx_raw_dali.get()) def reset_dali_response(self) -> None: """Drop any DALI responses left buffered from a previous exchange. @@ -605,7 +611,7 @@ def _insert_checksum(in_ints: list[int]) -> None: async def send_dali_command( self, tx: command.Command, priority: int = 1 - ) -> None: + ) -> int: """ Sends a variable length DALI command (16 or 24 bits), waiting until the LUBA device confirms it has sent the message before @@ -618,7 +624,7 @@ async def send_dali_command( # data before we send await self.rx_idle.wait() - dali_ints = tx.frame.as_byte_sequence + dali_ints = list(tx.frame.as_byte_sequence) if not len(dali_ints) in (2, 3): raise ValueError( f"Only works with 16 or 24 bit messages, not {8*len(dali_ints)}" @@ -636,7 +642,7 @@ async def send_dali_command( dali_ints[1], # Data, big-endian 0 if len(dali_ints) == 2 else dali_ints[2], # Data, big-endian 0, # Pad out to four bytes, to fit LUBA frame - None, # Checksum + 0, # Checksum ] # Fill in the checksum self._insert_checksum(tx_ints) @@ -648,7 +654,7 @@ async def send_dali_command( _LOG.trace( f"LUBA frame to send: {[f'0x{data:02x}' for data in tx_ints]}" ) - self.transport.write(bytearray(tx_ints)) + self._tx.write(bytearray(tx_ints)) # Wait for the LUBA device to respond, considering whether the # message is sent twice or not @@ -680,7 +686,7 @@ async def send_dali_command( confirm_count -= 1 # Release the lock - return confirm.tx_id + return cast(int, confirm.tx_id) async def send_device_info_query(self) -> None: """ @@ -694,7 +700,7 @@ async def send_device_info_query(self) -> None: DriverLubaRs232.LubaCmd.QUERY_DEVICE_INFO_CMD.value, # LUBA Command 1, # Length 0, # Request set "0" information - None, # Checksum + 0, # Checksum ] # Fill in the checksum self._insert_checksum(tx_ints) @@ -702,7 +708,7 @@ async def send_device_info_query(self) -> None: _LOG.trace( f"LUBA frame to send: {[f'0x{data:02x}' for data in tx_ints]}" ) - self.transport.write(bytearray(tx_ints)) + self._tx.write(bytearray(tx_ints)) # Wait for the LUBA device to respond dev_info = await asyncio.wait_for( @@ -729,14 +735,14 @@ async def send_device_descriptor_query( 0x59, # ASCII 'Y' DriverLubaRs232.LubaCmd.QUERY_DEVICE_DESCRIPTOR_CMD.value, 0, # Length, this command carries no data - None, # Checksum + 0, # Checksum ] self._insert_checksum(tx_ints) _LOG.trace( f"LUBA frame to send: {[f'0x{data:02x}' for data in tx_ints]}" ) - self.transport.write(bytearray(tx_ints)) + self._tx.write(bytearray(tx_ints)) descriptor = await asyncio.wait_for( self._queue_rx_luba_cmd.get(), @@ -768,14 +774,14 @@ async def send_read_status( DriverLubaRs232.LubaCmd.READ_STATUS_CMD.value, 1, # Length 0, # Line 0 (devices with a single DALI line) - None, # Checksum + 0, # Checksum ] self._insert_checksum(tx_ints) _LOG.trace( f"LUBA frame to send: {[f'0x{data:02x}' for data in tx_ints]}" ) - self.transport.write(bytearray(tx_ints)) + self._tx.write(bytearray(tx_ints)) line_status = await asyncio.wait_for( self._queue_rx_luba_cmd.get(), @@ -823,7 +829,7 @@ async def send_device_settings(self) -> None: mode_settings, event_settings, hardware_settings, - None, # Checksum + 0, # Checksum ] # Fill in the checksum self._insert_checksum(tx_ints) @@ -831,7 +837,7 @@ async def send_device_settings(self) -> None: _LOG.trace( f"LUBA frame to send: {[f'0x{data:02x}' for data in tx_ints]}" ) - self.transport.write(bytearray(tx_ints)) + self._tx.write(bytearray(tx_ints)) # Wait for the LUBA device to respond dev_settings = await asyncio.wait_for( @@ -984,7 +990,7 @@ def _process_byte(self, rx_int: int) -> None: else: raise RuntimeError(f"Invalid state: {self._rx_state}") - def _process_luba_event(self, received_data: tuple): + def _process_luba_event(self, received_data: Sequence[int]) -> None: """ Handle a LUBA 'event' message, typically these are received when a DALI frame was observed on the bus by the LUBA device @@ -1022,7 +1028,7 @@ def _process_luba_event(self, received_data: tuple): # Decode the command, for logging and debugging try: dali_command = command.Command.from_frame( - frame.Frame(bits=8 * len(tx_dali), data=tx_dali), + frame.ForwardFrame(bits=8 * len(tx_dali), data=tx_dali), devicetype=self._prev_tx_enable_dt, dev_inst_map=self._dev_inst_map, ) @@ -1090,7 +1096,7 @@ def _process_luba_event(self, received_data: tuple): else: # A 16 or 24-bit frame is an intercepted DALI command, # it can be deciphered into a Command object - dali_frame = frame.Frame( + dali_frame = frame.ForwardFrame( bits=8 * len(rx_dali), data=rx_dali ) try: @@ -1114,7 +1120,7 @@ def _process_luba_event(self, received_data: tuple): _LOG.debug(f"Adding DALI command to queue: {dali_command}") self._queue_rx_dali.distribute(dali_command) - def _process_luba_response_dali_frame_to_tx(self, received_data: tuple): + def _process_luba_response_dali_frame_to_tx(self, received_data: Sequence[int]) -> None: """ Handle a LUBA 'ADD DALI FRAME TO TX BUFFER' response message """ @@ -1149,7 +1155,7 @@ def _process_luba_response_dali_frame_to_tx(self, received_data: tuple): f"Invalid LUBA response length: {payload_length}" ) - def _process_luba_response_device_info(self, received_data: tuple): + def _process_luba_response_device_info(self, received_data: Sequence[int]) -> None: """ Handle a received "QUERY DEVICE INFO" response message """ @@ -1206,8 +1212,8 @@ def _process_luba_response_device_info(self, received_data: tuple): self._queue_rx_luba_cmd.put_nowait(info) def _process_luba_response_device_descriptor( - self, received_data: tuple - ): + self, received_data: Sequence[int] + ) -> None: """ Handle a received "QUERY DEVICE DESCRIPTOR" response message """ @@ -1234,7 +1240,7 @@ def _process_luba_response_device_descriptor( _LOG.info(f"Received device descriptor: {descriptor}") self._queue_rx_luba_cmd.put_nowait(descriptor) - def _process_luba_response_status(self, received_data: tuple): + def _process_luba_response_status(self, received_data: Sequence[int]) -> None: """ Handle a received "READ STATUS" response message """ @@ -1262,7 +1268,7 @@ def _process_luba_response_status(self, received_data: tuple): _LOG.info(f"Received line status: {line_status}") self._queue_rx_luba_cmd.put_nowait(line_status) - def _process_luba_response_settings(self, received_data: tuple): + def _process_luba_response_settings(self, received_data: Sequence[int]) -> None: """ Handle a received "READ / WRITE SETTINGS" response message """ @@ -1280,17 +1286,17 @@ def _process_luba_response_settings(self, received_data: tuple): _LOG.info(f"Received device settings: {settings}") self._queue_rx_luba_cmd.put_nowait(settings) - def connection_made(self, transport): - self.transport = transport + def connection_made(self, transport: asyncio.BaseTransport) -> None: + self.transport = cast(serialx.SerialTransport, transport) _LOG.info(f"Serial port opened: {transport}") self._connected.set() - def data_received(self, data): - _LOG.trace(f"Serial data received: {data}") + def data_received(self, data: bytes) -> None: + _LOG.trace(f"Serial data received: {data!r}") for rx in data: self._process_byte(rx) - def connection_lost(self, exc): + def connection_lost(self, exc: Optional[Exception]) -> None: _LOG.info("Serial port closed") self._connected.clear() @@ -1317,6 +1323,13 @@ def __init__( self._transport: Optional[serialx.SerialTransport] = None self._protocol: Optional[DriverLubaRs232.LubaProtocol] = None + @property + def _proto(self) -> DriverLubaRs232.LubaProtocol: + """The connected protocol, or raise if the driver is not connected.""" + if self._protocol is None: + raise IOError("DALI driver not connected") + return self._protocol + async def connect(self, *, scan_dev_inst: bool = False) -> None: if self.is_connected: _LOG.warning( @@ -1325,18 +1338,17 @@ async def connect(self, *, scan_dev_inst: bool = False) -> None: return # TODO: Add failure/retry handling - ( - self._transport, - self._protocol, - ) = await serialx.create_serial_connection( + transport, protocol = await serialx.create_serial_connection( loop=asyncio.get_event_loop(), protocol_factory=DriverLubaRs232.LubaProtocol, url=self.serial_path, baudrate=38400, ) + self._transport = transport + self._protocol = cast(DriverLubaRs232.LubaProtocol, protocol) try: await asyncio.wait_for( - self._protocol.connected.wait(), + self._proto.connected.wait(), timeout=DriverLubaRs232.timeout_connect, ) except asyncio.exceptions.TimeoutError as exc: @@ -1345,9 +1357,9 @@ async def connect(self, *, scan_dev_inst: bool = False) -> None: for attempt in range(1, DriverLubaRs232.handshake_attempts + 1): try: - await self._protocol.send_device_info_query() - self._protocol.bus_power = self.bus_power - await self._protocol.send_device_settings() + await self._proto.send_device_info_query() + self._proto.bus_power = self.bus_power + await self._proto.send_device_settings() break except (asyncio.TimeoutError, TimeoutError): if attempt >= DriverLubaRs232.handshake_attempts: @@ -1359,8 +1371,8 @@ async def connect(self, *, scan_dev_inst: bool = False) -> None: _LOG.warning( "LUBA handshake attempt %d timed out, retrying", attempt ) - self._protocol.reset_luba_response() - self._protocol.dev_inst_map = self.dev_inst_map + self._proto.reset_luba_response() + self._proto.dev_inst_map = self.dev_inst_map self._connected.set() @@ -1385,7 +1397,7 @@ async def _wait_for_bus_power(self) -> None: loop = asyncio.get_event_loop() deadline = loop.time() + DriverLubaRs232.timeout_bus_power while loop.time() < deadline: - status = await self._protocol.send_read_status() + status = await self._proto.send_read_status() if status is not None and not status.bus_voltage_error: return await asyncio.sleep(0.1) @@ -1397,13 +1409,13 @@ async def query_device_descriptor( """Return the interface's device descriptor (feature flags).""" if not self.is_connected: raise IOError("DALI driver cannot query, not connected") - return await self._protocol.send_device_descriptor_query() + return await self._proto.send_device_descriptor_query() async def read_status(self) -> Optional[DriverLubaRs232.LubaLineStatus]: """Return the status of the DALI line (e.g. bus power state).""" if not self.is_connected: raise IOError("DALI driver cannot query, not connected") - return await self._protocol.send_read_status() + return await self._proto.send_read_status() async def _send_frame( self, cmd: command.Command, *, priority: int = 1 @@ -1417,28 +1429,29 @@ async def _send_frame( # Make sure the received command buffer is empty, so that an # unexpected response can't accidentally be used - self._protocol.reset_dali_response() - await self._protocol.send_dali_command(cmd, priority=priority) - if cmd.is_query: + self._proto.reset_dali_response() + await self._proto.send_dali_command(cmd, priority=priority) + response_type = cmd.response + if response_type is not None: # Default to the command's own response type so that a query # which times out is reported as a missing value; a bare # Response(None) passes check_bad_rsp and leaks None to callers. - response = cmd.response(None) + response = response_type(None) while True: try: raw_rsp = await asyncio.wait_for( - self._protocol.wait_dali_raw_response(), + self._proto.wait_dali_raw_response(), timeout=DriverLubaRs232.timeout_rx, ) except asyncio.exceptions.TimeoutError: _LOG.debug(f"DALI response timeout, from message: {cmd}") break if raw_rsp is _DALI_FRAMING_ERROR: - response = cmd.response(frame.BackwardFrameError(255)) + response = response_type(frame.BackwardFrameError(255)) _LOG.debug(f"DALI framing error, from message: {cmd}") break if isinstance(raw_rsp, int): - response = cmd.response(frame.BackwardFrame(raw_rsp)) + response = response_type(frame.BackwardFrame(raw_rsp)) _LOG.debug(f"DALI response received: {raw_rsp}") break else: @@ -1452,7 +1465,7 @@ async def _send_frame( return response def new_dali_rx_queue(self) -> DistributorQueue: - return DistributorQueue(self._protocol.queue_rx_dali) + return DistributorQueue(self._proto.queue_rx_dali) @@ -1546,19 +1559,19 @@ class SCIRS232MsgTxConf(NamedTuple): def __init__(self) -> None: super().__init__() - self.transport = None + self.transport: Optional[serialx.SerialTransport] = None self._queue_rx_dali = DistributorQueue() - self._queue_rx_raw_dali = asyncio.Queue() - self._queue_rx_info = asyncio.Queue() + self._queue_rx_raw_dali: asyncio.Queue[Any] = asyncio.Queue() + self._queue_rx_info: asyncio.Queue[Any] = asyncio.Queue() self._prev_rx_enable_dt = 0 self._prev_tx_enable_dt = 0 self._tx_lock = asyncio.Lock() - self._rx_state = None + self._rx_state = self.ReadState.WAIT_STATUS self.rx_idle = asyncio.Event() - self._buffer = None - self._rx_expected_len = None - self._rx_received_len = None + self._buffer: list[int] = [] + self._rx_expected_len: Optional[int] = None + self._rx_received_len: int = 0 self._connected = asyncio.Event() self._dev_info: Optional[DriverSCIRS232.SCIRS232DeviceReply] = None self._dev_inst_map: Optional[DeviceInstanceTypeMapper] = None @@ -1569,12 +1582,20 @@ def __init__(self) -> None: self.reset() + @property + def _tx(self) -> serialx.SerialTransport: + """The connected serial transport, or raise if not connected.""" + if self.transport is None: + raise IOError("serial transport not connected") + return self.transport + @property def rx_state(self) -> ReadState: + assert self._rx_state is not None return self._rx_state @rx_state.setter - def rx_state(self, state: ReadState): + def rx_state(self, state: ReadState) -> None: if not isinstance(state, self.ReadState): raise TypeError( f"rx_state must be a ReadState enum, not {type(state)}" @@ -1591,19 +1612,19 @@ def dev_inst_map(self) -> Optional[DeviceInstanceTypeMapper]: return self._dev_inst_map @dev_inst_map.setter - def dev_inst_map(self, value: DeviceInstanceTypeMapper): + def dev_inst_map(self, value: DeviceInstanceTypeMapper) -> None: self._dev_inst_map = value @property def queue_rx_dali(self) -> DistributorQueue: return self._queue_rx_dali - def reset(self): + def reset(self) -> None: """ Returns the state machine to "WAIT_STATUS" """ self.rx_state = self.ReadState.WAIT_STATUS - self._buffer = [None] * self.MAX_LEN + self._buffer = [0] * self.MAX_LEN self._rx_expected_len = None self._rx_received_len = 0 @@ -1614,7 +1635,7 @@ async def wait_dali_raw_response(self) -> int: :return: A received DALI frame, as an int """ - return await self._queue_rx_raw_dali.get() + return cast(int, await self._queue_rx_raw_dali.get()) def reset_dali_response(self) -> None: """Drop any DALI frames left buffered from a previous exchange. @@ -1656,7 +1677,7 @@ async def send_dali_command(self, tx: command.Command) -> None: # data before we send await self.rx_idle.wait() - dali_ints = tx.frame.as_byte_sequence + dali_ints = list(tx.frame.as_byte_sequence) if not len(dali_ints) in (1, 2, 3): raise ValueError( f"Only works with 8, 16 or 24 bit messages, not {8*len(dali_ints)}" @@ -1675,7 +1696,7 @@ async def send_dali_command(self, tx: command.Command) -> None: dali_ints[0], 0 if len(dali_ints) < 2 else dali_ints[1], 0 if len(dali_ints) < 3 else dali_ints[2], - None, # Checksum + 0, # Checksum ] # Fill in the checksum self._insert_checksum(tx_ints) @@ -1687,7 +1708,7 @@ async def send_dali_command(self, tx: command.Command) -> None: _LOG.trace( f"SCI RS232 frame to send: {[f'0x{data:02x}' for data in tx_ints]}" ) - self.transport.write(bytearray(tx_ints)) + self._tx.write(bytearray(tx_ints)) confirm = await asyncio.wait_for( self._queue_rx_info.get(), @@ -1711,7 +1732,7 @@ async def send_device_info_query(self) -> None: 0, 0, 0, - None, # Checksum + 0, # Checksum ] # Fill in the checksum self._insert_checksum(tx_ints) @@ -1724,7 +1745,7 @@ async def send_device_info_query(self) -> None: _LOG.trace( f"SCI RS232 frame to send: {[f'0x{data:02x}' for data in tx_ints]}" ) - self.transport.write(bytearray(tx_ints)) + self._tx.write(bytearray(tx_ints)) # Wait for the SCI RS232 device to respond dev_info = await asyncio.wait_for( @@ -1833,12 +1854,12 @@ def _process_byte(self, rx_int: int) -> None: else: raise RuntimeError(f"Invalid state: {self._rx_state}") - def _process_system_message(self, data : int): + def _process_system_message(self, data: int) -> None: device_id_info = DriverSCIRS232.SCIRS232DeviceReply( id=(data & 0xf0) >> 4,code=data&0xf) self._queue_rx_info.put_nowait(device_id_info) - def _process_error(self, data : tuple): + def _process_error(self, data: Sequence[int]) -> None: try: error_type = DriverSCIRS232.SCIRS232Protocol.ErrorType(data[3]) except ValueError: @@ -1863,7 +1884,7 @@ def _process_error(self, data : tuple): _LOG.error(f"SCI RS232 reports {error_str} error ({error_type})") self._process_system_message(data[0]) - def _process_dali_frame(self, received_data: tuple): + def _process_dali_frame(self, received_data: Sequence[int]) -> None: """ Handle a DALI 'event' message, typically these are received when a DALI frame was observed on the bus by the SCI RS232 device @@ -1884,7 +1905,7 @@ def _process_dali_frame(self, received_data: tuple): else: # A 16 or 24-bit frame is an intercepted DALI command, # it can be deciphered into a Command object - dali_frame = frame.Frame( + dali_frame = frame.ForwardFrame( bits=8 * len(received_data), data=received_data ) try: @@ -1908,17 +1929,17 @@ def _process_dali_frame(self, received_data: tuple): _LOG.debug(f"Adding DALI command to queue: {dali_command}") self._queue_rx_dali.distribute(dali_command) - def connection_made(self, transport): - self.transport = transport + def connection_made(self, transport: asyncio.BaseTransport) -> None: + self.transport = cast(serialx.SerialTransport, transport) _LOG.info(f"Serial port opened: {transport}") self._connected.set() - def data_received(self, data): - _LOG.trace(f"Serial data received: {data}") + def data_received(self, data: bytes) -> None: + _LOG.trace(f"Serial data received: {data!r}") for rx in data: self._process_byte(rx) - def connection_lost(self, exc): + def connection_lost(self, exc: Optional[Exception]) -> None: _LOG.info("Serial port closed") self._connected.clear() @@ -1943,6 +1964,13 @@ def __init__( self._transport: Optional[serialx.SerialTransport] = None self._protocol: Optional[DriverSCIRS232.SCIRS232Protocol] = None + @property + def _proto(self) -> DriverSCIRS232.SCIRS232Protocol: + """The connected protocol, or raise if the driver is not connected.""" + if self._protocol is None: + raise IOError("DALI driver not connected") + return self._protocol + async def connect(self, *, scan_dev_inst: bool = False) -> None: if self.is_connected: _LOG.warning( @@ -1955,27 +1983,26 @@ async def connect(self, *, scan_dev_inst: bool = False) -> None: ) # TODO: Add failure/retry handling - ( - self._transport, - self._protocol, - ) = await serialx.create_serial_connection( + transport, protocol = await serialx.create_serial_connection( loop=asyncio.get_event_loop(), protocol_factory=DriverSCIRS232.SCIRS232Protocol, url=self.serial_path, baudrate=38400, ) + self._transport = transport + self._protocol = cast(DriverSCIRS232.SCIRS232Protocol, protocol) try: await asyncio.wait_for( - self._protocol._connected.wait(), + self._proto._connected.wait(), timeout=DriverSCIRS232.timeout_connect, ) except asyncio.exceptions.TimeoutError as exc: _LOG.critical(f"Timeout waiting for driver to connect: {exc}") raise - await self._protocol.send_device_info_query() - self._protocol.dev_inst_map = self.dev_inst_map + await self._proto.send_device_info_query() + self._proto.dev_inst_map = self.dev_inst_map self._connected.set() @@ -2003,25 +2030,26 @@ async def _send_frame( # Make sure the received command buffer is empty, so that an # unexpected response can't accidentally be used - self._protocol.reset_dali_response() - await self._protocol.send_dali_command(cmd) - if cmd.is_query: + self._proto.reset_dali_response() + await self._proto.send_dali_command(cmd) + response_type = cmd.response + if response_type is not None: # Default to the command's own response type so that a query # which times out is reported as a missing value; a bare # Response(None) passes check_bad_rsp and leaks None to callers. - response = cmd.response(None) + response = response_type(None) while True: try: raw_rsp = await asyncio.wait_for( - self._protocol._queue_rx_raw_dali.get(), - #self._protocol.wait_dali_raw_response(), + self._proto._queue_rx_raw_dali.get(), + #self._proto.wait_dali_raw_response(), timeout=DriverSCIRS232.timeout_rx, ) except asyncio.exceptions.TimeoutError: _LOG.debug(f"DALI response timeout, from message: {cmd}") break if isinstance(raw_rsp, int): - response = cmd.response(frame.BackwardFrame(raw_rsp)) + response = response_type(frame.BackwardFrame(raw_rsp)) _LOG.debug(f"DALI response received: {raw_rsp}") break else: @@ -2035,4 +2063,4 @@ async def _send_frame( return response def new_dali_rx_queue(self) -> DistributorQueue: - return DistributorQueue(self._protocol.queue_rx_dali) + return DistributorQueue(self._proto.queue_rx_dali) From 0a8edbb909042adf448d885462826250f2c92fa4 Mon Sep 17 00:00:00 2001 From: Markus Becker Date: Mon, 19 Oct 2026 09:56:05 +0200 Subject: [PATCH 3/3] Type the HID drivers for mypy --strict Annotate every function in the shared HID base and the tridonic and hasseb drivers, declare the instance attributes (file descriptor, reconnect/bus-watch tasks, outstanding-command map, response buffers) at class scope, and add _send_raw/_power_supply/_raw_write to the base so the shared send paths type-check. Narrow the optional file descriptor and decoded command before use, type the frame/response union locals, and fix a command._response typo to command.response. Typing only, no behaviour change. Co-Authored-By: Claude Opus 4.8 --- dali/driver/hid.py | 128 +++++++++++++++++++++++++++++---------------- 1 file changed, 83 insertions(+), 45 deletions(-) diff --git a/dali/driver/hid.py b/dali/driver/hid.py index 83a07a19..2efd4c4f 100644 --- a/dali/driver/hid.py +++ b/dali/driver/hid.py @@ -11,6 +11,7 @@ import random import glob import warnings +from typing import Any, Callable, Generator, Optional from dali.exceptions import UnsupportedFrameTypeError, CommunicationError from dali.sequences import sleep as seq_sleep from dali.sequences import progress as seq_progress @@ -21,38 +22,40 @@ import dali.gear from dali.gear.general import EnableDeviceType -def _hex(b): +def _hex(b: bytes) -> str: return ''.join("%02X" % x for x in b) -def _single_command_sequence(cmd): +def _single_command_sequence( + cmd: dali.command.Command, +) -> Generator[dali.command.Command, Optional[dali.command.Response], Optional[dali.command.Response]]: """Wrap a single command as a trivial sequence that returns its response.""" return (yield cmd) class _callback: """Helper class for callback registration """ - def __init__(self, parent): + def __init__(self, parent: Any) -> None: self._parent = parent - self._callbacks = {} + self._callbacks: dict[Any, Callable[..., Any]] = {} class _callback_handle: """Callback handle Call unregister() to remove this callback. """ - def __init__(self, callback): + def __init__(self, callback: "_callback") -> None: self._callback = callback - def unregister(self): + def unregister(self) -> None: del self._callback._callbacks[self] - def register(self, func): + def register(self, func: Callable[..., Any]) -> "_callback._callback_handle": wrapper = self._callback_handle(self) self._callbacks[wrapper] = func return wrapper - def _invoke(self, *args): + def _invoke(self, *args: Any) -> None: for func in self._callbacks.values(): asyncio.get_running_loop().call_soon(func, self._parent, *args) @@ -60,8 +63,14 @@ def _invoke(self, *args): class hid: """Shared code for drivers that work with HID devices """ - def __init__(self, path, reconnect_interval=1, reconnect_limit=None, - glob=False, dev_inst_map=None): + _f: Optional[int] + _reconnect_task: Optional[asyncio.Task[None]] + firmware_version: Optional[str] + serial: Optional[str] + + def __init__(self, path: str, reconnect_interval: float = 1, + reconnect_limit: Optional[int] = None, + glob: bool = False, dev_inst_map: Any = None) -> None: self._log = logging.getLogger() self._path = path self._reconnect_interval = reconnect_interval @@ -107,7 +116,7 @@ def __init__(self, path, reconnect_interval=1, reconnect_limit=None, self.firmware_version = None self.serial = None - def connect(self): + def connect(self) -> bool: """Attempt to connect to the device. Attempts to open the device. If this fails, schedules a @@ -152,7 +161,7 @@ def connect(self): self.connection_status_callback._invoke("connected") return True - async def _reconnect(self): + async def _reconnect(self) -> None: self._reconnect_count += 1 if self._reconnect_limit is not None \ and self._reconnect_count > self._reconnect_limit: @@ -165,7 +174,7 @@ async def _reconnect(self): self._reconnect_task = None self.connect() - def disconnect(self, reconnect=False): + def disconnect(self, reconnect: bool = False) -> None: self._log.debug("disconnecting") if self._reconnect_task: self._reconnect_task.cancel() @@ -180,7 +189,10 @@ def disconnect(self, reconnect=False): if reconnect: self._reconnect_task = asyncio.ensure_future(self._reconnect()) - async def send(self, msg, *, priority=5, progress=None, exceptions=None): + async def send(self, msg: dali.command.Command | Generator[Any, Any, Any], *, + priority: int = 5, + progress: Optional[Callable[..., Any]] = None, + exceptions: Optional[bool] = None) -> Any: """Send a command or run a sequence as a single transaction A bare Command is wrapped in a trivial sequence, so a single command @@ -246,7 +258,8 @@ async def send(self, msg, *, priority=5, progress=None, exceptions=None): self.transaction_lock.release() seq.close() - async def power_supply(self, supply_on, in_transaction=False, exceptions=None): + async def power_supply(self, supply_on: bool, in_transaction: bool = False, + exceptions: Optional[bool] = None) -> Any: """ TODO """ @@ -269,7 +282,8 @@ async def power_supply(self, supply_on, in_transaction=False, exceptions=None): if not in_transaction: self.transaction_lock.release() - async def run_sequence(self, seq, progress=None): + async def run_sequence(self, seq: Generator[Any, Any, Any], + progress: Optional[Callable[..., Any]] = None) -> Any: """Deprecated alias for send(), which now accepts commands and sequences """ warnings.warn( @@ -279,7 +293,7 @@ async def run_sequence(self, seq, progress=None): ) return await self.send(seq, progress=progress) - def _initialise_device(self): + def _initialise_device(self) -> None: """Send any device-specific initialisation commands """ # Some devices may need to send initialisation commands and await @@ -287,7 +301,7 @@ def _initialise_device(self): # they set self.connected once initialisation is complete self.connected.set() - def _shutdown_device(self): + def _shutdown_device(self) -> None: """Shut down everything that is waiting for the device The device has gone away. Everything that is waiting for the @@ -295,7 +309,8 @@ def _shutdown_device(self): """ pass - def _reader(self): + def _reader(self) -> None: + assert self._f is not None try: # No need to retry on InterruptedError since 3.5 data = os.read(self._f, 64) @@ -307,10 +322,26 @@ def _reader(self): return self._handle_read(data) - def _handle_read(self, data): + def _handle_read(self, data: bytes) -> None: pass + def _raw_write(self, buf: bytes) -> None: + assert self._f is not None + os.write(self._f, buf) + + async def _send_raw( + self, command: dali.command.Command + ) -> Optional[dali.command.Response]: + raise NotImplementedError + + async def _power_supply(self, supply_on: bool) -> Optional[dali.command.Response]: + raise NotImplementedError + class tridonic(hid): + _outstanding: dict[int, tuple[asyncio.Event, list[Any]]] + _bus_watch_data: list[bytes] + _bus_watch_task: Optional[asyncio.Task[None]] + # Commands sent to the interface # cmd, seq, ctrl, mode, frame (4 bytes), dtr, prio, devtype _cmdtmpl = struct.Struct(">4B4s3B53x") @@ -380,7 +411,7 @@ class tridonic(hid): _BUS_STATUS_DALI_MODE = 6 @staticmethod - def _seqnum(i): + def _seqnum(i: int) -> Generator[int, None, None]: """Sequence number generator """ while True: @@ -389,7 +420,7 @@ def _seqnum(i): if i > 0xff: i = 1 - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self._log = self._log.getChild("tridonic") @@ -409,16 +440,17 @@ def __init__(self, *args, **kwargs): self._bus_watch_data_available = asyncio.Event() self._bus_watch_data = [] - def _raw_write(self, buf: bytes): + def _raw_write(self, buf: bytes) -> None: self._log.debug(f"_raw_write: {_hex(buf[0:self._cmdtmpl_useful_size])}") + assert self._f is not None os.write(self._f, buf) - def _initialise_device(self): + def _initialise_device(self) -> None: # Read firmware version; pick up the reply in _handle_read self._raw_write(self._cmd( tridonic._CMD_INIT, tridonic._CMD_INIT_READVERSION)) - async def _power_supply(self, supply_on): + async def _power_supply(self, supply_on: bool) -> Optional[dali.command.Response]: await self.connected.wait() async with self._command_semaphore: data = self._cmd(self._CMD_POWERSUPPLY, @@ -432,9 +464,10 @@ async def _power_supply(self, supply_on): self._log.debug("fail on transmit, disconnecting") self.disconnect(reconnect=True) raise CommunicationError + return None @staticmethod - def _command_mode(frame): + def _command_mode(frame: dali.frame.Frame) -> int: if len(frame) == 8: return tridonic._SEND_MODE_DALI8 if len(frame) == 16: @@ -443,7 +476,7 @@ def _command_mode(frame): return tridonic._SEND_MODE_DALI24 raise UnsupportedFrameTypeError - async def _send_raw(self, command): + async def _send_raw(self, command: dali.command.Command) -> Optional[dali.command.Response]: frame = command.frame if len(frame) not in (16, 24): raise UnsupportedFrameTypeError @@ -452,7 +485,7 @@ async def _send_raw(self, command): seq = next(self._cmd_seq) self._log.debug(f"Sending with {seq=:02x}") event = asyncio.Event() - messages = [] + messages: list[Any] = [] # If seq is in self._outstanding this means we've wrapped # around the whole sequence number space with an event # still outstanding: clearly a bug! @@ -473,7 +506,7 @@ async def _send_raw(self, command): raise CommunicationError outstanding_transmissions = 2 if command.sendtwice else 1 - response = None + response: dali.frame.BackwardFrame | dali.frame.BackwardFrameError | str | None = None while outstanding_transmissions or response is None: self._log.debug(f"_send_raw waiting for {outstanding_transmissions=} " f"{response=!s}") @@ -515,11 +548,12 @@ async def _send_raw(self, command): del self._outstanding[seq], event, messages if command.response: # Construct response and return it - if response == "no": + if isinstance(response, str): return command.response(None) return command.response(response) + return None - async def _bus_watch(self): + async def _bus_watch(self) -> None: # Why is this a task, and not just run from _handle_read()? # It's so that when we see a forward frame on the bus that # didn't originate with us, we can apply a timeout after which @@ -557,6 +591,7 @@ async def _bus_watch(self): if origin not in (self._MODE_OBSERVE, self._MODE_RESPONSE): self._log.warning("bus_watch: unexpected packet mode, ignoring") continue + frame: dali.frame.ForwardFrame | dali.frame.BackwardFrame | dali.frame.BackwardFrameError | str if rtype == self._RESPONSE_FRAME_DALI16: frame = dali.frame.ForwardFrame(16, raw_frame) elif rtype == self._RESPONSE_FRAME_DALI24: @@ -612,20 +647,21 @@ async def _bus_watch(self): else: self._log.error("Unexpected response waiting for retransmit of config command, frame = %s", frame) elif current_command.response: + response_type = current_command.response # We are waiting for a response if timeout or frame == "no": # The response is "No". self._log.debug("Command %s response \'No\'", current_command) self.bus_traffic._invoke( - current_command, current_command.response(None), False) + current_command, response_type(None), False) current_command = None continue elif isinstance(frame, dali.frame.BackwardFrame): # There's a response self._log.debug("Command %s response %s", - current_command, current_command.response(frame)) + current_command, response_type(frame)) self.bus_traffic._invoke( - current_command, current_command.response(frame), False) + current_command, response_type(frame), False) current_command = None continue else: @@ -634,7 +670,7 @@ async def _bus_watch(self): self._log.debug("Command %s response \'No\' (on new frame)", current_command) self.bus_traffic._invoke( - current_command, current_command.response(None), False) + current_command, response_type(None), False) current_command = None # If we reach here, there is no current_command and there @@ -646,6 +682,7 @@ async def _bus_watch(self): command = dali.command.from_frame( frame, devicetype=devicetype, dev_inst_map=self.dev_inst_map) + assert command is not None devicetype = 0 if command.sendtwice or command.response: # We need more information. Stash the command and wait. @@ -661,7 +698,7 @@ async def _bus_watch(self): self._log.debug("Unexpected backward frame %s", frame) # self._log.debug("End of loop") - def _handle_read(self, data): + def _handle_read(self, data: bytes) -> None: self._log.debug("_handle_read %s", _hex(data[0:self._resptmpl_useful_size])) if data[0] == self._MODE_INFO: # Response to initialisation command @@ -704,7 +741,7 @@ def _handle_read(self, data): else: self._log.debug("Unknown response mode %x", data[0]) - def _shutdown_device(self): + def _shutdown_device(self) -> None: # All outstanding commands need to be woken up and told they've failed for event, messages in self._outstanding.values(): messages.append("fail") @@ -721,8 +758,9 @@ def _shutdown_device(self): self.serial = None @staticmethod - def _cmd(cmd, serial, ctrl=0, mode=0, frame=bytes(), dtr0=0, prio=0, - devtype=0): + def _cmd(cmd: int, serial: int, ctrl: int = 0, mode: int = 0, + frame: bytes = bytes(), dtr0: int = 0, prio: int = 0, + devtype: int = 0) -> bytes: """Return 64 bytes for the specified command """ return tridonic._cmdtmpl.pack( @@ -747,14 +785,14 @@ class hasseb(hid): _OK = 2 _INVALID_ANSWER = 3 - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self._log = self._log.getChild("hasseb") self._command_lock = asyncio.Lock() self._response_available = asyncio.Event() - self._response = None + self._response: Any = None - async def _send_raw(self, command): + async def _send_raw(self, command: dali.command.Command) -> Optional[dali.command.Response]: frame = command.frame if len(frame) != 16: raise UnsupportedFrameTypeError @@ -786,7 +824,7 @@ async def _send_raw(self, command): elif self._response[0] == self._OK: response = command.response(dali.frame.BackwardFrame(self._response[1])) elif self._response[0] == self._INVALID_ANSWER: - response = command._response(dali.frame.BackwardFrameError( + response = command.response(dali.frame.BackwardFrameError( self._response[1])) else: self._log.debug("Unknown response code %x", self._response[0]) @@ -794,7 +832,7 @@ async def _send_raw(self, command): self.bus_traffic._invoke(command, response, False) return response - def _handle_read(self, data): + def _handle_read(self, data: bytes) -> None: # Response should be two bytes. First byte is status, second # byte is optional response data. The hasseb appears to send # "NO DATA AVAILABLE" reports continuously when it is idle. @@ -802,7 +840,7 @@ def _handle_read(self, data): self._response = data self._response_available.set() - def _shutdown_device(self): + def _shutdown_device(self) -> None: # If there's a command in progress, tell it it's failed self._response = "fail" self._response_available.set()