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