diff --git a/dali/device/sequences.py b/dali/device/sequences.py index acf0c0ca..3dc5797e 100644 --- a/dali/device/sequences.py +++ b/dali/device/sequences.py @@ -6,14 +6,18 @@ import types from typing import Generator, Optional, Type -from dali.address import DeviceShort, InstanceNumber +from dali.address import DeviceBroadcast, DeviceShort, InstanceNumber from dali.command import Command, Response from dali.device.general import ( DTR0, DTR1, DTR2, + Compare, EventScheme, + Initialise, InstanceEventFilter, + ProgramShortAddress, + QueryDeviceStatus, QueryResolution, QueryInputValue, QueryInputValueLatch, @@ -22,11 +26,22 @@ QueryEventFilterM, QueryEventScheme, QueryEventSchemeResponse, + Randomise, + SearchAddrH, + SearchAddrL, + SearchAddrM, SetEventFilter, SetEventScheme, + SetShortAddress, + StartQuiescentMode, + StopQuiescentMode, + Terminate, + VerifyShortAddress, + Withdraw, ) from dali.device.helpers import check_bad_rsp -from dali.exceptions import DALISequenceError +from dali.exceptions import DALISequenceError, ProgramShortAddressFailure +from dali.sequences import progress, sleep def SetEventSchemes( @@ -275,3 +290,161 @@ def query_input_value( value >>= 8 - resolution return value + + +# INITIALISE data byte values for control devices, from IEC 62386-103:2014 +# Table 21. Unlike control gear (part 102), devices use 0xff for "all" and +# 0x7f for "no short address" (short address == MASK). +_INITIALISE_ALL = 0xFF +_INITIALISE_UNADDRESSED = 0x7F + + +def _find_next( + low: int, high: int +) -> Generator[Command, Optional[Response], Optional[int | str]]: + """Binary search the 24-bit random address space for a responding device. + + Returns the random address of the device with the lowest address in the + ``low..high`` range, ``"clash"`` if two devices share the current address, + or ``None`` if no device answers in the range. + """ + yield SearchAddrH((high >> 16) & 0xFF) + yield SearchAddrM((high >> 8) & 0xFF) + yield SearchAddrL(high & 0xFF) + + r = yield Compare() + + if low == high: + if r.value is True: + return "clash" if r.raw_value.error else low + return None + + if r.value is True: + midpoint = (low + high) // 2 + res = yield from _find_next(low, midpoint) + if res is not None: + return res + return (yield from _find_next(midpoint + 1, high)) + + return None + + +def Commissioning( + available_addresses: Optional[list[int]] = None, + readdress: bool = False, + dry_run: bool = False, +) -> Generator[Command, Optional[Response], list[DeviceShort]]: + """Assign short addresses to 24-bit (part 103) control devices. + + This mirrors :func:`dali.sequences.Commissioning` for control gear, using + the part 103 addressing commands. Use with an appropriate DALI driver + instance, through its ``run_sequence()`` method. + + The bus is placed in quiescent mode for the duration so that spontaneous + event messages do not collide with the addressing backward frames. + + :param available_addresses: If passed, only these short addresses will be + assigned; otherwise all 64 short addresses are considered available. + :param readdress: If set, all existing short addresses are cleared first + and every device is readdressed; otherwise only devices that are + currently unaddressed are assigned a short address. + :param dry_run: If set, no short addresses are actually programmed. Useful + for testing the search without altering the bus. + :return: The list of short addresses that are in use once the sequence + completes. + """ + used_addresses: list[DeviceShort] = [] + if available_addresses is None: + available_addresses = list(range(64)) + else: + available_addresses = list(available_addresses) + + if readdress: + if dry_run: + yield progress( + message="dry_run is set: not deleting existing short addresses" + ) + else: + yield DTR0(255) + yield SetShortAddress(DeviceBroadcast()) + else: + # Devices that already have a short address answer QueryDeviceStatus at + # that address, so those addresses are considered already in use. + for a in list(available_addresses): + in_use = yield QueryDeviceStatus(DeviceShort(a)) + if not check_bad_rsp(in_use): + available_addresses.remove(a) + used_addresses.append(DeviceShort(a)) + yield progress(message=f"Available addresses: {available_addresses}") + + yield Terminate() + yield StartQuiescentMode(DeviceBroadcast()) + yield Initialise(_INITIALISE_ALL if readdress else _INITIALISE_UNADDRESSED) + + # A verify failure must still leave the bus addressable, so the error is + # recorded and raised only after quiescent mode has been exited below; + # yielding during exception unwinding is unsafe because the sequence runner + # closes the generator via GeneratorExit on failure. + failed_address: Optional[int] = None + finished = False + # Loop to cope with multiple devices picking the same random search + # address; when that clash is detected we re-randomise and begin again. + # Devices that already received addresses are unaffected. + while not finished: + yield Randomise() + # Randomise can take up to 200ms to settle. + yield sleep(0.2) + + low = 0 + high = 0xFFFFFF + + while low is not None: + yield progress(completed=low, size=high) + low = yield from _find_next(low, high) + if low == "clash": + yield progress( + message="Multiple devices picked the same random " + "address; restarting" + ) + break + if low is None: + finished = True + break + yield progress(message=f"Device found at address {low:#x}") + if available_addresses: + new_addr = available_addresses.pop(0) + if dry_run: + yield progress( + message="Not programming short address " + f"{new_addr} because dry_run is set" + ) + else: + yield progress( + message=f"Programming short address {new_addr}" + ) + yield ProgramShortAddress(new_addr) + r = yield VerifyShortAddress(new_addr) + if r.value is not True: + failed_address = new_addr + finished = True + break + used_addresses.append(DeviceShort(new_addr)) + else: + yield progress( + message="Device found but no short addresses left" + ) + yield Withdraw() + if low < high: + low = low + 1 + else: + low = None + finished = True + + yield Terminate() + yield StopQuiescentMode(DeviceBroadcast()) + + if failed_address is not None: + raise ProgramShortAddressFailure(failed_address) + + yield progress(message="Addressing complete") + return used_addresses diff --git a/dali/tests/fakes.py b/dali/tests/fakes.py index 3885cccf..8bb317bd 100644 --- a/dali/tests/fakes.py +++ b/dali/tests/fakes.py @@ -488,15 +488,22 @@ def __init__( shortaddr: Optional[address.DeviceShort] = None, groups: Optional[Iterable[address.DeviceGroup]] = None, memory_banks: Optional[Iterable[Type[FakeMemoryBank]]] = (FakeDeviceBank0,), + random_preload: Optional[Iterable[int]] = None, ): # Store parameters self.shortaddr = shortaddr self.groups = set(groups) if groups else set() + self.random_preload = list(random_preload) if random_preload else [] # Configure internal variables self.dtr0: int = 0 self.dtr1: int = 0 self.dtr2: int = 0 self.enable_write_memory: bool = False + self.initialising: bool = False + self.withdrawn: bool = False + self.quiescent: bool = False + self.randomaddr = frame.Frame(24) + self.searchaddr = frame.Frame(24) self.memory_banks = {} for fake_bank in memory_banks: bank_number = fake_bank.bank.address @@ -504,6 +511,24 @@ def __init__( raise ValueError(f"Duplicate memory bank {bank_number}") self.memory_banks[bank_number] = fake_bank() + def _next_random_address(self) -> int: + if self.random_preload: + return self.random_preload.pop(0) + return random.randrange(0, 0x1000000) + + def _initialise_reacts(self, param: int) -> bool: + """Whether an INITIALISE with this data byte selects this device. + + Per IEC 62386-103:2014 Table 21: 0xff addresses all devices, 0x7f + addresses devices without a short address, and 0x00..0x3f addresses + the device with that short address. + """ + if param == 0xFF: + return True + if param == 0x7F: + return self.shortaddr is None + return self.shortaddr is not None and self.shortaddr.address == param + def valid_address(self, cmd: Command) -> bool: """Should we respond to this command?""" if len(cmd.frame) != 24: @@ -591,6 +616,47 @@ def send(self, cmd: Command) -> Optional[int]: return self._device_status elif isinstance(cmd, device.general.QueryNumberOfInstances): return len(self._instances) + elif isinstance(cmd, device.general.Terminate): + self.initialising = False + self.withdrawn = False + elif isinstance(cmd, device.general.Initialise): + if self._initialise_reacts(cmd.param): + self.initialising = True + self.withdrawn = False + elif isinstance(cmd, device.general.Randomise): + self.randomaddr = frame.Frame(24, self._next_random_address()) + elif isinstance(cmd, device.general.Compare): + if ( + self.initialising + and not self.withdrawn + and self.randomaddr.as_integer <= self.searchaddr.as_integer + ): + return _yes + elif isinstance(cmd, device.general.Withdraw): + if self.initialising and self.randomaddr == self.searchaddr: + self.withdrawn = True + elif isinstance(cmd, device.general.SearchAddrH): + self.searchaddr[23:16] = cmd.param + elif isinstance(cmd, device.general.SearchAddrM): + self.searchaddr[15:8] = cmd.param + elif isinstance(cmd, device.general.SearchAddrL): + self.searchaddr[7:0] = cmd.param + elif isinstance(cmd, device.general.ProgramShortAddress): + if self.initialising and self.randomaddr == self.searchaddr: + if cmd.param == 0xFF: + self.shortaddr = None + else: + self.shortaddr = address.DeviceShort(cmd.param) + elif isinstance(cmd, device.general.VerifyShortAddress): + if self.shortaddr is not None and self.shortaddr.address == cmd.param: + return _yes + elif isinstance(cmd, device.general.SetShortAddress): + if self.dtr0 == 0xFF: + self.shortaddr = None + elif isinstance(cmd, device.general.StartQuiescentMode): + self.quiescent = True + elif isinstance(cmd, device.general.StopQuiescentMode): + self.quiescent = False elif isinstance(cmd, device.general.EnableWriteMemory): self.enable_write_memory = True diff --git a/dali/tests/test_device_sequences.py b/dali/tests/test_device_sequences.py index 910a8e14..5e6c7773 100644 --- a/dali/tests/test_device_sequences.py +++ b/dali/tests/test_device_sequences.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import pytest from dali.address import DeviceShort, InstanceNumber @@ -19,6 +21,7 @@ from dali.device.helpers import DeviceInstanceTypeMapper, check_bad_rsp from dali.device.pushbutton import InstanceEventFilter as EventFilter_pb from dali.device.sequences import ( + Commissioning, QueryEventFilters, SetEventFilters, SetEventSchemes, @@ -363,3 +366,84 @@ def test_query_input_values_10bit(): except StopIteration as r: ret = r.value assert ret == 434 + + +def _short_addresses(devices: list[fakes.Device]) -> list[int]: + """Return the sorted short addresses of the devices that have one.""" + return sorted( + d.shortaddr.address for d in devices if d.shortaddr is not None + ) + + +def test_device_commissioning(): + devices = [fakes.Device() for _ in range(10)] + bus = fakes.Bus(devices) + + used = bus.run_sequence(Commissioning()) + + assert _short_addresses(devices) == list(range(10)) + assert sorted(a.address for a in used) == list(range(10)) + # Quiescent mode must be exited for every device once addressing completes. + assert all(not d.quiescent for d in devices) + + +def test_device_commissioning_adds_new_without_readdress(): + devices = [fakes.Device() for _ in range(10)] + bus = fakes.Bus(devices) + bus.run_sequence(Commissioning()) + + # Move one device out to address 30 and add 10 fresh unaddressed devices. + # bus.gear is the same list object, so appending once adds them to the bus. + devices[5].shortaddr = DeviceShort(30) + for _ in range(10): + devices.append(fakes.Device()) + + bus.run_sequence(Commissioning()) + + # Existing addresses are kept; new devices fill the lowest free addresses. + assert _short_addresses(devices) == list(range(19)) + [30] + + +def test_device_commissioning_readdress_dry_run_keeps_addresses(): + devices = [fakes.Device() for _ in range(10)] + bus = fakes.Bus(devices) + bus.run_sequence(Commissioning()) + before = _short_addresses(devices) + + bus.run_sequence(Commissioning(readdress=True, dry_run=True)) + + assert _short_addresses(devices) == before + + +def test_device_commissioning_readdress(): + devices = [fakes.Device(DeviceShort(i + 20)) for i in range(10)] + bus = fakes.Bus(devices) + + bus.run_sequence(Commissioning(readdress=True)) + + assert _short_addresses(devices) == list(range(10)) + + +def test_device_commissioning_clash(): + # At least one device picks the same random address as another on the + # first pass, forcing a re-randomise. + randoms = list(range(0, 0xFFFFFF, 0x82000)) + randoms[8] = randoms[4] + devices = [fakes.Device(random_preload=[x]) for x in randoms] + bus = fakes.Bus(devices) + + bus.run_sequence(Commissioning()) + + assert _short_addresses(devices) == list(range(len(devices))) + + +def test_device_commissioning_restricted_addresses(): + devices = [fakes.Device() for _ in range(10)] + bus = fakes.Bus(devices) + + bus.run_sequence(Commissioning(available_addresses=range(10, 15))) + + addressed = _short_addresses(devices) + assert all(10 <= a < 15 for a in addressed) + assert len(addressed) == 5 + assert sum(d.shortaddr is None for d in devices) == 5