Add initial power socket support (#27)

elro/__init__.py

@@ -1,3 +1,3 @@
 """Elro connects P1 API."""
 
-__version__ = "0.5.5.3"
+__version__ = "0.6.0"

elro/command.py

@@ -155,6 +155,26 @@ class CommandAttributes(TypedDict):
     content_transformer=None,
 )
 
+SOCKET_OFF = CommandAttributes(
+    cmd_id=Command.EQUIPMENT_CONTROL,
+    attribute_transformer=None,
+    additional_attributes={"device_ID": 0, "device_status": "01000000"},
+    receive_types=[Command.ANSWER_YES_OR_NO],
+    content_field="answer_yes_or_no",
+    content_sync_finished=2,
+    content_transformer=None,
+)
+
+SOCKET_ON = CommandAttributes(
+    cmd_id=Command.EQUIPMENT_CONTROL,
+    attribute_transformer=None,
+    additional_attributes={"device_ID": 0, "device_status": "01010000"},
+    receive_types=[Command.ANSWER_YES_OR_NO],
+    content_field="answer_yes_or_no",
+    content_sync_finished=2,
+    content_transformer=None,
+)
+
 # GET_SCENES returns a dict[{scene_id}, None]
 # NOTE: If queried frequently not all data is provisioned all the time
 GET_SCENES = CommandAttributes(

elro/device.py

@@ -61,6 +61,7 @@ def __repr__(self):
 ATTR_BATTERY_LEVEL = "battery"
 ATTR_DEVICE_TYPE = "device_type"
 ATTR_DEVICE_STATE = "device_state"
+ATTR_DEVICE_VALUE = "device_value"
 ATTR_SIGNAL = "signal"
 
 STATE_ALARM = "ALARM"
@@ -75,6 +76,9 @@ def __repr__(self):
 STATES_ON = (STATE_ALARM, STATE_FIRE_ALARM, STATE_TEST_ALARM)
 STATES_OFFLINE = (STATE_OFFLINE,)
 
+DEVICE_VALUE_OFF = "off"
+DEVICE_VALUE_ON = "on"
+
 # Represent the state of a device.
 DEVICE_STATE = {
     "01": STATE_NORMAL,
@@ -91,3 +95,8 @@ def __repr__(self):
     "FE": STATE_UNKNOWN,
     "FF": STATE_OFFLINE,
 }
+
+DEVICE_VALUE = {
+    0: DEVICE_VALUE_OFF,
+    1: DEVICE_VALUE_ON,
+}

elro/utils.py

@@ -7,7 +7,7 @@
 import collections
 from typing import Any
 
-from elro.device import DeviceType, DEVICE_STATE
+from elro.device import DEVICE_VALUE, DeviceType, DEVICE_STATE
 
 
 # From the ByteUtil class, needed by CRC_maker
@@ -281,15 +281,19 @@ def get_device_states(content: list) -> dict[str, Any]:
             # Unsupported record, skip and continue silently
             continue
         device_state = hexdata["device_status"][4:6]
+        device_value_data = int(hexdata["device_status"][6:8], 16)
         return_dict[hexdata["device_ID"]] = {
             "device_type": device_type,
             "signal": int(hexdata["device_status"][0:2], 16),
             "battery": int(hexdata["device_status"][2:4], 16),
             "device_state": DEVICE_STATE.get(
                 device_state, device_state
             ),  # return hex device state if it is not known
+            "device_value": DEVICE_VALUE.get(
+                device_value_data, hex(device_value_data)
+            ),  # return hex device value if it is not known
             "device_status_data": hexdata,
-            "device_value_data": int(hexdata["device_status"][6:8], 16),
+            "device_value_data": device_value_data,
         }
     return return_dict
 

examples/socket_demo.py

@@ -64,12 +64,12 @@ async def async_test_socket(self, device_id: int, command_code_hex: str) -> None
 
 
 if __name__ == "__main__":
-    # argv: 1 = IP_ADDRESS, 2 = API_KEY, 3 = device_id, 4 = command_hex
+    # argv: 1=IP_ADDRESS, 2=API_KEY, 3=device_id, 4=command_hex (0100=off, 0101=on)
     if len(sys.argv) == 1:
         print(
             f"Elro Connects socket test util.\n"
             f"Query only:\n{sys.argv[0]} ip-adress api-key\n\n"
-            f"Execute a command:\n{sys.argv[0]} "
+            f"Execute a command (0100=off, 0101=on):\n{sys.argv[0]} "
             "ip-adress api-key device_id command_hex"
         )
         sys.exit(0)

tests/test_api.py

@@ -202,6 +202,38 @@ async def test_sync_device_status(mock_k1_connector):
     assert result[3]["device_state"] == "UNKNOWN"
 
 
+@pytest.mark.asyncio
+async def test_sync_socket_status(mock_k1_connector):
+    """Test get device status."""
+    await mock_k1_connector.async_connect()
+
+    help_mock_command_reply(mock_k1_connector, MOCK_SOCKET_STATUS_OFF_RESPONSE)
+
+    result = await mock_k1_connector.async_process_command(
+        SYN_DEVICE_STATUS, sence_group=0
+    )
+
+    assert result[1]["device_type"] == "SOCKET"
+    assert result[1]["signal"] == 4
+    assert result[1]["battery"] == 255
+    assert result[1]["device_state"] == "NORMAL"
+    assert result[1]["device_value_data"] == 0
+    assert result[1]["device_value"] == "off"
+
+    help_mock_command_reply(mock_k1_connector, MOCK_SOCKET_STATUS_ON_RESPONSE)
+
+    result = await mock_k1_connector.async_process_command(
+        SYN_DEVICE_STATUS, sence_group=0
+    )
+
+    assert result[1]["device_type"] == "SOCKET"
+    assert result[1]["signal"] == 4
+    assert result[1]["battery"] == 255
+    assert result[1]["device_state"] == "NORMAL"
+    assert result[1]["device_value_data"] == 1
+    assert result[1]["device_value"] == "on"
+
+
 @pytest.mark.asyncio
 async def test_api_access_properties(mock_k1_connector):
     """Test api access properties."""