implement get-flow-for-speed and set-flow-for-speed (#15)

* Implement get-flow-for-speed method in lib and CLI

* reuse get_single_property

* implement set-flow-for-speed, make PdoTypes enum

* simplify TYPE_CN_BOOL conversion

* Bump dependency to Python 3.10 so we can use match

---------

Co-authored-by: Michaël Arnauts <[email protected]>

Author: iSOcH

.github/workflows/ci.yml

@@ -21,7 +21,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ ubuntu-latest ]
-        python-version: [ "3.8", "3.9", "3.10", "3.11" ]
+        python-version: [ "3.10", "3.11" ]
     steps:
       - name: Check out ${{ github.sha }} from repository ${{ github.repository }}
         uses: actions/checkout@v3

README.md

@@ -3,7 +3,7 @@
 `aiocomfoconnect` is an asyncio Python 3 library for communicating with a Zehnder ComfoAir Q350/450/600 ventilation system. It's the successor of
 [comfoconnect](https://github.com/michaelarnauts/comfoconnect).
 
-It's compatible with Python 3.8 and higher.
+It's compatible with Python 3.10 and higher.
 
 ## Installation
 

aiocomfoconnect/__main__.py

@@ -51,6 +51,12 @@ async def main(args):
     elif args.action == "get-property":
         await run_get_property(args.host, args.uuid, args.node_id, args.unit, args.subunit, args.property_id, args.property_type)
 
+    elif args.action == "get-flow-for-speed":
+        await run_get_flow_for_speed(args.host, args.uuid, args.speed)
+
+    elif args.action == "set-flow-for-speed":
+        await run_set_flow_for_speed(args.host, args.uuid, args.speed, args.flow)
+
     else:
         raise Exception("Unknown action: " + args.action)
 
@@ -310,6 +316,46 @@ async def run_get_property(host: str, uuid: str, node_id: int, unit: int, subuni
     await comfoconnect.disconnect()
 
 
+async def run_get_flow_for_speed(host: str, uuid: str, speed: Literal["away", "low", "medium", "high"]):
+    """Get the configured airflow for the specified speed."""
+    # Discover bridge so we know the UUID
+    bridges = await discover_bridges(host)
+    if not bridges:
+        raise Exception("No bridge found")
+
+    # Connect to the bridge
+    comfoconnect = ComfoConnect(bridges[0].host, bridges[0].uuid)
+    try:
+        await comfoconnect.connect(uuid)
+    except ComfoConnectNotAllowed:
+        print("Could not connect to bridge. Please register first.")
+        sys.exit(1)
+
+    print(await comfoconnect.get_flow_for_speed(speed))
+
+    await comfoconnect.disconnect()
+
+
+async def run_set_flow_for_speed(host: str, uuid: str, speed: Literal["away", "low", "medium", "high"], desired_flow: int):
+    """Set the configured airflow for the specified speed."""
+    # Discover bridge so we know the UUID
+    bridges = await discover_bridges(host)
+    if not bridges:
+        raise Exception("No bridge found")
+
+    # Connect to the bridge
+    comfoconnect = ComfoConnect(bridges[0].host, bridges[0].uuid)
+    try:
+        await comfoconnect.connect(uuid)
+    except ComfoConnectNotAllowed:
+        print("Could not connect to bridge. Please register first.")
+        sys.exit(1)
+
+    await comfoconnect.set_flow_for_speed(speed, desired_flow)
+
+    await comfoconnect.disconnect()
+
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()
     parser.add_argument("--debug", "-d", help="Enable debug logging", default=False, action="store_true")
@@ -365,6 +411,17 @@ async def run_get_property(host: str, uuid: str, node_id: int, unit: int, subuni
     p_sensor.add_argument("--host", help="Host address of the bridge")
     p_sensor.add_argument("--uuid", help="UUID of this app", default=DEFAULT_UUID)
 
+    p_get_flow_speed = subparsers.add_parser("get-flow-for-speed", help="Get m³/h for given speed")
+    p_get_flow_speed.add_argument("speed", help="Fan speed", choices=["low", "medium", "high", "away"])
+    p_get_flow_speed.add_argument("--host", help="Host address of the bridge")
+    p_get_flow_speed.add_argument("--uuid", help="UUID of this app", default=DEFAULT_UUID)
+
+    p_set_flow_speed = subparsers.add_parser("set-flow-for-speed", help="Set m³/h for given speed")
+    p_set_flow_speed.add_argument("speed", help="Fan speed", choices=["low", "medium", "high", "away"])
+    p_set_flow_speed.add_argument("flow", help="Desired airflow in m³/h", type=int)
+    p_set_flow_speed.add_argument("--host", help="Host address of the bridge")
+    p_set_flow_speed.add_argument("--uuid", help="UUID of this app", default=DEFAULT_UUID)
+
     arguments = parser.parse_args()
 
     if arguments.debug:

aiocomfoconnect/comfoconnect.py

@@ -15,17 +15,11 @@
     SUBUNIT_06,
     SUBUNIT_07,
     SUBUNIT_08,
-    TYPE_CN_BOOL,
-    TYPE_CN_INT8,
-    TYPE_CN_INT16,
-    TYPE_CN_INT64,
-    TYPE_CN_STRING,
-    TYPE_CN_UINT8,
-    TYPE_CN_UINT16,
-    TYPE_CN_UINT32,
+    PdoType,
     UNIT_ERROR,
     UNIT_SCHEDULE,
     UNIT_TEMPHUMCONTROL,
+    UNIT_VENTILATIONCONFIG,
     VentilationBalance,
     VentilationMode,
     VentilationSetting,
@@ -34,7 +28,7 @@
 )
 from aiocomfoconnect.properties import Property
 from aiocomfoconnect.sensors import Sensor
-from aiocomfoconnect.util import bytearray_to_bits, bytestring
+from aiocomfoconnect.util import bytearray_to_bits, bytestring, encode_pdo_value
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -102,13 +96,13 @@ async def get_single_property(self, unit: int, subunit: int, property_id: int, p
         """Get a property and convert to the right type."""
         result = await self.cmd_rmi_request(bytes([0x01, unit, subunit, 0x10, property_id]), node_id=node_id)
 
-        if property_type == TYPE_CN_STRING:
+        if property_type == PdoType.TYPE_CN_STRING:
             return result.message.decode("utf-8").rstrip("\x00")
-        if property_type in [TYPE_CN_INT8, TYPE_CN_INT16, TYPE_CN_INT64]:
+        if property_type in [PdoType.TYPE_CN_INT8, PdoType.TYPE_CN_INT16, PdoType.TYPE_CN_INT64]:
             return int.from_bytes(result.message, byteorder="little", signed=True)
-        if property_type in [TYPE_CN_UINT8, TYPE_CN_UINT16, TYPE_CN_UINT32]:
+        if property_type in [PdoType.TYPE_CN_UINT8, PdoType.TYPE_CN_UINT16, PdoType.TYPE_CN_UINT32]:
             return int.from_bytes(result.message, byteorder="little", signed=False)
-        if property_type in [TYPE_CN_BOOL]:
+        if property_type in [PdoType.TYPE_CN_BOOL]:
             return result.message[0] == 1
 
         return result.message
@@ -125,6 +119,15 @@ async def set_property(self, unit: int, subunit: int, property_id: int, value: i
 
         return result.message
 
+    async def set_property_typed(self, unit: int, subunit: int, property_id: int, value: int, pdo_type: PdoType, node_id=1) -> any:
+        """Set a typed property."""
+        value_bytes = encode_pdo_value(value, pdo_type)
+        message_bytes = bytes([0x03, unit, subunit, property_id]) + value_bytes
+
+        result = await self.cmd_rmi_request(message_bytes, node_id=node_id)
+
+        return result.message
+
     def _sensor_callback(self, sensor_id, sensor_value):
         """Callback function for sensor updates."""
         if self._sensor_callback_fn is None:
@@ -212,6 +215,36 @@ async def set_speed(self, speed: Literal["away", "low", "medium", "high"]):
         else:
             raise ValueError(f"Invalid speed: {speed}")
 
+    async def get_flow_for_speed(self, speed: Literal["away", "low", "medium", "high"]) -> int:
+        """Get the targeted airflow in m³/h for the given VentilationSpeed (away / low / medium / high)."""
+
+        match speed:
+            case VentilationSpeed.AWAY:
+                property_id = 3
+            case VentilationSpeed.LOW:
+                property_id = 4
+            case VentilationSpeed.MEDIUM:
+                property_id = 5
+            case VentilationSpeed.HIGH:
+                property_id = 6
+
+        return await self.get_single_property(UNIT_VENTILATIONCONFIG, SUBUNIT_01, property_id, PdoType.TYPE_CN_INT16)
+
+    async def set_flow_for_speed(self, speed: Literal["away", "low", "medium", "high"], desired_flow: int):
+        """Set the targeted airflow in m³/h for the given VentilationSpeed (away / low / medium / high)."""
+
+        match speed:
+            case VentilationSpeed.AWAY:
+                property_id = 3
+            case VentilationSpeed.LOW:
+                property_id = 4
+            case VentilationSpeed.MEDIUM:
+                property_id = 5
+            case VentilationSpeed.HIGH:
+                property_id = 6
+
+        await self.set_property_typed(UNIT_VENTILATIONCONFIG, SUBUNIT_01, property_id, desired_flow, PdoType.TYPE_CN_INT16)
+
     async def get_bypass(self):
         """Get the bypass mode (auto / on / off)."""
         result = await self.cmd_rmi_request(bytes([0x83, UNIT_SCHEDULE, SUBUNIT_02, 0x01]))

aiocomfoconnect/const.py

@@ -1,16 +1,19 @@
 """ Constants """
 
 # PDO Types
-TYPE_CN_BOOL = 0x00
-TYPE_CN_UINT8 = 0x01
-TYPE_CN_UINT16 = 0x02
-TYPE_CN_UINT32 = 0x03
-TYPE_CN_INT8 = 0x05
-TYPE_CN_INT16 = 0x06
-TYPE_CN_INT64 = 0x08
-TYPE_CN_STRING = 0x09
-TYPE_CN_TIME = 0x10
-TYPE_CN_VERSION = 0x11
+class PdoType:
+    """Defines a PDO type."""
+
+    TYPE_CN_BOOL = 0x00
+    TYPE_CN_UINT8 = 0x01
+    TYPE_CN_UINT16 = 0x02
+    TYPE_CN_UINT32 = 0x03
+    TYPE_CN_INT8 = 0x05
+    TYPE_CN_INT16 = 0x06
+    TYPE_CN_INT64 = 0x08
+    TYPE_CN_STRING = 0x09
+    TYPE_CN_TIME = 0x10
+    TYPE_CN_VERSION = 0x11
 
 
 # ComfoConnect Units

aiocomfoconnect/properties.py

@@ -4,8 +4,7 @@
 from dataclasses import dataclass
 
 from .const import (
-    TYPE_CN_STRING,
-    TYPE_CN_UINT32,
+    PdoType,
     UNIT_NODE,
     UNIT_NODECONFIGURATION,
     UNIT_TEMPHUMCONTROL,
@@ -22,15 +21,15 @@ class Property:
     property_type: int
 
 
-PROPERTY_SERIAL_NUMBER = Property(UNIT_NODE, 0x01, 0x04, TYPE_CN_STRING)
-PROPERTY_FIRMWARE_VERSION = Property(UNIT_NODE, 0x01, 0x06, TYPE_CN_UINT32)
-PROPERTY_MODEL = Property(UNIT_NODE, 0x01, 0x08, TYPE_CN_STRING)
-PROPERTY_ARTICLE = Property(UNIT_NODE, 0x01, 0x0B, TYPE_CN_STRING)
-PROPERTY_COUNTRY = Property(UNIT_NODE, 0x01, 0x0D, TYPE_CN_STRING)
-PROPERTY_NAME = Property(UNIT_NODE, 0x01, 0x14, TYPE_CN_STRING)
+PROPERTY_SERIAL_NUMBER = Property(UNIT_NODE, 0x01, 0x04, PdoType.TYPE_CN_STRING)
+PROPERTY_FIRMWARE_VERSION = Property(UNIT_NODE, 0x01, 0x06, PdoType.TYPE_CN_UINT32)
+PROPERTY_MODEL = Property(UNIT_NODE, 0x01, 0x08, PdoType.TYPE_CN_STRING)
+PROPERTY_ARTICLE = Property(UNIT_NODE, 0x01, 0x0B, PdoType.TYPE_CN_STRING)
+PROPERTY_COUNTRY = Property(UNIT_NODE, 0x01, 0x0D, PdoType.TYPE_CN_STRING)
+PROPERTY_NAME = Property(UNIT_NODE, 0x01, 0x14, PdoType.TYPE_CN_STRING)
 
-PROPERTY_MAINTAINER_PASSWORD = Property(UNIT_NODECONFIGURATION, 0x01, 0x03, TYPE_CN_STRING)
+PROPERTY_MAINTAINER_PASSWORD = Property(UNIT_NODECONFIGURATION, 0x01, 0x03, PdoType.TYPE_CN_STRING)
 
-PROPERTY_SENSOR_VENTILATION_TEMP_PASSIVE = Property(UNIT_TEMPHUMCONTROL, 0x01, 0x04, TYPE_CN_UINT32)
-PROPERTY_SENSOR_VENTILATION_HUMIDITY_COMFORT = Property(UNIT_TEMPHUMCONTROL, 0x01, 0x06, TYPE_CN_UINT32)
-PROPERTY_SENSOR_VENTILATION_HUMIDITY_PROTECTION = Property(UNIT_TEMPHUMCONTROL, 0x01, 0x07, TYPE_CN_UINT32)
+PROPERTY_SENSOR_VENTILATION_TEMP_PASSIVE = Property(UNIT_TEMPHUMCONTROL, 0x01, 0x04, PdoType.TYPE_CN_UINT32)
+PROPERTY_SENSOR_VENTILATION_HUMIDITY_COMFORT = Property(UNIT_TEMPHUMCONTROL, 0x01, 0x06, PdoType.TYPE_CN_UINT32)
+PROPERTY_SENSOR_VENTILATION_HUMIDITY_PROTECTION = Property(UNIT_TEMPHUMCONTROL, 0x01, 0x07, PdoType.TYPE_CN_UINT32)