Extracting OPC UA Nodes for Chlorine Residuals

The precise engineering task on this page is turning the free chlorine, total chlorine, and monochloramine nodes exposed by a treatment plant’s OPC UA server into audit-ready residual records — each carrying a validated value in mg/L, a quality tier, and an authoritative UTC sample time — without ever writing to the control network. This is the disinfection-metric reference implementation within the parent OPC UA Data Extraction section, written for the Python automation builders who own the ingestion service and the environmental compliance teams who sign the resulting reports. Chlorine residual is the operational backbone of disinfection compliance under the EPA’s Stage 2 Disinfectants and Disinfection Byproducts Rule and 40 CFR Part 141, so a missed or mis-quality-flagged reading is not a data nuisance — it is a monitoring gap in the legal record. Reliable extraction demands deterministic namespace resolution, strict StatusCode validation, and SourceTimestamp preservation before a residual value ever enters the broader SCADA Data Ingestion & Time-Series Sync architecture.

Prerequisites & Environment Setup

The implementation targets Python 3.10+ and the asynchronous asyncua client, which is the practical baseline for subscription-based extraction from OPC UA servers such as Ignition, Siemens PCS 7, Rockwell FactoryTalk, and AVEVA System Platform. Pin the library version explicitly; the subscription and security APIs used below are stable on the 1.x line but the handler contract has shifted across earlier releases. pydantic is recommended once records flow into the compliance pipeline so that every emitted envelope is schema-validated before archival.

Two things do not come from a package. First, network access: chlorine analyzers live on the operational-technology (OT) control network, and the extractor must reach the server only through the read-only egress path described in Security Boundary Design — never as a writable client. Second, a server profile: the endpoint URL (opc.tcp://host:4840), the client and server certificates for SignAndEncrypt mutual authentication, and the stable namespace URIs for each vendor’s tag tree. Without the namespace URIs you are guessing at numeric indexes, which is the single most common cause of a silently mis-addressed residual.

python3 -m venv .venv && source .venv/bin/activate
pip install "asyncua==1.1.5" "pydantic==2.7.*"
# Generate a client certificate for SignAndEncrypt mutual auth
openssl req -x509 -newkey rsa:2048 -keyout client_key.pem \
    -out client_cert.pem -days 365 -nodes -subj "/CN=chlorine-extractor"

Step-by-Step Implementation

The extractor is a strictly passive consumer: it browses, resolves, subscribes, and validates, but never writes a node. The five steps below build up a single ChlorineResidualMonitor class that a plant can run as a long-lived service.

Step 1 — Resolve the namespace index from its stable URI

OPC UA servers rarely standardize chlorine node paths, and the numeric namespace index (ns=2, ns=3, ns=4) can renumber on a firmware upgrade even when the underlying tag is unchanged. Hardcoding the index therefore produces brittle integrations. Instead, resolve the index from its stable URI at session startup, so a renumbering is caught as an explicit resolution failure at connect time rather than as a wrong-but-plausible reading. Avoid ns=0, which holds OPC Foundation base types, for process telemetry.

Step 2 — Construct node identifiers and validate node class and data type

Chlorine residuals are published under three identifier formats — a string browse path (ns=2;s=PLC01.Analyzers.FreeCl2.PV), a numeric identifier (ns=3;i=10452), or a GUID (ns=4;g=8a9b2c3d-...). Whichever the server uses, the extractor must confirm before ingesting that the node’s NodeClass is Variable and that its data type resolves to Float or Double. A residual must be numeric for contact-time (CT) and rolling-average math; a Boolean or String node is an alarm state or a text descriptor and must be rejected outright.

Step 3 — Subscribe to monitored items instead of polling

For compliance capture, use monitored-item subscriptions rather than synchronous polling. Subscriptions push only changed values, capture timestamped transitions, reduce network load on the OT segment, and preserve the SourceTimestamp integrity that the 40 CFR 141 data-availability calculation depends on. The client registers each validated node with a sampling_interval and receives a datachange_notification when the analyzer value crosses its deadband. The full extraction session — connect, authenticate, resolve namespace, subscribe, validate, route — is summarized below.

Step 4 — Validate the StatusCode and route by quality tier

Every OPC UA value envelope carries a StatusCode that determines whether the reading is usable for a regulatory average. A Good code means the analyzer is within calibrated parameters. An Uncertain code (sensor warming, calibration in progress) must be flagged in compliance logs, not silently averaged. A Bad code (communication loss, hardware fault) must trigger deterministic fallback routing so it never corrupts a CT calculation or fires a false exceedance. The three-tier pathway is: primary — live Good data; secondary — last-known-good carried with an Uncertain flag into an exception queue; tertiary — historical interpolation or a logged manual override with operator ID and reason code per 40 CFR Part 141 audit rules.

Step 5 — Preserve SourceTimestamp and emit the compliance envelope

OPC UA supplies two timestamps: SourceTimestamp, stamped at the analyzer or PLC, and ServerTimestamp, applied by the server on ingestion. For CT and 40 CFR Part 141 reporting, SourceTimestamp is authoritative and must be propagated to the time-series database; using ServerTimestamp lets server-side clock drift or network latency shift a compliance window. Because SourceTimestamp is optional in the protocol, fall back to a timezone-aware server clock when it is absent, and never store a naive timestamp. The following class implements all five steps end to end.

import asyncio
import logging
from datetime import datetime, timezone
from typing import Dict, List, Optional, Tuple

from asyncua import Client, Node, ua

# Structured logging for compliance audit trails.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S%z",
)
logger = logging.getLogger("chlorine_compliance")

