From 3ee6c7b6e3df546304316c5feb688b0827db9949 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Tue, 18 Aug 2026 15:10:54 +0200 Subject: [PATCH] feat(s7commplus): add alarm handling --- s7commplus/__init__.py | 4 + s7commplus/alarm.py | 422 +++++++++++++++++++++++++++++++++++++ s7commplus/async_client.py | 44 ++++ s7commplus/client.py | 59 ++++++ s7commplus/connection.py | 20 ++ s7commplus/protocol.py | 24 +++ tests/test_s7_alarm.py | 167 +++++++++++++++ 7 files changed, 740 insertions(+) create mode 100644 s7commplus/alarm.py create mode 100644 tests/test_s7_alarm.py diff --git a/s7commplus/__init__.py b/s7commplus/__init__.py index 7e5199d5..ac57773c 100644 --- a/s7commplus/__init__.py +++ b/s7commplus/__init__.py @@ -18,6 +18,7 @@ from .async_client import S7CommPlusAsyncClient as AsyncClient from .server import S7CommPlusServer as Server, DataBlock, CPUState from .connection import S7CommPlusConnection +from .alarm import Alarm, AlarmNotification, AlarmText from .tag_browser import ( Tag, Member, @@ -34,6 +35,9 @@ "DataBlock", "CPUState", "S7CommPlusConnection", + "Alarm", + "AlarmNotification", + "AlarmText", "decompress_blob", "find_and_decompress", "Tag", diff --git a/s7commplus/alarm.py b/s7commplus/alarm.py new file mode 100644 index 00000000..46897def --- /dev/null +++ b/s7commplus/alarm.py @@ -0,0 +1,422 @@ +"""Alarm models and wire decoders for S7CommPlus notifications.""" + +from __future__ import annotations + +import struct +from dataclasses import dataclass, field +from typing import Any + +from .protocol import DataType, ElementID, Ids, Opcode +from .vlq import decode_int32_vlq, decode_int64_vlq, decode_uint32_vlq, decode_uint64_vlq, encode_uint32_vlq + +_ALARM_SUBSCRIPTION_RELATION_ID = 0x7FFFC001 +_ALARM_REFERENCE_RELATION_ID = 0x51010001 + + +@dataclass(frozen=True) +class AlarmText: + """The texts for one alarm in one PLC language.""" + + language_id: int + info_text: str = "" + alarm_text: str = "" + additional_texts: tuple[str, ...] = () + + +@dataclass(frozen=True) +class Alarm: + """Current state of a PLC alarm.""" + + cpu_alarm_id: int + all_states_info: int + domain: int + message_type: int + sequence_counter: int + name: str = "" + state: str = "unknown" + timestamp: int | None = None + acknowledge_timestamp: int | None = None + hmi_info: bytes = b"" + associated_values: tuple[bytes, ...] = () + texts: dict[int, AlarmText] = field(default_factory=dict) + + +@dataclass(frozen=True) +class AlarmNotification: + """An unsolicited S7CommPlus alarm notification.""" + + subscription_id: int + credit_tick: int + sequence_number: int + subscription_change_counter: int + timestamp: int | None + alarms: tuple[Alarm, ...] + + +@dataclass +class _Object: + relation_id: int + class_id: int + attributes: dict[int, Any] = field(default_factory=dict) + children: list[_Object] = field(default_factory=list) + + +@dataclass(frozen=True) +class _Blob: + root_id: int + value: bytes + + +def _attribute(attribute_id: int, datatype: int, value: bytes, flags: int = 0) -> bytes: + return bytes([ElementID.ATTRIBUTE]) + encode_uint32_vlq(attribute_id) + bytes([flags, datatype]) + value + + +def _wstring(value: str) -> bytes: + encoded = value.encode("utf-8") + return encode_uint32_vlq(len(encoded)) + encoded + + +def _udint_array(values: list[int], flags: int = 0x20) -> bytes: + return ( + bytes([flags, DataType.UDINT]) + encode_uint32_vlq(len(values)) + b"".join(encode_uint32_vlq(value) for value in values) + ) + + +def _uint_array(values: list[int], flags: int = 0x10) -> bytes: + return bytes([flags, DataType.UINT]) + encode_uint32_vlq(len(values)) + b"".join(struct.pack(">H", value) for value in values) + + +def build_alarm_subscription_request( + session_id: int, + language_ids: list[int] | None = None, + domains: list[int] | None = None, + credit_limit: int = -1, +) -> bytes: + """Build an alarm-subscription CREATE_OBJECT payload.""" + if not -1 <= credit_limit <= 255: + raise ValueError("credit_limit must be -1 (unlimited) or between 0 and 255") + languages = [] if language_ids is None else language_ids + domain_filter = [0xFFFF] if domains is None else domains + if any(not 0 <= value <= 0xFFFF for value in domain_filter): + raise ValueError("alarm domains must be UInt16 values") + if any(not 0 <= value <= 0xFFFFFFFF for value in languages): + raise ValueError("language IDs must be UInt32 values") + + payload = bytearray() + payload += struct.pack(">I", session_id) + payload += bytes([0, DataType.UDINT]) + encode_uint32_vlq(0) + payload += struct.pack(">I", 0) + payload += bytes([ElementID.START_OF_OBJECT]) + payload += struct.pack(">I", _ALARM_SUBSCRIPTION_RELATION_ID) + payload += encode_uint32_vlq(Ids.CLASS_SUBSCRIPTION) + payload += encode_uint32_vlq(0) + encode_uint32_vlq(0) + payload += _attribute( + Ids.OBJECT_VARIABLE_TYPE_NAME, DataType.WSTRING, _wstring(f"PyAlarm_{_ALARM_SUBSCRIPTION_RELATION_ID:#x}") + ) + payload += _attribute(Ids.SUBSCRIPTION_FUNCTION_CLASS_ID, DataType.USINT, b"\x02") + payload += _attribute(Ids.SUBSCRIPTION_MISSED_SENDINGS, DataType.UINT, struct.pack(">H", 0)) + payload += _attribute(Ids.SUBSCRIPTION_SUBSYSTEM_ERROR, DataType.LINT, encode_uint32_vlq(0)) + payload += _attribute(Ids.SUBSCRIPTION_ROUTE_MODE, DataType.USINT, b"\x02") + payload += _attribute(Ids.SUBSCRIPTION_ACTIVE, DataType.BOOL, b"\x01") + payload += bytes([ElementID.ATTRIBUTE]) + encode_uint32_vlq(Ids.SUBSCRIPTION_REFERENCE_LIST) + payload += _udint_array([0x80010000, 0, 0]) + payload += _attribute(Ids.SUBSCRIPTION_CYCLE_TIME, DataType.UDINT, encode_uint32_vlq(0)) + payload += _attribute(Ids.SUBSCRIPTION_DELAY_TIME, DataType.UDINT, encode_uint32_vlq(0)) + payload += _attribute(Ids.SUBSCRIPTION_DISABLED, DataType.USINT, b"\x00") + payload += _attribute(Ids.SUBSCRIPTION_COUNT, DataType.USINT, b"\x00") + payload += _attribute(Ids.SUBSCRIPTION_CREDIT_LIMIT, DataType.INT, struct.pack(">h", credit_limit)) + payload += _attribute(Ids.SUBSCRIPTION_TICKS, DataType.UINT, struct.pack(">H", 0xFFFF)) + + payload += bytes([ElementID.START_OF_OBJECT]) + payload += struct.pack(">I", _ALARM_REFERENCE_RELATION_ID) + payload += encode_uint32_vlq(Ids.ALARM_SUBSCRIPTION_REF_CLASS_RID) + payload += encode_uint32_vlq(0) + encode_uint32_vlq(0) + payload += _attribute(Ids.OBJECT_VARIABLE_TYPE_NAME, DataType.WSTRING, _wstring("python-snap7 alarms")) + payload += _attribute(Ids.SUBSCRIPTION_REFERENCE_TRIGGER_MODE, DataType.USINT, b"\x03") + payload += bytes([ElementID.ATTRIBUTE]) + encode_uint32_vlq(Ids.ALARM_SUBSCRIPTION_REF_ALARM_DOMAIN) + payload += _uint_array([0] * 10) + payload += bytes([ElementID.ATTRIBUTE]) + encode_uint32_vlq(Ids.ALARM_SUBSCRIPTION_REF_ALARM_DOMAIN_FILTER) + payload += _uint_array(domain_filter, flags=0x20) + payload += bytes([ElementID.ATTRIBUTE]) + encode_uint32_vlq(Ids.ALARM_SUBSCRIPTION_REF_TEXT_LANGUAGES) + payload += _udint_array(languages) + payload += _attribute(Ids.ALARM_SUBSCRIPTION_REF_SEND_TEXTS, DataType.BOOL, b"\x01") + payload += bytes([ElementID.RELATION]) + payload += encode_uint32_vlq(Ids.ALARM_SUBSCRIPTION_REF_ITS_ALARM_SUBSYSTEM) + payload += struct.pack(">I", Ids.NATIVE_THE_ALARM_SUBSYSTEM_RID) + payload += bytes([ElementID.TERMINATING_OBJECT, ElementID.TERMINATING_OBJECT]) + payload += struct.pack(">I", 0) + return bytes(payload) + + +def build_alarm_explore_request() -> bytes: + """Build an EXPLORE request for the current alarm state.""" + attributes = [ + Ids.ALARM_DAI_CPU_ALARM_ID, + Ids.ALARM_DAI_ALL_STATES_INFO, + Ids.ALARM_DAI_DOMAIN, + Ids.ALARM_DAI_COMING, + Ids.ALARM_DAI_GOING, + Ids.ALARM_DAI_MESSAGE_TYPE, + Ids.ALARM_DAI_HMI_INFO, + Ids.OBJECT_VARIABLE_TYPE_NAME, + Ids.ALARM_DAI_SEQUENCE_COUNTER, + Ids.ALARM_DAI_TEXTS, + ] + payload = bytearray(struct.pack(">I", Ids.NATIVE_THE_ALARM_SUBSYSTEM_RID)) + payload += encode_uint32_vlq(Ids.ALARM_SUBSYSTEM_UPDATE_RELEVANT_DAI) + payload += b"\x01\x01\x00\x00" + payload += encode_uint32_vlq(len(attributes)) + for attribute_id in attributes: + payload += encode_uint32_vlq(attribute_id) + payload += struct.pack(">I", 0) + b"\x00" + return bytes(payload) + + +def _read_vlq32(data: bytes, offset: int) -> tuple[int, int]: + value, consumed = decode_uint32_vlq(data, offset) + return value, offset + consumed + + +def _read_vlq64(data: bytes, offset: int) -> tuple[int, int]: + value, consumed = decode_uint64_vlq(data, offset) + return value, offset + consumed + + +def _decode_blob(data: bytes, offset: int) -> tuple[_Blob, int]: + root_id, offset = _read_vlq32(data, offset) + if root_id > 1: + if offset + 9 > len(data): + raise ValueError("Truncated typed alarm blob") + offset += 8 + blob_type = data[offset] + offset += 1 + if blob_type not in (2, 3): + raise ValueError(f"Unsupported alarm blob type: {blob_type}") + size, offset = _read_vlq32(data, offset) + end = offset + size + if end > len(data): + raise ValueError("Truncated alarm blob") + return _Blob(root_id, bytes(data[offset:end])), end + + +def _decode_scalar(data: bytes, offset: int, datatype: int) -> tuple[Any, int]: + if datatype == DataType.NULL: + return None, offset + if datatype == DataType.BOOL: + return bool(data[offset]), offset + 1 + if datatype in (DataType.USINT, DataType.BYTE): + return data[offset], offset + 1 + if datatype == DataType.SINT: + return struct.unpack_from(">b", data, offset)[0], offset + 1 + if datatype in (DataType.UINT, DataType.WORD): + return struct.unpack_from(">H", data, offset)[0], offset + 2 + if datatype == DataType.INT: + return struct.unpack_from(">h", data, offset)[0], offset + 2 + if datatype in (DataType.UDINT, DataType.AID): + return _read_vlq32(data, offset) + if datatype == DataType.DINT: + value, consumed = decode_int32_vlq(data, offset) + return value, offset + consumed + if datatype == DataType.ULINT: + return _read_vlq64(data, offset) + if datatype in (DataType.LINT, DataType.TIMESPAN): + value, consumed = decode_int64_vlq(data, offset) + return value, offset + consumed + if datatype in (DataType.DWORD, DataType.RID): + return struct.unpack_from(">I", data, offset)[0], offset + 4 + if datatype == DataType.LWORD: + return struct.unpack_from(">Q", data, offset)[0], offset + 8 + if datatype == DataType.REAL: + return struct.unpack_from(">f", data, offset)[0], offset + 4 + if datatype == DataType.LREAL: + return struct.unpack_from(">d", data, offset)[0], offset + 8 + if datatype == DataType.TIMESTAMP: + return struct.unpack_from(">Q", data, offset)[0], offset + 8 + if datatype == DataType.WSTRING: + size, offset = _read_vlq32(data, offset) + end = offset + size + return data[offset:end].decode("utf-8", errors="replace"), end + if datatype == DataType.BLOB: + return _decode_blob(data, offset) + if datatype == DataType.STRUCT: + struct_id = struct.unpack_from(">I", data, offset)[0] + offset += 4 + members: dict[int, Any] = {0: struct_id} + while offset < len(data) and data[offset] != 0: + member_id, offset = _read_vlq32(data, offset) + value, offset = _decode_value(data, offset) + members[member_id] = value + return members, offset + 1 + raise ValueError(f"Unsupported alarm value datatype: {datatype:#x}") + + +def _decode_value(data: bytes, offset: int) -> tuple[Any, int]: + if offset + 2 > len(data): + raise ValueError("Truncated alarm value") + flags, datatype = data[offset], data[offset + 1] + offset += 2 + if flags == 0x40: + values: dict[int, Any] = {} + key, offset = _read_vlq32(data, offset) + while key: + if datatype == DataType.BLOB: + value, offset = _decode_blob(data, offset) + else: + value, offset = _decode_scalar(data, offset, datatype) + values[key] = value + key, offset = _read_vlq32(data, offset) + return values, offset + if flags in (0x10, 0x20): + count, offset = _read_vlq32(data, offset) + array_values: list[Any] = [] + for _ in range(count): + value, offset = _decode_scalar(data, offset, datatype) + array_values.append(value) + return array_values, offset + return _decode_scalar(data, offset, datatype) + + +def _decode_object(data: bytes, offset: int) -> tuple[_Object, int]: + if data[offset] != ElementID.START_OF_OBJECT: + raise ValueError("Expected S7CommPlus object") + offset += 1 + relation_id = struct.unpack_from(">I", data, offset)[0] + offset += 4 + class_id, offset = _read_vlq32(data, offset) + _, offset = _read_vlq32(data, offset) # class flags + _, offset = _read_vlq32(data, offset) # attribute id + result = _Object(relation_id, class_id) + while offset < len(data): + tag = data[offset] + if tag == ElementID.TERMINATING_OBJECT: + return result, offset + 1 + if tag == ElementID.START_OF_OBJECT: + child, offset = _decode_object(data, offset) + result.children.append(child) + continue + if tag == ElementID.ATTRIBUTE: + attribute_id, value_offset = _read_vlq32(data, offset + 1) + value, offset = _decode_value(data, value_offset) + result.attributes[attribute_id] = value + continue + if tag == ElementID.RELATION: + _, offset = _read_vlq32(data, offset + 1) + offset += 4 + continue + raise ValueError(f"Unsupported object element: {tag:#x}") + raise ValueError("Unterminated S7CommPlus object") + + +def _decode_objects(data: bytes, offset: int) -> tuple[list[_Object], int]: + objects = [] + while offset < len(data) and data[offset] == ElementID.START_OF_OBJECT: + obj, offset = _decode_object(data, offset) + objects.append(obj) + return objects, offset + + +def _alarm_texts(value: Any, language_ids: set[int] | None) -> dict[int, AlarmText]: + grouped: dict[int, dict[int, str]] = {} + if not isinstance(value, dict): + return {} + for key, blob in value.items(): + if not isinstance(key, int) or not isinstance(blob, _Blob): + continue + language_id, text_id = key >> 16, key & 0xFFFF + if language_ids is not None and language_id not in language_ids: + continue + grouped.setdefault(language_id, {})[text_id] = blob.value.decode("utf-8", errors="replace") + result = {} + for language_id, texts in grouped.items(): + additional = tuple(texts.get(i, "") for i in range(3, 12)) + result[language_id] = AlarmText(language_id, texts.get(1, ""), texts.get(2, ""), additional) + return result + + +def _alarm_from_object(obj: _Object, language_ids: set[int] | None) -> Alarm: + attrs = obj.attributes + state_id = Ids.ALARM_DAI_COMING if Ids.ALARM_DAI_COMING in attrs else Ids.ALARM_DAI_GOING + state_value = attrs.get(state_id) + state = "coming" if state_id == Ids.ALARM_DAI_COMING else "going" + timestamp: int | None = None + acknowledge_timestamp: int | None = None + associated_values: tuple[bytes, ...] = () + if isinstance(state_value, dict): + timestamp_value = state_value.get(3475) + acknowledge_value = state_value.get(3646) + timestamp = timestamp_value if isinstance(timestamp_value, int) else None + acknowledge_timestamp = acknowledge_value if isinstance(acknowledge_value, int) else None + raw_values = state_value.get(3476) + if isinstance(raw_values, list): + associated_values = tuple(value.value for value in raw_values if isinstance(value, _Blob)) + hmi = attrs.get(Ids.ALARM_DAI_HMI_INFO) + return Alarm( + cpu_alarm_id=int(attrs.get(Ids.ALARM_DAI_CPU_ALARM_ID, 0)), + all_states_info=int(attrs.get(Ids.ALARM_DAI_ALL_STATES_INFO, 0)), + domain=int(attrs.get(Ids.ALARM_DAI_DOMAIN, 0)), + message_type=int(attrs.get(Ids.ALARM_DAI_MESSAGE_TYPE, 0)), + sequence_counter=int(attrs.get(Ids.ALARM_DAI_SEQUENCE_COUNTER, 0)), + name=str(attrs.get(Ids.OBJECT_VARIABLE_TYPE_NAME, "")), + state=state if state_value is not None else "unknown", + timestamp=timestamp, + acknowledge_timestamp=acknowledge_timestamp, + hmi_info=hmi.value if isinstance(hmi, _Blob) else b"", + associated_values=associated_values, + texts=_alarm_texts(attrs.get(Ids.ALARM_DAI_TEXTS), language_ids), + ) + + +def parse_alarm_explore_response(response: bytes, language_ids: list[int] | None = None) -> list[Alarm]: + """Parse the payload returned by an alarm-subsystem EXPLORE request.""" + return_value, offset = _read_vlq64(response, 0) + if return_value != 0: + raise RuntimeError(f"Alarm browse failed: PLC returned {return_value:#x}") + if offset + 4 > len(response): + raise ValueError("Alarm browse response is truncated") + offset += 4 # ExploreId + # IntegrityId is between ExploreId and the object list on V2+ responses. + while offset < len(response) and response[offset] != ElementID.START_OF_OBJECT: + _, offset = _read_vlq32(response, offset) + objects, _ = _decode_objects(response, offset) + wanted = set(language_ids) if language_ids is not None else None + return [_alarm_from_object(obj, wanted) for obj in objects if obj.class_id == Ids.ALARM_DAI_CLASS_RID] + + +def parse_alarm_notification(frame: bytes, language_ids: list[int] | None = None) -> AlarmNotification: + """Parse one complete S7CommPlus notification frame.""" + if len(frame) < 5 or frame[0] != 0x72: + raise ValueError("Invalid S7CommPlus notification frame") + data_length = struct.unpack_from(">H", frame, 2)[0] + data = frame[4 : 4 + data_length] + if not data or data[0] != Opcode.NOTIFICATION: + raise ValueError("Expected S7CommPlus notification opcode") + offset = 1 + subscription_id = struct.unpack_from(">I", data, offset)[0] + offset += 10 # subscription id plus three unknown UInt16 fields + credit_tick = data[offset] + offset += 1 + sequence_number, offset = _read_vlq32(data, offset) + change_counter = data[offset] + timestamp: int | None = None + if change_counter: + offset += 1 + else: + timestamp = struct.unpack_from(">Q", data, offset)[0] + offset += 9 # timestamp plus additional change counter + # Skip the data-change value list. Alarm-only subscriptions terminate it with zero. + while offset < len(data) and data[offset] != 0: + raise ValueError("Mixed data/alarm notifications are not supported") + offset += 1 + alarms: list[Alarm] = [] + if offset < len(data) and data[offset] != 0: + alarm_subscription_id = struct.unpack_from(">I", data, offset)[0] + offset += 6 + if data[offset] != 0x81: + raise ValueError(f"Unsupported alarm notification return value: {data[offset]:#x}") + offset += 1 + objects, _ = _decode_objects(data, offset) + wanted = set(language_ids) if language_ids is not None else None + alarms = [_alarm_from_object(obj, wanted) for obj in objects if obj.class_id == Ids.ALARM_DAI_CLASS_RID] + if subscription_id == 0: + subscription_id = alarm_subscription_id + return AlarmNotification(subscription_id, credit_tick, sequence_number, change_counter, timestamp, tuple(alarms)) diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index f632365b..5e3f8bae 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -47,6 +47,14 @@ _parse_explore_datablocks, _build_subscription_request, ) +from .alarm import ( + Alarm, + AlarmNotification, + build_alarm_explore_request, + build_alarm_subscription_request, + parse_alarm_explore_response, + parse_alarm_notification, +) from . import typeinfo from .protocol import Ids @@ -597,6 +605,42 @@ async def delete_subscription(self, subscription_id: int) -> None: await self._send_request(FunctionCode.DELETE_OBJECT, payload) logger.info(f"Subscription {subscription_id:#x} deleted") + async def create_alarm_subscription( + self, + language_ids: Optional[list[int]] = None, + domains: Optional[list[int]] = None, + credit_limit: int = -1, + ) -> int: + """Subscribe to PLC alarm events and return the subscription ID.""" + payload = build_alarm_subscription_request(self._session_id, language_ids, domains, credit_limit) + response = await self._send_request(FunctionCode.CREATE_OBJECT, payload) + object_ids, _, return_value = parse_create_object_session_id(response) + if return_value != 0 or not object_ids: + raise RuntimeError(f"Alarm subscription failed: PLC returned {return_value:#x}") + return object_ids[0] + + async def delete_alarm_subscription(self, subscription_id: int) -> None: + """Delete an alarm subscription created by this client.""" + await self.delete_subscription(subscription_id) + + async def receive_alarm_notification( + self, language_ids: Optional[list[int]] = None, timeout: Optional[float] = None + ) -> AlarmNotification: + """Wait for one alarm notification, optionally with a timeout in seconds.""" + async with self._lock: + if not self._connected: + raise RuntimeError("Not connected") + receive = self._recv_cotp_dt() + frame = await asyncio.wait_for(receive, timeout) if timeout is not None else await receive + return parse_alarm_notification(frame, language_ids) + + async def browse_alarms(self, language_ids: Optional[list[int]] = None) -> list[Alarm]: + """Return the PLC's current active alarm state.""" + response = await self._send_request( + FunctionCode.EXPLORE, build_alarm_explore_request(), integrity_tail=5, reassemble=True + ) + return parse_alarm_explore_response(response, language_ids) + async def read_symbolic(self, access_area: int, lids: list[int], symbol_crc: int = 0) -> bytes: """Read a variable using S7CommPlus symbolic (LID-based) access. diff --git a/s7commplus/client.py b/s7commplus/client.py index 7bef905a..c7667624 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -8,6 +8,14 @@ from typing import Any, Optional from . import typeinfo +from .alarm import ( + Alarm, + AlarmNotification, + build_alarm_explore_request, + build_alarm_subscription_request, + parse_alarm_explore_response, + parse_alarm_notification, +) from .blob_decompressor import find_and_decompress from .connection import S7CommPlusConnection from .protocol import FunctionCode, Ids, ElementID, DataType, ObjectId @@ -17,6 +25,7 @@ encode_object_qualifier, encode_pvalue_blob, decode_pvalue_to_bytes, + parse_create_object_session_id, ) logger = logging.getLogger(__name__) @@ -533,6 +542,56 @@ def delete_subscription(self, subscription_id: int) -> None: self._connection.send_request(FunctionCode.DELETE_OBJECT, payload) logger.info(f"Subscription {subscription_id:#x} deleted") + def create_alarm_subscription( + self, + language_ids: Optional[list[int]] = None, + domains: Optional[list[int]] = None, + credit_limit: int = -1, + ) -> int: + """Subscribe to PLC alarm events. + + Args: + language_ids: Windows LCIDs for texts included with notifications. + ``None`` requests every configured language. + domains: Alarm-domain IDs to include. ``None`` subscribes to all. + credit_limit: Notification credit limit; ``-1`` means unlimited. + + Returns: + Subscription object ID assigned by the PLC. + """ + if self._connection is None: + raise RuntimeError("Not connected") + payload = build_alarm_subscription_request(self._connection.session_id, language_ids, domains, credit_limit) + response = self._connection.send_request(FunctionCode.CREATE_OBJECT, payload) + object_ids, _, return_value = parse_create_object_session_id(response) + if return_value != 0 or not object_ids: + raise RuntimeError(f"Alarm subscription failed: PLC returned {return_value:#x}") + return object_ids[0] + + def delete_alarm_subscription(self, subscription_id: int) -> None: + """Delete an alarm subscription created by this client.""" + self.delete_subscription(subscription_id) + + def receive_alarm_notification(self, language_ids: Optional[list[int]] = None) -> AlarmNotification: + """Block until the PLC sends one alarm notification.""" + if self._connection is None: + raise RuntimeError("Not connected") + return parse_alarm_notification(self._connection.receive_notification(), language_ids) + + def browse_alarms(self, language_ids: Optional[list[int]] = None) -> list[Alarm]: + """Return the PLC's current active alarm state. + + Args: + language_ids: Optional Windows LCIDs used to filter returned texts. + Omitting the filter retains every language sent by the PLC. + """ + if self._connection is None: + raise RuntimeError("Not connected") + response = self._connection.send_request( + FunctionCode.EXPLORE, build_alarm_explore_request(), integrity_tail=5, reassemble=True + ) + return parse_alarm_explore_response(response, language_ids) + def __enter__(self) -> "S7CommPlusClient": return self diff --git a/s7commplus/connection.py b/s7commplus/connection.py index c33f5a08..d4ed1a9d 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -692,6 +692,26 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: return resp_payload + def receive_notification(self) -> bytes: + """Receive one unsolicited S7CommPlus notification frame. + + The call blocks up to the connection's configured socket timeout. It must + not run concurrently with :meth:`send_request`, because both consume the + same protocol stream. + """ + if not self._connected: + from snap7.error import S7ConnectionError + + raise S7ConnectionError("Not connected") + frame = self._recv_s7_data() + _, data_length, consumed = decode_header(frame) + data = frame[consumed : consumed + data_length] + if not data or data[0] != Opcode.NOTIFICATION: + from snap7.error import S7ConnectionError + + raise S7ConnectionError("Expected an S7CommPlus notification") + return frame + # Sanity caps for fragment reassembly — generous vs. any real PLC EXPLORE response, # but bounded so a malformed/adversarial stream can't drive unbounded allocation. _MAX_REASSEMBLED_BYTES = 16 * 1024 * 1024 diff --git a/s7commplus/protocol.py b/s7commplus/protocol.py index fb9aa3cb..d23d3579 100644 --- a/s7commplus/protocol.py +++ b/s7commplus/protocol.py @@ -195,11 +195,35 @@ class Ids(IntEnum): SUBSCRIPTION_CREDIT_LIMIT = 1053 SUBSCRIPTION_REFERENCE_LIST = 1048 SUBSCRIPTION_FUNCTION_CLASS_ID = 1082 + SUBSCRIPTION_MISSED_SENDINGS = 1002 + SUBSCRIPTION_SUBSYSTEM_ERROR = 1003 + SUBSCRIPTION_REFERENCE_TRIGGER_MODE = 1005 + SUBSCRIPTION_ROUTE_MODE = 1040 + SUBSCRIPTION_DELAY_TIME = 1050 + SUBSCRIPTION_DISABLED = 1051 + SUBSCRIPTION_COUNT = 1052 + SUBSCRIPTION_TICKS = 1054 # Alarm subscription ALARM_SUBSCRIPTION_REF_CLASS_RID = 2662 ALARM_SUBSCRIPTION_REF_ALARM_DOMAIN = 2659 ALARM_SUBSCRIPTION_REF_ITS_ALARM_SUBSYSTEM = 2660 + ALARM_SUBSCRIPTION_REF_ALARM_DOMAIN_FILTER = 7731 + ALARM_SUBSCRIPTION_REF_SEND_TEXTS = 8173 + ALARM_SUBSCRIPTION_REF_TEXT_LANGUAGES = 8181 + + # Alarm objects and text libraries + ALARM_SUBSYSTEM_UPDATE_RELEVANT_DAI = 2667 + ALARM_DAI_CPU_ALARM_ID = 2670 + ALARM_DAI_ALL_STATES_INFO = 2671 + ALARM_DAI_DOMAIN = 2672 + ALARM_DAI_COMING = 2673 + ALARM_DAI_GOING = 2677 + ALARM_DAI_CLASS_RID = 2681 + ALARM_DAI_TEXTS = 2715 + ALARM_DAI_MESSAGE_TYPE = 4079 + ALARM_DAI_HMI_INFO = 7813 + ALARM_DAI_SEQUENCE_COUNTER = 7917 # DB AccessArea base (add DB number to get area ID) DB_ACCESS_AREA_BASE = 0x8A0E0000 diff --git a/tests/test_s7_alarm.py b/tests/test_s7_alarm.py new file mode 100644 index 00000000..c6e29838 --- /dev/null +++ b/tests/test_s7_alarm.py @@ -0,0 +1,167 @@ +"""Tests for S7CommPlus alarm subscriptions, browsing, and notifications.""" + +import struct +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from s7commplus import Alarm, AlarmNotification, AlarmText +from s7commplus.alarm import ( + build_alarm_explore_request, + build_alarm_subscription_request, + parse_alarm_explore_response, + parse_alarm_notification, +) +from s7commplus.async_client import S7CommPlusAsyncClient +from s7commplus.client import S7CommPlusClient +from s7commplus.protocol import DataType, ElementID, FunctionCode, Ids, Opcode, ProtocolVersion +from s7commplus.vlq import encode_uint32_vlq, encode_uint64_vlq + + +def _attribute(attribute_id: int, datatype: int, value: bytes, flags: int = 0) -> bytes: + return bytes([ElementID.ATTRIBUTE]) + encode_uint32_vlq(attribute_id) + bytes([flags, datatype]) + value + + +def _alarm_object() -> bytes: + result = bytearray([ElementID.START_OF_OBJECT]) + result += struct.pack(">I", 0x8A7E0001) + result += encode_uint32_vlq(Ids.ALARM_DAI_CLASS_RID) + result += encode_uint32_vlq(0) + encode_uint32_vlq(0) + result += _attribute(Ids.ALARM_DAI_CPU_ALARM_ID, DataType.LWORD, struct.pack(">Q", 0x8A7E0001002A0000)) + result += _attribute(Ids.ALARM_DAI_ALL_STATES_INFO, DataType.USINT, b"\x03") + result += _attribute(Ids.ALARM_DAI_DOMAIN, DataType.UINT, struct.pack(">H", 256)) + result += _attribute(Ids.ALARM_DAI_MESSAGE_TYPE, DataType.DINT, encode_uint32_vlq(1)) + result += _attribute(Ids.ALARM_DAI_SEQUENCE_COUNTER, DataType.UDINT, encode_uint32_vlq(17)) + result += _attribute(Ids.OBJECT_VARIABLE_TYPE_NAME, DataType.WSTRING, encode_uint32_vlq(6) + b"Motor1") + result += _attribute(Ids.ALARM_DAI_HMI_INFO, DataType.BLOB, b"\x00\x03hmi") + + coming = bytearray(struct.pack(">I", Ids.ALARM_DAI_COMING)) + coming += encode_uint32_vlq(3475) + bytes([0, DataType.TIMESTAMP]) + struct.pack(">Q", 123456789) + coming += b"\x00" + result += _attribute(Ids.ALARM_DAI_COMING, DataType.STRUCT, bytes(coming)) + + texts = bytearray() + for language_id, text_id, text in ((1031, 1, "Info"), (1031, 2, "Alarm"), (1033, 2, "Alert")): + raw = text.encode() + texts += encode_uint32_vlq((language_id << 16) | text_id) + texts += encode_uint32_vlq(0) + encode_uint32_vlq(len(raw)) + raw + texts += b"\x00" + result += _attribute(Ids.ALARM_DAI_TEXTS, DataType.BLOB, bytes(texts), flags=0x40) + result += bytes([ElementID.TERMINATING_OBJECT]) + return bytes(result) + + +def _explore_response() -> bytes: + return encode_uint64_vlq(0) + struct.pack(">I", Ids.NATIVE_THE_ALARM_SUBSYSTEM_RID) + encode_uint32_vlq(4) + _alarm_object() + + +def _notification_frame() -> bytes: + body = bytearray([Opcode.NOTIFICATION]) + body += struct.pack(">IHHH", 0x11223344, 0, 0, 0) + body += b"\x05" + encode_uint32_vlq(12) + b"\x01" + body += b"\x00" # end of data-change values + body += struct.pack(">IH", 0x11223344, 0) + b"\x81" + _alarm_object() + return struct.pack(">BBH", 0x72, ProtocolVersion.V2, len(body)) + body + struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + + +def test_alarm_models_are_public() -> None: + assert Alarm.__module__ == "s7commplus.alarm" + assert AlarmNotification.__module__ == "s7commplus.alarm" + assert AlarmText.__module__ == "s7commplus.alarm" + + +def test_build_alarm_subscription_request_contains_filters() -> None: + payload = build_alarm_subscription_request(0x12345678, [1031, 1033], [256, 257]) + assert payload.startswith(struct.pack(">I", 0x12345678)) + assert encode_uint32_vlq(Ids.ALARM_SUBSCRIPTION_REF_CLASS_RID) in payload + assert encode_uint32_vlq(1031) in payload + assert struct.pack(">H", 257) in payload + assert payload.endswith(bytes([ElementID.TERMINATING_OBJECT, ElementID.TERMINATING_OBJECT]) + b"\x00\x00\x00\x00") + + +@pytest.mark.parametrize("credit_limit", [-2, 256]) +def test_build_alarm_subscription_rejects_invalid_credit(credit_limit: int) -> None: + with pytest.raises(ValueError, match="credit_limit"): + build_alarm_subscription_request(1, credit_limit=credit_limit) + + +def test_build_alarm_explore_request_targets_alarm_subsystem() -> None: + payload = build_alarm_explore_request() + assert payload.startswith(struct.pack(">I", Ids.NATIVE_THE_ALARM_SUBSYSTEM_RID)) + assert encode_uint32_vlq(Ids.ALARM_SUBSYSTEM_UPDATE_RELEVANT_DAI) in payload + assert encode_uint32_vlq(Ids.ALARM_DAI_TEXTS) in payload + + +def test_parse_alarm_explore_response_with_language_filter() -> None: + alarms = parse_alarm_explore_response(_explore_response(), [1031]) + assert len(alarms) == 1 + alarm = alarms[0] + assert alarm.cpu_alarm_id == 0x8A7E0001002A0000 + assert alarm.name == "Motor1" + assert alarm.state == "coming" + assert alarm.timestamp == 123456789 + assert alarm.hmi_info == b"hmi" + assert alarm.texts == {1031: AlarmText(1031, "Info", "Alarm", ("",) * 9)} + + +def test_parse_alarm_notification() -> None: + notification = parse_alarm_notification(_notification_frame()) + assert notification.subscription_id == 0x11223344 + assert notification.credit_tick == 5 + assert notification.sequence_number == 12 + assert len(notification.alarms) == 1 + assert notification.alarms[0].texts[1033].alarm_text == "Alert" + + +def test_sync_alarm_client_apis() -> None: + client = S7CommPlusClient() + connection = MagicMock() + connection.session_id = 0x1234 + connection.send_request.side_effect = [ + encode_uint64_vlq(0) + b"\x01" + encode_uint32_vlq(0x55667788), + _explore_response(), + b"\x00", + ] + connection.receive_notification.return_value = _notification_frame() + client._connection = connection + + assert client.create_alarm_subscription([1031]) == 0x55667788 + assert client.browse_alarms([1031])[0].texts[1031].alarm_text == "Alarm" + assert client.receive_alarm_notification().alarms[0].cpu_alarm_id == 0x8A7E0001002A0000 + client.delete_alarm_subscription(0x55667788) + assert connection.send_request.call_args_list[-1].args[0] == FunctionCode.DELETE_OBJECT + + +@pytest.mark.asyncio +async def test_async_alarm_client_apis() -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._session_id = 0x1234 + client._send_request = AsyncMock( + side_effect=[ + encode_uint64_vlq(0) + b"\x01" + encode_uint32_vlq(0x55667788), + _explore_response(), + b"\x00", + ] + ) + client._recv_cotp_dt = AsyncMock(return_value=_notification_frame()) + + assert await client.create_alarm_subscription([1031]) == 0x55667788 + assert (await client.browse_alarms([1031]))[0].name == "Motor1" + assert (await client.receive_alarm_notification(timeout=1)).credit_tick == 5 + await client.delete_alarm_subscription(0x55667788) + + +@pytest.mark.parametrize( + "method,args", + [ + ("create_alarm_subscription", ()), + ("browse_alarms", ()), + ("receive_alarm_notification", ()), + ("delete_alarm_subscription", (1,)), + ], +) +def test_sync_alarm_methods_require_connection(method: str, args: tuple[object, ...]) -> None: + client = S7CommPlusClient() + with pytest.raises(RuntimeError, match="Not connected"): + getattr(client, method)(*args)