iec61850 0.12.0

Async-first IEC 61850 client for Python.

iec61850

Async-first, type-hinted IEC 61850 client for Python.

Scope and protocol boundaries

This package is an MMS-only façade over the rust_61850 workspace. It covers the connection-oriented, application-level services that fit Python's runtime characteristics naturally:

• Client: read / write / control / dataset / reporting (URCB + BRCB) / log query / SCL parsing / directory queries / TLS
• Server hosting: MMS server with URCB / BRCB / LCB / SGCB / control handlers, declared from SCL or from a Python dict spec

Out of scope by design: GOOSE (IEC 61850-8-1) and Sampled Values (IEC 61850-9-2) are hard-real-time L2 protocols (GOOSE T1 = 4 ms, SV = 250 µs / 4 kHz). Their timing budgets fit Rust + SCHED_FIFO, not a GIL-bound runtime with non-deterministic GC pauses. Misrepresenting them as "supported" in Python would invite production protection-scheme failures.

If you need GOOSE / SV publish or subscribe, use the upstream Rust crates directly (same author, same wire-level implementation):

• iec61850-goose — GOOSE publisher / subscriber / receiver (Ethernet L2 raw socket, frame template, retransmission state machine)
• iec61850-sv — Sampled Values 4 kHz publisher + subscriber (Linux SCHED_FIFO publish loop, p99 jitter target < 200 µs)

The PICS table below uses a four-column Py-C / Py-S / Rust-C / Rust-S breakdown precisely so an unsupported Python row points you at the right Rust crate.

Features

• TCP connect / disconnect with timeout
• TLS connect (IEC 62351-3 cipher whitelist, mutual TLS, TLS 1.2 + 1.3, known-peer pinning, CRL, configurable validation knobs)
• Per-connection tuning: request timeout, max outstanding invocations, local max PDU size
• High-level Iec61850Client async context manager wrapping the lifecycle of an IedConnection plus an optional background ReportDispatcher
• Typed scalar read / write: bool, int32, int64, uint32, float, float64, string, timestamp (decoded to datetime), quality (decoded to a Quality dataclass)
• Generic read / write with array-element and sub-component selection
• Directory queries: get_server_directory, get_logical_device_directory, get_logical_node_directory(AcsiClass), get_data_directory
• Schema introspection: get_variable_specification(ref, fc) (recursive MMS type tree) and get_device_model() (per-LD named-variable index)
• Dataset admin: create_data_set, delete_data_set, get_data_set_values, set_data_set_values
• Connection control: disconnect (graceful) and abort (rude close — drop TCP without sending MMS Conclude)
• URCB / BRCB reporting: get_rcb_values, set_rcb_values, install_report_handler, poll_reports, background ReportDispatcher
• Log service: query_journal_by_time and query_journal_after_entry for paginating Log Control Block contents
• SCL / ICD / CID document parser: load_scl(path) / parse_scl(xml) → Scl handle exposing IED inventory, the full document as a nested dict, and a canonical text summary per IED
• Server hosting: IedServer.from_scl(path, ied_name=...) instantiates an MMS server from an SCL document, or IedServer.from_model_spec(spec) from a declarative Python dict when the IED is generated in code rather than authored as XML. Bind, configure (vendor / model name / max connections), enter via async with, push value updates with typed update_* methods, intercept reads / writes with on_read / on_write callbacks, serve control commands (SBO / direct, normal / enhanced) with on_control — operate and wait callbacks may be either sync or async — expose URCB / BRCB reporting via add_dataset / register_urcb / register_brcb, host Log Control Blocks via register_log_control + log_value (in-memory journal backend, ReadJournal over MMS), wire Setting Group callbacks per LD with register_setting_group_handler + force_active_setting_group; atomic multi-attribute updates with with server.batch():; terminate TLS at the server with with_tls() (same IEC 62351-3 profile as the client, plus client-cert pinning)
• Control: select, select_with_value, operate, cancel across the four IEC 61850 control models (direct-normal / direct-enhanced / sbo-normal / sbo-enhanced)
• Typed exception hierarchy: IedError, IedConnectionError, IedTimeoutError, IedDataAccessError, IedServiceError, IedControlError, IedServerError
• SNTP / NTP client: query_sntp(addr, timeout_s) returns an SntpResponse with server time, clock offset, and round-trip estimate computed per RFC 4330 §5
• Type stubs (_native.pyi) and a py.typed marker for full mypy / pyright coverage of the public surface
• Cookbook of runnable examples under examples/ (quickstart client, server from SCL, server from model dict, BRCB reporting, control SBO, SNTP query)

Install

pip install iec61850

Requires Python 3.11+. Wheels are published for Windows x86_64 and Linux x86_64 (manylinux 2014).

Conformance (PICS)

ACSI service support per IEC 61850-7-2 Edition 2.1.

