Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 13 additions & 5 deletions dali/driver/serial.py
Original file line number Diff line number Diff line change
Expand Up @@ -1329,7 +1329,13 @@ def __init__(
self._transport: Optional[serialx.SerialTransport] = None
self._protocol: Optional[DriverLubaRs232.LubaProtocol] = None

async def connect(self, *, scan_dev_inst: bool = False) -> None:
async def connect(
self, *, scan_dev_inst: bool = False, apply_settings: bool = True
) -> None:
# ``apply_settings=False`` connects without writing the device settings,
# leaving the interface's hardware configuration (including its bus-power
# state) untouched. Use it to probe a discovered interface read-only so
# the probe cannot depower a bus the integrated supply is already driving.
if self.is_connected:
_LOG.warning(
f"'connect()' called but luba232 driver already connected"
Expand Down Expand Up @@ -1358,8 +1364,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()
if apply_settings:
self._protocol.bus_power = self.bus_power
await self._protocol.send_device_settings()
break
except (asyncio.TimeoutError, TimeoutError):
if attempt >= DriverLubaRs232.handshake_attempts:
Expand All @@ -1378,8 +1385,9 @@ async def connect(self, *, scan_dev_inst: bool = False) -> None:

# 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:
# the bus to be powered before scanning. Skipped when the settings were
# not applied, as the bus-power state was then left untouched.
if apply_settings and self.bus_power:
await self._wait_for_bus_power()

# Scan the bus for control devices, and create a mapping of addresses
Expand Down
19 changes: 19 additions & 0 deletions dali/tests/test_serial_driver.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,25 @@ async def test_connect_retries_handshake_on_timeout(monkeypatch):
assert protocol.reset_luba_response.call_count == 2


@pytest.mark.asyncio
async def test_connect_without_applying_settings_leaves_hardware_untouched(
monkeypatch,
):
"""connect(apply_settings=False) must not write the device settings.

A read-only probe of a discovered interface must not overwrite the
hardware configuration, otherwise it could depower a bus the integrated
supply is already driving.
"""
driver, protocol = _connecting_driver(monkeypatch, [None])

await driver.connect(apply_settings=False)

assert driver.is_connected
assert protocol.send_device_info_query.await_count == 1
assert protocol.send_device_settings.await_count == 0


@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."""
Expand Down
Loading