transformation added 2
This commit is contained in:
@@ -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}, "
|
||||
|
||||
Reference in New Issue
Block a user