"""Dali Gateway""" import asyncio import contextlib from enum import Enum, auto import json import logging import random import ssl import threading import time from typing import Any, Callable, Dict, List, Sequence, Tuple, Union, cast import paho.mqtt.client as paho_mqtt # Backward compatibility with paho-mqtt < 2.0.0 try: from paho.mqtt.enums import CallbackAPIVersion HAS_CALLBACK_API_VERSION = True except ImportError: # paho-mqtt < 2.0.0 doesn't have CallbackAPIVersion HAS_CALLBACK_API_VERSION = False # pyright: ignore[reportConstantRedefinition] from .const import ( BUS_SCAN_TIMEOUT, CA_CERT_PATH, DEVICE_MODEL_MAP, DEVICE_PARAM_KEY_MAP, DEVICE_PARAM_PROTOCOL_KEY_MAP, DPID_ENERGY, INBOUND_CALLBACK_BATCH_WINDOW_MS, MAX_CONCURRENT_READS, SENSOR_PARAM_KEY_MAP, SENSOR_PARAM_PROTOCOL_KEY_MAP, ) from .device import Device from .exceptions import BusScanCancelledError, DaliGatewayError from .group import Group from .helper import ( gen_device_name, gen_device_unique_id, gen_group_unique_id, gen_scene_unique_id, is_illuminance_sensor, is_light_device, is_motion_sensor, is_panel_device, parse_illuminance_status, parse_light_status, parse_motion_status, parse_panel_status, ) from .scene import Scene from .types import ( CallbackEventType, DeviceParamCommand, DeviceParamType, EnergyData, GroupDeviceType, IlluminanceStatus, LightStatus, MotionStatus, PanelStatus, SceneDeviceType, SensorParamType, ) from .udp_client import send_identify_gateway _LOGGER = logging.getLogger(__name__) # Connection parameters _CONNECTION_TIMEOUT = 30.0 # seconds - gateway broker may respond slowly # Reconnection parameters _RECONNECT_INITIAL_DELAY = 1.0 # seconds _RECONNECT_MAX_DELAY = 60.0 # seconds _RECONNECT_BACKOFF_MULTIPLIER = 2.0 _RECONNECT_JITTER = 0.1 # ±10% class ConnectionState(Enum): """Connection state machine for gateway.""" DISCONNECTED = auto() CONNECTING = auto() CONNECTED = auto() RECONNECTING = auto() class DaliGateway: """Dali Gateway""" def __init__( self, gw_sn: str, gw_ip: str, port: int, username: str, passwd: str, *, name: str | None = None, channel_total: Sequence[int] | None = None, is_tls: bool = False, loop: asyncio.AbstractEventLoop | None = None, ) -> None: self._gw_sn = gw_sn self._gw_ip = gw_ip self._port = port self._name = name or gw_sn self._username = username self._passwd = passwd self._is_tls = is_tls self._channel_total = ( [int(ch) for ch in channel_total] if channel_total else [0] ) self.software_version: str = "" self.firmware_version: str = "" # Event loop for thread-safe callback dispatch # Can be provided at __init__ or will be auto-detected in connect() self._loop: asyncio.AbstractEventLoop | None = loop # Connection state machine self._connection_state = ConnectionState.DISCONNECTED self._reconnect_task: asyncio.TimerHandle | None = None self._reconnect_delay = _RECONNECT_INITIAL_DELAY self._shutdown_requested = False self._connection_lock: asyncio.Lock | None = None # Initialized in connect() self._sub_topic = f"/{self._gw_sn}/client/reciver/" self._pub_topic = f"/{self._gw_sn}/server/publish/" # MQTT client - handle compatibility between paho-mqtt versions # Use timestamp in client_id to ensure uniqueness across reconnections. # MQTT brokers may reject connections from clients with duplicate IDs. client_id = f"ha_dali_center_{self._gw_sn}_{int(time.time() * 1000)}" if HAS_CALLBACK_API_VERSION: # paho-mqtt >= 2.0.0 self._mqtt_client = paho_mqtt.Client( CallbackAPIVersion.VERSION2, # pyright: ignore[reportPossiblyUnboundVariable] client_id=client_id, protocol=paho_mqtt.MQTTv311, ) else: # paho-mqtt < 2.0.0 self._mqtt_client = paho_mqtt.Client( client_id=client_id, protocol=paho_mqtt.MQTTv311, ) self._connect_result: int | None = None self._connection_event = asyncio.Event() self._mqtt_client.on_connect = self._on_connect self._mqtt_client.on_disconnect = self._on_disconnect self._mqtt_client.on_message = self._on_message self._scenes_received = asyncio.Event() self._groups_received = asyncio.Event() self._devices_received = asyncio.Event() self._bus_scan_complete = asyncio.Event() self._scenes_result: list[Scene] = [] self._groups_result: list[Group] = [] self._devices_result: list[Device] = [] self._devices_seen_ids: set[str] = set() self._bus_scan_result: list[Device] = [] self._bus_scan_seen_ids: set[str] = set() self._bus_scan_cancelled = False self._bus_scan_channels: list[int] = [] self._bus_scanning = False self._read_group_events: Dict[Tuple[int, int], asyncio.Event] = {} self._read_group_results: Dict[Tuple[int, int], Dict[str, Any]] = {} self._read_scene_events: Dict[Tuple[int, int], asyncio.Event] = {} self._read_scene_results: Dict[Tuple[int, int], Dict[str, Any]] = {} self._device_listeners: Dict[ CallbackEventType, Dict[str, List[Callable[..., None]]] ] = { CallbackEventType.ONLINE_STATUS: {}, CallbackEventType.LIGHT_STATUS: {}, CallbackEventType.MOTION_STATUS: {}, CallbackEventType.ILLUMINANCE_STATUS: {}, CallbackEventType.PANEL_STATUS: {}, CallbackEventType.ENERGY_REPORT: {}, CallbackEventType.ENERGY_DATA: {}, CallbackEventType.SENSOR_ON_OFF: {}, CallbackEventType.DEV_PARAM: {}, CallbackEventType.SENSOR_PARAM: {}, CallbackEventType.VERSION_UPDATED: {}, } self._pending_requests: Dict[str, Dict[str, Dict[str, Any]]] = {} self._batch_timer: Dict[str, asyncio.TimerHandle] = {} # cmd -> timer # Inbound callback batching with smart merging # Key: (event_type, dev_id, listener_id) -> (listener, merged_data) self._pending_callbacks: Dict[ Tuple[CallbackEventType, str, int], Tuple[Callable[..., None], Any] ] = {} self._callback_lock = threading.Lock() self._batch_scheduled = False # MQTT command dispatch table (built once, not per-message) self._command_handlers: Dict[str, Callable[[Dict[str, Any]], None]] = { "devStatus": self._process_device_status, "readDevRes": self._process_device_status, "writeDevRes": self._noop_handler, "writeGroupRes": self._noop_handler, "writeSceneRes": self._noop_handler, "onlineStatus": self._process_online_status, "reportEnergy": self._process_energy_report, "searchDevRes": self._process_search_device_response, "getSceneRes": self._process_get_scene_response, "getGroupRes": self._process_get_group_response, "getVersionRes": self._process_get_version_response, "readGroupRes": self._process_read_group_response, "readSceneRes": self._process_read_scene_response, "restartGatewayRes": self._process_restart_gateway_response, "getEnergyRes": self._process_get_energy_response, "setSensorOnOffRes": self._noop_handler, "getSensorOnOffRes": self._process_get_sensor_on_off_response, "setSensorArgvRes": self._noop_handler, "getSensorArgvRes": self._process_get_sensor_argv_response, "setDevParamRes": self._noop_handler, "getDevParamRes": self._process_get_dev_param_response, "identifyDevRes": self._process_identify_dev_response, } def _get_device_key(self, dev_type: str, channel: int, address: int) -> str: return f"{dev_type}_{channel}_{address}" def _build_parameter(self, param: DeviceParamType) -> Dict[str, Any]: """Convert DeviceParamType to protocol format using snake_case to camelCase mapping.""" param_dict = dict(param) return { DEVICE_PARAM_KEY_MAP[python_key]: param_dict[python_key] for python_key in param_dict if python_key in DEVICE_PARAM_KEY_MAP } def add_request( self, cmd: str, dev_type: str, channel: int, address: int, data: Dict[str, Any] ) -> None: if cmd not in self._pending_requests: self._pending_requests[cmd] = {} device_key = self._get_device_key(dev_type, channel, address) # Merge properties instead of overwriting the entire data if device_key in self._pending_requests[cmd]: existing_data = self._pending_requests[cmd][device_key] if "property" in existing_data and "property" in data: # Merge properties, avoiding duplicates by dpid existing_properties = { prop["dpid"]: prop for prop in existing_data["property"] } new_properties = {prop["dpid"]: prop for prop in data["property"]} existing_properties.update(new_properties) data["property"] = list(existing_properties.values()) self._pending_requests[cmd][device_key] = data if self._batch_timer.get(cmd) is None: if self._loop is None or not self._loop.is_running(): # Fallback: flush immediately if no event loop available self._flush_batch(cmd) return self._batch_timer[cmd] = self._loop.call_later( INBOUND_CALLBACK_BATCH_WINDOW_MS / 1000.0, self._flush_batch, cmd ) def _flush_batch(self, cmd: str) -> None: if not self._pending_requests.get(cmd): return batch_data: List[Dict[str, Any]] = list(self._pending_requests[cmd].values()) command: Dict[str, Any] = { "cmd": cmd, "msgId": str(int(time.time())), "gwSn": self._gw_sn, "data": batch_data, } self._mqtt_client.publish(self._pub_topic, json.dumps(command)) _LOGGER.debug( "Gateway %s: Sent batch %s %s", self._gw_sn, cmd, json.dumps(command) ) self._pending_requests[cmd].clear() self._batch_timer.pop(cmd, None) def __repr__(self) -> str: return ( f"DaliGateway(gw_sn={self._gw_sn}, gw_ip={self._gw_ip}, " f"port={self._port}, name={self._name})" ) def _publish_command(self, cmd: str, **kwargs: Any) -> None: """Publish a command to the MQTT broker. Args: cmd: The command name (e.g., 'writeScene', 'getSensorOnOff') **kwargs: Additional fields to include in the command payload """ payload: Dict[str, Any] = { "cmd": cmd, "msgId": str(int(time.time())), "gwSn": self._gw_sn, **kwargs, } self._mqtt_client.publish(self._pub_topic, json.dumps(payload)) @property def gw_sn(self) -> str: return self._gw_sn @property def gw_ip(self) -> str: return self._gw_ip @property def port(self) -> int: return self._port @property def username(self) -> str: return self._username @property def passwd(self) -> str: return self._passwd @property def channel_total(self) -> List[int]: return list(self._channel_total) @property def is_tls(self) -> bool: return self._is_tls @property def name(self) -> str: return self._name @property def is_connected(self) -> bool: """Return True if the gateway is connected.""" return self._connection_state == ConnectionState.CONNECTED @property def connection_state(self) -> ConnectionState: """Return the current connection state.""" return self._connection_state @property def bus_scanning(self) -> bool: """Return True if a bus scan is currently in progress.""" return self._bus_scanning def _set_event_threadsafe(self, event: asyncio.Event) -> None: """Set an asyncio.Event in a thread-safe manner.""" if self._loop is not None and self._loop.is_running(): self._loop.call_soon_threadsafe(event.set) else: event.set() def register_listener( self, event_type: CallbackEventType, listener: Union[ Callable[[bool], None], Callable[[LightStatus], None], Callable[[MotionStatus], None], Callable[[IlluminanceStatus], None], Callable[[PanelStatus], None], Callable[[float], None], Callable[[EnergyData], None], Callable[[DeviceParamType], None], Callable[[SensorParamType], None], Callable[[Tuple[str, str]], None], ], dev_id: str, ) -> Callable[[], None]: """Register a listener for a specific event type. Args: event_type: The type of event to listen for listener: The callback function to invoke dev_id: Device ID to filter events for (required) """ if event_type not in self._device_listeners: return lambda: None # Register device-specific listener if dev_id not in self._device_listeners[event_type]: self._device_listeners[event_type][dev_id] = [] self._device_listeners[event_type][dev_id].append(listener) def _unsubscribe() -> None: listeners = self._device_listeners.get(event_type, {}).get(dev_id, []) with contextlib.suppress(ValueError): listeners.remove(listener) return _unsubscribe def _notify_listeners( self, event_type: CallbackEventType, dev_id: str, data: Union[ bool, LightStatus, MotionStatus, IlluminanceStatus, PanelStatus, float, EnergyData, DeviceParamType, SensorParamType, Tuple[str, str], ], ) -> None: """Queue callbacks for batched dispatch to prevent event loop overload. Smart merging: same device + same listener merges dict fields, keeping latest value for each field. Non-dict types are replaced. """ listeners = self._device_listeners.get(event_type, {}).get(dev_id, []) if not listeners: return # Fallback: if no event loop, call directly (backward compatibility) if self._loop is None or not self._loop.is_running(): for listener in listeners: listener(data) return with self._callback_lock: for listener in listeners: key = (event_type, dev_id, id(listener)) if key in self._pending_callbacks: _, existing_data = self._pending_callbacks[key] if isinstance(existing_data, dict) and isinstance(data, dict): # Explicit cast for TypedDict compatibility existing_dict = cast("Dict[str, Any]", existing_data) new_dict = cast("Dict[str, Any]", data) merged: Dict[str, Any] = { **existing_dict, **{k: v for k, v in new_dict.items() if v is not None}, } self._pending_callbacks[key] = (listener, merged) else: self._pending_callbacks[key] = (listener, data) else: self._pending_callbacks[key] = (listener, data) if not self._batch_scheduled: self._batch_scheduled = True self._loop.call_soon_threadsafe(self._schedule_flush) def _schedule_flush(self) -> None: """Schedule flush after batch window. Must be called from event loop.""" if self._loop is not None: self._loop.call_later( INBOUND_CALLBACK_BATCH_WINDOW_MS / 1000.0, self._flush_callbacks, ) def _flush_callbacks(self) -> None: """Flush all pending callbacks. Runs in the event loop thread.""" with self._callback_lock: pending = self._pending_callbacks self._pending_callbacks = {} self._batch_scheduled = False for listener, data in pending.values(): listener(data) def _on_connect( self, client: paho_mqtt.Client, userdata: Any, flags: Any, rc: int, properties: Any = None, ) -> None: self._connect_result = rc # Thread-safe Event.set() self._set_event_threadsafe(self._connection_event) if rc == 0: # Update connection state self._connection_state = ConnectionState.CONNECTED self._reconnect_delay = _RECONNECT_INITIAL_DELAY # Reset backoff _LOGGER.debug( "Gateway %s: MQTT connection established to %s:%s", self._gw_sn, self._gw_ip, self._port, ) self._mqtt_client.subscribe(self._sub_topic) _LOGGER.debug( "Gateway %s: Subscribed to MQTT topic %s", self._gw_sn, self._sub_topic ) # Notify gateway-level listeners (thread-safe via _notify_listeners) self._notify_listeners(CallbackEventType.ONLINE_STATUS, self._gw_sn, True) # Notify all device-specific listeners that gateway is online for device_id in self._device_listeners[CallbackEventType.ONLINE_STATUS]: if device_id != self._gw_sn: self._notify_listeners( CallbackEventType.ONLINE_STATUS, device_id, True ) else: _LOGGER.error( "Gateway %s: MQTT connection failed with code %s", self._gw_sn, rc ) def _on_disconnect( self, client: paho_mqtt.Client, userdata: Any, *args: Any, ) -> None: # Handle different paho-mqtt versions: # v1.6.x: (client, userdata, rc) # v2.0.0+: (client, userdata, disconnect_flags, reason_code, properties) if HAS_CALLBACK_API_VERSION and len(args) >= 2: # paho-mqtt >= 2.0.0 reason_code = args[1] # disconnect_flags, reason_code, properties elif len(args) >= 1: # paho-mqtt < 2.0.0 reason_code = args[0] # rc else: reason_code = 0 was_connected = self._connection_state == ConnectionState.CONNECTED unexpected_disconnect = reason_code != 0 and was_connected if unexpected_disconnect: _LOGGER.warning( "Gateway %s: Unexpected MQTT disconnection (%s:%s) - Reason code: %s", self._gw_sn, self._gw_ip, self._port, reason_code, ) # Set state to RECONNECTING if we should attempt reconnection if not self._shutdown_requested: self._connection_state = ConnectionState.RECONNECTING self._schedule_reconnect() else: self._connection_state = ConnectionState.DISCONNECTED else: self._connection_state = ConnectionState.DISCONNECTED # Notify gateway-level listeners (thread-safe via _notify_listeners) self._notify_listeners(CallbackEventType.ONLINE_STATUS, self._gw_sn, False) # Notify all device-specific listeners that gateway is offline for device_id in self._device_listeners[CallbackEventType.ONLINE_STATUS]: if device_id != self._gw_sn: self._notify_listeners( CallbackEventType.ONLINE_STATUS, device_id, False ) def _on_message( self, client: paho_mqtt.Client, userdata: Any, msg: paho_mqtt.MQTTMessage ) -> None: try: payload_json = json.loads(msg.payload.decode("utf-8", errors="replace")) _LOGGER.debug( "Gateway %s: Received MQTT message on topic %s: %s", self._gw_sn, msg.topic, payload_json, ) cmd = payload_json.get("cmd") if not cmd: _LOGGER.warning( "Gateway %s: Received MQTT message without cmd field", self._gw_sn ) return handler = self._command_handlers.get(cmd) if handler: handler(payload_json) else: _LOGGER.debug( "Gateway %s: Unhandled MQTT command '%s', payload: %s", self._gw_sn, cmd, payload_json, ) except json.JSONDecodeError: _LOGGER.error( "Gateway %s: Failed to decode MQTT message payload: %s", self._gw_sn, msg.payload, ) except (ValueError, KeyError, TypeError) as e: _LOGGER.error( "Gateway %s: Error processing MQTT message: %s", self._gw_sn, str(e) ) def _process_online_status(self, payload: Dict[str, Any]) -> None: data_list = payload.get("data") if not data_list: _LOGGER.warning( "Gateway %s: Received onlineStatus with no data: %s", self._gw_sn, payload, ) return for data in data_list: dev_id = gen_device_unique_id( data.get("devType"), data.get("channel"), data.get("address"), self._gw_sn, ) available: bool = data.get("status", False) self._notify_listeners(CallbackEventType.ONLINE_STATUS, dev_id, available) def _process_device_status(self, payload: Dict[str, Any]) -> None: data = payload.get("data") if not data: _LOGGER.warning( "Gateway %s: Received devStatus with no data: %s", self._gw_sn, payload ) return dev_id = gen_device_unique_id( data.get("devType"), data.get("channel"), data.get("address"), self._gw_sn ) if not dev_id: _LOGGER.warning("Failed to generate device ID from data: %s", data) return property_list = data.get("property", []) dev_type = data.get("devType") if dev_type and is_light_device(dev_type): light_status = parse_light_status(property_list) self._notify_listeners(CallbackEventType.LIGHT_STATUS, dev_id, light_status) elif dev_type and is_motion_sensor(dev_type): motion_statuses = parse_motion_status(property_list) for motion_status in motion_statuses: self._notify_listeners( CallbackEventType.MOTION_STATUS, dev_id, motion_status ) elif dev_type and is_illuminance_sensor(dev_type): illuminance_statuses = parse_illuminance_status(property_list) for illuminance_status in illuminance_statuses: self._notify_listeners( CallbackEventType.ILLUMINANCE_STATUS, dev_id, illuminance_status ) elif dev_type and is_panel_device(dev_type): panel_statuses = parse_panel_status(property_list) for panel_status in panel_statuses: self._notify_listeners( CallbackEventType.PANEL_STATUS, dev_id, panel_status ) else: # Warn if no callback handler exists for this device type _LOGGER.warning( "Gateway %s: No callback handler for device type %s (device: %s). " "Property data: %s", self._gw_sn, dev_type, dev_id, property_list, ) def _noop_handler(self, payload: Dict[str, Any]) -> None: """No-op handler for responses that need no processing beyond logging.""" def _process_identify_dev_response(self, payload: Dict[str, Any]) -> None: """Process identifyDev response. This method exists as a hook for subclasses to override. The response is already logged by _on_message. """ def _process_energy_report(self, payload: Dict[str, Any]) -> None: data = payload.get("data") if not data: _LOGGER.warning( "Gateway %s: Received reportEnergy with no data: %s", self._gw_sn, payload, ) return dev_id = gen_device_unique_id( data.get("devType"), data.get("channel"), data.get("address"), self._gw_sn ) if not dev_id: _LOGGER.warning("Failed to generate device ID from data: %s", data) return property_list = data.get("property", []) for prop in property_list: if prop.get("dpid") == DPID_ENERGY: try: energy_value = float(prop.get("value", "0")) self._notify_listeners( CallbackEventType.ENERGY_REPORT, dev_id, energy_value ) except (ValueError, TypeError) as e: _LOGGER.error("Error converting energy value: %s", str(e)) def _process_get_version_response(self, payload_json: Dict[str, Any]) -> None: self.software_version = payload_json.get("data", {}).get("swVersion", "") self.firmware_version = payload_json.get("data", {}).get("fwVersion", "") self._notify_listeners( CallbackEventType.VERSION_UPDATED, self._gw_sn, (self.software_version, self.firmware_version), ) def _process_get_energy_response(self, payload_json: Dict[str, Any]) -> None: data_list = payload_json.get("data") if not data_list: _LOGGER.warning( "Gateway %s: Received getEnergyRes with no data: %s", self._gw_sn, payload_json, ) return for data in data_list: dev_id = gen_device_unique_id( data.get("devType"), data.get("channel"), data.get("address"), self._gw_sn, ) if not dev_id: _LOGGER.warning("Failed to generate device ID from data: %s", data) continue energy_data: EnergyData = { "yearEnergy": data.get("yearEnergy", {}), "monthEnergy": data.get("monthEnergy", {}), "dayEnergy": data.get("dayEnergy", {}), "hourEnergy": data.get("hourEnergy", []), } self._notify_listeners(CallbackEventType.ENERGY_DATA, dev_id, energy_data) def _parse_device_from_raw(self, raw: Dict[str, Any]) -> Device: """Parse a raw device dict from searchDevRes into a Device object.""" dev_type = str(raw.get("devType", "")) channel = int(raw.get("channel", 0)) address = int(raw.get("address", 0)) unique_id = gen_device_unique_id(dev_type, channel, address, self._gw_sn) dev_id = str(raw.get("devId") or unique_id) name = str(raw.get("name") or gen_device_name(dev_type, channel, address)) return Device( self, unique_id=unique_id, dev_id=dev_id, name=name, dev_type=dev_type, channel=channel, address=address, status=str(raw.get("status", "")), dev_sn=str(raw.get("devSn", "")), area_name=str(raw.get("areaName", "")), area_id=str(raw.get("areaId", "")), model=DEVICE_MODEL_MAP.get(dev_type, "Unknown"), properties=[], ) @staticmethod def _append_unique( device: Device, result: list[Device], seen_ids: set[str] ) -> bool: """Append device to result list if unique_id is not already present.""" if device.unique_id in seen_ids: return False seen_ids.add(device.unique_id) result.append(device) return True def _process_search_device_response(self, payload_json: Dict[str, Any]) -> None: search_flag = payload_json.get("searchFlag", "exited") search_status = payload_json.get("searchStatus") # Route based on search mode if search_flag == "busDevice": # Bus scan mode - handle state machine _LOGGER.debug("Gateway %s: Bus scan status %s", self._gw_sn, search_status) if search_status == 2: # Scanning in progress - track actual gateway state self._bus_scanning = True _LOGGER.info("Gateway %s: Bus scan in progress...", self._gw_sn) elif search_status == 3: # Device data reported - accumulate devices count_before = len(self._bus_scan_result) for raw in payload_json.get("data", []): self._append_unique( self._parse_device_from_raw(raw), self._bus_scan_result, self._bus_scan_seen_ids, ) _LOGGER.info( "Gateway %s: Received %d devices (total: %d)", self._gw_sn, len(self._bus_scan_result) - count_before, len(self._bus_scan_result), ) elif search_status in {0, 1}: # Scan complete - track actual gateway state self._bus_scanning = False if search_status == 0: _LOGGER.info( "Gateway %s: Bus scan complete - no devices found", self._gw_sn ) else: _LOGGER.info( "Gateway %s: Bus scan complete - found %d device(s)", self._gw_sn, len(self._bus_scan_result), ) self._set_event_threadsafe(self._bus_scan_complete) else: # Legacy exited mode - original behavior for raw in payload_json.get("data", []): self._append_unique( self._parse_device_from_raw(raw), self._devices_result, self._devices_seen_ids, ) if search_status in {0, 1}: self._set_event_threadsafe(self._devices_received) def _process_get_scene_response(self, payload_json: Dict[str, Any]) -> None: self._scenes_result.clear() for channel_scenes in payload_json.get("scene", []): channel = channel_scenes.get("channel", 0) for scene_data in channel_scenes.get("data", []): scene_id = int(scene_data.get("sceneId", 0)) name = str(scene_data.get("name", "")) area_id = str(scene_data.get("areaId", "")) if any( existing.unique_id == gen_scene_unique_id(scene_id, channel, self._gw_sn) for existing in self._scenes_result ): continue self._scenes_result.append( Scene( self, scene_id=scene_id, name=name, channel=channel, area_id=area_id, devices=[], ) ) self._set_event_threadsafe(self._scenes_received) def _process_get_group_response(self, payload_json: Dict[str, Any]) -> None: self._groups_result.clear() for channel_groups in payload_json.get("group", []): channel = channel_groups.get("channel", 0) for group_data in channel_groups.get("data", []): group_id = int(group_data.get("groupId", 0)) name = str(group_data.get("name", "")) area_id = str(group_data.get("areaId", "")) if any( existing.unique_id == gen_group_unique_id(group_id, channel, self._gw_sn) for existing in self._groups_result ): continue self._groups_result.append( Group( self, group_id=group_id, name=name, channel=channel, area_id=area_id, devices=[], ) ) self._set_event_threadsafe(self._groups_received) def _process_read_group_response(self, payload: Dict[str, Any]) -> None: group_id = payload.get("groupId", 0) group_name = payload.get("name", "") channel = payload.get("channel", 0) group_key = (group_id, channel) raw_devices = payload.get("data", []) # Create GroupDeviceType objects from raw device data devices: List[GroupDeviceType] = [] for device_data in raw_devices: dev_type = str(device_data.get("devType", "")) channel_id = int(device_data.get("channel", 0)) address = int(device_data.get("address", 0)) device: GroupDeviceType = { "unique_id": gen_device_unique_id( dev_type, channel_id, address, self._gw_sn ), "id": str(device_data.get("devId", "")), "name": gen_device_name(dev_type, channel_id, address), "dev_type": dev_type, "channel": channel_id, "address": address, "status": "", "dev_sn": "", "area_name": "", "area_id": "", "model": DEVICE_MODEL_MAP.get(dev_type, "Unknown"), "prop": [], } devices.append(device) self._read_group_results[group_key] = { "unique_id": gen_group_unique_id(group_id, channel, self._gw_sn), "id": group_id, "name": group_name, "channel": channel, "area_id": "", "devices": devices, } # Signal completion for this specific group if group_key in self._read_group_events: self._set_event_threadsafe(self._read_group_events[group_key]) def _process_read_scene_response(self, payload: Dict[str, Any]) -> None: scene_id = payload.get("sceneId", 0) scene_name = payload.get("name", "") channel = payload.get("channel", 0) scene_key = (scene_id, channel) data: Dict[str, Any] | None = payload.get("data") if data is None: _LOGGER.error( "Gateway %s: Received readSceneRes with no data for scene %s channel %s", self._gw_sn, scene_id, channel, ) # Mark as received even with error to unblock waiting coroutine if scene_key in self._read_scene_events: self._set_event_threadsafe(self._read_scene_events[scene_key]) return raw_devices: List[Dict[str, Any]] = data.get("device", []) # Create SceneDeviceType objects from raw device data devices: List[SceneDeviceType] = [] for device_data in raw_devices: # Convert raw property data to LightStatus using parse_light_status raw_properties = device_data.get("property", []) light_status = parse_light_status(raw_properties) dev_type = str(device_data.get("devType", "")) channel_id = int(device_data.get("channel", 0)) address = int(device_data.get("address", 0)) gw_sn_obj = str(device_data.get("gwSnObj", "")) if dev_type == "0401": unique_id = gen_group_unique_id(address, channel_id, self._gw_sn) else: unique_id = gen_device_unique_id( dev_type, channel_id, address, self._gw_sn ) device: SceneDeviceType = { "unique_id": unique_id, "dev_type": dev_type, "channel": channel_id, "address": address, "gw_sn_obj": gw_sn_obj, "property": light_status, } devices.append(device) self._read_scene_results[scene_key] = { "unique_id": gen_scene_unique_id(scene_id, channel, self._gw_sn), "id": scene_id, "name": scene_name, "channel": channel, "area_id": "", "devices": devices, } # Signal completion for this specific scene if scene_key in self._read_scene_events: self._set_event_threadsafe(self._read_scene_events[scene_key]) def _process_get_sensor_on_off_response(self, payload: Dict[str, Any]) -> None: dev_id = gen_device_unique_id( payload.get("devType", ""), payload.get("channel", 0), payload.get("address", 0), self._gw_sn, ) value = payload.get("value", False) self._notify_listeners(CallbackEventType.SENSOR_ON_OFF, dev_id, value) def _process_get_sensor_argv_response(self, payload: Dict[str, Any]) -> None: """Process getSensorArgv response and emit parameters to listeners.""" dev_type = payload.get("devType", "") channel = payload.get("channel", 0) address = payload.get("address", 0) data = payload.get("data", {}) if not data: return # Build SensorParamType from response using protocol to Python key mapping sensor_param_dict: Dict[str, Any] = { SENSOR_PARAM_PROTOCOL_KEY_MAP[protocol_key]: value for protocol_key, value in data.items() if protocol_key in SENSOR_PARAM_PROTOCOL_KEY_MAP } sensor_param = cast("SensorParamType", sensor_param_dict) dev_id = gen_device_unique_id(dev_type, channel, address, self._gw_sn) self._notify_listeners(CallbackEventType.SENSOR_PARAM, dev_id, sensor_param) def _process_get_dev_param_response(self, payload: Dict[str, Any]) -> None: """Process getDevParam response and emit parameters to listeners.""" dev_type = payload.get("devType", "") channel = payload.get("channel", 0) address = payload.get("address", 0) paramer = payload.get("paramer", {}) if not paramer: return # Build DeviceParamType from response using protocol to Python key mapping device_param_dict: Dict[str, Any] = { DEVICE_PARAM_PROTOCOL_KEY_MAP[protocol_key]: value for protocol_key, value in paramer.items() if protocol_key in DEVICE_PARAM_PROTOCOL_KEY_MAP } device_param = cast("DeviceParamType", device_param_dict) dev_id = gen_device_unique_id(dev_type, channel, address, self._gw_sn) self._notify_listeners(CallbackEventType.DEV_PARAM, dev_id, device_param) def _process_restart_gateway_response(self, payload: Dict[str, Any]) -> None: ack = payload.get("ack", False) _LOGGER.info( "Gateway %s: Received restart confirmation, ack: %s. Gateway will restart shortly.", self._gw_sn, ack, ) async def _setup_ssl(self) -> None: try: loop = asyncio.get_running_loop() await loop.run_in_executor(None, self._setup_ssl_sync) except Exception as e: _LOGGER.error("Failed to configure SSL/TLS: %s", str(e)) raise DaliGatewayError( f"SSL/TLS configuration failed: {e}", self._gw_sn ) from e def _setup_ssl_sync(self) -> None: context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH) context.load_verify_locations(str(CA_CERT_PATH)) context.check_hostname = False context.verify_mode = ssl.CERT_REQUIRED self._mqtt_client.tls_set_context(context) # pyright: ignore[reportUnknownMemberType] _LOGGER.debug("SSL/TLS configured with CA certificate: %s", CA_CERT_PATH) def get_credentials(self) -> tuple[str, str]: return self._username, self._passwd def _schedule_reconnect(self) -> None: """Schedule a reconnection attempt with exponential backoff. Called from paho-mqtt thread, so must be thread-safe. """ if self._shutdown_requested: _LOGGER.debug( "Gateway %s: Shutdown requested, skipping reconnection", self._gw_sn ) return # Cancel any existing reconnect task to prevent duplicate reconnections if self._reconnect_task is not None: self._reconnect_task.cancel() self._reconnect_task = None # Add jitter to prevent thundering herd jitter = self._reconnect_delay * _RECONNECT_JITTER * (2 * random.random() - 1) delay = self._reconnect_delay + jitter _LOGGER.info( "Gateway %s: Scheduling reconnection in %.1f seconds", self._gw_sn, delay, ) # Schedule reconnection on the event loop if self._loop is not None and self._loop.is_running(): self._reconnect_task = self._loop.call_later( delay, lambda: self._loop.create_task(self._reconnect()), # type: ignore[union-attr] ) else: _LOGGER.warning( "Gateway %s: No event loop available for reconnection", self._gw_sn, ) # Increase delay for next attempt (exponential backoff) self._reconnect_delay = min( self._reconnect_delay * _RECONNECT_BACKOFF_MULTIPLIER, _RECONNECT_MAX_DELAY, ) async def _reconnect(self) -> None: """Attempt to reconnect to the gateway.""" if self._shutdown_requested: _LOGGER.debug( "Gateway %s: Shutdown requested, aborting reconnection", self._gw_sn ) return # Use connection lock for state check if available if self._connection_lock is not None: async with self._connection_lock: # Skip if already connected to prevent duplicate connections if self._connection_state == ConnectionState.CONNECTED: _LOGGER.debug( "Gateway %s: Already connected, skipping reconnection", self._gw_sn, ) return # Reset connection state while holding lock self._connection_event.clear() self._connect_result = None else: # Fallback without lock if self._connection_state == ConnectionState.CONNECTED: _LOGGER.debug( "Gateway %s: Already connected, skipping reconnection", self._gw_sn ) return self._connection_event.clear() self._connect_result = None _LOGGER.info("Gateway %s: Attempting reconnection...", self._gw_sn) try: # Attempt reconnection self._mqtt_client.reconnect() # Wait for connection result await asyncio.wait_for(self._connection_event.wait(), timeout=10) if self._connect_result == 0: # pyright: ignore[reportUnnecessaryComparison] _LOGGER.info( "Gateway %s: Reconnection successful", self._gw_sn, ) # Request version information self._request_version() else: _LOGGER.warning( "Gateway %s: Reconnection failed with code %s", self._gw_sn, self._connect_result, ) # Schedule another reconnection attempt if not self._shutdown_requested: self._connection_state = ConnectionState.RECONNECTING self._schedule_reconnect() except (asyncio.TimeoutError, OSError, ConnectionRefusedError) as err: _LOGGER.warning( "Gateway %s: Reconnection attempt failed: %s", self._gw_sn, err, ) # Schedule another reconnection attempt if not self._shutdown_requested: self._connection_state = ConnectionState.RECONNECTING self._schedule_reconnect() def stop_reconnection(self) -> None: """Stop any pending reconnection attempts. Should be called before disconnect() to prevent auto-reconnection. """ self._shutdown_requested = True if self._reconnect_task is not None: self._reconnect_task.cancel() self._reconnect_task = None _LOGGER.debug("Gateway %s: Cancelled pending reconnection", self._gw_sn) async def connect(self) -> None: # Auto-detect event loop if not provided at __init__ if self._loop is None: self._loop = asyncio.get_running_loop() # Initialize connection lock if not already done if self._connection_lock is None: self._connection_lock = asyncio.Lock() # Acquire lock for state transition check async with self._connection_lock: if self._connection_state == ConnectionState.CONNECTED: _LOGGER.debug( "Gateway %s: Already connected, skipping connect", self._gw_sn ) return if self._connection_state == ConnectionState.CONNECTING: _LOGGER.debug("Gateway %s: Connection already in progress", self._gw_sn) return self._connection_event.clear() self._connect_result = None self._shutdown_requested = False # Reset shutdown flag for new connection self._reconnect_delay = _RECONNECT_INITIAL_DELAY # Reset backoff self._connection_state = ConnectionState.CONNECTING self._mqtt_client.username_pw_set(self._username, self._passwd) if self._is_tls: await self._setup_ssl() try: _LOGGER.info( "Attempting connection to gateway %s at %s:%s (TLS: %s)", self._gw_sn, self._gw_ip, self._port, self._is_tls, ) self._mqtt_client.connect(self._gw_ip, self._port) self._mqtt_client.loop_start() await asyncio.wait_for( self._connection_event.wait(), timeout=_CONNECTION_TIMEOUT ) if self._connect_result is not None and self._connect_result == 0: _LOGGER.info( "Successfully connected to gateway %s at %s:%s", self._gw_sn, self._gw_ip, self._port, ) # Request version information self._request_version() return except asyncio.TimeoutError as err: # Critical: Stop the paho loop to prevent leaked background threads # that could cause duplicate message processing if a new gateway # instance is created later with the same client_id. self._mqtt_client.loop_stop() self._connection_state = ConnectionState.DISCONNECTED _LOGGER.error( "Connection timeout to gateway %s at %s:%s after %.0f seconds - check network connectivity", self._gw_sn, self._gw_ip, self._port, _CONNECTION_TIMEOUT, ) raise DaliGatewayError( f"Connection timeout to gateway {self._gw_sn}", self._gw_sn ) from err except (ConnectionRefusedError, OSError) as err: _LOGGER.error( "Network error connecting to gateway %s at %s:%s: %s - check if gateway is powered on and accessible", self._gw_sn, self._gw_ip, self._port, str(err), ) raise DaliGatewayError( f"Network error connecting to gateway {self._gw_sn}: {err}", self._gw_sn ) from err # Connection failed - clean up the paho loop before raising self._mqtt_client.loop_stop() self._connection_state = ConnectionState.DISCONNECTED if self._connect_result is not None and self._connect_result in (4, 5): _LOGGER.error( "Authentication failed for gateway %s (code %s) with credentials user='%s'. " "Please press the gateway button and retry", self._gw_sn, self._connect_result, self._username, ) raise DaliGatewayError( f"Authentication failed for gateway {self._gw_sn}. " "Please press the gateway button and retry", self._gw_sn, ) _LOGGER.error( "Connection failed for gateway %s with result code %s", self._gw_sn, self._connect_result, ) raise DaliGatewayError( f"Connection failed for gateway {self._gw_sn} " f"with code {self._connect_result}" ) async def disconnect(self) -> None: # Stop any pending reconnection attempts first self.stop_reconnection() # Use connection lock if available if self._connection_lock is not None: async with self._connection_lock: await self._disconnect_impl() else: await self._disconnect_impl() async def _disconnect_impl(self) -> None: """Internal disconnect implementation.""" try: self._mqtt_client.loop_stop() self._mqtt_client.disconnect() self._connection_event.clear() self._connection_state = ConnectionState.DISCONNECTED _LOGGER.info("Successfully disconnected from gateway %s", self._gw_sn) except Exception as exc: # pylint: disable=broad-exception-caught _LOGGER.error( "Error during disconnect from gateway %s: %s", self._gw_sn, exc ) raise DaliGatewayError( f"Failed to disconnect from gateway {self._gw_sn}: {exc}" ) from exc def _request_version(self) -> None: """Request gateway version information via MQTT.""" payload = { "cmd": "getVersion", "msgId": str(int(time.time())), "gwSn": self._gw_sn, } _LOGGER.debug("Gateway %s: Requesting version information", self._gw_sn) self._mqtt_client.publish(self._pub_topic, json.dumps(payload)) async def identify_gateway(self) -> None: """Make the gateway's indicator light blink to identify it physically. This command is sent via UDP multicast (not MQTT) according to the protocol specification. """ _LOGGER.debug("Sending identify command to gateway %s via UDP", self._gw_sn) await send_identify_gateway(self._gw_sn) async def read_group(self, group_id: int, channel: int = 0) -> Dict[str, Any]: group_key = (group_id, channel) # Create event for this specific group read self._read_group_events[group_key] = asyncio.Event() self._read_group_results.pop(group_key, None) # Clear any previous result payload: Dict[str, Any] = { "cmd": "readGroup", "msgId": str(int(time.time())), "gwSn": self._gw_sn, "channel": channel, "groupId": group_id, } _LOGGER.debug( "Gateway %s: Sending read group command for group %s channel %s", self._gw_sn, group_id, channel, ) self._mqtt_client.publish(self._pub_topic, json.dumps(payload)) try: await asyncio.wait_for( self._read_group_events[group_key].wait(), timeout=30.0 ) except asyncio.TimeoutError as err: _LOGGER.error( "Gateway %s: Timeout waiting for read group response for group %s channel %s", self._gw_sn, group_id, channel, ) # Cleanup self._read_group_events.pop(group_key, None) self._read_group_results.pop(group_key, None) raise DaliGatewayError( f"Timeout reading group {group_id} channel {channel} from gateway {self._gw_sn}", self._gw_sn, ) from err # Get result result = self._read_group_results.get(group_key) # Cleanup self._read_group_events.pop(group_key, None) self._read_group_results.pop(group_key, None) if not result: _LOGGER.error( "Gateway %s: Failed to read group %s channel %s - group may not exist", self._gw_sn, group_id, channel, ) raise DaliGatewayError( f"Group {group_id} channel {channel} not found on gateway {self._gw_sn}", self._gw_sn, ) _LOGGER.info( "Gateway %s: Group read completed - ID: %s, Channel: %s, Name: %s, Devices: %d", self._gw_sn, result["id"], result["channel"], result["name"], len(result["devices"]), ) return result async def read_scene(self, scene_id: int, channel: int = 0) -> Dict[str, Any]: scene_key = (scene_id, channel) # Create event for this specific scene read self._read_scene_events[scene_key] = asyncio.Event() self._read_scene_results.pop(scene_key, None) # Clear any previous result payload: Dict[str, Any] = { "cmd": "readScene", "msgId": str(int(time.time())), "gwSn": self._gw_sn, "channel": channel, "sceneId": scene_id, } _LOGGER.debug( "Gateway %s: Sending read scene command for scene %s channel %s", self._gw_sn, scene_id, channel, ) self._mqtt_client.publish(self._pub_topic, json.dumps(payload)) try: await asyncio.wait_for( self._read_scene_events[scene_key].wait(), timeout=30.0 ) except asyncio.TimeoutError as err: _LOGGER.error( "Gateway %s: Timeout waiting for read scene response for scene %s channel %s", self._gw_sn, scene_id, channel, ) # Cleanup self._read_scene_events.pop(scene_key, None) self._read_scene_results.pop(scene_key, None) raise DaliGatewayError( f"Timeout reading scene {scene_id} channel {channel} from gateway {self._gw_sn}", self._gw_sn, ) from err # Get result result = self._read_scene_results.get(scene_key) # Cleanup self._read_scene_events.pop(scene_key, None) self._read_scene_results.pop(scene_key, None) if not result: _LOGGER.error( "Gateway %s: Failed to read scene %s channel %s - scene may not exist", self._gw_sn, scene_id, channel, ) raise DaliGatewayError( f"Scene {scene_id} channel {channel} not found on gateway {self._gw_sn}", self._gw_sn, ) _LOGGER.info( "Gateway %s: Scene read completed - ID: %s, Channel: %s, Name: %s, Devices: %d", self._gw_sn, result["id"], result["channel"], result["name"], len(result["devices"]), ) return result async def discover_devices(self) -> list[Device]: self._devices_received = asyncio.Event() self._devices_result.clear() self._devices_seen_ids.clear() search_payload = { "cmd": "searchDev", "searchFlag": "exited", "msgId": str(int(time.time())), "gwSn": self._gw_sn, } _LOGGER.debug("Gateway %s: Sending device discovery command", self._gw_sn) self._mqtt_client.publish(self._pub_topic, json.dumps(search_payload)) try: await asyncio.wait_for(self._devices_received.wait(), timeout=30.0) except asyncio.TimeoutError: _LOGGER.warning( "Gateway %s: Timeout waiting for device discovery response", self._gw_sn ) _LOGGER.info( "Gateway %s: Device discovery completed, found %d device(s)", self._gw_sn, len(self._devices_result), ) return self._devices_result async def scan_bus(self, channels: list[int]) -> list[Device]: """Scan DALI bus for physical devices. This sends a searchDev command with searchFlag: "busDevice" to perform a physical bus scan, as opposed to discover_devices() which reads from gateway cache (searchFlag: "exited"). Args: channels: List of channel numbers to scan Returns: List of Device objects found on the bus Raises: asyncio.TimeoutError: If scan does not complete within BUS_SCAN_TIMEOUT seconds """ self._bus_scan_complete = asyncio.Event() self._bus_scan_result.clear() self._bus_scan_seen_ids.clear() self._bus_scan_cancelled = False self._bus_scanning = True self._bus_scan_channels = channels search_payload = { "cmd": "searchDev", "searchFlag": "busDevice", "channel": channels, "AddrAssignment": "auto", "msgId": str(int(time.time())), "gwSn": self._gw_sn, } _LOGGER.debug( "Gateway %s: Sending bus scan command for channels %s", self._gw_sn, channels, ) _LOGGER.debug( "Gateway %s: Bus scan payload: %s", self._gw_sn, json.dumps(search_payload) ) self._mqtt_client.publish(self._pub_topic, json.dumps(search_payload)) try: await asyncio.wait_for( self._bus_scan_complete.wait(), timeout=BUS_SCAN_TIMEOUT ) except asyncio.TimeoutError: _LOGGER.warning( "Gateway %s: Timeout waiting for bus scan to complete", self._gw_sn ) # Discard partial results on timeout self._bus_scan_result.clear() self._bus_scan_seen_ids.clear() self._bus_scanning = False raise self._bus_scanning = False # Check if scan was cancelled if self._bus_scan_cancelled: _LOGGER.info("Gateway %s: Bus scan was cancelled", self._gw_sn) # Discard partial results on cancellation self._bus_scan_result.clear() self._bus_scan_seen_ids.clear() raise BusScanCancelledError( f"Bus scan cancelled for gateway {self._gw_sn}", gw_sn=self._gw_sn, ) _LOGGER.info( "Gateway %s: Bus scan completed, found %d device(s)", self._gw_sn, len(self._bus_scan_result), ) return self._bus_scan_result async def stop_scan(self) -> None: """Stop an in-progress bus scan. This sends a searchDev command with searchFlag: "stop" to halt the current bus scan operation. The method proactively sets the completion event to unblock scan_bus(), regardless of gateway response. Note: Gateway response behavior to stop command is not fully documented and needs validation with real devices. """ stop_payload: dict[str, Any] = { "cmd": "searchDev", "searchFlag": "stop", "msgId": str(int(time.time())), "gwSn": self._gw_sn, } if self._bus_scan_channels: stop_payload["channel"] = self._bus_scan_channels stop_json = json.dumps(stop_payload) _LOGGER.debug( "Gateway %s: Sending bus scan stop command: %s", self._gw_sn, stop_json ) self._mqtt_client.publish(self._pub_topic, stop_json) # Mark as cancelled and proactively unblock scan_bus() # Don't rely on gateway response as its behavior is not fully known self._bus_scan_cancelled = True self._set_event_threadsafe(self._bus_scan_complete) _LOGGER.info("Gateway %s: Bus scan stop signal sent", self._gw_sn) async def discover_groups(self) -> list[Group]: """Discover all groups and read their detailed configuration with limited concurrency. Uses a semaphore to limit parallel read_group() calls, preventing MQTT message storms that can overload the event loop during multi-gateway startup. Returns only groups that were successfully read. Groups that fail to read (timeout, errors, etc.) are logged but not included in the result. """ # Phase 1: Discover basic group list self._groups_received = asyncio.Event() self._groups_result.clear() search_payload = { "cmd": "getGroup", "msgId": str(int(time.time())), "getFlag": "exited", "gwSn": self._gw_sn, } _LOGGER.debug("Gateway %s: Sending group discovery command", self._gw_sn) self._mqtt_client.publish(self._pub_topic, json.dumps(search_payload)) try: await asyncio.wait_for(self._groups_received.wait(), timeout=30.0) except asyncio.TimeoutError: _LOGGER.warning( "Gateway %s: Timeout waiting for group discovery response", self._gw_sn ) return [] if not self._groups_result: _LOGGER.info("Gateway %s: No groups found", self._gw_sn) return [] _LOGGER.info( "Gateway %s: Found %d group(s), reading details with limited concurrency...", self._gw_sn, len(self._groups_result), ) # Phase 2: Read detailed group data with limited concurrency # Limit concurrent reads to avoid MQTT message storms read_semaphore = asyncio.Semaphore(MAX_CONCURRENT_READS) async def read_group_with_limit(group_id: int, channel: int) -> Dict[str, Any]: async with read_semaphore: return await self.read_group(group_id, channel) # Store basic group info for reconstruction basic_groups: List[Tuple[int, str, int, str]] = [ (group.group_id, group.name, group.channel, group.area_id) for group in self._groups_result ] # Create read tasks with semaphore limit read_tasks = [ read_group_with_limit(group_id, channel) for group_id, _, channel, _ in basic_groups ] # Execute all reads with exception handling results = await asyncio.gather(*read_tasks, return_exceptions=True) # Phase 3: Construct Group objects with device data groups_with_devices: list[Group] = [] for (group_id, name, channel, area_id), result in zip(basic_groups, results): if isinstance(result, Exception): _LOGGER.error( "Gateway %s: Failed to read group %s (channel %s): %s", self._gw_sn, group_id, channel, result, ) continue # Successfully read group data - result is Dict[str, Any] result_dict = cast("Dict[str, Any]", result) try: group = Group( command_client=self, group_id=group_id, name=result_dict.get("name", name), channel=channel, area_id=result_dict.get("area_id", area_id), devices=result_dict.get("devices", []), ) groups_with_devices.append(group) except (KeyError, TypeError, ValueError) as e: _LOGGER.error( "Gateway %s: Failed to create Group object for group %s: %s", self._gw_sn, group_id, e, ) _LOGGER.info( "Gateway %s: Group discovery completed, %d/%d group(s) successfully read", self._gw_sn, len(groups_with_devices), len(basic_groups), ) return groups_with_devices async def discover_scenes(self) -> list[Scene]: """Discover all scenes and read their detailed configuration with limited concurrency. Uses a semaphore to limit parallel read_scene() calls, preventing MQTT message storms that can overload the event loop during multi-gateway startup. Returns only scenes that were successfully read. Scenes that fail to read (timeout, errors, etc.) are logged but not included in the result. """ # Phase 1: Discover basic scene list self._scenes_received = asyncio.Event() self._scenes_result.clear() search_payload = { "cmd": "getScene", "msgId": str(int(time.time())), "getFlag": "exited", "gwSn": self._gw_sn, } _LOGGER.debug("Gateway %s: Sending scene discovery command", self._gw_sn) self._mqtt_client.publish(self._pub_topic, json.dumps(search_payload)) try: await asyncio.wait_for(self._scenes_received.wait(), timeout=30.0) except asyncio.TimeoutError: _LOGGER.warning( "Gateway %s: Timeout waiting for scene discovery response", self._gw_sn ) return [] if not self._scenes_result: _LOGGER.info("Gateway %s: No scenes found", self._gw_sn) return [] _LOGGER.info( "Gateway %s: Found %d scene(s), reading details with limited concurrency...", self._gw_sn, len(self._scenes_result), ) # Phase 2: Read detailed scene data with limited concurrency # Limit concurrent reads to avoid MQTT message storms read_semaphore = asyncio.Semaphore(MAX_CONCURRENT_READS) async def read_scene_with_limit(scene_id: int, channel: int) -> Dict[str, Any]: async with read_semaphore: return await self.read_scene(scene_id, channel) # Store basic scene info for reconstruction basic_scenes: List[Tuple[int, str, int, str]] = [ (scene.scene_id, scene.name, scene.channel, scene.area_id) for scene in self._scenes_result ] # Create read tasks with semaphore limit read_tasks = [ read_scene_with_limit(scene_id, channel) for scene_id, _, channel, _ in basic_scenes ] # Execute all reads with exception handling results = await asyncio.gather(*read_tasks, return_exceptions=True) # Phase 3: Construct Scene objects with device data scenes_with_devices: list[Scene] = [] for (scene_id, name, channel, area_id), result in zip(basic_scenes, results): if isinstance(result, Exception): _LOGGER.error( "Gateway %s: Failed to read scene %s (channel %s): %s", self._gw_sn, scene_id, channel, result, ) continue # Successfully read scene data - result is Dict[str, Any] result_dict = cast("Dict[str, Any]", result) try: scene = Scene( command_client=self, scene_id=scene_id, name=result_dict.get("name", name), channel=channel, area_id=result_dict.get("area_id", area_id), devices=result_dict.get("devices", []), ) scenes_with_devices.append(scene) except (KeyError, TypeError, ValueError) as e: _LOGGER.error( "Gateway %s: Failed to create Scene object for scene %s: %s", self._gw_sn, scene_id, e, ) _LOGGER.info( "Gateway %s: Scene discovery completed, %d/%d scene(s) successfully read", self._gw_sn, len(scenes_with_devices), len(basic_scenes), ) return scenes_with_devices def command_write_dev( self, dev_type: str, channel: int, address: int, properties: List[Dict[str, Any]], ) -> None: self.add_request( "writeDev", dev_type, channel, address, { "devType": dev_type, "channel": channel, "address": address, "property": properties, }, ) def command_read_dev(self, dev_type: str, channel: int, address: int) -> None: self.add_request( "readDev", dev_type, channel, address, {"devType": dev_type, "channel": channel, "address": address}, ) def command_get_energy( self, dev_type: str, channel: int, address: int, year: int, month: int, day: int ) -> None: self.add_request( "getEnergy", dev_type, channel, address, { "devType": dev_type, "channel": channel, "address": address, "condition": {"year": year, "month": month, "day": day, "hour": []}, }, ) def command_write_group( self, group_id: int, channel: int, properties: List[Dict[str, Any]] ) -> None: command: Dict[str, Any] = { "cmd": "writeGroup", "msgId": str(int(time.time())), "gwSn": self._gw_sn, "channel": channel, "groupId": group_id, "data": properties, } command_json = json.dumps(command) _LOGGER.debug( "Gateway %s: Sending writeGroup %s", self._gw_sn, command_json, ) self._mqtt_client.publish(self._pub_topic, command_json) def command_write_scene(self, scene_id: int, channel: int) -> None: self._publish_command("writeScene", channel=channel, sceneId=scene_id) def command_set_sensor_on_off( self, dev_type: str, channel: int, address: int, value: bool ) -> None: self._publish_command( "setSensorOnOff", devType=dev_type, channel=channel, address=address, value=value, ) def command_get_sensor_on_off( self, dev_type: str, channel: int, address: int ) -> None: self._publish_command( "getSensorOnOff", devType=dev_type, channel=channel, address=address, ) def command_set_sensor_argv( self, dev_type: str, channel: int, address: int, param: SensorParamType ) -> None: """Set sensor parameters. Args: dev_type: Sensor device type code (e.g., "0201") channel: DALI channel number address: Device address param: Dictionary of sensor parameters to set (only provided fields will be set) """ # Convert snake_case Python keys to camelCase protocol keys param_dict = dict(param) data: Dict[str, Any] = { SENSOR_PARAM_KEY_MAP[python_key]: param_dict[python_key] for python_key in param_dict if python_key in SENSOR_PARAM_KEY_MAP } if not data: _LOGGER.warning( "Gateway %s: No valid parameters provided for setSensorArgv", self._gw_sn, ) return command: Dict[str, Any] = { "cmd": "setSensorArgv", "msgId": str(int(time.time())), "gwSn": self._gw_sn, "devType": dev_type, "channel": channel, "address": address, "data": data, } command_json = json.dumps(command) _LOGGER.debug( "Gateway %s: Sending setSensorArgv command: %s", self._gw_sn, command ) self._mqtt_client.publish(self._pub_topic, command_json) def command_get_sensor_argv( self, dev_type: str, channel: int, address: int ) -> None: """Get sensor parameters. Args: dev_type: Sensor device type code (e.g., "0201") channel: DALI channel number address: Device address """ self._publish_command( "getSensorArgv", devType=dev_type, channel=channel, address=address, ) def command_identify_dev(self, dev_type: str, channel: int, address: int) -> None: self._publish_command( "identifyDev", data={"devType": dev_type, "channel": channel, "address": address}, ) def command_get_dev_param(self, dev_type: str, channel: int, address: int) -> None: self._publish_command( "getDevParam", devType=dev_type, channel=channel, address=address, fromBus=False, ) def command_set_dev_param( self, dev_type: str, channel: int, address: int, param: DeviceParamType ) -> None: """Set device parameters. Args: dev_type: Device type code (e.g., "0101") channel: DALI channel number address: Device address param: Dictionary of device parameters to set (only provided fields will be set) """ paramer = self._build_parameter(param) if not paramer: _LOGGER.warning( "Gateway %s: No valid parameters provided for setDevParam", self._gw_sn ) return command: Dict[str, Any] = { "cmd": "setDevParam", "msgId": str(int(time.time())), "gwSn": self._gw_sn, "data": [ { "devType": dev_type, "channel": channel, "address": address, "paramer": paramer, } ], } command_json = json.dumps(command) _LOGGER.debug( "Gateway %s: Sending setDevParam command: %s", self._gw_sn, command ) self._mqtt_client.publish(self._pub_topic, command_json) def command_set_dev_params(self, items: Sequence[DeviceParamCommand]) -> None: """Set parameters for multiple targets in one MQTT message.""" data: List[Dict[str, Any]] = [] for item in items: paramer = self._build_parameter(item["param"]) if not paramer: _LOGGER.warning( "Gateway %s: No valid parameters provided for %s", self._gw_sn, item, ) continue data.append( { "devType": item["dev_type"], "channel": item["channel"], "address": item["address"], "paramer": paramer, } ) if not data: _LOGGER.warning( "Gateway %s: No valid setDevParam payloads provided for batch send", self._gw_sn, ) return command: Dict[str, Any] = { "cmd": "setDevParam", "msgId": str(int(time.time())), "gwSn": self._gw_sn, "data": data, } command_json = json.dumps(command) _LOGGER.debug( "Gateway %s: Sending batch setDevParam command: %s", self._gw_sn, command, ) self._mqtt_client.publish(self._pub_topic, command_json) def restart_gateway(self) -> None: """Restart the gateway.""" _LOGGER.debug("Gateway %s: Sending restart command", self._gw_sn) self._publish_command("restartGateway")