diff --git a/OWNd/message/__init__.py b/OWNd/message/__init__.py index 5d4aa9c..a20b60a 100644 --- a/OWNd/message/__init__.py +++ b/OWNd/message/__init__.py @@ -33,6 +33,8 @@ OWNCENPlusEvent, OWNCenCommand, OWNCenPlusCommand, + OWNScenarioPlusCommand, + OWNScenarioPlusEvent, OWNDryContactCommand, OWNDryContactEvent, ) @@ -40,6 +42,7 @@ MESSAGE_TYPE_ACTIVE_POWER, MESSAGE_TYPE_CURRENT_DAY_CONSUMPTION, MESSAGE_TYPE_CURRENT_MONTH_CONSUMPTION, + MESSAGE_TYPE_AUTO_UPDATE_INTERVAL, MESSAGE_TYPE_DAILY_CONSUMPTION, MESSAGE_TYPE_ENERGY_TOTALIZER, MESSAGE_TYPE_HOURLY_CONSUMPTION, diff --git a/OWNd/message/base.py b/OWNd/message/base.py index dbea1c0..f174e64 100644 --- a/OWNd/message/base.py +++ b/OWNd/message/base.py @@ -222,6 +222,11 @@ def who(self) -> int | None: """The 'who' ID of the subject of this message""" return self._who + @property + def what(self) -> int | None: + """The raw WHAT of a command or status frame, None for dimension frames.""" + return self._what + @property def where(self) -> str | None: """The 'where' ID of the subject of this message""" diff --git a/OWNd/message/cen.py b/OWNd/message/cen.py index 2dfeb71..ee31b8f 100644 --- a/OWNd/message/cen.py +++ b/OWNd/message/cen.py @@ -155,6 +155,66 @@ def human_readable_log(self) -> str: +class OWNScenarioPlusEvent(OWNEvent): + """WHO 25 WHAT 11-15: scenario-plus / dimmer-like commands on a WHERE. + + libqtdevices ScenarioPlusDevice sends 11#0 on, 12 off, 13#0#5 increase, + 14#0#5 decrease, 15 stop (scenario_device.cpp:37-41, 163-186); mhs1 bt_luci + accepts all five and echoes them on the monitor port. + """ + + _NAMES = {11: "on", 12: "off", 13: "increase", 14: "decrease", 15: "stop"} + + def __init__(self, data: str) -> None: + super().__init__(data) + self._action = self._NAMES.get(self._what) if self._what is not None else None + self.object = self._where + _params = "#".join(self._what_param) if self._what_param else "" + self._human_readable_log = ( + f"Scenario plus {self._where}: {self._action}" + + (f" (parameters {_params})" if _params else "") + + "." + ) + + @property + def action(self) -> str | None: + return self._action + + +class OWNScenarioPlusCommand(OWNCommand): + """Builders for the five WHO 25 frames libqtdevices ScenarioPlusDevice sends.""" + + @classmethod + def turn_on(cls, where: str | int) -> OWNScenarioPlusCommand: + message = cls(f"*25*11#0*{where}##") + message._human_readable_log = f"Scenario plus {where}: on." + return message + + @classmethod + def turn_off(cls, where: str | int) -> OWNScenarioPlusCommand: + message = cls(f"*25*12*{where}##") + message._human_readable_log = f"Scenario plus {where}: off." + return message + + @classmethod + def increase(cls, where: str | int) -> OWNScenarioPlusCommand: + message = cls(f"*25*13#0#5*{where}##") + message._human_readable_log = f"Scenario plus {where}: increase." + return message + + @classmethod + def decrease(cls, where: str | int) -> OWNScenarioPlusCommand: + message = cls(f"*25*14#0#5*{where}##") + message._human_readable_log = f"Scenario plus {where}: decrease." + return message + + @classmethod + def stop(cls, where: str | int) -> OWNScenarioPlusCommand: + message = cls(f"*25*15*{where}##") + message._human_readable_log = f"Scenario plus {where}: stop." + return message + + class OWNDryContactCommand(OWNCommand): @classmethod def status(cls, where: str | int) -> OWNDryContactCommand: @@ -280,6 +340,8 @@ def _parse_who25_event(data: str) -> OWNEvent: what_code = None if what_code is not None and 21 <= what_code <= 28: return OWNCENPlusEvent(data) + if what_code is not None and 11 <= what_code <= 15: + return OWNScenarioPlusEvent(data) if what_code is not None and what_code not in (31, 32): return OWNEvent(data) return OWNDryContactEvent(data) @@ -296,6 +358,8 @@ def _parse_who25_command(data: str) -> OWNCommand: what_code = None if what_code is not None and 21 <= what_code <= 28: return OWNCenPlusCommand(data) + if what_code is not None and 11 <= what_code <= 15: + return OWNScenarioPlusCommand(data) # Same rule as _parse_who25_event: only WHAT 31/32 are dry contacts. if what_code is not None and what_code not in (31, 32): return OWNCommand(data) diff --git a/OWNd/message/energy.py b/OWNd/message/energy.py index 5be4132..08a4400 100644 --- a/OWNd/message/energy.py +++ b/OWNd/message/energy.py @@ -17,6 +17,7 @@ MESSAGE_TYPE_MONTHLY_CONSUMPTION = "monthly_consumption" MESSAGE_TYPE_CURRENT_DAY_CONSUMPTION = "current_day_partial_consumption" MESSAGE_TYPE_CURRENT_MONTH_CONSUMPTION = "current_month_partial_consumption" +MESSAGE_TYPE_AUTO_UPDATE_INTERVAL = "auto_update_interval" def _integer_value( @@ -45,6 +46,8 @@ def __init__(self, data: str) -> None: self._current_day_partial_consumption = 0 self._monthly_consumption: dict[str, Any] = {} self._current_month_partial_consumption = 0 + self._update_interval: int | None = None + self._energy_type: int | None = None if not where.startswith(("1", "5", "7")): return @@ -158,6 +161,16 @@ def __init__(self, data: str) -> None: self._dimension_value, 0 ) self._human_readable_log = f"Sensor {self._sensor} is reporting a power consumption of {self._monthly_consumption['value']} Wh for {self._monthly_consumption['date'].strftime('%B %Y')}." # pylint: disable=line-too-long + elif self._dimension == 1200 and self._dimension_value: + # *#18*W*1200#type*time##: the meter confirms (time > 0) or + # ends (time = 0) the automatic power updates; libqtdevices + # re-arms the stream on time = 0 (energy_device.cpp:289-322, + # 719-726). type: 1 electricity, 2 gas, 3 heat, 4 water + # (energy_device.cpp:50-56). + self._type = MESSAGE_TYPE_AUTO_UPDATE_INTERVAL + self._update_interval = _integer_value(self._dimension_value, 0) + self._energy_type = _integer_value(self._dimension_param, 0, 0) if self._dimension_param else None + self._human_readable_log = f"Sensor {self._sensor} automatic updates every {self._update_interval} s (0 = stopped)." # pylint: disable=line-too-long elif self._dimension == 53: self._type = MESSAGE_TYPE_CURRENT_MONTH_CONSUMPTION self._current_month_partial_consumption = _integer_value( @@ -173,6 +186,16 @@ def message_type(self) -> str | None: def sensor(self) -> str: return self._sensor + @property + def update_interval(self) -> int | None: + """Seconds between automatic power updates from a 1200 reply, 0 when stopped.""" + return self._update_interval + + @property + def energy_type(self) -> int | None: + """Energy type of a 1200 reply (1 electricity, 2 gas, 3 heat, 4 water).""" + return self._energy_type + @property def active_power(self) -> int: return self._active_power diff --git a/OWNd/message/heating.py b/OWNd/message/heating.py index a9fda25..2170678 100644 --- a/OWNd/message/heating.py +++ b/OWNd/message/heating.py @@ -183,6 +183,9 @@ def __init__(self, data: str) -> None: self._measured_temperature = None self._secondary_temperature = None self._measured_humidity = None + self._program: int | None = None + self._scenario: int | None = None + self._holiday_days: int | None = None self._holiday_end_date: tuple[int, int, int] | None = None self._holiday_end_time: tuple[int, int] | None = None self._manual_timed_duration: tuple[int, int] | None = None @@ -211,6 +214,7 @@ def __init__(self, data: str) -> None: self._mode in [0, 210, 211, 212, 215] or (self._mode >= 2101 and self._mode <= 2103) or (self._mode >= 2201 and self._mode <= 2216) + or (self._mode >= 23001 and self._mode <= 23255) ): self._type = MESSAGE_TYPE_MODE self._mode_name = CLIMATE_MODE_COOL @@ -221,6 +225,7 @@ def __init__(self, data: str) -> None: self._mode in [1, 110, 111, 112, 115] or (self._mode >= 1101 and self._mode <= 1103) or (self._mode >= 1201 and self._mode <= 1216) + or (self._mode >= 13001 and self._mode <= 13255) ): self._type = MESSAGE_TYPE_MODE self._mode_name = CLIMATE_MODE_HEAT @@ -232,8 +237,6 @@ def __init__(self, data: str) -> None: or (self._mode >= 3101 and self._mode <= 3116) or (self._mode >= 3201 and self._mode <= 3216) or (self._mode >= 33001 and self._mode <= 33255) - or (self._mode >= 23001 and self._mode <= 23255) - or (self._mode >= 13001 and self._mode <= 13255) ): self._type = MESSAGE_TYPE_MODE self._mode_name = CLIMATE_MODE_AUTO @@ -250,15 +253,60 @@ def __init__(self, data: str) -> None: self._human_readable_log = ( f"Zone {self._zone}'s remote control is enabled" ) + elif self._mode in (22, 23, 24, 30, 31): + # Central unit system status (Legrand WHO 4 p. 5, p. 64): + # 22 at least one probe OFF, 23 at least one probe in + # protection, 24 at least one probe in manual, 30 failure + # discovered, 31 central unit battery KO. + self._mode_name = None + _status_text = { + 22: "at least one probe is OFF", + 23: "at least one probe is in protection", + 24: "at least one probe is in manual mode", + 30: "a failure was discovered", + 31: "the central unit battery is KO", + }[self._mode] + self._human_readable_log = ( + f"Zone {self._zone}'s central unit reports {_status_text}" + ) else: self._mode_name = None self._human_readable_log = f"Zone {self._zone}'s mode is unknown" + # Program / scenario / holiday-days numbers carried in the WHAT + # (Legrand WHO 4 p. 5 and p. 64; libqtdevices thermal_device.cpp:209-211, + # 246-262, 291-307): 11xx/21xx/31xx program, 12xx/22xx/32xx scenario, + # 13xxx/23xxx/33xxx holiday days. + if self._type == MESSAGE_TYPE_MODE: + if 1101 <= self._mode <= 1199 or 2101 <= self._mode <= 2199 or 3101 <= self._mode <= 3199: + self._program = self._mode % 100 + elif 1201 <= self._mode <= 1299 or 2201 <= self._mode <= 2299 or 3201 <= self._mode <= 3299: + self._scenario = self._mode % 100 + elif 13001 <= self._mode <= 13999 or 23001 <= self._mode <= 23999 or 33001 <= self._mode <= 33999: + self._holiday_days = self._mode % 1000 + if ( + self._mode in (115, 215, 315) + and self._what_param + and self._what_param[0] + ): + # Holiday daily plan: the parameter is the weekly program the + # central unit resumes afterwards, 1101-1103 / 2101-2103 + # (Legrand WHO 4 p. 56 and p. 64, "115#parameterH"); libqtdevices + # reads it as whatArgN(0) % 100 (thermal_device.cpp:241, 286). + # It is not a temperature. + try: + self._program = int(self._what_param[0]) % 100 + self._human_readable_log += f" (program {self._program})." + except ValueError: + self._human_readable_log += "." + elif ( self._type == MESSAGE_TYPE_MODE and self._what_param and self._what_param[0] is not None ): + # 110#T / 210#T manual with temperature (Legrand WHO 4 p. 23, + # p. 56), 312#T timed manual (libqtdevices thermal_device.cpp:378). self._type = MESSAGE_TYPE_MODE_TARGET self._set_temperature = who4_temperature(self._what_param[0]) if self._set_temperature is not None: @@ -657,6 +705,21 @@ def cooling_fan_speed(self) -> int | None: def cooling_fan_on(self) -> bool | None: return self._cooling_fan_on + @property + def program(self) -> int | None: + """Weekly program number (1..16) named by the WHAT, else None.""" + return self._program + + @property + def scenario(self) -> int | None: + """Scenario number (1..16) named by the WHAT, else None.""" + return self._scenario + + @property + def holiday_days(self) -> int | None: + """Holiday days (1..255) named by a 13xxx/23xxx/33xxx WHAT, else None.""" + return self._holiday_days + @property def holiday_end_date(self) -> tuple[int, int, int] | None: """Holiday / weekend end date (day, month, year), or None.""" diff --git a/OWNd/message/lighting.py b/OWNd/message/lighting.py index 0fe5472..edeb5b6 100644 --- a/OWNd/message/lighting.py +++ b/OWNd/message/lighting.py @@ -408,6 +408,83 @@ def switch_off(cls, where: str | int, _transition: int | None = None) -> OWNLigh message._human_readable_log = f"Switching OFF light or switch {message._where}{message._interface_log_text}." return message + @classmethod + def set_brightness_preset(cls, where: str | int, preset: int) -> OWNLightingCommand: + """Dimmer preset WHAT 2..10 (20 %..100 %), the frame libqtdevices + DimmerDevice::setLevel100 sends (lighting_device.cpp:240-245).""" + preset = int(preset) + if not 2 <= preset <= 10: + raise ValueError("preset must be between 2 and 10") + message = cls(f"*1*{preset}*{where}##") + message._human_readable_log = f"Setting light {message._where}{message._interface_log_text} to preset level {preset}." # pylint: disable=line-too-long + return message + + @classmethod + def step_up( + cls, where: str | int, delta: int | None = None, speed: int | None = None + ) -> OWNLightingCommand: + """One level up (WHAT 30), or ``30#delta#speed`` for a dimmer 100 + (libqtdevices lighting_device.cpp:230-233, 409-412).""" + return cls._step(where, 30, "up", delta, speed) + + @classmethod + def step_down( + cls, where: str | int, delta: int | None = None, speed: int | None = None + ) -> OWNLightingCommand: + """One level down (WHAT 31), or ``31#delta#speed`` for a dimmer 100 + (libqtdevices lighting_device.cpp:235-238, 414-417).""" + return cls._step(where, 31, "down", delta, speed) + + @classmethod + def _step( + cls, where: str | int, what: int, direction: str, delta: int | None, speed: int | None + ) -> OWNLightingCommand: + if delta is None: + message = cls(f"*1*{what}*{where}##") + message._human_readable_log = f"Dimming light {message._where}{message._interface_log_text} one level {direction}." # pylint: disable=line-too-long + return message + delta = int(delta) + speed = int(speed if speed is not None else 0) + if not 1 <= delta <= 100 or not 0 <= speed <= 255: + raise ValueError("delta must be 1..100 and speed 0..255") + message = cls(f"*1*{what}#{delta}#{speed}*{where}##") + message._human_readable_log = f"Dimming light {message._where}{message._interface_log_text} {direction} by {delta}% at speed {speed}." # pylint: disable=line-too-long + return message + + _FIXED_TIMERS = {11: 60, 12: 120, 13: 180, 14: 240, 15: 300, 16: 900, 17: 30, 18: 0.5} + + @classmethod + def switch_on_timed(cls, where: str | int, what: int) -> OWNLightingCommand: + """Fixed timer WHAT 11..18 (1, 2, 3, 4, 5, 15 min, 30 s, 0.5 s), the + frame libqtdevices LightingDevice::fixedTiming sends (lighting_device.cpp:110-117).""" + what = int(what) + if what not in cls._FIXED_TIMERS: + raise ValueError("timer WHAT must be between 11 and 18") + message = cls(f"*1*{what}*{where}##") + message._human_readable_log = f"Switching ON light {message._where}{message._interface_log_text} for {cls._FIXED_TIMERS[what]}s." # pylint: disable=line-too-long + return message + + @classmethod + def set_variable_timer( + cls, where: str | int, hours: int, minutes: int, seconds: int + ) -> OWNLightingCommand: + """``*#1*W*#2*H*M*S##`` (libqtdevices LightingDevice::variableTiming, + lighting_device.cpp:119-124).""" + h, m, s = int(hours), int(minutes), int(seconds) + if not (0 <= h <= 255 and 0 <= m <= 59 and 0 <= s <= 59): + raise ValueError("hours 0..255, minutes and seconds 0..59") + message = cls(f"*#1*{where}*#2*{h}*{m}*{s}##") + message._human_readable_log = f"Setting light {message._where}{message._interface_log_text} timer to {h}h {m}m {s}s." # pylint: disable=line-too-long + return message + + @classmethod + def get_variable_timer(cls, where: str | int) -> OWNLightingCommand: + """``*#1*W*2##`` (libqtdevices LightingDevice::requestVariableTiming, + lighting_device.cpp:131-134).""" + message = cls(f"*#1*{where}*2##") + message._human_readable_log = f"Requesting light {message._where}{message._interface_log_text} timer." + return message + @classmethod def set_brightness( cls, where: str | int, _level: int = 30, _transition: int = 0 diff --git a/OWNd/message/scenario.py b/OWNd/message/scenario.py index 4c1f0aa..9888bad 100644 --- a/OWNd/message/scenario.py +++ b/OWNd/message/scenario.py @@ -13,7 +13,38 @@ def __init__(self, data: str) -> None: self._scenario = self._what self._control_panel = self._where - self._human_readable_log = f"Scenario {self._scenario} from control panel {self._control_panel} has been launched." # pylint: disable=line-too-long + self._programming_scenario: int | None = None + self._event: str | None = None + # WHAT 40-46 are scenario-module programming states, not scenarios + # (libqtdevices scenario_device.cpp:27-35, 93-156; mhs1 bt_luci + # accepts *0*40#N*W##, 41#N, 42, 42#N, 43, 44 and emits 40#N..46#N). + _programming = { + 40: "programming started", + 41: "programming stopped", + 42: "deleted", + 43: "locked", + 44: "unlocked", + 45: "status 45", + 46: "status 46", + } + if self._what in _programming: + self._event = _programming[self._what] + self._scenario = None + try: + self._programming_scenario = ( + int(self._what_param[0]) if self._what_param else None + ) + except (TypeError, ValueError): + self._programming_scenario = None + _target = ( + f"scenario {self._programming_scenario}" + if self._programming_scenario is not None + else "all scenarios" + ) + self._human_readable_log = f"Scenario module {self._control_panel}: {_target} {self._event}." # pylint: disable=line-too-long + else: + self._event = "launched" + self._human_readable_log = f"Scenario {self._scenario} from control panel {self._control_panel} has been launched." # pylint: disable=line-too-long @property def scenario(self) -> int | None: @@ -23,6 +54,16 @@ def scenario(self) -> int | None: def control_panel(self) -> str | None: return self._control_panel + @property + def event(self) -> str | None: + """'launched' for WHAT 1-31, else the programming/lock state (WHAT 40-46).""" + return self._event + + @property + def programming_scenario(self) -> int | None: + """Scenario named by a WHAT 40/41/42 parameter, None when it targets all.""" + return self._programming_scenario + class OWNAuxEvent(OWNEvent): diff --git a/tests/test_issue_77_firmware_findings.py b/tests/test_issue_77_firmware_findings.py index f799176..cd1b47e 100644 --- a/tests/test_issue_77_firmware_findings.py +++ b/tests/test_issue_77_firmware_findings.py @@ -25,6 +25,8 @@ import pytest from OWNd.message import ( + OWNScenarioPlusCommand, + OWNScenarioPlusEvent, CLIMATE_MODE_AUTO, CLIMATE_MODE_COOL, CLIMATE_MODE_HEAT, @@ -396,15 +398,22 @@ def test_fix8_interface_is_read_for_scenarios_and_who14(frame: str) -> None: # ── Fix 9: other WHO 25 messages ──────────────────────────────────────────── -@pytest.mark.parametrize("frame", ["*25*11#121*10##", "*25*12*131##"]) -def test_fix9_other_who25_messages_are_not_dry_contacts(frame: str) -> None: +@pytest.mark.parametrize( + ("frame", "action"), + [("*25*11#121*10##", "on"), ("*25*12*131##", "off"), ("*25*13#0#5*11##", "increase")], +) +def test_fix9_other_who25_messages_are_not_dry_contacts(frame: str, action: str) -> None: """Emitted: the firmware sends WHAT 11..15 on ``1xx`` addresses. - Encyclopedia: dry contacts are WHAT 31 and 32 only. + Encyclopedia: dry contacts are WHAT 31 and 32 only. libqtdevices + ScenarioPlusDevice sends 11#0 / 12 / 13#0#5 / 14#0#5 / 15 + (scenario_device.cpp:37-41) and mhs1 bt_luci accepts and echoes all five + (oracle3/out/luci.tsv), so they are scenario-plus events. """ event = OWNEvent.parse(frame) - assert type(event) is OWNEvent + assert isinstance(event, OWNScenarioPlusEvent) + assert event.action == action assert not isinstance(event, OWNDryContactEvent) @@ -425,8 +434,8 @@ def test_fix9_captured_who25_messages_keep_their_class(frame: str, cls: type) -> @pytest.mark.parametrize( ("frame", "cls"), [ - ("*25*11#121*10##", OWNCommand), - ("*25*12*131##", OWNCommand), + ("*25*11#121*10##", OWNScenarioPlusCommand), + ("*25*12*131##", OWNScenarioPlusCommand), ("*25*31#1*339##", OWNDryContactCommand), ("*25*32#1*33##", OWNDryContactCommand), ("*#25*331##", OWNDryContactCommand), diff --git a/tests/test_message_edges.py b/tests/test_message_edges.py index 4d3ac4d..93f9319 100644 --- a/tests/test_message_edges.py +++ b/tests/test_message_edges.py @@ -196,15 +196,41 @@ def test_mode_auto_variant_315(self): assert isinstance(msg, OWNHeatingEvent) assert msg.mode == CLIMATE_MODE_AUTO - def test_mode_auto_weekly_23001(self): + def test_mode_holiday_23001_is_conditioning(self): + # Legrand WHO 4 p. 5 and p. 64: 23xxx = holiday days in conditioning + # mode; libqtdevices thermal_device.cpp:258-262 (SUM_HOLIDAY, summer). msg = OWNEvent.parse("*4*23001*1##") assert isinstance(msg, OWNHeatingEvent) - assert msg.mode == CLIMATE_MODE_AUTO + assert msg.mode == CLIMATE_MODE_COOL + assert msg.holiday_days == 1 - def test_mode_auto_weekly_13001(self): + def test_mode_holiday_13001_is_heating(self): + # Legrand WHO 4 p. 5 and p. 64: 13xxx = holiday days in heating mode; + # libqtdevices thermal_device.cpp:303-307 (WIN_HOLIDAY, winter). msg = OWNEvent.parse("*4*13001*1##") assert isinstance(msg, OWNHeatingEvent) + assert msg.mode == CLIMATE_MODE_HEAT + assert msg.holiday_days == 1 + + def test_mode_holiday_33004_is_generic(self): + msg = OWNEvent.parse("*4*33004*#0##") assert msg.mode == CLIMATE_MODE_AUTO + assert msg.holiday_days == 4 + + def test_holiday_plan_parameter_is_a_program_not_a_temperature(self): + # Legrand WHO 4 p. 56 / p. 64: 115#parameterH, parameterH = 1101-1103; + # libqtdevices thermal_device.cpp:241 reads whatArgN(0) % 100. + msg = OWNEvent.parse("*4*115#1102*#0##") + assert msg.mode == CLIMATE_MODE_HEAT + assert msg.program == 2 + assert msg.set_temperature is None + assert msg.message_type == "hvac_mode" + + def test_manual_with_temperature_parameter(self): + # Legrand WHO 4 p. 23 / p. 56: 110#T manual heating with temperature. + msg = OWNEvent.parse("*4*110#0215*#0##") + assert msg.set_temperature == 21.5 + assert msg.message_type == "hvac_mode_target" def test_unknown_mode(self): msg = OWNEvent.parse("*4*999*1##")