• Py-C / Py-S — exposed through this Python package (client / server).
• Rust-C / Rust-S — supported somewhere in the upstream rust_61850 workspace.

Legend: ✓ supported · — not yet implemented · × by design deliberately not in Python (use the named Rust crate instead) · n/a not applicable · (n) footnote.

Application Association (§7)

Service	Py-C	Py-S	Rust-C	Rust-S
Associate (Two-Party, MMS/TCP)	✓	✓	✓	✓
Abort (graceful Conclude)	✓	✓	✓	✓
Abort (rude — TCP drop)	✓	✓	✓	✓
Release	✓	✓	✓	✓
TLS 1.2 / 1.3 (IEC 62351-3)	✓ (1)	✓ (1)	✓	✓
Mutual TLS / pinned peer / CRL	✓	✓	✓	✓
Authentication (ACSE password)	—	—	—	—

1. Includes 62351-3 cipher whitelist, verify_hostname knob, version pinning, and known-peer profile. Backed by iec61850-tls.

Server class (§8)

Service	Py-C	Py-S	Rust-C	Rust-S
GetServerDirectory	✓	✓	✓	✓
GetServerCapabilities (Ed.2)	—	—	—	—

Logical Device class (§9)

Service	Py-C	Py-S	Rust-C	Rust-S
GetLogicalDeviceDirectory	✓	✓	✓	✓

Logical Node class (§10)

Service	Py-C	Py-S	Rust-C	Rust-S
GetLogicalNodeDirectory	✓	✓	✓	✓
GetAllDataValues	—	—	—	—

Data class (§11)

Service	Py-C	Py-S	Rust-C	Rust-S
GetDataValues	✓	✓ (2)	✓	✓
SetDataValues	✓	✓ (2)	✓	✓
GetDataDirectory	✓	✓	✓	✓
GetDataDefinition / GetVariableSpec	✓	✓	✓	✓

2. Server applies the configured WriteAccessPolicies (default SP | SV | SE) and any registered on_read / on_write callbacks.

Data Set class (§12)

Service	Py-C	Py-S	Rust-C	Rust-S
GetDataSetValues	✓	✓	✓	✓
SetDataSetValues	✓	✓	✓	✓
CreateDataSet (dynamic)	✓	✓	✓	✓
DeleteDataSet	✓	✓	✓	✓
GetDataSetDirectory	—	—	—	—
Static (SCL-defined) datasets	✓	✓ (3)	✓	✓

3. Server registers static datasets through add_dataset(); either bound to a URCB or exposed standalone for GetDataSetValues.

Substitution (§13)

Service	Py-C	Py-S	Rust-C	Rust-S
Set substituted value (FC=SV writes)	✓	✓	✓	✓
Dedicated Substitution service API	—	—	—	—

Setting Group Control Block — SGCB (§14)

Service	Py-C	Py-S	Rust-C	Rust-S
SelectActiveSG / SelectEditSG	✓ (6)	✓ (6)	✓	✓
GetSGCBValues / SetSGCBValues	✓ (6)	✓ (6)	✓	✓
ConfirmEditSGValues	✓ (6)	✓ (6)	✓	✓
Setting access (FC=SG / FC=SE writes)	✓	✓	✓	✓

6. Server-side SGCB is declared via SCL <SettingControl numOfSGs="N"/> on LN0; runtime ActSG / EditSG / ConfirmEditSG state machine plus reservation timeout are owned by the server. Python applications install per-LD callbacks (act_sg_changed / edit_sg_changed / confirm_edit_sg) through register_setting_group_handler(), and can force_active_setting_group() on startup. Clients drive SGCB through the existing write() API on the special MMS path (e.g. write("IED1LD0/LLN0.SGCB.ActSG", FC.SP, 2)).

Reporting — URCB / BRCB (§17)

Service	Py-C	Py-S	Rust-C	Rust-S
URCB — GetURCBValues / SetURCBValues	✓	✓	✓	✓
URCB — Report (TrgOps, OptFlds, BufTm, IntgPd)	✓	✓	✓	✓
URCB — General Interrogation	✓	✓	✓	✓
BRCB — GetBRCBValues / SetBRCBValues / Report	✓	✓ (4)	✓	✓
Background report dispatcher	✓	n/a	✓	n/a

4. Server BRCB hosting via register_brcb() with in-memory ring buffer (entry-count semantics), reconnect resync, and update_typed trigger path.

Logging — LCB / Log (§15)

Service	Py-C	Py-S	Rust-C	Rust-S
ReadJournal (by time / by entry)	✓	✓ (5)	✓	✓
LCB — GetLCBValues / SetLCBValues	—	—	—	—
QueryLogByTime / QueryLogAfter	✓	✓ (5)	✓	✓
Log purging	—	—	—	—

5. Server LCB hosting via register_log_control() with an in-memory journal backend (optionally capacity-bounded, evicts oldest on overflow). Triggers are explicit (log_value); auto-trigger on update_* is not yet wired. LogEna toggles at runtime via set_log_ena.

Generic Substation Event — GOOSE / GSE (§18) (7)

Service	Py-C	Py-S	Rust-C	Rust-S
GoCB — Get / Set values	× by design	× by design	✓	✓
GOOSE publish	× by design	× by design	n/a	✓
GOOSE subscribe	× by design	× by design	✓	n/a

7. By design, not on the Python roadmap. GOOSE is a hard-real-time L2 protocol (IEC 61850-8-1; T1 retransmission floor = 4 ms; protection schemes require deterministic dispatch). Python's GIL + non-deterministic GC make it unsuitable for the GOOSE hot path. Use iec61850-goose directly — it provides GoosePublisher (frame template + retransmission state machine), GooseSubscriber (smpCnt continuity + Q4), and a typestate GooseReceiver. MMS-level GoCB administration is also handled there (via GoCBRegistry in iec61850-server).

Transmission of Sampled Values — SVCB (§19) (8)

Service	Py-C	Py-S	Rust-C	Rust-S
MSVCB / SVCB — Get / Set values	× by design	× by design	—	—
Sampled-value publish	× by design	× by design	n/a	✓
Sampled-value subscribe	× by design	× by design	✓	n/a

8. By design, not on the Python roadmap. SV runs at 4 kHz with a p99 jitter budget < 200 µs (protection profile 256 samples/cycle is even tighter). This is achievable on Linux with SCHED_FIFO + raw socket — not from a GIL-bound runtime. Use iec61850-sv directly — it provides SvPublisher with frame-template + hot-path setters, a Linux publish_thread (clock_nanosleep-based), and SvSubscriber with smpCnt continuity tracking.

Control (§20)

Service	Py-C	Py-S	Rust-C	Rust-S
status-only	✓	✓	✓	✓
direct-normal	✓	✓	✓	✓
sbo-normal	✓	✓	✓	✓
direct-enhanced	✓	✓	✓	✓
sbo-enhanced	✓	✓	✓	✓
Select / SelectWithValue	✓	✓	✓	✓
Operate / Cancel	✓	✓	✓	✓
Test mode, ctlNum, origin	✓	✓	✓	✓
TimeActivatedOperate	—	—	—	—
AddCause feedback	✓	✓	✓	✓

Time and Time Synchronization (§21)

Service	Py-C	Py-S	Rust-C	Rust-S
UTC time read (Timestamp DA)	✓	✓	✓	✓
Time-quality flags on update	n/a	✓	n/a	✓
SNTP / NTP responder	—	—	n/a	✓ (9)
SNTP / NTP client	✓ (10)	n/a	✓ (10)	n/a

9. SNTPv4 unicast server in iec61850-sntp (mode 3 → mode 4 reply). Not yet exposed through this Python package; the package targets the client side for time sync.
10. SNTPv4 unicast client. From Python use iec61850.query_sntp(addr, timeout_s); from Rust use iec61850_sntp::SntpClient. Single-sample offset / round-trip via RFC 4330 §5 four-timestamp formula.

File Transfer (§23)

Service	Py-C	Py-S	Rust-C	Rust-S
GetFile / SetFile / DeleteFile	— (11)	— (11)	— (11)	— (11)
GetFileAttributeValues	— (11)	— (11)	— (11)	— (11)
GetServerDirectory(FILE)	— (11)	— (11)	— (11)	— (11)

11. File services are deferred until the upstream Rust workspace ships PDU encode / decode, client API, and server-side dispatch. The work is tracked under the iec61850-rs workspace's Phase 2 plan (docs/tasks/Phase2-file-services.md). The Python package will pick them up automatically once the Rust client surface lands.

Tooling (out-of-band)

Capability	Python	Rust
SCL / ICD / CID parser	✓	✓
Two-stage SCL pipeline (XML + cross-element resolution)	✓	✓
Typed-spec introspection (TypeSpec)	✓	✓
Code-driven IedModel construction	✓	✓
Async SNTP / NTP client	✓	✓

Quick start

import asyncio
import iec61850

async def main():
    conn = await iec61850.IedConnection.connect("127.0.0.1:102", timeout_ms=5000)
    try:
        status = await conn.read_int32("simpleIOGenericIO/LLN0.Mod.stVal", iec61850.FC.ST)
        vendor = await conn.read_string("simpleIOGenericIO/LLN0.NamPlt.vendor", iec61850.FC.DC)
        quality = await conn.read_quality("simpleIOGenericIO/GGIO1.Ind1.q", iec61850.FC.ST)
        print(status, vendor, quality.validity)
    finally:
        await conn.disconnect()

asyncio.run(main())

TLS

ca_pem = open("ca.pem", "rb").read()
tls = iec61850.TlsConfig(ca_pem=ca_pem)

conn = await iec61850.IedConnection.connect_tls(
    "ied.example.com:3782",
    tls,
    server_name="ied.example.com",
    timeout_ms=5000,
)

Mutual TLS adds a client cert and key:

tls = iec61850.TlsConfig(
    ca_pem=open("ca.pem", "rb").read(),
    client_cert_pem=open("client.crt", "rb").read(),
    client_key_pem=open("client.key", "rb").read(),
)

Defaults: TLS 1.2–1.3, IEC 62351-3 cipher whitelist, chain and time validation on, session resumption on. Set verify_hostname=False on TlsConfig to skip SNI / SAN hostname matching for closed-network commissioning (other validation still applies).

Pinned peers, CRL, version pinning

tls = iec61850.TlsConfig(
    ca_pem=open("ca.pem", "rb").read(),
    # Restrict accepted server certificates to a fixed allow-list
    # (IEC 62351-3 known-peer profile).
    allow_only_known_peers=True,
    known_peer_pems=(open("ied1.crt", "rb").read(),),
    # Pin a single TLS version.
    min_version=iec61850.TlsVersion.TLS_1_3,
    max_version=iec61850.TlsVersion.TLS_1_3,
    # Revocation checks.
    crl_pems=(open("ca.crl.pem", "rb").read(),),
)

High-level client

Iec61850Client wraps IedConnection as an async context manager and optionally runs a background report dispatcher:

cfg = iec61850.Iec61850ClientConfig(
    address="ied.example.com",
    port=102,
    timeout_ms=5000,
    # Tuning that flows down into the underlying MMS client.
    request_timeout_ms=3000,
    max_outstanding=4,
    local_max_pdu_size=16384,
    # Background dispatcher; None to disable.
    report_dispatcher_interval_ms=100,
)

async with iec61850.Iec61850Client(cfg) as cli:
    val = await cli.connection.read_float(
        "simpleIOGenericIO/GGIO1.AnIn1.mag.f", iec61850.FC.MX
    )

For TLS, pass a TlsConfig on the config and (optionally) override the SNI:

cfg = iec61850.Iec61850ClientConfig(
    address="10.0.0.1",            # network address
    port=3782,
    tls=iec61850.TlsConfig(ca_pem=ca_pem),
    tls_server_name="ied.example.com",  # SNI; defaults to `address`
)

The same request_timeout_ms / max_outstanding / local_max_pdu_size keyword arguments are also accepted on IedConnection.connect and IedConnection.connect_tls for callers that prefer to manage the connection lifecycle directly.

Generic read / write

# Native Python types: scalars surface as bool / int / float / str;
# bytes-like kinds as bytes; arrays and structures as list.
value = await conn.read("simpleIOGenericIO/GGIO1.AnIn1.mag.f", iec61850.FC.MX)

await conn.write(
    "simpleIOGenericIO/LLN0.NamPlt.vendor", iec61850.FC.DC, "Acme"
)

Array elements and sub-components are addressed with keyword arguments:

# Reads the third element of an array DA.
elem = await conn.read("LD/LN.Arr", iec61850.FC.ST, array_index=2)

# Reads `stVal` inside the third element.
sub = await conn.read(
    "LD/LN.Arr", iec61850.FC.ST, array_index=2, component="stVal"
)

Schema introspection

# Per-variable MMS TypeSpecification, returned as a nested dict.
ts = await conn.get_variable_specification(
    "simpleIOGenericIO/LLN0.Mod", iec61850.FC.ST
)
# ts == {"kind": "structure", "components": [
#   {"name": "stVal", "type": {"kind": "integer", "width_bits": 32}},
#   {"name": "q",     "type": {"kind": "bit_string", "bits": 13}},
#   ...
# ]}

# Whole device-model index — list of logical devices with their MMS
# NamedVariable names. First call fetches; subsequent calls hit a cache.
model = await conn.get_device_model()
for ld in model["logical_devices"]:
    print(ld["name"], len(ld["variables"]))

# Force a re-fetch if the server model may have changed.
fresh = await conn.get_device_model(refresh=True)

Every type-spec node carries a "kind" discriminator. Scalar kinds add payload fields appropriate for the type (width_bits, format_width / exponent_width, max_chars, bits, ...). "array" adds element_count plus a recursive element_type. "structure" adds components — a list of {"name", "type"} entries. "unknown" surfaces the raw ASN.1 tag for forward compatibility.

Datasets

await conn.create_data_set(
    "simpleIOGenericIO/LLN0.ds1",
    [
        iec61850.DataSetMember("simpleIOGenericIO/GGIO1.AnIn1.mag.f", iec61850.FC.MX),
        iec61850.DataSetMember("simpleIOGenericIO/GGIO1.Ind1.stVal", iec61850.FC.ST),
    ],
)

values = await conn.get_data_set_values("simpleIOGenericIO/LLN0.ds1")
# values is a list ordered to match the dataset members.

await conn.set_data_set_values("simpleIOGenericIO/LLN0.ds1", [3.14, True])

deleted = await conn.delete_data_set("simpleIOGenericIO/LLN0.ds1")

DataSetMember accepts optional array_index / component to target an array element or a sub-component (component requires array_index); the facade composes the alternate-access reference for you.

get_data_set_values and set_data_set_values raise IedDataAccessError when any single entry's access or write fails on the server, with the entry index in the error message.

Connection control

# Normal close — MMS Conclude exchange, then TCP shutdown.
await conn.disconnect()

# Rude close — drop the TCP socket without negotiation. Use when the peer
# stops responding or a normal disconnect would block.
await conn.abort()

Log service

LOG_REF = "IED1LD0/LLN0$LG$evlog"

# First page — by time range. ``more_follows`` signals that the server
# truncated the response and the caller should resume.
entries, more = await conn.query_journal_by_time(LOG_REF, start_ms, end_ms)
for e in entries:
    print(e.time_ms, e.entry_id.hex(), len(e.variables))

# Resume from the last seen entry. Both arguments — the entry's ``time_ms``
# and 8-byte ``entry_id`` — are applied as filters server-side.
cursor = entries[-1]
more_entries, _ = await conn.query_journal_after_entry(
    LOG_REF, cursor.time_ms, cursor.entry_id
)

JournalEntry.variables is a tuple of JournalEntryVariable(data_ref, value, reason_code). Values follow the same conversion rules as IedConnection.read — scalars surface natively; bytes-like kinds (BIT_STRING / OCTET_STRING / UTC_TIME / BINARY_TIME) as bytes; composites as list.

Reporting

def on_report(report: iec61850.ClientReport) -> None:
    print(report.rcb_reference, len(report.entries))

rcb = await conn.get_rcb_values("simpleIOGenericIO/LLN0$RP$urcb01")
rcb.resv = True
rcb.rpt_ena = True
await conn.set_rcb_values(rcb, iec61850.RcbWriteMask.fields("resv", "rpt_ena"))
await conn.install_report_handler(rcb.object_reference, on_report)

dispatcher = conn.spawn_report_dispatcher(interval_ms=100)
try:
    await asyncio.sleep(10)
finally:
    await dispatcher.aclose()

Control

spc = conn.create_control_object(
    "IED1LD0/GGIO1.SPCSO1",
    iec61850.ControlModel.SBO_ENHANCED,
)
spc.set_origin(iec61850.OriginValue(or_cat=3, or_ident=b"py-client"))

if (await spc.select_with_value(True)).success:
    outcome = await spc.operate(True)
    if not outcome.success:
        print("operate failed:", outcome.add_cause)

SCL / ICD / CID parser

Load an IED configuration document and inspect it as plain Python data:

scl = iec61850.load_scl("MyDevice.icd")

scl.ieds()                          # ['IED1']

doc = scl.to_dict()
doc["ieds"][0]["name"]              # 'IED1'
doc["ieds"][0]["manufacturer"]      # 'ACME'

ld = doc["ieds"][0]["access_points"][0]["server"]["logical_devices"][0]
ld["inst"]                          # 'GenericIO'
[ln["ln_class"] for ln in ld["logical_nodes"]]   # ['LLN0', 'GGIO', ...]

# Resolve a logical-node's data objects via the DataTypeTemplates section:
ln_type_id = ld["logical_nodes"][0]["ln_type"]
ln_type = doc["data_type_templates"]["ln_node_types"][ln_type_id]
[do["name"] for do in ln_type["dos"]]            # ['Mod', 'Beh', ...]

Both load_scl and parse_scl run the full two-stage pipeline (XML syntax → cross-element type-reference resolution). The returned dict mirrors the SCL XML structure; type references stay as strings so callers can index into doc["data_type_templates"] (ln_node_types / do_types / da_types / enum_types) themselves.

For a stable text representation — useful as a regression / diff oracle — ask for the canonical summary of a single IED:

print(scl.summary("IED1"))
# IED name=IED1
#   lds count=1
#     LD inst=GenericIO ld_name=<None> lns=2
#       LN class=LLN0 inst= prefix= dos=2 ...
#       ...

Parse failures surface as SclError, a subclass of IedError, with line, column, element_path, attribute, kind, and message attributes set so the offending location is directly reachable:

try:
    iec61850.load_scl("broken.icd")
except iec61850.SclError as e:
    print(e.kind, "at", e.line, ":", e.column, "→", e.element_path, "@", e.attribute)
    # e.g. UnresolvedTypeReference at 42 : 7 → SCL/IED[name="IED1"]/.../LN[...] @ lnType

Server hosting

Host an IED defined by an SCL / ICD / CID document as an MMS server. IedServer.from_scl() builds the runtime model, bind() selects the TCP address (port 0 requests an OS-assigned port), and async with manages the lifecycle. While running, push value updates with the typed update_* methods, addressing data attributes by "<LD>/<LN>.<DO>.<DA>[.<sub>]*".

import asyncio
import iec61850

async def main():
    server = iec61850.IedServer.from_scl("plant.icd", ied_name="IED1")
    server.bind("0.0.0.0:0")
    server.vendor = "ACME"
    server.model_name = "Generic-IO"
    server.max_connections = 5

    async with server:
        print("listening on", server.bound_addr)
        while True:
            server.update_bool("GenericIO/GGIO1.Ind1.stVal", True)
            server.update_float32("GenericIO/MMXU1.TotW.mag.f", measure_power())
            await asyncio.sleep(0.1)

asyncio.run(main())

Configuration setters (vendor, model_name, revision, max_connections) must be called before start(). Updating an unknown path raises KeyError; a type mismatch (e.g. pushing update_float32 to a BOOLEAN attribute) raises IedDataAccessError. Bind failures and other lifecycle errors surface as IedServerError.

The supported typed updates are update_bool, update_int32, update_int64, update_uint32, update_float32, update_float64, and update_string.

Server-side TLS

with_tls() wraps the listener in an IEC 62351-3 TLS acceptor. Only valid before start(); calling it more than once raises RuntimeError.

server.with_tls(
    server_cert_pem=open("server.crt", "rb").read(),
    server_key_pem=open("server.key", "rb").read(),
)

Mutual TLS adds a CA bundle for client-chain validation and, optionally, a pinned peer list (only certificates whose SPKI matches one of the known peers are accepted):

server.with_tls(
    server_cert_pem=server_cert,
    server_key_pem=server_key,
    client_ca_pem=ca_pem,
    allow_only_known_peers=True,
    known_peer_pems=[peer1_pem, peer2_pem],
    crl_pems=[crl_pem],
)

Defaults: TLS 1.2 – 1.3, IEC 62351-3 cipher whitelist, chain validation on, time validation on, session resumption on, no client-cert pinning. min_tls_version and max_tls_version accept "tls1.2" / "tls1.3".

Read / write callbacks

Register per-attribute callbacks to override cached reads or intercept incoming writes. Both on_read and on_write may be called before start() (queued and installed at startup) or while the server is running (installed immediately). Re-registering the same path replaces the previous callback.

def measure_indication(path: str) -> bool:
    # Sampled from physical I/O on every client read.
    return read_io(path)

def validate_setpoint(path: str, value: int) -> bool:
    if value < 0 or value > 100:
        err = iec61850.IedDataAccessError("setpoint out of range")
        err.code = "ObjectValueInvalid"
        raise err
    apply_setpoint(value)
    return True   # also store value in the server-side cache

server.on_read("GenericIO/GGIO1.Ind1.stVal", measure_indication)
server.on_write("GenericIO/GGIO1.SetPt1.setVal", validate_setpoint)

on_read return values:

Return	Behaviour
any scalar	the value is returned to the client
None	fall through to the cached value
raises	read fails with IedDataAccessError

on_write return values:

Return	Behaviour
True	accept; cache is updated with the incoming value
False / None	accept; cache is not updated (you manage it)
raises	reject; client sees IedDataAccessError

Set a code attribute on the raised exception to control the reported DataAccessError variant — "HardwareFault", "TemporarilyUnavailable", "ObjectAccessDenied", "ObjectValueInvalid", etc. Without code the server reports ObjectAccessDenied.

Control callbacks

on_control binds the server-side execution of a control object. Address it at the DO ("<LD>/<LN>.<DO>") and declare which IEC 61850 control model the DO uses. Up to three callbacks may be supplied:

• check — sync. Static validation before the operate phase fires (interlocks, mode, permissions). Raise to reject; the return value is ignored.
• operate — sync or async. The actual command execution. Raise on failure; the return value is ignored.
• wait — sync or async. Dynamic check during the operate phase for sbo-enhanced controls (e.g. wait for synchro-check confirmation).

Each callback receives (path, ctl_val, action). action is a dict with ctl_num, test, synchro_check, interlock_check, is_select, ctl_time_ms, and origin (a sub-dict with or_cat and or_ident).

async def operate(path: str, value: bool, action: dict) -> None:
    if action["test"]:
        return                                  # test command — no I/O
    await drive_breaker(path, value)

def check(_path: str, _value: bool, _action: dict) -> None:
    if interlock_blocked():
        err = iec61850.IedControlError("interlocked")
        err.add_cause = "BlockedByInterlocking"
        raise err

server.on_control(
    "GenericIO/GGIO1.SPCSO1",
    ctl_model="direct-normal",
    check=check,
    operate=operate,
)

ctl_model is one of "status-only", "direct-normal", "sbo-normal", "direct-enhanced", "sbo-enhanced". For SBO models the optional sbo_timeout_ms (default 30000) and sbo_class ("operate-once" or "operate-many", default "operate-once") configure the select-phase behaviour.

Raise an exception from any callback to reject the command. Set add_cause on the exception to the variant name (e.g. "BlockedByInterlocking", "BlockedByProcess", "NotSupported") or its numeric MMS code; absence falls back to "Unknown".

Datasets and unbuffered reporting

Declare a server-side dataset and bind it to an Unbuffered Report Control Block (URCB) before start(). The same client APIs (get_rcb_values, set_rcb_values, install_report_handler) consume reports from the URCB once the server is running.

server.add_dataset(
    "GGIO1$ds1",
    [
        "GenericIO/GGIO1.Ind1.stVal",
        "GenericIO/GGIO1.AnIn1.mag.f",
    ],
)

server.register_urcb(
    "GenericIO/LLN0.urcb01",
    dataset="GGIO1$ds1",
    trg_ops=["data_changed", "gi"],
    opt_flds=["seq_num", "time_stamp", "reason", "data_set"],
    buf_tm_ms=50,
)

Dataset names follow the IEC 61850 convention "<LN>$<dsName>". Every entry in a dataset must belong to the same logical device. register_urcb accepts rpt_id (defaults to "<domain>/<LN>$RP$<rcb_name>"), conf_rev, trg_ops, opt_flds, buf_tm_ms, and intg_pd_ms.

Trigger options: "data_changed", "quality_changed", "data_update", "integrity", "gi", plus the aliases "all" and "none".

Optional fields: "seq_num", "time_stamp", "reason", "data_set", "data_reference", "conf_rev", "buffer_overflow", "entry_id". Per IEC 61850-7-2 §15, buffer_overflow and entry_id are masked out on the wire for unbuffered reports.

Datasets without a URCB are still reachable via get_data_set_values.

Buffered reporting (BRCB)

register_brcb mirrors register_urcb but binds the dataset to a Buffered Report Control Block ($BR$ MMS path). Reports are held in a per-RCB ring buffer until a client connects, so transient disconnects do not lose updates.

server.register_brcb(
    "GenericIO/LLN0.brcb01",
    dataset="GGIO1$ds1",
    trg_ops=["data_changed", "gi"],
    opt_flds=[
        "seq_num", "time_stamp", "reason", "data_set",
        "buffer_overflow", "entry_id",
    ],
    buf_tm_ms=50,
    buffer_capacity=128,
)

Per IEC 61850-7-2 §15, buffer_overflow and entry_id are honoured on the wire for BRCBs (URCBs mask them out). Additional knobs:

• buffer_capacity (default 64) — entry-count ring size; the buffer evicts oldest entries when full and surfaces overflow through the buffer_overflow field on the next report.
• with_resv_tms (default True) — expose the Edition 2+ ResvTms field for client reservation.
• with_owner (default False) — expose the Edition 2+ Owner field.

The same dataset can back both a URCB and a BRCB; client-side APIs (get_rcb_values, set_rcb_values, install_report_handler, ReportDispatcher) handle both transparently — RcbHandle.is_buffered discriminates them.

Logging (LCB)

register_log_control declares a Log Control Block ($LG$ MMS path). Each block backs onto an in-memory journal (InMemoryLogStorage), which can be unbounded (default) or capped to a fixed entry count (oldest entries are evicted on overflow). Clients pull the journal contents over MMS ReadJournal via query_journal_by_time / query_journal_after_entry.

server.register_log_control(
    "GenericIO/LLN0.lcb01",
    dataset="LLN0$evlogds",
    trg_ops=["data_changed"],
    storage_capacity=1000,
)

async with server:
    server.log_value(
        "GenericIO/LLN0.lcb01",
        data_ref="IED1GenericIO/GGIO1$ST$Ind1$stVal",
        value=True,
        reason_code=0x02,            # bit 1 = data_changed
    )

Triggers are explicit: log_value writes one entry per call rather than auto-tracking update_*. log_value returns the 8-byte entry id (as an int) on success, or None when the block's LogEna is disabled and the trigger was silently skipped. Toggle the enable state at runtime with set_log_ena(path, on).

Setting groups (SGCB)

A Setting Group Control Block is declared in SCL on <LN0>:

<LN0 lnClass="LLN0" inst="" lnType="LLN0_0">
  <SettingControl numOfSGs="3" actSG="1" resvTms="60"/>
</LN0>

The server tracks ActSG / EditSG / ConfirmEditSG state, enforces the single-client edit-session lock, and times out abandoned reservations. Python applications opt into the three veto / commit points per LD:

def on_act_sg(new_sg: int, conn_id: int) -> bool:
    return new_sg in allowed_sgs        # return False → ObjectAccessDenied

def on_confirm(edit_sg: int, conn_id: int) -> None:
    persist_pending_settings(edit_sg)   # commit FC=SE staging buffer

server.register_setting_group_handler(
    "GenericIO",
    on_act_sg=on_act_sg,
    on_confirm=on_confirm,
)

get_setting_group_info(ld_inst) returns the live snapshot (num_of_sg / act_sg / edit_sg / cnf_edit / last_act_tm_ms / resv_tms_s). force_active_setting_group(ld_inst, sg) switches the active group without consulting the callback — intended for startup state restoration. Calling register_setting_group_handler again at runtime atomically replaces the previous handler.

Clients drive SGCB through the regular write API on the special MMS path:

await conn.write("IED1GenericIO/LLN0.SGCB.ActSG", FC.SP, 2)    # SelectActiveSG
await conn.write("IED1GenericIO/LLN0.SGCB.EditSG", FC.SP, 2)   # open edit
await conn.write("IED1GenericIO/LLN0.SGCB.CnfEdit", FC.SP, True)  # commit

Building a server from a model dict (no SCL)

When the IED schema is generated in Python — code-driven test rigs, dynamic device skeletons, or runtimes that prefer dict-driven configuration over XML — from_model_spec consumes a declarative spec dict that maps onto the same model the SCL parser produces. The RCB / LCB / SGCB declarations land in the model and are picked up by start() exactly as SCL-derived ones would be; every callback and register_* method works against the same "<LD>/<LN>.<DO>[.<DA>]*" paths.

spec = {
    "ied_name": "IED1",
    "lds": [{
        "inst": "GenericIO",
        "lns": [
            {
                "lln0": True,
                "dos": [{
                    "name": "Mod",
                    "das": [
                        {"name": "stVal", "fc": "ST", "type": "Enumerated",
                         "trg_ops": ["data_changed"],
                         "value": {"type": "int", "value": 1}},
                        {"name": "q", "fc": "ST", "type": "Quality"},
                        {"name": "t", "fc": "ST", "type": "Timestamp"},
                    ],
                }],
                "sgcb": {"num_of_sg": 3, "act_sg": 1},
            },
            {
                "class": "GGIO", "inst": "1",
                "dos": [{
                    "name": "Ind1",
                    "das": [
                        {"name": "stVal", "fc": "ST", "type": "Boolean"},
                        {"name": "q", "fc": "ST", "type": "Quality"},
                        {"name": "t", "fc": "ST", "type": "Timestamp"},
                    ],
                }],
                "datasets": [{
                    "name": "Events", "entries": [
                        {"ln_name": "GGIO1", "fc": "ST",
                         "do_path": ["Ind1", "stVal"]},
                    ],
                }],
                "rcbs": [{
                    "name": "Events01", "buffered": False,
                    "dataset_ref": "Events", "conf_rev": 1,
                    "trg_ops": ["data_changed", "integrity"],
                    "opt_flds": ["seq_num", "time_stamp", "reason"],
                    "buf_tm_ms": 100,
                }],
            },
        ],
    }],
}

server = iec61850.IedServer.from_model_spec(spec)
server.bind("127.0.0.1:0")
async with server:
    ...

type accepts every IEC 61850-7-3 spelling ("Boolean", "Int32", "Float32", "Enumerated", "Timestamp", "Quality", …); sized variants use the object form {"type": "OctetString", "max_len": 64}. value is a tagged dict ({"type": "int", "value": 1}, {"type": "bit_string", "padding": 3, "data": b"\x00\x00"}, …) or {"type": "default"} for the type's zero. Constructed (SDA-bearing) DAs go under "constructed_das" with a "children" list; nested DOs go under "sub_dos". The underlying IedModelBuilder enforces all invariants (LLN0 first, SGCB only on LLN0, dataset entries resolve to real LN/DO) and surfaces violations as ValueError.

Atomic batch updates

server.batch() returns a synchronous context manager that holds the server's data-model lock for the duration of the with block. Concurrent batches raise RuntimeError rather than deadlocking, so callers can choose to retry or fail fast.

with server.batch():
    server.update_bool("GenericIO/GGIO1.Ind1.stVal", True)
    server.update_float32("GenericIO/GGIO1.AnIn1.mag.f", 12.5)
    server.update_int32("GenericIO/GGIO1.SetPt1.setVal", 7)

Error handling

try:
    conn = await iec61850.IedConnection.connect("10.0.0.1:102", timeout_ms=2000)
except iec61850.IedTimeoutError:
    ...   # connection timed out
except iec61850.IedConnectionError:
    ...   # TCP / OSI stack failure
except iec61850.IedError:
    ...   # catch-all base for any IEC 61850 error

License

Apache-2.0