Reliability patches (fork): liveness, write retries, RSSI decay, transitions

.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

.gitignore

@@ -1,10 +1,14 @@
 /*
 /*/
+!.gitignore
 !gitignore
 !README.md
 !LICENSE
 !setup.py
 !setup.cfg
 !pyplejd/
+!tests/
+!pytest.ini
+!.github/
 **/__pycache__/
-!compiling.md
+!compiling.md

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()