diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index c73cd871..af85f06c 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -8,7 +8,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ['3.8', '3.9', '3.10', '3.11', '3.12'] + python-version: ['3.10', '3.11', '3.12'] steps: - uses: actions/checkout@v3 diff --git a/dali/device/helpers.py b/dali/device/helpers.py index 4c250a7b..5f736448 100644 --- a/dali/device/helpers.py +++ b/dali/device/helpers.py @@ -19,7 +19,7 @@ ) from dali.exceptions import MissingResponse, ResponseError from dali.frame import BackwardFrameError -from dali.sequences import progress as seq_progress +from dali.sequences import progress as seq_progress, sleep def check_bad_rsp(r: Response | None) -> bool: @@ -85,7 +85,10 @@ def mapping(self) -> dict[(int, int), int]: return self._mapping def autodiscover( - self, addresses: int | tuple[int, int] | Iterable[int] = (0, 63) + self, + addresses: int | tuple[int, int] | Iterable[int] = (0, 63), + quiescent_settle: float = 0.2, + status_attempts: int = 3, ) -> Generator[Command, Response, None]: """ A generator sequence to scan a DALI bus for control device instances, @@ -99,6 +102,11 @@ def autodiscover( which case all addresses between the provided values will be scanned; or finally can be an iterable of ints in which case each address, in the iterator will be scanned. + :param quiescent_settle: Seconds to wait after entering quiescent mode + before probing, so an event frame already in flight does not corrupt + the first address probe. + :param status_attempts: How many times to probe an address whose status + reply collides with a spontaneous event before treating it as empty. :return: A generator function, to use with e.g. `driver.run_sequence()` Needs to be used through an appropriate driver, with `run_sequence()`, @@ -111,6 +119,12 @@ def autodiscover( # Use quiescent mode to reduce bus contention from input devices yield StartQuiescentMode(DeviceBroadcast()) + # Quiescent mode is not instant and an input device may already have an + # event frame in flight. Let the bus settle so the first probe is not + # corrupted by a colliding backward frame, which would otherwise make a + # present device look absent. + if quiescent_settle: + yield sleep(quiescent_settle) if isinstance(addresses, int): addresses = (n for n in range(0, addresses)) @@ -120,8 +134,16 @@ def autodiscover( for addr_int in addresses: addr = DeviceShort(addr_int) - # Check that the device exists and responds - rsp = yield QueryDeviceStatus(device=addr) + # Check that the device exists and responds. A framing error means a + # device answered but its backward frame collided with a spontaneous + # event, so retry; a missing response means the address is empty, so + # stop probing it immediately. + for _attempt in range(status_attempts): + rsp = yield QueryDeviceStatus(device=addr) + if not check_bad_rsp(rsp): + break + if not isinstance(getattr(rsp, "raw_value", None), BackwardFrameError): + break if check_bad_rsp(rsp): continue if isinstance(rsp, QueryDeviceStatusResponse): @@ -140,7 +162,13 @@ def autodiscover( rsp = yield QueryNumberOfInstances(device=addr) if check_bad_rsp(rsp): continue - num_inst = rsp.value + # Instance numbers are 0..31 (IEC 62386-103), so a device can have + # at most 32 instances. Cap the count defensively: a device + # reporting more, or a value corrupted by a colliding event frame, + # would otherwise build an out-of-range InstanceNumber and abort the + # whole scan. Non-existent instances are skipped when they do not + # respond to QueryInstanceEnabled below. + num_inst = min(rsp.value, 32) # For each instance, check it is enabled and then query the type for inst_int in range(num_inst): diff --git a/dali/driver/serial.py b/dali/driver/serial.py index 24ba70b2..0d54bdd5 100644 --- a/dali/driver/serial.py +++ b/dali/driver/serial.py @@ -29,16 +29,27 @@ from typing import Any, Callable, Generator, NamedTuple, Optional from urllib.parse import ParseResult, urlparse, urlunparse -import serial_asyncio +import serialx import dali.gear from dali import command, frame, gear, sequences from dali.driver import trace_logging # noqa: F401 from dali.device.helpers import DeviceInstanceTypeMapper +from dali.exceptions import TransmissionError _LOG = logging.getLogger("dali.driver") +def _drain_queue(queue: asyncio.Queue) -> list: + """Remove and return every currently-queued item without blocking.""" + items = [] + while True: + try: + items.append(queue.get_nowait()) + except asyncio.QueueEmpty: + return items + + class DistributorQueue(asyncio.Queue): def __init__(self, parent: Optional[DistributorQueue] = None, *, maxsize=0): """ @@ -147,6 +158,8 @@ def __init__( if dev_inst_map is None: self.dev_inst_map = DeviceInstanceTypeMapper() 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): @@ -173,9 +186,19 @@ def is_connected(self) -> bool: """ Flags whether the underlying transport is connected and ready for use + The serial device can disappear at any time, e.g. a USB interface + being unplugged. The driver's own event only records that the + handshake completed once, so for drivers backed by an asyncio + transport the protocol, which is told about the loss through + 'connection_lost()', has the last word. + :return: Boolean, true if connection is ready """ - return self._connected.is_set() + if not self._connected.is_set(): + return False + if self._protocol is None: + return True + return self._protocol.connected.is_set() async def wait_connected(self) -> None: """ @@ -301,6 +324,14 @@ class DriverLubaRs232(DriverSerialBase): timeout_rx = 0.025 timeout_tx_confirm = 1.0 # TX might take some time if the bus is busy timeout_connect = 1.0 + timeout_bus_power = 5.0 # Time for the integrated PS to bring the bus up + # The interface can be slow to acknowledge the initial handshake while the + # bus is busy (e.g. an input device streaming events), so retry it a few + # times rather than failing the whole connect on a single timeout. + handshake_attempts = 5 + + # Lunatone article numbers this driver has been tested against. + TESTED_ARTICLE_NUMBERS = frozenset({24166096, 24138246}) class LubaCmd(Enum): """ @@ -315,6 +346,8 @@ class LubaCmd(Enum): READ_STATUS_RSP = 0x2D QUERY_DEVICE_INFO_CMD = 0x20 QUERY_DEVICE_INFO_RSP = 0x21 + QUERY_DEVICE_DESCRIPTOR_CMD = 0x28 + QUERY_DEVICE_DESCRIPTOR_RSP = 0x29 EVENT_MESSAGE = 0x31 ADD_DALI_FRAME_TO_TX_CMD = 0x32 ADD_DALI_FRAME_TO_TX_RSP = 0x33 @@ -323,6 +356,21 @@ class LubaCmd(Enum): ADD_24DALI_FRAME_TO_TX_CMD = 0x36 ADD_24DALI_FRAME_TO_TX_RSP = 0x37 + class LubaTxError(Enum): + """Reason codes reported when the interface rejects a DALI frame. + + Refer ยง2.1.1 "ADD DALI FRAME TO TX BUFFER" of Lunatone's LUBA + documentation. + """ + + BUS_VOLTAGE_ERROR = 1 + DALI_INITIALIZE_MODE = 2 + DALI_QUIESCENT_MODE = 3 + SEND_BUFFER_FULL = 4 + LINE_NOT_AVAILABLE = 5 + SYNTAX_ERROR = 6 + MACRO_RUNNING = 7 + class LubaDeviceInfo(NamedTuple): """ Named tuple for storing a set of information about the LUBA device @@ -342,6 +390,26 @@ class LubaDeviceSettings(NamedTuple): mode: int event_filter: int + class LubaDeviceDescriptor(NamedTuple): + """ + Named tuple describing the functionality of the LUBA interface, as + returned by the "QUERY DEVICE DESCRIPTOR" command + """ + + # HardwareFeature bit 7: device has an integrated, switchable bus + # power supply + has_bus_power_supply: bool + + class LubaLineStatus(NamedTuple): + """ + Named tuple for the status of a DALI line, as returned by the + "READ STATUS" command + """ + + # Status bit 7: a bus voltage error is present (i.e. the bus is not + # powered) + bus_voltage_error: bool + class LubaProtocol(asyncio.Protocol): """ This class is internally used by DriverLubaRs232 to implement a state @@ -372,6 +440,7 @@ class LubaMsgTxConf(NamedTuple): tx_id: int message: Optional[command.Command] = None + error_code: Optional[int] = None def __init__(self) -> None: super().__init__() @@ -392,6 +461,8 @@ def __init__(self) -> None: self._connected = asyncio.Event() self._dev_info: Optional[DriverLubaRs232.LubaDeviceInfo] = None self._dev_inst_map: Optional[DeviceInstanceTypeMapper] = None + # Whether to switch on the interface's integrated bus power supply + self.bus_power = False self.reset() @@ -443,20 +514,28 @@ async def wait_dali_raw_response(self) -> int: return await self._queue_rx_raw_dali.get() def reset_dali_response(self) -> None: + """Drop any DALI responses left buffered from a previous exchange. + + A backward frame that arrives after its query's receive timeout has + no matching request, so it is cleared before the next transaction. + On a busy bus this is routine housekeeping, hence debug level. """ - Forces the queue of received DALI responses to be cleared, logging - any responses that are dropped if the queue is not empty - """ - qlen = self._queue_rx_raw_dali.qsize() - if qlen: - _LOG.critical( - f"LUBA RX DALI queue not empty! {qlen} items in queue!" + discarded = _drain_queue(self._queue_rx_raw_dali) + if discarded: + _LOG.debug( + "LUBA RX DALI queue: discarded %d stale response(s): %s", + len(discarded), + discarded, ) - try: - item = self._queue_rx_raw_dali.get_nowait() - _LOG.critical(f"LUBA RX DALI queue discarding: {item}") - except asyncio.QueueEmpty: - pass + + def reset_luba_response(self) -> None: + """Drop any LUBA command replies left buffered from a prior attempt. + + A reply that arrives after its command's confirm timeout would + otherwise be mistaken for the next command's reply, so it is cleared + before a handshake is retried. + """ + _drain_queue(self._queue_rx_luba_cmd) @staticmethod def _insert_checksum(in_ints: list[int]) -> None: @@ -525,6 +604,8 @@ async def send_dali_command(self, tx: command.Command) -> None: self._queue_tx_conf.get(), timeout=DriverLubaRs232.timeout_tx_confirm, ) + if confirm.error_code is not None: + raise TransmissionError(confirm.error_code) if hasattr(confirm.message, "frame"): if confirm.message.frame == tx.frame: _LOG.trace( @@ -538,7 +619,7 @@ async def send_dali_command(self, tx: command.Command) -> None: f"for '{confirm.message}'" ) else: - _LOG.warning( + _LOG.debug( f"Unable to decode message id {confirm.tx_id}, but " "LUBA device reports it was sent successfully" ) @@ -580,6 +661,81 @@ async def send_device_info_query(self) -> None: _LOG.error(f"Expected a LubaDeviceInfo, but got: {dev_info}") return + async def send_device_descriptor_query( + self, + ) -> Optional[DriverLubaRs232.LubaDeviceDescriptor]: + """ + Query the LUBA device descriptor, describing which features (such + as a switchable bus power supply) the interface supports + """ + # Use a mutex to ensure only one message is sent at a time + async with self._tx_lock: + _LOG.debug("Querying LUBA device descriptor") + tx_ints = [ + 0x59, # ASCII 'Y' + DriverLubaRs232.LubaCmd.QUERY_DEVICE_DESCRIPTOR_CMD.value, + 0, # Length, this command carries no data + None, # 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)) + + descriptor = await asyncio.wait_for( + self._queue_rx_luba_cmd.get(), + timeout=DriverLubaRs232.timeout_tx_confirm, + ) + # Release transmit mutex + + if not isinstance( + descriptor, DriverLubaRs232.LubaDeviceDescriptor + ): + _LOG.error( + f"Expected a LubaDeviceDescriptor, but got: {descriptor}" + ) + return None + return descriptor + + async def send_read_status( + self, + ) -> Optional[DriverLubaRs232.LubaLineStatus]: + """ + Read the status of the (single) DALI line, which reports whether + the bus is powered + """ + # Use a mutex to ensure only one message is sent at a time + async with self._tx_lock: + _LOG.debug("Reading LUBA line status") + tx_ints = [ + 0x59, # ASCII 'Y' + DriverLubaRs232.LubaCmd.READ_STATUS_CMD.value, + 1, # Length + 0, # Line 0 (devices with a single DALI line) + None, # 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)) + + line_status = await asyncio.wait_for( + self._queue_rx_luba_cmd.get(), + timeout=DriverLubaRs232.timeout_tx_confirm, + ) + # Release transmit mutex + + if not isinstance(line_status, DriverLubaRs232.LubaLineStatus): + _LOG.error( + f"Expected a LubaLineStatus, but got: {line_status}" + ) + return None + return line_status + async def send_device_settings(self) -> None: """ The implementation of the LUBA protocol assumes a certain @@ -603,7 +759,7 @@ async def send_device_settings(self) -> None: # 2: 1 = deactivates including the line number in events # 1: 1 = Deactivates events for macros # 0: reserved - hardware_settings = 0b00000000 + hardware_settings = 0b10000000 if self.bus_power else 0b00000000 # 7: 1 = turn on bus power supply # 6..0: Reserved tx_ints = [ @@ -749,6 +905,15 @@ def _process_byte(self, rx_int: int) -> None: rx_cmd == DriverLubaRs232.LubaCmd.QUERY_DEVICE_INFO_RSP ): self._process_luba_response_device_info(received_data) + elif ( + rx_cmd + == DriverLubaRs232.LubaCmd.QUERY_DEVICE_DESCRIPTOR_RSP + ): + self._process_luba_response_device_descriptor( + received_data + ) + elif rx_cmd == DriverLubaRs232.LubaCmd.READ_STATUS_RSP: + self._process_luba_response_status(received_data) elif ( rx_cmd == DriverLubaRs232.LubaCmd.READ_WRITE_SETTINGS_RSP @@ -807,7 +972,13 @@ def _process_luba_event(self, received_data: tuple): devicetype=self._prev_tx_enable_dt, dev_inst_map=self._dev_inst_map, ) - except: + except Exception: + _LOG.debug( + "Failed to decode transmitted DALI frame " + f"{[f'0x{b:02x}' for b in tx_dali]} " + f"(devicetype {self._prev_tx_enable_dt})", + exc_info=True, + ) dali_command = None # Store the last seen Device Type command, there will likely # be a subsequent command that we transmit which relies on @@ -904,6 +1075,13 @@ def _process_luba_response_dali_frame_to_tx(self, received_data: tuple): _LOG.error( f"LUBA device reports error in transmission: {error_code}" ) + # No DALI frame reaches the bus, so the "frame sent" event that + # normally confirms a transmission will never arrive. Surface + # the error to the waiting 'send_dali_command' so it fails fast + # instead of blocking until the confirmation timeout. + self._queue_tx_conf.put_nowait( + self.LubaMsgTxConf(tx_id=-1, error_code=error_code) + ) elif payload_length == 2: tx_id = received_data[3] _LOG.trace( @@ -959,14 +1137,74 @@ def _process_luba_response_device_info(self, received_data: tuple): ), ) _LOG.info(f"Received device info: {info}") - if info.article_num != 24166096: + if info.article_num not in DriverLubaRs232.TESTED_ARTICLE_NUMBERS: + tested = ", ".join( + str(num) for num in sorted(DriverLubaRs232.TESTED_ARTICLE_NUMBERS) + ) _LOG.warning( "DriverLubaRs232 has only been tested with Lunatone article " - f"nr. 24166096, not nr. {info.article_num}" + f"nr. {tested}, not nr. {info.article_num}" ) self._dev_info = info self._queue_rx_luba_cmd.put_nowait(info) + def _process_luba_response_device_descriptor( + self, received_data: tuple + ): + """ + Handle a received "QUERY DEVICE DESCRIPTOR" response message + """ + if ( + DriverLubaRs232.LubaCmd(self._buffer[1]) + != DriverLubaRs232.LubaCmd.QUERY_DEVICE_DESCRIPTOR_RSP + ): + raise ValueError( + f"Wrong event type 0x{self._buffer[1]:02x}, expected 0x29" + ) + + payload_length = received_data[2] + # The DeviceDescriptor is 19 bytes, of which HardwareFeature is the + # last one. Only its bit 7 (switchable bus power supply) is used. + if payload_length < 19: + raise ValueError( + f"Unexpected payload length {payload_length} for QUERY " + "DEVICE DESCRIPTOR response" + ) + hardware_feature = received_data[3 + 18] + descriptor = DriverLubaRs232.LubaDeviceDescriptor( + has_bus_power_supply=bool(hardware_feature & 0b10000000), + ) + _LOG.info(f"Received device descriptor: {descriptor}") + self._queue_rx_luba_cmd.put_nowait(descriptor) + + def _process_luba_response_status(self, received_data: tuple): + """ + Handle a received "READ STATUS" response message + """ + if ( + DriverLubaRs232.LubaCmd(self._buffer[1]) + != DriverLubaRs232.LubaCmd.READ_STATUS_RSP + ): + raise ValueError( + f"Wrong event type 0x{self._buffer[1]:02x}, expected 0x2D" + ) + + payload_length = received_data[2] + # Payload layout: Line, Tick (2), ID, NrOfEntries, Status. The + # Status byte (payload index 5) is only present for an available + # DALI line; a length of 1 means the requested line does not exist. + if payload_length < 6: + raise ValueError( + f"READ STATUS response without status byte (payload " + f"length {payload_length})" + ) + status = received_data[3 + 5] + line_status = DriverLubaRs232.LubaLineStatus( + bus_voltage_error=bool(status & 0b10000000), + ) + _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): """ Handle a received "READ / WRITE SETTINGS" response message @@ -997,7 +1235,7 @@ def data_received(self, data): def connection_lost(self, exc): _LOG.info("Serial port closed") - self.transport.loop.stop() + self._connected.clear() @property def connected(self) -> asyncio.Event: @@ -1011,12 +1249,15 @@ def __init__( self, uri: str | ParseResult, dev_inst_map: Optional[DeviceInstanceTypeMapper] = None, + bus_power: bool = False, ): super().__init__(uri=uri, dev_inst_map=dev_inst_map) self.serial_path = self.uri.path _LOG.info(f"Initialising luba232 driver for '{self.serial_path}'") - self._transport: Optional[serial_asyncio.SerialTransport] = None + # Whether to switch on the interface's integrated bus power supply + self.bus_power = bus_power + self._transport: Optional[serialx.SerialTransport] = None self._protocol: Optional[DriverLubaRs232.LubaProtocol] = None async def connect(self, *, scan_dev_inst: bool = False) -> None: @@ -1030,7 +1271,7 @@ async def connect(self, *, scan_dev_inst: bool = False) -> None: ( self._transport, self._protocol, - ) = await serial_asyncio.create_serial_connection( + ) = await serialx.create_serial_connection( loop=asyncio.get_event_loop(), protocol_factory=DriverLubaRs232.LubaProtocol, url=self.serial_path, @@ -1045,12 +1286,33 @@ async def connect(self, *, scan_dev_inst: bool = False) -> None: _LOG.critical(f"Timeout waiting for driver to connect: {exc}") raise - await self._protocol.send_device_info_query() - await self._protocol.send_device_settings() + 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() + break + except (asyncio.TimeoutError, TimeoutError): + if attempt >= DriverLubaRs232.handshake_attempts: + _LOG.error( + "LUBA handshake timed out after %d attempts", + DriverLubaRs232.handshake_attempts, + ) + raise + _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._connected.set() + # After switching on the integrated power supply the bus voltage takes + # a moment to rise; sending DALI frames before then fails, so wait for + # the bus to be powered before scanning. + if self.bus_power: + await self._wait_for_bus_power() + # Scan the bus for control devices, and create a mapping of addresses # to instance types if scan_dev_inst: @@ -1061,6 +1323,31 @@ async def connect(self, *, scan_dev_inst: bool = False) -> None: "device instances" ) + async def _wait_for_bus_power(self) -> None: + """Wait for the integrated power supply to bring up the bus voltage.""" + loop = asyncio.get_event_loop() + deadline = loop.time() + DriverLubaRs232.timeout_bus_power + while loop.time() < deadline: + status = await self._protocol.send_read_status() + if status is not None and not status.bus_voltage_error: + return + await asyncio.sleep(0.1) + _LOG.warning("DALI bus voltage not detected after enabling bus power") + + async def query_device_descriptor( + self, + ) -> Optional[DriverLubaRs232.LubaDeviceDescriptor]: + """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() + + 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() + async def send( self, msg: command.Command, in_transaction: bool = False ) -> Optional[command.Response]: @@ -1079,7 +1366,10 @@ async def send( self._protocol.reset_dali_response() await self._protocol.send_dali_command(msg) if msg.is_query: - response = command.Response(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 = msg.response(None) while True: try: raw_rsp = await asyncio.wait_for( @@ -1274,34 +1564,28 @@ async def wait_dali_raw_response(self) -> int: return await self._queue_rx_raw_dali.get() def reset_dali_response(self) -> None: - """ - Forces the queue of received DALI responses to be cleared, logging - any responses that are dropped if the queue is not empty - """ + """Drop any DALI frames left buffered from a previous exchange. - # remove backward frames - qlen = self._queue_rx_raw_dali.qsize() - if qlen: - _LOG.critical( - f"SCI RS232 RX DALI queue not empty! {qlen} items in queue!" + Backward frames and information frames that arrive after their + query's receive timeout have no matching request, so they are + cleared before the next transaction. On a busy bus this is routine + housekeeping, hence debug level. + """ + discarded = _drain_queue(self._queue_rx_raw_dali) + if discarded: + _LOG.debug( + "SCI RS232 RX DALI queue: discarded %d stale response(s): %s", + len(discarded), + discarded, ) - try: - item = self._queue_rx_raw_dali.get_nowait() - _LOG.critical(f"SCI RS232 RX DALI queue discarding: {item}") - except asyncio.QueueEmpty: - pass - - # remove information frames (includes errors and sent confirmations) - qlen = self._queue_rx_info.qsize() - if qlen: - _LOG.critical( - f"SCI RS232 RX info DALI queue not empty! {qlen} items in queue!" + + discarded_info = _drain_queue(self._queue_rx_info) + if discarded_info: + _LOG.debug( + "SCI RS232 RX info queue: discarded %d stale frame(s): %s", + len(discarded_info), + discarded_info, ) - try: - item = self._queue_rx_raw_dali.get_nowait() - _LOG.critical(f"SCI RS232 RX info DALI queue discarding: {item}") - except asyncio.QueueEmpty: - pass @staticmethod def _insert_checksum(in_ints: list[int]) -> None: @@ -1583,7 +1867,7 @@ def data_received(self, data): def connection_lost(self, exc): _LOG.info("Serial port closed") - self.transport.loop.stop() + self._connected.clear() @property def connected(self) -> asyncio.Event: @@ -1603,7 +1887,7 @@ def __init__( self.serial_path = self.uri.path _LOG.info(f"Initialising SCI RS232 driver for '{self.serial_path}'") - self._transport: Optional[serial_asyncio.SerialTransport] = None + self._transport: Optional[serialx.SerialTransport] = None self._protocol: Optional[DriverSCIRS232.SCIRS232Protocol] = None async def connect(self, *, scan_dev_inst: bool = False) -> None: @@ -1621,7 +1905,7 @@ async def connect(self, *, scan_dev_inst: bool = False) -> None: ( self._transport, self._protocol, - ) = await serial_asyncio.create_serial_connection( + ) = await serialx.create_serial_connection( loop=asyncio.get_event_loop(), protocol_factory=DriverSCIRS232.SCIRS232Protocol, url=self.serial_path, @@ -1670,7 +1954,10 @@ async def send( self._protocol.reset_dali_response() await self._protocol.send_dali_command(msg) if msg.is_query: - response = command.Response(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 = msg.response(None) while True: try: raw_rsp = await asyncio.wait_for( diff --git a/dali/exceptions.py b/dali/exceptions.py index 91dbbc9a..fd06e535 100644 --- a/dali/exceptions.py +++ b/dali/exceptions.py @@ -119,6 +119,19 @@ class CommunicationError(DriverError): """ +class TransmissionError(DriverError): + """The interface rejected transmission of a DALI frame. + + ``error_code`` is the interface-specific reason for the rejection. + """ + + def __init__(self, error_code): + self.error_code = error_code + super().__init__( + f"Frame transmission rejected, error code {error_code}" + ) + + class UnsupportedFrameTypeError(DriverError): """Device driver does not support this type of frame """ diff --git a/dali/tests/test_device_sequences.py b/dali/tests/test_device_sequences.py index 910a8e14..330eaf76 100644 --- a/dali/tests/test_device_sequences.py +++ b/dali/tests/test_device_sequences.py @@ -13,6 +13,7 @@ QueryEventSchemeResponse, QueryInputValue, QueryInputValueLatch, + QueryNumberOfInstances, QueryResolution, SetEventFilter, ) @@ -110,6 +111,83 @@ def test_device_autodiscover_skip_no_instances(fakes_bus): assert len(dev_inst_map.mapping) == 12 +def test_device_autodiscover_retries_collided_status(fakes_bus): + """A status probe that collides with an event is retried, not skipped. + + A spontaneous event from a device can corrupt the backward frame of the + first status probe; without a retry the whole device (and all its + instances) would be silently dropped from the scan. + """ + dev_inst_map = DeviceInstanceTypeMapper() + real_send = fakes_bus.send + collided = {"done": False} + + def send_with_one_collision(cmd): + if ( + not collided["done"] + and isinstance(cmd, QueryDeviceStatus) + and getattr(cmd.destination, "address", None) == 0 + ): + collided["done"] = True + return QueryDeviceStatusResponse(BackwardFrameError(0xFF)) + return real_send(cmd) + + fakes_bus.send = send_with_one_collision + fakes_bus.run_sequence(dev_inst_map.autodiscover()) + + assert collided["done"] + assert any(addr == 0 for addr, _instance in dev_inst_map.mapping) + assert len(dev_inst_map.mapping) == 12 + + +def test_device_autodiscover_absent_address_probed_once(fakes_bus): + """An empty address returns a missing (not collided) status, so it is not + retried; only a framing-error reply triggers the extra probes.""" + probes: list[int] = [] + real_send = fakes_bus.send + + def counting_send(cmd): + if isinstance(cmd, QueryDeviceStatus): + probes.append(getattr(cmd.destination, "address", None)) + return real_send(cmd) + + fakes_bus.send = counting_send + fakes_bus.run_sequence( + DeviceInstanceTypeMapper().autodiscover(addresses=(0, 5)) + ) + + # Addresses 3-5 have no device, so each is probed exactly once. + assert probes.count(3) == 1 + assert probes.count(4) == 1 + assert probes.count(5) == 1 + + +def test_device_autodiscover_caps_instance_count(fakes_bus): + """A device reporting more than 32 instances does not abort the scan. + + Instance numbers are 0..31, so a count above 32 (a non-conformant device, + or a value whose backward frame collided with an event) must not build an + out-of-range InstanceNumber and crash the whole scan; the count is capped + and the device's real instances are still enumerated. + """ + dev_inst_map = DeviceInstanceTypeMapper() + real_send = fakes_bus.send + + def send_with_bad_count(cmd): + if isinstance(cmd, QueryNumberOfInstances) and ( + getattr(cmd.destination, "address", None) == 0 + ): + return NumericResponse(BackwardFrame(255)) + return real_send(cmd) + + fakes_bus.send = send_with_bad_count + fakes_bus.run_sequence(dev_inst_map.autodiscover()) + + # Device 0's four real instances are still found despite the bogus count. + assert sum(1 for addr, _instance in dev_inst_map.mapping if addr == 0) == 4 + assert len(dev_inst_map.mapping) == 12 + + def test_device_query_event_scheme(fakes_bus): rsp = fakes_bus.send(QueryEventScheme(DeviceShort(1), InstanceNumber(1))) assert isinstance(rsp, QueryEventSchemeResponse) diff --git a/dali/tests/test_serial_driver.py b/dali/tests/test_serial_driver.py new file mode 100644 index 00000000..5559d5f3 --- /dev/null +++ b/dali/tests/test_serial_driver.py @@ -0,0 +1,159 @@ +import asyncio +import logging +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from dali.address import DeviceShort, InstanceNumber +from dali.command import NumericResponse +from dali.device.general import QueryInstanceType +from dali.device.helpers import check_bad_rsp +from dali.driver.serial import DriverLubaRs232 + + +@pytest.mark.asyncio +async def test_timed_out_query_reports_missing_value(): + """A query that times out must return the command's own response type. + + A bare Response(None) passes check_bad_rsp while its value is None, which + leaked into control-device autodiscovery and crashed on int(None). + """ + driver = DriverLubaRs232("luba232:/dev/ttyACM0") + driver._connected.set() + protocol = MagicMock() + protocol.reset_dali_response = MagicMock() + protocol.send_dali_command = AsyncMock() + protocol.wait_dali_raw_response = AsyncMock(side_effect=asyncio.TimeoutError) + driver._protocol = protocol + + response = await driver.send( + QueryInstanceType(device=DeviceShort(0), instance=InstanceNumber(0)) + ) + + assert isinstance(response, NumericResponse) + assert response.value == "(missing)" + assert check_bad_rsp(response) + + +@pytest.mark.asyncio +async def test_reset_dali_response_drains_all_without_critical(caplog): + """Stale buffered responses must all be cleared and not logged as CRITICAL. + + A late backward frame that missed its query is routine on a busy bus; the + old code discarded only one item and logged it at CRITICAL, spamming logs. + """ + protocol = DriverLubaRs232.LubaProtocol() + protocol._queue_rx_raw_dali.put_nowait(6) + protocol._queue_rx_raw_dali.put_nowait(7) + protocol._queue_rx_raw_dali.put_nowait(8) + + with caplog.at_level(logging.DEBUG, logger="dali.driver"): + protocol.reset_dali_response() + + assert protocol._queue_rx_raw_dali.empty() + assert not [r for r in caplog.records if r.levelno >= logging.CRITICAL] + assert any("discarded 3 stale" in r.getMessage() for r in caplog.records) + + +def _connecting_driver(monkeypatch, info_side_effect): + """Return a driver whose transport is faked and handshake is scripted.""" + driver = DriverLubaRs232("luba232:/dev/ttyACM0") + protocol = MagicMock() + connected = asyncio.Event() + connected.set() + protocol.connected = connected + protocol.send_device_info_query = AsyncMock(side_effect=info_side_effect) + protocol.send_device_settings = AsyncMock() + protocol.reset_luba_response = MagicMock() + monkeypatch.setattr( + "dali.driver.serial.serialx.create_serial_connection", + AsyncMock(return_value=(MagicMock(), protocol)), + ) + return driver, protocol + + +@pytest.mark.asyncio +async def test_connect_retries_handshake_on_timeout(monkeypatch): + """A busy bus can time out the first handshake frames; connect must retry. + + The interface is slow to acknowledge the info/settings exchange while a + device streams events, so a single timeout should not fail the connect. + """ + driver, protocol = _connecting_driver( + monkeypatch, [asyncio.TimeoutError, asyncio.TimeoutError, None] + ) + + await driver.connect() + + assert driver.is_connected + assert protocol.send_device_info_query.await_count == 3 + assert protocol.send_device_settings.await_count == 1 + assert protocol.reset_luba_response.call_count == 2 + + +@pytest.mark.asyncio +async def test_connect_raises_after_exhausting_handshake_attempts(monkeypatch): + """When every handshake attempt times out, connect must give up and raise.""" + driver, protocol = _connecting_driver(monkeypatch, asyncio.TimeoutError) + + with pytest.raises(asyncio.TimeoutError): + await driver.connect() + + assert not driver.is_connected + attempts = DriverLubaRs232.handshake_attempts + assert protocol.send_device_info_query.await_count == attempts + assert protocol.reset_luba_response.call_count == attempts - 1 + + +@pytest.mark.asyncio +async def test_lost_transport_is_reported_as_disconnected(monkeypatch): + """An interface that goes away must stop claiming to be connected. + + The driver's own event only records that the handshake once succeeded, so + an unplugged USB interface still looked connected and every frame ran into + its transmit timeout instead of failing fast. + """ + driver, protocol = _connecting_driver(monkeypatch, [None]) + await driver.connect() + assert driver.is_connected + + # What LubaProtocol.connection_lost() does when the port disappears + protocol.connected.clear() + + assert not driver.is_connected + + +@pytest.mark.asyncio +async def test_connect_reopens_the_port_after_a_lost_transport(monkeypatch): + """Once the interface is back, connect() must open the port again.""" + driver, protocol = _connecting_driver(monkeypatch, [None]) + await driver.connect() + protocol.connected.clear() + + replacement = MagicMock() + connected = asyncio.Event() + connected.set() + replacement.connected = connected + replacement.send_device_info_query = AsyncMock() + replacement.send_device_settings = AsyncMock() + monkeypatch.setattr( + "dali.driver.serial.serialx.create_serial_connection", + AsyncMock(return_value=(MagicMock(), replacement)), + ) + + await driver.connect() + + assert driver.is_connected + assert driver._protocol is replacement + + +@pytest.mark.asyncio +async def test_reset_luba_response_drains_all(caplog): + """Stale LUBA command replies must be cleared before a handshake retry.""" + protocol = DriverLubaRs232.LubaProtocol() + protocol._queue_rx_luba_cmd.put_nowait("stale-1") + protocol._queue_rx_luba_cmd.put_nowait("stale-2") + + protocol.reset_luba_response() + + assert protocol._queue_rx_luba_cmd.empty() diff --git a/setup.cfg b/setup.cfg index e2d4b4eb..f0511cb4 100644 --- a/setup.cfg +++ b/setup.cfg @@ -20,7 +20,7 @@ classifiers = Programming Language :: Python :: 3 [options] -python_requires = >=3.7 +python_requires = >=3.10 packages = dali dali.device @@ -33,11 +33,11 @@ packages = driver-unipi = pyusb pymodbus -driver-serial = pyserial-asyncio +driver-serial = serialx>=1.11.0 test = pytest pytest-asyncio - pyserial-asyncio + serialx>=1.11.0 [flake8] exclude = */driver/