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.
    
    zigbee2mqtt uses same format as abstract protocol (no transformation needed).
    """
    return payload


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


# ============================================================================
# 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,
}


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