Subscribing to OPC UA Alarms and Conditions

The engineering task on this page is to attach a passive listener to a treatment plant’s OPC UA server so that every alarm it raises — an analyzer fault, a low free-chlorine annunciation, a turbidity high-high trip — arrives as a structured event with its severity, its acknowledgement state, and an authoritative time, and is then mapped onto the compliance-relevant signals that downstream rules act on. This is the event-model counterpart within the parent OPC UA Data Extraction section: where Extracting OPC UA Nodes for Chlorine Residuals subscribes to the numeric process variable and asks “what is the value”, this page subscribes to the Alarms & Conditions model and asks “what did the plant declare abnormal, and has anyone acknowledged it”. Those are different address-space mechanisms, and conflating them is how a utility ends up with a residual number in the historian but no record that the analyzer was flagged faulty when it produced that number. The listener never writes process data; the only calls it makes are the standard Acknowledge and Confirm methods that the condition model itself exposes.

Prerequisites & Environment Setup

The implementation targets Python 3.10+ and the asynchronous asyncua client. Alarms & Conditions is an optional server facet, so before writing code confirm the server actually advertises it: an Ignition gateway with the Alarm Notification module, Siemens PCS 7, AVEVA System Platform, and most Kepware-fronted PLCs expose AlarmConditionType events, but a bare data-only server will accept an event subscription and simply never deliver a notification. Pin the library version, because the event-filter construction helpers on the 1.x line differ from earlier releases.

Two things do not come from a package. First, the read-only egress path to the OT network described in Security Boundary Design — the alarm listener is a consumer of the server’s Server object and needs no write permission whatsoever. Second, a server profile that tells you which event types the vendor emits and which optional condition fields (ConditionName, ActiveState, AckedState, Retain) are populated, because vendors vary in how completely they implement the model.

python3 -m venv .venv && source .venv/bin/activate
pip install "asyncua==1.1.5" "pydantic==2.7.*"
# Reuse the SignAndEncrypt client certificate from the residual extractor
openssl req -x509 -newkey rsa:2048 -keyout client_key.pem \
    -out client_cert.pem -days 365 -nodes -subj "/CN=alarm-listener"

Step-by-Step Implementation

The listener connects, subscribes for events on the server’s event notifier, receives each condition transition through a synchronous callback, translates it into an internal compliance signal, and — for acknowledgeable alarms — calls Acknowledge back on the condition. The five steps below build a single AlarmConditionListener class.

Step 1 — Subscribe for events on the Server object, not on a variable

Unlike a residual, an alarm is not read from a node — it is delivered from an event notifier. The root notifier every conformant server exposes is the Server object (ns=0;i=2253), which fans out every condition raised anywhere in the address space. A data-change subscription and an event subscription are distinct: the former monitors an attribute’s Value, the latter registers a MonitoredItem against the EventNotifier attribute. You create the subscription once, then call subscribe_events with the source notifier and the event type whose fields you want delivered.

Step 2 — Choose the event type and let the filter select its fields

Passing AlarmConditionType (rather than BaseEventType) tells asyncua to build a select clause covering the full alarm field set — ActiveState, AckedState, ConfirmedState, Retain, ConditionName, Severity, plus the base EventId, SourceNode, Time, and Message. This matters because a plain BaseEventType subscription strips the condition-specific fields and leaves you unable to tell an active alarm from a cleared one. Behind the select clause sits a where clause you can tighten later — for example, filtering server-side on a minimum Severity so trivial informational events never cross the OT boundary — but during commissioning it is safer to accept every alarm type and filter in Python, so that an unexpected but compliance-relevant condition is never dropped by an over-eager server-side predicate. The event you receive carries SourceNode, which links the alarm back to the analyzer whose value the sibling extractor is reading, so the two streams reconcile on a shared node identifier without any string matching on tag names.

Step 3 — Handle the alarm state machine