# Vendor namespace URI plus the browse-name suffix for each analyzer tag.
# Node IDs are resolved at runtime from the live namespace index rather than
# hardcoded, so the pipeline survives namespace re-indexing on vendor upgrades.
ANALYZER_NAMESPACE_URI = "urn:plc01:analyzers"
CHLORINE_TAGS = {
    "free_cl2": "PLC01.Analyzers.FreeCl2.PV",
    "total_cl2": "PLC01.Analyzers.TotalCl2.PV",
    "monochloramine": "PLC01.Analyzers.MonoClamine.PV",
}

NUMERIC_TYPES = (ua.VariantType.Float, ua.VariantType.Double)


class ChlorineResidualMonitor:
    def __init__(
        self,
        endpoint: str,
        cert_path: Optional[str] = None,
        key_path: Optional[str] = None,
    ):
        self.endpoint = endpoint
        self.client = Client(url=endpoint)
        self.subscription = None
        self.last_known_values: Dict[str, float] = {}
        self._configure_security(cert_path, key_path)

    def _configure_security(self, cert_path: Optional[str], key_path: Optional[str]) -> None:
        if cert_path and key_path:
            self.client.set_security_string(
                f"Basic256Sha256,SignAndEncrypt,{cert_path},{key_path}"
            )
        else:
            logger.warning(
                "No security credentials provided. Using an unsecured connection "
                "for development and testing only."
            )

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

    async def setup_monitoring(self) -> None:
        # get_namespace_index is a coroutine and must be awaited; it returns the
        # live integer index for the vendor namespace URI (Step 1).
        ns_index = await self.client.get_namespace_index(ANALYZER_NAMESPACE_URI)
        logger.info("Resolved namespace '%s' to index %d.", ANALYZER_NAMESPACE_URI, ns_index)

        valid_nodes: List[Tuple[str, Node]] = []
        for name, tag in CHLORINE_TAGS.items():
            node = self.client.get_node(f"ns={ns_index};s={tag}")
            node_class = await node.read_attribute(ua.AttributeIds.NodeClass)
            variant_type = await node.read_data_type_as_variant_type()

            # Step 2: reject anything that is not a numeric Variable.
            if node_class.Value.Value == ua.NodeClass.Variable and variant_type in NUMERIC_TYPES:
                valid_nodes.append((name, node))
            else:
                logger.error(
                    "Invalid node class or data type for %s. Skipping compliance ingestion.",
                    tag,
                )

        if not valid_nodes:
            raise RuntimeError("No valid chlorine residual nodes found in the address space.")

        # Step 3: create_subscription(period_ms, handler) returns the subscription;
        # subscribe_data_change then registers each monitored item.
        self.subscription = await self.client.create_subscription(1000, self)
        for _name, node in valid_nodes:
            await self.subscription.subscribe_data_change(node)
        logger.info("Subscribed to %d validated chlorine nodes.", len(valid_nodes))

    def datachange_notification(self, node: Node, val, data) -> None:
        # asyncua invokes this callback synchronously, so it must not be a
        # coroutine and must not await. Defer async fallback work to the loop.
        node_id = str(node.nodeid)
        data_value = data.monitored_item.Value
        status_code = data_value.StatusCode
        source_ts = data_value.SourceTimestamp or datetime.now(timezone.utc)

        # Step 4: StatusCode governs whether the value is usable.
        if status_code.is_bad():
            logger.warning(
                "Bad quality code for %s: %s. Triggering fallback routing.",
                node_id,
                status_code,
            )
            self._trigger_fallback(node_id, source_ts)
            return

        if not status_code.is_good():
            logger.info(
                "Uncertain quality for %s. Logging with a compliance exception flag.",
                node_id,
            )

        self.last_known_values[node_id] = val
        # Step 5: SourceTimestamp is preserved verbatim into the audit record.
        logger.info(
            "COMPLIANCE_RECORD | Node: %s | Value: %.3f mg/L | Quality: %s | SourceTimestamp: %s",
            node_id,
            val,
            status_code.name,
            source_ts.isoformat(),
        )

    def _trigger_fallback(self, node_id: str, ts: datetime) -> None:
        if node_id in self.last_known_values:
            fallback_val = self.last_known_values[node_id]
            logger.warning(
                "FALLBACK_ROUTING | Node: %s | Using last-known-good: %.3f mg/L | Time: %s",
                node_id,
                fallback_val,
                ts.isoformat(),
            )
            # Route to the compliance exception queue or historical interpolation service.
        else:
            logger.error(
                "FALLBACK_FAILED | Node: %s | No historical cache. Flagging as a mandatory data gap.",
                node_id,
            )

    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:
    monitor = ChlorineResidualMonitor("opc.tcp://scada-server:4840")
    try:
        await monitor.connect()
        await monitor.setup_monitoring()
        while True:
            await asyncio.sleep(3600)
    except Exception as exc:
        logger.critical("Monitoring pipeline failed: %s", exc, exc_info=True)
    finally:
        await monitor.disconnect()


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

