Support for shades and other API endpoints

Author: retoo

dingz/constants.py

@@ -21,19 +21,23 @@
 LIGHT = "light"
 SCHEDULE = "schedule"
 TIMER = "timer"
+SHADE = "shade"
+STATE = "state"
 
 # Special endpoints
 LOG = "/log"
 FIRMWARE = "/load"
 
 SETTINGS = "settings"
+SYSTEM_CONFIG = "system_config"
 FRONT_LED_GET = "led/get"
 FRONT_LED_SET = "led/set"
 BUTTON_ACTIONS = "action"
 WIFI_SCAN = "scan"
 
 # Configuration endpoints
 PIR_CONFIGURATION = "pir_config"
+BLIND_CONFIGURATION = "blind_config"
 THERMOSTAT_CONFIGURATION = "thermostat_config"
 INPUT_CONFIGURATION = "input_config"
 

dingz/dingz.py

@@ -21,6 +21,11 @@
     WIFI_SCAN,
     TIMER,
     SCHEDULE,
+    SHADE,
+    INFO,
+    STATE,
+    SYSTEM_CONFIG,
+    BLIND_CONFIGURATION,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -35,6 +40,7 @@ def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> N
         self._host = host
         self._session = session
         self._device_details = None
+        self._info = None
         self._wifi_networks = None
         self._settings = None
         self._catch_all = {}
@@ -47,15 +53,26 @@ def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> N
         self._motion = None
         self._schedule = None
         self._timer = None
+        self._state = {}
+        self._blind_config = None
+        self._system_config = None
+
         self.uri = URL.build(scheme="http", host=self._host).join(URL(API))
 
     async def get_device_info(self) -> None:
         """Get the details from the dingz."""
         url = URL(self.uri).join(URL(DEVICE_INFO))
         response = await make_call(self, uri=url)
-        self._device_details = response
+        # response is:  "mac" => { device_details }
+        self._device_details = next(iter(response.values()))
 
     async def get_info(self) -> None:
