diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..f093f28 --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,26 @@ +name: tests + +on: + push: + branches: [reliability, master] + pull_request: + +jobs: + pytest: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.11", "3.12"] + steps: + - uses: actions/checkout@v4 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r tests/requirements.txt + pip install -e . + - name: Run tests + run: pytest -q diff --git a/.gitignore b/.gitignore index 16faff7..cee2c73 100644 --- a/.gitignore +++ b/.gitignore @@ -1,10 +1,14 @@ /* /*/ +!.gitignore !gitignore !README.md !LICENSE !setup.py !setup.cfg !pyplejd/ +!tests/ +!pytest.ini +!.github/ **/__pycache__/ -!compiling.md \ No newline at end of file +!compiling.md diff --git a/pyplejd/__init__.py b/pyplejd/__init__.py index 75763d9..1417e22 100644 --- a/pyplejd/__init__.py +++ b/pyplejd/__init__.py @@ -9,7 +9,7 @@ from .ble.debug import rec_log from .cloud import PlejdCloudSite -from .errors import AuthenticationError, ConnectionError +from .errors import AuthenticationError, ConnectionError, PlejdWriteError from .interface import ( outputDeviceClass, inputDeviceClass, @@ -24,6 +24,7 @@ "DeviceTypes", "AuthenticationError", "ConnectionError", + "PlejdWriteError", "PLEJD_SERVICE", ] @@ -35,14 +36,20 @@ class PlejdManager: - def __init__(self, username: str, password: str, siteId: str): + def __init__( + self, + username: str, + password: str, + siteId: str, + connect_workaround: bool = False, + ): self.credentials = { "username": username, "password": password, "siteId": siteId, } - self.mesh = PlejdMesh(self) + self.mesh = PlejdMesh(self, connect_workaround=connect_workaround) self.devices: list[dt.PlejdDevice | dt.PlejdScene] = [] self.hardware: dict[str, dt.PlejdHardware] = {} self._blacklist = set() # TODO: MAKE WORK @@ -166,6 +173,41 @@ async def broadcast_time(self): await self.mesh.broadcast_time() return + @property + def groups(self) -> list[dict]: + """EXPERIMENTAL: list of {roomId, title, address} room/group targets.""" + return list(self.cloud.groups) + + async def dim_group(self, address: int, dim: int | None = None): + """EXPERIMENTAL: actuate a whole room/group with a single mesh write. + + `address` is a room/group mesh address (see `groups`). With dim=None + the group is switched on; dim<=0 switches it off; otherwise the group + is set to the given brightness (0-255). The command format is the same + group output command used for individual outputs, addressed to the + group instead. This is untested against live hardware. + """ + if dim is None: + cmd = LastData( + address=address, + command=LastData.CMD_GROUP_OUTPUT_STATE, + payload=[0x1], + ) + elif dim <= 0: + cmd = LastData( + address=address, + command=LastData.CMD_GROUP_OUTPUT_STATE, + payload=[0x0], + ) + else: + dim = int(dim) + cmd = LastData( + address=address, + command=LastData.CMD_GROUP_OUTPUT_STATE_AND_LEVEL, + payload=[0x1, dim, dim], + ) + await self.mesh.write(cmd.hex) + async def disconnect(self): await self.mesh.disconnect() diff --git a/pyplejd/ble/__init__.py b/pyplejd/ble/__init__.py index 3b15e85..e716340 100644 --- a/pyplejd/ble/__init__.py +++ b/pyplejd/ble/__init__.py @@ -10,6 +10,7 @@ from bleak.backends.device import BLEDevice from bleak_retry_connector import establish_connection, BleakClientWithServiceCache +from ..errors import PlejdWriteError from .crypto import auth_response, encrypt_decrypt from . import ble_characteristics as gatt from . import payload_encode @@ -30,18 +31,46 @@ class MeshDevice: bleDevice: BLEDevice = None is_gateway: bool = False - def see(self, rssi, bleDevice: BLEDevice) -> bool: + # RSSI is decayed toward this floor between sightings so that a gateway + # candidate that has gone quiet loses its ranking instead of holding a + # historical-max value forever. + RSSI_FLOOR = -100 + RSSI_HALFLIFE_S = 300.0 # 5 minutes + + # Monotonic timestamp (seconds) of the last sighting, used solely for RSSI + # decay math. Kept separate from `last_seen` (a wall-clock datetime that is + # surfaced to users via the last-seen sensor) so that an NTP step does not + # over- or under-decay the stored RSSI. + _rssi_ts: float = None + + def current_rssi(self, now: float = None) -> float | None: + """RSSI as of monotonic time `now`, decayed toward RSSI_FLOOR.""" + if self.rssi is None or self._rssi_ts is None: + return self.rssi + now = time.monotonic() if now is None else now + elapsed = max(0.0, now - self._rssi_ts) + factor = 0.5 ** (elapsed / self.RSSI_HALFLIFE_S) + return self.RSSI_FLOOR + (self.rssi - self.RSSI_FLOOR) * factor + + def see(self, rssi, bleDevice: BLEDevice, now: float = None) -> bool: # Returns true if first seen - if first_seen := (self.rssi is None): - self.bleDevice = bleDevice - self.rssi = rssi - + now = time.monotonic() if now is None else now + first_seen = self.rssi is None + + # Always refresh the BLEDevice handle - the connectable proxy, MAC + # details or advertisement backing it may have changed since we last + # saw this node. Keeping the first-ever handle led to connecting + # through a proxy that is no longer the best (or is gone). + self.bleDevice = bleDevice + + decayed = self.current_rssi(now) + self.rssi = rssi if decayed is None else max(decayed, rssi) + self._rssi_ts = now self.last_seen = datetime.now() - self.rssi = max(self.rssi, rssi) return first_seen - def update(): + def update(self): pass @@ -50,18 +79,31 @@ def normalize_address(addr: str) -> str: class PlejdMesh: - def __init__(self, manager): + def __init__(self, manager, connect_workaround: bool = False): self.manager = manager self._mesh_devices: dict[str, MeshDevice] = {} self._gateway_node: MeshDevice | None = None self._crypto_key: bytearray = None self._client: BleakClient = None + # When True, always perform the connect->disconnect->sleep(5)->connect + # "BT proxy workaround" up front. When False (default) we try a direct + # connect+authenticate first and only fall back to the workaround if + # that first authentication attempt fails. See connect(). + self._connect_workaround = connect_workaround + self._ble_lock = asyncio.Lock() + # Serialises connect attempts so that a burst of failed writes (each + # trying to reconnect) results in a single establish_connection storm, + # not N concurrent ones. + self._connect_lock = asyncio.Lock() @property def connected(self): - return self._client is not None + # A stored client handle is not enough - the underlying link may have + # dropped without the disconnect callback firing (see hass_plejd #162, + # #125, #147). Require the client to actually report a live connection. + return self._client is not None and self._client.is_connected def expect_device(self, node: MeshDevice = None): self._mesh_devices[node.BLEaddress] = node @@ -77,7 +119,10 @@ def set_key(self, key: str): self._crypto_key = key async def disconnect(self): - if not self.connected: + # Use the raw handle rather than `self.connected` so that a stale + # client (handle set but link already dropped) is still torn down and + # cleared, instead of being leaked. + if self._client is None: return False try: @@ -91,8 +136,16 @@ async def disconnect(self): self.manager.connect_callback(False) async def connect(self): + # Single-flight: if a connect is already in progress, wait for it and + # reuse its result instead of launching a competing establish_connection. if self.connected: return True + async with self._connect_lock: + if self.connected: + return True + return await self._connect() + + async def _connect(self): _CONNECTION_LOG.debug("Trying to connect to BLE mesh") def _disconnect(client: BleakClient): @@ -104,12 +157,16 @@ def _disconnect(client: BleakClient): self._gateway_node = None self.manager.connect_callback(False) - # Try to connect to nodes in order of decreasing RSSI + # Try to connect to nodes in order of decreasing (time-decayed) RSSI + # so a node that has recently gone quiet is deprioritised. + now = time.monotonic() filtered_nodes = filter( - lambda n: n.connectable and n.rssi is not None, + lambda n: n.connectable and n.current_rssi(now) is not None, self._mesh_devices.values(), ) - sorted_nodes = sorted(filtered_nodes, key=lambda n: n.rssi, reverse=True) + sorted_nodes = sorted( + filtered_nodes, key=lambda n: n.current_rssi(now), reverse=True + ) if not sorted_nodes: return False @@ -117,29 +174,45 @@ def _disconnect(client: BleakClient): for node in sorted_nodes: try: _CONNECTION_LOG.debug("Attempting to connect to %s", node) + # Pass the disconnect callback to establish_connection; bleak + # 3.x has no BleakClient.set_disconnected_callback (removed in + # ~0.19), so it must be supplied at connection time. client = await establish_connection( BleakClientWithServiceCache, node.bleDevice, node.bleDevice.name, + _disconnect, max_attempts=2, ) - # Workaround for problem in plejd firmware 2026-05-20 - # Disconnect and connect again - _CONNECTION_LOG.debug( - "BT Proxy workaround - Disconnecting for 5 seconds." - ) - await client.disconnect() - await asyncio.sleep(5) - _CONNECTION_LOG.debug("BT Proxy workaround - Reconnecting") - client = await establish_connection( - BleakClientWithServiceCache, - node.bleDevice, - node.bleDevice.name, - _disconnect, - ) + authenticated = False + if not self._connect_workaround: + # Fast path: try to authenticate on the first connection. + authenticated = await self._authenticate(client) + if not authenticated: + _CONNECTION_LOG.debug( + "Direct authentication failed; " + "falling back to BT Proxy workaround" + ) + + if not authenticated: + # Workaround for problem in plejd firmware 2026-05-20: + # disconnect, wait, and connect again before authenticating. + _CONNECTION_LOG.debug( + "BT Proxy workaround - Disconnecting for 5 seconds." + ) + await client.disconnect() + await asyncio.sleep(5) + _CONNECTION_LOG.debug("BT Proxy workaround - Reconnecting") + client = await establish_connection( + BleakClientWithServiceCache, + node.bleDevice, + node.bleDevice.name, + _disconnect, + ) + authenticated = await self._authenticate(client) - if not await self._authenticate(client): + if not authenticated: await client.disconnect() continue self._gateway_node = node @@ -198,7 +271,11 @@ async def poll(self): ) async def poll_buttons(self): - await self.write(LastData(command=LastData.CMD_EVENT_PREPARE).hex) + # Best-effort maintenance traffic - don't propagate write failures. + try: + await self.write(LastData(command=LastData.CMD_EVENT_PREPARE).hex) + except PlejdWriteError as e: + _LOGGER.debug("poll_buttons write failed: %s", e) async def ping(self): async with self._ble_lock: @@ -216,7 +293,11 @@ async def poll_time(self, address: int): return payloads = payload_encode.request_time(self, address) - await self.write(payloads) + try: + await self.write(payloads) + except PlejdWriteError as e: + _LOGGER.debug("poll_time write failed: %s", e) + return retval = await self._client.read_gatt_char(gatt.PLEJD_LASTDATA) data = encrypt_decrypt(self._crypto_key, self._gateway_node.BLEaddress, retval) @@ -231,13 +312,48 @@ async def poll_time(self, address: int): async def broadcast_time(self): payloads = payload_encode.set_time(self) - await self.write(payloads) + try: + await self.write(payloads) + except PlejdWriteError as e: + _LOGGER.debug("broadcast_time write failed: %s", e) async def write(self, *payloads: list[str]): + """Write one or more hex command payloads to the mesh. + + Raises PlejdWriteError if the command could not be delivered, even + after a single quick reconnect-and-retry. Callers that treat writes + as best-effort maintenance traffic should catch PlejdWriteError. + """ + _LOGGER.debug(f"Write: {payloads}") + await self._write(payloads) + + async def _write(self, payloads): + # Raw hex payloads are kept unencrypted until the moment of writing so + # that a retry after reconnecting re-encrypts against the (possibly + # new) gateway address. + try: + await self._write_once(payloads) + return True + except (PlejdWriteError, BleakError, asyncio.TimeoutError) as e: + _LOGGER.warning( + "Writing to plejd mesh failed (%s); reconnecting and retrying once", + str(e), + ) + + try: + await self.connect() + await self._write_once(payloads) + return True + except (PlejdWriteError, BleakError, asyncio.TimeoutError) as e: + raise PlejdWriteError( + f"Failed to write to plejd mesh after retry: {e}" + ) from e + + async def _write_once(self, payloads): if not self.connected: - return + raise PlejdWriteError("Not connected to plejd mesh") - pl = [ + encrypted = [ encrypt_decrypt( self._crypto_key, self._gateway_node.BLEaddress, @@ -245,24 +361,12 @@ async def write(self, *payloads: list[str]): ) for payload in payloads ] - _LOGGER.debug(f"Write: {payloads}") - await self._write(pl) - - async def _write(self, payloads): - if not self.connected: - return - - try: - async with self._ble_lock: - for payload in payloads: - _LOGGER.debug("Writing to plejd mesh: %s", payload.hex()) - await self._client.write_gatt_char( - gatt.PLEJD_DATA, payload, response=True - ) - except (BleakError, asyncio.TimeoutError) as e: - _LOGGER.warning("Writing to plejd mesh failed: %s", str(e)) - return False - return True + async with self._ble_lock: + for payload in encrypted: + _LOGGER.debug("Writing to plejd mesh: %s", payload.hex()) + await self._client.write_gatt_char( + gatt.PLEJD_DATA, payload, response=True + ) async def _ping(self, client): if client is None: diff --git a/pyplejd/cloud/__init__.py b/pyplejd/cloud/__init__.py index 4d35815..49bc3b5 100644 --- a/pyplejd/cloud/__init__.py +++ b/pyplejd/cloud/__init__.py @@ -230,6 +230,26 @@ def inputs(self) -> Generator[PlejdSceneData, None, None]: "first_device": firstDevice, } + @property + def groups(self) -> Generator[dict, None, None]: + """EXPERIMENTAL: room/group mesh addresses. + + Each Plejd room has its own mesh address (roomAddress) that a single + group command can be sent to in order to actuate every output in the + room at once. Yields {roomId, title, address}. + """ + details = self.details + if not details: + raise RuntimeError("No site details have been fetched") + + for roomId, address in details.roomAddress.items(): + room = details.find_room(roomId) + yield { + "roomId": roomId, + "title": room.title if room else roomId, + "address": address, + } + @property def scenes(self) -> Generator[dict, None, None]: if not self.details: diff --git a/pyplejd/errors.py b/pyplejd/errors.py index 77c3a44..e86156f 100644 --- a/pyplejd/errors.py +++ b/pyplejd/errors.py @@ -4,3 +4,12 @@ class AuthenticationError(Exception): class ConnectionError(Exception): """Connection failed.""" + + +class PlejdWriteError(Exception): + """Writing a command to the Plejd mesh failed. + + Raised after a single quick reconnect-and-retry has also failed, so that + callers (e.g. light turn_on/turn_off) can surface the failure instead of + silently dropping the command. + """ diff --git a/pyplejd/interface/plejd_light.py b/pyplejd/interface/plejd_light.py index 937fbf9..4e50b30 100644 --- a/pyplejd/interface/plejd_light.py +++ b/pyplejd/interface/plejd_light.py @@ -1,6 +1,44 @@ +import asyncio +import contextlib +import logging + from .plejd_device import PlejdOutput, PlejdTraits, PlejdDeviceType from ..ble import LastData, MiniPkg, LightLevel from ..ble.debug import rec_log +from ..errors import PlejdWriteError + +_LOGGER = logging.getLogger(__name__) + +# Interval between individual dim writes when performing a software transition. +# Plejd allows roughly 5-20 writes/s, so ~150 ms keeps well within budget. +RAMP_INTERVAL = 0.15 + + +def ramp_steps(start, target, transition, interval=RAMP_INTERVAL): + """Generate the sequence of dim levels for a software transition. + + Pure function: returns the intermediate brightness levels (0-255) to write, + stepping linearly from `start` to `target` over `transition` seconds at + roughly `interval`-second spacing. The returned list always finishes + exactly on `target` and never contains a redundant leading duplicate of the + starting level. Consecutive duplicate levels are collapsed so a slow ramp + over a small range does not emit dozens of identical writes. + """ + start = int(start) + target = int(target) + n = max(1, round(transition / interval)) + steps = [] + prev = start + for i in range(1, n + 1): + level = round(start + (target - start) * i / n) + if level != prev: + steps.append(level) + prev = level + # Always finish exactly on target (also covers start == target, where the + # loop produces no movement). + if not steps or steps[-1] != target: + steps.append(target) + return steps class PlejdLight(PlejdOutput): @@ -17,6 +55,9 @@ def __init__(self, *args, **kwargs): ): self.colortemp = [ct.minTemperature, ct.maxTemperature] + # Currently running software-transition task, if any. + self._ramp_task: asyncio.Task | None = None + async def parse_lightlevel(self, level: LightLevel): state = self._state state.update( @@ -59,55 +100,125 @@ async def parse_lastdata(self, data: LastData): for listener in self._listeners: listener(self._state) - async def turn_on(self, dim=None, colortemp=None): + def _current_level(self) -> int: + """Best known current brightness (0 when the light is off).""" + if not self._state.get("state"): + return 0 + return int(self._state.get("dim") or 0) + + def _dim_command(self, dim: int) -> LastData: + dim = int(dim) + return LastData( + address=self.address, + command=LastData.CMD_GROUP_OUTPUT_STATE_AND_LEVEL, + payload=[0x1, dim, dim], + ) + + def _on_command(self) -> LastData: + return LastData( + address=self.address, + command=LastData.CMD_GROUP_OUTPUT_STATE, + payload=[0x1], + ) + + def _off_command(self) -> LastData: + return LastData( + address=self.address, + command=LastData.CMD_GROUP_OUTPUT_STATE, + payload=[0x0], + ) + + def _colortemp_command(self, colortemp: float) -> LastData: + colortemp = int(1e6 / colortemp) + return LastData( + address=self.address, + command=LastData.CMD_OUTPUT_SET, + payload=[ + MiniPkg(type=MiniPkg.TPE_SOURCE, payload=[MiniPkg.SRC_MANUAL]), + MiniPkg( + type=MiniPkg.TPE_WHITEBALANCE, + payload=colortemp.to_bytes(2, byteorder="big"), + ), + ], + ) + + async def _cancel_ramp(self): + """Cancel any in-flight software transition (latest command wins). + + Awaits the cancelled task so a new ramp cannot briefly overlap writes + with the old one. + """ + task = self._ramp_task + self._ramp_task = None + if task is not None and not task.done(): + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + async def _run_ramp(self, commands: list[LastData]): + """Write the remaining ramp steps, one every RAMP_INTERVAL seconds. + + Never lets an exception escape the task: a write failure mid-fade is + logged once (with device context) and stops the ramp, and cancellation + propagates so the awaiting canceller can settle. + """ + try: + for cmd in commands: + await asyncio.sleep(RAMP_INTERVAL) + await self._mesh.write(cmd.hex) + except asyncio.CancelledError: + raise + except PlejdWriteError as err: + _LOGGER.warning( + "Transition aborted for Plejd device %s: %s", self.address, err + ) + + async def turn_on(self, dim=None, colortemp=None, transition=None): if not self._mesh: return + + # A new command always supersedes a running transition. + await self._cancel_ramp() + + # Software transition: only meaningful when ramping brightness to a + # known target. Fall through to the immediate path otherwise. + if transition and dim is not None: + levels = ramp_steps(self._current_level(), int(dim), float(transition)) + # The colour temperature and the first brightness step are written + # synchronously so an immediate failure raises to the caller (and + # is surfaced by Home Assistant); remaining steps run in the task. + if colortemp is not None: + await self._mesh.write(self._colortemp_command(colortemp).hex) + await self._mesh.write(self._dim_command(levels[0]).hex) + remaining = [self._dim_command(level) for level in levels[1:]] + if remaining: + self._ramp_task = asyncio.ensure_future(self._run_ramp(remaining)) + return + commands: list[LastData] = [] if dim is not None: - dim = int(dim) - commands.append( - LastData( - address=self.address, - command=LastData.CMD_GROUP_OUTPUT_STATE_AND_LEVEL, - payload=[0x1, dim, dim], - ) - ) + commands.append(self._dim_command(dim)) else: - commands.append( - LastData( - address=self.address, - command=LastData.CMD_GROUP_OUTPUT_STATE, - payload=[0x1], - ) - ) + commands.append(self._on_command()) if colortemp is not None: - colortemp = int(1e6 / colortemp) - commands.append( - LastData( - address=self.address, - command=LastData.CMD_OUTPUT_SET, - payload=[ - MiniPkg( - type=MiniPkg.TPE_SOURCE, - payload=[MiniPkg.SRC_MANUAL], - ), - MiniPkg( - type=MiniPkg.TPE_WHITEBALANCE, - payload=colortemp.to_bytes(2), - ), - ], - ) - ) + commands.append(self._colortemp_command(colortemp)) await self._mesh.write(*(c.hex for c in commands)) - async def turn_off(self): + async def turn_off(self, transition=None): if not self._mesh: return - cmd = LastData( - address=self.address, - command=LastData.CMD_GROUP_OUTPUT_STATE, - payload=[0x0], - ) - await self._mesh.write(cmd.hex) + await self._cancel_ramp() + + # Fade to off when a transition is requested and the light is currently + # on, then send an explicit off command as the final step. + if transition and self._current_level() > 0: + levels = ramp_steps(self._current_level(), 0, float(transition)) + await self._mesh.write(self._dim_command(levels[0]).hex) + remaining = [self._dim_command(level) for level in levels[1:]] + remaining.append(self._off_command()) + self._ramp_task = asyncio.ensure_future(self._run_ramp(remaining)) + return + + await self._mesh.write(self._off_command().hex) diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..78c5011 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +asyncio_mode = auto +testpaths = tests diff --git a/setup.py b/setup.py index 63ce742..7770389 100644 --- a/setup.py +++ b/setup.py @@ -1,8 +1,8 @@ from setuptools import find_packages, setup -MIN_PY_VERSION = "3.10" +MIN_PY_VERSION = "3.11" PACKAGES = find_packages() -VERSION = "0.21.3" +VERSION = "0.22.0+fork.1" setup( name="pyplejd", diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/requirements.txt b/tests/requirements.txt new file mode 100644 index 0000000..ed7ee95 --- /dev/null +++ b/tests/requirements.txt @@ -0,0 +1,7 @@ +pytest +pytest-asyncio +aiohttp +bleak +bleak_retry_connector +pydantic +cryptography diff --git a/tests/test_connect.py b/tests/test_connect.py new file mode 100644 index 0000000..237045d --- /dev/null +++ b/tests/test_connect.py @@ -0,0 +1,103 @@ +"""Regression tests for the connect path (review round 1: C1 phantom bleak API +and P2 single-flight connect).""" + +from types import SimpleNamespace + +import asyncio + +from pyplejd import ble as ble_mod +from pyplejd.ble import PlejdMesh, MeshDevice + +KEY = "00112233445566778899aabbccddeeff" +GATEWAY = "112233445566" + + +class SpecFakeClient: + """A deliberately spec'd fake exposing ONLY the bleak 3.x client methods + pyplejd actually uses. + + Unlike MagicMock, accessing an unknown attribute (e.g. the removed + ``set_disconnected_callback``) raises AttributeError - which is exactly the + regression guard for finding C1. + """ + + def __init__(self): + self.is_connected = True + self.started = [] + + async def start_notify(self, char, cb): + self.started.append(char) + + async def write_gatt_char(self, char, data, response=True): + pass + + async def disconnect(self): + self.is_connected = False + + +class FakeManager: + def __init__(self): + self.connected_calls = [] + + def connect_callback(self, connected): + self.connected_calls.append(connected) + + +def _mesh_with_node(): + mesh = PlejdMesh(FakeManager()) + mesh.set_key(KEY) + node = MeshDevice() + node.connectable = True + node.BLEaddress = GATEWAY + node.bleDevice = SimpleNamespace(name="gw") + node.see(-50, node.bleDevice, now=1000.0) + mesh._mesh_devices = {GATEWAY: node} + return mesh + + +async def test_connect_uses_only_available_bleak_api(monkeypatch): + # C1: the connect path must not call the removed + # BleakClient.set_disconnected_callback. The spec'd fake would raise + # AttributeError if it were called. + mesh = _mesh_with_node() + fake = SpecFakeClient() + assert not hasattr(fake, "set_disconnected_callback") + + captured = {} + + async def fake_establish(cls, device, name, disconnected_callback=None, **kwargs): + captured["cb"] = disconnected_callback + return fake + + monkeypatch.setattr(ble_mod, "establish_connection", fake_establish) + + async def fake_auth(client): + return True + + monkeypatch.setattr(mesh, "_authenticate", fake_auth) + + ok = await mesh.connect() + assert ok is True + assert mesh.connected is True + # The disconnect callback is supplied at connection time, not via the + # removed setter. + assert callable(captured["cb"]) + + +async def test_connect_is_single_flight(monkeypatch): + # P2: concurrent connect attempts must collapse into a single _connect(). + mesh = PlejdMesh(FakeManager()) + calls = 0 + + async def slow_connect(): + nonlocal calls + calls += 1 + await asyncio.sleep(0.05) + mesh._client = SpecFakeClient() + return True + + monkeypatch.setattr(mesh, "_connect", slow_connect) + + results = await asyncio.gather(mesh.connect(), mesh.connect(), mesh.connect()) + assert results == [True, True, True] + assert calls == 1 diff --git a/tests/test_group.py b/tests/test_group.py new file mode 100644 index 0000000..9c84bb9 --- /dev/null +++ b/tests/test_group.py @@ -0,0 +1,38 @@ +"""Tests for the experimental group (room) dim command construction (patch 6).""" + +from pyplejd import PlejdManager + + +class CaptureMesh: + def __init__(self): + self.writes = [] + + async def write(self, *payloads): + self.writes.extend(payloads) + + +def make_manager(): + mgr = PlejdManager("user", "pass", "site") + mgr.mesh = CaptureMesh() + return mgr + + +async def test_dim_group_sets_level(): + mgr = make_manager() + await mgr.dim_group(5, 128) + # address 05, version 01, cmd_type 10, command 0098, payload 01 80 80 + assert mgr.mesh.writes == ["0501100098018080"] + + +async def test_dim_group_none_turns_on(): + mgr = make_manager() + await mgr.dim_group(5) + # command 0097 (group output state), payload 01 (on) + assert mgr.mesh.writes == ["050110009701"] + + +async def test_dim_group_zero_turns_off(): + mgr = make_manager() + await mgr.dim_group(5, 0) + # command 0097 (group output state), payload 00 (off) + assert mgr.mesh.writes == ["050110009700"] diff --git a/tests/test_ramp.py b/tests/test_ramp.py new file mode 100644 index 0000000..02976ab --- /dev/null +++ b/tests/test_ramp.py @@ -0,0 +1,149 @@ +"""Tests for software-transition ramp generation, cancellation and error +containment (patch 5 + review round 1 C2/C3).""" + +import logging + +import pytest + +from pyplejd.errors import PlejdWriteError +from pyplejd.interface.plejd_light import ramp_steps, PlejdLight, RAMP_INTERVAL + + +def test_ramp_finishes_exactly_on_target(): + steps = ramp_steps(0, 255, transition=1.0) + assert steps[-1] == 255 + + +def test_ramp_is_monotonic_up(): + steps = ramp_steps(10, 200, transition=2.0) + assert steps == sorted(steps) + assert steps[-1] == 200 + + +def test_ramp_is_monotonic_down(): + steps = ramp_steps(200, 10, transition=2.0) + assert steps == sorted(steps, reverse=True) + assert steps[-1] == 10 + + +def test_ramp_step_count_matches_duration(): + # ~150 ms per step -> a 1.5 s transition is about 10 steps. + steps = ramp_steps(0, 100, transition=1.5) + expected = round(1.5 / RAMP_INTERVAL) + assert 1 <= len(steps) <= expected + 1 + assert steps[-1] == 100 + + +def test_ramp_no_redundant_duplicates(): + steps = ramp_steps(50, 55, transition=5.0) + assert all(a != b for a, b in zip(steps, steps[1:])) + assert steps[-1] == 55 + + +def test_ramp_equal_start_target_still_writes_target(): + steps = ramp_steps(100, 100, transition=1.0) + assert steps == [100] + + +def test_ramp_tiny_transition_is_single_step(): + steps = ramp_steps(0, 128, transition=0.01) + assert steps == [128] + + +class _FakeMesh: + def __init__(self, fail_at=None): + # fail_at: raise PlejdWriteError once the Nth payload (1-indexed) is + # reached. + self.writes = [] + self.fail_at = fail_at + + async def write(self, *payloads): + for p in payloads: + self.writes.append(p) + if self.fail_at is not None and len(self.writes) >= self.fail_at: + raise PlejdWriteError("simulated write failure") + + +def _make_light(fail_at=None, state=True, dim=0): + light = PlejdLight.__new__(PlejdLight) + light._mesh = _FakeMesh(fail_at=fail_at) + light._state = {"state": state, "dim": dim} + light._ramp_task = None + light.address = 1 + return light + + +@pytest.mark.asyncio +async def test_ramp_task_runs_and_reaches_target(): + light = _make_light() + await light.turn_on(dim=255, transition=0.5) + assert light._ramp_task is not None + await light._ramp_task + # Final write is the target dim command (0xff appears in the payload). + assert "ff" in light._mesh.writes[-1] + + +@pytest.mark.asyncio +async def test_first_step_written_synchronously_and_failure_raises(): + # C2: an immediate failure on the first step must raise to the caller. + light = _make_light(fail_at=1) + with pytest.raises(PlejdWriteError): + await light.turn_on(dim=255, transition=0.5) + # No detached task is left behind. + assert light._ramp_task is None + + +@pytest.mark.asyncio +async def test_mid_ramp_failure_is_contained(caplog): + # C2: a failure during the async part of the ramp is logged once and does + # not escape the task (no "exception was never retrieved"). + caplog.set_level(logging.WARNING) + light = _make_light(fail_at=2) # first (sync) step ok, next step fails + await light.turn_on(dim=255, transition=0.5) + task = light._ramp_task + assert task is not None + await task # must not raise + assert task.done() + assert task.exception() is None + assert any("Transition aborted" in r.getMessage() for r in caplog.records) + + +@pytest.mark.asyncio +async def test_latest_command_cancels_previous_ramp(): + light = _make_light() + await light.turn_on(dim=255, transition=5.0) + first = light._ramp_task + # New command supersedes and awaits cancellation of the running transition. + await light.turn_on(dim=100, transition=5.0) + assert first.cancelled() + assert light._ramp_task is not first + await light._cancel_ramp() + + +@pytest.mark.asyncio +async def test_turn_off_cancels_ramp(): + light = _make_light() + await light.turn_on(dim=255, transition=5.0) + task = light._ramp_task + await light.turn_off() + assert task.cancelled() + assert light._ramp_task is None + + +@pytest.mark.asyncio +async def test_turn_off_with_transition_fades_then_off(): + light = _make_light(state=True, dim=255) + await light.turn_off(transition=0.5) + task = light._ramp_task + assert task is not None + await task + # Final write is the explicit off command (group output state, payload 00). + assert light._mesh.writes[-1] == "010110009700" + + +@pytest.mark.asyncio +async def test_turn_off_without_transition_is_immediate(): + light = _make_light(state=True, dim=255) + await light.turn_off() + assert light._ramp_task is None + assert light._mesh.writes == ["010110009700"] diff --git a/tests/test_rssi_decay.py b/tests/test_rssi_decay.py new file mode 100644 index 0000000..101dbfe --- /dev/null +++ b/tests/test_rssi_decay.py @@ -0,0 +1,71 @@ +"""Tests for MeshDevice RSSI decay and BLEDevice handle refresh (patch 3). + +Time is expressed as a monotonic-clock float (seconds), matching the +production code which uses time.monotonic() for decay math. +""" + +from datetime import datetime + +from pyplejd.ble import MeshDevice + +T0 = 1000.0 # arbitrary monotonic base + + +def _dev(): + d = MeshDevice() + d.connectable = True + return d + + +def test_first_sighting_stores_rssi_and_handle(): + d = _dev() + first = d.see(-50, "handle-1", now=T0) + assert first is True + assert d.rssi == -50 + assert d.bleDevice == "handle-1" + # last_seen is a wall-clock datetime for the user-facing sensor. + assert isinstance(d.last_seen, datetime) + + +def test_handle_refreshed_on_every_advertisement(): + d = _dev() + d.see(-50, "handle-1", now=T0) + second = d.see(-60, "handle-2", now=T0 + 1) + assert second is False + # Handle is always refreshed to the most recent advertisement. + assert d.bleDevice == "handle-2" + + +def test_current_rssi_decays_toward_floor(): + d = _dev() + d.see(-40, "h", now=T0) + # After exactly one half-life, value should be halfway to the -100 floor. + half = T0 + MeshDevice.RSSI_HALFLIFE_S + expected = MeshDevice.RSSI_FLOOR + (-40 - MeshDevice.RSSI_FLOOR) * 0.5 + assert abs(d.current_rssi(half) - expected) < 1e-6 + # Far in the future it approaches, but never passes, the floor. + far = T0 + 3600 + assert d.current_rssi(far) > MeshDevice.RSSI_FLOOR + assert d.current_rssi(far) < -95 + + +def test_stronger_recent_sighting_beats_decayed_value(): + d = _dev() + d.see(-40, "h", now=T0) + # Long silence, then a weak sighting: stored value should not be the old + # historical max, but the fresh (weak) reading, since decay dropped below. + d.see(-70, "h", now=T0 + 1800) + assert d.rssi == -70 + + +def test_strong_sighting_after_short_gap_keeps_max(): + d = _dev() + d.see(-40, "h", now=T0) + # Almost no decay after 1s; a weaker reading keeps the decayed strong one. + d.see(-55, "h", now=T0 + 1) + assert d.rssi > -55 + + +def test_current_rssi_none_when_never_seen(): + d = _dev() + assert d.current_rssi() is None diff --git a/tests/test_write_retry.py b/tests/test_write_retry.py new file mode 100644 index 0000000..077c3d1 --- /dev/null +++ b/tests/test_write_retry.py @@ -0,0 +1,103 @@ +"""Tests for connected-liveness and the write reconnect-retry (patches 1 & 2).""" + +import pytest +from bleak import BleakError + +from pyplejd.ble import PlejdMesh, MeshDevice +from pyplejd.errors import PlejdWriteError + +KEY = "00112233445566778899aabbccddeeff" # 16-byte AES key as hex +GATEWAY = "112233445566" +# LastData(address=1, command=0x0097, payload=[1]) -> group ON command. +PAYLOAD = "010110009701" + + +class FakeClient: + def __init__(self, connected=True, fail=0): + self.is_connected = connected + self._fail = fail + self.writes = [] + + async def write_gatt_char(self, char, payload, response=True): + if self._fail > 0: + self._fail -= 1 + raise BleakError("simulated write failure") + self.writes.append(payload) + + +class FakeManager: + def connect_callback(self, connected): + pass + + +def make_mesh(client): + mesh = PlejdMesh(FakeManager()) + mesh.set_key(KEY) + gw = MeshDevice() + gw.BLEaddress = GATEWAY + mesh._gateway_node = gw + mesh._client = client + return mesh + + +def test_connected_requires_live_link(): + mesh = make_mesh(FakeClient(connected=False)) + assert mesh.connected is False + mesh._client = FakeClient(connected=True) + assert mesh.connected is True + mesh._client = None + assert mesh.connected is False + + +async def test_write_once_raises_when_disconnected(): + mesh = make_mesh(FakeClient(connected=False)) + with pytest.raises(PlejdWriteError): + await mesh._write_once([PAYLOAD]) + + +async def test_write_succeeds_without_retry(): + client = FakeClient(connected=True) + mesh = make_mesh(client) + await mesh.write(PAYLOAD) + assert len(client.writes) == 1 + + +async def test_write_retries_after_transient_failure(): + bad = FakeClient(connected=True, fail=1) + good = FakeClient(connected=True) + mesh = make_mesh(bad) + + async def fake_connect(): + mesh._client = good + return True + + mesh.connect = fake_connect + await mesh.write(PAYLOAD) + # First (failing) attempt wrote nothing; retry after reconnect succeeded. + assert bad.writes == [] + assert len(good.writes) == 1 + + +async def test_write_reconnects_when_initially_disconnected(): + good = FakeClient(connected=True) + mesh = make_mesh(FakeClient(connected=False)) + + async def fake_connect(): + mesh._client = good + return True + + mesh.connect = fake_connect + await mesh.write(PAYLOAD) + assert len(good.writes) == 1 + + +async def test_write_raises_after_retry_also_fails(): + bad = FakeClient(connected=True, fail=2) + mesh = make_mesh(bad) + + async def fake_connect(): + return True # client stays broken + + mesh.connect = fake_connect + with pytest.raises(PlejdWriteError): + await mesh.write(PAYLOAD)