An AlarmConditionType instance is a small state machine. It is Inactive until the monitored condition asserts, then Active; an operator (or an automated agent) moves it to Acked; a Confirm call moves it to Confirmed; and when the underlying condition clears it returns to Inactive. Retain=true marks the alarm as still requiring attention, so the listener can distinguish a live standing alarm from a historical transition being replayed. Modelling these transitions explicitly is what keeps a flapping analyzer from being logged as dozens of independent violations.

Step 4 — Acknowledge acknowledgeable conditions

An AcknowledgeableConditionType (the supertype of AlarmConditionType) exposes an Acknowledge method that takes the triggering EventId and a comment. Calling it is the one action the listener performs on the server, and it is a method call on the condition instance — never a write to a process variable, so it stays within the read-only posture the OT boundary demands. The EventId is a per-transition opaque byte string, so you must acknowledge the exact event you received, not a stale one; acknowledging with the wrong EventId returns BadEventIdUnknown. Only acknowledge alarms your policy designates for automatic clearing — a self-resetting nuisance alarm below the operational band, or an instrument-fault flag whose remediation is already ticketed elsewhere. Safety- and compliance-critical alarms such as a low free-chlorine annunciation should remain for a human operator, because auto-acknowledging one would erase the very evidence a primacy-agency audit expects to find, and the listener’s job there is to record the transition and its dwell time, not to silence it.

Step 5 — Map the event onto a compliance signal and emit the envelope

The final step translates the vendor’s alarm vocabulary into the internal signals the rule engine understands. OPC UA Severity is a 1–1000 integer; normalise it to a unit scale with

so that a downstream weighting is vendor-independent. To quantify how long a compliance-relevant alarm stood before anyone acted, accumulate the active dwell time across the day as $T_{active} = \sum_{k}\left(t^{ack}_{k} - t^{active}_{k}\right)$, which feeds annunciation-response reporting. The class below implements all five steps end to end.

import asyncio
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Dict, Optional

from asyncua import Client, ua

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S%z",
)
logger = logging.getLogger("alarm_listener")

# Map a SourceName (or SourceNode browse name) onto an internal compliance
# signal. Alarms not in this map are logged but not promoted to a signal.
COMPLIANCE_SIGNALS = {
    "FreeCl2.Analyzer": "ANALYZER_FAULT",
    "FreeCl2.Low": "LOW_CHLORINE_ALARM",
    "Turbidity.HiHi": "TURBIDITY_HIGH_HIGH",
}

# SourceNames whose alarms this service is permitted to auto-acknowledge.
AUTO_ACK_ALLOWLIST = {"FreeCl2.Analyzer"}


@dataclass
class AlarmSignal:
    signal: str
    source_name: str
    severity: int
    normalized_severity: float
    active: bool
    acked: bool
    retain: bool
    event_time: datetime
    message: str