Once records leave this service they are reconciled onto a single axis by the Time-Series Alignment Strategies module, and heavy subscription fan-out across many plants is handed to Async Batch Processing Setup when one event loop should not carry every tag load. When mapping residuals to regulatory thresholds, compute rolling averages aligned to the analyzer’s native clock — for a 4-hour window of $N$ samples the mean is $\overline{C}_{4h} = \frac{1}{N}\sum_{i=1}^{N} C_i$ — and apply a deadband of 0.01 mg/L to suppress telemetry noise without masking genuine disinfectant decay.

Configuration Reference

Pull the namespace URIs and browse paths directly from the server’s address space; treat them as versioned configuration, never as inline constants.

OPC UA value envelope fields

Node identifier formats and validation

Parameters and quality-tier codes

Verification & Testing

Confirm the quality-routing decision with a deterministic unit test that feeds synthetic StatusCode values through the classification logic and asserts the resulting tier, so a regression in the fallback path is caught before it silently corrupts a compliance average.

from asyncua import ua


def classify(status_code: ua.StatusCode) -> str:
    """Mirror of the datachange_notification routing decision (Step 4)."""
    if status_code.is_bad():
        return "FALLBACK"
    if not status_code.is_good():
        return "UNCERTAIN"
    return "VALID"


def test_good_status_is_valid():
    assert classify(ua.StatusCode(ua.StatusCodes.Good)) == "VALID"


def test_uncertain_status_is_flagged_not_averaged():
    uncertain = ua.StatusCode(ua.StatusCodes.UncertainLastUsableValue)
    assert classify(uncertain) == "UNCERTAIN"


def test_bad_status_triggers_fallback():
    bad = ua.StatusCode(ua.StatusCodes.BadCommunicationError)
    assert classify(bad) == "FALLBACK"

Acceptance criteria before promoting the extractor to production:

• Namespace URIs and browse paths are on file and resolved from the URI at session start, never hardcoded as numeric indexes.
• Every subscribed node is confirmed NodeClass == Variable with a Float/Double data type before ingestion.
• SignAndEncrypt with certificate-based mutual authentication is enabled; the unsecured path is disabled outside development.
• Good, Uncertain, and Bad StatusCodes each resolve to VALID, UNCERTAIN, and FALLBACK/GAP respectively — never a raw numeric average of a bad reading.
• SourceTimestamp is preserved verbatim; absence falls back to a timezone-aware UTC clock, never a naive timestamp.
• Every Bad event and fallback activation emits a structured, auditable log line.

Troubleshooting & Gotchas

• Residuals silently address the wrong tag after a firmware upgrade. The vendor renumbered the namespace array and a hardcoded ns=2 now points elsewhere. Always resolve the index from the namespace URI at session start; a resolution failure at connect time is the desired outcome, not a plausible-but-wrong reading.
• A monitoring point goes silent and is read as “no change”. Subscriptions push only on a deadband crossing, so a frozen analyzer produces no datachange_notification. Treat prolonged silence as a candidate gap and cross-check against the cadence from Monitoring Frequency Scheduling rather than assuming the value held steady.
• Compliance windows drift by minutes. The pipeline is averaging on ServerTimestamp. Switch to SourceTimestamp, and normalize every record through aligning irregular SCADA timestamps to UTC before any window is computed.
• A Bad reading fires a false low-residual violation. The extractor averaged an out-of-quality value. Route every Bad StatusCode through the fallback tiers so it becomes a logged FALLBACK/GAP, and let the Violation Detection Rule Engine evaluate exceedances only on VALID and flagged records — the same discipline applied when handling missing sensor readings without triggering false violations.
• Certificates expire and extraction stops overnight. A silent authentication failure looks identical to a network outage. Rotate client and server certificates before expiry, and ship every connection error to a centralized SIEM so an expiring certificate is alerted, not discovered at the next audit.

Related

• OPC UA Data Extraction — parent section and the end-to-end extraction contract
• Parsing Modbus Registers for Turbidity Sensors — sibling extraction workflow for a register-based protocol
• Aligning Irregular SCADA Timestamps to UTC — normalizing the timestamps this extractor emits
• Handling Missing Sensor Readings Without Triggering False Violations — the detection-side counterpart to fallback routing
• Security Boundary Design — the read-only OT egress path this service must use