Merge pull request #69 from plugwise/add_lock_state

Add support for getting and setting the lock-state of Plugs, Circles, etc.

Author: bouwew

.github/workflows/verify.yml

@@ -4,7 +4,7 @@
 name: Latest commit
 
 env:
-  CACHE_VERSION: 3
+  CACHE_VERSION: 2
   DEFAULT_PYTHON: 3.9
   PRE_COMMIT_HOME: ~/.cache/pre-commit
 

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.9.3 - Add lock-state switches
+
+- Support for getting and setting the lock-state of Plugs-, Circles-, Stealth-switches. A set lock-state prevents a switch from being turned off.
+- There is no lock_state available for the following special Plugwise classes: `central heating pump` and `value actuator`
+
 ## 0.9.2 - Smile Optimize 
 
 - Functions not called by the plugwise(-beta) code have been moved to helper.py in which they are part of the subclass SmileHelper

plugwise/__init__.py

@@ -1,6 +1,6 @@
 """Plugwise module."""
 
-__version__ = "0.9.2"
+__version__ = "0.9.3"
 
 from plugwise.smile import Smile
 from plugwise.stick import stick

plugwise/helper.py

@@ -518,7 +518,6 @@ def appliance_data(self, dev_id):
         """
         data = {}
         search = self._appliances
-
         if self._smile_legacy and self.smile_type != "stretch":
             search = self._domain_objects
 
@@ -531,6 +530,7 @@ def appliance_data(self, dev_id):
                     **DEVICE_MEASUREMENTS,
                     **HEATER_CENTRAL_MEASUREMENTS,
                 }.items()
+
             for measurement, attrs in measurements:
 
                 p_locator = (
@@ -565,9 +565,10 @@ def appliance_data(self, dev_id):
                 if appliance.find(i_locator) is not None:
                     name = f"{measurement}_interval"
                     measure = appliance.find(i_locator).text
-
                     data[name] = format_measure(measure, ENERGY_WATT_HOUR)
 
+            data.update(self.get_lock_state(appliance))
+
         # Fix for Adam + Anna: heating_state also present under Anna, remove
         if "temperature" in data:
             data.pop("heating_state", None)
@@ -914,3 +915,20 @@ def object_value(self, obj_type, obj_id, measurement):
             return val
 
         return None
+
+    def get_lock_state(self, xml):
+        """Adam & Stretches: find relay switch lock state."""
+        data = {}
+        actuator = "actuator_functionalities"
+        func_type = "relay_functionality"
+        if self.smile_type == "stretch" and self.smile_version[1].major == 2:
+            actuator = "actuators"
+            func_type = "relay"
+        appl_class = xml.find("type").text
+        if appl_class not in ["central_heating_pump", "valve_actuator"]:
+            locator = f".//{actuator}/{func_type}/lock"
+            if xml.find(locator) is not None:
+                measure = xml.find(locator).text
+                data["lock"] = format_measure(measure, None)
+
+        return data

plugwise/smile.py

@@ -371,11 +371,17 @@ async def set_temperature(self, loc_id, temperature):
     async def set_switch_state(self, appl_id, members, model, state):
         """Switch the Switch off/on."""
         actuator = "actuator_functionalities"
-        func_type = "relay_functionality"
         device = "relay"
+        func_type = "relay_functionality"
+        func = "state"
         if model == "dhw_cm_switch":
-            func_type = "toggle_functionality"
             device = "toggle"
+            func_type = "toggle_functionality"
+
+        if model == "lock":
+            func = "lock"
+            state = "false" if state == "off" else "true"
+
         stretch_v2 = self.smile_type == "stretch" and self.smile_version[1].major == 2
         if stretch_v2:
             actuator = "actuators"
@@ -399,8 +405,17 @@ async def set_switch_state(self, appl_id, members, model, state):
         uri = f"{APPLIANCES};id={appl_id}/{device};id={switch_id}"
         if stretch_v2:
             uri = f"{APPLIANCES};id={appl_id}/{device}"
-        state = str(state)
-        data = f"<{func_type}><state>{state}</state></{func_type}>"
+        data = f"<{func_type}><{func}>{state}</{func}></{func_type}>"
+
+        if model == "relay":
+            locator = f'appliance[@id="{appl_id}"]/{actuator}/{func_type}/lock'
+            lock_state = self._appliances.find(locator).text
+            print("Lock state: ", lock_state)
+            # Don't bother switching a relay when the corresponding lock-state is true
+            if lock_state == "true":
+                return False
+            await self.request(uri, method="put", data=data)
+            return True
 
         await self.request(uri, method="put", data=data)
         return True

plugwise/util.py

@@ -114,9 +114,9 @@ def format_measure(measure, unit):
                 elif abs(float(measure)) >= 100:
                     measure = int(round(float(measure)))
         except ValueError:
-            if measure == "on":
+            if measure in ["on", "true"]:
                 measure = True
-            elif measure == "off":
+            if measure in ["off", "false"]:
                 measure = False
     return measure
 

tests/test_smile.py

@@ -408,19 +408,18 @@ async def device_test(self, smile=pw_smile.Smile, testdata=None):
 
     @pytest.mark.asyncio
     async def tinker_switch(
-        self, smile, dev_id=None, members=None, model=None, unhappy=False
+        self, smile, dev_id=None, members=None, model="relay", unhappy=False
     ):
         """Turn a Switch on and off to test functionality."""
         _LOGGER.info("Asserting modifying settings for switch devices:")
         _LOGGER.info("- Devices (%s):", dev_id)
+        switch_change = False
         for new_state in [False, True, False]:
             _LOGGER.info("- Switching %s", new_state)
             try:
                 switch_change = await smile.set_switch_state(
                     dev_id, members, model, new_state
                 )
-                assert switch_change
-                _LOGGER.info("  + worked as intended")
             except (
                 pw_exceptions.ErrorSendingCommandError,
                 pw_exceptions.ResponseError,
@@ -430,6 +429,7 @@ async def tinker_switch(
                 else:  # pragma: no cover
                     _LOGGER.info("  - failed unexpectedly")
                     raise self.UnexpectedError
+        return switch_change
 
     @pytest.mark.asyncio
     async def tinker_thermostat(self, smile, loc_id, good_schemas=None, unhappy=False):
@@ -941,6 +941,7 @@ async def test_connect_adam_plus_anna(self):
             # Plug MediaCenter
             "aa6b0002df0a46e1b1eb94beb61eddfe": {
                 "electricity_consumed": 10.3,
+                "lock": False,
                 "relay": True,
             },
         }
@@ -967,7 +968,10 @@ async def test_connect_adam_plus_anna(self):
         await self.tinker_thermostat(
             smile, "009490cc2f674ce6b576863fbb64f867", good_schemas=["Weekschema"]
         )
-        await self.tinker_switch(smile, "aa6b0002df0a46e1b1eb94beb61eddfe")
+        switch_change = await self.tinker_switch(
+            smile, "aa6b0002df0a46e1b1eb94beb61eddfe"
+        )
+        assert switch_change
         await smile.close_connection()
         await self.disconnect(server, client)
 
@@ -978,9 +982,10 @@ async def test_connect_adam_plus_anna(self):
             good_schemas=["Weekschema"],
             unhappy=True,
         )
-        await self.tinker_switch(
+        switch_change = await self.tinker_switch(
             smile, "aa6b0002df0a46e1b1eb94beb61eddfe", unhappy=True
         )
+        assert not switch_change
         await smile.close_connection()
         await self.disconnect(server, client)
 
@@ -1009,15 +1014,24 @@ async def test_connect_adam_plus_anna_new(self):
         await self.device_test(smile, testdata)
         assert smile.active_device_present
 
-        await self.tinker_switch(
+        switch_change = await self.tinker_switch(
             smile,
             "b83f9f9758064c0fab4af6578cba4c6d",
             ["aa6b0002df0a46e1b1eb94beb61eddfe", "f2be121e4a9345ac83c6e99ed89a98be"],
         )
-        await self.tinker_switch(
+        assert switch_change
+        switch_change = await self.tinker_switch(
             smile, "2743216f626f43948deec1f7ab3b3d70", model="dhw_cm_switch"
         )
-
+        assert switch_change
+        switch_change = await self.tinker_switch(
+            smile, "40ec6ebe67844b21914c4a5382a3f09f", model="lock"
+        )
+        assert switch_change
+        switch_change = await self.tinker_switch(
+            smile, "f2be121e4a9345ac83c6e99ed89a98be"
+        )
+        assert not switch_change
         await smile.close_connection()
         await self.disconnect(server, client)
 
@@ -1087,7 +1101,10 @@ async def test_connect_adam_zone_per_device(self):
         await self.tinker_thermostat(
             smile, "82fa13f017d240daa0d0ea1775420f24", good_schemas=["CV Jessie"]
         )
-        await self.tinker_switch(smile, "675416a629f343c495449970e2ca37b5")
+        switch_change = await self.tinker_switch(
+            smile, "675416a629f343c495449970e2ca37b5"
+        )
+        assert not switch_change
         await smile.close_connection()
         await self.disconnect(server, client)
 
@@ -1181,7 +1198,10 @@ async def test_connect_adam_multiple_devices_per_zone(self):
         await self.tinker_thermostat(
             smile, "82fa13f017d240daa0d0ea1775420f24", good_schemas=["CV Jessie"]
         )
-        await self.tinker_switch(smile, "675416a629f343c495449970e2ca37b5")
+        switch_change = await self.tinker_switch(
+            smile, "675416a629f343c495449970e2ca37b5"
+        )
+        assert not switch_change
         await smile.close_connection()
         await self.disconnect(server, client)
 
@@ -1459,6 +1479,7 @@ async def test_connect_stretch_v23(self):
             # Tv hoek 25F6790
             "c71f1cb2100b42ca942f056dcb7eb01f": {
                 "electricity_consumed": 33.3,
+                "lock": False,
                 "relay": True,
             },
             # Wasdroger 043AECA
@@ -1481,12 +1502,16 @@ async def test_connect_stretch_v23(self):
         _LOGGER.info(" # Assert no master thermostat")
         assert smile.single_master_thermostat() is None  # it's not a thermostat :)
 
-        await self.tinker_switch(smile, "2587a7fcdd7e482dab03fda256076b4b")
-        await self.tinker_switch(
+        switch_change = await self.tinker_switch(
+            smile, "2587a7fcdd7e482dab03fda256076b4b"
+        )
+        assert switch_change
+        switch_change = await self.tinker_switch(
             smile,
             "f7b145c8492f4dd7a4de760456fdef3e",
             ["407aa1c1099d463c9137a3a9eda787fd"],
         )
+        assert switch_change
 
         smile.get_all_devices()
         await self.device_test(smile, testdata)

userdata/adam_plus_anna_new/core.appliances.xml

@@ -71,7 +71,7 @@
 		<actuator_functionalities>
 			<relay_functionality id='dc6998c6abcb4fec94db2043f78a29c1'>
 				<updated_date>2020-11-10T15:49:54.788+01:00</updated_date>
-				<lock>false</lock>
+				<lock>true</lock>
 				<state>on</state>
 				<relay id='9f35a3b3d6094816952e801f66964849'/>
 			</relay_functionality>