diff --git a/CHANGES.md b/CHANGES.md index 4474ffe9..a630fe21 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -12,6 +12,7 @@ Major release: new `s7commplus` package with S7CommPlus protocol support. * S7CommPlus PLC start/stop via INVOKE * S7CommPlus object browsing via EXPLORE * S7CommPlus live symbol browsing (`client.browse()`) and datablock listing (experimental) +* S7CommPlus active-alarm browsing and alarm subscriptions (experimental) * TIA Portal XML import for SymbolTable (`SymbolTable.from_tia_xml()`) (experimental) * S7CommPlus CPU state reading and block transfer (upload/download) * **Symbolic (LID-based) access for optimized DBs** (experimental): diff --git a/s7commplus/__init__.py b/s7commplus/__init__.py index 7e5199d5..115dda0f 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, LanguageId from .tag_browser import ( Tag, Member, @@ -34,6 +35,10 @@ "DataBlock", "CPUState", "S7CommPlusConnection", + "Alarm", + "AlarmNotification", + "AlarmText", + "LanguageId", "decompress_blob", "find_and_decompress", "Tag", diff --git a/s7commplus/alarm.py b/s7commplus/alarm.py new file mode 100644 index 00000000..49becb45 --- /dev/null +++ b/s7commplus/alarm.py @@ -0,0 +1,464 @@ +"""Alarm models and wire decoders for S7CommPlus notifications.""" + +from __future__ import annotations + +import struct +from collections.abc import Sequence +from dataclasses import dataclass, field +from enum import IntEnum +from typing import Any + +from .codec import encode_object_qualifier +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 + + +class LanguageId(IntEnum): + """Windows locale identifiers (LCIDs) commonly supported by Siemens HMIs.""" + + CHINESE_TRADITIONAL = 1028 + CZECH = 1029 + DANISH = 1030 + GERMAN_GERMANY = 1031 + GREEK = 1032 + ENGLISH_UNITED_STATES = 1033 + SPANISH_TRADITIONAL = 1034 + FINNISH = 1035 + FRENCH_FRANCE = 1036 + HUNGARIAN = 1038 + ITALIAN_ITALY = 1040 + JAPANESE = 1041 + KOREAN = 1042 + DUTCH_NETHERLANDS = 1043 + POLISH = 1045 + PORTUGUESE_BRAZIL = 1046 + RUSSIAN = 1049 + SWEDISH = 1053 + TURKISH = 1055 + CHINESE_SIMPLIFIED = 2052 + DUTCH_BELGIUM = 2067 + PORTUGUESE_PORTUGAL = 2070 + + +@dataclass(frozen=True) +class AlarmText: + """The texts for one alarm in one PLC language.""" + + language_id: LanguageId | 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( + subscription_container_id: int, + language_ids: Sequence[LanguageId | int] | None = None, + domains: list[int] | None = None, + credit_limit: int = 10, +) -> 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", subscription_container_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"Subscription_{_ALARM_SUBSCRIPTION_RELATION_ID}"), + ) + 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("S7pDriver_Alarming")) + 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_delete_alarm_subscription_request(subscription_container_id: int, protocol_version: int) -> bytes: + """Build the DeleteObject payload used for an alarm subscription container.""" + return ( + struct.pack(">I", subscription_container_id) + + b"\x00" + + encode_object_qualifier(protocol_version=protocol_version) + + struct.pack(">I", 0) + ) + + +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[LanguageId | 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[LanguageId | 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: Sequence[LanguageId | 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: Sequence[LanguageId | 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 9ca7600c..28a45abc 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -43,6 +43,16 @@ _parse_get_var_substreamed_response, _set_s7_groups, ) +from .alarm import ( + Alarm, + AlarmNotification, + LanguageId, + build_alarm_explore_request, + build_alarm_subscription_request, + build_delete_alarm_subscription_request, + parse_alarm_explore_response, + parse_alarm_notification, +) from .protocol import ( READ_FUNCTION_CODES, S7COMMPLUS_LOCAL_TSAP, @@ -75,6 +85,7 @@ def __init__(self) -> None: self._reader: Optional[asyncio.StreamReader] = None self._writer: Optional[asyncio.StreamWriter] = None self._session_id: int = 0 + self._subscription_container_id: int = 0 self._sequence_number: int = 0 self._protocol_version: int = 0 self._connected = False @@ -109,6 +120,11 @@ def protocol_version(self) -> int: def session_id(self) -> int: return self._session_id + @property + def subscription_container_id(self) -> int: + """Object ID assigned to the session's subscription container.""" + return self._subscription_container_id + @property def session_setup_ok(self) -> bool: """Whether the S7CommPlus session setup succeeded for data operations.""" @@ -366,6 +382,7 @@ async def disconnect(self) -> None: self._connected = False self._session_id = 0 + self._subscription_container_id = 0 self._sequence_number = 0 self._protocol_version = 0 self._with_integrity_id = False @@ -548,6 +565,48 @@ 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[LanguageId | int]] = None, + domains: Optional[list[int]] = None, + credit_limit: int = 10, + ) -> int: + """Subscribe to PLC alarm events and return the subscription ID.""" + if self._subscription_container_id == 0: + raise RuntimeError("PLC did not provide a subscription container object") + payload = build_alarm_subscription_request(self._subscription_container_id, language_ids, domains, credit_limit) + response = await self._send_request(FunctionCode.CREATE_OBJECT, payload, integrity_tail=len(payload) - 11) + 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.""" + if self._subscription_container_id == 0: + raise RuntimeError("PLC did not provide a subscription container object") + payload = build_delete_alarm_subscription_request(self._subscription_container_id, self._protocol_version) + await self._send_request(FunctionCode.DELETE_OBJECT, payload) + logger.info(f"Alarm subscription {subscription_id:#x} deleted") + + async def receive_alarm_notification( + self, language_ids: Optional[list[LanguageId | 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[LanguageId | 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. @@ -901,8 +960,10 @@ async def _create_session(self) -> None: object_ids, obj_end, return_value = parse_create_object_session_id(body) if object_ids: self._session_id = object_ids[0] + self._subscription_container_id = object_ids[1] if len(object_ids) > 1 else 0 else: self._session_id = struct.unpack_from(">I", response, 9)[0] + self._subscription_container_id = 0 self._protocol_version = version if return_value != 0: diff --git a/s7commplus/client.py b/s7commplus/client.py index 106d6f60..e19fb43b 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -10,12 +10,23 @@ from snap7.error import S7ConnectionError from . import typeinfo +from .alarm import ( + Alarm, + AlarmNotification, + LanguageId, + build_alarm_explore_request, + build_alarm_subscription_request, + build_delete_alarm_subscription_request, + parse_alarm_explore_response, + parse_alarm_notification, +) from .blob_decompressor import find_and_decompress from .codec import ( decode_pvalue_to_bytes, encode_item_address, encode_object_qualifier, encode_pvalue_blob, + parse_create_object_session_id, ) from .connection import S7CommPlusConnection from .protocol import DataType, ElementID, FunctionCode, Ids, ObjectId, ProtocolVersion @@ -648,6 +659,73 @@ 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[LanguageId | int]] = None, + domains: Optional[list[int]] = None, + credit_limit: int = 10, + ) -> 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. The default of 10 matches + the working S7-1500 reference trace. + + Returns: + Subscription object ID assigned by the PLC. + """ + if self._connection is None: + raise RuntimeError("Not connected") + if self._connection.subscription_container_id == 0: + raise RuntimeError("PLC did not provide a subscription container object") + payload = build_alarm_subscription_request( + self._connection.subscription_container_id, language_ids, domains, credit_limit + ) + response = self._connection.send_request( + FunctionCode.CREATE_OBJECT, + payload, + integrity_tail=len(payload) - 11, + ) + 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.""" + if self._connection is None: + raise RuntimeError("Not connected") + if self._connection.subscription_container_id == 0: + raise RuntimeError("PLC did not provide a subscription container object") + payload = build_delete_alarm_subscription_request( + self._connection.subscription_container_id, self._connection.protocol_version + ) + self._connection.send_request(FunctionCode.DELETE_OBJECT, payload) + logger.info(f"Alarm subscription {subscription_id:#x} deleted") + + def receive_alarm_notification(self, language_ids: Optional[list[LanguageId | 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[LanguageId | 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 e1f6bd83..f44e7f50 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -263,6 +263,7 @@ def __init__( self._incoming_bio: Optional[ssl.MemoryBIO] = None self._outgoing_bio: Optional[ssl.MemoryBIO] = None self._session_id: int = 0 + self._subscription_container_id: int = 0 self._sequence_number: int = 0 self._protocol_version: int = 0 # Detected from PLC response self._tls_active: bool = False @@ -313,6 +314,11 @@ def session_id(self) -> int: """Session ID assigned by the PLC.""" return self._session_id + @property + def subscription_container_id(self) -> int: + """Object ID assigned to the session's subscription container.""" + return self._subscription_container_id + @property def tls_active(self) -> bool: """Whether TLS encryption is active on this connection.""" @@ -607,6 +613,7 @@ def disconnect(self) -> None: self._outgoing_bio = None self._oms_secret = None self._session_id = 0 + self._subscription_container_id = 0 self._sequence_number = 0 self._protocol_version = 0 self._server_session_version = None @@ -779,6 +786,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 @@ -1017,8 +1044,9 @@ def _wstring_attr(attr_id: int, s: str) -> bytes: raise S7ConnectionError("CreateObject response has no session ObjectId") - # First ObjectId is the new session id; second (if any) is for notifications. + # First ObjectId is the session; the second is its subscription container. self._session_id = object_ids[0] + self._subscription_container_id = object_ids[1] if len(object_ids) > 1 else 0 self._protocol_version = version logger.debug( diff --git a/s7commplus/protocol.py b/s7commplus/protocol.py index 3e13a627..68052530 100644 --- a/s7commplus/protocol.py +++ b/s7commplus/protocol.py @@ -200,11 +200,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..2fa539a3 --- /dev/null +++ b/tests/test_s7_alarm.py @@ -0,0 +1,200 @@ +"""Tests for S7CommPlus alarm subscriptions, browsing, and notifications.""" + +import struct +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from s7commplus import Alarm, AlarmNotification, AlarmText, LanguageId +from s7commplus.alarm import ( + build_alarm_explore_request, + build_alarm_subscription_request, + build_delete_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" + assert LanguageId.ENGLISH_UNITED_STATES == 1033 + + +def test_alarm_subscription_matches_real_plc_reference_trace() -> None: + payload = build_alarm_subscription_request(0x70000CB8) + captured = bytes.fromhex( + "70000cb80004000000000002a17fffc00187690000a38169001517" + "537562736372697074696f6e5f32313437343637323635a3883a000202" + "a3876a00030000a3876b000900a38810000202a38811000101a388182004" + "0388808480000000a38819000400a3881a000400a3881b000200a3881c000200" + "a3881d0007000aa3881e0003ffffa15101000194660000a38169001512" + "5337704472697665725f416c61726d696e67a3876d000203a3946310030a" + "0000000000000000000000000000000000000000a3bc33200301ffff" + "a3bf75200400a3bf6d000101a4946400000008a2a200000000" + ) + # The write IntegrityId 2 is inserted by _send_request at offset 11. + assert payload == captured[:11] + captured[12:] + + +def test_alarm_delete_matches_real_plc_reference_trace() -> None: + payload = build_delete_alarm_subscription_request(0x70000CB8, ProtocolVersion.V2) + wire_payload = payload[:-4] + b"\x03" + payload[-4:] + assert wire_payload == bytes.fromhex("70000cb800000004e88969001200000000896a001300896b000400000300000000") + + +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.subscription_container_id = 0x1235 + connection.protocol_version = ProtocolVersion.V2 + 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 + create_call = connection.send_request.call_args_list[0] + assert create_call.kwargs["integrity_tail"] == len(create_call.args[1]) - 11 + assert create_call.args[1].startswith(struct.pack(">I", 0x1235)) + 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) + delete_call = connection.send_request.call_args_list[-1] + assert delete_call.args[0] == FunctionCode.DELETE_OBJECT + assert delete_call.args[1].startswith(struct.pack(">I", 0x1235)) + + +@pytest.mark.asyncio +async def test_async_alarm_client_apis() -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._session_id = 0x1234 + client._subscription_container_id = 0x1235 + client._protocol_version = ProtocolVersion.V2 + 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)