home-automation/apps/abstraction/transformation.py

"""Payload transformation functions for vendor-specific device communication.

This module implements a registry-pattern for vendor-specific transformations:
- Each (device_type, technology, direction) tuple maps to a specific handler function
- Handlers transform payloads between abstract and vendor-specific formats
- Unknown combinations fall back to pass-through (no transformation)
"""

import logging
from typing import Any, Callable

logger = logging.getLogger(__name__)


# ============================================================================
# HANDLER FUNCTIONS: simulator technology
# ============================================================================

def _transform_light_simulator_to_vendor(payload: dict[str, Any]) -> dict[str, Any]:
    """Transform abstract light payload to simulator format.
    
    Simulator uses same format as abstract protocol (no transformation needed).
    """
    return payload


def _transform_light_simulator_to_abstract(payload: dict[str, Any]) -> dict[str, Any]:
    """Transform simulator light payload to abstract format.
    
    Simulator uses same format as abstract protocol (no transformation needed).
    """
    return payload


def _transform_thermostat_simulator_to_vendor(payload: dict[str, Any]) -> dict[str, Any]:
    """Transform abstract thermostat payload to simulator format.
    
    Simulator uses same format as abstract protocol (no transformation needed).
    """
    return payload


def _transform_thermostat_simulator_to_abstract(payload: dict[str, Any]) -> dict[str, Any]:
    """Transform simulator thermostat payload to abstract format.
    
    Simulator uses same format as abstract protocol (no transformation needed).
    """
    return payload


# ============================================================================
# HANDLER FUNCTIONS: zigbee2mqtt technology
# ============================================================================

def _transform_light_zigbee2mqtt_to_vendor(payload: dict[str, Any]) -> dict[str, Any]:
    """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)):
            # Convert percentage (0-100) to zigbee2mqtt range (0-254)
            vendor_payload["brightness"] = round(abstract_brightness * 254 / 100)
    
    return vendor_payload


def _transform_light_zigbee2mqtt_to_abstract(payload: dict[str, Any]) -> 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 = payload.copy()
    
    # 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)):
            # Convert zigbee2mqtt range (0-254) to percentage (0-100)
            abstract_payload["brightness"] = round(vendor_brightness * 100 / 254)
    
    return abstract_payload


def _transform_thermostat_zigbee2mqtt_to_vendor(payload: dict[str, Any]) -> dict[str, Any]:
    """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:
        # zigbee2mqtt expects current_heating_setpoint as string
        vendor_payload["current_heating_setpoint"] = str(payload["target"])
    
    return vendor_payload


