vendor transformations separated

apps/abstraction/vendors/zigbee2mqtt.py

@@ -0,0 +1,209 @@
+"""Zigbee2MQTT vendor transformations."""
+
+import json
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+def transform_light_to_vendor(payload: dict[str, Any]) -> str:
+    """Transform abstract light payload to zigbee2mqtt format.
+    
+    Transformations:
+    - power: 'on'/'off' -> state: 'ON'/'OFF'
+    - brightness: 0-100 -> brightness: 0-254
+    
+    Example:
+    - Abstract: {'power': 'on', 'brightness': 100}
+    - zigbee2mqtt: {'state': 'ON', 'brightness': 254}
+    """
+    vendor_payload = payload.copy()
+    
+    # Transform power -> state with uppercase values
+    if "power" in vendor_payload:
+        power_value = vendor_payload.pop("power")
+        vendor_payload["state"] = power_value.upper() if isinstance(power_value, str) else power_value
+    
+    # Transform brightness: 0-100 (%) -> 0-254 (zigbee2mqtt range)
+    if "brightness" in vendor_payload:
+        abstract_brightness = vendor_payload["brightness"]
+        if isinstance(abstract_brightness, (int, float)):
+            vendor_payload["brightness"] = round(abstract_brightness * 254 / 100)
+    
+    return json.dumps(vendor_payload)
+
+
+def transform_light_to_abstract(payload: str) -> dict[str, Any]:
+    """Transform zigbee2mqtt light payload to abstract format.
+    
+    Transformations:
+    - state: 'ON'/'OFF' -> power: 'on'/'off'
+    - brightness: 0-254 -> brightness: 0-100
+    
+    Example:
+    - zigbee2mqtt: {'state': 'ON', 'brightness': 254}
+    - Abstract: {'power': 'on', 'brightness': 100}
+    """
+    abstract_payload = json.loads(payload)
+    
+    # Transform state -> power with lowercase values
+    if "state" in abstract_payload:
+        state_value = abstract_payload.pop("state")
+        abstract_payload["power"] = state_value.lower() if isinstance(state_value, str) else state_value
+    
+    # Transform brightness: 0-254 (zigbee2mqtt range) -> 0-100 (%)
+    if "brightness" in abstract_payload:
+        vendor_brightness = abstract_payload["brightness"]
+        if isinstance(vendor_brightness, (int, float)):
+            abstract_payload["brightness"] = round(vendor_brightness * 100 / 254)
+    
+    return abstract_payload
+
+
+def transform_thermostat_to_vendor(payload: dict[str, Any]) -> str:
+    """Transform abstract thermostat payload to zigbee2mqtt format.
+    
+    Transformations:
+    - target -> current_heating_setpoint (as string)
+    - mode is ignored (zigbee2mqtt thermostats use system_mode in state only)
+    
+    Example:
+    - Abstract: {'target': 22.0}
+    - zigbee2mqtt: {'current_heating_setpoint': '22.0'}
+    """
+    vendor_payload = {}
+    
+    if "target" in payload:
+        vendor_payload["current_heating_setpoint"] = str(payload["target"])
+    
+    return json.dumps(vendor_payload)
+
+
+def transform_thermostat_to_abstract(payload: str) -> dict[str, Any]:
+    """Transform zigbee2mqtt thermostat payload to abstract format.
+    
+    Transformations:
+    - current_heating_setpoint -> target (as float)
+    - local_temperature -> current (as float)
+    - system_mode -> mode
+    
+    Example:
+    - zigbee2mqtt: {'current_heating_setpoint': 15, 'local_temperature': 23, 'system_mode': 'heat'}
+    - Abstract: {'target': 15.0, 'current': 23.0, 'mode': 'heat'}
+    """
+    payload = json.loads(payload)
+    abstract_payload = {}
+    
+    if "current_heating_setpoint" in payload:
+        setpoint = payload["current_heating_setpoint"]
+        abstract_payload["target"] = float(setpoint)
+    
+    if "local_temperature" in payload:
+        current = payload["local_temperature"]
+        abstract_payload["current"] = float(current)
+    
+    if "system_mode" in payload:
+        abstract_payload["mode"] = payload["system_mode"]
+    
+    return abstract_payload
+
+
+def transform_contact_sensor_to_vendor(payload: dict[str, Any]) -> str:
+    """Transform abstract contact sensor payload to zigbee2mqtt format.
+    
+    Contact sensors are read-only, so this should not be called for SET commands.
+    """
+    logger.warning("Contact sensors are read-only - SET commands should not be used")
+    return json.dumps(payload)
+
+
+def transform_contact_sensor_to_abstract(payload: str) -> dict[str, Any]:
+    """Transform zigbee2mqtt contact sensor payload to abstract format.
+    
+    Transformations:
+    - contact: bool -> "open" | "closed"
+      - zigbee2mqtt semantics: False = OPEN, True = CLOSED (inverted!)
+    
+    Example:
+    - zigbee2mqtt: {"contact": false, "battery": 100}
+    - Abstract: {"contact": "open", "battery": 100}
+    """
+    payload = json.loads(payload)
+    abstract_payload = {}
+    
+    if "contact" in payload:
+        contact_bool = payload["contact"]
+        abstract_payload["contact"] = "closed" if contact_bool else "open"
+    
+    # Pass through optional fields
+    for field in ["battery", "linkquality", "device_temperature", "voltage"]:
+        if field in payload:
+            abstract_payload[field] = payload[field]
+    
+    return abstract_payload
+
+
+def transform_temp_humidity_sensor_to_vendor(payload: dict[str, Any]) -> str:
+    """Transform abstract temp/humidity sensor payload to zigbee2mqtt format.
+    
+    Temp/humidity sensors are read-only.
+    """
+    return json.dumps(payload)
+
+
+def transform_temp_humidity_sensor_to_abstract(payload: str) -> dict[str, Any]:
+    """Transform zigbee2mqtt temp/humidity sensor payload to abstract format.
+    
+    Passthrough - zigbee2mqtt provides temperature, humidity, battery, linkquality directly.
+    """
+    payload = json.loads(payload)
+    return payload
+
+
+def transform_relay_to_vendor(payload: dict[str, Any]) -> str:
+    """Transform abstract relay payload to zigbee2mqtt format.
+    
+    - power: 'on'/'off' -> state: 'ON'/'OFF'
+    """
+    vendor_payload = payload.copy()
+    
+    if "power" in vendor_payload:
+        power_value = vendor_payload.pop("power")
+        vendor_payload["state"] = power_value.upper() if isinstance(power_value, str) else power_value
+    
+    return json.dumps(vendor_payload)
+
+
+def transform_relay_to_abstract(payload: str) -> dict[str, Any]:
+    """Transform zigbee2mqtt relay payload to abstract format.
+    
+    - state: 'ON'/'OFF' -> power: 'on'/'off'
+    """
+    payload = json.loads(payload)
+    abstract_payload = payload.copy()
+    
+    if "state" in abstract_payload:
+        state_value = abstract_payload.pop("state")
+        abstract_payload["power"] = state_value.lower() if isinstance(state_value, str) else state_value
+    
+    return abstract_payload
+
+
+# Registry of handlers for this vendor
+HANDLERS = {
+    ("light", "to_vendor"): transform_light_to_vendor,
+    ("light", "to_abstract"): transform_light_to_abstract,
+    ("thermostat", "to_vendor"): transform_thermostat_to_vendor,
+    ("thermostat", "to_abstract"): transform_thermostat_to_abstract,
+    ("contact_sensor", "to_vendor"): transform_contact_sensor_to_vendor,
+    ("contact_sensor", "to_abstract"): transform_contact_sensor_to_abstract,
+    ("contact", "to_vendor"): transform_contact_sensor_to_vendor,
+    ("contact", "to_abstract"): transform_contact_sensor_to_abstract,
+    ("temp_humidity_sensor", "to_vendor"): transform_temp_humidity_sensor_to_vendor,
+    ("temp_humidity_sensor", "to_abstract"): transform_temp_humidity_sensor_to_abstract,
+    ("temp_humidity", "to_vendor"): transform_temp_humidity_sensor_to_vendor,
+    ("temp_humidity", "to_abstract"): transform_temp_humidity_sensor_to_abstract,
+    ("relay", "to_vendor"): transform_relay_to_vendor,
+    ("relay", "to_abstract"): transform_relay_to_abstract,
+}