class AlarmConditionListener:
    def __init__(self, endpoint: str, cert: Optional[str] = None, key: Optional[str] = None):
        self.client = Client(url=endpoint)
        self.subscription = None
        self.loop = asyncio.get_event_loop()
        self._active_since: Dict[str, datetime] = {}
        if cert and key:
            self.client.set_security_string(f"Basic256Sha256,SignAndEncrypt,{cert},{key}")
        else:
            logger.warning("No credentials supplied; unsecured connection is for testing only.")

    async def connect(self) -> None:
        await self.client.connect()
        logger.info("Connected to OPC UA endpoint.")

    async def start(self) -> None:
        # Step 1 + 2: one subscription, events sourced from the Server object,
        # typed as AlarmConditionType so the full condition field set is selected.
        self.subscription = await self.client.create_subscription(500, self)
        server_obj = self.client.get_node(ua.ObjectIds.Server)
        alarm_type = self.client.get_node(ua.ObjectIds.AlarmConditionType)
        await self.subscription.subscribe_events(server_obj, alarm_type)
        logger.info("Subscribed to AlarmConditionType events on the Server object.")

    def event_notification(self, event) -> None:
        # asyncua invokes this synchronously; never await here. Fields are read
        # off the event object, which carries the AlarmConditionType select set.
        source_name = getattr(event, "SourceName", "") or ""
        severity = int(getattr(event, "Severity", 0) or 0)
        active = bool(getattr(getattr(event, "ActiveState", None), "Value", False))
        acked = bool(getattr(getattr(event, "AckedState", None), "Value", False))
        retain = bool(getattr(event, "Retain", False))
        event_time = getattr(event, "Time", None) or datetime.now(timezone.utc)
        message = str(getattr(getattr(event, "Message", None), "Text", "") or "")

        signal_name = COMPLIANCE_SIGNALS.get(source_name)
        if signal_name is None:
            logger.debug("Unmapped alarm from %s; not promoted.", source_name)
            return

        # Step 3: track dwell time across the Active -> Acked transition.
        if active and source_name not in self._active_since:
            self._active_since[source_name] = event_time
        elif not active:
            self._active_since.pop(source_name, None)

        signal = AlarmSignal(
            signal=signal_name,
            source_name=source_name,
            severity=severity,
            normalized_severity=(severity - 1) / 999 if severity else 0.0,
            active=active,
            acked=acked,
            retain=retain,
            event_time=event_time,
            message=message,
        )
        logger.info(
            "COMPLIANCE_ALARM | %s | active=%s acked=%s sev=%d | %s | %s",
            signal.signal, signal.active, signal.acked, signal.severity,
            signal.event_time.isoformat(), signal.message,
        )

        # Step 4: schedule an acknowledge for allowlisted, unacknowledged alarms.
        if active and not acked and source_name in AUTO_ACK_ALLOWLIST:
            event_id = getattr(event, "EventId", None)
            condition_id = getattr(event, "ConditionId", None)
            if event_id and condition_id:
                self.loop.create_task(self._acknowledge(condition_id, event_id))

    async def _acknowledge(self, condition_id, event_id: bytes) -> None:
        condition = self.client.get_node(condition_id)
        try:
            # Standard AcknowledgeableConditionType::Acknowledge(EventId, Comment).
            await condition.call_method(
                ua.NodeId(ua.ObjectIds.AcknowledgeableConditionType_Acknowledge),
                ua.Variant(event_id, ua.VariantType.ByteString),
                ua.LocalizedText("Acknowledged by compliance listener"),
            )
            logger.info("Acknowledged condition %s.", condition_id)
        except ua.UaStatusCodeError as exc:
            logger.error("Acknowledge failed for %s: %s", condition_id, exc)

    async def disconnect(self) -> None:
        if self.subscription:
            await self.subscription.delete()
        await self.client.disconnect()
        logger.info("Disconnected from OPC UA server.")


async def main() -> None:
    listener = AlarmConditionListener("opc.tcp://scada-server:4840")
    try:
        await listener.connect()
        await listener.start()
        while True:
            await asyncio.sleep(3600)
    except Exception as exc:
        logger.critical("Alarm listener failed: %s", exc, exc_info=True)
    finally:
        await listener.disconnect()


if __name__ == "__main__":
    asyncio.run(main())

Promoted signals leave this service as typed events. A LOW_CHLORINE_ALARM becomes an input to the Violation Detection & Rule Engine Logic, which decides whether a genuine exceedance has occurred, while a standing ANALYZER_FAULT is exactly the corroborating evidence the Monitoring Gap Detection Algorithms need to classify an absent residual as an instrument outage rather than a false compliance breach. The alarm envelopes themselves are archived alongside the numeric readings through the Historian Integration Patterns for Compliance Pipelines, so the record shows not just what the residual was but what the plant declared about it.

Configuration Reference

The first table is the event field set the AlarmConditionType filter selects; treat the source-to-signal map as versioned configuration pulled from the plant’s alarm register, never as inline constants.

AlarmConditionType event fields

Subscription and routing parameters

Verification & Testing

Confirm the mapping and dwell logic deterministically, without a live server, by driving synthetic events through the same field-extraction path so a regression in signal promotion or acknowledgement gating is caught before it reaches production.

from types import SimpleNamespace


