home-automation/apps/abstraction/vendors/zigbee2mqtt.py

"""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,
}