def _transform_thermostat_zigbee2mqtt_to_abstract(payload: dict[str, Any]) -> 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'}
    """
    abstract_payload = {}
    
    # Extract target temperature
    if "current_heating_setpoint" in payload:
        setpoint = payload["current_heating_setpoint"]
        abstract_payload["target"] = float(setpoint)
    
    # Extract current temperature
    if "local_temperature" in payload:
        current = payload["local_temperature"]
        abstract_payload["current"] = float(current)
    
    # Extract mode
    if "system_mode" in payload:
        abstract_payload["mode"] = payload["system_mode"]
    
    return abstract_payload


# ============================================================================
# HANDLER FUNCTIONS: contact_sensor - zigbee2mqtt technology
# ============================================================================

def _transform_contact_sensor_zigbee2mqtt_to_vendor(payload: dict[str, Any]) -> dict[str, Any]:
    """Transform abstract contact sensor payload to zigbee2mqtt format.
    
    Contact sensors are read-only, so this should not be called for SET commands.
    Returns payload as-is for compatibility.
    """
    logger.warning("Contact sensors are read-only - SET commands should not be used")
    return payload


def _transform_contact_sensor_zigbee2mqtt_to_abstract(payload: dict[str, Any]) -> dict[str, Any]:
    """Transform zigbee2mqtt contact sensor payload to abstract format.
    
    Transformations:
    - contact: bool -> "open" | "closed"
      - zigbee2mqtt semantics: False = OPEN, True = CLOSED (inverted!)
    - battery: pass through (already 0-100)
    - linkquality: pass through
    - device_temperature: pass through (if present)
    - voltage: pass through (if present)
    
    Example:
    - zigbee2mqtt: {"contact": false, "battery": 100, "linkquality": 87}
    - Abstract: {"contact": "open", "battery": 100, "linkquality": 87}
    """
    abstract_payload = {}
    
    # Transform contact state (inverted logic!)
    if "contact" in payload:
        contact_bool = payload["contact"]
        # zigbee2mqtt: False = OPEN, True = CLOSED
        abstract_payload["contact"] = "closed" if contact_bool else "open"
    
    # Pass through optional fields
    if "battery" in payload:
        abstract_payload["battery"] = payload["battery"]
    
    if "linkquality" in payload:
        abstract_payload["linkquality"] = payload["linkquality"]
    
    if "device_temperature" in payload:
        abstract_payload["device_temperature"] = payload["device_temperature"]
    
    if "voltage" in payload:
        abstract_payload["voltage"] = payload["voltage"]
    
    return abstract_payload


# ============================================================================
# HANDLER FUNCTIONS: contact_sensor - max technology (Homegear MAX!)
# ============================================================================

def _transform_contact_sensor_max_to_vendor(payload: dict[str, Any]) -> dict[str, Any]:
    """Transform abstract contact sensor payload to MAX! format.
    
    Contact sensors are read-only, so this should not be called for SET commands.
    Returns payload as-is for compatibility.
    """
    logger.warning("Contact sensors are read-only - SET commands should not be used")
    return payload


def _transform_contact_sensor_max_to_abstract(payload: str | bool | dict[str, Any]) -> dict[str, Any]:
    """Transform MAX! (Homegear) contact sensor payload to abstract format.
    
    MAX! sends "true"/"false" (string or bool) on STATE topic.
    
    Transformations:
    - "true" or True -> "open" (window/door open)
    - "false" or False -> "closed" (window/door closed)
    
    Example:
    - MAX!: "true" or True
    - Abstract: {"contact": "open"}
    """
    try:
        # Handle string, bool, or dict input
        if isinstance(payload, dict):
            # If already a dict, extract contact field
            contact_value = payload.get("contact", False)
        elif isinstance(payload, str):
            # Parse string to bool
            contact_value = payload.strip().lower() == "true"
        elif isinstance(payload, bool):
            # Use bool directly
            contact_value = payload
        else:
            logger.warning(f"MAX! contact sensor unexpected payload type: {type(payload)}, value: {payload}")
            contact_value = False
        
        # MAX! semantics: True = OPEN, False = CLOSED
        return {
            "contact": "open" if contact_value else "closed"
        }
    except (ValueError, TypeError) as e:
        logger.error(f"MAX! contact sensor failed to parse: {payload}, error: {e}")
        return {
            "contact": "closed"  # Default to closed on error
        }


# ============================================================================
# HANDLER FUNCTIONS: max technology (Homegear MAX!)
# ============================================================================

def _transform_thermostat_max_to_vendor(payload: dict[str, Any]) -> str:
    """Transform abstract thermostat payload to MAX! (Homegear) format.
    
    MAX! expects only the integer temperature value (no JSON).
    
    Transformations:
    - Extract 'target' temperature from payload
    - Convert float to integer (MAX! only accepts integers)
    - Return as plain string value
    
    Example:
    - Abstract: {'mode': 'heat', 'target': 22.5}
    - MAX!: "22"
    
    Note: MAX! ignores mode - it's always in heating mode
    """
    if "target" not in payload:
        logger.warning(f"MAX! thermostat payload missing 'target': {payload}")
        return "21"  # Default fallback
    
    target_temp = payload["target"]
    
    # Convert to integer (MAX! protocol requirement)
    if isinstance(target_temp, (int, float)):
        int_temp = int(round(target_temp))
        return str(int_temp)
    
    logger.warning(f"MAX! invalid target temperature type: {type(target_temp)}, value: {target_temp}")
    return "21"


def _transform_thermostat_max_to_abstract(payload: str | int | float) -> dict[str, Any]:
    """Transform MAX! (Homegear) thermostat payload to abstract format.
    
    MAX! sends only the integer temperature value (no JSON).
    
    Transformations:
    - Parse plain string/int value
    - Convert to float for abstract protocol
    - Wrap in abstract payload structure with mode='heat'
    
    Example:
    - MAX!: "22" or 22
    - Abstract: {'target': 22.0, 'mode': 'heat'}
    
    Note: MAX! doesn't send current temperature via SET_TEMPERATURE topic
    """
    try:
        # Handle both string and numeric input
        if isinstance(payload, str):
            target_temp = float(payload.strip())
        elif isinstance(payload, (int, float)):
            target_temp = float(payload)
        else:
            logger.warning(f"MAX! unexpected payload type: {type(payload)}, value: {payload}")
            target_temp = 21.0
        
        return {
            "target": target_temp,
            "mode": "heat"  # MAX! is always in heating mode
        }
    except (ValueError, TypeError) as e:
        logger.error(f"MAX! failed to parse temperature: {payload}, error: {e}")
        return {
            "target": 21.0,
            "mode": "heat"
        }


# ============================================================================
# REGISTRY: Maps (device_type, technology, direction) -> handler function
# ============================================================================

TransformHandler = Callable[[dict[str, Any]], dict[str, Any]]

TRANSFORM_HANDLERS: dict[tuple[str, str, str], TransformHandler] = {
    # Light transformations
    ("light", "simulator", "to_vendor"): _transform_light_simulator_to_vendor,
    ("light", "simulator", "to_abstract"): _transform_light_simulator_to_abstract,
    ("light", "zigbee2mqtt", "to_vendor"): _transform_light_zigbee2mqtt_to_vendor,
    ("light", "zigbee2mqtt", "to_abstract"): _transform_light_zigbee2mqtt_to_abstract,
    
    # Thermostat transformations
    ("thermostat", "simulator", "to_vendor"): _transform_thermostat_simulator_to_vendor,
    ("thermostat", "simulator", "to_abstract"): _transform_thermostat_simulator_to_abstract,
    ("thermostat", "zigbee2mqtt", "to_vendor"): _transform_thermostat_zigbee2mqtt_to_vendor,
    ("thermostat", "zigbee2mqtt", "to_abstract"): _transform_thermostat_zigbee2mqtt_to_abstract,
    ("thermostat", "max", "to_vendor"): _transform_thermostat_max_to_vendor,
    ("thermostat", "max", "to_abstract"): _transform_thermostat_max_to_abstract,
    
    # Contact sensor transformations (support both 'contact' and 'contact_sensor' types)
    ("contact_sensor", "zigbee2mqtt", "to_vendor"): _transform_contact_sensor_zigbee2mqtt_to_vendor,
    ("contact_sensor", "zigbee2mqtt", "to_abstract"): _transform_contact_sensor_zigbee2mqtt_to_abstract,
    ("contact_sensor", "max", "to_vendor"): _transform_contact_sensor_max_to_vendor,
    ("contact_sensor", "max", "to_abstract"): _transform_contact_sensor_max_to_abstract,
    ("contact", "zigbee2mqtt", "to_vendor"): _transform_contact_sensor_zigbee2mqtt_to_vendor,
    ("contact", "zigbee2mqtt", "to_abstract"): _transform_contact_sensor_zigbee2mqtt_to_abstract,
    ("contact", "max", "to_vendor"): _transform_contact_sensor_max_to_vendor,
    ("contact", "max", "to_abstract"): _transform_contact_sensor_max_to_abstract,
}


def _get_transform_handler(
    device_type: str,
    device_technology: str,
    direction: str
) -> TransformHandler:
    """Get transformation handler for given device type, technology and direction.
    
    Args:
        device_type: Type of device (e.g., "light", "thermostat")
        device_technology: Technology/vendor (e.g., "simulator", "zigbee2mqtt")
        direction: Transformation direction ("to_vendor" or "to_abstract")
    
    Returns:
        Handler function for transformation, or pass-through if not found
    """
    key = (device_type, device_technology, direction)
    handler = TRANSFORM_HANDLERS.get(key)
    
    if handler is None:
        logger.warning(
            f"No transformation handler for {key}, using pass-through. "
            f"Available: {list(TRANSFORM_HANDLERS.keys())}"
        )
        return lambda payload: payload  # Pass-through fallback
    
    return handler


# ============================================================================
# PUBLIC API: Main transformation functions
# ============================================================================

def transform_abstract_to_vendor(
    device_type: str,
    device_technology: str,
    abstract_payload: dict[str, Any]
) -> dict[str, Any]:
    """Transform abstract payload to vendor-specific format.
    
    Args:
        device_type: Type of device (e.g., "light", "thermostat")
        device_technology: Technology/vendor (e.g., "simulator", "zigbee2mqtt")
        abstract_payload: Payload in abstract home protocol format
    
    Returns:
        Payload in vendor-specific format
    """
    logger.debug(
        f"transform_abstract_to_vendor IN: type={device_type}, tech={device_technology}, "
        f"payload={abstract_payload}"
    )
    
    handler = _get_transform_handler(device_type, device_technology, "to_vendor")
    vendor_payload = handler(abstract_payload)
    
    logger.debug(
        f"transform_abstract_to_vendor OUT: type={device_type}, tech={device_technology}, "
        f"payload={vendor_payload}"
    )
    return vendor_payload


def transform_vendor_to_abstract(
    device_type: str,
    device_technology: str,
    vendor_payload: dict[str, Any]
) -> dict[str, Any]:
    """Transform vendor-specific payload to abstract format.
    
    Args:
        device_type: Type of device (e.g., "light", "thermostat")
        device_technology: Technology/vendor (e.g., "simulator", "zigbee2mqtt")
        vendor_payload: Payload in vendor-specific format
    
    Returns:
        Payload in abstract home protocol format
    """
    logger.debug(
        f"transform_vendor_to_abstract IN: type={device_type}, tech={device_technology}, "
        f"payload={vendor_payload}"
    )
    
    handler = _get_transform_handler(device_type, device_technology, "to_abstract")
    abstract_payload = handler(vendor_payload)
    
    logger.debug(
        f"transform_vendor_to_abstract OUT: type={device_type}, tech={device_technology}, "
        f"payload={abstract_payload}"
    )
    return abstract_payload