iec61850 0.9.0

Async-first IEC 61850 client for Python.

iec61850

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

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
• 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

Install

pip install iec61850

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

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

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