"""This represents a velbus module.""" from __future__ import annotations import asyncio from collections.abc import Awaitable, Callable import importlib.resources import inspect import json import logging import pathlib import struct import sys from typing import TYPE_CHECKING from aiofile import async_open if TYPE_CHECKING: from velbusaio.controller import Controller from velbusaio import channels as channels_module, properties as properties_module from velbusaio.channels import ( Button, ButtonCounter, Channel, Dimmer, Temperature as TemperatureChannelType, ) from velbusaio.command_registry import commandRegistry from velbusaio.const import PRIORITY_LOW, SCAN_MODULEINFO_TIMEOUT_INITIAL from velbusaio.helpers import h2, handle_match, keys_exists from velbusaio.message import Message from velbusaio.messages.blind_status import ( BlindStatusMessage, BlindStatusNg20Message, BlindStatusNgMessage, ) from velbusaio.messages.bus_error_counter_status import BusErrorCounterStatusMessage from velbusaio.messages.cancel_forced_off import CancelForcedOff from velbusaio.messages.cancel_forced_on import CancelForcedOn from velbusaio.messages.cancel_inhibit import CancelInhibit from velbusaio.messages.channel_name_part1 import ( ChannelNamePart1Message, ChannelNamePart1Message2, ChannelNamePart1Message3, ) from velbusaio.messages.channel_name_part2 import ( ChannelNamePart2Message, ChannelNamePart2Message2, ChannelNamePart2Message3, ) from velbusaio.messages.channel_name_part3 import ( ChannelNamePart3Message, ChannelNamePart3Message2, ChannelNamePart3Message3, ) from velbusaio.messages.channel_name_request import ( COMMAND_CODE as CHANNEL_NAME_REQUEST_COMMAND_CODE, ChannelNameRequestMessage, ) from velbusaio.messages.clear_led import ClearLedMessage from velbusaio.messages.counter_status import CounterStatusMessage from velbusaio.messages.counter_status_request import CounterStatusRequestMessage from velbusaio.messages.counter_value import CounterValueMessage from velbusaio.messages.dali_device_settings import ( DaliDeviceSettingMsg, DeviceType as DaliDeviceType, DeviceTypeMsg as DaliDeviceTypeMsg, MemberOfGroupMsg, ) from velbusaio.messages.dali_device_settings_request import ( COMMAND_CODE as DALI_DEVICE_SETTINGS_REQUEST_COMMAND_CODE, DaliDeviceSettingsRequest, ) from velbusaio.messages.dali_dim_value_status import DimValueStatus from velbusaio.messages.dimmer_channel_status import DimmerChannelStatusMessage from velbusaio.messages.dimmer_status import DimmerStatusMessage from velbusaio.messages.fast_blinking_led import FastBlinkingLedMessage from velbusaio.messages.forced_off import ForcedOff from velbusaio.messages.forced_on import ForcedOn from velbusaio.messages.inhibit import Inhibit from velbusaio.messages.memory_data import MemoryDataMessage from velbusaio.messages.memory_data_block import MemoryDataBlockMessage from velbusaio.messages.module_status import ( ModuleStatusGP4PirMessage, ModuleStatusMessage, ModuleStatusMessage2, ModuleStatusPirMessage, ) from velbusaio.messages.module_status_request import ModuleStatusRequestMessage from velbusaio.messages.module_type_request import ModuleTypeRequestMessage from velbusaio.messages.psu_load import PsuLoadMessage from velbusaio.messages.psu_values import PsuValuesMessage from velbusaio.messages.push_button_status import PushButtonStatusMessage from velbusaio.messages.raw import MeteoRawMessage, SensorRawMessage from velbusaio.messages.read_data_block_from_memory import ( ReadDataBlockFromMemoryMessage, ) from velbusaio.messages.read_data_from_memory import ReadDataFromMemoryMessage from velbusaio.messages.relay_status import ( RelayStatusMessage, RelayStatusMessage2, RelayStatusMessage3, ) from velbusaio.messages.sensor_temperature import SensorTemperatureMessage from velbusaio.messages.set_led import SetLedMessage from velbusaio.messages.slider_status import SliderStatusMessage from velbusaio.messages.slow_blinking_led import SlowBlinkingLedMessage from velbusaio.messages.temp_sensor_status import TempSensorStatusMessage from velbusaio.messages.update_led_status import UpdateLedStatusMessage from velbusaio.properties import Property class Module: """Abstract class for Velbus hardware modules.""" @classmethod def factory( cls, module_address: int, module_type: int, serial: int | None = None, memorymap: int | None = None, build_year: int | None = None, build_week: int | None = None, cache_dir: str | None = None, on_module_found: Callable[[Module], Awaitable[None]] | None = None, ) -> Module: """Module factory method.""" if module_type in {0x45, 0x5A}: return VmbDali( module_address, module_type, serial, memorymap, build_year, build_week, cache_dir, on_module_found, ) return Module( module_address, module_type, serial, memorymap, build_year, build_week, cache_dir, on_module_found, ) def __init__( self, module_address: int, module_type: int, serial: int | None = None, memorymap: int | None = None, build_year: int | None = None, build_week: int | None = None, cache_dir: str | None = None, on_module_found: Callable[[Module], Awaitable[None]] | None = None, ) -> None: """Initialize Module object.""" self._address = module_address self._type = int(module_type) self._data = {} self._name = None self._name_buffer: dict[ int, str ] = {} # temporary buffer while assembling name from memory blocks self._sub_address = {} self.serial = serial self.memory_map_version = memorymap self.build_year = build_year self.build_week = build_week self._cache_dir = cache_dir self._is_loading = False self._got_status = asyncio.Event() self._got_status.clear() self._channels: dict[int, Channel] = {} self._properties: dict[str, Property] = {} self.loaded = False self._use_cache = True self._loaded_cache = {} self._on_module_found: Callable[[Module], Awaitable[None]] | None = ( on_module_found ) async def wait_for_status_messages(self) -> None: """Wait for status messages to be received.""" try: await asyncio.wait_for(self._got_status.wait(), 2) except TimeoutError: self._log.warning(f"Timeout waiting for status messages for: {self}") def get_initial_timeout(self) -> int: """Get initial timeout for scanning module info.""" return SCAN_MODULEINFO_TIMEOUT_INITIAL async def initialize( self, writer: Callable[[Message], Awaitable[None]], controller: Controller ) -> None: """Initialize the module.""" self._controller = controller self._log = logging.getLogger("velbus-module") # Build message handler dispatch table self._message_handlers = self._build_message_handlers() # load the protocol data try: # Load global.json first global_data = {} try: if sys.version_info >= (3, 13): with importlib.resources.path( __name__, "module_spec/global.json" ) as fspath: async with async_open(fspath) as global_file: global_data = json.loads(await global_file.read()) else: async with async_open( str( importlib.resources.files(__name__.split(".")[0]).joinpath( "module_spec/global.json" ) ) ) as global_file: global_data = json.loads(await global_file.read()) self._log.debug("Global module spec loaded") except FileNotFoundError: self._log.debug("No global module spec found") # Load module-specific data if sys.version_info >= (3, 13): with importlib.resources.path( __name__, f"module_spec/{h2(self._type)}.json" ) as fspath: async with async_open(fspath) as protocol_file: self._data = json.loads(await protocol_file.read()) else: async with async_open( str( importlib.resources.files(__name__.split(".")[0]).joinpath( f"module_spec/{h2(self._type)}.json" ) ) ) as protocol_file: self._data = json.loads(await protocol_file.read()) # Merge global data into module data (module-specific takes precedence) for key, value in global_data.items(): if key not in self._data: self._data[key] = value elif isinstance(value, dict) and isinstance(self._data[key], dict): # Deep merge for nested dictionaries self._data[key] = {**value, **self._data[key]} self._log.debug(f"Module spec {h2(self._type)} loaded") except FileNotFoundError: self._log.warning(f"No module spec for {h2(self._type)}") self._data = {} # set some params from the velbus controller self._writer = writer for chan in self._channels.values(): chan.set_writer(writer) def cleanupSubChannels(self) -> None: """Cleanup subchannels that are not defined.""" # TODO: 21/11/2022 DannyDeGaspari: Fix needed # Care should be taken for this function, not all subaddresses have their channels on multiples of 8. # The last subaddress contain typically the temperature channels, has more then 8 channels # and doesn't start on a boundary of 8. # E.g. The VMBGP4 has one subaddress, so since the second subaddress is not defined, # this function will delete channels 17-24 while 17 and 18 belong to the temperature channels. # # The solution would be that this functions knows were the temperature channels are located # and/or what the max number of subaddresses are for each module. # if self._sub_address == {} and self.loaded: # raise Exception("No subaddresses defined") for sub in range(1, 4): if sub not in self._sub_address: for i in range(((sub * 8) + 1), (((sub + 1) * 8) + 1)): if i in self._channels and not isinstance( self._channels[i], TemperatureChannelType ): del self._channels[i] async def _cache(self) -> None: if not self._use_cache: return cfile = pathlib.Path(f"{self._cache_dir}/{self._address}.json") async with async_open(cfile, "w") as fl: await fl.write(json.dumps(self.to_cache(), indent=4)) def __getstate__(self) -> dict: """Get state for pickling.""" d = self.__dict__ return {k: d[k] for k in d if k not in {"_writer", "_log", "_controller"}} def __setstate__(self, state: dict) -> None: """Set state for unpickling.""" self.__dict__ = state def __repr__(self) -> str: """Return string representation of the module.""" return f"<{self._name} type:{self._type} address:{self._address} loaded:{self.loaded} loading:{self._is_loading} channels: {self._channels} properties: {self._properties}>" def __str__(self) -> str: """Return string representation of the module.""" return self.__repr__() def to_cache(self) -> dict: """Build cache dict.""" d = { "name": self._name if isinstance(self._name, str) else "", "channels": {}, "sub_addresses": {}, "properties": {}, } for num, chan in self._channels.items(): d["channels"][num] = chan.to_cache() for num, address in self._sub_address.items(): d["sub_addresses"][num] = address for num, prop in self._properties.items(): d["properties"][num] = prop.to_cache() return d def get_address(self) -> int: """Get the module address.""" return self._address def get_addresses(self) -> list: """Get all addresses for this module.""" res = [self._address] res.extend(self._sub_address.values()) return res def get_sub_address_dict(self) -> dict[int, int]: """Return the sub addresses dict.""" return self._sub_address def is_channel_active(self, channel_num: int) -> bool: """Check if a channel is active based on sub-address configuration.""" # Channels 1-8 are always on the master address (active) if channel_num <= 8: return True # Calculate sub-address index (1, 2, 3...) # Block 1: 1-8 (index 0 / master) # Block 2: 9-16 (index 1) # Block 3: 17-24 (index 2) # Block 4: 25-32 (index 3) sub_idx = (channel_num - 1) // 8 # Check if this sub-index exists in the active sub-addresses return sub_idx in self._sub_address def add_subaddress(self, num, addr) -> None: """Add a subaddress to this module.""" self._sub_address[num] = addr def get_type(self) -> int: """Get the module type.""" return self._type def get_type_name(self) -> str: """Get the module type name.""" if "Type" in self._data: return self._data["Type"] return "UNKNOWN" def get_serial(self) -> str | None: """Get the module serial number.""" return self.serial def get_name(self) -> str: """Get the module name.""" if isinstance(self._name, dict): return "" return self._name def get_sw_version(self) -> str: """Get the module software version.""" return f"{self.build_year}.{self.build_week}" def calc_channel_offset(self, address: int) -> int: """Calculate channel offset based on address.""" _channel_offset = 0 if self._address != address: for _sub_addr_key, _sub_addr_val in self._sub_address.items(): if _sub_addr_val == address: _channel_offset = 8 * _sub_addr_key break return _channel_offset def on_connect(self, meth: Callable[[], Awaitable[None]]) -> None: """Register a coroutine to be called on connect.""" self._controller.add_connect_callback(meth) def remove_on_connect(self, meth: Callable[[], Awaitable[None]]) -> None: """Remove a previously registered on connect coroutine.""" self._controller.remove_connect_callback(meth) def on_disconnect(self, meth: Callable[[], Awaitable[None]]) -> None: """Register a coroutine to be called on disconnect.""" self._controller.add_disconnect_callback(meth) def remove_on_disconnect(self, meth: Callable[[], Awaitable[None]]) -> None: """Remove a previously registered on disconnect coroutine.""" self._controller.remove_disconnect_callback(meth) async def _trigger_load_finished_callbacks(self) -> None: """Trigger all registered on load finished callbacks.""" if self._on_module_found: try: await self._on_module_found(self) except (RuntimeError, ValueError, TypeError, AttributeError) as e: self._log.error(f"Error in on_module_found callback: {e}") @property def is_connected(self) -> bool: """Return if the module is connected.""" return self._controller.connected def _build_message_handlers(self) -> dict: """Build dispatch table for message handlers.""" return { # Channel name messages ChannelNamePart1Message: self._handle_channel_name_part1, ChannelNamePart1Message2: self._handle_channel_name_part1, ChannelNamePart1Message3: self._handle_channel_name_part1, ChannelNamePart2Message: self._handle_channel_name_part2, ChannelNamePart2Message2: self._handle_channel_name_part2, ChannelNamePart2Message3: self._handle_channel_name_part2, ChannelNamePart3Message: self._handle_channel_name_part3, ChannelNamePart3Message2: self._handle_channel_name_part3, ChannelNamePart3Message3: self._handle_channel_name_part3, # Memory messages MemoryDataMessage: self._process_memory_data_message, MemoryDataBlockMessage: self._process_memory_data_block_message, # Status messages BusErrorCounterStatusMessage: self._handle_bus_error_counter, RelayStatusMessage: self._handle_relay_status, RelayStatusMessage2: self._handle_relay_status, RelayStatusMessage3: self._handle_relay_status3, ForcedOn: self._handle_forced_on, ForcedOff: self._handle_forced_off, Inhibit: self._handle_inhibit, CancelForcedOn: self._handle_cancel_forced, CancelForcedOff: self._handle_cancel_forced, CancelInhibit: self._handle_cancel_forced, # Temperature messages SensorTemperatureMessage: self._handle_sensor_temperature, TempSensorStatusMessage: self._handle_temp_sensor_status, # Button and module status messages PushButtonStatusMessage: self._handle_push_button_status, ModuleStatusMessage: self._handle_module_status, ModuleStatusMessage2: self._handle_module_status2, CounterStatusMessage: self._handle_counter_status, ModuleStatusPirMessage: self._handle_module_status_pir, ModuleStatusGP4PirMessage: self._handle_module_status_gp4_pir, # LED messages UpdateLedStatusMessage: self._handle_led_status, SetLedMessage: self._handle_set_led, ClearLedMessage: self._handle_clear_led, SlowBlinkingLedMessage: self._handle_slow_blinking_led, FastBlinkingLedMessage: self._handle_fast_blinking_led, # Dimmer and blind messages DimmerChannelStatusMessage: self._handle_dimmer_status, DimmerStatusMessage: self._handle_dimmer_status, SliderStatusMessage: self._handle_slider_status, BlindStatusNgMessage: self._handle_blind_status_ng, BlindStatusNg20Message: self._handle_blind_status_ng20, BlindStatusMessage: self._handle_blind_status, # Sensor messages MeteoRawMessage: self._handle_meteo_raw, SensorRawMessage: self._handle_sensor_raw, CounterValueMessage: self._handle_counter_value, DimValueStatus: self._handle_dim_value_status, # PSU messages PsuLoadMessage: self._handle_psu_load, PsuValuesMessage: self._handle_psu_values, } async def _handle_channel_name_part1(self, message: Message) -> None: """Handle channel name part 1 messages.""" await self._process_channel_name_message(1, message) await self._cache() async def _handle_channel_name_part2(self, message: Message) -> None: """Handle channel name part 2 messages.""" await self._process_channel_name_message(2, message) await self._cache() async def _handle_channel_name_part3(self, message: Message) -> None: """Handle channel name part 3 messages.""" await self._process_channel_name_message(3, message) await self._cache() async def _handle_bus_error_counter( self, message: BusErrorCounterStatusMessage ) -> None: """Handle bus error counter status messages.""" await self._update_property( "BusErrorTx", {"_cur": message.transmit_error_counter} ) await self._update_property( "BusErrorRx", {"_cur": message.receive_error_counter} ) await self._update_property("BusOffCounter", {"_cur": message.bus_off_counter}) async def _handle_relay_status( self, message: RelayStatusMessage | RelayStatusMessage2 ) -> None: """Handle relay status messages.""" await self._update_channel( message.channel, { "on": message.is_on(), "inhibit": message.is_inhibited(), "forced_on": message.is_forced_on(), "disabled": message.is_disabled(), }, ) async def _handle_relay_status3(self, message: RelayStatusMessage3) -> None: """Handle relay status 3 messages.""" for channel in range(1, 9): await self._update_channel( channel, { "on": message.is_on(channel), "inhibit": message.is_inhibited(channel), "forced_on": message.is_forced_on(channel), "forced_off": message.is_forced_off(channel), "disabled": message.is_program_disabled(channel), }, ) async def _handle_forced_on(self, message: ForcedOn) -> None: """Handle forced on messages.""" await self._update_channel( message.channel, {"forced_on": True, "forced_off": False, "inhibit": False}, ) async def _handle_forced_off(self, message: ForcedOff) -> None: """Handle forced off messages.""" await self._update_channel( message.channel, {"forced_on": False, "forced_off": True, "inhibit": False}, ) async def _handle_inhibit(self, message: Inhibit) -> None: """Handle inhibit messages.""" await self._update_channel( message.channel, {"forced_on": False, "forced_off": False, "inhibit": True}, ) async def _handle_cancel_forced( self, message: CancelForcedOn | CancelForcedOff | CancelInhibit ) -> None: """Handle cancel forced/inhibit messages.""" await self._update_channel( message.channel, {"forced_on": False, "forced_off": False, "inhibit": False}, ) async def _handle_sensor_temperature( self, message: SensorTemperatureMessage ) -> None: """Handle sensor temperature messages.""" chan = self._translate_channel_name(self._data["TemperatureChannel"]) await self._channels[chan].maybe_update_temperature( message.getCurTemp(), 1 / 64 ) await self._update_channel( chan, { "min": message.getMinTemp(), "max": message.getMaxTemp(), }, ) async def _handle_temp_sensor_status( self, message: TempSensorStatusMessage ) -> None: """Handle temperature sensor status messages.""" chan = self._translate_channel_name(self._data["TemperatureChannel"]) if chan in self._channels: await self._update_channel( chan, { "target": message.target_temp, "cmode": message.mode_str, "cstatus": message.status_str, "sleep_timer": message.sleep_timer, "cool_mode": message.cool_mode, }, ) await self._channels[chan].maybe_update_temperature( message.current_temp, 1 / 2 ) # Update thermostat channels channel_name_to_msg_prop_map = { "Heater": "heater", "Boost heater/cooler": "boost", "Pump": "pump", "Cooler": "cooler", "Temperature alarm 1": "alarm1", "Temperature alarm 2": "alarm2", "Temperature alarm 3": "alarm3", "Temperature alarm 4": "alarm4", } for channel_str in self._data["Channels"]: if keys_exists(self._data, "Channels", channel_str, "Type"): if self._data["Channels"][channel_str]["Type"] == "ThermostatChannel": channel = self._translate_channel_name(channel_str) channel_name = self._data["Channels"][channel_str]["Name"] if ( channel in self._channels and channel_name in channel_name_to_msg_prop_map ): await self._update_channel( channel, { "closed": getattr( message, channel_name_to_msg_prop_map[channel_name] ) }, ) async def _handle_push_button_status( self, message: PushButtonStatusMessage, channel_offset: int ) -> None: """Handle push button status messages.""" _update_buttons = False for channel_types in self._data["Channels"]: if keys_exists(self._data, "Channels", channel_types, "Type"): if self._data["Channels"][channel_types]["Type"] in ( "Button", "Sensor", "ButtonCounter", "Relay", ): _update_buttons = True break if _update_buttons: for channel_id in range(1, 9): channel = self._translate_channel_name(channel_id + channel_offset) if channel_id in message.closed: await self._update_channel(channel, {"closed": True, "on": True}) if channel_id in message.closed_long: await self._update_channel(channel, {"long": True}) if channel_id in message.opened: await self._update_channel( channel, {"closed": False, "long": False, "on": False} ) async def _handle_module_status( self, message: ModuleStatusMessage, channel_offset: int ) -> None: """Handle module status messages.""" for channel_id in range(1, 9): channel = self._translate_channel_name(channel_id + channel_offset) if channel_id in message.closed: await self._update_channel(channel, {"closed": True}) elif channel in self._channels and isinstance( self._channels[channel], (Button, ButtonCounter) ): await self._update_channel(channel, {"closed": False}) async def _handle_module_status2( self, message: ModuleStatusMessage2, channel_offset: int ) -> None: """Handle module status 2 messages.""" for channel_id in range(1, 9): channel = self._translate_channel_name(channel_id + channel_offset) if channel_id in message.closed: await self._update_channel(channel, {"closed": True}) elif channel in self._channels and isinstance( self._channels[channel], (Button, ButtonCounter) ): await self._update_channel(channel, {"closed": False}) if channel_id in message.enabled: await self._update_channel(channel, {"enabled": True}) elif channel in self._channels and isinstance( self._channels[channel], (Button, ButtonCounter) ): await self._update_channel(channel, {"enabled": False}) await self._update_property( "selected_program", {"selected_program_str": message.selected_program_str}, ) async def _handle_counter_status(self, message: CounterStatusMessage) -> None: """Handle counter status messages.""" if isinstance(self._channels.get(message.channel), ButtonCounter): channel = self._translate_channel_name(message.channel) await self._update_channel( channel, { "pulses": message.pulses, "counter": message.counter, "delay": message.delay, }, ) async def _handle_module_status_pir(self, message: ModuleStatusPirMessage) -> None: """Handle PIR module status messages.""" await self._update_property("light_value", {"cur": message.light_value}) await self._update_channel(1, {"closed": message.dark}) await self._update_channel(2, {"closed": message.light}) await self._update_channel(3, {"closed": message.motion1}) await self._update_channel(4, {"closed": message.light_motion1}) await self._update_channel(5, {"closed": message.motion2}) await self._update_channel(6, {"closed": message.light_motion2}) if 7 in self._channels: await self._update_channel(7, {"closed": message.low_temp_alarm}) if 8 in self._channels: await self._update_channel(8, {"closed": message.high_temp_alarm}) await self._update_property( "selected_program", {"selected_program_str": message.selected_program_str}, ) async def _handle_module_status_gp4_pir( self, message: ModuleStatusGP4PirMessage, channel_offset: int ) -> None: """Handle GP4 PIR module status messages.""" await self._update_property("light_value", {"cur": message.light_value}) for channel_id in range(1, 9): channel = self._translate_channel_name(channel_id + channel_offset) await self._update_channel( channel, {"closed": channel_id in message.closed} ) if channel in self._channels and isinstance( self._channels[channel], Button ): await self._update_channel( channel, {"enabled": channel_id in message.enabled} ) await self._update_property( "selected_program", {"selected_program_str": message.selected_program_str}, ) async def _handle_led_status( self, message: UpdateLedStatusMessage, channel_offset: int ) -> None: """Handle LED status update messages.""" for channel_id in range(1, 9): channel = self._translate_channel_name(channel_id + channel_offset) if channel_id in message.led_slow_blinking: await self._update_channel(channel, {"led_state": "slow"}) elif channel_id in message.led_fast_blinking: await self._update_channel(channel, {"led_state": "fast"}) elif channel_id in message.led_on: await self._update_channel(channel, {"led_state": "on"}) else: await self._update_channel(channel, {"led_state": "off"}) async def _handle_set_led( self, message: SetLedMessage, channel_offset: int ) -> None: """Handle set LED messages.""" for channel_id in range(1, 9): channel = self._translate_channel_name(channel_id + channel_offset) if channel_id in message.leds: await self._update_channel(channel, {"led_state": "on"}) async def _handle_clear_led( self, message: ClearLedMessage, channel_offset: int ) -> None: """Handle clear LED messages.""" for channel_id in range(1, 9): channel = self._translate_channel_name(channel_id + channel_offset) if channel_id in message.leds: await self._update_channel(channel, {"led_state": "off"}) async def _handle_slow_blinking_led( self, message: SlowBlinkingLedMessage, channel_offset: int ) -> None: """Handle slow blinking LED messages.""" for channel_id in range(1, 9): channel = self._translate_channel_name(channel_id + channel_offset) if channel_id in message.leds: await self._update_channel(channel, {"led_state": "slow"}) async def _handle_fast_blinking_led( self, message: FastBlinkingLedMessage, channel_offset: int ) -> None: """Handle fast blinking LED messages.""" for channel_id in range(1, 9): channel = self._translate_channel_name(channel_id + channel_offset) if channel_id in message.leds: await self._update_channel(channel, {"led_state": "fast"}) async def _handle_dimmer_status( self, message: DimmerChannelStatusMessage | DimmerStatusMessage ) -> None: """Handle dimmer status messages.""" channel = self._translate_channel_name(message.channel) await self._update_channel(channel, {"state": message.cur_dimmer_state()}) async def _handle_slider_status(self, message: SliderStatusMessage) -> None: """Handle slider status messages.""" channel = self._translate_channel_name(message.channel) await self._update_channel(channel, {"state": message.cur_slider_state()}) async def _handle_blind_status_ng(self, message: BlindStatusNgMessage) -> None: """Handle blind status NG messages.""" channel = self._translate_channel_name(message.channel) await self._update_channel( channel, {"state": message.status, "position": message.position} ) async def _handle_blind_status_ng20(self, message: BlindStatusNg20Message) -> None: """Handle blind status NG20 messages.""" if message.status == 0: # stopped messages contain the position for both channels, so we need to update both channels for i in range(2): channel = self._translate_channel_name(message.channel[i]) await self._update_channel( channel, {"state": message.status, "position": message.position[i]} ) else: channel = self._translate_channel_name(message.channel) await self._update_channel( channel, {"state": message.status, "position": message.position} ) async def _handle_blind_status(self, message: BlindStatusMessage) -> None: """Handle blind status messages.""" channel = self._translate_channel_name(message.channel) await self._update_channel(channel, {"state": message.status}) async def _handle_meteo_raw(self, message: MeteoRawMessage) -> None: """Handle meteo raw messages.""" await self._update_channel(11, {"cur": message.rain}) await self._update_channel(12, {"cur": message.light}) await self._update_channel(13, {"cur": message.wind}) async def _handle_sensor_raw(self, message: SensorRawMessage) -> None: """Handle sensor raw messages.""" await self._update_channel( message.sensor, {"cur": message.value, "unit": message.unit} ) async def _handle_counter_value(self, message: CounterValueMessage) -> None: """Handle counter value messages.""" await self._update_channel( message.channel, {"power": message.power, "energy": message.energy} ) async def _handle_dim_value_status(self, message: DimValueStatus) -> None: """Handle dim value status messages.""" for offset, dim_value in enumerate(message.dim_values): channel = message.channel + offset await self._update_channel(channel, {"state": dim_value}) async def _handle_psu_load(self, message: PsuLoadMessage) -> None: """Handle PSU load messages.""" await self._update_property("psu_load_out", {"cur": message.out}) await self._update_property("psu_load_1", {"cur": message.load_1}) await self._update_property("psu_load_2", {"cur": message.load_2}) async def _handle_psu_values(self, message: PsuValuesMessage) -> None: """Handle PSU values messages.""" suffix = "out" if message.channel == 3 else f"{message.channel}" await self._update_property(f"psu_power_{suffix}", {"cur": message.watt}) await self._update_property(f"psu_voltage_{suffix}", {"cur": message.volt}) await self._update_property(f"psu_current_{suffix}", {"cur": message.amp}) async def on_message(self, message: Message) -> None: """Process received message.""" self._log.debug(f"RX: {message}") _channel_offset = self.calc_channel_offset(message.address) # Use dispatch table for message handling handler = self._message_handlers.get(type(message)) if handler: # Check if handler needs channel_offset parameter sig = inspect.signature(handler) if "channel_offset" in sig.parameters: await handler(message, _channel_offset) else: await handler(message) # Notify status self._got_status.set() async def _update_channel(self, channel: int, updates: dict): try: await self._channels[channel].update(updates) except KeyError: self._log.info( f"channel {channel} does not exist for module @ address {self}" ) async def _update_property(self, property_name: str, updates: dict): try: await self._properties[property_name].update(updates) except KeyError: self._log.info( f"property {property_name} does not exist for module @ address {self}" ) def get_channels(self) -> dict: """List all channels for this module.""" return self._channels def get_properties(self) -> dict[str, Property]: """List all properties for this module.""" return self._properties async def load_from_vlp(self, vlp_data) -> None: """Initialize the module from VLP data.""" self._is_loading = True self._use_cache = False self._name = vlp_data.get_name() self._data["Channels"] = vlp_data.get_channels() await self._load_default_channels() await self._load_properties() for chan in self._channels.values(): chan.set_loaded(True) self.loaded = True self._is_loading = False await self._request_module_status() await self._trigger_load_finished_callbacks() async def load(self, from_cache: bool = False) -> None: """Initialize the module.""" # start the loading self._is_loading = True # see if we have a cache cache = await self._get_cache() self._loaded_cache = cache # load default channels await self._load_default_channels() await self._load_properties() # load the data from memory ( the stuff that we need) if "name" in cache and isinstance(cache["name"], str) and cache["name"] != "": self._name = cache["name"] else: await self.__load_memory() # load the sub addresses from cache if it's available if (self._use_cache or from_cache) and "sub_addresses" in cache: for num, addr in cache["sub_addresses"].items(): self._sub_address[int(num)] = int(addr) else: # Submit ModuleType request to trigger discovery of sub addresses. # Do not cache immediately here: sub addresses are populated asynchronously # when the response is handled; caching now could store an empty mapping. await self._writer(ModuleTypeRequestMessage(self._address)) # load the module status # await self._request_module_status() # load the channel names if "channels" in cache: for num, chan in cache["channels"].items(): chan_num = self._translate_channel_name(num) if chan_num not in self._channels: continue self._channels[chan_num].set_name(chan["name"]) if "subdevice" in chan: self._channels[chan_num].set_sub_device(chan["subdevice"]) else: self._channels[chan_num].set_sub_device(False) if "Unit" in chan: self._channels[chan_num].set_unit(chan["Unit"]) self._channels[chan_num].set_loaded(True) else: await self._request_channel_name() # load the module specific stuff self._load() # stop the loading self._is_loading = False await self._request_module_status() await self._trigger_load_finished_callbacks() async def _get_cache(self): cfile = pathlib.Path(f"{self._cache_dir}/{self._address}.json") try: async with async_open(cfile, "r") as fl: cache = json.loads(await fl.read()) except OSError: cache = {} except (json.JSONDecodeError, KeyError, ValueError): self._log.warning( "Cache file for module %s is corrupt, removing it", self._address, ) cfile.unlink(missing_ok=True) cache = {} return cache def _load(self) -> None: """Method for per module type loading.""" def number_of_channels(self) -> int: """Retrieve the number of available channels in this module. :return: int """ if not len(self._channels): return 0 return max(self._channels.keys()) async def set_memo_text(self, txt: str) -> None: """Set memo text property.""" if "memo_text" not in self._properties: return await self._properties["memo_text"].set(txt) async def _process_memory_data_block_message( self, message: MemoryDataBlockMessage ) -> None: addr = f"{message.high_address:02X}{message.low_address:02X}" if "Memory" not in self._data: return # TODO this can also be SensorName, implement that also if "ModuleName" not in self._data["Memory"]: return addr_data = self._data["Memory"]["ModuleName"] if isinstance(self._name, str): # Name already fully assembled, nothing to do return # Parse address ranges: "00DD-00E9;01DD-01E9;02DD-02E9;03DD-03E9;04DD-04E8" ranges = [] byte_offset = 0 for block in addr_data.split(";"): start_str, end_str = block.split("-") start_addr = int("0x" + start_str, 0) end_addr = int("0x" + end_str, 0) ranges.append((start_addr, end_addr, byte_offset)) byte_offset += end_addr - start_addr + 1 # Check if incoming address falls within any range incoming_addr = int("0x" + addr, 0) for start_addr, end_addr, range_byte_offset in ranges: if start_addr <= incoming_addr <= end_addr: # Calculate the byte position within the overall module name position_in_range = incoming_addr - start_addr base_position = range_byte_offset + position_in_range # Store each byte in the temporary buffer, not in _name itself. # _name is only set to a string once all bytes have been received. for i, byte_val in enumerate(message.data): char_position = base_position + i self._name_buffer[char_position] = chr(byte_val) # Check if all expected bytes have been received total_bytes = ranges[-1][2] + (ranges[-1][1] - ranges[-1][0] + 1) if len(self._name_buffer) >= total_bytes: # All bytes received: assemble the name string, skip 0xFF (unprogrammed bytes) self._name = "".join( str(x) for x in self._name_buffer.values() if x != chr(0xFF) ) self._name_buffer = {} # release buffer await self._cache() break async def _process_memory_data_message(self, message: MemoryDataMessage) -> None: addr_int = (message.high_address << 8) + message.low_address addr = f"{message.high_address:02X}{message.low_address:02X}" if "Memory" not in self._data: return if "Address" not in self._data["Memory"]: return if addr not in self._data["Memory"]["Address"]: return mdata = self._data["Memory"]["Address"][addr] if "Match" in mdata: for chan, chan_data in handle_match(mdata["Match"], message.data).items(): data = chan_data.copy() # Special handling for 16-bit PulsePerUnits if "PulsePerUnits" in data: current_pulses = getattr(self._channels[chan], "_pulses", 0) or 0 # If this is the high byte (even address in VMB8IN) if addr_int % 4 == 0: # 02E8, 02EC... new_pulses = (message.data << 8) + (current_pulses & 0xFF) else: # 02E9, 02ED... new_pulses = (current_pulses & 0xFF00) + message.data data["pulses"] = new_pulses await self._update_channel(chan, data) async def _process_channel_name_message(self, part: int, message: Message) -> None: channel = self._translate_channel_name(message.channel) if channel not in self._channels: return if self._channels[channel].set_name_part(part, message.name) and self.loaded: await self._channels[channel].status_update() def _translate_channel_name(self, channel: str) -> int: if keys_exists( self._data, "ChannelNumbers", "Name", "Map", f"{int(channel):02X}", ): return int( self._data["ChannelNumbers"]["Name"]["Map"][f"{int(channel):02X}"] ) return int(channel) async def is_loaded(self) -> bool: """Check if all name messages have been received.""" # if we are loaded, just return if self.loaded: return True if self._is_loading: return False # Name loading is still in progress if the buffer is non-empty. # _name is an empty dict (initial state), None (no name spec), or a str (fully loaded). if self._name_buffer: return False # all channel names should be loaded for chan in self._channels.values(): if not chan.is_loaded(): return False # set that we finished the module loading self.loaded = True await self._cache() return True async def _request_module_status(self) -> None: """Request current state of channels.""" if "Channels" not in self._data: # some modules have no channels return self._log.info(f"Request module status {self._address}") mod_stat_req_msg = ModuleStatusRequestMessage(self._address) counter_msg = None if keys_exists(self._data, "AllChannelStatus"): mod_stat_req_msg.channels = self._data["AllChannelStatus"] else: for chan, chan_data in self._data["Channels"].items(): if int(chan) < 9 and chan_data["Type"] in ("Blind", "Dimmer", "Relay"): mod_stat_req_msg.channels.append(int(chan)) if chan_data["Type"] == "ButtonCounter": if counter_msg is None: counter_msg = CounterStatusRequestMessage(self._address) counter_msg.channels.append(int(chan)) await self._writer(mod_stat_req_msg) if counter_msg is not None: await self._writer(counter_msg) async def _request_channel_name(self) -> None: # request the module channel names if keys_exists(self._data, "AllChannelStatus"): msg = ChannelNameRequestMessage(self._address) msg.priority = PRIORITY_LOW msg.channels = 0xFF await self._writer(msg) else: msg_type = commandRegistry.get_command( CHANNEL_NAME_REQUEST_COMMAND_CODE, self.get_type() ) msg = msg_type(self._address) msg.priority = PRIORITY_LOW msg.channels = list(range(1, (self.number_of_channels() + 1))) await self._writer(msg) async def __load_memory(self) -> None: """Request all needed memory addresses.""" if "Memory" not in self._data: self._name = None return if self._type == 0x0C: self._name = None return for memory_key, memory_part in self._data["Memory"].items(): if memory_key == "Address": for addr_int in memory_part: addr = struct.unpack( ">BB", struct.pack(">h", int("0x" + addr_int, 0)) ) msg = ReadDataFromMemoryMessage(self._address) msg.priority = PRIORITY_LOW msg.high_address = addr[0] msg.low_address = addr[1] await self._writer(msg) elif memory_key in {"ModuleName", "SensorName"}: # example: # "ModuleName": "00DD-00E9;01DD-01E9;02DD-02E9;03DD-03E9;04DD-04E8", # request using MermoryDataBlock message, requests 4 bytes for block in memory_part.split(";"): addr_start_str, addr_end_str = block.split("-") # Convert to integer addresses addr_start_int = int("0x" + addr_start_str, 0) addr_end_int = int("0x" + addr_end_str, 0) # Split into 4-byte blocks current_addr = addr_start_int while current_addr <= addr_end_int: block_end = min( current_addr + 3, addr_end_int ) # 4 bytes = current + 3 # Convert back to high/low address bytes addr_start = struct.unpack( ">BB", struct.pack(">h", current_addr) ) msg = ReadDataBlockFromMemoryMessage(self._address) msg.priority = PRIORITY_LOW msg.high_address = addr_start[0] msg.low_address = addr_start[1] await self._writer(msg) current_addr = block_end + 1 async def _load_properties(self) -> None: """Method for per module type loading of properties.""" if "Properties" not in self._data: return for prop, prop_data in self._data["Properties"].items(): if "Type" not in prop_data: continue prop_type = prop_data["Type"] try: cls = getattr(properties_module, prop_type) except AttributeError: self._log.error( "Unknown property type '%s' for property '%s' on module address %s", prop_type, prop, getattr(self, "_address", "unknown"), ) continue self._properties[prop] = cls( module=self, name=prop_data.get("Name", prop), writer=self._writer, ) async def _load_default_channels(self) -> None: if "Channels" not in self._data: return for chan, chan_data in self._data["Channels"].items(): edit = True sub = True if "Editable" not in chan_data or chan_data["Editable"] != "yes": edit = False if "Subdevice" not in chan_data or chan_data["Subdevice"] != "yes": sub = False chan_type = chan_data["Type"] try: cls = getattr(channels_module, chan_type) except AttributeError: self._log.error( "Unknown channel type '%s' for channel '%s' on module address %s", chan_type, chan, getattr(self, "_address", "unknown"), ) continue chan_num = self._translate_channel_name(chan) self._channels[chan_num] = cls( module=self, num=chan_num, name=chan_data["Name"], nameEditable=edit, subDevice=sub, writer=self._writer, address=self._address, ) if chan_data["Type"] == "Temperature": if "Thermostat" in self._data or ( "ThermostatAddr" in self._data and self._data["ThermostatAddr"] != 0 ): await self._update_channel(chan_num, {"thermostat": True}) if chan_data["Type"] == "Dimmer" and "sliderScale" in self._data: self._channels[chan_num].slider_scale = self._data["sliderScale"] class VmbDali(Module): """DALI has a variable number of channels. Therefore we create a module that first creates 64 placeholder channels. After that it requests the DALI device settings to determine the actual channels. """ def __init__( self, module_address: int, module_type: int, serial: int | None = None, memorymap: int | None = None, build_year: int | None = None, build_week: int | None = None, cache_dir: str | None = None, on_load_finished: Callable[[Module], Awaitable[None]] | None = None, ) -> None: """Initialize DALI module.""" super().__init__( module_address, module_type, serial, memorymap, build_year, build_week, cache_dir, on_load_finished, ) self.group_members: dict[int, set[int]] = {} def get_initial_timeout(self) -> int: """Get initial timeout for loading this module.""" return 100000 async def _load_default_channels(self) -> None: for chan in range(1, 64 + 1): self._channels[chan] = Channel( module=self, num=chan, name="placeholder", nameEditable=True, subDevice=True, writer=self._writer, address=self._address, ) # Placeholders will keep this module loading # Until the DaliDeviceSettings messages either delete or replace these placeholder's # with actual channels await self._request_dali_channels() async def _request_dali_channels(self): msg_type = commandRegistry.get_command( DALI_DEVICE_SETTINGS_REQUEST_COMMAND_CODE, self.get_type() ) msg: DaliDeviceSettingsRequest = msg_type(self._address) msg.priority = PRIORITY_LOW msg.channel = 81 # all msg.settings = None # all await self._writer(msg) async def on_message(self, message: Message) -> None: """Process received message.""" if isinstance(message, DaliDeviceSettingMsg): if isinstance(message.data, DaliDeviceTypeMsg): if message.data.device_type == DaliDeviceType.NoDevicePresent: if message.channel in self._channels: del self._channels[message.channel] elif message.data.device_type == DaliDeviceType.LedModule: cache = self._loaded_cache if ( "channels" in cache and str(message.channel) in cache["channels"] and cache["channels"][str(message.channel)]["type"] == "Dimmer" ): # If we have a cached dimmer channel, use that name name = cache["channels"][str(message.channel)]["name"] self._channels[message.channel] = Dimmer( self, message.channel, name, False, # set False to enable an already loaded Dimmer True, self._writer, self._address, slider_scale=254, ) elif self._channels.get(message.channel).__class__ != Dimmer: # New or changed type, replace channel: self._channels[message.channel] = Dimmer( self, message.channel, None, True, True, self._writer, self._address, slider_scale=254, ) await self._request_single_channel_name(message.channel) elif isinstance(message.data, MemberOfGroupMsg): for group in range(15 + 1): this_group_members = self.group_members.setdefault(group, set()) if message.data.member_of_group[group]: this_group_members.add(message.channel) elif message.channel in this_group_members: this_group_members.remove(message.channel) elif isinstance(message, PushButtonStatusMessage): _channel_offset = self.calc_channel_offset(message.address) for channel in message.opened: if _channel_offset + channel > 64: # ignore groups continue await self._update_channel((_channel_offset + channel), {"state": 0}) # ignore message.closed: we don't know at what dimlevel they're started elif isinstance(message, DimValueStatus): for offset, dim_value in enumerate(message.dim_values): channel = message.channel + offset if channel <= 64: # channel await self._update_channel(channel, {"state": dim_value}) elif channel <= 80: # group group_num = channel - 65 for chan in self.group_members.get(group_num, []): await self._update_channel(chan, {"state": dim_value}) else: # broadcast for chan in self._channels.values(): await chan.update({"state": dim_value}) elif isinstance( message, ( SetLedMessage, ClearLedMessage, FastBlinkingLedMessage, SlowBlinkingLedMessage, ), ): pass else: return await super().on_message(message) return None async def _request_channel_name(self) -> None: # Channel names are requested after channel scan # don't do them here (at initialization time) pass async def _request_single_channel_name(self, channel_num: int) -> None: msg_type = commandRegistry.get_command( CHANNEL_NAME_REQUEST_COMMAND_CODE, self.get_type() ) msg = msg_type(self._address) msg.priority = PRIORITY_LOW msg.channels = channel_num await self._writer(msg)