def promote(source_name, active, acked, severity, signals, allowlist):
    """Mirror of event_notification's promotion and auto-ack decision."""
    signal = signals.get(source_name)
    if signal is None:
        return None
    normalized = (severity - 1) / 999 if severity else 0.0
    auto_ack = active and not acked and source_name in allowlist
    return SimpleNamespace(signal=signal, normalized=normalized, auto_ack=auto_ack)


SIGNALS = {"FreeCl2.Low": "LOW_CHLORINE_ALARM", "FreeCl2.Analyzer": "ANALYZER_FAULT"}
ALLOWLIST = {"FreeCl2.Analyzer"}


def test_low_chlorine_alarm_is_promoted_but_not_auto_acked():
    result = promote("FreeCl2.Low", True, False, 800, SIGNALS, ALLOWLIST)
    assert result.signal == "LOW_CHLORINE_ALARM"
    assert result.auto_ack is False  # safety-relevant: humans acknowledge


def test_analyzer_fault_on_allowlist_is_auto_acked():
    result = promote("FreeCl2.Analyzer", True, False, 500, SIGNALS, ALLOWLIST)
    assert result.auto_ack is True


def test_severity_normalizes_to_unit_scale():
    assert promote("FreeCl2.Low", True, False, 1000, SIGNALS, ALLOWLIST).normalized == 1.0
    assert promote("FreeCl2.Low", True, False, 1, SIGNALS, ALLOWLIST).normalized == 0.0


def test_unmapped_source_is_dropped():
    assert promote("Pump3.Trip", True, False, 400, SIGNALS, ALLOWLIST) is None

Acceptance criteria before promoting the listener to production:

• The subscription is created against the Server object with AlarmConditionType, and a test alarm on the server delivers a notification carrying ActiveState and AckedState.
• SourceNode/SourceName on each event resolves to the same node the residual extractor reads, so alarm and value reconcile.
• Only source names on the auto-acknowledge allowlist are acknowledged; low-chlorine and other safety-relevant alarms are recorded, never silenced.
• Acknowledge is called with the exact EventId from the received event and handles BadEventIdUnknown without crashing the loop.
• Severity is normalised to $[0,1]$ and every promoted signal carries a timezone-aware Time, never a naive timestamp.
• Every alarm transition, mapped or unmapped, emits a structured, auditable log line.

Troubleshooting & Gotchas

• The subscription succeeds but no alarms ever arrive. The server exposes data but not the Alarms & Conditions facet, or its Server object’s EventNotifier bit is not set for SubscribeToEvents. Browse the server’s capabilities and confirm at least one condition instance exists; a data-only server accepts subscribe_events and then stays silent forever.
• Every field on the event is None. You subscribed with BaseEventType instead of AlarmConditionType, so the filter’s select clause never requested the condition fields. Subscribe with the alarm type (or supply an explicit select list) so ActiveState, AckedState, and Retain are populated.
• Acknowledge returns BadEventIdUnknown. The EventId is per-transition and expires when the condition changes state; you acknowledged a stale event. Acknowledge from inside the handler using that notification’s own EventId, and treat the error as benign — it usually means the alarm already cleared.
• A flapping analyzer floods the historian with violations. Each Active/Inactive cycle is a separate event, and without state tracking each one looks like a fresh breach. Debounce on the condition state machine and let the Monitoring Gap Detection Algorithms collapse a burst of transitions into a single instrument-fault interval.
• Alarm times disagree with the residual timeline by minutes. The server is stamping events on receipt rather than at the source. Prefer the condition’s Time field over wall-clock receipt time, and reconcile it through the same UTC normalisation the numeric readings pass through before any reporting window is computed.

Related

• OPC UA Data Extraction — parent section and the end-to-end extraction contract
• Extracting OPC UA Nodes for Chlorine Residuals — the value-reading counterpart these alarms annotate
• Historian Integration Patterns for Compliance Pipelines — where alarm envelopes are archived beside numeric readings
• Monitoring Gap Detection Algorithms — uses analyzer-fault alarms to distinguish an outage from a breach
• Violation Detection & Rule Engine Logic — consumes promoted alarm signals for exceedance evaluation