diff --git a/OWNd/message/__init__.py b/OWNd/message/__init__.py index ae949806..9c235c98 100644 --- a/OWNd/message/__init__.py +++ b/OWNd/message/__init__.py @@ -171,6 +171,9 @@ MESSAGE_TYPE_MOTION_TIMEOUT, MESSAGE_TYPE_PIR_SENSITIVITY, PIR_SENSITIVITY_MAPPING, + area_of_where, + is_apl_address, + normalize_where, OWNLightingCommand, OWNLightingEvent, ) @@ -220,6 +223,9 @@ "MESSAGE_TYPE_ILLUMINANCE", "MESSAGE_TYPE_MOTION_TIMEOUT", "PIR_SENSITIVITY_MAPPING", + "is_apl_address", + "area_of_where", + "normalize_where", "OWNLightingEvent", "OWNLightingCommand", # Automation (WHO 2) diff --git a/OWNd/message/base.py b/OWNd/message/base.py index eba9c0ae..821f3d22 100644 --- a/OWNd/message/base.py +++ b/OWNd/message/base.py @@ -257,10 +257,35 @@ def interface(self) -> str | None: else None ) + @property + def what(self) -> int | None: + """The 'what' ID of this message""" + return getattr(self, "_what", None) + + @property + def what_param(self) -> list[str]: + """The 'what' parameters of this message""" + return getattr(self, "_what_param", []) + + @property + def where_param(self) -> list[str]: + """The 'where' parameters of this message""" + return getattr(self, "_where_param", []) + @property def dimension(self) -> int | None: """The 'dimension' ID of this message""" - return self._dimension + return getattr(self, "_dimension", None) + + @property + def dimension_param(self) -> list[str]: + """The 'dimension' parameters of this message""" + return getattr(self, "_dimension_param", []) + + @property + def dimension_value(self) -> list[str]: + """The 'dimension' values of this message""" + return getattr(self, "_dimension_value", []) @property def entity(self) -> str: diff --git a/OWNd/message/lighting.py b/OWNd/message/lighting.py index af449904..344e29a9 100644 --- a/OWNd/message/lighting.py +++ b/OWNd/message/lighting.py @@ -21,6 +21,74 @@ PIR_SENSITIVITY_MAPPING = ["low", "medium", "high", "very high"] +def is_apl_address(base: str) -> bool: + """Check if base address is a valid OpenWebNet Point-to-Point (APL) address. + + Point-to-point addressing combinations: + - A = 0; PL [1-9] -> 2 digits (e.g. 02 = Area 0, PL 2; accepted by BTicino firmware) + - A = 00; PL [01-15] -> 4 digits (e.g. 0015 = Area 00, PL 15) + - A [1-9]; PL [1-9] -> 2 digits (e.g. 15 = Area 1, PL 5) + - A = 10; PL [01-15] -> 4 digits (e.g. 1015 = Area 10, PL 15) + - A [01-09]; PL [10-15] -> 4 digits (e.g. 0115 = Area 1, PL 15) + """ + if not base.isdigit(): + return False + if len(base) == 2: + a = int(base[0]) + pl = int(base[1]) + if a == 0: + return 1 <= pl <= 9 + return 1 <= a <= 9 and 1 <= pl <= 9 + if len(base) == 4: + a = int(base[:2]) + pl = int(base[2:]) + if a == 0: + return 1 <= pl <= 15 + if 1 <= a <= 9: + return 10 <= pl <= 15 + if a == 10: + return 1 <= pl <= 15 + return False + + +def area_of_where(where: str | int | None) -> str | None: + """Return the area status WHERE for a given Point-to-Point (APL) WHERE.""" + if where is None: + return None + where_str = str(where).strip() + parts = where_str.split("#", 1) + base = parts[0] + if not is_apl_address(base): + return None + if len(base) == 2: + if base[0] == "0": + return "00" + return base[0] + a = base[:2] + if a == "00": + return "00" + if a == "10": + return "100" + return str(int(a)) + + +def normalize_where(where: str | int | None) -> str: + """Normalize OpenWebNet address while preserving Point-to-Point (APL) addressing.""" + if where is None: + return "" + where_str = str(where).strip() + if not where_str: + return "" + parts = where_str.split("#", 1) + base = parts[0] + if is_apl_address(base) or base in ("00", "100", "0"): + norm_base = base + elif base.isdigit() and base.startswith("00") and int(base) > 15: + norm_base = str(int(base)) + else: + norm_base = base + return f"{norm_base}#{parts[1]}" if len(parts) > 1 else norm_base + class OWNLightingEvent(OWNEvent): def __init__(self, data: str) -> None: @@ -272,6 +340,13 @@ def motion_timeout(self) -> datetime.timedelta | None: def color_temp(self) -> int | None: return self._color_temp + @property + def color_temp_kelvin(self) -> int | None: + """Color temperature in Kelvin when a valid mired reading (> 1) is present.""" + if self._color_temp is not None and self._color_temp > 1: + return int(1000000 / self._color_temp) + return None + @property def supports_color_temp(self) -> bool | None: return self._supports_color_temp @@ -292,6 +367,10 @@ def value(self) -> int | None: def hs(self) -> tuple[int, int] | None: return self._hs + @property + def hs_color(self) -> tuple[int, int] | None: + return self._hs + @property def hsv(self) -> tuple[int, int, int] | None: return self._hsv @@ -311,6 +390,55 @@ def supports_rgb(self) -> bool | None: class OWNLightingCommand(OWNCommand): + @property + def brightness(self) -> int | None: + """Decoded brightness percentage (0..100) for Dimension 1/4 write commands.""" + if self._dimension in (1, 4) and self._dimension_value: + try: + raw = int(self._dimension_value[0]) + except (TypeError, ValueError): + return None + if 100 <= raw <= 200: + return raw - 100 + return None + + @property + def color_temp(self) -> int | None: + """Decoded color temperature in mireds for Dimension 14 write commands.""" + if self._dimension == 14 and self._dimension_value: + try: + val = int(self._dimension_value[0]) + except (TypeError, ValueError): + return None + if val > 1: + return val + return None + + @property + def color_temp_kelvin(self) -> int | None: + """Decoded color temperature in Kelvin for Dimension 14 write commands.""" + ct = self.color_temp + return int(1000000 / ct) if ct is not None and ct > 1 else None + + @property + def hs(self) -> tuple[int, int] | None: + """Decoded (hue, saturation) tuple for Dimension 12 write commands.""" + if self._dimension == 12 and len(self._dimension_value) >= 3: + try: + h = int(self._dimension_value[0]) + s = int(self._dimension_value[1]) + v = int(self._dimension_value[2]) + except (TypeError, ValueError): + return None + if h <= 360 and (h, s, v) != (511, 127, 255): + return (h, s) + return None + + @property + def hs_color(self) -> tuple[int, int] | None: + """Alias for :attr:`hs`.""" + return self.hs + @classmethod def status(cls, where: str | int) -> OWNLightingCommand: message = cls(f"*#1*{where}##") diff --git a/OWNd/message/sound.py b/OWNd/message/sound.py index 6a280df1..5f7828d7 100644 --- a/OWNd/message/sound.py +++ b/OWNd/message/sound.py @@ -29,6 +29,7 @@ def __init__(self, data: str) -> None: self._environment = self._zone[1] if self._is_routing_event else None self._routed_source = self._zone[2] if self._is_routing_event else None self._volume: int | None = None + self._volume_step_delta: int | None = None self._frequency_khz: int | None = None self._track: int | None = None self._rds_text: str | None = None @@ -54,6 +55,16 @@ def __init__(self, data: str) -> None: elif self._state == 100: self._is_source_busy = True self._human_readable_log = f"{subject} is BUSY." + elif self._state is not None and 1001 <= self._state <= 1015: + self._volume_step_delta = self._state - 1000 + self._human_readable_log = ( + f"{subject} increase volume by {self._volume_step_delta} step(s)." + ) + elif self._state is not None and 1101 <= self._state <= 1115: + self._volume_step_delta = -(self._state - 1100) + self._human_readable_log = ( + f"{subject} decrease volume by {abs(self._volume_step_delta)} step(s)." + ) elif self._state == 5000: self._is_seek_up = True self._human_readable_log = f"{subject} seek forward." @@ -162,6 +173,11 @@ def zone(self) -> str: def volume(self) -> int | None: return self._volume + @property + def volume_step_delta(self) -> int | None: + """Signed volume step delta (+1..+15 for WHAT 1001..1015, -1..-15 for WHAT 1101..1115).""" + return self._volume_step_delta + @property def frequency_khz(self) -> int | None: return self._frequency_khz diff --git a/tests/test_sound.py b/tests/test_sound.py index 44c7239b..f1a49f6e 100644 --- a/tests/test_sound.py +++ b/tests/test_sound.py @@ -275,3 +275,24 @@ def test_who16_sound_source_events() -> None: blank_rds = OWNSoundEvent("*#16*101*8*32*32*32*32*32*32*32*32##") assert blank_rds.rds_text is None assert "RDS text: ''" in blank_rds.human_readable_log + + +def test_who16_volume_step_delta_events() -> None: + """Verify WHO 16 stepped volume WHAT 1001..1015 and 1101..1115 decode volume_step_delta.""" + vol_up_1 = OWNMessage.parse("*16*1001*21##") + assert isinstance(vol_up_1, OWNSoundEvent) + assert vol_up_1.volume_step_delta == 1 + assert "increase volume by 1 step(s)" in vol_up_1.human_readable_log + + vol_up_15 = OWNMessage.parse("*16*1015*21##") + assert isinstance(vol_up_15, OWNSoundEvent) + assert vol_up_15.volume_step_delta == 15 + + vol_down_1 = OWNMessage.parse("*16*1101*21##") + assert isinstance(vol_down_1, OWNSoundEvent) + assert vol_down_1.volume_step_delta == -1 + assert "decrease volume by 1 step(s)" in vol_down_1.human_readable_log + + vol_down_15 = OWNMessage.parse("*16*1115*21##") + assert isinstance(vol_down_15, OWNSoundEvent) + assert vol_down_15.volume_step_delta == -15 diff --git a/tests/test_who1_ts10_builders.py b/tests/test_who1_ts10_builders.py index 688bc83f..76e606e4 100644 --- a/tests/test_who1_ts10_builders.py +++ b/tests/test_who1_ts10_builders.py @@ -45,3 +45,88 @@ def test_variable_timer(): assert str(OWNLightingCommand.get_variable_timer(11)) == "*#1*11*2##" with pytest.raises(ValueError): OWNLightingCommand.set_variable_timer(11, 0, 60, 0) + + +def test_apl_addressing_helpers(): + from OWNd.message import area_of_where, is_apl_address, normalize_where + + assert is_apl_address("02") is True + assert is_apl_address("15") is True + assert is_apl_address("0015") is True + assert is_apl_address("0115") is True + assert is_apl_address("1015") is True + assert is_apl_address("00") is False + assert is_apl_address("100") is False + assert is_apl_address("#6") is False + assert is_apl_address("1101") is False + + assert area_of_where("02") == "00" + assert area_of_where("15") == "1" + assert area_of_where("0015") == "00" + assert area_of_where("0115") == "1" + assert area_of_where("1015") == "100" + assert area_of_where("#6") is None + assert area_of_where(None) is None + + assert normalize_where("02") == "02" + assert normalize_where("0015") == "0015" + assert normalize_where("0021") == "21" + assert normalize_where("0109") == "0109" + assert normalize_where("15#4#01") == "15#4#01" + assert normalize_where("#5") == "#5" + assert normalize_where(None) == "" + assert normalize_where(" ") == "" + + +def test_lighting_dimension_properties(): + from OWNd.message import OWNLightingEvent, OWNMessage + + cmd_b = OWNMessage.parse("*#1*#6*#1*150*0##") + assert isinstance(cmd_b, OWNLightingCommand) + assert cmd_b.brightness == 50 + assert cmd_b.dimension_value == ["150", "0"] + assert cmd_b.where_param == [] + assert cmd_b.dimension_param == [] + assert cmd_b.what is None + assert cmd_b.what_param == [] + assert cmd_b.color_temp is None + assert cmd_b.color_temp_kelvin is None + assert cmd_b.hs is None + + cmd_b_bad = OWNMessage.parse("*#1*#6*#1*50*0##") + assert isinstance(cmd_b_bad, OWNLightingCommand) + assert cmd_b_bad.brightness is None + + cmd_b._dimension_value = ["abc"] + assert cmd_b.brightness is None + + cmd_ct = OWNMessage.parse("*#1*#6*#14*153##") + assert isinstance(cmd_ct, OWNLightingCommand) + assert cmd_ct.color_temp == 153 + assert cmd_ct.color_temp_kelvin == int(1000000 / 153) + assert cmd_ct.brightness is None + + cmd_ct._dimension_value = ["1"] + assert cmd_ct.color_temp is None + cmd_ct._dimension_value = ["abc"] + assert cmd_ct.color_temp is None + + cmd_hs = OWNMessage.parse("*#1*#6*#12*120*50*80##") + assert isinstance(cmd_hs, OWNLightingCommand) + assert cmd_hs.hs == (120, 50) + assert cmd_hs.hs_color == (120, 50) + + cmd_hs._dimension_value = ["511", "127", "255"] + assert cmd_hs.hs is None + cmd_hs._dimension_value = ["abc", "50", "80"] + assert cmd_hs.hs is None + + ev_ct = OWNMessage.parse("*#1*#6*14*153##") + assert isinstance(ev_ct, OWNLightingEvent) + assert ev_ct.color_temp_kelvin == int(1000000 / 153) + + ev_hs = OWNMessage.parse("*#1*#6*12*120*50*80##") + assert isinstance(ev_hs, OWNLightingEvent) + assert ev_hs.hs_color == (120, 50) + assert ev_hs.color_temp_kelvin is None +