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)

Vendor-specific implementations are in the vendors/ subdirectory.
"""

import logging
from typing import Any, Callable

from apps.abstraction.vendors import (
    simulator,
    zigbee2mqtt,
    max,
    shelly,
    tasmota,
    hottis_pv_modbus,
    hottis_wago_modbus,
    hottis_wifi_relay,
    hottis_led_stripe
)

logger = logging.getLogger(__name__)


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

TransformHandler = Callable[[Any], Any]

# Build registry from vendor modules
TRANSFORM_HANDLERS: dict[tuple[str, str, str], TransformHandler] = {}

# Register handlers from each vendor module
for vendor_name, vendor_module in [
    ("simulator", simulator),
    ("zigbee2mqtt", zigbee2mqtt),
    ("max", max),
    ("shelly", shelly),
    ("tasmota", tasmota),
    ("hottis_pv_modbus", hottis_pv_modbus),
    ("hottis_wago_modbus", hottis_wago_modbus),
    ("hottis_wifi_relay", hottis_wifi_relay),
    ("hottis_led_stripe", hottis_led_stripe),
]:
    for (device_type, direction), handler in vendor_module.HANDLERS.items():
        key = (device_type, vendor_name, direction)
        TRANSFORM_HANDLERS[key] = handler


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]
) -> str:
    """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 (as string)
    """
    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: str
) -> 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 (as string)
    
    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