diff --git a/CHANGES.md b/CHANGES.md index 13e5a2a5..fb4aaf69 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) * S7CommPlus symbolic data subscriptions and notification decoding (experimental) * TIA Portal XML import for SymbolTable (`SymbolTable.from_tia_xml()`) (experimental) * S7CommPlus CPU state reading and block transfer (upload/download) diff --git a/s7commplus/__init__.py b/s7commplus/__init__.py index 0289ef38..709270b4 100644 --- a/s7commplus/__init__.py +++ b/s7commplus/__init__.py @@ -14,6 +14,7 @@ """ from .async_client import S7CommPlusAsyncClient as AsyncClient +from .alarm import Alarm, AlarmNotification, AlarmText, LanguageId from .blob_decompressor import decompress_blob, find_and_decompress from .client import S7CommPlusClient as Client from .connection import S7CommPlusConnection @@ -32,11 +33,15 @@ ) __all__ = [ + "Alarm", + "AlarmNotification", + "AlarmText", "AsyncClient", "CPUState", "Client", "DataBlock", "ExploreDataBlock", + "LanguageId", "Member", "S7CommPlusConnection", "Server", diff --git a/s7commplus/alarm.py b/s7commplus/alarm.py new file mode 100644 index 00000000..02342439 --- /dev/null +++ b/s7commplus/alarm.py @@ -0,0 +1,543 @@ +"""Alarm models and wire decoders for S7CommPlus notifications.""" + +from __future__ import annotations + +import re +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 +_ASSOCIATED_VALUE_TOKEN = re.compile(r"@(\d+)(%[^@]+)@") + +_SOFTDATATYPE_BOOL = 0x02000001 +_SOFTDATATYPE_BYTE = 0x02000002 +_SOFTDATATYPE_CHAR = 0x02000003 +_SOFTDATATYPE_WORD = 0x02000004 +_SOFTDATATYPE_INT = 0x02000005 +_SOFTDATATYPE_DWORD = 0x02000006 +_SOFTDATATYPE_DINT = 0x02000007 +_SOFTDATATYPE_REAL = 0x02000008 +_SOFTDATATYPE_STRING = 0x02000013 +_SOFTDATATYPE_LREAL = 0x02000030 +_SOFTDATATYPE_USINT = 0x02000034 +_SOFTDATATYPE_UINT = 0x02000035 +_SOFTDATATYPE_UDINT = 0x02000036 +_SOFTDATATYPE_SINT = 0x02000037 +_SOFTDATATYPE_WCHAR = 0x0200003D +_SOFTDATATYPE_WSTRING = 0x0200003E +_STRING_TYPE_MIN = 0x020A0000 +_STRING_TYPE_MAX = 0x020AFFFF +_WSTRING_TYPE_MIN = 0x020B0000 +_WSTRING_TYPE_MAX = 0x020BFFFF + + +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 _decode_associated_value(blob: _Blob) -> object | None: + """Decode an alarm SD value using its Softdatatype root ID.""" + root_id, value = blob.root_id, blob.value + try: + if root_id == _SOFTDATATYPE_STRING or _STRING_TYPE_MIN <= root_id <= _STRING_TYPE_MAX: + size = value[1] + return value[2 : 2 + size].decode("latin-1", errors="replace") + if root_id == _SOFTDATATYPE_WSTRING or _WSTRING_TYPE_MIN <= root_id <= _WSTRING_TYPE_MAX: + size = struct.unpack_from(">H", value, 2)[0] + return value[4 : 4 + 2 * size].decode("utf-16-be", errors="replace") + if root_id == _SOFTDATATYPE_BOOL: + return bool(value[0]) + if root_id in (_SOFTDATATYPE_BYTE, _SOFTDATATYPE_USINT): + return value[0] + if root_id == _SOFTDATATYPE_CHAR: + return value[:1].decode("latin-1", errors="replace") + if root_id == _SOFTDATATYPE_SINT: + return struct.unpack(">b", value)[0] + if root_id in (_SOFTDATATYPE_WORD, _SOFTDATATYPE_UINT): + return struct.unpack(">H", value)[0] + if root_id == _SOFTDATATYPE_INT: + return struct.unpack(">h", value)[0] + if root_id in (_SOFTDATATYPE_DWORD, _SOFTDATATYPE_UDINT): + return struct.unpack(">I", value)[0] + if root_id == _SOFTDATATYPE_DINT: + return struct.unpack(">i", value)[0] + if root_id == _SOFTDATATYPE_REAL: + return struct.unpack(">f", value)[0] + if root_id == _SOFTDATATYPE_LREAL: + return struct.unpack(">d", value)[0] + if root_id == _SOFTDATATYPE_WCHAR: + return value[:2].decode("utf-16-be", errors="replace") + except (IndexError, struct.error): + return None + return None + + +def _interpolate_associated_values(text: str, values: Sequence[_Blob]) -> str: + def replace(match: re.Match[str]) -> str: + index = int(match.group(1)) + if index >= len(values): + return match.group(0) + value = _decode_associated_value(values[index]) + if value is None: + return match.group(0) + try: + return match.group(2) % value + except (TypeError, ValueError): + return match.group(0) + + return _ASSOCIATED_VALUE_TOKEN.sub(replace, text) + + +def _alarm_texts( + value: Any, language_ids: set[LanguageId | int] | None, associated_values: Sequence[_Blob] +) -> 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 + text = blob.value.decode("utf-8", errors="replace") + grouped.setdefault(language_id, {})[text_id] = _interpolate_associated_values(text, associated_values) + 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_blobs: tuple[_Blob, ...] = () + 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_blobs = tuple(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=tuple(blob.value for blob in associated_blobs), + texts=_alarm_texts(attrs.get(Ids.ALARM_DAI_TEXTS), language_ids, associated_blobs), + ) + + +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 c5c992e7..83b8c55b 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -44,6 +44,16 @@ _parse_protection_level_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 .legitimation import ( build_legacy_response, build_new_response, @@ -86,6 +96,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 @@ -122,6 +133,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.""" @@ -442,6 +458,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 @@ -625,6 +642,52 @@ 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. + + Do not run this alongside a data-subscription receive loop on the same + connection: mixed notification dispatch is not supported yet. + """ + 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 read_alarms(self, language_ids: Optional[list[LanguageId | int]] = None) -> list[Alarm]: + """Return a snapshot of the PLC's active alarms without consuming notifications.""" + 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. @@ -978,8 +1041,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 5e2f611c..531d33d9 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -11,6 +11,16 @@ 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, @@ -703,6 +713,80 @@ 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. + + Do not run this alongside a data-subscription receive loop on the same + connection: mixed notification dispatch is not supported yet. + """ + if self._connection is None: + raise RuntimeError("Not connected") + return parse_alarm_notification(self._connection.receive_notification(), language_ids) + + def read_alarms(self, language_ids: Optional[list[LanguageId | int]] = None) -> list[Alarm]: + """Return the PLC's current active alarm state. + + This is a snapshot read and does not create or consume a subscription, + so it can be used before or while an alarm subscription exists. + + 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 8757178b..3f8e0cdb 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -1170,7 +1170,7 @@ 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 diff --git a/s7commplus/protocol.py b/s7commplus/protocol.py index 9d49472b..9a18c360 100644 --- a/s7commplus/protocol.py +++ b/s7commplus/protocol.py @@ -203,6 +203,8 @@ class Ids(IntEnum): SUBSCRIPTION_CREDIT_LIMIT = 1053 SUBSCRIPTION_REFERENCE_LIST = 1048 SUBSCRIPTION_FUNCTION_CLASS_ID = 1082 + SUBSCRIPTION_REFERENCE_TRIGGER_MODE = 1005 + SUBSCRIPTION_DELAY_TIME = 1050 SUBSCRIPTION_DISABLED = 1051 SUBSCRIPTION_COUNT = 1052 SUBSCRIPTION_TICKS = 1054 @@ -211,6 +213,22 @@ class Ids(IntEnum): 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 # Session's effective protection level, readable via GetVarSubStreamed EFFECTIVE_PROTECTION_LEVEL = 1842 diff --git a/tests/test_s7_alarm.py b/tests/test_s7_alarm.py new file mode 100644 index 00000000..d5b2f157 --- /dev/null +++ b/tests/test_s7_alarm.py @@ -0,0 +1,221 @@ +"""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 _blob(root_id: int, value: bytes) -> bytes: + metadata = bytes(8) + b"\x03" if root_id > 1 else b"" + return encode_uint32_vlq(root_id) + metadata + encode_uint32_vlq(len(value)) + 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) + associated_values = [ + _blob(1, b"type-info"), + _blob(0x02000005, struct.pack(">h", 4)), + _blob(1, b""), + _blob(1, b""), + _blob(1, b""), + _blob(0x020A00FE, b"\xfe\x09=F6+S2-G1"), + ] + coming += encode_uint32_vlq(3476) + bytes([0x10, DataType.BLOB]) + encode_uint32_vlq(len(associated_values)) + coming += b"".join(associated_values) + 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 @1%d@ @5%s@"), + (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.associated_values[1] == b"\x00\x04" + assert alarm.associated_values[5] == b"\xfe\x09=F6+S2-G1" + assert alarm.texts == {1031: AlarmText(1031, "Info", "Alarm 4 =F6+S2-G1", ("",) * 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.read_alarms([1031])[0].texts[1031].alarm_text == "Alarm 4 =F6+S2-G1" + 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.read_alarms([1031]))[0].texts[1031].alarm_text == "Alarm 4 =F6+S2-G1" + 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", ()), + ("read_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)