CherryPickPoints

examples/example_specific_points.py

@@ -0,0 +1,48 @@
+"""List defined zones."""
+import asyncio
+import aiohttp
+
+from my_token import MY_TOKEN
+
+from myuplink.auth import Auth
+from myuplink import MyUplinkAPI
+
+PARAMETERS = ["40004", "40940", "40005"]
+
+
+async def main():
+    """Connect and print test data."""
+    async with aiohttp.ClientSession() as session:
+        auth = Auth(
+            session,
+            "https://api.myuplink.com",
+            MY_TOKEN,
+        )
+        api = MyUplinkAPI(auth)
+
+        systems = await api.async_get_systems()
+        for system in systems:
+            print(f"System id: {system.id}")
+            print(f"System name: {system.name}")
+
+            print(f"No of devices in system: {len(system.devices)}")
+            for sys_device in system.devices:
+                device = await api.async_get_device(sys_device.deviceId)
+                print(device.id)
+                print(device.productName)
+                print(device.productSerialNumber)
+                print(device.firmwareCurrent)
+                print(device.firmwareDesired)
+                print(device.connectionState)
+
+                points = await api.async_get_device_points(
+                    sys_device.deviceId, points=PARAMETERS
+                )
+                for point in points:
+                    print(
+                        f"{point.parameter_id} | {point.parameter_name} | {point.value}"
+                    )
+                print()
+
+
+asyncio.run(main())

src/myuplink/api.py

@@ -55,19 +55,24 @@ async def async_get_device_json(self, device_id) -> dict:
         return await resp.json()
 
     async def async_get_device_points(
-        self, device_id, language: str = "en-GB"
+        self, device_id, language: str = "en-GB", points: list[str] | None = None
     ) -> list[DevicePoint]:
         """Return device points."""
-        array = await self.async_get_device_points_json(device_id, language)
+        array = await self.async_get_device_points_json(device_id, language, points)
         return [DevicePoint(point_data) for point_data in array]
 
     async def async_get_device_points_json(
-        self, device_id, language: str = "en-GB"
+        self, device_id, language: str = "en-GB", points: list[str] | None = None
     ) -> dict:
         """Return device points as json."""
         headers = {"Accept-Language": language}
+        if points is None:
+            points = []
+        params = ""
+        if len(points) > 0:
+            params = "?parameters=" + ",".join(points)
         resp = await self.auth.request(
-            "get", f"v2/devices/{device_id}/points", headers=headers
+            "get", f"v2/devices/{device_id}/points{params}", headers=headers
         )
         resp.raise_for_status()
         return await resp.json()