diff --git a/OWNd/message/__init__.py b/OWNd/message/__init__.py index 48e1ae0..0ce5af4 100644 --- a/OWNd/message/__init__.py +++ b/OWNd/message/__init__.py @@ -36,6 +36,43 @@ OWNDryContactCommand, OWNDryContactEvent, ) +from .door_entry import ( + BROADCAST_WHERE, + END_ALL_CALLS, + KIND_AUTOSWITCH, + KIND_EXTERNAL_INTERCOM, + KIND_FLOOR, + KIND_INTERNAL_INTERCOM, + KIND_PAGER, + KIND_PE1, + KIND_PE2, + KIND_PE3, + KIND_PE4, + MESSAGE_TYPE_AMPLIFIER_MUTE, + MESSAGE_TYPE_CALL, + MESSAGE_TYPE_CAMERA, + MESSAGE_TYPE_LOCK, + MESSAGE_TYPE_PTZ, + MESSAGE_TYPE_SESSION, + MESSAGE_TYPE_STAIRCASE_LIGHT, + MESSAGE_TYPE_TELELOOP, + MESSAGE_TYPE_VCT_INIT, + MMTYPE_AUDIO, + MMTYPE_AUDIO_VIDEO, + MMTYPE_VIDEO, + OWNDoorEntryCommand, + OWNDoorEntryEvent, + OWNIntercomCommand, + OWNIntercomEvent, + OWNLockCommand, + OWNLockEvent, + PTZ_ACTION_PRESS, + PTZ_ACTION_RELEASE, + PTZ_MOVE_DOWN, + PTZ_MOVE_LEFT, + PTZ_MOVE_RIGHT, + PTZ_MOVE_UP, +) from .energy import ( MESSAGE_TYPE_ACTIVE_POWER, MESSAGE_TYPE_CURRENT_DAY_CONSUMPTION, @@ -179,6 +216,42 @@ # Burglar Alarm (WHO 5) "OWNAlarmEvent", "OWNAlarmCommand", + # Door Entry & Lock / Intercom (WHO 6 & 8) + "OWNDoorEntryEvent", + "OWNDoorEntryCommand", + "OWNLockEvent", + "OWNLockCommand", + "OWNIntercomEvent", + "OWNIntercomCommand", + "KIND_PE1", + "KIND_PE2", + "KIND_PE3", + "KIND_PE4", + "KIND_AUTOSWITCH", + "KIND_INTERNAL_INTERCOM", + "KIND_EXTERNAL_INTERCOM", + "KIND_FLOOR", + "KIND_PAGER", + "MMTYPE_AUDIO", + "MMTYPE_VIDEO", + "MMTYPE_AUDIO_VIDEO", + "END_ALL_CALLS", + "BROADCAST_WHERE", + "PTZ_MOVE_UP", + "PTZ_MOVE_DOWN", + "PTZ_MOVE_LEFT", + "PTZ_MOVE_RIGHT", + "PTZ_ACTION_PRESS", + "PTZ_ACTION_RELEASE", + "MESSAGE_TYPE_LOCK", + "MESSAGE_TYPE_STAIRCASE_LIGHT", + "MESSAGE_TYPE_CAMERA", + "MESSAGE_TYPE_PTZ", + "MESSAGE_TYPE_CALL", + "MESSAGE_TYPE_SESSION", + "MESSAGE_TYPE_AMPLIFIER_MUTE", + "MESSAGE_TYPE_TELELOOP", + "MESSAGE_TYPE_VCT_INIT", # Gateway (WHO 13) "_validate_gateway_clock_values", "_gateway_timezone", diff --git a/OWNd/message/base.py b/OWNd/message/base.py index cd2ad53..3f8decf 100644 --- a/OWNd/message/base.py +++ b/OWNd/message/base.py @@ -15,6 +15,7 @@ "alarm", "automation", "cen", + "door_entry", "energy", "gateway", "heating", @@ -66,6 +67,13 @@ class OWNMessage: _ALARM_EMPTY_WHERE = re.compile( r"^\*5\*(?P\d+)(?P(?:#\d+)*)\*##$" ) # *5*WHAT*## + # WHO 6: camera OFF published form has no WHERE field (*6*9##), but field + # units may also emit a single trailing star delimiter (*6*9*##). + # The double-star form (*6*9**##) has WHERE="*" and is matched by _STATUS. + # Only WHAT 9 may omit WHERE; any other short WHO 6 frame is not valid. + _DOOR_ENTRY_SHORT = re.compile( + r"^\*6\*(?P9)\*?##$" + ) # *6*9##, *6*9*## _STATUS_REQUEST = re.compile( r"^\*#(?P\d+)(?:\*(?P#?\d+)(?P(?:#\d+)*))?##$" ) # *#WHO*WHERE## or *#WHO## @@ -128,6 +136,15 @@ def __init__(self, data: str) -> None: # not turn this spelling into a device. self._where = "" + elif match := self._DOOR_ENTRY_SHORT.match(self._raw): + self._is_valid_message = True + self._match = match + self._family = "EVENT" + self._message_type = "STATUS" + self._who = 6 + self._what = int(match.group("what")) + self._where = "" + elif match := self._STATUS_REQUEST.match(self._raw): self._is_valid_message = True self._match = match @@ -186,6 +203,7 @@ def parse(cls, data: str) -> OWNMessage | None: if ( cls._STATUS.match(data) or cls._ALARM_EMPTY_WHERE.match(data) + or cls._DOOR_ENTRY_SHORT.match(data) or cls._DIMENSION_REQUEST_REPLY.match(data) ): return OWNEvent.parse(data) @@ -390,7 +408,7 @@ def parse(cls, data: str) -> OWNCommand | None: return parser(data) if _who in (0, 3, 14, 22, 24) or _who > 1000: return cls(data) - if _who in (6, 7, 9): + if _who in (7, 9): return ( OWNStatusRequest(data) if cls._STATUS_REQUEST.match(data) diff --git a/OWNd/message/door_entry.py b/OWNd/message/door_entry.py new file mode 100644 index 0000000..cda70b8 --- /dev/null +++ b/OWNd/message/door_entry.py @@ -0,0 +1,740 @@ +"""WHO 6 and WHO 8: Door entry, lock, and video intercom events and commands.""" + +from __future__ import annotations + +from .base import OWNCommand, OWNEvent, register_command_parser, register_event_parser + +# Call kinds +KIND_PE1 = 1 +KIND_PE2 = 2 +KIND_PE3 = 3 +KIND_PE4 = 4 +KIND_AUTOSWITCH = 5 +KIND_INTERNAL_INTERCOM = 6 +KIND_EXTERNAL_INTERCOM = 7 +KIND_FLOOR = 13 +KIND_PAGER = 14 + +# Multimedia types +MMTYPE_AUDIO = 2 +MMTYPE_VIDEO = 3 +MMTYPE_AUDIO_VIDEO = 4 + +# Special values +END_ALL_CALLS = 4 +BROADCAST_WHERE = "4" + +# PTZ opcodes and actions +PTZ_MOVE_UP = 59 +PTZ_MOVE_DOWN = 60 +PTZ_MOVE_LEFT = 61 +PTZ_MOVE_RIGHT = 62 + +PTZ_ACTION_PRESS = 1 +PTZ_ACTION_RELEASE = 2 + +# Message types for WHO 8 +MESSAGE_TYPE_LOCK = "door_lock" +MESSAGE_TYPE_STAIRCASE_LIGHT = "staircase_light" +MESSAGE_TYPE_CAMERA = "camera" +MESSAGE_TYPE_PTZ = "ptz" +MESSAGE_TYPE_CALL = "intercom_call" +MESSAGE_TYPE_SESSION = "session" +MESSAGE_TYPE_AMPLIFIER_MUTE = "amplifier_mute" +MESSAGE_TYPE_TELELOOP = "teleloop" +MESSAGE_TYPE_VCT_INIT = "vct_init" + +_PTZ_DIRECTIONS: dict[str, int] = { + "up": PTZ_MOVE_UP, + "down": PTZ_MOVE_DOWN, + "left": PTZ_MOVE_LEFT, + "right": PTZ_MOVE_RIGHT, +} +_PTZ_OPCODE_NAMES: dict[int, str] = {v: k for k, v in _PTZ_DIRECTIONS.items()} + + +class OWNDoorEntryEvent(OWNEvent): + """Event reported by a WHO 6 door entry system or lock.""" + + def __init__(self, data: str) -> None: + super().__init__(data) + + self._where = self._where or "" + self._is_incoming_call = self._what == 6 + self._is_broadcast_call = self._what == 6 and self._where == "4100" + self._is_chime = self._what == 20 + self._is_call = self._is_incoming_call or self._is_chime + self._is_lock_open = self._what in (10, 22) + self._is_staircase_on = self._what == 12 + self._is_staircase_off = self._what == 11 + self._is_camera_on = self._what == 0 + self._is_camera_off = self._what == 9 + + if self._is_broadcast_call: + self._human_readable_log = "Incoming door entry broadcast call." + elif self._is_incoming_call: + self._human_readable_log = ( + f"Incoming door entry call to handset {self._where}." + ) + elif self._is_chime: + self._human_readable_log = f"Door entry chime active at {self._where}." + elif self._what == 10: + self._human_readable_log = ( + f"Door lock released at entrance panel {self._where}." + ) + elif self._what == 22: + self._human_readable_log = ( + f"Door lock released during conversation with entrance panel {self._where}." + ) + elif self._is_staircase_on: + self._human_readable_log = ( + f"Staircase light switched ON from door entry at {self._where}." + ) + elif self._is_staircase_off: + self._human_readable_log = ( + f"Staircase light switched OFF from door entry at {self._where}." + ) + elif self._is_camera_on: + self._human_readable_log = ( + f"Door entry camera switched ON at {self._where}." + ) + elif self._is_camera_off: + self._human_readable_log = "Door entry camera switched OFF." + elif self._what is not None: + self._human_readable_log = ( + f"Door entry device {self._where} received event: {self._what}." + ) + + @property + def is_call(self) -> bool: + """True if the event represents an incoming call or active chime.""" + return self._is_call + + @property + def is_incoming_call(self) -> bool: + """True if an incoming call frame (*6*6*WHERE##) was received.""" + return self._is_incoming_call + + @property + def is_broadcast_call(self) -> bool: + """True if a broadcast call frame (*6*6*4100##) was received.""" + return self._is_broadcast_call + + @property + def is_chime(self) -> bool: + """True if a chime event (*6*20*WHERE##) was received. + + WHAT 20 is not in the published WHO 6 table (WHAT 0, 6, 9, 10, 11, 12, + 18, 22) and no capture of it exists in this repository, so treat it as + unverified. + """ + return self._is_chime + + @property + def is_lock_open(self) -> bool: + """True if a door lock open event (*6*10*WHERE## or *6*22*WHERE##) was received.""" + return self._is_lock_open + + @property + def is_staircase_on(self) -> bool: + """True if a staircase light ON event (*6*12*WHERE##) was received.""" + return self._is_staircase_on + + @property + def is_staircase_off(self) -> bool: + """True if a staircase light OFF event (*6*11*WHERE##) was received.""" + return self._is_staircase_off + + @property + def is_camera_on(self) -> bool: + """True if a camera ON event (*6*0*WHERE##) was received.""" + return self._is_camera_on + + @property + def is_camera_off(self) -> bool: + """True if a camera OFF event (*6*9##) was received.""" + return self._is_camera_off + + +class OWNDoorEntryCommand(OWNCommand): + """WHO 6 commands for door locks, staircase lighting, and cameras.""" + + @classmethod + def open_lock(cls, where: str | int = 0) -> OWNDoorEntryCommand: + """Open / release door lock at the specified entrance panel address.""" + ep = str(where).strip() + message = cls(f"*6*10*{ep}##") + message._human_readable_log = ( + f"Opening door lock at entrance panel {ep}." + ) + return message + + @classmethod + def staircase_light_on(cls, where: str | int = 0) -> OWNDoorEntryCommand: + """Switch ON the staircase light from the door entry system.""" + ep = str(where).strip() + message = cls(f"*6*12*{ep}##") + message._human_readable_log = ( + f"Switching ON staircase light from door entry at {ep}." + ) + return message + + @classmethod + def staircase_light_off(cls, where: str | int = 0) -> OWNDoorEntryCommand: + """Switch OFF the staircase light from the door entry system.""" + ep = str(where).strip() + message = cls(f"*6*11*{ep}##") + message._human_readable_log = ( + f"Switching OFF staircase light from door entry at {ep}." + ) + return message + + @classmethod + def camera_on(cls, where: str | int) -> OWNDoorEntryCommand: + """Switch ON video door entry camera at entrance panel.""" + ep = str(where).strip() + message = cls(f"*6*0*{ep}##") + message._human_readable_log = ( + f"Switching ON camera at entrance panel {ep}." + ) + return message + + @classmethod + def camera_off(cls, where: str | int | None = None) -> OWNDoorEntryCommand: + """Switch OFF video door entry camera.""" + del where + message = cls("*6*9##") + message._human_readable_log = "Switching OFF video door entry camera." + return message + + +class OWNIntercomEvent(OWNEvent): + """Event reported by a WHO 8 video door entry / lock actuator or intercom device.""" + + def __init__(self, data: str) -> None: + super().__init__(data) + + self._where = self._where or "" + what = self._what + params = self._what_param + + self._type: str | None = None + self._is_unlocked: bool = what == 19 + self._is_locked: bool = what == 20 + self._is_lock_open: bool | None = ( + True if what == 19 else (False if what == 20 else None) + ) + self._is_staircase_on: bool = what == 21 + self._is_staircase_off: bool = what == 22 + self._is_light_on: bool | None = ( + True if what == 21 else (False if what == 22 else None) + ) + # WHAT 1 is any call session; the kind (first WHAT parameter) tells a + # door-panel ring (1..4) from an internal handset call (6) or a pager (14). + self._is_call: bool = what == 1 + self._is_incoming_call: bool = False + self._is_internal_call: bool = False + self._is_muted: bool | None = ( + True if what == 63 else (False if what == 64 else None) + ) + self._call_kind: int | None = None + self._multimedia_type: int | None = None + self._caller: str | None = None + self._callee: str | None = None + self._is_pager: bool = False + self._camera: str | None = None + self._is_cycle: bool = False + self._ptz_direction: str | None = None + self._ptz_action: str | None = None + self._session_action: str | None = None + self._teleloop_mode: int | None = None + self._vct_mode: int | None = None + + if what is None: + return + + if what in (19, 20): + self._type = MESSAGE_TYPE_LOCK + action = "unlocked / open" if self._is_unlocked else "locked / released" + self._human_readable_log = f"Lock actuator {self._where} {action}." + elif what in (21, 22): + self._type = MESSAGE_TYPE_STAIRCASE_LIGHT + action = "switched ON" if self._is_staircase_on else "switched OFF" + self._human_readable_log = f"Staircase light {self._where} is {action}." + elif what in (63, 64): + self._type = MESSAGE_TYPE_AMPLIFIER_MUTE + action = "silenced (muted)" if self._is_muted else "restored (unmuted)" + self._human_readable_log = f"Multimedia amplifier {self._where} is {action}." + elif what == 4: + self._type = MESSAGE_TYPE_CAMERA + self._caller = params[0] if params else None + self._camera = self._where + self._human_readable_log = ( + f"Camera {self._camera} switched ON by caller {self._caller}." + if self._caller + else f"Camera {self._camera} switched ON." + ) + elif what == 6: + self._type = MESSAGE_TYPE_CAMERA + self._is_cycle = True + self._caller = params[0] if params else None + self._camera = self._where + self._human_readable_log = ( + f"External camera cycling triggered by caller {self._caller} (target {self._camera})." + if self._caller + else f"External camera cycling triggered (target {self._camera})." + ) + elif what in (59, 60, 61, 62): + self._type = MESSAGE_TYPE_PTZ + self._camera = self._where + self._ptz_direction = _PTZ_OPCODE_NAMES.get(what) + action_code = ( + int(params[0]) if params and params[0].isdigit() else PTZ_ACTION_PRESS + ) + self._ptz_action = "press" if action_code == PTZ_ACTION_PRESS else "release" + self._human_readable_log = ( + f"Camera {self._camera} PTZ move {self._ptz_direction} ({self._ptz_action})." + ) + elif what == 1: + self._type = MESSAGE_TYPE_CALL + if len(params) >= 2: + self._call_kind = int(params[0]) if params[0].isdigit() else None + self._multimedia_type = int(params[1]) if params[1].isdigit() else None + if len(params) >= 3: + self._caller = params[2] + self._callee = self._where + self._is_pager = ( + self._call_kind == KIND_PAGER or self._callee == BROADCAST_WHERE + ) + self._is_incoming_call = self._call_kind in ( + KIND_PE1, + KIND_PE2, + KIND_PE3, + KIND_PE4, + ) + self._is_internal_call = ( + self._call_kind == KIND_INTERNAL_INTERCOM and not self._is_pager + ) + desc = "Pager broadcast" if self._is_pager else "Intercom call" + if self._caller: + self._human_readable_log = ( + f"{desc} from {self._caller} to {self._callee} " + f"(kind {self._call_kind}, mm {self._multimedia_type})." + ) + else: + self._human_readable_log = ( + f"{desc} to {self._callee} (kind {self._call_kind}, mm {self._multimedia_type})." + ) + elif what == 2: + self._type = MESSAGE_TYPE_SESSION + self._session_action = "answer" + if len(params) >= 2: + self._call_kind = int(params[0]) if params[0].isdigit() else None + self._multimedia_type = int(params[1]) if params[1].isdigit() else None + self._human_readable_log = f"Call answered on {self._where}." + elif what == 3: + self._type = MESSAGE_TYPE_SESSION + if len(params) >= 2 and params[1] == "3": + self._session_action = "stop_video" + self._human_readable_log = f"Video stopped on {self._where}." + else: + self._session_action = "end" + self._human_readable_log = f"Call ended on {self._where}." + if len(params) >= 2: + self._call_kind = int(params[0]) if params[0].isdigit() else None + self._multimedia_type = int(params[1]) if params[1].isdigit() else None + elif what == 9: + self._type = MESSAGE_TYPE_SESSION + self._session_action = "caller_notification" + self._caller = self._where + if len(params) >= 2: + self._call_kind = int(params[0]) if params[0].isdigit() else None + self._multimedia_type = int(params[1]) if params[1].isdigit() else None + self._human_readable_log = f"Caller notification: {self._caller}." + elif what == 37: + self._type = MESSAGE_TYPE_VCT_INIT + if params and params[0].isdigit(): + self._vct_mode = int(params[0]) + self._human_readable_log = ( + f"VCT initialized on {self._where} (mode {self._vct_mode})." + ) + elif what == 40: + self._type = MESSAGE_TYPE_SESSION + self._session_action = "rearm" + self._caller = self._where + self._human_readable_log = f"Session rearmed for {self._caller}." + elif what in (76, 77, 78, 79): + self._type = MESSAGE_TYPE_TELELOOP + if what == 76: + self._session_action = "teleloop_start" + elif what == 77: + self._session_action = "teleloop_association" + if params and params[0].isdigit(): + self._teleloop_mode = int(params[0]) + elif what == 78: + self._session_action = "teleloop_timeout" + else: + self._session_action = "teleloop_active" + self._human_readable_log = ( + f"Teleloop event {self._session_action} on {self._where}." + ) + else: + self._human_readable_log = ( + f"Lock actuator {self._where} received event: {what}." + ) + + @property + def message_type(self) -> str | None: + """Subsystem message type category.""" + return self._type + + @property + def is_unlocked(self) -> bool: + """True if lock actuator is unlocked (*8*19*WHERE##).""" + return self._is_unlocked + + @property + def is_locked(self) -> bool: + """True if lock actuator is locked (*8*20*WHERE##).""" + return self._is_locked + + @property + def is_lock_open(self) -> bool | None: + """True if lock is open, False if locked, None for non-lock frames.""" + return self._is_lock_open + + @property + def is_call(self) -> bool: + """True for any call session start (*8*1#...), doorbell or not. + + Use ``is_incoming_call`` to react to an entrance panel ringing. + """ + return self._is_call + + @property + def is_incoming_call(self) -> bool: + """True if an entrance panel (PE1..PE4, call kind 1..4) is calling. + + Internal handset calls (kind 6) and pager broadcasts (kind 14) are not + doorbell rings; see ``is_internal_call`` and ``is_pager``. + """ + return self._is_incoming_call + + @property + def is_internal_call(self) -> bool: + """True if the call is between two handsets (call kind 6).""" + return self._is_internal_call + + @property + def is_staircase_on(self) -> bool: + """True if staircase light is ON (*8*21*WHERE##).""" + return self._is_staircase_on + + @property + def is_staircase_off(self) -> bool: + """True if staircase light is OFF (*8*22*WHERE##).""" + return self._is_staircase_off + + @property + def is_light_on(self) -> bool | None: + """True if staircase light is ON, False if OFF, None if not lighting.""" + return self._is_light_on + + @property + def is_muted(self) -> bool | None: + """True if amplifier is muted, False if restored, None if not mute frame.""" + return self._is_muted + + @property + def call_kind(self) -> int | None: + """Intercom call kind (KIND_PE1..PE4, KIND_INTERNAL_INTERCOM, etc.).""" + return self._call_kind + + @property + def multimedia_type(self) -> int | None: + """Call multimedia type (MMTYPE_AUDIO, MMTYPE_VIDEO, MMTYPE_AUDIO_VIDEO).""" + return self._multimedia_type + + @property + def caller(self) -> str | None: + """Calling unit address.""" + return self._caller + + @property + def callee(self) -> str | None: + """Target handset or device address.""" + return self._callee + + @property + def is_pager(self) -> bool: + """True if call is a pager / broadcast call.""" + return self._is_pager + + @property + def camera(self) -> str | None: + """Target camera address.""" + return self._camera + + @property + def is_cycle(self) -> bool: + """True if camera cycling was triggered.""" + return self._is_cycle + + @property + def ptz_direction(self) -> str | None: + """PTZ movement direction ('up', 'down', 'left', 'right').""" + return self._ptz_direction + + @property + def ptz_action(self) -> str | None: + """PTZ action ('press', 'release').""" + return self._ptz_action + + @property + def session_action(self) -> str | None: + """Session control action ('answer', 'end', 'stop_video', 'caller_notification', etc.).""" + return self._session_action + + @property + def teleloop_mode(self) -> int | None: + """Teleloop mode parameter.""" + return self._teleloop_mode + + @property + def vct_mode(self) -> int | None: + """VCT process mode parameter.""" + return self._vct_mode + + +class OWNIntercomCommand(OWNCommand): + """WHO 8 commands for video door entry, intercoms, and lock actuators.""" + + def __init__(self, data: str) -> None: + super().__init__(data) + if self._message_type == "STATUS_REQUEST": + self._human_readable_log = ( + f"Requesting intercom status for {self._where}." + ) + + @classmethod + def unlock(cls, where: str | int) -> OWNIntercomCommand: + """Activate / unlock door lock actuator at address WHERE.""" + target = str(where).strip() + message = cls(f"*8*19*{target}##") + message._human_readable_log = ( + f"Unlocking lock actuator at address {target}." + ) + return message + + @classmethod + def lock(cls, where: str | int) -> OWNIntercomCommand: + """Release / lock door lock actuator at address WHERE.""" + target = str(where).strip() + message = cls(f"*8*20*{target}##") + message._human_readable_log = ( + f"Releasing lock actuator at address {target}." + ) + return message + + @classmethod + def open_door_lock(cls, where: str | int) -> OWNIntercomCommand: + """Activate/open door lock actuator.""" + return cls.unlock(where) + + @classmethod + def release_door_lock(cls, where: str | int) -> OWNIntercomCommand: + """Release/close door lock actuator.""" + return cls.lock(where) + + @classmethod + def staircase_light_on(cls, where: str | int) -> OWNIntercomCommand: + """Turn ON staircase light actuator.""" + target = str(where).strip() + message = cls(f"*8*21*{target}##") + message._human_readable_log = f"Turning ON staircase light {target}." + return message + + @classmethod + def staircase_light_off(cls, where: str | int) -> OWNIntercomCommand: + """Turn OFF staircase light actuator.""" + target = str(where).strip() + message = cls(f"*8*22*{target}##") + message._human_readable_log = f"Turning OFF staircase light {target}." + return message + + @classmethod + def turn_on_staircase_light(cls, where: str | int) -> OWNIntercomCommand: + return cls.staircase_light_on(where) + + @classmethod + def turn_off_staircase_light(cls, where: str | int) -> OWNIntercomCommand: + return cls.staircase_light_off(where) + + @classmethod + def switch_camera(cls, caller: str | int, camera: str | int) -> OWNIntercomCommand: + """Auto-switch / turn on a camera from a caller console.""" + caller_str = str(caller).strip() + cam_str = str(camera).strip() + message = cls(f"*8*4#{caller_str}*{cam_str}##") + message._human_readable_log = ( + f"Auto-switching camera {cam_str} from console {caller_str}." + ) + return message + + @classmethod + def cycle_camera( + cls, caller: str | int, master_caller: str | int + ) -> OWNIntercomCommand: + """Cycle to the next external unit / camera.""" + caller_str = str(caller).strip() + master_str = str(master_caller).strip() + message = cls(f"*8*6#{caller_str}*{master_str}##") + message._human_readable_log = ( + f"Cycling camera from console {caller_str} (master {master_str})." + ) + return message + + @classmethod + def ptz_move( + cls, camera: str | int, direction: str, action: str = "press" + ) -> OWNIntercomCommand: + """Send PTZ movement press or release to a camera.""" + direction_lower = direction.strip().lower() + if direction_lower not in _PTZ_DIRECTIONS: + raise ValueError( + f"direction must be one of {list(_PTZ_DIRECTIONS.keys())}, got '{direction}'" + ) + opcode = _PTZ_DIRECTIONS[direction_lower] + action_lower = action.strip().lower() + if action_lower not in ("press", "release"): + raise ValueError("action must be 'press' or 'release'") + action_code = ( + PTZ_ACTION_PRESS if action_lower == "press" else PTZ_ACTION_RELEASE + ) + cam_str = str(camera).strip() + message = cls(f"*8*{opcode}#{action_code}*{cam_str}##") + message._human_readable_log = ( + f"PTZ move {direction_lower} ({action_lower}) on camera {cam_str}." + ) + return message + + @classmethod + def call_internal( + cls, caller: str | int, callee: str | int, video: bool = False + ) -> OWNIntercomCommand: + """Initiate internal intercom call.""" + mm = MMTYPE_AUDIO_VIDEO if video else MMTYPE_AUDIO + caller_str = str(caller).strip() + callee_str = str(callee).strip() + message = cls( + f"*8*1#{KIND_INTERNAL_INTERCOM}#{mm}#{caller_str}*{callee_str}##" + ) + message._human_readable_log = ( + f"Initiating internal {'audio/video' if video else 'audio'} call " + f"from {caller_str} to {callee_str}." + ) + return message + + @classmethod + def call_external( + cls, caller: str | int, callee: str | int, video: bool = False + ) -> OWNIntercomCommand: + """Initiate external intercom call.""" + mm = MMTYPE_AUDIO_VIDEO if video else MMTYPE_AUDIO + caller_str = str(caller).strip() + callee_str = str(callee).strip() + message = cls( + f"*8*1#{KIND_EXTERNAL_INTERCOM}#{mm}#{caller_str}*{callee_str}##" + ) + message._human_readable_log = ( + f"Initiating external {'audio/video' if video else 'audio'} call " + f"from {caller_str} to {callee_str}." + ) + return message + + @classmethod + def call_pager( + cls, caller: str | int, broadcast_where: str | int = BROADCAST_WHERE + ) -> OWNIntercomCommand: + """Initiate pager broadcast call.""" + caller_str = str(caller).strip() + bc_str = str(broadcast_where).strip() + message = cls(f"*8*1#{KIND_PAGER}#{MMTYPE_AUDIO}#{caller_str}*{bc_str}##") + message._human_readable_log = ( + f"Initiating pager broadcast from {caller_str} to {bc_str}." + ) + return message + + @classmethod + def answer_call( + cls, + where: str | int, + kind: int = KIND_INTERNAL_INTERCOM, + mm_type: int = MMTYPE_AUDIO, + ) -> OWNIntercomCommand: + """Answer an incoming call.""" + target = str(where).strip() + message = cls(f"*8*2#{kind}#{mm_type}*{target}##") + message._human_readable_log = f"Answering call on {target}." + return message + + @classmethod + def end_call( + cls, + where: str | int = END_ALL_CALLS, + kind: int = KIND_INTERNAL_INTERCOM, + mm_type: int = MMTYPE_AUDIO, + ) -> OWNIntercomCommand: + """Terminate call session.""" + target = str(where).strip() + message = cls(f"*8*3#{kind}#{mm_type}*{target}##") + message._human_readable_log = f"Ending call session on {target}." + return message + + @classmethod + def stop_video( + cls, where: str | int, kind: int = KIND_INTERNAL_INTERCOM + ) -> OWNIntercomCommand: + """Stop video stream for active call.""" + target = str(where).strip() + message = cls(f"*8*3#{kind}#3*{target}##") + message._human_readable_log = f"Stopping video feed on {target}." + return message + + @classmethod + def mute_amplifier(cls, where: str | int) -> OWNIntercomCommand: + """Silence multimedia amplifier.""" + target = str(where).strip() + message = cls(f"*8*63*{target}##") + message._human_readable_log = ( + f"Silencing multimedia amplifier {target}." + ) + return message + + @classmethod + def unmute_amplifier(cls, where: str | int) -> OWNIntercomCommand: + """Restore multimedia amplifier audio.""" + target = str(where).strip() + message = cls(f"*8*64*{target}##") + message._human_readable_log = ( + f"Restoring multimedia amplifier {target}." + ) + return message + + @classmethod + def status(cls, where: str | int) -> OWNIntercomCommand: + """Request device/intercom status.""" + target = str(where).strip() + message = cls(f"*#8*{target}##") + message._human_readable_log = f"Requesting intercom status for {target}." + return message + + +# Backwards-compatible names: WHO 8 is mostly call routing, locks are one part. +OWNLockEvent = OWNIntercomEvent +OWNLockCommand = OWNIntercomCommand + +register_event_parser(6, OWNDoorEntryEvent) +register_command_parser(6, OWNDoorEntryCommand) +register_event_parser(8, OWNIntercomEvent) +register_command_parser(8, OWNIntercomCommand) diff --git a/OWNd/profiles.py b/OWNd/profiles.py index 4a68283..dbda8d4 100644 --- a/OWNd/profiles.py +++ b/OWNd/profiles.py @@ -10,6 +10,9 @@ WHO_LOAD_CONTROL = 3 WHO_HEATING = 4 WHO_ALARM = 5 +WHO_DOOR_ENTRY = 6 +WHO_LOCK = 8 +WHO_INTERCOM = 8 WHO_CEN = 15 WHO_SOUND = 16 WHO_SCENARIO = 17 @@ -393,6 +396,12 @@ class MH200Profile(GatewayProfile): page 13 "Gateways that allow the function": MH200 NO, MH200N 03565 YES), the legacy MH200 firmware supports neither, supporting only classic CEN (WHO 15). + + WHO 6 and 8 (video door entry) are listed because a live MH200 (firmware + 2.1.0, MyHOME#667) relayed ``*8*1#1#4*74##``, ``*8*9#1#4*73##`` and + ``*6*9##`` from a 2-wire entrance panel. They are bus-level subsystems: a + plant without a door entry system never emits them. No other gateway + model has a capture either way, so they are not in ``DEFAULT_SUPPORTED_WHO``. """ def __init__(self) -> None: @@ -406,6 +415,8 @@ def __init__(self) -> None: WHO_LIGHTING, WHO_AUTOMATION, WHO_HEATING, + WHO_DOOR_ENTRY, + WHO_LOCK, WHO_CEN, WHO_SOUND, WHO_SCENARIO, @@ -419,6 +430,11 @@ class MH200NProfile(GatewayProfile): Supports sound system discovery (WHO 16). Hardware verified answering ``*#16*0*5##`` without NACK and returning full source and amplifier inventory (MyHOME#427 / comment 5848181845). + + WHO 6 and 8 are listed by family with the MH200, whose capture proves the + relay (see ``MH200Profile``). Nothing has been captured on an MH200N itself; + its 1.1.8 image ships a ``bt_vct`` translator, which the firmware oracle + target does not run, so the oracle's WHO 8 NACKs for it are not evidence. """ def __init__(self) -> None: @@ -433,6 +449,8 @@ def __init__(self) -> None: WHO_LIGHTING, WHO_AUTOMATION, WHO_HEATING, + WHO_DOOR_ENTRY, + WHO_LOCK, WHO_CEN, WHO_SCENARIO, WHO_CEN_PLUS, diff --git a/tests/test_door_entry.py b/tests/test_door_entry.py new file mode 100644 index 0000000..52f0d79 --- /dev/null +++ b/tests/test_door_entry.py @@ -0,0 +1,485 @@ +"""Tests for WHO 6 and WHO 8 door entry and lock messages.""" + +from __future__ import annotations + +import pytest + +from OWNd.message import ( + BROADCAST_WHERE, + END_ALL_CALLS, + KIND_EXTERNAL_INTERCOM, + KIND_INTERNAL_INTERCOM, + KIND_PAGER, + KIND_PE1, + MESSAGE_TYPE_AMPLIFIER_MUTE, + MESSAGE_TYPE_CALL, + MESSAGE_TYPE_CAMERA, + MESSAGE_TYPE_LOCK, + MESSAGE_TYPE_PTZ, + MESSAGE_TYPE_SESSION, + MESSAGE_TYPE_STAIRCASE_LIGHT, + MESSAGE_TYPE_TELELOOP, + MESSAGE_TYPE_VCT_INIT, + MMTYPE_AUDIO, + MMTYPE_AUDIO_VIDEO, + OWNCommand, + OWNDoorEntryCommand, + OWNDoorEntryEvent, + OWNIntercomCommand, + OWNIntercomEvent, + OWNLockCommand, + OWNLockEvent, + OWNMessage, +) + + +def test_door_entry_event_incoming_call() -> None: + event = OWNDoorEntryEvent("*6*6*1##") + assert event.is_incoming_call is True + assert event.is_call is True + assert event.is_broadcast_call is False + assert event.where == "1" + assert event.human_readable_log == "Incoming door entry call to handset 1." + + +def test_door_entry_event_broadcast_call() -> None: + event = OWNDoorEntryEvent("*6*6*4100##") + assert event.is_incoming_call is True + assert event.is_call is True + assert event.is_broadcast_call is True + assert event.where == "4100" + assert event.human_readable_log == "Incoming door entry broadcast call." + + +def test_door_entry_event_chime() -> None: + event = OWNDoorEntryEvent("*6*20*1##") + assert event.is_chime is True + assert event.is_call is True + assert event.is_incoming_call is False + assert event.where == "1" + assert event.human_readable_log == "Door entry chime active at 1." + + +def test_door_entry_event_lock_open() -> None: + event1 = OWNDoorEntryEvent("*6*10*4001##") + assert event1.is_lock_open is True + assert event1.is_call is False + assert event1.human_readable_log == "Door lock released at entrance panel 4001." + + event2 = OWNDoorEntryEvent("*6*22*4001##") + assert event2.is_lock_open is True + assert ( + event2.human_readable_log + == "Door lock released during conversation with entrance panel 4001." + ) + + +def test_door_entry_event_staircase() -> None: + on_event = OWNDoorEntryEvent("*6*12*1##") + assert on_event.is_staircase_on is True + assert on_event.is_staircase_off is False + assert ( + on_event.human_readable_log + == "Staircase light switched ON from door entry at 1." + ) + + off_event = OWNDoorEntryEvent("*6*11*1##") + assert off_event.is_staircase_on is False + assert off_event.is_staircase_off is True + assert ( + off_event.human_readable_log + == "Staircase light switched OFF from door entry at 1." + ) + + +def test_door_entry_event_camera() -> None: + cam_on = OWNDoorEntryEvent("*6*0*4001##") + assert cam_on.is_camera_on is True + assert cam_on.is_camera_off is False + assert cam_on.human_readable_log == "Door entry camera switched ON at 4001." + + # All three camera OFF frame forms (standard, double-star, and single-star trailing delimiters) + for raw, expected_where in (("*6*9##", ""), ("*6*9**##", "*"), ("*6*9*##", "")): + cam_off = OWNDoorEntryEvent(raw) + assert cam_off.is_camera_off is True + assert cam_off.is_camera_on is False + assert cam_off.where == expected_where + assert cam_off.human_readable_log == "Door entry camera switched OFF." + + parsed = OWNMessage.parse(raw) + assert isinstance(parsed, OWNDoorEntryEvent) + assert parsed.is_camera_off is True + assert parsed.where == expected_where + + +def test_door_entry_commands() -> None: + cmd_lock = OWNDoorEntryCommand.open_lock(4001) + assert str(cmd_lock) == "*6*10*4001##" + assert ( + cmd_lock.human_readable_log == "Opening door lock at entrance panel 4001." + ) + + cmd_riser = OWNDoorEntryCommand.open_lock("4001#2") + assert str(cmd_riser) == "*6*10*4001#2##" + + cmd_light_on = OWNDoorEntryCommand.staircase_light_on(1) + assert str(cmd_light_on) == "*6*12*1##" + + cmd_light_off = OWNDoorEntryCommand.staircase_light_off(1) + assert str(cmd_light_off) == "*6*11*1##" + + cmd_cam_on = OWNDoorEntryCommand.camera_on(4001) + assert str(cmd_cam_on) == "*6*0*4001##" + + cmd_cam_off = OWNDoorEntryCommand.camera_off() + assert str(cmd_cam_off) == "*6*9##" + + cmd_cam_off_star = OWNDoorEntryCommand.camera_off("*") + assert str(cmd_cam_off_star) == "*6*9##" + + +def test_door_entry_event_fallback() -> None: + event = OWNDoorEntryEvent("*6*99*1##") + assert event.human_readable_log == "Door entry device 1 received event: 99." + + event_none = OWNDoorEntryEvent("*#6*1##") + assert event_none.human_readable_log == "*#6*1##" + + +def test_lock_actuator_event_and_command() -> None: + evt_unlock = OWNLockEvent("*8*19*20##") + assert evt_unlock.is_unlocked is True + assert evt_unlock.is_locked is False + assert evt_unlock.is_lock_open is True + assert evt_unlock.message_type == MESSAGE_TYPE_LOCK + assert evt_unlock.human_readable_log == "Lock actuator 20 unlocked / open." + + evt_lock = OWNLockEvent("*8*20*20##") + assert evt_lock.is_locked is True + assert evt_lock.is_unlocked is False + assert evt_lock.is_lock_open is False + assert evt_lock.human_readable_log == "Lock actuator 20 locked / released." + + evt_fallback = OWNLockEvent("*8*99*20##") + assert evt_fallback.human_readable_log == "Lock actuator 20 received event: 99." + + evt_none = OWNLockEvent("*#8*20##") + assert evt_none.human_readable_log == "*#8*20##" + + cmd_unlock = OWNLockCommand.unlock(20) + assert str(cmd_unlock) == "*8*19*20##" + assert str(OWNLockCommand.open_door_lock(20)) == "*8*19*20##" + + cmd_lock = OWNLockCommand.lock(20) + assert str(cmd_lock) == "*8*20*20##" + assert str(OWNLockCommand.release_door_lock(20)) == "*8*20*20##" + + +def test_who8_staircase_light_events_and_commands() -> None: + ev_on = OWNLockEvent("*8*21*11##") + assert ev_on.message_type == MESSAGE_TYPE_STAIRCASE_LIGHT + assert ev_on.is_staircase_on is True + assert ev_on.is_staircase_off is False + assert ev_on.is_light_on is True + assert ev_on.human_readable_log == "Staircase light 11 is switched ON." + + ev_off = OWNLockEvent("*8*22*0##") + assert ev_off.message_type == MESSAGE_TYPE_STAIRCASE_LIGHT + assert ev_off.is_staircase_on is False + assert ev_off.is_staircase_off is True + assert ev_off.is_light_on is False + assert ev_off.human_readable_log == "Staircase light 0 is switched OFF." + + cmd_on = OWNLockCommand.turn_on_staircase_light("11") + assert str(cmd_on) == "*8*21*11##" + assert str(OWNLockCommand.staircase_light_on("11")) == "*8*21*11##" + + cmd_off = OWNLockCommand.turn_off_staircase_light("0") + assert str(cmd_off) == "*8*22*0##" + assert str(OWNLockCommand.staircase_light_off("0")) == "*8*22*0##" + + +def test_who8_camera_auto_switch_cycling_and_ptz() -> None: + ev_cam = OWNLockEvent("*8*4#11*20##") + assert ev_cam.message_type == MESSAGE_TYPE_CAMERA + assert ev_cam.caller == "11" + assert ev_cam.camera == "20" + assert "Camera 20 switched ON by caller 11." in ev_cam.human_readable_log + + ev_cam_nocaller = OWNLockEvent("*8*4*20##") + assert ev_cam_nocaller.caller is None + assert ev_cam_nocaller.human_readable_log == "Camera 20 switched ON." + + ev_cycle = OWNLockEvent("*8*6#11*20##") + assert ev_cycle.message_type == MESSAGE_TYPE_CAMERA + assert ev_cycle.is_cycle is True + assert "cycling triggered by caller 11" in ev_cycle.human_readable_log + + ev_cycle_nocaller = OWNLockEvent("*8*6*20##") + assert ev_cycle_nocaller.caller is None + assert "cycling triggered (target 20)" in ev_cycle_nocaller.human_readable_log + + ev_ptz_up = OWNLockEvent("*8*59#1*20##") + assert ev_ptz_up.message_type == MESSAGE_TYPE_PTZ + assert ev_ptz_up.ptz_direction == "up" + assert ev_ptz_up.ptz_action == "press" + + ev_ptz_down = OWNLockEvent("*8*60#2*20##") + assert ev_ptz_down.ptz_direction == "down" + assert ev_ptz_down.ptz_action == "release" + + ev_ptz_left = OWNLockEvent("*8*61#1*20##") + assert ev_ptz_left.ptz_direction == "left" + assert ev_ptz_left.ptz_action == "press" + + ev_ptz_right = OWNLockEvent("*8*62#2*20##") + assert ev_ptz_right.ptz_direction == "right" + assert ev_ptz_right.ptz_action == "release" + + cmd_cam = OWNLockCommand.switch_camera(11, 20) + assert str(cmd_cam) == "*8*4#11*20##" + + cmd_cycle = OWNLockCommand.cycle_camera(11, 20) + assert str(cmd_cycle) == "*8*6#11*20##" + + cmd_up = OWNLockCommand.ptz_move(20, "up", "press") + assert str(cmd_up) == "*8*59#1*20##" + cmd_down = OWNLockCommand.ptz_move(20, "down", "release") + assert str(cmd_down) == "*8*60#2*20##" + cmd_left = OWNLockCommand.ptz_move(20, "left", "press") + assert str(cmd_left) == "*8*61#1*20##" + cmd_right = OWNLockCommand.ptz_move(20, "right", "release") + assert str(cmd_right) == "*8*62#2*20##" + + with pytest.raises(ValueError, match="direction must be one of"): + OWNLockCommand.ptz_move(20, "diagonal", "press") + with pytest.raises(ValueError, match="action must be"): + OWNLockCommand.ptz_move(20, "up", "hold") + + +def test_who8_intercom_calling_and_session() -> None: + # Real trace from MH200 (#667): *8*1#1#4*74## (PE1, Audio/Video to handset 74) + ev_mh200_call = OWNLockEvent("*8*1#1#4*74##") + assert ev_mh200_call.message_type == MESSAGE_TYPE_CALL + assert ev_mh200_call.is_call is True + assert ev_mh200_call.is_incoming_call is True + assert ev_mh200_call.call_kind == KIND_PE1 + assert ev_mh200_call.multimedia_type == MMTYPE_AUDIO_VIDEO + assert ev_mh200_call.caller is None + assert ev_mh200_call.callee == "74" + assert ev_mh200_call.human_readable_log == "Intercom call to 74 (kind 1, mm 4)." + + # Bare call without params: *8*1*74## + ev_bare_call = OWNLockEvent("*8*1*74##") + assert ev_bare_call.message_type == MESSAGE_TYPE_CALL + assert ev_bare_call.is_call is True + assert ev_bare_call.call_kind is None + assert ev_bare_call.is_incoming_call is False + assert ev_bare_call.callee == "74" + + # Internal audio call: *8*1#6#2#11*16## (caller 11, callee 16, kind 6, mm 2) + ev_call = OWNLockEvent("*8*1#6#2#11*16##") + assert ev_call.message_type == MESSAGE_TYPE_CALL + assert ev_call.is_call is True + # An internal handset call is not a doorbell ring. + assert ev_call.is_incoming_call is False + assert ev_call.is_internal_call is True + assert ev_call.call_kind == KIND_INTERNAL_INTERCOM + assert ev_call.multimedia_type == MMTYPE_AUDIO + assert ev_call.caller == "11" + assert ev_call.callee == "16" + assert ev_call.is_pager is False + assert "Intercom call from 11 to 16" in ev_call.human_readable_log + + # Pager call: *8*1#14#2#11*4## + ev_pager = OWNLockEvent("*8*1#14#2#11*4##") + assert ev_pager.is_pager is True + assert ev_pager.is_call is True + assert ev_pager.is_incoming_call is False + assert ev_pager.is_internal_call is False + assert "Pager broadcast from 11 to 4" in ev_pager.human_readable_log + + # Answer call: *8*2#6#2*11## + ev_ans = OWNLockEvent("*8*2#6#2*11##") + assert ev_ans.message_type == MESSAGE_TYPE_SESSION + assert ev_ans.session_action == "answer" + assert ev_ans.call_kind == KIND_INTERNAL_INTERCOM + assert ev_ans.multimedia_type == MMTYPE_AUDIO + assert ev_ans.human_readable_log == "Call answered on 11." + + # Answer without params + ev_ans_noparams = OWNLockEvent("*8*2*11##") + assert ev_ans_noparams.session_action == "answer" + + # End call: *8*3#6#2*411## + ev_end = OWNLockEvent("*8*3#6#2*411##") + assert ev_end.message_type == MESSAGE_TYPE_SESSION + assert ev_end.session_action == "end" + assert ev_end.human_readable_log == "Call ended on 411." + + # End without params + ev_end_noparams = OWNLockEvent("*8*3*11##") + assert ev_end_noparams.session_action == "end" + + # Stop video: *8*3#1#3*11## + ev_stop_vid = OWNLockEvent("*8*3#1#3*11##") + assert ev_stop_vid.session_action == "stop_video" + assert ev_stop_vid.human_readable_log == "Video stopped on 11." + + # Real trace from MH200 (#667): *8*9#1#4*73## (caller notification from PE 73) + ev_notif = OWNLockEvent("*8*9#1#4*73##") + assert ev_notif.message_type == MESSAGE_TYPE_SESSION + assert ev_notif.session_action == "caller_notification" + assert ev_notif.caller == "73" + assert ev_notif.call_kind == KIND_PE1 + assert ev_notif.multimedia_type == MMTYPE_AUDIO_VIDEO + assert ev_notif.human_readable_log == "Caller notification: 73." + + # Caller notification without params + ev_notif_noparams = OWNLockEvent("*8*9*11##") + assert ev_notif_noparams.session_action == "caller_notification" + + # Session rearm: *8*40*11## + ev_rearm = OWNLockEvent("*8*40*11##") + assert ev_rearm.session_action == "rearm" + assert ev_rearm.caller == "11" + assert ev_rearm.human_readable_log == "Session rearmed for 11." + + # Builders + cmd_int_audio = OWNLockCommand.call_internal("11", "16", video=False) + assert str(cmd_int_audio) == "*8*1#6#2#11*16##" + + cmd_int_video = OWNLockCommand.call_internal("11", "16", video=True) + assert str(cmd_int_video) == "*8*1#6#4#11*16##" + + cmd_ext_audio = OWNLockCommand.call_external("11", "16", video=False) + assert str(cmd_ext_audio) == "*8*1#7#2#11*16##" + + cmd_ext_video = OWNLockCommand.call_external("11", "16", video=True) + assert str(cmd_ext_video) == "*8*1#7#4#11*16##" + + cmd_pager = OWNLockCommand.call_pager("11", BROADCAST_WHERE) + assert str(cmd_pager) == "*8*1#14#2#11*4##" + + cmd_ans = OWNLockCommand.answer_call( + "11", kind=KIND_INTERNAL_INTERCOM, mm_type=MMTYPE_AUDIO + ) + assert str(cmd_ans) == "*8*2#6#2*11##" + + cmd_end = OWNLockCommand.end_call( + "411", kind=KIND_INTERNAL_INTERCOM, mm_type=MMTYPE_AUDIO + ) + assert str(cmd_end) == "*8*3#6#2*411##" + assert str(OWNLockCommand.end_call()) == f"*8*3#6#2*{END_ALL_CALLS}##" + + cmd_stop_vid = OWNLockCommand.stop_video("11", kind=KIND_PE1) + assert str(cmd_stop_vid) == "*8*3#1#3*11##" + + +def test_who8_amplifier_mute_teleloop_and_vct() -> None: + # Mute: *8*63*11## + ev_mute = OWNLockEvent("*8*63*11##") + assert ev_mute.message_type == MESSAGE_TYPE_AMPLIFIER_MUTE + assert ev_mute.is_muted is True + assert "silenced (muted)" in ev_mute.human_readable_log + + # Unmute: *8*64*11## + ev_unmute = OWNLockEvent("*8*64*11##") + assert ev_unmute.is_muted is False + assert "unmuted" in ev_unmute.human_readable_log + + # Teleloop start: *8*76*11## + ev_tl0 = OWNLockEvent("*8*76*11##") + assert ev_tl0.message_type == MESSAGE_TYPE_TELELOOP + assert ev_tl0.session_action == "teleloop_start" + + # Teleloop assoc: *8*77#5*11## + ev_tl1 = OWNLockEvent("*8*77#5*11##") + assert ev_tl1.message_type == MESSAGE_TYPE_TELELOOP + assert ev_tl1.session_action == "teleloop_association" + assert ev_tl1.teleloop_mode == 5 + + # Teleloop assoc without params + ev_tl_noparam = OWNLockEvent("*8*77*11##") + assert ev_tl_noparam.session_action == "teleloop_association" + + # Teleloop timeout: *8*78*11## + ev_tl2 = OWNLockEvent("*8*78*11##") + assert ev_tl2.message_type == MESSAGE_TYPE_TELELOOP + assert ev_tl2.session_action == "teleloop_timeout" + + # Teleloop active: *8*79*11## + ev_tl_active = OWNLockEvent("*8*79*11##") + assert ev_tl_active.session_action == "teleloop_active" + + # VCT init: *8*37#1*11## + ev_vct = OWNLockEvent("*8*37#1*11##") + assert ev_vct.message_type == MESSAGE_TYPE_VCT_INIT + assert ev_vct.vct_mode == 1 + + # VCT without params + ev_vct_noparam = OWNLockEvent("*8*37*11##") + assert ev_vct_noparam.message_type == MESSAGE_TYPE_VCT_INIT + + # Builders + cmd_mute = OWNLockCommand.mute_amplifier("11") + assert str(cmd_mute) == "*8*63*11##" + + cmd_unmute = OWNLockCommand.unmute_amplifier("11") + assert str(cmd_unmute) == "*8*64*11##" + + cmd_status = OWNLockCommand.status("11") + assert str(cmd_status) == "*#8*11##" + assert cmd_status.human_readable_log == "Requesting intercom status for 11." + + # *#8*WHERE*19## has no documented meaning; no builder is offered for it. + assert not hasattr(OWNIntercomCommand, "lock_status") + + +def test_who8_primary_classes_and_lock_aliases() -> None: + assert OWNLockEvent is OWNIntercomEvent + assert OWNLockCommand is OWNIntercomCommand + assert OWNIntercomEvent.__name__ == "OWNIntercomEvent" + + +@pytest.mark.parametrize("raw", ["*6*3##", "*6*42##", "*6*6##", "*6*10##", "*6*0*##"]) +def test_who6_short_form_only_accepts_camera_off(raw: str) -> None: + """Only WHAT 9 may omit WHERE; other short WHO 6 frames are not valid.""" + assert not isinstance(OWNMessage.parse(raw), OWNDoorEntryEvent) + + +def test_message_registry_parsing() -> None: + msg6_lock = OWNMessage.parse("*6*10*4001##") + assert isinstance(msg6_lock, OWNDoorEntryEvent) + + msg6_call = OWNMessage.parse("*6*6*1##") + assert isinstance(msg6_call, OWNDoorEntryEvent) + + msg6_cam_off = OWNMessage.parse("*6*9##") + assert isinstance(msg6_cam_off, OWNDoorEntryEvent) + + cmd6 = OWNCommand.parse("*6*10*4001##") + assert isinstance(cmd6, OWNDoorEntryCommand) + + # WHO 8 registry parsing + msg8_lock = OWNMessage.parse("*8*19*20##") + assert isinstance(msg8_lock, OWNLockEvent) + + msg8_rel = OWNMessage.parse("*8*20*20##") + assert isinstance(msg8_rel, OWNLockEvent) + + msg8_call = OWNMessage.parse("*8*1#1#4*74##") + assert isinstance(msg8_call, OWNLockEvent) + + msg8_notif = OWNMessage.parse("*8*9#1#4*73##") + assert isinstance(msg8_notif, OWNLockEvent) + + cmd8_unlock = OWNCommand.parse("*8*19*20##") + assert isinstance(cmd8_unlock, OWNLockCommand) + + cmd8_lock = OWNCommand.parse("*8*20*20##") + assert isinstance(cmd8_lock, OWNLockCommand) + + # Aliases + assert OWNIntercomEvent is OWNLockEvent + assert OWNIntercomCommand is OWNLockCommand diff --git a/tests/test_firmware_oracle_conformance.py b/tests/test_firmware_oracle_conformance.py index 7cd8d2c..6c87ecc 100644 --- a/tests/test_firmware_oracle_conformance.py +++ b/tests/test_firmware_oracle_conformance.py @@ -23,6 +23,8 @@ OWNEvent, OWNLightingCommand, OWNLightingEvent, + OWNLockCommand, + OWNLockEvent, OWNMessage, OWNSignaling, ) @@ -172,10 +174,9 @@ def test_oracle_input_frames_ownd_parser_resilience(): Guarantees that: 1. Zero input frames cause unhandled parser crashes (AttributeError, IndexError, ValueError). - 2. Exactly 297 frames parse into valid typed OWNMessage instances across subsystems + 2. Exactly 303 frames parse into valid typed OWNMessage instances across subsystems WHO 0, 1, 2, 3, 4, 8, 14, 15, 18, 22, 25. - 3. Exactly 6 frames (WHO 8 global status queries `*#8*...##` without WHERE) safely return None. - 4. For all 297 parsed frames, msg.who matches the subsystem extracted from the frame syntax. + 3. For all 303 parsed frames, msg.who matches the subsystem extracted from the frame syntax. """ parsed_count = 0 unparsed_count = 0 @@ -196,11 +197,24 @@ def test_oracle_input_frames_ownd_parser_resilience(): ) else: unparsed_count += 1 - # Unparsed frames are exclusively WHO 8 status queries without target address - assert frame.startswith("*#8*"), f"Unexpected unparsed frame {frame}" - assert parsed_count == 297 - assert unparsed_count == 6 + assert parsed_count == 303 + assert unparsed_count == 0 + + +def test_oracle_who8_inputs_parse_into_specialized_classes() -> None: + """Verify that all authentic oracle inputs for WHO 8 parse into specialized classes.""" + who8_count = 0 + for inp in ALL_VERDICTS: + if inp.startswith("*8*") or inp.startswith("*#8*"): + who8_count += 1 + msg = OWNMessage.parse(inp) + assert msg is not None, f"Failed to parse WHO 8 message: {inp}" + assert isinstance(msg, (OWNLockEvent, OWNLockCommand)), ( + f"Expected OWNLockEvent/Command for {inp}, got {type(msg)}" + ) + assert who8_count == 59 + def test_golden_corpus_bidirectional_cross_validation(): diff --git a/tests/test_profiles.py b/tests/test_profiles.py index cf4740a..08dcd16 100644 --- a/tests/test_profiles.py +++ b/tests/test_profiles.py @@ -14,10 +14,12 @@ WHO_AUTOMATION, WHO_CEN, WHO_CEN_PLUS, + WHO_DOOR_ENTRY, WHO_ENERGY, WHO_HEATING, WHO_LIGHTING, WHO_LOAD_CONTROL, + WHO_LOCK, WHO_SCENARIO, WHO_SOUND, WHO_SOUND_DIFFUSION, @@ -63,8 +65,22 @@ def test_mh200_keeps_the_mh200n_pacing() -> None: assert set(mh200n.supported_who) - set(mh200.supported_who) == {WHO_CEN_PLUS} assert mh200.supports_who(WHO_CEN) is True assert not mh200.supports_who(WHO_CEN_PLUS) + assert mh200.supports_who(WHO_DOOR_ENTRY) is True + assert mh200.supports_who(WHO_LOCK) is True assert mh200n.supports_who(WHO_CEN) is True assert mh200n.supports_who(WHO_CEN_PLUS) is True + assert mh200n.supports_who(WHO_DOOR_ENTRY) is True + assert mh200n.supports_who(WHO_LOCK) is True + + +def test_door_entry_not_advertised_by_default() -> None: + """Only the MH200 has a door-entry capture; other profiles must not claim it.""" + assert WHO_DOOR_ENTRY not in DEFAULT_SUPPORTED_WHO + assert WHO_LOCK not in DEFAULT_SUPPORTED_WHO + for model in ("MH201", "F452", "F454", "MyHomeServer1"): + profile = get_gateway_profile(model) + assert not profile.supports_who(WHO_DOOR_ENTRY), model + assert not profile.supports_who(WHO_LOCK), model def test_mh200n_audio_enabled_and_verified() -> None: