Ingesting SCADA Fault Events with Python
When a distribution breaker trips or a recloser locks out, the SCADA system knows it within milliseconds — but that knowledge arrives as a device-status change on a point in the historian, not as a fault on a named feature of the network. Turning that raw operation into an outage event a trace engine can act on is the job of the ingestion consumer, and doing it by hand or with a brittle script is exactly what fails under storm load. A single trip re-reported three times by a flapping communications link becomes three outages and three dispatched crews; a lockout snapped to the wrong device isolates the wrong section. This procedure builds a resilient consumer that resolves each SCADA signal to a feature and emits an idempotent, auditable event, implementing the resolution and deduplication contract established in SCADA and AMI telemetry integration and feeding the impact and dispatch machinery of outage routing and impact automation. The snapping tolerance it enforces inherits directly from the CRS alignment and geodetic transformations discipline that keeps the SCADA point cache and the GIS network on one spatial frame.
Environment Prerequisites
Lock the following before pointing a consumer at a live SCADA feed. A misconfigured environment produces the most dangerous outcome — a consumer that reports healthy while attaching trips to the wrong features.
- Python runtime: Python 3.11 in an isolated environment (
conda create -n scada-ingest python=3.11) so tolerance and projection behavior stay reproducible across engineers. - Dependencies:
geopandas>=1.0,shapely>=2.0, andpyproj>=3.0for feature resolution;hashlib,dataclasses,logging, anddatetimecome from the standard library. - Durable queue: A message queue in front of the consumer (Kafka, RabbitMQ, Azure Service Bus, or an equivalent) so alarm bursts buffer and a consumer restart replays committed offsets rather than dropping events.
- Device layer: The SCADA-controllable devices (breakers, reclosers, sectionalizers) and monitored spans as a point/line layer readable by
geopandas.read_file, exported from a reconciled, unversioned snapshot on a currently validated topology. - CRS and tolerance: A single authoritative projected EPSG code shared by the SCADA point cache and the GIS network, and a snapping tolerance equal to the network design accuracy (typically 0.1–0.5 m). Parameterize both.
- Clock discipline: A recorded skew budget between the SCADA historian clock and the ingestion host, so the event-time correlation window can be sized against a known worst case.
- Idempotent sink: A staging store keyed on the dedup key that supports upsert, so replays and re-reports reconcile instead of duplicating.
Schema-Aware Validation Protocol — Run Before the Consumer
Most SCADA ingestion failures originate in the setup, not the messages. Work this ordered checklist first; the earliest item is the most frequent culprit.
- Confirm the SCADA cache and the network share one projected CRS. Compare by EPSG code (
gdf.crs.to_epsg()), never by layer name. A sub-meter offset between the historian’s point cache and the GIS network makes the nearest-feature snap attach a trip to an adjacent device. - Verify the device layer is current. Resolution against a stale snapshot taken before a field edit snaps trips to devices that no longer connect as expected. Export from a reconciled, unversioned snapshot immediately before the run.
- Pin the tolerance to design accuracy. A tolerance wider than the spacing between parallel feeders manufactures cross-feeder mis-snaps; one tighter than survey accuracy orphans real trips. Match the junction-edge connectivity tolerance the network already enforces.
- Constrain the status vocabulary. Confirm every inbound status token maps to a known SCADA operation (
TRIP,LOCKOUT,RECLOSE,FAULT_PASSED). An unrecognized token must route to review, never silently drop. - Test the dedup key on a replay. Feed the consumer the same batch twice and assert the deduplicated count equals the batch size on the second pass. A key computed before resolution, or on the raw message id, will fail this and let re-reports through.
Minimal Reproducible Implementation
The consumer below is the centerpiece. It pulls a batch of SCADA messages from a durable queue, normalizes each into the canonical envelope, resolves it to a network feature within tolerance, computes a deterministic deduplication key, and upserts so that duplicate trip re-reports collapse to a single outage event. Every failure mode — a missing field, an unknown status, an out-of-tolerance snap, a transient sink error — is caught and classified rather than allowed to halt the loop, and the function returns a structured result the build gate and audit log both consume. The queue and sink are represented by small protocols so the pattern runs against any concrete transport.
from __future__ import annotations
import hashlib
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Protocol
import geopandas as gpd
from shapely.geometry import Point
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
VALID_STATUS = {"TRIP", "LOCKOUT", "RECLOSE", "FAULT_PASSED"}
class Queue(Protocol):
def poll(self, max_messages: int) -> list[dict]: ...
def ack(self, message_id: str) -> None: ...
class EventSink(Protocol):
def upsert(self, dedup_key: str, event: dict) -> bool:
"""Insert if absent; return True on insert, False if the key existed."""
...
@dataclass
class IngestResult:
"""Structured, gateable outcome of one consume cycle."""
processed: int = 0
inserted: int = 0
deduplicated: int = 0
orphaned: int = 0
review: int = 0
errors: list[str] = field(default_factory=list)
@property
def ok(self) -> bool:
# A run is healthy when orphans stay under 2% of processed.
return not self.errors and self.orphaned <= max(1, self.processed // 50)
def _dedup_key(feature_id: str, event_time: datetime, window_seconds: int) -> str:
"""Deterministic identity: resolved feature + event-time window bucket."""
bucket = int(event_time.astimezone(timezone.utc).timestamp() // window_seconds)
return hashlib.sha256(f"{feature_id}|{bucket}".encode()).hexdigest()[:24]
def _resolve(x: float, y: float, devices: gpd.GeoDataFrame,
id_col: str, tolerance_m: float) -> str | None:
"""Snap a SCADA location to the nearest device within tolerance, else None."""
distances = devices.geometry.distance(Point(x, y))
idx = distances.idxmin()
if float(distances.loc[idx]) > tolerance_m:
return None
return str(devices.loc[idx, id_col])
def consume_scada_faults(
queue: Queue,
sink: EventSink,
devices: gpd.GeoDataFrame,
id_col: str = "DEVICE_ID",
tolerance_m: float = 0.25,
window_seconds: int = 30,
batch_size: int = 500,
) -> IngestResult:
"""Consume one batch of SCADA fault messages into idempotent outage events.
Each message is normalized, resolved to a network feature, deduplicated,
and upserted. Failures are classified and counted, never fatal. Returns a
structured IngestResult for the build gate and audit log.
"""
if devices.crs is None:
raise ValueError("device layer has no defined CRS; align before ingesting")
result = IngestResult()
for msg in queue.poll(batch_size):
result.processed += 1
message_id = msg.get("id", "<unknown>")
try:
# 1. Validate and normalize into the canonical envelope.
status = msg["status"]
if status not in VALID_STATUS:
result.review += 1
logging.warning("msg %s: unknown status %r -> review", message_id, status)
queue.ack(message_id)
continue
event_time = datetime.fromisoformat(msg["event_time"])
x, y = float(msg["x"]), float(msg["y"])
# 2. Resolve the signal to a network feature within tolerance.
feature_id = _resolve(x, y, devices, id_col, tolerance_m)
if feature_id is None:
result.orphaned += 1
logging.warning("msg %s: no feature within %.2f m -> orphan",
message_id, tolerance_m)
queue.ack(message_id)
continue
# 3. Build the idempotent event and upsert on the dedup key.
key = _dedup_key(feature_id, event_time, window_seconds)
event = {
"dedup_key": key,
"source": "SCADA",
"origin_id": msg.get("origin_id"),
"feature_id": feature_id,
"status": status,
"event_time": event_time.isoformat(),
"arrival_time": datetime.now(timezone.utc).isoformat(),
}
if sink.upsert(key, event):
result.inserted += 1
else:
result.deduplicated += 1
queue.ack(message_id)
except (KeyError, ValueError) as exc:
# Malformed payload: record, ack to avoid a poison-message loop.
result.errors.append(f"{message_id}: {exc}")
logging.error("msg %s: malformed payload: %s", message_id, exc)
queue.ack(message_id)
logging.info(
"SCADA ingest cycle: processed=%d inserted=%d dedup=%d orphan=%d review=%d",
result.processed, result.inserted, result.deduplicated,
result.orphaned, result.review,
)
return resultThree details make this consumer safe under real load. Acking a malformed message after recording it prevents a poison message from stalling the partition forever, while the error is still preserved for the audit trail. The IngestResult.ok property turns the run into a build-gateable signal: an orphan spike above two percent fails the gate, which is precisely the symptom of a CRS drift or an over-tight tolerance. And because the dedup key is derived from the resolved feature and the event-time bucket — not the message id — replaying the same batch produces inserts on the first pass and pure deduplications on the second, which is the property that lets several workers consume partitions in parallel without corrupting outage state.
Production Deployment Pattern
A working consumer becomes an engineering control when it is deployed with the same rigor as the model it feeds.
- Run against a reconciled snapshot of the device layer. Refresh the cached
devicesGeoDataFrame from a read-only replica on the topology-validation cadence, never inside the message loop, so resolution never blocks on a geodatabase read or contends with editor locks. - Apply bounded retry with backoff on the sink. Wrap
sink.upsertin a three-attempt exponential backoff so a momentary lock or network blip retries rather than dropping an event or spiking the error count. Idempotency makes the retry safe. - Wire
IngestResultinto CI/CD. Assertresult.okin the deployment pipeline and fail the build on an excessive orphan rate; parse the orphan and review queues to open remediation tickets without blocking the merge. - Promote confirmed events downstream. Once staged, corroborated events flow into impact analysis — the resolved feature ids drive computing affected-customer counts with NetworkX and the isolation traces built on upstream and downstream tracing algorithms.
- Persist an immutable audit record. Append each cycle’s counts, the baseline EPSG, the tolerance and window parameters, and the library versions to a timestamped, version-controlled log — the chain of custody a reliability or rate-case review expects, satisfied automatically rather than reconstructed on demand.
Conclusion
A SCADA fault feed is only as useful as the features it resolves to. By normalizing every trip and lockout into one envelope, snapping it to a device within the network’s design tolerance, and collapsing re-reports through a deterministic dedup key, this consumer turns a noisy device-status stream into idempotent, auditable outage events. The structured result makes the pipeline gateable, so a projection drift or a tolerance error fails a build instead of misrouting a crew. Deploy it against a reconciled snapshot with bounded retries and an immutable log, and SCADA ingestion becomes a dependable first link in the outage-automation chain rather than its weakest one.