"""Zigbee2MQTT vendor transformations.""" import json import logging from typing import Any logger = logging.getLogger(__name__) def transform_light_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)): vendor_payload["brightness"] = round(abstract_brightness * 254 / 100) return json.dumps(vendor_payload) def transform_light_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)): abstract_payload["brightness"] = round(vendor_brightness * 100 / 254) return abstract_payload def transform_thermostat_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: vendor_payload["current_heating_setpoint"] = str(payload["target"]) return json.dumps(vendor_payload) def transform_thermostat_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 = {} if "current_heating_setpoint" in payload: setpoint = payload["current_heating_setpoint"] abstract_payload["target"] = float(setpoint) if "local_temperature" in payload: current = payload["local_temperature"] abstract_payload["current"] = float(current) if "system_mode" in payload: abstract_payload["mode"] = payload["system_mode"] return abstract_payload def transform_contact_sensor_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. """ logger.warning("Contact sensors are read-only - SET commands should not be used") return json.dumps(payload) def transform_contact_sensor_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!) Example: - zigbee2mqtt: {"contact": false, "battery": 100} - Abstract: {"contact": "open", "battery": 100} """ payload = json.loads(payload) abstract_payload = {} if "contact" in payload: contact_bool = payload["contact"] abstract_payload["contact"] = "closed" if contact_bool else "open" # Pass through optional fields for field in ["battery", "linkquality", "device_temperature", "voltage"]: if field in payload: abstract_payload[field] = payload[field] return abstract_payload def transform_temp_humidity_sensor_to_vendor(payload: dict[str, Any]) -> str: """Transform abstract temp/humidity sensor payload to zigbee2mqtt format. Temp/humidity sensors are read-only. """ return json.dumps(payload) def transform_temp_humidity_sensor_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 def transform_relay_to_vendor(payload: dict[str, Any]) -> str: """Transform abstract relay payload to zigbee2mqtt format. - 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_to_abstract(payload: str) -> dict[str, Any]: """Transform zigbee2mqtt relay payload to abstract format. - 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 # Registry of handlers for this vendor HANDLERS = { ("light", "to_vendor"): transform_light_to_vendor, ("light", "to_abstract"): transform_light_to_abstract, ("thermostat", "to_vendor"): transform_thermostat_to_vendor, ("thermostat", "to_abstract"): transform_thermostat_to_abstract, ("contact_sensor", "to_vendor"): transform_contact_sensor_to_vendor, ("contact_sensor", "to_abstract"): transform_contact_sensor_to_abstract, ("contact", "to_vendor"): transform_contact_sensor_to_vendor, ("contact", "to_abstract"): transform_contact_sensor_to_abstract, ("temp_humidity_sensor", "to_vendor"): transform_temp_humidity_sensor_to_vendor, ("temp_humidity_sensor", "to_abstract"): transform_temp_humidity_sensor_to_abstract, ("temp_humidity", "to_vendor"): transform_temp_humidity_sensor_to_vendor, ("temp_humidity", "to_abstract"): transform_temp_humidity_sensor_to_abstract, ("relay", "to_vendor"): transform_relay_to_vendor, ("relay", "to_abstract"): transform_relay_to_abstract, }