Metadata-Version: 2.4
Name: iec61850
Version: 0.8.0
Classifier: Development Status :: 2 - Pre-Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: POSIX :: Linux
Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: MacOS
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Networking
Summary: Async-first IEC 61850 client for Python.
Keywords: iec61850,scada,ics,mms,goose,energy,ems
Author: Cheng Sin Pang
License: Apache-2.0
Requires-Python: >=3.11
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Repository, https://github.com/csp0924/iec61850-rust

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

```bash
pip install iec61850
```

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

## Quick start

```python
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

```python
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:

```python
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

```python
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:

```python
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:

```python
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

```python
# 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:

```python
# 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

```python
# 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

```python
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

```python
# 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

```python
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

```python
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

```python
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)
```

## Error handling

```python
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

