From 9629850ebbbeac9412eea7ada803d6ea7a4eafce Mon Sep 17 00:00:00 2001 From: Wolfgang Hottgenroth Date: Mon, 8 Dec 2025 16:48:23 +0100 Subject: [PATCH] vendor transformations separated 2 --- apps/abstraction/transformation_old.py | 712 ------------------------- 1 file changed, 712 deletions(-) delete mode 100644 apps/abstraction/transformation_old.py diff --git a/apps/abstraction/transformation_old.py b/apps/abstraction/transformation_old.py deleted file mode 100644 index 146d356..0000000 --- a/apps/abstraction/transformation_old.py +++ /dev/null @@ -1,712 +0,0 @@ -"""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 -import json -from typing import Any, Callable - -logger = logging.getLogger(__name__) - - -# ============================================================================ -# HANDLER FUNCTIONS: simulator technology -# ============================================================================ - -def _transform_light_simulator_to_vendor(payload: dict[str, Any]) -> str: - """Transform abstract light payload to simulator format. - - Simulator uses same format as abstract protocol (no transformation needed). - """ - return json.dumps(payload) - - -def _transform_light_simulator_to_abstract(payload: str) -> dict[str, Any]: - """Transform simulator light payload to abstract format. - - Simulator uses same format as abstract protocol (no transformation needed). - """ - - payload = json.loads(payload) - - return payload - - -def _transform_thermostat_simulator_to_vendor(payload: dict[str, Any]) -> str: - """Transform abstract thermostat payload to simulator format. - - Simulator uses same format as abstract protocol (no transformation needed). - """ - return json.dumps(payload) - - -def _transform_thermostat_simulator_to_abstract(payload: str) -> dict[str, Any]: - """Transform simulator thermostat payload to abstract format. - - Simulator uses same format as abstract protocol (no transformation needed). - """ - - payload = json.loads(payload) - - return payload - - -# ============================================================================ -# HANDLER FUNCTIONS: zigbee2mqtt technology -# ============================================================================ - -def _transform_light_zigbee2mqtt_to_vendor(payload: dict[str, Any]) -> str: - """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 json.dumps(vendor_payload) - - -def _transform_light_zigbee2mqtt_to_abstract(payload: str) -> 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 = json.loads(payload) - - # 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]) -> str: - """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 json.dumps(vendor_payload) - - -def _transform_thermostat_zigbee2mqtt_to_abstract(payload: str) -> 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'} - """ - payload = json.loads(payload) - 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]) -> str: - """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 json.dumps(payload) - - -def _transform_contact_sensor_zigbee2mqtt_to_abstract(payload: str) -> 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} - """ - payload = json.loads(payload) - 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]) -> str: - """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 json.dumps(payload) - - -def _transform_contact_sensor_max_to_abstract(payload: str) -> 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: - contact_value = payload.strip().lower() == "true" - - # 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: temp_humidity_sensor - zigbee2mqtt technology -# ============================================================================ - -def _transform_temp_humidity_sensor_zigbee2mqtt_to_vendor(payload: dict[str, Any]) -> str: - """Transform abstract temp/humidity sensor payload to zigbee2mqtt format. - - Temp/humidity sensors are read-only, so this should not be called for SET commands. - Returns payload as-is for compatibility. - """ - return json.dumps(payload) - - -def _transform_temp_humidity_sensor_zigbee2mqtt_to_abstract(payload: str) -> dict[str, Any]: - """Transform zigbee2mqtt temp/humidity sensor payload to abstract format. - - Passthrough - zigbee2mqtt provides temperature, humidity, battery, linkquality directly. - """ - payload = json.loads(payload) - return payload - - - - -# ============================================================================ -# HANDLER FUNCTIONS: relay - zigbee2mqtt technology -# ============================================================================ - -def _transform_relay_zigbee2mqtt_to_vendor(payload: dict[str, Any]) -> str: - """Transform abstract relay payload to zigbee2mqtt format. - - Relay only has power on/off, same transformation as light. - - power: 'on'/'off' -> state: 'ON'/'OFF' - """ - vendor_payload = payload.copy() - - 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 json.dumps(vendor_payload) - - -def _transform_relay_zigbee2mqtt_to_abstract(payload: str) -> dict[str, Any]: - """Transform zigbee2mqtt relay payload to abstract format. - - Relay only has power on/off, same transformation as light. - - state: 'ON'/'OFF' -> power: 'on'/'off' - """ - payload = json.loads(payload) - abstract_payload = payload.copy() - - 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 - - -# ============================================================================ -# HANDLER FUNCTIONS: relay - shelly technology -# ============================================================================ - -def _transform_relay_shelly_to_vendor(payload: dict[str, Any]) -> str: - """Transform abstract relay payload to Shelly format. - - Shelly expects plain text 'on' or 'off' (not JSON). - - power: 'on'/'off' -> 'on'/'off' (plain string) - - Example: - - Abstract: {'power': 'on'} - - Shelly: 'on' - """ - power = payload.get("power", "off") - return power - - -def _transform_relay_shelly_to_abstract(payload: str) -> dict[str, Any]: - """Transform Shelly relay payload to abstract format. - - Shelly sends plain text 'on' or 'off' (not JSON). - - 'on'/'off' -> power: 'on'/'off' - - Example: - - Shelly: 'on' - - Abstract: {'power': 'on'} - """ - return {"power": payload.strip()} - -# ============================================================================ -# HANDLER FUNCTIONS: relay - tasmota technology -# ============================================================================ - -def _transform_relay_tasmota_to_vendor(payload: dict[str, Any]) -> str: - """Transform abstract relay payload to Tasmota format. - - Tasmota expects plain text 'on' or 'off' (not JSON). - - power: 'on'/'off' -> 'on'/'off' (plain string) - - Example: - - Abstract: {'power': 'on'} - - Tasmota: 'on' - """ - power = payload.get("power", "off") - return power - - -def _transform_relay_tasmota_to_abstract(payload: str) -> dict[str, Any]: - """Transform Tasmota relay payload to abstract format. - - Tasmota sends plain text 'on' or 'off' (not JSON). - - 'on'/'off' -> power: 'on'/'off' - - Example: - - Tasmota: 'ON' - - Abstract: {'power': 'on'} - """ - return {"power": payload.strip().lower()} - -# ============================================================================ -# HANDLER FUNCTIONS: relay - hottis_pv_modbus technology -# ============================================================================ - -def _transform_relay_hottis_pv_modbus_to_vendor(payload: dict[str, Any]) -> str: - """Transform abstract relay payload to Hottis Modbus format. - - Hottis Modbus expects plain text 'on' or 'off' (not JSON). - - power: 'on'/'off' -> 'on'/'off' (plain string) - - Example: - - Abstract: {'power': 'on'} - - Hottis Modbus: 'on' - """ - power = payload.get("power", "off") - return power - - -def _transform_relay_hottis_pv_modbus_to_abstract(payload: str) -> dict[str, Any]: - def _transform_relay_hottis_pv_modbus_to_abstract(payload: str) -> dict[str, Any]: - """Transform Hottis Modbus relay payload to abstract format. - - Hottis Modbus sends JSON like: - {"status": "Ok", "timestamp": "...", "state": false, "cnt": 528} - - We only care about the 'state' field: - - state: true -> power: 'on' - - state: false -> power: 'off' - """ - data = json.loads(payload) - - state = data.get("state", False) - power = "on" if bool(state) else "off" - - - return {"power": payload.strip()} - - -# ============================================================================ -# HANDLER FUNCTIONS: three_phase_powermeter - hottis_pv_modbus technology -# ============================================================================ - -def _transform_three_phase_powermeter_hottis_pv_modbus_to_vendor(payload: dict[str, Any]) -> str: - """Transform abstract three_phase_powermeter payload to hottis_pv_modbus format. - energy: float = Field(..., description="Total energy in kWh") - total_power: float = Field(..., description="Total power in W") - phase1_power: float = Field(..., description="Power for phase 1 in W") - phase2_power: float = Field(..., description="Power for phase 2 in W") - phase3_power: float = Field(..., description="Power for phase 3 in W") - phase1_voltage: float = Field(..., description="Voltage for phase 1 in V") - phase2_voltage: float = Field(..., description="Voltage for phase 2 in V") - phase3_voltage: float = Field(..., description="Voltage for phase 3 in V") - phase1_current: float = Field(..., description="Current for phase 1 in A") - phase2_current: float = Field(..., description="Current for phase 2 in A") - phase3_current: float = Field(..., description="Current for phase 3 in A") - - - """ - - vendor_payload = { - "energy": payload.get("energy", 0.0), - "total_power": payload.get("total_power", 0.0), - "phase1_power": payload.get("phase1_power", 0.0), - "phase2_power": payload.get("phase2_power", 0.0), - "phase3_power": payload.get("phase3_power", 0.0), - "phase1_voltage": payload.get("phase1_voltage", 0.0), - "phase2_voltage": payload.get("phase2_voltage", 0.0), - "phase3_voltage": payload.get("phase3_voltage", 0.0), - "phase1_current": payload.get("phase1_current", 0.0), - "phase2_current": payload.get("phase2_current", 0.0), - "phase3_current": payload.get("phase3_current", 0.0), - } - - return json.dumps(vendor_payload) - - -def _transform_three_phase_powermeter_hottis_pv_modbus_to_abstract(payload: str) -> dict[str, Any]: - """Transform hottis_pv_modbus three_phase_powermeter payload to abstract format. - - Transformations: - - Map vendor field names to abstract field names - - totalImportEnergy -> energy - - powerL1/powerL2/powerL3 -> phase1_power/phase2_power/phase3_power - - voltageL1/voltageL2/voltageL3 -> phase1_voltage/phase2_voltage/phase3_voltage - - currentL1/currentL2/currentL3 -> phase1_current/phase2_current/phase3_current - - Sum of powerL1..3 -> total_power - """ - data = json.loads(payload) - - # Helper to read numeric values uniformly as float - def _get_float(key: str, default: float = 0.0) -> float: - return float(data.get(key, default)) - - # Read all numeric values via helper for consistent error handling - phase1_power = _get_float("powerL1") - phase2_power = _get_float("powerL2") - phase3_power = _get_float("powerL3") - - phase1_voltage = _get_float("voltageL1") - phase2_voltage = _get_float("voltageL2") - phase3_voltage = _get_float("voltageL3") - - phase1_current = _get_float("currentL1") - phase2_current = _get_float("currentL2") - phase3_current = _get_float("currentL3") - - energy = _get_float("totalImportEnergy") - - abstract_payload = { - "energy": energy, - "total_power": phase1_power + phase2_power + phase3_power, - "phase1_power": phase1_power, - "phase2_power": phase2_power, - "phase3_power": phase3_power, - "phase1_voltage": phase1_voltage, - "phase2_voltage": phase2_voltage, - "phase3_voltage": phase3_voltage, - "phase1_current": phase1_current, - "phase2_current": phase2_current, - "phase3_current": phase3_current, - } - - return abstract_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) -> 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 - """ - - # Handle both string and numeric input - target_temp = float(payload.strip()) - - return { - "target": target_temp, - "mode": "heat" # MAX! is always in heating mode - } - - -# ============================================================================ -# 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, - - # Temperature & humidity sensor transformations (support both type aliases) - ("temp_humidity_sensor", "zigbee2mqtt", "to_vendor"): _transform_temp_humidity_sensor_zigbee2mqtt_to_vendor, - ("temp_humidity_sensor", "zigbee2mqtt", "to_abstract"): _transform_temp_humidity_sensor_zigbee2mqtt_to_abstract, - ("temp_humidity", "zigbee2mqtt", "to_vendor"): _transform_temp_humidity_sensor_zigbee2mqtt_to_vendor, - ("temp_humidity", "zigbee2mqtt", "to_abstract"): _transform_temp_humidity_sensor_zigbee2mqtt_to_abstract, - - # Relay transformations - ("relay", "zigbee2mqtt", "to_vendor"): _transform_relay_zigbee2mqtt_to_vendor, - ("relay", "zigbee2mqtt", "to_abstract"): _transform_relay_zigbee2mqtt_to_abstract, - ("relay", "shelly", "to_vendor"): _transform_relay_shelly_to_vendor, - ("relay", "shelly", "to_abstract"): _transform_relay_shelly_to_abstract, - ("relay", "hottis_pv_modbus", "to_vendor"): _transform_relay_hottis_pv_modbus_to_vendor, - ("relay", "hottis_pv_modbus", "to_abstract"): _transform_relay_hottis_pv_modbus_to_abstract, - ("relay", "tasmota", "to_vendor"): _transform_relay_tasmota_to_vendor, - ("relay", "tasmota", "to_abstract"): _transform_relay_tasmota_to_abstract, - - # Three-Phase Powermeter transformations - ("three_phase_powermeter", "hottis_pv_modbus", "to_vendor"): _transform_three_phase_powermeter_hottis_pv_modbus_to_vendor, - ("three_phase_powermeter", "hottis_pv_modbus", "to_abstract"): _transform_three_phase_powermeter_hottis_pv_modbus_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: 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 - - 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