+        """Get general information fro the dingz."""
+        url = URL(self.uri).join(URL(INFO))
+        response = await make_call(self, uri=url)
+        self._info = response
+
+    async def get_all_info(self) -> None:
         """Get everything from the dingz unit."""
         for endpoint in [
             PUCK,
@@ -118,6 +135,21 @@ async def get_light(self) -> None:
         self._intensity = response["intensity"]
         self._hour_of_day = response["state"]
 
+    async def get_state(self) -> None:
+        url = URL(self.uri).join(URL(STATE))
+        response = await make_call(self, uri=url)
+        self._state = response
+
+    async def get_blind_config(self) -> None:
+        url = URL(self.uri).join(URL(BLIND_CONFIGURATION))
+        response = await make_call(self, uri=url)
+        self._blind_config = response
+
+    async def get_system_config(self) -> None:
+        url = URL(self.uri).join(URL(SYSTEM_CONFIG))
+        response = await make_call(self, uri=url)
+        self._system_config = response
+
     async def enabled(self) -> bool:
         """Return true if front LED is on."""
         url = URL(self.uri).join(URL(FRONT_LED_GET))
@@ -136,12 +168,69 @@ async def turn_off(self) -> None:
         url = URL(self.uri).join(URL(FRONT_LED_SET))
         await make_call(self, uri=url, method="POST", data=data)
 
+    async def operate_shade(self, shade_no, blind=None, lamella=None) -> None:
+        """operate the lamella and blind
+        blind: 0 fully closed, 100 fully open
+        lamella: 0 lamellas closed, 100 lamellas open
+        """
+
+        # With newer versions of dingz, we can just leave
+        # either lamella or blind None (i.e. do not chang)
+        # but currently we need to lookup the current state
+        # of the shade first.
+        if blind is None or lamella is None:
+            await self.get_state()
+
+            if blind is None:
+                blind = self.current_blind_level(shade_no)
+
+            if lamella is None:
+                if self.is_shade_opened(shade_no):
+                    # if the shade is currently completely opened (i.e. up), the lamella
+                    # value is not really relevant (has no effect). We assume the
+                    # lamella value to be 0, ie. closed.
+                    # i.e. we set lamella to 45, raise blind to the top, and then back down again
+                    # => de we expect the lamella to be set to 45 again, or does it get resetted to 0?
+                    lamella = 0
+                else:
+                    lamella = self.current_lamella_level(shade_no)
+
+        url = URL(self.uri).join(URL("%s/%s" % (SHADE, shade_no)))
+        params = {"blind": str(blind), "lamella": str(lamella)}
+        await make_call(self, uri=url, method="POST", parameters=params)
+
+    async def shade_up(self, shade_no) -> None:
+        await self.shade_command(shade_no, "up")
+
+    async def shade_down(self, shade_no) -> None:
+        await self.shade_command(shade_no, "down")
+
+    async def shade_stop(self, shade_no) -> None:
+        await self.shade_command(shade_no, "stop")
+
+    async def lamella_open(self, shade_no) -> None:
+        await self.operate_shade(shade_no, lamella=100)
+
+    async def lamella_close(self, shade_no) -> None:
+        await self.operate_shade(shade_no, lamella=0)
+
+    async def lamella_stop(self, shade_no) -> None:
+        await self.shade_stop(shade_no)
+
+    async def shade_command(self, shade_no, verb):
+        url = URL(self.uri).join(URL("%s/%s/%s" % (SHADE, shade_no, verb)))
+        await make_call(self, uri=url, method="POST")
+
     async def set_timer(self, data) -> None:
         """Set a timer."""
         print(data)
         url = URL(self.uri).join(URL(TIMER))
         await make_call(self, uri=url, method="POST", json_data=data)
 
+    @property
+    def dingz_name(self) -> str:
+        return self._system_config["dingz_name"]
+
     @property
     def device_details(self) -> str:
         """Return the current device details."""
@@ -173,7 +262,7 @@ def wifi_networks(self) -> str:
         return self._wifi_networks
 
     @property
-    def everything(self) -> str:
+    def everything(self) -> dict:
         """Return the all available device details."""
         return self._catch_all
 
@@ -207,6 +296,76 @@ def intensity(self) -> float:
         """Return the current light intensity in lux."""
         return round(self._intensity, 1)
 
+    @property
+    def version(self) -> str:
+        return self._info["version"]
+
+    @property
+    def type(self) -> str:
+        return self._info["type"]
+
+    @property
+    def mac(self) -> str:
+        return self._info["mac"]
+
+    @property
+    def front_hw_model(self) -> str:
+        return self._device_details["front_hw_model"]
+
+    @property
+    def puck_hw_model(self) -> str:
+        return self._device_details["puck_hw_model"]
+
+    @property
+    def front_sn(self) -> str:
+        return self._device_details["front_sn"]
+
+    @property
+    def puck_sn(self) -> str:
+        return self._device_details["puck_sn"]
+
+    @property
+    def hw_version(self) -> str:
+        return self._device_details["hw_version"]
+
+    @property
+    def fw_version(self) -> str:
+        return self._device_details["fw_version"]
+
+    def _shade_current_state(self, shade_no: int):
+        return self._state["blinds"][shade_no]
+
+    def current_blind_level(self, shade_no):
+        return self._shade_current_state(shade_no)["position"]
+
+    def current_lamella_level(self, shade_no):
+        return self._shade_current_state(shade_no)["lamella"]
+
+    def is_shade_closed(self, shade_no):
+        # when closed, we care if the lamellas are opened or not
+        return (
+            self.current_blind_level(shade_no) == 0
+            and self.current_lamella_level(shade_no) == 0
+        )
+
+    def is_shade_opened(self, shade_no):
+        return self.current_blind_level(shade_no) == 100
+
+    def is_lamella_closed(self, shade_no):
+        return self.current_lamella_level(shade_no) == 0
+
+    def is_lamella_opened(self, shade_no):
+        return self.current_lamella_level(shade_no) == 100
+
+    def is_blind_opening(self, shade_no):
+        return self._shade_current_state(shade_no)["moving"] == "up"
+
+    def is_blind_closing(self, shade_no):
+        return self._shade_current_state(shade_no)["moving"] == "down"
+
+    def blind_name(self, shade_no):
+        return self._blind_config["blinds"][shade_no]["name"]
+
     # See "Using Asyncio in Python" by Caleb Hattingh for implementation
     # details.
     async def close(self) -> None:

dingz/discovery.py

@@ -0,0 +1,80 @@
+
+import asyncio
+import logging
+from typing import Optional, List
+
+_LOGGER = logging.getLogger(__name__)
+
+class DiscoveredDevice(object):
+    mac: str
+    type: int
+    is_child: bool
+    mystrom_registered: bool
+    mystrom_online: bool
+    restarted: bool
+
+    @staticmethod
+    def create_from_announce_msg(raw_addr, announce_msg):
+        if len(announce_msg) != 8:
+            raise RuntimeError("unexpected announcement, %s" % announce_msg)
+
+        device = DiscoveredDevice(host=raw_addr[0],
+                                  mac=announce_msg[0:6].hex(":"))
+        device.type = announce_msg[6]
+        status = announce_msg[7]
+        device.is_child = status & 1 != 0
+        device.mystrom_registered = status & 2 != 0
+        device.mystrom_online = status & 4 != 0
+        device.restarted = status & 8 != 0
+        return device
+
+    def __init__(self, host, mac):
+        self.host = host
+        self.mac = mac
+
+
+class DiscoveryProtocol(asyncio.DatagramProtocol):
+    def __init__(self, registry):
+        super().__init__()
+        self.registry = registry
+
+    def connection_made(self, transport):
+        self.transport = transport
+
+    def datagram_received(self, data, addr):
+        device = DiscoveredDevice.create_from_announce_msg(addr, data)
+        self.registry.register(device)
+
+    def connection_lost(self, exc: Optional[Exception]) -> None:
+        super().connection_lost(exc)
+
+
+class DeviceRegistry(object):
+    def __init__(self):
+        self.devices_by_mac = {}
+
+    def register(self, device):
+        self.devices_by_mac[device.mac] = device
+
+    def devices(self):
+        return list(self.devices_by_mac.values())
+
+
+async def discover_dingz_devices(timeout=7) -> List[DiscoveredDevice]:
+    registry = DeviceRegistry()
+    loop = asyncio.get_event_loop()
+    (transport, protocol) = await loop.create_datagram_endpoint(lambda: DiscoveryProtocol(registry), local_addr=("0.0.0.0", 7979))
+
+    await asyncio.sleep(timeout)
+    transport.close()
+    devices = registry.devices()
+    return devices
+
+
+if __name__ == "__main__":
+    loop = asyncio.get_event_loop()
+    devices = asyncio.run(discover_dingz_devices())
+
+    print("found %s devices" % len(devices))
+    for device in devices:
+        print(f'{device.mac} {device.host}')