From b1e9b201d1b022a2bc87ad068603d6f3e254696f Mon Sep 17 00:00:00 2001 From: Wolfgang Hottgenroth Date: Sun, 9 Nov 2025 13:26:55 +0100 Subject: [PATCH] transformation added 2 --- apps/abstraction/transformation.py | 201 +++++++++++++++++++++++------ 1 file changed, 164 insertions(+), 37 deletions(-) diff --git a/apps/abstraction/transformation.py b/apps/abstraction/transformation.py index 5a0ac48..cdadac8 100644 --- a/apps/abstraction/transformation.py +++ b/apps/abstraction/transformation.py @@ -1,46 +1,182 @@ """Payload transformation functions for vendor-specific device communication. -This module provides transformation functions to translate between abstract -home protocol payloads and vendor-specific device payloads. +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 +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. + + zigbee2mqtt uses 'state' instead of 'power': + - Abstract: {'power': 'on', 'brightness': 100} + - zigbee2mqtt: {'state': 'ON', 'brightness': 100} + """ + 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 + + return vendor_payload + + +def _transform_light_zigbee2mqtt_to_abstract(payload: dict[str, Any]) -> dict[str, Any]: + """Transform zigbee2mqtt light payload to abstract format. + + zigbee2mqtt uses 'state' instead of 'power': + - zigbee2mqtt: {'state': 'ON', 'brightness': 100} + - 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 + + 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 payload for SET commands. - - This function allows technology-specific transformations when sending commands - to devices. For example, different vendors might use different field names or - value formats for the same abstract concept. + """Transform abstract payload to vendor-specific format. Args: - device_type: Type of device (e.g., 'light', 'thermostat') - device_technology: Technology identifier (e.g., 'zigbee2mqtt', 'tasmota') - abstract_payload: Abstract payload following home protocol - + 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: - Vendor-specific payload for the device - - Example: - Input: {'power': 'on', 'brightness': 75} - Output: {'state': 'ON', 'brightness': 75} # hypothetical vendor format + Payload in vendor-specific format """ logger.debug( f"transform_abstract_to_vendor IN: type={device_type}, tech={device_technology}, " f"payload={abstract_payload}" ) - # TODO: Implement technology-specific transformations here - # Currently pass-through: return payload unchanged - vendor_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}, " @@ -54,32 +190,23 @@ def transform_vendor_to_abstract( device_technology: str, vendor_payload: dict[str, Any] ) -> dict[str, Any]: - """Transform vendor-specific payload to abstract payload for STATE messages. - - This function allows technology-specific transformations when receiving state - updates from devices. For example, different vendors might report state using - different field names or value formats. + """Transform vendor-specific payload to abstract format. Args: - device_type: Type of device (e.g., 'light', 'thermostat') - device_technology: Technology identifier (e.g., 'zigbee2mqtt', 'tasmota') - vendor_payload: Vendor-specific payload from the device - + 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: - Abstract payload following home protocol - - Example: - Input: {'state': 'ON', 'brightness': 75} # hypothetical vendor format - Output: {'power': 'on', 'brightness': 75} + Payload in abstract home protocol format """ logger.debug( f"transform_vendor_to_abstract IN: type={device_type}, tech={device_technology}, " f"payload={vendor_payload}" ) - # TODO: Implement technology-specific transformations here - # Currently pass-through: return payload unchanged - abstract_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}, "