Skip to content

DER profile

The IEEE 1815.2 point map, and the builder that turns a map and a binding into an outstation. Serving a DER is the guide; this is the reference.

Map

py1815.profile.model

The point model: what a loaded profile table says about one point.

A map is a set of points with absolute indices. The tables state some indices relative to a block (the second inverter's fourth analog input), and those are resolved when the map is loaded for a DER of known composition (D37), so nothing here or after it does index arithmetic.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

Kind

Bases: StrEnum

The five kinds of point the profile defines, by the tables' own names.

Source code in src/py1815/profile/model.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class Kind(StrEnum):
    """The five kinds of point the profile defines, by the tables' own names."""

    BO = "BO"
    BI = "BI"
    AO = "AO"
    AI = "AI"
    CTR = "CTR"

    @property
    def is_output(self) -> bool:
        """Whether a master commands this kind, and does not merely read it."""
        return self in (Kind.BO, Kind.AO)

    @property
    def is_analog(self) -> bool:
        """Whether this kind carries a scaled number."""
        return self in (Kind.AO, Kind.AI)
is_output property 
is_output: bool

Whether a master commands this kind, and does not merely read it.

is_analog property 
is_analog: bool

Whether this kind carries a scaled number.

Composition dataclass

How many of each repeating component a DER has.

The equipment blocks of the tables are stated once per unit, and the number of units is the DER's. Zero of a component means its block is not served at all.

Source code in src/py1815/profile/model.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@dataclass(frozen=True)
class Composition:
    """How many of each repeating component a DER has.

    The equipment blocks of the tables are stated once per unit, and the
    number of units is the DER's. Zero of a component means its block is not
    served at all.
    """

    meters: int = 0
    der_units: int = 0
    inverters: int = 0
    batteries: int = 0

    def __post_init__(self) -> None:
        for name in ("meters", "der_units", "inverters", "batteries"):
            if getattr(self, name) < 0:
                raise ValueError(f"{name} is {getattr(self, name)}; a count cannot be negative")

    def count(self, component: str) -> int:
        """The number of units of a component, by the tables' key for it."""
        return int(getattr(self, component))
count 
count(component: str) -> int

The number of units of a component, by the tables' key for it.

Source code in src/py1815/profile/model.py 
62
63
64
def count(self, component: str) -> int:
    """The number of units of a component, by the tables' key for it."""
    return int(getattr(self, component))

Point dataclass

One point, at the index this DER serves it at.

Source code in src/py1815/profile/model.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@dataclass(frozen=True)
class Point:
    """One point, at the index this DER serves it at."""

    kind: Kind
    index: int
    name: str
    #: The heading the tables list the point under.
    section: str | None = None
    #: What the tables say the point is for; functions share one purpose.
    purpose: str | None = None
    #: 1, 2 or 3 for a point that reports events, 0 for one read only as
    #: static data, and None for one that belongs to no class at all -- which
    #: is how the tables mark a point left out of a class 0 response.
    event_class: int | None = None
    #: Whether the profile requires every outstation to implement the point.
    mandatory: bool = False
    #: The transmitted value times this, plus ``offset``, is the engineering
    #: value. None where the tables state no scaling, and the value travels
    #: as it is.
    multiplier: float | None = None
    offset: float = 0.0
    #: The range of the *transmitted* value, where the tables give one.
    minimum: float | None = None
    maximum: float | None = None
    units: str | None = None
    #: For a binary point, what the tables call its two states: the name of
    #: the state when the value is 0, then when it is 1.
    states: tuple[str, str] | None = None
    #: The point this one is paired with: an output's status input, or an
    #: input's commanding output.
    associated: Address | None = None
    #: For a "supports" input, the output that enables the function it
    #: reports on. Support is derived from whether that output is bound (D40).
    enabled_by: Address | None = None
    #: The repeating block the point belongs to and which unit of it, one-based.
    block: str | None = None
    unit: int | None = None
    #: A value the tables themselves fix, in engineering units: the profile
    #: version, or a block's starting index in the advertisement block.
    fixed_value: float | None = None
    #: For a counter: whether a frozen counter exists beside it, and the class
    #: its freeze events report in.
    frozen: bool = False
    frozen_event_class: int | None = None

    @property
    def address(self) -> Address:
        """The kind and index together, which is what identifies a point."""
        return (self.kind, self.index)

    @property
    def in_class_0(self) -> bool:
        """Whether an integrity poll carries this point, by the tables alone."""
        return self.event_class is not None

    def to_wire(self, value: float) -> float:
        """An engineering value as the number that travels."""
        if self.multiplier is None:
            return value
        return (value - self.offset) / self.multiplier

    def from_wire(self, raw: float) -> float:
        """A transmitted number as the engineering value it stands for."""
        if self.multiplier is None:
            return raw
        return raw * self.multiplier + self.offset

    def in_range(self, raw: float) -> bool:
        """Whether a transmitted value is inside the range the tables give."""
        if self.minimum is not None and raw < self.minimum:
            return False
        return not (self.maximum is not None and raw > self.maximum)
address property 
address: Address

The kind and index together, which is what identifies a point.

in_class_0 property 
in_class_0: bool

Whether an integrity poll carries this point, by the tables alone.

to_wire 
to_wire(value: float) -> float

An engineering value as the number that travels.

Source code in src/py1815/profile/model.py 
123
124
125
126
127
def to_wire(self, value: float) -> float:
    """An engineering value as the number that travels."""
    if self.multiplier is None:
        return value
    return (value - self.offset) / self.multiplier
from_wire 
from_wire(raw: float) -> float

A transmitted number as the engineering value it stands for.

Source code in src/py1815/profile/model.py 
129
130
131
132
133
def from_wire(self, raw: float) -> float:
    """A transmitted number as the engineering value it stands for."""
    if self.multiplier is None:
        return raw
    return raw * self.multiplier + self.offset
in_range 
in_range(raw: float) -> bool

Whether a transmitted value is inside the range the tables give.

Source code in src/py1815/profile/model.py 
135
136
137
138
139
def in_range(self, raw: float) -> bool:
    """Whether a transmitted value is inside the range the tables give."""
    if self.minimum is not None and raw < self.minimum:
        return False
    return not (self.maximum is not None and raw > self.maximum)

MapError

Bases: ValueError

The tables, resolved for this composition, do not form a usable map.

Source code in src/py1815/profile/model.py 
142
143
class MapError(ValueError):
    """The tables, resolved for this composition, do not form a usable map."""

PointMap dataclass

Every point of the profile, resolved for one DER.

Source code in src/py1815/profile/model.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@dataclass(frozen=True)
class PointMap:
    """Every point of the profile, resolved for one DER."""

    points: Mapping[Address, Point]
    composition: Composition = field(default_factory=Composition)
    edition: str = ""
    profile_version: str | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "points", MappingProxyType(dict(self.points)))

    def __contains__(self, address: object) -> bool:
        return address in self.points

    def __len__(self) -> int:
        return len(self.points)

    def get(self, kind: Kind, index: int) -> Point | None:
        """The point at an address, or None if the map holds none there."""
        return self.points.get((kind, index))

    def point(self, kind: Kind, index: int) -> Point:
        """The point at an address, or the error that says the map has none."""
        found = self.points.get((kind, index))
        if found is None:
            raise MapError(f"the map holds no {kind.value}{index}")
        return found

    def of(self, kind: Kind) -> Iterator[Point]:
        """Every point of one kind, in index order."""
        return iter(sorted((p for p in self.points.values() if p.kind is kind), key=_by_index))
get 
get(kind: Kind, index: int) -> Point | None

The point at an address, or None if the map holds none there.

Source code in src/py1815/profile/model.py 
164
165
166
def get(self, kind: Kind, index: int) -> Point | None:
    """The point at an address, or None if the map holds none there."""
    return self.points.get((kind, index))
point 
point(kind: Kind, index: int) -> Point

The point at an address, or the error that says the map has none.

Source code in src/py1815/profile/model.py 
168
169
170
171
172
173
def point(self, kind: Kind, index: int) -> Point:
    """The point at an address, or the error that says the map has none."""
    found = self.points.get((kind, index))
    if found is None:
        raise MapError(f"the map holds no {kind.value}{index}")
    return found
of 
of(kind: Kind) -> Iterator[Point]

Every point of one kind, in index order.

Source code in src/py1815/profile/model.py 
175
176
177
def of(self, kind: Kind) -> Iterator[Point]:
    """Every point of one kind, in index order."""
    return iter(sorted((p for p in self.points.values() if p.kind is kind), key=_by_index))

py1815.profile.load

Load the profile tables and resolve them into a map for one DER.

The tables are a JSON document py1815.profile.extract writes from the IEEE 1815.2 companion workbook. This library does not carry them (D36): they are read from wherever the caller keeps its own copy, and :func:default_tables says where that is unless told otherwise.

Resolution is the one place block arithmetic happens (D37). A point the tables address relative to an equipment block is repeated once per unit of that equipment, each at block start + unit * block length + offset; everything downstream sees absolute indices.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

default_tables 

default_tables() -> pathlib.Path

Where the tables are read from when no path is given.

The environment variable if it is set, otherwise a file under the user's home directory. Never a path inside the installed package: the tables are the caller's copy of a document this library may not redistribute, and a location inside the package is one a packaging step could sweep up.

Source code in src/py1815/profile/load.py 
59
60
61
62
63
64
65
66
67
68
69
70
def default_tables() -> pathlib.Path:
    """Where the tables are read from when no path is given.

    The environment variable if it is set, otherwise a file under the user's
    home directory. Never a path inside the installed package: the tables are
    the caller's copy of a document this library may not redistribute, and a
    location inside the package is one a packaging step could sweep up.
    """
    named = os.environ.get(TABLES_VARIABLE)
    if named:
        return pathlib.Path(named)
    return pathlib.Path.home() / ".py1815" / TABLES_NAME

read_tables 

read_tables(
    path: str | Path | None = None,
) -> dict[str, Any]

The tables document at path, or at the default location.

Source code in src/py1815/profile/load.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def read_tables(path: str | pathlib.Path | None = None) -> dict[str, Any]:
    """The tables document at *path*, or at the default location."""
    where = pathlib.Path(path) if path is not None else default_tables()
    if not where.is_file():
        raise MapError(
            f"no profile tables at {where}. They are generated from the IEEE 1815.2 "
            "companion workbook, which this library does not carry: run "
            "`py1815-der tables fetch`, or `py1815-der tables build <workbook.xlsx>` "
            "with a copy you have downloaded."
        )
    try:
        document = json.loads(where.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise MapError(f"{where} is not a profile tables document: {exc}") from exc
    if not isinstance(document, dict) or "points" not in document or "key" not in document:
        raise MapError(f"{where} is not a profile tables document")
    return document

load 

load(
    path: str | Path | None = None,
    composition: Composition | None = None,
) -> PointMap

Read the tables and resolve them for a DER of the given composition.

Source code in src/py1815/profile/load.py 
92
93
94
95
96
def load(
    path: str | pathlib.Path | None = None, composition: Composition | None = None
) -> PointMap:
    """Read the tables and resolve them for a DER of the given composition."""
    return resolve(read_tables(path), composition or Composition())

resolve 

resolve(
    document: dict[str, Any], composition: Composition
) -> PointMap

The tables as a flat map of absolute indices for one composition.

Source code in src/py1815/profile/load.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
def resolve(document: dict[str, Any], composition: Composition) -> PointMap:
    """The tables as a flat map of absolute indices for one composition."""
    key = document["key"]
    blocks = _block_starts(key, composition)
    advertisement = document.get("index_points") or {}
    first_advertised = int(advertisement.get("first", MAX_INDEX + 1))
    last_advertised = first_advertised + int(advertisement.get("count", 0)) - 1

    points: dict[Address, Point] = {}

    def add(point: Point, row: dict[str, Any]) -> None:
        if not 0 <= point.index <= MAX_INDEX:
            raise MapError(
                f"{point.kind.value} {row['name']!r} resolves to index {point.index}, "
                f"outside 0..{MAX_INDEX}"
            )
        if point.address in points:
            raise MapError(
                f"{point.kind.value}{point.index} is claimed twice: by "
                f"{points[point.address].name!r} and by {point.name!r}"
            )
        points[point.address] = point

    for kind in Kind:
        for row in document["points"].get(kind.value, []):
            if row.get("index") is not None:
                index = int(row["index"])
                advertised = kind is Kind.AI and first_advertised <= index <= last_advertised
                fixed = _advertised(row, key, composition) if advertised else _fixed_value(row)
                if advertised and fixed is None:
                    # Nothing to advertise for this block and kind, so the
                    # point is not served rather than served as a zero that
                    # would read as "starts at index 0".
                    continue
                add(
                    _point(
                        kind,
                        index,
                        row,
                        blocks=blocks,
                        unit=None,
                        fixed_value=fixed,
                        advertised=advertised,
                    ),
                    row,
                )
                continue
            block = blocks.get(row.get("block"))
            if block is None:
                raise MapError(f"{kind.value} {row['name']!r} names an unknown block")
            repeating = block["component"] is not None
            for position in range(block["units"]):
                index = (
                    block["start"] + position * block["lengths"][kind.value] + int(row["offset"])
                )
                add(
                    _point(
                        kind,
                        index,
                        row,
                        blocks=blocks,
                        unit=position + 1 if repeating else None,
                        fixed_value=None,
                    ),
                    row,
                )

    _pair_supports(points)
    return PointMap(
        points=points,
        composition=composition,
        edition=str(document.get("edition", "")),
        profile_version=(document.get("source") or {}).get("profile_version"),
    )

Binding

py1815.profile.binding

Binding: how a caller says where each point's value comes from.

A point is bound by callables, one per direction (D38). An input is bound to something that returns its value; an output is bound to something that accepts one. Everything between that and the wire -- scaling, flags, class 0, events, the echo of a control -- is the builder's, so a simulator and a data store bind the same way and neither touches an object encoder.

Values cross this interface in engineering units: watts, volts, percent, a boolean. The tables say how each travels, and that conversion is not the caller's to repeat.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

Quality

Bases: Enum

What a source knows about the value it is handing over.

Four states, because four have a representation on the wire. A value that is merely old has none -- static objects carry no timestamp -- so a source decides for itself when old becomes COMM_LOST.

Source code in src/py1815/profile/binding.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class Quality(Enum):
    """What a source knows about the value it is handing over.

    Four states, because four have a representation on the wire. A value
    that is merely old has none -- static objects carry no timestamp -- so a
    source decides for itself when old becomes ``COMM_LOST``.
    """

    GOOD = "good"
    #: The source cannot be reached; the value is the last one it gave.
    COMM_LOST = "comm-lost"
    #: Nothing has been read since this outstation started.
    NEVER_READ = "never-read"
    #: The value is there and is not in effect: it belongs to a function
    #: that is disabled. Reported with no flag set at all.
    OFFLINE = "offline"

Reading dataclass

One value from a source, with what the source knows about it.

Source code in src/py1815/profile/binding.py 
44
45
46
47
48
49
50
51
52
53
54
@dataclass(frozen=True)
class Reading:
    """One value from a source, with what the source knows about it."""

    value: float | bool
    quality: Quality = Quality.GOOD
    #: Measurement time in milliseconds since the Unix epoch, UTC. None means
    #: "now". Events from a ``GOOD`` reading use this time, shifted by any
    #: master time write. Events from any other quality use the outstation's
    #: clock.
    timestamp_ms: int | None = None

Output dataclass

One bound output.

Source code in src/py1815/profile/binding.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dataclass(frozen=True)
class Output:
    """One bound output."""

    #: Called on operate, never on select. None means the value is stored and
    #: reported back, which is all a plain setting needs.
    apply: Writer | None = None
    #: What the output's status, and the input that mirrors it, report before
    #: any write. None reports the point as never read.
    initial: float | bool | None = None
    #: Checked on select and again on operate, before ``apply``. Returns the
    #: status to refuse with, or None to allow. For a refusal that depends on
    #: state rather than on the value's range, which the tables already bound.
    check: Callable[[float], CommandStatus | None] | None = None
    #: Where the output's status comes from, when it is not simply the last
    #: value written. For an output that is a window onto something else: one
    #: of a multiplexed curve's points shows whichever curve is selected.
    status: Reader | None = None

Binding

Every point a caller serves, and how.

Built up by calls and handed to the builder, which reads it once. Binding the same address twice is refused: the second binding would silently replace the first, and the caller that wrote both believes both are live.

Source code in src/py1815/profile/binding.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class Binding:
    """Every point a caller serves, and how.

    Built up by calls and handed to the builder, which reads it once. Binding
    the same address twice is refused: the second binding would silently
    replace the first, and the caller that wrote both believes both are live.
    """

    def __init__(self) -> None:
        self.readers: dict[Address, Reader] = {}
        self.outputs: dict[Address, Output] = {}
        #: Event deadbands, in transmitted units. An analog input absent from
        #: here reports every change of its transmitted value, unless an
        #: event policy gives it a deadband.
        self.deadbands: dict[int, float] = {}

    def read(
        self, kind: Kind, index: int, reader: Reader, *, deadband: float | None = None
    ) -> None:
        """Bind an input (binary, analog or counter) to where its value comes from."""
        if kind.is_output:
            raise ValueError(f"{kind.value}{index} is an output; bind it with output()")
        self._claim((kind, index), self.readers)
        self.readers[(kind, index)] = reader
        if deadband is not None:
            if kind is not Kind.AI:
                raise ValueError("only an analog input has a deadband")
            self.deadbands[index] = deadband

    def output(
        self,
        kind: Kind,
        index: int,
        apply: Writer | None = None,
        *,
        initial: float | bool | None = None,
        check: Callable[[float], CommandStatus | None] | None = None,
        status: Reader | None = None,
    ) -> None:
        """Bind an output (binary or analog) to what a command on it does."""
        if not kind.is_output:
            raise ValueError(f"{kind.value}{index} is an input; bind it with read()")
        self._claim((kind, index), self.outputs)
        self.outputs[(kind, index)] = Output(
            apply=apply, initial=initial, check=check, status=status
        )

    @staticmethod
    def _claim(address: Address, table: Mapping[Address, object]) -> None:
        if address in table:
            raise ValueError(f"{address[0].value}{address[1]} is already bound")
read 
read(
    kind: Kind,
    index: int,
    reader: Reader,
    *,
    deadband: float | None = None,
) -> None

Bind an input (binary, analog or counter) to where its value comes from.

Source code in src/py1815/profile/binding.py 
102
103
104
105
106
107
108
109
110
111
112
113
def read(
    self, kind: Kind, index: int, reader: Reader, *, deadband: float | None = None
) -> None:
    """Bind an input (binary, analog or counter) to where its value comes from."""
    if kind.is_output:
        raise ValueError(f"{kind.value}{index} is an output; bind it with output()")
    self._claim((kind, index), self.readers)
    self.readers[(kind, index)] = reader
    if deadband is not None:
        if kind is not Kind.AI:
            raise ValueError("only an analog input has a deadband")
        self.deadbands[index] = deadband
output 
output(
    kind: Kind,
    index: int,
    apply: Writer | None = None,
    *,
    initial: float | bool | None = None,
    check: Callable[[float], CommandStatus | None]
    | None = None,
    status: Reader | None = None,
) -> None

Bind an output (binary or analog) to what a command on it does.

Source code in src/py1815/profile/binding.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def output(
    self,
    kind: Kind,
    index: int,
    apply: Writer | None = None,
    *,
    initial: float | bool | None = None,
    check: Callable[[float], CommandStatus | None] | None = None,
    status: Reader | None = None,
) -> None:
    """Bind an output (binary or analog) to what a command on it does."""
    if not kind.is_output:
        raise ValueError(f"{kind.value}{index} is an input; bind it with read()")
    self._claim((kind, index), self.outputs)
    self.outputs[(kind, index)] = Output(
        apply=apply, initial=initial, check=check, status=status
    )

Event policy

Which points report events, in which class, and past what deadband: data a deployment hands to the builder as event_policy. The guide's Setting the event policy has an example and the order in which a point's own rule, its kind's rule and the tables are consulted.

py1815.profile.policy

Event policy: which points report events, in which class, past what deadband.

The tables give every point a default event class, and a default is all it is. What a controlling station wants reported differs from one deployment to the next, so the choice is taken as data: a rule for each kind of point, and exceptions for the points named one at a time. A deployment edits that data and not the code that binds its points.

The policy is a mapping, or the dataclasses a mapping becomes. This module reads no file and defines no file format (D66): a caller that keeps its policy in JSON or YAML loads it with whatever it already uses and hands over the result.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

EventRule dataclass

What one point, or every point of one kind, is told about its events.

Each field left as None says nothing, and whatever stands behind the rule decides: behind a point's rule, the rule for its kind; behind that, the tables.

Source code in src/py1815/profile/policy.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@dataclass(frozen=True)
class EventRule:
    """What one point, or every point of one kind, is told about its events.

    Each field left as None says nothing, and whatever stands behind the rule
    decides: behind a point's rule, the rule for its kind; behind that, the
    tables.
    """

    #: The class the events are reported in: 1, 2 or 3.
    event_class: int | None = None
    #: False turns events off, and the point is then static data only. It
    #: stays in a class 0 response exactly as before. True turns them back on
    #: for one point of a kind whose rule turned them off.
    events: bool | None = None
    #: How far an analog input has to move before the change is an event, in
    #: engineering units. The builder converts it with the point's scaling.
    deadband: float | None = None

    def __post_init__(self) -> None:
        chosen = self.event_class
        # An int and not a bool: ``1.0 == 1`` and ``True == 1``, and either in
        # force would hand back something that is not the class it claims.
        if chosen is not None and (
            not isinstance(chosen, int) or isinstance(chosen, bool) or chosen not in (1, 2, 3)
        ):
            raise ValueError(f"an event class is 1, 2 or 3, not {chosen!r}")
        if self.events is not None and not isinstance(self.events, bool):
            raise ValueError(f"events is true or false, not {self.events!r}")
        if self.events is False and chosen is not None:
            raise ValueError(f"events are turned off and given class {chosen} in one rule")
        deadband = self.deadband
        if deadband is None:
            return
        if isinstance(deadband, bool) or not isinstance(deadband, (int, float)):
            raise ValueError(f"a deadband is a number, not {deadband!r}")
        if not math.isfinite(deadband) or deadband < 0:
            raise ValueError(f"a deadband is zero or more, not {deadband!r}")

    @property
    def decides_reporting(self) -> bool:
        """Whether the rule says anything about a point reporting at all."""
        return self.event_class is not None or self.events is not None

    @classmethod
    def from_mapping(cls, data: Mapping[str, Any]) -> EventRule:
        """A rule from its plain form: ``class``, ``events`` and ``deadband``."""
        if not isinstance(data, Mapping):
            raise ValueError(f"an event rule is a mapping, not {data!r}")
        stray = sorted(str(key) for key in data if key not in _RULE_KEYS)
        if stray:
            raise ValueError(
                f"an event rule holds {', '.join(_RULE_KEYS)}; it does not hold {', '.join(stray)}"
            )
        return cls(
            event_class=data.get("class"),
            events=data.get("events"),
            deadband=data.get("deadband"),
        )
decides_reporting property 
decides_reporting: bool

Whether the rule says anything about a point reporting at all.

from_mapping classmethod 
from_mapping(data: Mapping[str, Any]) -> EventRule

A rule from its plain form: class, events and deadband.

Source code in src/py1815/profile/policy.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@classmethod
def from_mapping(cls, data: Mapping[str, Any]) -> EventRule:
    """A rule from its plain form: ``class``, ``events`` and ``deadband``."""
    if not isinstance(data, Mapping):
        raise ValueError(f"an event rule is a mapping, not {data!r}")
    stray = sorted(str(key) for key in data if key not in _RULE_KEYS)
    if stray:
        raise ValueError(
            f"an event rule holds {', '.join(_RULE_KEYS)}; it does not hold {', '.join(stray)}"
        )
    return cls(
        event_class=data.get("class"),
        events=data.get("events"),
        deadband=data.get("deadband"),
    )

EventPolicy dataclass

A deployment's event reporting: a rule per kind, and exceptions per point.

Three kinds report events and may be named: binary inputs, analog inputs, and counters, whose rule governs the event each freeze logs. A deadband belongs to an analog input and to nothing else.

What can be checked without a map is checked here, so a policy that cannot be right is refused when it is made. Whether the points it names exist is the builder's to check, when it has the map beside it.

Source code in src/py1815/profile/policy.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@dataclass(frozen=True)
class EventPolicy:
    """A deployment's event reporting: a rule per kind, and exceptions per point.

    Three kinds report events and may be named: binary inputs, analog inputs,
    and counters, whose rule governs the event each freeze logs. A deadband
    belongs to an analog input and to nothing else.

    What can be checked without a map is checked here, so a policy that
    cannot be right is refused when it is made. Whether the points it names
    exist is the builder's to check, when it has the map beside it.
    """

    defaults: Mapping[Kind, EventRule] = field(default_factory=dict)
    points: Mapping[Address, EventRule] = field(default_factory=dict)

    def __post_init__(self) -> None:
        for kind, rule in self.defaults.items():
            _check(kind, rule, f"the default for {getattr(kind, 'value', kind)!s}")
        for address, rule in self.points.items():
            if not isinstance(address, tuple) or len(address) != 2:
                raise ValueError(f"a point is named by its kind and index, not {address!r}")
            kind, index = address
            if isinstance(index, bool) or not isinstance(index, int) or index < 0:
                raise ValueError(f"a point's index is zero or more, not {index!r}")
            _check(kind, rule, f"{getattr(kind, 'value', kind)!s}{index}")
        object.__setattr__(self, "defaults", MappingProxyType(dict(self.defaults)))
        object.__setattr__(self, "points", MappingProxyType(dict(self.points)))

    @classmethod
    def from_mapping(cls, data: Mapping[str, Any]) -> EventPolicy:
        """A policy from plain data, as a configuration file would hold it.

        ::

            {
                "defaults": {"AI": {"class": 2, "deadband": 0.5}},
                "points": {
                    "AI537": {"class": 1, "deadband": 500},
                    "BI12": {"events": False},
                },
            }

        ``defaults`` is keyed by kind (``BI``, ``AI`` or ``CTR``) and
        ``points`` by kind and index together. Either may be left out. A key
        this does not know is refused, since a misspelled one would otherwise
        be a rule that silently does nothing.
        """
        if not isinstance(data, Mapping):
            raise ValueError(f"an event policy is a mapping, not {data!r}")
        stray = sorted(str(key) for key in data if key not in _POLICY_KEYS)
        if stray:
            raise ValueError(
                f"an event policy holds {' and '.join(_POLICY_KEYS)}; "
                f"it does not hold {', '.join(stray)}"
            )
        defaults = _section(data, "defaults")
        points = _section(data, "points")
        return cls(
            defaults={_kind(key): EventRule.from_mapping(rule) for key, rule in defaults.items()},
            points={_address(key): EventRule.from_mapping(rule) for key, rule in points.items()},
        )
from_mapping classmethod 
from_mapping(data: Mapping[str, Any]) -> EventPolicy

A policy from plain data, as a configuration file would hold it.

::

{
    "defaults": {"AI": {"class": 2, "deadband": 0.5}},
    "points": {
        "AI537": {"class": 1, "deadband": 500},
        "BI12": {"events": False},
    },
}

defaults is keyed by kind (BI, AI or CTR) and points by kind and index together. Either may be left out. A key this does not know is refused, since a misspelled one would otherwise be a rule that silently does nothing.

Source code in src/py1815/profile/policy.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@classmethod
def from_mapping(cls, data: Mapping[str, Any]) -> EventPolicy:
    """A policy from plain data, as a configuration file would hold it.

    ::

        {
            "defaults": {"AI": {"class": 2, "deadband": 0.5}},
            "points": {
                "AI537": {"class": 1, "deadband": 500},
                "BI12": {"events": False},
            },
        }

    ``defaults`` is keyed by kind (``BI``, ``AI`` or ``CTR``) and
    ``points`` by kind and index together. Either may be left out. A key
    this does not know is refused, since a misspelled one would otherwise
    be a rule that silently does nothing.
    """
    if not isinstance(data, Mapping):
        raise ValueError(f"an event policy is a mapping, not {data!r}")
    stray = sorted(str(key) for key in data if key not in _POLICY_KEYS)
    if stray:
        raise ValueError(
            f"an event policy holds {' and '.join(_POLICY_KEYS)}; "
            f"it does not hold {', '.join(stray)}"
        )
    defaults = _section(data, "defaults")
    points = _section(data, "points")
    return cls(
        defaults={_kind(key): EventRule.from_mapping(rule) for key, rule in defaults.items()},
        points={_address(key): EventRule.from_mapping(rule) for key, rule in points.items()},
    )

Outstation

py1815.profile.outstation

The builder: a point map and a binding in, an IEEE 1815.2 outstation out.

:class:DerOutstation is everything a :class:~py1815.session.Session needs from the device side. It answers reads from the map, in the variations and class membership the profile selects; it accepts controls, scaled and range checked by the tables; it records events with the class each point is given; and it freezes the counters. The caller supplied values in engineering units through a :class:~py1815.profile.binding.Binding and never sees an object.

What the profile fixes, and this module therefore does not ask about:

  • Binary inputs, analog inputs, counters and frozen counters answer a class 0 read. Output status does not: it is read by naming its group.
  • The points that advertise block starting indices are served, and are left out of class 0.
  • A "supports" input is static only. It never produces an event.
  • A binary output behaves as latched whatever operation commanded it.
  • A freeze does not clear the counters it freezes.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

DerOutstation

An IEEE 1815.2 outstation's device side, built from a map and a binding.

Implements the read, control and freeze providers a session takes, so one object is handed to :class:~py1815.session.Session three times, or :meth:session does it.

Source code in src/py1815/profile/outstation.py 
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
class DerOutstation:
    """An IEEE 1815.2 outstation's device side, built from a map and a binding.

    Implements the read, control and freeze providers a session takes, so one
    object is handed to :class:`~py1815.session.Session` three times, or
    :meth:`session` does it.
    """

    def __init__(
        self,
        point_map: PointMap,
        binding: Binding,
        *,
        strict: bool = True,
        event_capacity: int = 2000,
        block_octets: int = 1024,
        clock_ms: Callable[[], int] = wall_clock_ms,
        level2: bool = False,
        disabled_offline: bool = True,
        event_policy: EventPolicy | Mapping[str, Any] | None = None,
        read_only: bool = False,
        read_only_status: CommandStatus = CommandStatus.NOT_AUTHORIZED,
    ) -> None:
        """
        Args:
            point_map: The profile's points, resolved for this DER.
            binding: Where each served point's value comes from.
            strict: Refuse to build an outstation that leaves a point the
                profile makes mandatory unserved. Turned off only to serve a
                deliberately partial map, which is then not a conformant one.
            event_capacity: Events each class holds before its oldest is lost.
            block_octets: The most octets one static block may occupy. A
                response is made of whole blocks, so this has to fit the
                fragment a master can receive.
            clock_ms: Wall-clock milliseconds, UTC. Event and freeze times are
                taken from it, corrected by whatever time a master has written.
            level2: Answer as a DNP3 Subset Level 2 outstation and nothing
                more. The DER profile is built on Level 2 and adds to it;
                IEEE 1815.2 has every such addition be one an outstation can
                turn off, and this turns them off. Frozen counters are
                reported without their time, analog output status in 16
                bits, a freeze buffers no event, and freeze-and-clear clears.
                A master may still name the fuller variations outright.
            disabled_offline: Report the inputs of a function that is
                disabled with the ONLINE flag clear, as IEEE 1815.2
                clause 6.1.1 requires: the value is still sent, marked
                as not in effect. A function is enabled when its enable
                output stands at on, by the output's ``status`` reader if
                it has one and by the last accepted write if not. Turn it
                off for a controlling station
                that discards any value not flagged ONLINE, and so
                could not check a setting before enabling its function.
            event_policy: Which points report events, in which class, and
                past what deadband, where that is not to be the tables'
                choice: an :class:`~py1815.profile.policy.EventPolicy`, or
                the plain mapping one is made from. A policy that names a
                point the map does not hold, or asks of a point what it
                cannot do, is refused here. None leaves every class to the
                tables and every deadband to the binding.
            read_only: Report and do not command (D65). Every control on a
                bound output is refused with ``read_only_status`` and nothing
                reaches the binding. For an outstation that is one of several
                interfaces onto a device, when another of them holds control.
                Settable afterwards through :attr:`read_only`.
            read_only_status: What a refused control is answered with. The
                default says this master may not command here. A caller that
                knows another master holds the point may prefer
                ``BLOCKED_OTHER_MASTER``.
        """
        if block_octets < 32:
            raise ValueError(f"block_octets is {block_octets}; too small to hold an object range")
        if read_only_status is CommandStatus.SUCCESS:
            raise ValueError("read_only_status is SUCCESS, which refuses nothing")
        self._map = point_map
        self._binding = binding
        self._read_only = read_only
        self._read_only_status = read_only_status
        #: The sessions :meth:`session` wired to this outstation, so a change of
        #: role can withdraw the selects they granted under the old one.
        self._sessions: weakref.WeakSet[Session] = weakref.WeakSet()
        self._block_octets = block_octets
        self._clock_ms = clock_ms
        self._level2 = level2
        self._defaults = dict(_DEFAULT_VARIATIONS)
        if level2:
            self._defaults.update(_LEVEL_2_VARIATIONS)
        #: What each counter read when it was last cleared, so that a
        #: counter bound to a running total can be cleared without the
        #: source being asked to forget.
        self._cleared_at: dict[int, int] = {}
        #: What a master's time write moved this outstation's clock by.
        self._time_offset_ms = 0
        self.events = EventBuffers(capacity=event_capacity, analog_latest_only=True)

        unknown = sorted(
            f"{kind.value}{index}"
            for kind, index in (*binding.readers, *binding.outputs)
            if (kind, index) not in point_map
        )
        if unknown:
            raise MapError(f"the binding names points the map does not hold: {', '.join(unknown)}")

        #: The last value each output accepted, in engineering units.
        self._state: dict[Address, float | bool] = {
            address: output.initial
            for address, output in binding.outputs.items()
            if output.initial is not None
        }
        #: How each served point comes to be served. Kept for the coverage
        #: report; nothing that answers a master reads it.
        self._origins: dict[Address, Source] = dict.fromkeys(binding.outputs, Source.BOUND)
        self._sources = self._resolve_sources()
        #: For each input of a function, the output that enables the function.
        self._gates: dict[Address, Address] = self._function_gates() if disabled_offline else {}
        self._served: dict[Kind, list[Point]] = {
            kind: [p for p in point_map.of(kind) if self._is_served(p)] for kind in Kind
        }
        self._by_index: dict[Kind, dict[int, Point]] = {
            kind: {p.index: p for p in points} for kind, points in self._served.items()
        }
        #: Each frozen counter's value and the moment it was frozen.
        self._frozen: dict[int, tuple[CounterPoint, int]] = {}
        self._primed = False
        if event_policy is None:
            event_policy = EventPolicy()
        elif not isinstance(event_policy, EventPolicy):
            event_policy = EventPolicy.from_mapping(event_policy)
        #: The event class each input reports in, and 0 for one that reports
        #: none. For a counter it is the class its freezes are logged in.
        self._classes: dict[Address, int] = {}
        #: Each event-reporting analog input's deadband, in transmitted units.
        #: One absent from here reports every change of its transmitted value.
        self._deadbands: dict[int, float] = {}
        self._apply_policy(event_policy)

        if strict:
            missing = sorted(
                (p for p in point_map.points.values() if p.mandatory and not self._is_served(p)),
                key=lambda p: (p.kind.value, p.index),
            )
            if missing:
                listed = ", ".join(f"{p.kind.value}{p.index}" for p in missing[:20])
                more = f" and {len(missing) - 20} more" if len(missing) > 20 else ""
                raise MapError(
                    f"{len(missing)} point(s) the profile makes mandatory are not bound: "
                    f"{listed}{more}"
                )

    # ------------------------------------------------------------ assembly

    def _resolve_sources(self) -> dict[Address, Reader]:
        """Where each input's value comes from, the caller's word first.

        A bound reader wins. Failing that, an input paired with a bound output
        reports what that output last accepted; a "supports" input reports
        whether its function's enable output is bound (D40); and a point the
        tables fix the value of reports that value.
        """
        sources: dict[Address, Reader] = {}
        mirrors: dict[Address, Address] = {}
        for address in self._binding.outputs:
            point = self._map.point(*address)
            if point.associated is not None and not point.associated[0].is_output:
                mirrors[point.associated] = address
        for point in self._map.points.values():
            if point.kind.is_output or point.associated is None:
                continue
            if point.associated in self._binding.outputs:
                mirrors.setdefault(point.address, point.associated)

        for point in self._map.points.values():
            if point.kind.is_output:
                continue
            address = point.address
            if address in self._binding.readers:
                sources[address] = self._binding.readers[address]
                self._origins[address] = Source.BOUND
            elif address in mirrors:
                sources[address] = self._mirror(mirrors[address])
                self._origins[address] = Source.MIRROR
            elif point.enabled_by is not None:
                sources[address] = _constant(point.enabled_by in self._binding.outputs)
                self._origins[address] = Source.SUPPORTS
            elif point.fixed_value is not None:
                sources[address] = _constant(point.fixed_value)
                self._origins[address] = Source.FIXED
        return sources

    def _function_gates(self) -> dict[Address, Address]:
        """Which enable output each function input answers to.

        A function is the points the tables give one purpose, under one
        heading, around an enable output that a supports input is paired
        with. Two of its inputs are left out, because they say something
        about the function that is true whether or not it is enabled: the
        input that says it is supported, and the one that says whether it
        is enabled.
        """
        functions: dict[tuple[str, str | None], Point] = {}
        exempt: set[Address] = set()
        for point in self._map.points.values():
            if point.enabled_by is None or point.enabled_by not in self._binding.outputs:
                continue
            enable = self._map.point(*point.enabled_by)
            if not enable.purpose:
                continue
            functions[(enable.purpose.casefold(), enable.section)] = enable
            exempt.add(point.address)
            if enable.associated is not None:
                exempt.add(enable.associated)
        gates: dict[Address, Address] = {}
        for point in self._map.points.values():
            if point.kind.is_output or not point.purpose or point.address in exempt:
                continue
            gate = functions.get((point.purpose.casefold(), point.section))
            if gate is not None and point.address in self._sources:
                gates[point.address] = gate.address
        return gates

    def _apply_policy(self, policy: EventPolicy) -> None:
        """Settle each input's event class and deadband, the policy's word first.

        For the class, a rule naming the point outranks the rule for its
        kind, which outranks the tables. A kind's rule moves the points that
        already report and leaves alone the ones the tables keep static: a
        "supports" input is not made to report by a rule for every binary
        input. Naming such a point is how it is given events.

        For the deadband, a rule naming the point outranks the deadband given
        when the point was bound, which outranks the rule for its kind. The
        two from the policy are in engineering units and are converted here;
        the binding's is in transmitted units already (D66).
        """
        unknown = sorted(_name(address) for address in policy.points if address not in self._map)
        if unknown:
            raise MapError(
                f"the event policy names points the map does not hold: {', '.join(unknown)}"
            )
        for point in self._map.points.values():
            if point.kind.is_output:
                continue
            default = policy.defaults.get(point.kind)
            own = policy.points.get(point.address)
            reported = _class_in_force(point, default, own)
            self._classes[point.address] = reported
            if point.kind is not Kind.AI:
                continue
            if not reported:
                if own is not None and own.deadband is not None:
                    raise MapError(
                        f"the event policy gives {_name(point.address)} a deadband, "
                        "and the point reports no events"
                    )
                continue
            if own is not None and own.deadband is not None:
                self._deadbands[point.index] = _transmitted(point, own.deadband)
            elif point.index in self._binding.deadbands:
                self._deadbands[point.index] = self._binding.deadbands[point.index]
            elif default is not None and default.deadband is not None:
                self._deadbands[point.index] = _transmitted(point, default.deadband)

    def _mirror(self, output: Address) -> Reader:
        # The input paired with an output says what the output stands at,
        # which is the same question its status answers. Asking it the same
        # way is what keeps the two from disagreeing when the device applied
        # something other than what was written (D65).
        return lambda: self._standing(output)

    def _standing(self, address: Address) -> Reading:
        """What an output stands at: the value in force, and how far to trust it.

        The binding's status reader when it has one, since only the device
        knows what it applied. Without one, the last write this outstation
        accepted. A read-only outstation accepts none, and one that became
        read-only stops vouching for a write it took earlier, because another
        interface may have changed the value since: it reports the initial
        value the binding gave, or that it has nothing to report.
        """
        output = self._binding.outputs[address]
        if output.status is not None:
            try:
                result = output.status()
            except Exception:  # pylint: disable=broad-exception-caught
                logger.exception("profile: status of %s%d failed", address[0].value, address[1])
                return Reading(0, Quality.COMM_LOST)
            return result if isinstance(result, Reading) else Reading(result)
        value = output.initial if self.read_only else self._state.get(address)
        if value is None:
            # Nothing has been written and no initial value was given, so
            # there is no value to report: say so rather than report zero.
            return Reading(0, Quality.NEVER_READ)
        return Reading(value)

    def _is_served(self, point: Point) -> bool:
        if point.kind.is_output:
            return point.address in self._binding.outputs
        return point.address in self._sources

    @property
    def read_only(self) -> bool:
        """Whether this outstation reports and refuses every control (D65)."""
        return self._read_only

    @read_only.setter
    def read_only(self, value: bool) -> None:
        if bool(value) == self._read_only:
            return
        self._read_only = bool(value)
        # The writes this outstation accepted are no longer its to vouch for:
        # while it only reported, another interface may have changed every one
        # of them. Without a status reader, an output reports nothing until it
        # is written again under the new role.
        self._state.clear()
        if self._read_only:
            # A select granted while it commanded is left armed, so the operate
            # it was granted for reaches the refusal every other control gets
            # and is answered with the read-only status. Withdrawing it here
            # would answer NO_SELECT, which tells a master to select again when
            # the truth is that this interface may no longer command.
            return
        # Given control back. A select from before it only reported is not
        # permission now: another interface held the device in between.
        for session in list(self._sessions):
            session.abandon_select()

    @property
    def point_map(self) -> PointMap:
        """The map this outstation was built from."""
        return self._map

    @property
    def level2(self) -> bool:
        """Whether this outstation is held to Subset Level 2."""
        return self._level2

    def default_variation(self, group: int) -> int:
        """The variation a static group is reported in when none is named."""
        return self._defaults[group]

    def served(self, kind: Kind) -> list[Point]:
        """The points of one kind this outstation serves, in index order."""
        return list(self._served[kind])

    def event_class(self, kind: Kind, index: int) -> int:
        """The class a point's events are reported in: 1, 2 or 3, or 0 for none.

        What is in force, after the event policy and the tables have both had
        their say. For a counter it is the class its freezes are logged in.
        """
        return self._classes.get(self._map.point(kind, index).address, 0)

    def deadband(self, index: int) -> float:
        """An analog input's event deadband in force, in transmitted units.

        Zero reports every change of the transmitted value, and is also the
        answer for a point that reports no events at all.
        """
        return self._deadbands.get(self._map.point(Kind.AI, index).index, 0.0)

    def value(self, kind: Kind, index: int) -> float | bool | None:
        """The last value an output accepted, or None if it has none yet."""
        return self._state.get((kind, index))

    def frozen(self, index: int) -> int | None:
        """Return what counter *index* held at its last freeze, or None if never frozen."""
        held = self._frozen.get(index)
        return None if held is None else held[0].value

    def coverage(self) -> Coverage:
        """Every point of the map against what this outstation serves.

        For each point: whether it is bound, served without a binding (and
        how), or absent, and for a served point the quality it has at this
        moment. Where a point's value comes from was settled when the
        outstation was built and is the same in every report. The quality is
        read now, by asking each source once, so a report is a snapshot.

        A report, and no part of what a master is answered with: it buffers
        no event, changes no output and is read by nothing here (D67).
        """
        entries: list[Entry] = []
        for kind in Kind:
            for point in self._map.of(kind):
                source = self._origins.get(point.address, Source.ABSENT)
                if source is Source.ABSENT:
                    entries.append(Entry(point, source))
                    continue
                quality, online = self._standing_on_the_wire(point)
                entries.append(Entry(point, source, quality, online))
        return Coverage(tuple(entries))

    def _standing_on_the_wire(self, point: Point) -> tuple[Quality, bool]:
        """A served point's quality now, and whether it goes out with ONLINE set.

        One question to the source, answered both ways. They differ for a
        value with no number: the source calls it good, and the wire sends
        zero with ONLINE clear and a reference error, as it does for any NaN.
        """
        if point.kind.is_output:
            value, flags = self._output_status(point)
            quality = _FLAG_QUALITY[flags]
            online = quality is Quality.GOOD
            if online and point.kind is Kind.AO and value is not None:
                online = not math.isnan(point.to_wire(float(value)))
            return quality, online
        reading = self._reading(point)
        online = reading.quality is Quality.GOOD
        if online and point.kind is Kind.AI:
            online = not math.isnan(point.to_wire(float(reading.value)))
        return reading.quality, online

    # ---------------------------------------------------------------- time

    def now_ms(self) -> int:
        """The time, as this outstation's clock currently has it."""
        return self._clock_ms() + self._time_offset_ms

    def set_time(self, timestamp_ms: int) -> None:
        """Adopt the time a master wrote. Later events and freezes carry it."""
        self._time_offset_ms = int(timestamp_ms) - self._clock_ms()

    # ------------------------------------------------------------- readings

    def _reading(self, point: Point) -> Reading:
        try:
            result = self._sources[point.address]()
        except Exception:  # pylint: disable=broad-exception-caught
            # A source that raises is a source that could not be read. One bad
            # point must not take the whole response with it, and the master
            # is told this value cannot be trusted.
            logger.exception("profile: reading %s%d failed", point.kind.value, point.index)
            return Reading(0, Quality.COMM_LOST)
        reading = result if isinstance(result, Reading) else Reading(result)
        gate = self._gates.get(point.address)
        if gate is not None and reading.quality is Quality.GOOD and not self._enabled(gate):
            # The function this input belongs to is disabled. The value is
            # sent as it stands and marked as not in effect.
            return Reading(reading.value, Quality.OFFLINE, reading.timestamp_ms)
        return reading

    def _enabled(self, gate: Address) -> bool:
        """Whether the function behind an enable output is enabled, as far as is known.

        What the enable output stands at, asked as its status and its mirror
        ask it, so the three cannot disagree (D65). A device that reports its
        own state is believed over the last write accepted here: it may have
        been enabled before this outstation started, or by another interface,
        or have turned the function off itself. A standing value that cannot
        be vouched for, because the reader failed or nothing has been written,
        is not taken as enabled: not knowing is not grounds to tell a master
        that a setting is in force.
        """
        standing = self._standing(gate)
        return standing.quality is Quality.GOOD and bool(standing.value)

    def _event_time(self, reading: Reading, now: int) -> int:
        """Return the timestamp for an event raised by this reading (D79).

        Use the reading's own timestamp when it has one and its quality is
        ``GOOD``, shifted by the offset a master's time write applied to this
        outstation's clock. Otherwise use ``now``. A reading that is not
        ``GOOD`` still holds the time of its last measurement, but the event
        is the quality change, and that is happening now.
        """
        if reading.timestamp_ms is None or reading.quality is not Quality.GOOD:
            return now
        return int(reading.timestamp_ms) + self._time_offset_ms

    def _binary(self, point: Point) -> BinaryPoint:
        return self._binary_of(self._reading(point))

    @staticmethod
    def _binary_of(reading: Reading) -> BinaryPoint:
        return BinaryPoint(bool(reading.value), _QUALITY_FLAGS[reading.quality])

    def _analog(self, point: Point) -> AnalogPoint:
        return self._analog_of(point, self._reading(point))

    @staticmethod
    def _analog_of(point: Point, reading: Reading) -> AnalogPoint:
        flags = _QUALITY_FLAGS[reading.quality]
        raw = point.to_wire(float(reading.value))
        if math.isfinite(raw):
            raw = float(round(raw))
            if not point.in_range(raw):
                flags |= AnalogQuality.OVER_RANGE
        return AnalogPoint(raw, flags)

    def _counter(self, point: Point) -> CounterPoint:
        reading = self._reading(point)
        flags = {
            Quality.GOOD: CounterQuality.ONLINE,
            Quality.COMM_LOST: CounterQuality.COMM_LOST,
            Quality.NEVER_READ: CounterQuality.RESTART,
            Quality.OFFLINE: 0,
        }[reading.quality]
        count = max(0, round(float(reading.value)))
        return CounterPoint(max(0, count - self._cleared_at.get(point.index, 0)), int(flags))

    # ---------------------------------------------------------------- reads

    def read(self, headers: Sequence[ObjectHeader]) -> bytes:
        """The objects a read names, as one body. See :meth:`read_blocks`."""
        return b"".join(self.read_blocks(headers))

    def read_blocks(self, headers: Sequence[ObjectHeader]) -> Sequence[bytes]:
        """The objects a read names, as the blocks a response may be split at."""
        blocks: list[bytes] = []
        for header in headers:
            if header.group == CLASS_GROUP and header.event_class == 0:
                for group in _CLASS_0_GROUPS:
                    points = [p for p in self._points(group) if p.in_class_0]
                    blocks += self._ranges(group, self._defaults[group], points)
                continue
            variation = self._variation(header)
            if header.qualifier in _INDEXED:
                blocks += self._indexed(
                    header.group, variation, self._named(header), header.qualifier
                )
            else:
                blocks += self._ranges(header.group, variation, self._selected(header))
        return blocks

    def _variation(self, header: ObjectHeader) -> int:
        served = _SERVED_VARIATIONS.get(header.group)
        if served is None:
            raise UnknownObject(f"group {header.group} is not served")
        if header.variation == 0:
            return self._defaults[header.group]
        widened = _WIDENED_VARIATIONS.get(header.group, {}).get(header.variation)
        if widened is not None:
            return widened
        if header.variation not in served:
            raise UnknownObject(f"group {header.group} variation {header.variation} is not offered")
        return header.variation

    def _named(self, header: ObjectHeader) -> list[Point]:
        """The served points a read names one index at a time, in the order named.

        Not required of an outstation at any subset level, and answered
        because a master picking a few scattered points has no better way to
        ask. Every index has to be a point this outstation serves. A range is
        allowed to cross a gap, since a master cannot know where the gaps
        are; an index is a statement that a point is there, and one that is
        not there is the master's error to hear about (D63).
        """
        points = self._points(header.group)
        if not points:
            raise UnknownObject(f"group {header.group} holds no points")
        if not header.indices:
            raise ParameterError(f"an indexed read of group {header.group} names no index")
        by_index = {point.index: point for point in points}
        absent = sorted({index for index in header.indices if index not in by_index})
        if absent:
            raise ParameterError(f"group {header.group} has no point at index {absent[0]}")
        return [by_index[index] for index in header.indices]

    def _points(self, group: int) -> list[Point]:
        """Return the served points a static group reports. Only counters that freeze are frozen."""
        points = self._served[_GROUP_KINDS[group]]
        if group == GROUP_FROZEN_COUNTER:
            return [p for p in points if p.frozen]
        return points

    def _selected(self, header: ObjectHeader) -> list[Point]:
        """The served points a static read header names."""
        points = self._points(header.group)
        if not points:
            # A kind this outstation has none of is, to a master, an object
            # it does not have. Answering with nothing and no indication
            # would read as a kind that exists and happens to be empty.
            raise UnknownObject(f"group {header.group} holds no points")
        if header.qualifier is QualifierCode.ALL_OBJECTS:
            return points
        if header.qualifier in _RANGES and header.start is not None and header.stop is not None:
            named = [p for p in points if header.start <= p.index <= header.stop]
            # A range that runs past the last point, or lands between
            # points, names something that is not there. The object is
            # one this outstation knows; it is the range that is wrong.
            if not named or header.stop > points[-1].index:
                raise ParameterError(
                    f"group {header.group} has no points across {header.start}..{header.stop}"
                )
            return named
        raise UnknownObject(
            f"qualifier 0x{int(header.qualifier):02X} does not select static group {header.group}"
        )

    def _encode(self, group: int, variation: int, point: Point) -> tuple[int, bytes] | None:
        """One point as one object and the variation it took, or None for nothing to report.

        The variation is the one asked for unless that one has no flag
        octet and the point has something to say in it. A variation
        without flags stands for a point that is online and nothing else;
        a point that is anything else is sent in the matching variation
        with flags, so that asking for the compact form never hides that
        a value is not to be trusted.
        """
        if group == GROUP_BINARY_INPUT:
            return variation, encode_binary(self._binary(point))
        if group == GROUP_ANALOG_INPUT:
            analog = self._analog(point)
            if analog.flags != AnalogQuality.ONLINE:
                variation = _FLAGGED_ANALOG.get(variation, variation)
            return variation, encode_analog(analog, AnalogVariation(variation))
        if group == GROUP_COUNTER:
            return variation, encode_counter(self._counter(point))
        if group == GROUP_FROZEN_COUNTER:
            frozen = self._frozen.get(point.index)
            if frozen is None:
                # Never frozen, so there is no frozen value; the counter is
                # absent from this group until its first freeze.
                return None
            if (
                variation == FrozenCounterVariation.INT32
                and frozen[0].flags != CounterQuality.ONLINE
            ):
                variation = int(FrozenCounterVariation.INT32_WITH_FLAG)
            timed = variation == FrozenCounterVariation.INT32_WITH_FLAG_AND_TIME
            return variation, encode_frozen_counter(
                frozen[0],
                variation=FrozenCounterVariation(variation),
                timestamp_ms=frozen[1] if timed else None,
            )
        value, flags = self._output_status(point)
        if group == GROUP_BINARY_OUTPUT_STATUS:
            return variation, encode_binary_output_status(state=bool(value), flags=flags)
        raw = point.to_wire(float(value or 0))
        if math.isfinite(raw):
            # Rounded to the integer the variation carries. A value with no
            # number is passed as it is, and the encoder sends it flagged
            # instead of the read failing on it.
            raw = float(round(raw))
        return variation, encode_analog_output_status(raw, variation, flags=flags)

    def _output_status(self, point: Point) -> tuple[float | bool | None, int]:
        """What an output currently stands at, and the flags that go with it."""
        reading = self._standing(point.address)
        if reading.quality is Quality.NEVER_READ:
            # Never written and given no initial value: nothing to report yet.
            return None, _QUALITY_FLAGS[Quality.NEVER_READ]
        return reading.value, _QUALITY_FLAGS[reading.quality]

    def _indexed(
        self, group: int, variation: int, points: Iterable[Point], qualifier: QualifierCode
    ) -> list[bytes]:
        """Points as index-prefixed objects, in the order given, split at the block budget.

        The counterpart of :meth:`_ranges` for a read that named indices. A
        block holds one variation, so a point that takes a different one from
        the point before it starts a new block, exactly as a range does.

        Where it differs is a point with nothing to report, which today is a
        frozen counter that has never been frozen. A range passes over it. A
        read that named it is refused, for the reason an index that is not a
        point is refused: leaving it out would answer with clean indications
        and fewer objects than were asked for (D63).
        """
        prefix = _INDEXED[qualifier]
        # Group, variation and qualifier, then a count as wide as an index.
        header = 3 + prefix
        blocks: list[bytes] = []
        current = variation
        items: list[tuple[int, bytes]] = []
        size = 0

        def flush() -> None:
            nonlocal items, size
            if items:
                blocks.append(indexed_block(group, current, items, qualifier=qualifier))
            items, size = [], 0

        for point in points:
            result = self._encode(group, variation, point)
            if result is None:
                raise ParameterError(f"group {group} has nothing to report at index {point.index}")
            taken, encoded = result
            cost = prefix + len(encoded)
            if items and (taken != current or size + cost + header > self._block_octets):
                flush()
            if not items:
                current = taken
            items.append((point.index, encoded))
            size += cost
        flush()
        return blocks

    def _ranges(self, group: int, variation: int, points: Iterable[Point]) -> list[bytes]:
        """Points as object ranges: split at index gaps and at the block budget.

        Never inside an object. A run of contiguous indices is one header and
        its objects until it would outgrow a block, and then it is two. A
        point that takes a different variation from its neighbors starts a
        run of its own, since a header names one variation.
        """
        blocks: list[bytes] = []
        start = previous = -1
        current = variation
        objects: list[bytes] = []
        size = 0

        def flush() -> None:
            nonlocal objects, size
            if objects:
                header = range_header(group, current, start=start, stop=previous)
                blocks.append(header + b"".join(objects))
            objects, size = [], 0

        for point in points:
            result = self._encode(group, variation, point)
            if result is None:
                continue
            taken, encoded = result
            # Seven octets is the widest range header: group, variation,
            # qualifier and two sixteen-bit indices.
            if objects and (
                point.index != previous + 1
                or taken != current
                or size + len(encoded) + 7 > self._block_octets
            ):
                flush()
            if not objects:
                start = point.index
                current = taken
            objects.append(encoded)
            size += len(encoded)
            previous = point.index
        flush()
        return blocks

    # --------------------------------------------------------------- events

    def poll(self) -> int:
        """Read every event-reporting input and buffer what changed.

        Called on whatever cadence the caller's sources move at, and again
        after a control is operated so the inputs that mirror it report the
        change at once. The first call only records where every point stands:
        a master learns initial values from its integrity poll, and a buffer
        of them would be a few hundred events saying nothing happened.

        Returns:
            How many events were buffered.
        """
        now = self.now_ms()
        buffered = 0
        for point in self._served[Kind.BI]:
            reported = self._classes[point.address]
            if not reported:
                continue
            reading = self._reading(point)
            binary = self._binary_of(reading)
            if not self._primed:
                self.events.prime_binary(point.index, binary)
            elif self.events.record_binary(
                point.index,
                binary,
                event_class=EventClass(reported),
                timestamp_ms=self._event_time(reading, now),
            ):
                buffered += 1
        for point in self._served[Kind.AI]:
            reported = self._classes[point.address]
            if not reported:
                continue
            reading = self._reading(point)
            analog = self._analog_of(point, reading)
            if not self._primed:
                self.events.prime_analog(point.index, analog)
            elif self.events.record_analog(
                point.index,
                analog,
                event_class=EventClass(reported),
                deadband=self._deadbands.get(point.index, 0.0),
                timestamp_ms=self._event_time(reading, now),
            ):
                buffered += 1
        self._primed = True
        return buffered

    # -------------------------------------------------------------- freezes

    def freeze(self, headers: Sequence[ObjectHeader], *, clear: bool) -> None:
        """Freeze the counters a master named.

        Under the DER profile the counters are not cleared, whichever freeze
        was asked for: the profile has a controlling station compute an
        interval's energy by subtracting two frozen values, which a counter
        that restarted from zero would break. Held to Subset Level 2, a
        freeze-and-clear clears, as the base standard has it.
        """
        named: list[Point] = []
        for header in headers:
            if header.group != GROUP_COUNTER:
                raise UnknownObject(f"group {header.group} cannot be frozen")
            named += self._selected(header)
        if clear and self._level2:
            self._freeze(named, clear=True)
            return
        if clear:
            logger.info("profile: freeze-and-clear froze %d counter(s), clearing none", len(named))
        self._freeze(named)

    def freeze_all(self) -> int:
        """Freeze every counter at the same moment. Returns how many froze."""
        return self._freeze(self._served[Kind.CTR])

    def _freeze(self, points: Iterable[Point], *, clear: bool = False) -> int:
        now = self.now_ms()
        frozen = 0
        for point in points:
            if not point.frozen:
                continue
            value = self._counter(point)
            self._frozen[point.index] = (value, now)
            if clear:
                self._cleared_at[point.index] = self._cleared_at.get(point.index, 0) + value.value
            frozen += 1
            reported = self._classes[point.address]
            if self._level2 or not reported:
                # Frozen counter events are not Level 2 objects, and a policy
                # may turn a counter's off. The counter is frozen either way.
                continue
            self.events.record_frozen_counter(
                point.index, value, event_class=EventClass(reported), timestamp_ms=now
            )
        return frozen

    # ------------------------------------------------------------- controls

    def select(self, controls: Sequence[Control]) -> Sequence[CommandStatus]:
        """Whether each control could be operated. Nothing is executed."""
        return [self._command(control, execute=False) for control in controls]

    def operate(self, controls: Sequence[Control]) -> Sequence[CommandStatus]:
        """Execute each control through its binding, and say how each went."""
        statuses = [self._command(control, execute=True) for control in controls]
        if self._primed and CommandStatus.SUCCESS in statuses:
            # So the inputs mirroring what just changed report it in the
            # indications of this very response, not on the next cadence.
            self.poll()
        return statuses

    def _command(self, control: Control, *, execute: bool) -> CommandStatus:
        if control.group == GROUP_BINARY_OUTPUT_COMMAND:
            kind = Kind.BO
        elif control.group == GROUP_ANALOG_OUTPUT_COMMAND:
            kind = Kind.AO
        else:
            return CommandStatus.NOT_SUPPORTED
        address = (kind, control.index)
        output = self._binding.outputs.get(address)
        point = self._map.get(kind, control.index)
        if output is None or point is None:
            # No such output here. Per point rather than for the request: the
            # controls beside it may be perfectly good.
            return CommandStatus.NOT_SUPPORTED
        if self.read_only:
            # Before the value is looked at and before the binding is asked
            # anything: a refusal that depended on what was requested would
            # tell a master more about a device it may not command than "no".
            return self._read_only_status

        value: float | bool
        if isinstance(control.command, ControlRelayOutputBlock):
            state = _latched(control.command)
            if state is None:
                return CommandStatus.NOT_SUPPORTED
            value = state
        else:
            status, value = _setpoint(control.command, point)
            if status is not CommandStatus.SUCCESS:
                return status
        return self._apply(address, output, value, execute=execute)

    def _apply(
        self, address: Address, output: Output, value: float | bool, *, execute: bool
    ) -> CommandStatus:
        if output.check is not None:
            refusal = output.check(value)
            if refusal is not None and refusal is not CommandStatus.SUCCESS:
                return refusal
        if not execute:
            return CommandStatus.SUCCESS
        answer: Any = output.apply(value) if output.apply is not None else None
        if answer is None or answer is CommandStatus.SUCCESS:
            self._state[address] = value
            return CommandStatus.SUCCESS
        return CommandStatus(answer)

    # -------------------------------------------------------------- session

    def session(self, *, max_response: int = 2048, **options: Any) -> Session:
        """A session wired to this outstation, with the profile's options set.

        Analog events travel in the 32-bit variation without time, which is
        the profile's selection and the one a Level 2 master is certain to
        parse; the time is asked for from the first response; and the freeze
        functions are served only when there are counters to freeze. The
        first two are defaults a caller may override by naming the option.

        Every other option of :class:`~py1815.session.Session` passes
        through. ``unsolicited=True`` is the one a profile's consumers ask
        about: the session then reports the events :meth:`poll` buffers to a
        master that has enabled their class, without waiting to be polled.
        It is off unless named, and nothing about the map or the binding
        changes with it.
        """
        if self._block_octets + RESPONSE_HEADER_SIZE > max_response:
            raise ValueError(
                f"static blocks of up to {self._block_octets} octets do not fit a "
                f"{max_response}-octet response"
            )
        options.setdefault("need_time", True)
        options.setdefault("analog_event_variation", AnalogEventVariation.INT32)
        session = Session(
            self,
            control_provider=self if self._binding.outputs else None,
            events=self.events,
            # A freeze is offered only where a counter freezes. One accepted by a device
            # whose counters do not would store nothing a read can return (D100).
            freeze_provider=self if any(p.frozen for p in self._served[Kind.CTR]) else None,
            time_sink=self.set_time,
            max_response=max_response,
            **options,
        )
        self._sessions.add(session)
        return session
read_only property writable 
read_only: bool

Whether this outstation reports and refuses every control (D65).

point_map property 
point_map: PointMap

The map this outstation was built from.

level2 property 
level2: bool

Whether this outstation is held to Subset Level 2.

default_variation 
default_variation(group: int) -> int

The variation a static group is reported in when none is named.

Source code in src/py1815/profile/outstation.py 
498
499
500
def default_variation(self, group: int) -> int:
    """The variation a static group is reported in when none is named."""
    return self._defaults[group]
served 
served(kind: Kind) -> list[Point]

The points of one kind this outstation serves, in index order.

Source code in src/py1815/profile/outstation.py 
502
503
504
def served(self, kind: Kind) -> list[Point]:
    """The points of one kind this outstation serves, in index order."""
    return list(self._served[kind])
event_class 
event_class(kind: Kind, index: int) -> int

The class a point's events are reported in: 1, 2 or 3, or 0 for none.

What is in force, after the event policy and the tables have both had their say. For a counter it is the class its freezes are logged in.

Source code in src/py1815/profile/outstation.py 
506
507
508
509
510
511
512
def event_class(self, kind: Kind, index: int) -> int:
    """The class a point's events are reported in: 1, 2 or 3, or 0 for none.

    What is in force, after the event policy and the tables have both had
    their say. For a counter it is the class its freezes are logged in.
    """
    return self._classes.get(self._map.point(kind, index).address, 0)
deadband 
deadband(index: int) -> float

An analog input's event deadband in force, in transmitted units.

Zero reports every change of the transmitted value, and is also the answer for a point that reports no events at all.

Source code in src/py1815/profile/outstation.py 
514
515
516
517
518
519
520
def deadband(self, index: int) -> float:
    """An analog input's event deadband in force, in transmitted units.

    Zero reports every change of the transmitted value, and is also the
    answer for a point that reports no events at all.
    """
    return self._deadbands.get(self._map.point(Kind.AI, index).index, 0.0)
value 
value(kind: Kind, index: int) -> float | bool | None

The last value an output accepted, or None if it has none yet.

Source code in src/py1815/profile/outstation.py 
522
523
524
def value(self, kind: Kind, index: int) -> float | bool | None:
    """The last value an output accepted, or None if it has none yet."""
    return self._state.get((kind, index))
frozen 
frozen(index: int) -> int | None

Return what counter index held at its last freeze, or None if never frozen.

Source code in src/py1815/profile/outstation.py 
526
527
528
529
def frozen(self, index: int) -> int | None:
    """Return what counter *index* held at its last freeze, or None if never frozen."""
    held = self._frozen.get(index)
    return None if held is None else held[0].value
coverage 
coverage() -> Coverage

Every point of the map against what this outstation serves.

For each point: whether it is bound, served without a binding (and how), or absent, and for a served point the quality it has at this moment. Where a point's value comes from was settled when the outstation was built and is the same in every report. The quality is read now, by asking each source once, so a report is a snapshot.

A report, and no part of what a master is answered with: it buffers no event, changes no output and is read by nothing here (D67).

Source code in src/py1815/profile/outstation.py 
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
def coverage(self) -> Coverage:
    """Every point of the map against what this outstation serves.

    For each point: whether it is bound, served without a binding (and
    how), or absent, and for a served point the quality it has at this
    moment. Where a point's value comes from was settled when the
    outstation was built and is the same in every report. The quality is
    read now, by asking each source once, so a report is a snapshot.

    A report, and no part of what a master is answered with: it buffers
    no event, changes no output and is read by nothing here (D67).
    """
    entries: list[Entry] = []
    for kind in Kind:
        for point in self._map.of(kind):
            source = self._origins.get(point.address, Source.ABSENT)
            if source is Source.ABSENT:
                entries.append(Entry(point, source))
                continue
            quality, online = self._standing_on_the_wire(point)
            entries.append(Entry(point, source, quality, online))
    return Coverage(tuple(entries))
now_ms 
now_ms() -> int

The time, as this outstation's clock currently has it.

Source code in src/py1815/profile/outstation.py 
576
577
578
def now_ms(self) -> int:
    """The time, as this outstation's clock currently has it."""
    return self._clock_ms() + self._time_offset_ms
set_time 
set_time(timestamp_ms: int) -> None

Adopt the time a master wrote. Later events and freezes carry it.

Source code in src/py1815/profile/outstation.py 
580
581
582
def set_time(self, timestamp_ms: int) -> None:
    """Adopt the time a master wrote. Later events and freezes carry it."""
    self._time_offset_ms = int(timestamp_ms) - self._clock_ms()
read 
read(headers: Sequence[ObjectHeader]) -> bytes

The objects a read names, as one body. See :meth:read_blocks.

Source code in src/py1815/profile/outstation.py 
664
665
666
def read(self, headers: Sequence[ObjectHeader]) -> bytes:
    """The objects a read names, as one body. See :meth:`read_blocks`."""
    return b"".join(self.read_blocks(headers))
read_blocks 
read_blocks(
    headers: Sequence[ObjectHeader],
) -> Sequence[bytes]

The objects a read names, as the blocks a response may be split at.

Source code in src/py1815/profile/outstation.py 
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
def read_blocks(self, headers: Sequence[ObjectHeader]) -> Sequence[bytes]:
    """The objects a read names, as the blocks a response may be split at."""
    blocks: list[bytes] = []
    for header in headers:
        if header.group == CLASS_GROUP and header.event_class == 0:
            for group in _CLASS_0_GROUPS:
                points = [p for p in self._points(group) if p.in_class_0]
                blocks += self._ranges(group, self._defaults[group], points)
            continue
        variation = self._variation(header)
        if header.qualifier in _INDEXED:
            blocks += self._indexed(
                header.group, variation, self._named(header), header.qualifier
            )
        else:
            blocks += self._ranges(header.group, variation, self._selected(header))
    return blocks
poll 
poll() -> int

Read every event-reporting input and buffer what changed.

Called on whatever cadence the caller's sources move at, and again after a control is operated so the inputs that mirror it report the change at once. The first call only records where every point stands: a master learns initial values from its integrity poll, and a buffer of them would be a few hundred events saying nothing happened.

Returns:

Type Description
int

How many events were buffered.
Source code in src/py1815/profile/outstation.py 
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
def poll(self) -> int:
    """Read every event-reporting input and buffer what changed.

    Called on whatever cadence the caller's sources move at, and again
    after a control is operated so the inputs that mirror it report the
    change at once. The first call only records where every point stands:
    a master learns initial values from its integrity poll, and a buffer
    of them would be a few hundred events saying nothing happened.

    Returns:
        How many events were buffered.
    """
    now = self.now_ms()
    buffered = 0
    for point in self._served[Kind.BI]:
        reported = self._classes[point.address]
        if not reported:
            continue
        reading = self._reading(point)
        binary = self._binary_of(reading)
        if not self._primed:
            self.events.prime_binary(point.index, binary)
        elif self.events.record_binary(
            point.index,
            binary,
            event_class=EventClass(reported),
            timestamp_ms=self._event_time(reading, now),
        ):
            buffered += 1
    for point in self._served[Kind.AI]:
        reported = self._classes[point.address]
        if not reported:
            continue
        reading = self._reading(point)
        analog = self._analog_of(point, reading)
        if not self._primed:
            self.events.prime_analog(point.index, analog)
        elif self.events.record_analog(
            point.index,
            analog,
            event_class=EventClass(reported),
            deadband=self._deadbands.get(point.index, 0.0),
            timestamp_ms=self._event_time(reading, now),
        ):
            buffered += 1
    self._primed = True
    return buffered
freeze 
freeze(
    headers: Sequence[ObjectHeader], *, clear: bool
) -> None

Freeze the counters a master named.

Under the DER profile the counters are not cleared, whichever freeze was asked for: the profile has a controlling station compute an interval's energy by subtracting two frozen values, which a counter that restarted from zero would break. Held to Subset Level 2, a freeze-and-clear clears, as the base standard has it.

Source code in src/py1815/profile/outstation.py 
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
def freeze(self, headers: Sequence[ObjectHeader], *, clear: bool) -> None:
    """Freeze the counters a master named.

    Under the DER profile the counters are not cleared, whichever freeze
    was asked for: the profile has a controlling station compute an
    interval's energy by subtracting two frozen values, which a counter
    that restarted from zero would break. Held to Subset Level 2, a
    freeze-and-clear clears, as the base standard has it.
    """
    named: list[Point] = []
    for header in headers:
        if header.group != GROUP_COUNTER:
            raise UnknownObject(f"group {header.group} cannot be frozen")
        named += self._selected(header)
    if clear and self._level2:
        self._freeze(named, clear=True)
        return
    if clear:
        logger.info("profile: freeze-and-clear froze %d counter(s), clearing none", len(named))
    self._freeze(named)
freeze_all 
freeze_all() -> int

Freeze every counter at the same moment. Returns how many froze.

Source code in src/py1815/profile/outstation.py 
966
967
968
def freeze_all(self) -> int:
    """Freeze every counter at the same moment. Returns how many froze."""
    return self._freeze(self._served[Kind.CTR])
select 
select(
    controls: Sequence[Control],
) -> Sequence[CommandStatus]

Whether each control could be operated. Nothing is executed.

Source code in src/py1815/profile/outstation.py 
993
994
995
def select(self, controls: Sequence[Control]) -> Sequence[CommandStatus]:
    """Whether each control could be operated. Nothing is executed."""
    return [self._command(control, execute=False) for control in controls]
operate 
operate(
    controls: Sequence[Control],
) -> Sequence[CommandStatus]

Execute each control through its binding, and say how each went.

Source code in src/py1815/profile/outstation.py 
 997
 998
 999
1000
1001
1002
1003
1004
def operate(self, controls: Sequence[Control]) -> Sequence[CommandStatus]:
    """Execute each control through its binding, and say how each went."""
    statuses = [self._command(control, execute=True) for control in controls]
    if self._primed and CommandStatus.SUCCESS in statuses:
        # So the inputs mirroring what just changed report it in the
        # indications of this very response, not on the next cadence.
        self.poll()
    return statuses
session 
session(
    *, max_response: int = 2048, **options: Any
) -> Session

A session wired to this outstation, with the profile's options set.

Analog events travel in the 32-bit variation without time, which is the profile's selection and the one a Level 2 master is certain to parse; the time is asked for from the first response; and the freeze functions are served only when there are counters to freeze. The first two are defaults a caller may override by naming the option.

Every other option of :class:~py1815.session.Session passes through. unsolicited=True is the one a profile's consumers ask about: the session then reports the events :meth:poll buffers to a master that has enabled their class, without waiting to be polled. It is off unless named, and nothing about the map or the binding changes with it.

Source code in src/py1815/profile/outstation.py 
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
def session(self, *, max_response: int = 2048, **options: Any) -> Session:
    """A session wired to this outstation, with the profile's options set.

    Analog events travel in the 32-bit variation without time, which is
    the profile's selection and the one a Level 2 master is certain to
    parse; the time is asked for from the first response; and the freeze
    functions are served only when there are counters to freeze. The
    first two are defaults a caller may override by naming the option.

    Every other option of :class:`~py1815.session.Session` passes
    through. ``unsolicited=True`` is the one a profile's consumers ask
    about: the session then reports the events :meth:`poll` buffers to a
    master that has enabled their class, without waiting to be polled.
    It is off unless named, and nothing about the map or the binding
    changes with it.
    """
    if self._block_octets + RESPONSE_HEADER_SIZE > max_response:
        raise ValueError(
            f"static blocks of up to {self._block_octets} octets do not fit a "
            f"{max_response}-octet response"
        )
    options.setdefault("need_time", True)
    options.setdefault("analog_event_variation", AnalogEventVariation.INT32)
    session = Session(
        self,
        control_provider=self if self._binding.outputs else None,
        events=self.events,
        # A freeze is offered only where a counter freezes. One accepted by a device
        # whose counters do not would store nothing a read can return (D100).
        freeze_provider=self if any(p.frozen for p in self._served[Kind.CTR]) else None,
        time_sink=self.set_time,
        max_response=max_response,
        **options,
    )
    self._sessions.add(session)
    return session

Configuration

py1815.profile.config

The DER outstation's configuration: every setting in one JSON document.

.. code-block:: json

{
  "bind": "0.0.0.0:20000",
  "outstation_address": 10,
  "unsolicited": true,
  "composition": {"meters": 1, "inverters": 2}
}

py1815-der config prints the complete document, with every default filled in, for editing. py1815-der run --config FILE loads one, and so do points and profile.

Unknown settings and values of the wrong type are errors that name where they are, so a typing mistake is reported instead of ignored.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

ConfigError

Bases: ValueError

A configuration that cannot be used. The message says where and why.

Source code in src/py1815/settings.py 
23
24
class ConfigError(ValueError):
    """A configuration that cannot be used. The message says where and why."""

Identity dataclass

What the DNP3 Device Profile document says about the device.

Source code in src/py1815/profile/config.py 
41
42
43
44
45
46
47
48
49
50
@dataclass(frozen=True)
class Identity:
    """What the DNP3 Device Profile document says about the device."""

    vendor: str = "Not stated"
    device: str = "py1815 simulated DER outstation"
    hardware_version: str = "Not applicable (software)"
    #: Empty means this library's version.
    software_version: str = ""
    author: str = "py1815"

DerConfig dataclass

The DER outstation's whole configuration.

Source code in src/py1815/profile/config.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@dataclass(frozen=True)
class DerConfig:
    """The DER outstation's whole configuration."""

    #: Address and port to listen on. The default is this machine only.
    bind: str = f"127.0.0.1:{DEFAULT_PORT}"
    #: Path of the profile tables file, or None for the usual locations.
    tables: str | None = None
    #: This outstation's DNP3 link address.
    outstation_address: int = 1024
    #: The master's DNP3 link address, or None to serve whichever master
    #: speaks first on a connection.
    master_address: int | None = 1
    #: Send unsolicited responses for the event classes a master enables.
    unsolicited: bool = False
    #: Refuse every control, so a master can read and cannot command.
    read_only: bool = False
    #: Answer as a DNP3 Subset Level 2 outstation, without the profile's
    #: additions.
    level2: bool = False
    #: Clear the ONLINE flag on the inputs of a disabled function
    #: (IEEE 1815.2 clause 6.1.1).
    disabled_offline: bool = True
    #: Events each class holds before the oldest is dropped.
    event_capacity: int = 2000
    #: Largest response fragment to send, in octets.
    max_response: int = 2048
    #: Seconds a select stays valid.
    select_timeout: float = DEFAULT_SELECT_TIMEOUT
    #: Seconds to wait for a master to confirm a response, or None to wait
    #: without limit.
    confirm_timeout: float | None = DEFAULT_CONFIRM_TIMEOUT
    #: Seconds of silence before a connection is closed.
    idle_timeout: float = IDLE_TIMEOUT
    #: Which points report events, in which class and past what deadband, in
    #: the form :meth:`~py1815.profile.policy.EventPolicy.from_mapping` takes.
    #: None uses the profile tables' choices.
    event_policy: Mapping[str, Any] | None = None
    #: How many of each repeating component the DER has.
    composition: Composition = field(default_factory=Composition)
    #: Seed for the simulation's noise.
    seed: int = 0
    #: Seconds between simulation steps.
    tick: float = 1.0
    identity: Identity = field(default_factory=Identity)

    @classmethod
    def from_mapping(cls, given: Mapping[str, Any]) -> DerConfig:
        """Build a configuration from a decoded JSON document.

        Raises :class:`~py1815.settings.ConfigError` for an unknown setting or
        a value of the wrong type.
        """
        top_level = [each.name for each in fields(cls) if each.name not in ("seed", "tick")] + [
            "simulation"
        ]
        settings.section(given, "configuration", top_level)
        values: dict[str, Any] = {}
        for key, value in given.items():
            if key == "bind":
                values[key] = settings.text(value, key)
            elif key == "tables":
                values[key] = None if value is None else settings.text(value, key)
            elif key == "outstation_address":
                values[key] = settings.integer(value, key, 0, settings.MAX_ADDRESS)
            elif key == "master_address":
                values[key] = (
                    None if value is None else settings.integer(value, key, 0, settings.MAX_ADDRESS)
                )
            elif key in ("unsolicited", "read_only", "level2", "disabled_offline"):
                values[key] = settings.boolean(value, key)
            elif key == "event_capacity":
                values[key] = settings.integer(value, key, 1, 1_000_000)
            elif key == "max_response":
                values[key] = settings.integer(value, key, 249, 65535)
            elif key in ("select_timeout", "idle_timeout"):
                values[key] = settings.seconds(value, key)
            elif key == "confirm_timeout":
                values[key] = None if value is None else settings.seconds(value, key)
            elif key == "event_policy":
                values[key] = cls._event_policy(value)
            elif key == "composition":
                counts = settings.section(value, key, _COMPOSITION)
                values[key] = Composition(
                    **{
                        name: settings.integer(count, f"{key}.{name}", 0, _MAX_COUNT)
                        for name, count in counts.items()
                    }
                )
            elif key == "simulation":
                simulation = settings.section(value, key, _SIMULATION)
                if "seed" in simulation:
                    values["seed"] = settings.integer(
                        simulation["seed"], "simulation.seed", 0, 2**32 - 1
                    )
                if "tick" in simulation:
                    values["tick"] = settings.seconds(simulation["tick"], "simulation.tick")
            else:
                identity = settings.section(value, key, _IDENTITY)
                for name, entry in identity.items():
                    if not isinstance(entry, str):
                        raise ConfigError(f"{key}.{name} must be a string")
                values[key] = Identity(**identity)
        config = cls(**values)
        if config.master_address == config.outstation_address:
            raise ConfigError("outstation_address and master_address must be different")
        if config.unsolicited and config.confirm_timeout is None:
            # The session refuses this combination: with no limit on the wait
            # for a confirmation, one master that never confirms a read would
            # block unsolicited responses for good.
            raise ConfigError(
                "confirm_timeout must be a number of seconds when unsolicited is true"
            )
        return config

    @staticmethod
    def _event_policy(value: Any) -> dict[str, Any] | None:
        if value is None:
            return None
        if not isinstance(value, Mapping):
            raise ConfigError("event_policy must be an object or null")
        try:
            EventPolicy.from_mapping(value)
        except (ValueError, TypeError, KeyError) as error:
            raise ConfigError(f"event_policy: {error}") from None
        return dict(value)

    @classmethod
    def load(cls, path: str | pathlib.Path) -> DerConfig:
        """Read a configuration from a JSON file."""
        return cls.from_mapping(settings.read(path))

    def outstation_options(self) -> dict[str, Any]:
        """Return the keyword arguments for :class:`~py1815.profile.outstation.DerOutstation`."""
        return {
            "event_capacity": self.event_capacity,
            "level2": self.level2,
            "disabled_offline": self.disabled_offline,
            "read_only": self.read_only,
            "event_policy": self.event_policy,
        }

    def session_options(self) -> dict[str, Any]:
        """Return the keyword arguments for the outstation's session."""
        return {
            "outstation_address": self.outstation_address,
            "master_address": self.master_address,
            "unsolicited": self.unsolicited,
            "max_response": self.max_response,
            "select_timeout": self.select_timeout,
            "confirm_timeout": self.confirm_timeout,
        }

    def describe(self) -> dict[str, Any]:
        """Return the complete configuration as a JSON-compatible dict."""
        described: dict[str, Any] = {}
        for each in fields(self):
            value = getattr(self, each.name)
            if each.name in ("seed", "tick"):
                continue
            if each.name == "composition":
                value = {name: getattr(value, name) for name in _COMPOSITION}
                described[each.name] = value
                described["simulation"] = {"seed": self.seed, "tick": self.tick}
                continue
            if each.name == "identity":
                value = {name: getattr(value, name) for name in _IDENTITY}
            elif each.name == "event_policy" and value is not None:
                value = dict(value)
            described[each.name] = value
        return described

    def render(self) -> str:
        """Return the complete configuration as formatted JSON."""
        return settings.render(self.describe())
from_mapping classmethod 
from_mapping(given: Mapping[str, Any]) -> DerConfig

Build a configuration from a decoded JSON document.

Raises :class:~py1815.settings.ConfigError for an unknown setting or a value of the wrong type.

Source code in src/py1815/profile/config.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@classmethod
def from_mapping(cls, given: Mapping[str, Any]) -> DerConfig:
    """Build a configuration from a decoded JSON document.

    Raises :class:`~py1815.settings.ConfigError` for an unknown setting or
    a value of the wrong type.
    """
    top_level = [each.name for each in fields(cls) if each.name not in ("seed", "tick")] + [
        "simulation"
    ]
    settings.section(given, "configuration", top_level)
    values: dict[str, Any] = {}
    for key, value in given.items():
        if key == "bind":
            values[key] = settings.text(value, key)
        elif key == "tables":
            values[key] = None if value is None else settings.text(value, key)
        elif key == "outstation_address":
            values[key] = settings.integer(value, key, 0, settings.MAX_ADDRESS)
        elif key == "master_address":
            values[key] = (
                None if value is None else settings.integer(value, key, 0, settings.MAX_ADDRESS)
            )
        elif key in ("unsolicited", "read_only", "level2", "disabled_offline"):
            values[key] = settings.boolean(value, key)
        elif key == "event_capacity":
            values[key] = settings.integer(value, key, 1, 1_000_000)
        elif key == "max_response":
            values[key] = settings.integer(value, key, 249, 65535)
        elif key in ("select_timeout", "idle_timeout"):
            values[key] = settings.seconds(value, key)
        elif key == "confirm_timeout":
            values[key] = None if value is None else settings.seconds(value, key)
        elif key == "event_policy":
            values[key] = cls._event_policy(value)
        elif key == "composition":
            counts = settings.section(value, key, _COMPOSITION)
            values[key] = Composition(
                **{
                    name: settings.integer(count, f"{key}.{name}", 0, _MAX_COUNT)
                    for name, count in counts.items()
                }
            )
        elif key == "simulation":
            simulation = settings.section(value, key, _SIMULATION)
            if "seed" in simulation:
                values["seed"] = settings.integer(
                    simulation["seed"], "simulation.seed", 0, 2**32 - 1
                )
            if "tick" in simulation:
                values["tick"] = settings.seconds(simulation["tick"], "simulation.tick")
        else:
            identity = settings.section(value, key, _IDENTITY)
            for name, entry in identity.items():
                if not isinstance(entry, str):
                    raise ConfigError(f"{key}.{name} must be a string")
            values[key] = Identity(**identity)
    config = cls(**values)
    if config.master_address == config.outstation_address:
        raise ConfigError("outstation_address and master_address must be different")
    if config.unsolicited and config.confirm_timeout is None:
        # The session refuses this combination: with no limit on the wait
        # for a confirmation, one master that never confirms a read would
        # block unsolicited responses for good.
        raise ConfigError(
            "confirm_timeout must be a number of seconds when unsolicited is true"
        )
    return config
load classmethod 
load(path: str | Path) -> DerConfig

Read a configuration from a JSON file.

Source code in src/py1815/profile/config.py 
183
184
185
186
@classmethod
def load(cls, path: str | pathlib.Path) -> DerConfig:
    """Read a configuration from a JSON file."""
    return cls.from_mapping(settings.read(path))
outstation_options 
outstation_options() -> dict[str, Any]

Return the keyword arguments for :class:~py1815.profile.outstation.DerOutstation.

Source code in src/py1815/profile/config.py 
188
189
190
191
192
193
194
195
196
def outstation_options(self) -> dict[str, Any]:
    """Return the keyword arguments for :class:`~py1815.profile.outstation.DerOutstation`."""
    return {
        "event_capacity": self.event_capacity,
        "level2": self.level2,
        "disabled_offline": self.disabled_offline,
        "read_only": self.read_only,
        "event_policy": self.event_policy,
    }
session_options 
session_options() -> dict[str, Any]

Return the keyword arguments for the outstation's session.

Source code in src/py1815/profile/config.py 
198
199
200
201
202
203
204
205
206
207
def session_options(self) -> dict[str, Any]:
    """Return the keyword arguments for the outstation's session."""
    return {
        "outstation_address": self.outstation_address,
        "master_address": self.master_address,
        "unsolicited": self.unsolicited,
        "max_response": self.max_response,
        "select_timeout": self.select_timeout,
        "confirm_timeout": self.confirm_timeout,
    }
describe 
describe() -> dict[str, Any]

Return the complete configuration as a JSON-compatible dict.

Source code in src/py1815/profile/config.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def describe(self) -> dict[str, Any]:
    """Return the complete configuration as a JSON-compatible dict."""
    described: dict[str, Any] = {}
    for each in fields(self):
        value = getattr(self, each.name)
        if each.name in ("seed", "tick"):
            continue
        if each.name == "composition":
            value = {name: getattr(value, name) for name in _COMPOSITION}
            described[each.name] = value
            described["simulation"] = {"seed": self.seed, "tick": self.tick}
            continue
        if each.name == "identity":
            value = {name: getattr(value, name) for name in _IDENTITY}
        elif each.name == "event_policy" and value is not None:
            value = dict(value)
        described[each.name] = value
    return described
render 
render() -> str

Return the complete configuration as formatted JSON.

Source code in src/py1815/profile/config.py 
228
229
230
def render(self) -> str:
    """Return the complete configuration as formatted JSON."""
    return settings.render(self.describe())

py1815.settings

Helpers for reading and validating JSON configuration files.

Shared by the master's configuration (:mod:py1815.master.config) and the DER outstation's (:mod:py1815.profile.config). Each check takes the value and its location in the document, and raises :class:ConfigError with a message that names the location.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

ConfigError

Bases: ValueError

A configuration that cannot be used. The message says where and why.

Source code in src/py1815/settings.py 
23
24
class ConfigError(ValueError):
    """A configuration that cannot be used. The message says where and why."""

seconds 

seconds(
    value: Any,
    where: str,
    *,
    minimum: float = 0.0,
    unit: str = "seconds",
) -> float

Return a number greater than minimum, of seconds unless unit says otherwise.

Source code in src/py1815/settings.py 
27
28
29
30
31
32
33
34
35
36
def seconds(value: Any, where: str, *, minimum: float = 0.0, unit: str = "seconds") -> float:
    """Return a number greater than ``minimum``, of seconds unless ``unit`` says otherwise."""
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ConfigError(f"{where} must be a number of {unit}")
    # JSON such as 1e309 decodes to infinity, which is not a usable value.
    if not math.isfinite(value):
        raise ConfigError(f"{where} must be a finite number of {unit}")
    if not value > minimum:
        raise ConfigError(f"{where} must be greater than {minimum:g}")
    return float(value)

integer 

integer(value: Any, where: str, low: int, high: int) -> int

Return a whole number from low to high.

Source code in src/py1815/settings.py 
39
40
41
42
43
def integer(value: Any, where: str, low: int, high: int) -> int:
    """Return a whole number from ``low`` to ``high``."""
    if isinstance(value, bool) or not isinstance(value, int) or not low <= value <= high:
        raise ConfigError(f"{where} must be a whole number from {low} to {high}")
    return value

boolean 

boolean(value: Any, where: str) -> bool

Return true or false.

Source code in src/py1815/settings.py 
46
47
48
49
50
def boolean(value: Any, where: str) -> bool:
    """Return true or false."""
    if not isinstance(value, bool):
        raise ConfigError(f"{where} must be true or false")
    return value

text 

text(value: Any, where: str) -> str

Return a non-empty string.

Source code in src/py1815/settings.py 
53
54
55
56
57
def text(value: Any, where: str) -> str:
    """Return a non-empty string."""
    if not isinstance(value, str) or not value:
        raise ConfigError(f"{where} must be a non-empty string")
    return value

section 

section(
    value: Any, where: str, allowed: Sequence[str]
) -> Mapping[str, Any]

Return a JSON object whose keys are all in allowed.

Source code in src/py1815/settings.py 
60
61
62
63
64
65
66
67
68
69
def section(value: Any, where: str, allowed: Sequence[str]) -> Mapping[str, Any]:
    """Return a JSON object whose keys are all in ``allowed``."""
    if not isinstance(value, Mapping):
        raise ConfigError(f"{where} must be an object")
    for key in value:
        if key not in allowed:
            raise ConfigError(
                f"{where}.{key} is not a setting; expected one of {', '.join(allowed)}"
            )
    return value

read 

read(path: str | Path) -> dict[str, Any]

Read a JSON configuration file and return the decoded document.

Source code in src/py1815/settings.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
def read(path: str | pathlib.Path) -> dict[str, Any]:
    """Read a JSON configuration file and return the decoded document."""
    try:
        content = pathlib.Path(path).read_text(encoding="utf-8")
    except OSError as error:
        raise ConfigError(f"{path}: {error.strerror or error}") from None
    try:
        document = json.loads(content)
    except ValueError as error:
        raise ConfigError(f"{path} is not valid JSON: {error}") from None
    if not isinstance(document, dict):
        raise ConfigError(f"{path} must hold a JSON object")
    return document

render 

render(document: Mapping[str, Any]) -> str

Return a configuration as formatted JSON.

Source code in src/py1815/settings.py 
87
88
89
def render(document: Mapping[str, Any]) -> str:
    """Return a configuration as formatted JSON."""
    return json.dumps(document, indent=2) + "\n"

Coverage

py1815.profile.coverage

Coverage: how much of the profile a built outstation serves.

A deployment that starts from a partial map needs to see what it has and what it has left. :meth:~py1815.profile.outstation.DerOutstation.coverage answers with one :class:Entry for every point of the resolved map: where the point's value comes from, or that it has none, and how it stood when the report was made. Mandatory points are told apart from optional ones, so the report also says how far the map is from one a strict build would accept.

It is a report and nothing else. No request is answered from it, and taking one buffers no event and changes no output (D67).

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

Source

Bases: Enum

Where a point's value comes from, which is fixed when the outstation is built.

The first four are the ways the builder serves a point, in the order it tries them for an input. An output has only the first: it is served when it is bound, and not otherwise.

Source code in src/py1815/profile/coverage.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class Source(Enum):
    """Where a point's value comes from, which is fixed when the outstation is built.

    The first four are the ways the builder serves a point, in the order it
    tries them for an input. An output has only the first: it is served when
    it is bound, and not otherwise.
    """

    #: The caller bound it: a reader for an input, an output for an output.
    BOUND = "bound"
    #: An input paired with a bound output, reporting what that output last
    #: accepted.
    MIRROR = "mirror"
    #: A "supports" input, reporting whether its function's enable output is
    #: bound. Served whether or not it is, since "not supported" is an answer.
    SUPPORTS = "supports"
    #: A value the tables themselves fix, such as the profile version or a
    #: block's starting index.
    FIXED = "fixed"
    #: Nothing serves it. A class 0 read does not carry it and a read of its
    #: index is refused.
    ABSENT = "absent"

Entry dataclass

One point of the map, and how this outstation stands on it.

Source code in src/py1815/profile/coverage.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@dataclass(frozen=True)
class Entry:
    """One point of the map, and how this outstation stands on it."""

    point: Point
    source: Source
    #: What the point's source said of it when the report was made, after the
    #: builder's own rules: a source that raised is ``COMM_LOST``, and an input
    #: of a disabled function is ``OFFLINE``. None for a point that is absent.
    quality: Quality | None = None
    #: Whether the point was being sent with its ONLINE flag set. Taken from
    #: what goes on the wire and not from the quality alone: a good source can
    #: hand over a value with no number, which is sent with ONLINE clear.
    online: bool = False

    @property
    def served(self) -> bool:
        """Whether a master can read the point at all."""
        return self.source is not Source.ABSENT

    @property
    def address(self) -> str:
        """The point as the tables and the command line write it: ``AI537``."""
        return f"{self.point.kind.value}{self.point.index}"
served property 
served: bool

Whether a master can read the point at all.

address property 
address: str

The point as the tables and the command line write it: AI537.

Coverage dataclass

Every point of the map against what an outstation serves.

Two reports compare equal when every point stands alike in both, so a deployment can keep one and see what a new binding changed.

Source code in src/py1815/profile/coverage.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
@dataclass(frozen=True)
class Coverage:
    """Every point of the map against what an outstation serves.

    Two reports compare equal when every point stands alike in both, so a
    deployment can keep one and see what a new binding changed.
    """

    #: One entry per point of the map, by kind and then by index.
    entries: tuple[Entry, ...]

    def entry(self, kind: Kind, index: int) -> Entry:
        """The entry for one point, or the error that says the map has none."""
        for found in self.entries:
            if found.point.address == (kind, index):
                return found
        raise MapError(f"the map holds no {kind.value}{index}")

    def of(self, *sources: Source) -> tuple[Entry, ...]:
        """The entries whose value comes from any of the sources named."""
        return tuple(entry for entry in self.entries if entry.source in sources)

    @property
    def served(self) -> tuple[Entry, ...]:
        """The points a master can read, however each is served."""
        return tuple(entry for entry in self.entries if entry.served)

    @property
    def absent(self) -> tuple[Entry, ...]:
        """The points nothing serves, mandatory and optional alike."""
        return self.of(Source.ABSENT)

    @property
    def offline(self) -> tuple[Entry, ...]:
        """The served points that were not online when the report was made.

        Each is on the wire, with its ONLINE flag clear: its source could not
        be reached, it has never been read or written, or it belongs to a
        function that is disabled. The entry's quality says which.
        """
        return tuple(entry for entry in self.entries if entry.served and not entry.online)

    @property
    def missing(self) -> tuple[Entry, ...]:
        """The mandatory points nothing serves: what a strict build refuses over."""
        return tuple(entry for entry in self.absent if entry.point.mandatory)

    @property
    def conformant(self) -> bool:
        """Whether every mandatory point is served, which is what ``strict`` requires."""
        return not self.missing

    def changed_since(self, earlier: Coverage) -> tuple[Entry, ...]:
        """The entries that differ from an earlier report of the same map.

        Each is returned as it stands now. A point the earlier report did not
        hold counts as changed.
        """
        before = {entry.point.address: entry for entry in earlier.entries}
        return tuple(entry for entry in self.entries if before.get(entry.point.address) != entry)

    def render(self) -> str:
        """The report as text: a summary, a count by source, and a line per point.

        A point's line holds its address, ``M`` if it is mandatory, where its
        value comes from, its quality, and its name. The lines do not depend
        on one another, so comparing two renderings line by line shows the
        points that moved.
        """
        mandatory = [entry for entry in self.entries if entry.point.mandatory]
        optional = [entry for entry in self.entries if not entry.point.mandatory]
        served = len(self.served)
        lines = [
            f"{served} of {len(self.entries)} points served, {len(self.absent)} absent",
            f"  {_served(mandatory)} of {len(mandatory)} mandatory points served",
            f"  {_served(optional)} of {len(optional)} optional points served",
            f"  {len(self.offline)} served point(s) offline when this was reported",
        ]
        if self.missing:
            named = ", ".join(entry.address for entry in self.missing[:_NAMED])
            more = len(self.missing) - _NAMED
            lines.append(
                "not conformant; mandatory points not served: "
                + named
                + (f" and {more} more" if more > 0 else "")
            )
        else:
            lines.append("conformant: every mandatory point is served")
        lines += ["", f"{'':<10}{'mandatory':>10}{'optional':>10}"]
        for source in Source:
            lines.append(
                f"{source.value:<10}"
                f"{sum(1 for entry in mandatory if entry.source is source):>10}"
                f"{sum(1 for entry in optional if entry.source is source):>10}"
            )
        lines.append("")
        for entry in self.entries:
            marker = "M" if entry.point.mandatory else " "
            quality = "-" if entry.quality is None else entry.quality.value
            lines.append(
                f"{entry.address:<8} {marker} {entry.source.value:<8} {quality:<10} "
                f"{entry.point.name}"
            )
        return "\n".join(lines)

    def __str__(self) -> str:
        return self.render()
served property 
served: tuple[Entry, ...]

The points a master can read, however each is served.

absent property 
absent: tuple[Entry, ...]

The points nothing serves, mandatory and optional alike.

offline property 
offline: tuple[Entry, ...]

The served points that were not online when the report was made.

Each is on the wire, with its ONLINE flag clear: its source could not be reached, it has never been read or written, or it belongs to a function that is disabled. The entry's quality says which.

missing property 
missing: tuple[Entry, ...]

The mandatory points nothing serves: what a strict build refuses over.

conformant property 
conformant: bool

Whether every mandatory point is served, which is what strict requires.

entry 
entry(kind: Kind, index: int) -> Entry

The entry for one point, or the error that says the map has none.

Source code in src/py1815/profile/coverage.py 
89
90
91
92
93
94
def entry(self, kind: Kind, index: int) -> Entry:
    """The entry for one point, or the error that says the map has none."""
    for found in self.entries:
        if found.point.address == (kind, index):
            return found
    raise MapError(f"the map holds no {kind.value}{index}")
of 
of(*sources: Source) -> tuple[Entry, ...]

The entries whose value comes from any of the sources named.

Source code in src/py1815/profile/coverage.py 
96
97
98
def of(self, *sources: Source) -> tuple[Entry, ...]:
    """The entries whose value comes from any of the sources named."""
    return tuple(entry for entry in self.entries if entry.source in sources)
changed_since 
changed_since(earlier: Coverage) -> tuple[Entry, ...]

The entries that differ from an earlier report of the same map.

Each is returned as it stands now. A point the earlier report did not hold counts as changed.

Source code in src/py1815/profile/coverage.py 
130
131
132
133
134
135
136
137
def changed_since(self, earlier: Coverage) -> tuple[Entry, ...]:
    """The entries that differ from an earlier report of the same map.

    Each is returned as it stands now. A point the earlier report did not
    hold counts as changed.
    """
    before = {entry.point.address: entry for entry in earlier.entries}
    return tuple(entry for entry in self.entries if before.get(entry.point.address) != entry)
render 
render() -> str

The report as text: a summary, a count by source, and a line per point.

A point's line holds its address, M if it is mandatory, where its value comes from, its quality, and its name. The lines do not depend on one another, so comparing two renderings line by line shows the points that moved.

Source code in src/py1815/profile/coverage.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def render(self) -> str:
    """The report as text: a summary, a count by source, and a line per point.

    A point's line holds its address, ``M`` if it is mandatory, where its
    value comes from, its quality, and its name. The lines do not depend
    on one another, so comparing two renderings line by line shows the
    points that moved.
    """
    mandatory = [entry for entry in self.entries if entry.point.mandatory]
    optional = [entry for entry in self.entries if not entry.point.mandatory]
    served = len(self.served)
    lines = [
        f"{served} of {len(self.entries)} points served, {len(self.absent)} absent",
        f"  {_served(mandatory)} of {len(mandatory)} mandatory points served",
        f"  {_served(optional)} of {len(optional)} optional points served",
        f"  {len(self.offline)} served point(s) offline when this was reported",
    ]
    if self.missing:
        named = ", ".join(entry.address for entry in self.missing[:_NAMED])
        more = len(self.missing) - _NAMED
        lines.append(
            "not conformant; mandatory points not served: "
            + named
            + (f" and {more} more" if more > 0 else "")
        )
    else:
        lines.append("conformant: every mandatory point is served")
    lines += ["", f"{'':<10}{'mandatory':>10}{'optional':>10}"]
    for source in Source:
        lines.append(
            f"{source.value:<10}"
            f"{sum(1 for entry in mandatory if entry.source is source):>10}"
            f"{sum(1 for entry in optional if entry.source is source):>10}"
        )
    lines.append("")
    for entry in self.entries:
        marker = "M" if entry.point.mandatory else " "
        quality = "-" if entry.quality is None else entry.quality.value
        lines.append(
            f"{entry.address:<8} {marker} {entry.source.value:<8} {quality:<10} "
            f"{entry.point.name}"
        )
    return "\n".join(lines)

Curves

py1815.profile.curves

The generic curves of IEEE 1815.2 clause 6.1.3, held behind one window.

A DER function that follows a curve does not get two hundred points of its own. The profile gives every curve the same block of points and a selector that says which curve the block is showing, and each function names the curve it uses by number. This module is the store behind that block: the curves, which one is selected, which functions refer to which, and the three rules the standard attaches to them.

  • A selector naming a curve that does not exist is refused.
  • A curve referred to by an enabled function is locked: its type, its units and its points cannot be changed until the function is disabled or pointed at another curve.
  • A function can only be pointed at a curve of a type it follows.

The standard recommends the status for the first two and leaves the third to the implementation, which here answers that the operation is not supported for that curve. Zero as a function's curve number means it uses none.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

Curve dataclass

One curve, as the integers a controlling station wrote.

The scaling of a point depends on the units the curve declares, so the values are kept as they travel and whoever follows the curve applies the scaling for the units it understands.

Source code in src/py1815/profile/curves.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@dataclass
class Curve:
    """One curve, as the integers a controlling station wrote.

    The scaling of a point depends on the units the curve declares, so the
    values are kept as they travel and whoever follows the curve applies the
    scaling for the units it understands.
    """

    number: int
    fields: list[float] = field(default_factory=lambda: [0.0] * FIELDS)
    values: list[float] = field(default_factory=lambda: [0.0] * (2 * MAX_POINTS))

    @property
    def type(self) -> int:
        """What kind of curve this is; zero for one that is not defined."""
        return round(self.fields[TYPE])

    @property
    def count(self) -> int:
        """How many of its points are in use."""
        return max(0, min(MAX_POINTS, round(self.fields[POINT_COUNT])))

    @property
    def x_units(self) -> int:
        """The units of the independent values."""
        return round(self.fields[X_UNITS])

    @property
    def y_units(self) -> int:
        """The units of the dependent values."""
        return round(self.fields[Y_UNITS])

    @property
    def points(self) -> list[tuple[float, float]]:
        """The points in use, as written, lowest numbered first."""
        return [(self.values[2 * n], self.values[2 * n + 1]) for n in range(self.count)]

    def at(self, x: float) -> float | None:
        """The curve's value at *x*, in the units its points were written in.

        Straight lines between points, and level beyond either end. None for
        a curve with no points. The points are followed in the order of their
        independent values, so a curve that doubles back to describe
        hysteresis is not followed as one.
        """
        points = sorted(self.points, key=lambda point: point[0])
        if not points:
            return None
        if x <= points[0][0]:
            return points[0][1]
        for (x0, y0), (x1, y1) in pairwise(points):
            if x <= x1:
                return y0 if x1 == x0 else y0 + (y1 - y0) * (x - x0) / (x1 - x0)
        return points[-1][1]
type property 
type: int

What kind of curve this is; zero for one that is not defined.

count property 
count: int

How many of its points are in use.

x_units property 
x_units: int

The units of the independent values.

y_units property 
y_units: int

The units of the dependent values.

points property 
points: list[tuple[float, float]]

The points in use, as written, lowest numbered first.

at 
at(x: float) -> float | None

The curve's value at x, in the units its points were written in.

Straight lines between points, and level beyond either end. None for a curve with no points. The points are followed in the order of their independent values, so a curve that doubles back to describe hysteresis is not followed as one.

Source code in src/py1815/profile/curves.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def at(self, x: float) -> float | None:
    """The curve's value at *x*, in the units its points were written in.

    Straight lines between points, and level beyond either end. None for
    a curve with no points. The points are followed in the order of their
    independent values, so a curve that doubles back to describe
    hysteresis is not followed as one.
    """
    points = sorted(self.points, key=lambda point: point[0])
    if not points:
        return None
    if x <= points[0][0]:
        return points[0][1]
    for (x0, y0), (x1, y1) in pairwise(points):
        if x <= x1:
            return y0 if x1 == x0 else y0 + (y1 - y0) * (x - x0) / (x1 - x0)
    return points[-1][1]

CurveStore

The curves a DER holds, and the window a controlling station edits them through.

Source code in src/py1815/profile/curves.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
class CurveStore:
    """The curves a DER holds, and the window a controlling station edits them through."""

    def __init__(self, count: int = 10) -> None:
        """
        Args:
            count: How many curves there are, numbered from one. The profile
                leaves the number to the DER.
        """
        if count < 1:
            raise ValueError("a curve store holds at least one curve")
        self.curves: dict[int, Curve] = {n: Curve(n) for n in range(1, count + 1)}
        self.selected = 1
        self._references: list[_Reference] = []

    # ----------------------------------------------------------------- state

    @property
    def current(self) -> Curve:
        """The curve the window is showing."""
        return self.curves[self.selected]

    def referenced(self, number: int) -> bool:
        """Whether any function names this curve as the one it uses."""
        return any(reference.curve == number for reference in self._references)

    def locked(self, number: int) -> bool:
        """Whether a function that is enabled names this curve."""
        return any(
            reference.curve == number and reference.enabled() for reference in self._references
        )

    # --------------------------------------------------------------- binding

    def bind(
        self,
        binding: Binding,
        *,
        output: int,
        readback: int,
        referenced: int,
        check: Check | None = None,
    ) -> None:
        """Bind the curve block: its selector, the fields, the points and the indicator.

        Args:
            binding: Where the points are bound.
            output: The analog output index of the selector. The fields and
                then the points follow it.
            readback: The analog input index the selector reads back at; the
                inputs are laid out as the outputs are.
            referenced: The binary input saying whether the selected curve is
                named by a function.
            check: A refusal that applies to every write here, such as a
                lockout, consulted before this store's own.
        """

        def guarded(own: Check) -> Check:
            def consult(value: float) -> CommandStatus | None:
                if check is not None:
                    answer = check(value)
                    if answer is not None:
                        return answer
                return own(value)

            return consult

        def exists(value: float) -> CommandStatus | None:
            return None if round(value) in self.curves else CommandStatus.OUT_OF_RANGE

        def select(value: float) -> CommandStatus | None:
            self.selected = round(value)
            return None

        binding.output(Kind.AO, output, select, check=guarded(exists), status=lambda: self.selected)
        binding.read(Kind.AI, readback, lambda: self.selected)
        binding.read(Kind.BI, referenced, lambda: self.referenced(self.selected))

        def unlocked(position: int) -> Check:
            def consult(value: float) -> CommandStatus | None:
                if self.locked(self.selected):
                    return CommandStatus.AUTOMATION_INHIBIT
                if position == TYPE and not self._usable(self.selected, round(value)):
                    return CommandStatus.NOT_SUPPORTED
                return None

            return consult

        def cell(position: int) -> tuple[Callable[[float], None], Reader]:
            if position < FIELDS:

                def write_field(value: float) -> None:
                    self.current.fields[position] = float(value)

                return write_field, lambda: self.current.fields[position]

            def write_value(value: float) -> None:
                self.current.values[position - FIELDS] = float(value)

            return write_value, lambda: self.current.values[position - FIELDS]

        for position in range(FIELDS + 2 * MAX_POINTS):
            write, read = cell(position)
            binding.output(
                Kind.AO,
                output + 1 + position,
                write,
                check=guarded(unlocked(position)),
                status=read,
            )
            binding.read(Kind.AI, readback + 1 + position, read)

    def reference(
        self,
        binding: Binding,
        index: int,
        *,
        types: Iterable[int],
        enabled: Callable[[], bool],
        check: Check | None = None,
    ) -> Callable[[], Curve | None]:
        """Bind a function's curve number, and return how to get the curve it names.

        Args:
            binding: Where the output is bound.
            index: The analog output that holds the function's curve number.
            types: The curve types the function follows.
            enabled: Whether the function is enabled now, which is what locks
                the curve it names.
            check: A refusal consulted before this store's own.

        Returns:
            A callable giving the curve the function names, or None while it
            names none.
        """
        reference = _Reference(types=frozenset(types), enabled=enabled)
        self._references.append(reference)

        def consult(value: float) -> CommandStatus | None:
            if check is not None:
                answer = check(value)
                if answer is not None:
                    return answer
            number = round(value)
            if number == 0:
                return None
            if number not in self.curves:
                return CommandStatus.OUT_OF_RANGE
            if self.curves[number].type not in reference.types:
                return CommandStatus.NOT_SUPPORTED
            return None

        def assign(value: float) -> CommandStatus | None:
            reference.curve = round(value)
            return None

        binding.output(Kind.AO, index, assign, initial=0, check=consult)
        return lambda: self.curves.get(reference.curve)

    def _usable(self, number: int, kind: int) -> bool:
        """Whether every function naming curve *number* could follow one of type *kind*."""
        return all(
            kind in reference.types for reference in self._references if reference.curve == number
        )
current property 
current: Curve

The curve the window is showing.

referenced 
referenced(number: int) -> bool

Whether any function names this curve as the one it uses.

Source code in src/py1815/profile/curves.py 
135
136
137
def referenced(self, number: int) -> bool:
    """Whether any function names this curve as the one it uses."""
    return any(reference.curve == number for reference in self._references)
locked 
locked(number: int) -> bool

Whether a function that is enabled names this curve.

Source code in src/py1815/profile/curves.py 
139
140
141
142
143
def locked(self, number: int) -> bool:
    """Whether a function that is enabled names this curve."""
    return any(
        reference.curve == number and reference.enabled() for reference in self._references
    )
bind 
bind(
    binding: Binding,
    *,
    output: int,
    readback: int,
    referenced: int,
    check: Check | None = None,
) -> None

Bind the curve block: its selector, the fields, the points and the indicator.

Parameters:

Name Type Description Default
binding Binding

Where the points are bound.

 required
output int

The analog output index of the selector. The fields and then the points follow it.

 required
readback int

The analog input index the selector reads back at; the inputs are laid out as the outputs are.

 required
referenced int

The binary input saying whether the selected curve is named by a function.

 required
check Check | None

A refusal that applies to every write here, such as a lockout, consulted before this store's own.

 None
Source code in src/py1815/profile/curves.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def bind(
    self,
    binding: Binding,
    *,
    output: int,
    readback: int,
    referenced: int,
    check: Check | None = None,
) -> None:
    """Bind the curve block: its selector, the fields, the points and the indicator.

    Args:
        binding: Where the points are bound.
        output: The analog output index of the selector. The fields and
            then the points follow it.
        readback: The analog input index the selector reads back at; the
            inputs are laid out as the outputs are.
        referenced: The binary input saying whether the selected curve is
            named by a function.
        check: A refusal that applies to every write here, such as a
            lockout, consulted before this store's own.
    """

    def guarded(own: Check) -> Check:
        def consult(value: float) -> CommandStatus | None:
            if check is not None:
                answer = check(value)
                if answer is not None:
                    return answer
            return own(value)

        return consult

    def exists(value: float) -> CommandStatus | None:
        return None if round(value) in self.curves else CommandStatus.OUT_OF_RANGE

    def select(value: float) -> CommandStatus | None:
        self.selected = round(value)
        return None

    binding.output(Kind.AO, output, select, check=guarded(exists), status=lambda: self.selected)
    binding.read(Kind.AI, readback, lambda: self.selected)
    binding.read(Kind.BI, referenced, lambda: self.referenced(self.selected))

    def unlocked(position: int) -> Check:
        def consult(value: float) -> CommandStatus | None:
            if self.locked(self.selected):
                return CommandStatus.AUTOMATION_INHIBIT
            if position == TYPE and not self._usable(self.selected, round(value)):
                return CommandStatus.NOT_SUPPORTED
            return None

        return consult

    def cell(position: int) -> tuple[Callable[[float], None], Reader]:
        if position < FIELDS:

            def write_field(value: float) -> None:
                self.current.fields[position] = float(value)

            return write_field, lambda: self.current.fields[position]

        def write_value(value: float) -> None:
            self.current.values[position - FIELDS] = float(value)

        return write_value, lambda: self.current.values[position - FIELDS]

    for position in range(FIELDS + 2 * MAX_POINTS):
        write, read = cell(position)
        binding.output(
            Kind.AO,
            output + 1 + position,
            write,
            check=guarded(unlocked(position)),
            status=read,
        )
        binding.read(Kind.AI, readback + 1 + position, read)
reference 
reference(
    binding: Binding,
    index: int,
    *,
    types: Iterable[int],
    enabled: Callable[[], bool],
    check: Check | None = None,
) -> Callable[[], Curve | None]

Bind a function's curve number, and return how to get the curve it names.

Parameters:

Name Type Description Default
binding Binding

Where the output is bound.

 required
index int

The analog output that holds the function's curve number.

 required
types Iterable[int]

The curve types the function follows.

 required
enabled Callable[[], bool]

Whether the function is enabled now, which is what locks the curve it names.

 required
check Check | None

A refusal consulted before this store's own.

 None

Returns:

Type Description
Callable[[], Curve | None]

A callable giving the curve the function names, or None while it
Callable[[], Curve | None]

names none.
Source code in src/py1815/profile/curves.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def reference(
    self,
    binding: Binding,
    index: int,
    *,
    types: Iterable[int],
    enabled: Callable[[], bool],
    check: Check | None = None,
) -> Callable[[], Curve | None]:
    """Bind a function's curve number, and return how to get the curve it names.

    Args:
        binding: Where the output is bound.
        index: The analog output that holds the function's curve number.
        types: The curve types the function follows.
        enabled: Whether the function is enabled now, which is what locks
            the curve it names.
        check: A refusal consulted before this store's own.

    Returns:
        A callable giving the curve the function names, or None while it
        names none.
    """
    reference = _Reference(types=frozenset(types), enabled=enabled)
    self._references.append(reference)

    def consult(value: float) -> CommandStatus | None:
        if check is not None:
            answer = check(value)
            if answer is not None:
                return answer
        number = round(value)
        if number == 0:
            return None
        if number not in self.curves:
            return CommandStatus.OUT_OF_RANGE
        if self.curves[number].type not in reference.types:
            return CommandStatus.NOT_SUPPORTED
        return None

    def assign(value: float) -> CommandStatus | None:
        reference.curve = round(value)
        return None

    binding.output(Kind.AO, index, assign, initial=0, check=consult)
    return lambda: self.curves.get(reference.curve)

Device Profile

py1815.profile.device_profile

Write the DNP3 Device Profile document for an outstation built here.

IEEE Std 1815 has every device publish a Device Profile: a statement of what it implements and how it is configured, in an XML form the DNP Users Group maintains a schema for. This module writes that document, for schema version 2.12.00, from a built :class:~py1815.profile.outstation.DerOutstation and the :class:~py1815.session.Session serving it.

Everything in it is read from the objects it describes. The point lists come from what the outstation serves, the limits and options from the session's own configuration, and the implementation table from the same facts the session answers by. Nothing is restated by hand, because a profile that can drift from its device is the usual kind, and the reason people distrust them.

What it does not claim. A profile has places for figures only a measurement can supply (clock drift, response time, timestamp error) and for a conformance test result. None has been measured or run for this library, so those elements are left out, which the schema permits. An absent element says "not stated"; a guessed one would say something false.

The schema and its rendering stylesheet are the DNP Users Group's and are not carried here. The document names them, as the standard has an instance do, and validates against a copy the caller holds.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

Identity dataclass

Who made the device and what it is called: the facts no object holds.

Source code in src/py1815/profile/device_profile.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@dataclass(frozen=True)
class Identity:
    """Who made the device and what it is called: the facts no object holds."""

    vendor: str = "Not stated"
    device: str = "py1815 DER outstation"
    hardware_version: str = "Not applicable (software)"
    software_version: str = ""
    document_version: int = 1
    author: str = "py1815"
    #: Where the outstation listens, when that is known.
    host: str | None = None
    port: int | None = None

    def software(self) -> str:
        """The stated software version, or this library's own."""
        if self.software_version:
            return self.software_version
        try:
            return f"py1815 {importlib.metadata.version('py1815')}"
        except importlib.metadata.PackageNotFoundError:
            return "py1815"
software 
software() -> str

The stated software version, or this library's own.

Source code in src/py1815/profile/device_profile.py 
88
89
90
91
92
93
94
95
def software(self) -> str:
    """The stated software version, or this library's own."""
    if self.software_version:
        return self.software_version
    try:
        return f"py1815 {importlib.metadata.version('py1815')}"
    except importlib.metadata.PackageNotFoundError:
        return "py1815"

Row dataclass

One line of the implementation table.

Source code in src/py1815/profile/device_profile.py 
 98
 99
100
101
102
103
104
105
106
107
108
@dataclass(frozen=True)
class Row:
    """One line of the implementation table."""

    group: int
    variation: int
    description: str
    #: The request function code and the qualifiers accepted with it.
    request: tuple[int, tuple[int, ...]] | None = None
    #: The response function code and the qualifiers it may be sent with.
    response: tuple[int, tuple[int, ...]] | None = None

Implementation dataclass

What an outstation answers: object rows, and function codes that carry none.

Source code in src/py1815/profile/device_profile.py 
111
112
113
114
115
116
@dataclass(frozen=True)
class Implementation:
    """What an outstation answers: object rows, and function codes that carry none."""

    rows: tuple[Row, ...]
    function_codes: tuple[int, ...] = field(default=())

implementation 

implementation(
    outstation: DerOutstation, facts: SessionFacts
) -> Implementation

The implementation table for an outstation as it is configured.

Built from what is served and what the session was given, so a monitor with no outputs lists no controls and an outstation with no counters lists no freezes. The tests hold every row to the session's behavior.

Source code in src/py1815/profile/device_profile.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
def implementation(outstation: DerOutstation, facts: SessionFacts) -> Implementation:
    """The implementation table for an outstation as it is configured.

    Built from what is served and what the session was given, so a monitor
    with no outputs lists no controls and an outstation with no counters
    lists no freezes. The tests hold every row to the session's behavior.
    """
    served = _served(outstation)
    if outstation.read_only:
        # A control is answered, with a refusal, and is not something a master
        # may usefully send: it is left off the list of what is supported.
        facts = replace(facts, controls=False)
    rows: list[Row] = []

    def static(group: int, default: str, variations: dict[int, str]) -> None:
        # A read by index is answered by index, and every other read by range.
        select = tuple(_RANGE + _ALL + _INDEXED)
        rows.append(Row(group, 0, f"{default} - any variation", request=(READ, select)))
        for variation, description in variations.items():
            rows.append(
                Row(
                    group,
                    variation,
                    description,
                    request=(READ, select),
                    response=(RESPONSE, tuple(_RANGE + _INDEXED)),
                )
            )

    def widened(group: int, variations: dict[int, str]) -> None:
        # Read and answered in a 32-bit variation with flags, so the row lists
        # the request and no response of its own.
        select = tuple(_RANGE + _ALL + _INDEXED)
        for variation, description in variations.items():
            rows.append(Row(group, variation, description, request=(READ, select)))

    def events(group: int, name: str, variations: dict[int, str]) -> None:
        if not facts.events:
            return
        if group == 23 and outstation.level2:
            # Read, and answered with nothing: a Level 2 outstation buffers none.
            variations = {}
        select = tuple(_ALL + _COUNT)
        rows.append(Row(group, 0, f"{name} - any variation", request=(READ, select)))
        for variation, description in variations.items():
            rows.append(
                Row(
                    group,
                    variation,
                    description,
                    request=(READ, select),
                    response=(RESPONSE, tuple(_INDEXED)),
                )
            )

    def controls(group: int, variations: dict[int, str]) -> None:
        for variation, description in variations.items():
            for function in (SELECT, OPERATE, DIRECT):
                rows.append(
                    Row(
                        group,
                        variation,
                        description,
                        request=(function, tuple(_INDEXED)),
                        response=(RESPONSE, tuple(_INDEXED)),
                    )
                )
            rows.append(Row(group, variation, description, request=(DIRECT_NR, tuple(_INDEXED))))

    if served[Kind.BI]:
        static(1, "Binary Input", {2: "Binary Input - with flags"})
        events(
            2,
            "Binary Input Event",
            {
                1: "Binary Input Event - without time",
                2: "Binary Input Event - with absolute time",
                3: "Binary Input Event - with relative time",
            },
        )
    if served[Kind.BO]:
        static(10, "Binary Output", {2: "Binary Output - output status with flags"})
        if facts.controls:
            controls(12, {1: "Binary Output Command - control relay output block"})
    if served[Kind.CTR]:
        static(20, "Counter", {1: "Counter - 32-bit with flag"})
        widened(
            20,
            {
                2: "Counter - 16-bit with flag",
                5: "Counter - 32-bit without flag",
                6: "Counter - 16-bit without flag",
            },
        )
        if facts.freezes:
            for function in (FREEZE, FREEZE_NR, FREEZE_CLEAR, FREEZE_CLEAR_NR):
                rows.append(
                    Row(20, 0, "Counter - any variation", request=(function, tuple(_RANGE + _ALL)))
                )
        if any(point.frozen for point in served[Kind.CTR]):
            # Only counters that freeze have frozen counters, and a read of group 21
            # from a device with none is refused (D100).
            static(
                21,
                "Frozen Counter",
                {
                    1: "Frozen Counter - 32-bit with flag",
                    5: "Frozen Counter - 32-bit with flag and time",
                    9: "Frozen Counter - 32-bit without flag",
                },
            )
            widened(
                21,
                {
                    2: "Frozen Counter - 16-bit with flag",
                    10: "Frozen Counter - 16-bit without flag",
                },
            )
        if facts.events:
            for variation, description in (
                (0, "Counter Event - any variation"),
                (1, "Counter Event - 32-bit with flag"),
                (2, "Counter Event - 16-bit with flag"),
            ):
                # Read, and answered with nothing: none are ever buffered.
                rows.append(Row(22, variation, description, request=(READ, tuple(_ALL + _COUNT))))
        events(
            23,
            "Frozen Counter Event",
            {
                1: "Frozen Counter Event - 32-bit with flag",
                5: "Frozen Counter Event - 32-bit with flag and time",
            },
        )
    if served[Kind.AI]:
        static(
            30,
            "Analog Input",
            {
                1: "Analog Input - 32-bit with flag",
                2: "Analog Input - 16-bit with flag",
                3: "Analog Input - 32-bit without flag",
                4: "Analog Input - 16-bit without flag",
            },
        )
        events(
            32,
            "Analog Input Event",
            {
                1: "Analog Input Event - 32-bit without time",
                2: "Analog Input Event - 16-bit without time",
                3: "Analog Input Event - 32-bit with time",
                4: "Analog Input Event - 16-bit with time",
            },
        )
    if served[Kind.AO]:
        static(
            40,
            "Analog Output Status",
            {
                1: "Analog Output Status - 32-bit with flag",
                2: "Analog Output Status - 16-bit with flag",
            },
        )
        if facts.controls:
            controls(
                41,
                {
                    1: "Analog Output - 32-bit",
                    2: "Analog Output - 16-bit",
                    3: "Analog Output - single-precision floating point",
                    4: "Analog Output - double-precision floating point",
                },
            )
    if facts.time_write:
        rows.append(Row(50, 1, "Time and Date - absolute time", request=(WRITE, (0x07,))))
        rows.append(
            Row(
                50,
                3,
                "Time and Date - absolute time at last recorded time",
                request=(WRITE, (0x07,)),
            )
        )
    if facts.events and served[Kind.BI]:
        # What a binary event with relative time counts from, and whether the
        # clock that gave it had been set.
        rows.append(
            Row(
                51,
                1,
                "Time and Date CTO - absolute time, synchronized",
                response=(RESPONSE, (0x07,)),
            )
        )
        rows.append(
            Row(
                51,
                2,
                "Time and Date CTO - absolute time, unsynchronized",
                response=(RESPONSE, (0x07,)),
            )
        )
    rows.append(Row(52, 2, "Time Delay - fine", response=(RESPONSE, (0x07,))))
    rows.append(Row(60, 1, "Class Objects - class 0 data", request=(READ, tuple(_ALL))))
    if facts.events:
        for variation in (2, 3, 4):
            rows.append(
                Row(
                    60,
                    variation,
                    f"Class Objects - class {variation - 1} data",
                    request=(READ, tuple(_ALL + _COUNT)),
                )
            )
    if facts.unsolicited and facts.events:
        # What an unsolicited response carries: events, in the variations a
        # class poll is answered in, since it is built the same way. Binary
        # events stamped before the clock was set travel with relative time
        # behind a common time, as they do when polled.
        unsolicited = (UNSOLICITED_RESPONSE, tuple(_INDEXED))
        if served[Kind.BI]:
            rows.append(Row(2, 2, "Binary Input Event - with absolute time", response=unsolicited))
            rows.append(Row(2, 3, "Binary Input Event - with relative time", response=unsolicited))
            for variation, clock in ((1, "synchronized"), (2, "unsynchronized")):
                rows.append(
                    Row(
                        51,
                        variation,
                        f"Time and Date CTO - absolute time, {clock}",
                        response=(UNSOLICITED_RESPONSE, (0x07,)),
                    )
                )
        if served[Kind.CTR] and not outstation.level2:
            rows.append(
                Row(23, 5, "Frozen Counter Event - 32-bit with flag and time", response=unsolicited)
            )
        if served[Kind.AI]:
            rows.append(
                Row(
                    32,
                    facts.analog_event_variation,
                    "Analog Input Event - as configured",
                    response=unsolicited,
                )
            )
    if facts.unsolicited:
        # Enabled and disabled by class. Listed only for a session that has
        # unsolicited responses on: with them off the enable request is
        # refused and nothing is sent.
        for variation in (2, 3, 4):
            for function in (ENABLE_UNSOLICITED, DISABLE_UNSOLICITED):
                rows.append(
                    Row(
                        60,
                        variation,
                        f"Class Objects - class {variation - 1} data",
                        request=(function, tuple(_ALL)),
                    )
                )
    rows.append(Row(80, 1, "Internal Indications - packed format", request=(WRITE, tuple(_RANGE))))
    # Function codes that carry no object: confirm, delay measurement, and the
    # two that depend on how the session was built. Enable and disable
    # unsolicited are not among them: a session that sends unsolicited
    # responses takes them by class, above, and one that sends none refuses
    # both (D21).
    codes = [0, 23]
    if facts.cold_restart:
        codes.append(13)
    if facts.time_write:
        codes.append(RECORD_CURRENT_TIME)
    # What configuration has turned off is not something the device answers.
    off = facts.disabled_functions
    return Implementation(
        rows=tuple(r for r in rows if r.request is None or r.request[0] not in off),
        function_codes=tuple(sorted(code for code in codes if code not in off)),
    )

build 

build(
    outstation: DerOutstation,
    session: Session,
    identity: Identity | None = None,
    *,
    statuses: Sequence[CommandStatus] = (),
    today: date | None = None,
) -> ElementTree.Element

The Device Profile document for an outstation and the session serving it.

Parameters:

Name Type Description Default
outstation DerOutstation

What is served: the points, each one's scaling, and the event class and deadband in force for it.

 required
session Session

How it is served: addresses, fragment sizes, timeouts, and which of time, freezes, controls and events it was given.

 required
identity Identity | None

The vendor, device and version strings, and where it listens.

 None
statuses Sequence[CommandStatus]

Command statuses the caller's bindings can return, beyond the ones the session and builder answer with themselves.

 ()
today date | None

The date written into the revision history.

 None
Source code in src/py1815/profile/device_profile.py 
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
def build(
    outstation: DerOutstation,
    session: Session,
    identity: Identity | None = None,
    *,
    statuses: Sequence[CommandStatus] = (),
    today: datetime.date | None = None,
) -> ElementTree.Element:
    """The Device Profile document for an outstation and the session serving it.

    Args:
        outstation: What is served: the points, each one's scaling, and the
            event class and deadband in force for it.
        session: How it is served: addresses, fragment sizes, timeouts, and
            which of time, freezes, controls and events it was given.
        identity: The vendor, device and version strings, and where it listens.
        statuses: Command statuses the caller's bindings can return, beyond the
            ones the session and builder answer with themselves.
        today: The date written into the revision history.
    """
    identity = identity or Identity()
    facts = session.facts
    if outstation.read_only:
        # The document describes what a master may ask of this outstation, and
        # a read-only one offers no control, though it answers one out loud.
        facts = replace(facts, controls=False)
    served = _served(outstation)
    codes = sorted({int(status) for status in (*_OWN_STATUSES, *statuses)} - {0})

    ElementTree.register_namespace("", NAMESPACE)
    ElementTree.register_namespace("xsi", _XSI)
    root = ElementTree.Element(
        _tag("DNP3DeviceProfileDocument"),
        {
            "schemaVersion": SCHEMA_VERSION,
            f"{{{_XSI}}}schemaLocation": f"{NAMESPACE} {SCHEMA_FILE}",
        },
    )
    header = ElementTree.SubElement(root, _tag("documentHeader"))
    _add(
        header,
        [
            ("documentName", f"{identity.device} Device Profile"),
            (
                "documentDescription",
                "Capabilities and current configuration, generated from the running outstation",
            ),
            ("documentType", "Both"),
        ],
    )
    revision = ElementTree.SubElement(
        header, _tag("revisionHistory"), {"version": str(identity.document_version)}
    )
    _add(
        revision,
        [
            ("date", (today or datetime.date.today()).isoformat()),
            ("author", identity.author),
            ("reason", "Generated"),
        ],
    )

    device = ElementTree.SubElement(root, _tag("referenceDevice"))
    _add(
        device,
        [
            (
                "configuration",
                [
                    ("deviceConfig", _device(identity, facts)),
                    ("networkConfig", _network(identity, facts)),
                    ("linkConfig", _link(facts)),
                    ("applConfig", _application(facts, codes)),
                    ("outstationConfig", _outstation(facts)),
                    ("unsolicitedConfig", _unsolicited(facts)),
                    ("outstationPerformance", _performance(facts)),
                ],
            ),
            ("database", _database(served, facts, outstation)),
            ("implementationTable", _table(implementation(outstation, facts))),
            ("dataPointsList", _points(served, facts, outstation)),
        ],
    )
    return root

render 

render(root: Element) -> str

The document as text, with the declaration and stylesheet reference.

The stylesheet instruction is what lets a browser, or any XSLT processor, show the profile in the Users Group's own layout when the stylesheet sits beside the file.

Source code in src/py1815/profile/device_profile.py 
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
def render(root: ElementTree.Element) -> str:
    """The document as text, with the declaration and stylesheet reference.

    The stylesheet instruction is what lets a browser, or any XSLT processor,
    show the profile in the Users Group's own layout when the stylesheet sits
    beside the file.
    """
    ElementTree.indent(root, space="  ")
    body = ElementTree.tostring(root, encoding="unicode")
    return (
        '<?xml version="1.0" encoding="UTF-8"?>\n'
        f'<?xml-stylesheet type="text/xsl" href="{STYLESHEET_FILE}"?>\n'
        f"{body}\n"
    )

Device Profile, laid out

Reads the XML of any Device Profile document into facts, sections and tables, and writes them as one HTML page. The guide's Reading the document, and printing it says what the page holds.

py1815.profile.device_profile_view

Read a DNP3 Device Profile document and lay it out as a page a person can read.

A Device Profile is XML, and a person reads it as a document: what the device is, how each setting may be configured and how it is configured now, the implementation table, and the point lists. This module reads the XML of any Device Profile, whichever edition of the schema it was written for and whoever wrote it, into a plain model of facts, sections and tables, and writes that model as one standalone HTML page that prints cleanly.

The layout is this library's own. The DNP Users Group publishes a stylesheet that renders a Device Profile in the Users Group's layout. That stylesheet and the schema are the Users Group's and are not carried here (D98), so nothing in this module comes from either. A label is the element's own name made readable (maxReceivedFragmentSize becomes "Max received fragment size"), which is not the wording the Users Group prints.

What it does not know. It does not know the schema: it cannot say whether a document is valid, which elements an edition requires, or what an element means. It recognizes a handful of element names that every edition shares, such as the document header, the three parts of a setting and the point lists, and lays out everything else by its shape. An element it does not recognize is still shown, by its label and its text. It does not know the namespace either, since each edition has its own: only local names are compared. It states nothing the document does not: a value that is absent is an empty cell.

The document is untrusted input. A document that is not XML, or is not a Device Profile, is refused with a :class:DeviceProfileError that says what is wrong. See :func:read for the limits it enforces.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

DeviceProfileError

Bases: ValueError

A document that cannot be read as a Device Profile, and why.

Source code in src/py1815/profile/device_profile_view.py 
103
104
class DeviceProfileError(ValueError):
    """A document that cannot be read as a Device Profile, and why."""

Fact dataclass

One fact of the document's header: what it is, and its value.

Source code in src/py1815/profile/device_profile_view.py 
107
108
109
110
111
112
@dataclass(frozen=True)
class Fact:
    """One fact of the document's header: what it is, and its value."""

    label: str
    value: str

Table dataclass

A table: a title, column headings, and rows of text cells.

Every row has a cell for every column. A cell that states several things separates them with a newline, and a cell the document says nothing for is empty.

Source code in src/py1815/profile/device_profile_view.py 
115
116
117
118
119
120
121
122
123
124
125
126
@dataclass(frozen=True)
class Table:
    """A table: a title, column headings, and rows of text cells.

    Every row has a cell for every column. A cell that states several things
    separates them with a newline, and a cell the document says nothing for
    is empty.
    """

    title: str
    columns: tuple[str, ...]
    rows: tuple[tuple[str, ...], ...]

Section dataclass

A part of the document: a title and the tables under it.

Source code in src/py1815/profile/device_profile_view.py 
129
130
131
132
133
134
@dataclass(frozen=True)
class Section:
    """A part of the document: a title and the tables under it."""

    title: str
    tables: tuple[Table, ...]

Document dataclass

A Device Profile as a person reads it: a title, header facts and sections.

Source code in src/py1815/profile/device_profile_view.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
@dataclass(frozen=True)
class Document:
    """A Device Profile as a person reads it: a title, header facts and sections."""

    title: str
    facts: tuple[Fact, ...]
    sections: tuple[Section, ...]
    #: Whose layout this is, in one line.
    notice: str = NOTICE

    def describe(self) -> dict[str, Any]:
        """Return the document as JSON-compatible data."""
        return {
            "title": self.title,
            "notice": self.notice,
            "facts": [{"label": fact.label, "value": fact.value} for fact in self.facts],
            "sections": [
                {
                    "title": section.title,
                    "tables": [
                        {
                            "title": table.title,
                            "columns": list(table.columns),
                            "rows": [list(row) for row in table.rows],
                        }
                        for table in section.tables
                    ],
                }
                for section in self.sections
            ],
        }
describe 
describe() -> dict[str, Any]

Return the document as JSON-compatible data.

Source code in src/py1815/profile/device_profile_view.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def describe(self) -> dict[str, Any]:
    """Return the document as JSON-compatible data."""
    return {
        "title": self.title,
        "notice": self.notice,
        "facts": [{"label": fact.label, "value": fact.value} for fact in self.facts],
        "sections": [
            {
                "title": section.title,
                "tables": [
                    {
                        "title": table.title,
                        "columns": list(table.columns),
                        "rows": [list(row) for row in table.rows],
                    }
                    for table in section.tables
                ],
            }
            for section in self.sections
        ],
    }

read 

read(source: str | bytes) -> Document

Read the XML of a Device Profile document into a :class:Document.

The text is read as it is given. Octets are decoded as the document's own declaration says, UTF-8 when it has none.

The document is untrusted. One that declares a document type is refused, so no entity can be defined to expand or to fetch anything, and nothing the document refers to is followed. One of more than :data:MAX_SIZE characters or octets, more than :data:MAX_ELEMENTS elements, or elements nested more than :data:MAX_DEPTH deep is refused, and so is one whose tables would have more than :data:MAX_COLUMNS columns in one table or :data:MAX_CELLS cells between them.

Raises:

Type Description
DeviceProfileError

The text is not XML, is not a Device Profile, or is past one of the limits. The message says which.
Source code in src/py1815/profile/device_profile_view.py 
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
def read(source: str | bytes) -> Document:
    """Read the XML of a Device Profile document into a :class:`Document`.

    The text is read as it is given. Octets are decoded as the document's
    own declaration says, UTF-8 when it has none.

    The document is untrusted. One that declares a document type is
    refused, so no entity can be defined to expand or to fetch anything, and
    nothing the document refers to is followed. One of more than
    :data:`MAX_SIZE` characters or octets, more than :data:`MAX_ELEMENTS`
    elements, or elements nested more than :data:`MAX_DEPTH` deep is refused,
    and so is one whose tables would have more than :data:`MAX_COLUMNS`
    columns in one table or :data:`MAX_CELLS` cells between them.

    Raises:
        DeviceProfileError: The text is not XML, is not a Device Profile, or
            is past one of the limits. The message says which.
    """
    return _layout(_parse(source))

page 

page(document: Document) -> str

Return a Device Profile as one standalone HTML page.

The page carries its own style, runs no script and refers to nothing outside itself, so it can be saved, mailed or opened from a disk. Every piece of text from the document is escaped. Printed, or saved as PDF from a browser, it repeats each table's headings on every page and keeps a row on one page.

Source code in src/py1815/profile/device_profile_view.py 
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
def page(document: Document) -> str:
    """Return a Device Profile as one standalone HTML page.

    The page carries its own style, runs no script and refers to nothing
    outside itself, so it can be saved, mailed or opened from a disk. Every
    piece of text from the document is escaped. Printed, or saved as PDF from
    a browser, it repeats each table's headings on every page and keeps a row
    on one page.
    """
    lines = [
        "<!doctype html>",
        '<html lang="en">',
        "<head>",
        '<meta charset="utf-8">',
        '<meta name="viewport" content="width=device-width, initial-scale=1">',
        f"<title>{_text(document.title)}</title>",
        f"<style>{_STYLE}</style>",
        "</head>",
        "<body>",
        "<header>",
        f"<h1>{_text(document.title)}</h1>",
        f'<p class="notice">{_text(document.notice)}</p>',
    ]
    if document.facts:
        lines.append('<dl class="facts">')
        lines += [
            f"<dt>{_text(fact.label)}</dt><dd>{_text(fact.value)}</dd>" for fact in document.facts
        ]
        lines.append("</dl>")
    lines.append("</header>")
    if document.sections:
        lines.append('<nav class="contents" aria-label="Contents">')
        lines.append("<ol>")
        lines += [
            f'<li><a href="#section-{number}">{_text(section.title)}</a></li>'
            for number, section in enumerate(document.sections, start=1)
        ]
        lines.append("</ol>")
        lines.append("</nav>")
    lines.append("<main>")
    for number, section in enumerate(document.sections, start=1):
        lines.append(f'<section id="section-{number}">')
        lines.append(f"<h2>{_text(section.title)}</h2>")
        if not section.tables:
            lines.append('<p class="empty">Nothing is stated.</p>')
        for table in section.tables:
            lines += _table(table)
        lines.append("</section>")
    lines += ["</main>", "</body>", "</html>"]
    return "\n".join(lines) + "\n"

Simulated DER

py1815.profile.der

A simulated DER to stand behind the outstation, so it runs with nothing attached.

A three-phase, storage-coupled generator of modest size, modeled just far enough to make every point the profile requires mean something: power that follows a slowly varying source, a meter that reports it, energy counters that accumulate it, a state of charge that moves when the storage is dispatched, and six DER functions a controlling station can enable and watch take effect -- active power limit, charge/discharge, constant vars, constant power factor, and two that follow a curve of voltage: volt-var and volt-watt.

It is a stand-in, not a model of any product, and it is not where DER behavior belongs: how a DER carries out a function or follows a curve is out of scope for this package. What a real integration replaces is this module and nothing else: it binds the same points to its own device or DER simulator through the same :class:~py1815.profile.binding.Binding.

The indices below are the profile's fixed ones (IEEE 1815.2 clause 5.2 gives the configuration and system blocks absolute indices), named for what this simulation does with each. Names, units and scaling stay in the tables, which are loaded at run time and not carried here.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

Function dataclass

One DER function this simulation implements, by the points it uses.

Source code in src/py1815/profile/der.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@dataclass(frozen=True)
class Function:
    """One DER function this simulation implements, by the points it uses."""

    name: str
    #: The binary output that enables it.
    enable: int
    #: The first and last of its analog output settings.
    settings: tuple[int, int]
    #: Analog inputs it computes, beyond the ones that read its settings back.
    inputs: tuple[int, ...] = ()
    #: The setting that names its curve, and the curve types it follows.
    curve: int | None = None
    curve_types: tuple[int, ...] = ()

Ratings dataclass

The nameplate of the simulated DER.

Source code in src/py1815/profile/der.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
@dataclass(frozen=True)
class Ratings:
    """The nameplate of the simulated DER."""

    watts: float = 50_000.0
    volt_amperes: float = 55_000.0
    vars: float = 27_500.0
    #: Nominal line-to-line voltage; each phase is this over the root of three.
    volts: float = 480.0
    hertz: float = 60.0
    storage_watt_hours: float = 200_000.0

    @property
    def phase_volts(self) -> float:
        """Nominal line-to-neutral voltage."""
        return self.volts / math.sqrt(3)

    @property
    def amperes(self) -> float:
        """Rated current per phase."""
        return self.volt_amperes / (math.sqrt(3) * self.volts)
phase_volts property 
phase_volts: float

Nominal line-to-neutral voltage.

amperes property 
amperes: float

Rated current per phase.

ReferenceDer

The simulated device: its state, how it moves, and how it is bound.

Source code in src/py1815/profile/der.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
class ReferenceDer:
    """The simulated device: its state, how it moves, and how it is bound."""

    def __init__(self, ratings: Ratings | None = None, *, seed: int = 0) -> None:
        self.ratings = ratings or Ratings()
        self._random = random.Random(seed)
        #: Every setting a controlling station can write, by address, in
        #: engineering units. Seeded by :meth:`bind` with each one's default.
        self.settings: dict[Address, float | bool] = {}
        self.elapsed = 0.0
        self.started = True
        #: A start or stop under way: where it is headed, and the seconds left.
        self._transition: tuple[bool, float] | None = None
        self.watts = 0.0
        self.vars = 0.0
        self.phase_volts = self.ratings.phase_volts
        self.hertz = self.ratings.hertz
        self.state_of_charge = 60.0
        #: Running energy totals: watt-hours delivered and received, then
        #: var-hours delivered and received.
        self.energy = [0.0, 0.0, 0.0, 0.0]
        #: The curves, the window onto them, and which function names which.
        self.curves = CurveStore(CURVE_COUNT)
        #: How to get the curve each curve-following function names, by the
        #: index of the setting that names it. Filled in by :meth:`bind`.
        self._curve_of: dict[int, Callable[[], curves.Curve | None]] = {}

    # ------------------------------------------------------------- behavior

    def _on(self, index: int) -> bool:
        return bool(self.settings.get((BO, index), False))

    def _setting(self, index: int) -> float:
        return float(self.settings.get((AO, index), 0.0))

    @property
    def selected_curve(self) -> int:
        """The number of the curve the edit window is showing."""
        return self.curves.selected

    @property
    def line_volts(self) -> float:
        """Measured line-to-line voltage at the connection point."""
        return self.phase_volts * math.sqrt(3)

    @property
    def nominal_volts(self) -> float:
        """What a curve's percent voltage is a percentage of: the reference plus its offset."""
        return self._setting(AO_REFERENCE_VOLTAGE) + self._setting(AO_REFERENCE_VOLTAGE_OFFSET)

    def _along(self, setting: int, y_units: tuple[int, ...]) -> float | None:
        """Where the measured voltage falls on a function's curve, as a fraction of rating.

        None when the function names no curve, or one in units this
        simulation does not follow, or the nominal voltage is not set.
        """
        named = self._curve_of.get(setting)
        curve = named() if named is not None else None
        nominal = self.nominal_volts
        if curve is None or nominal <= 0:
            return None
        if curve.x_units != X_PERCENT_VOLTAGE or curve.y_units not in y_units:
            return None
        # Both axes travel in tenths of a percent.
        value = curve.at(self.line_volts / nominal * 1000.0)
        return None if value is None else value / 1000.0

    def volt_watt_limit(self) -> float | None:
        """The active power volt-watt allows at this voltage, or None when it sets no limit."""
        if not self._on(BO_ENABLE_VOLT_WATT):
            return None
        share = self._along(AO_VOLT_WATT_CURVE, (Y_PERCENT_MAX_WATTS,))
        return None if share is None else share * self.ratings.watts

    def volt_var_target(self) -> float | None:
        """The reactive power volt-var calls for at this voltage, or None when it calls for none."""
        if not self._on(BO_ENABLE_VOLT_VAR):
            return None
        share = self._along(AO_VOLT_VAR_CURVE, (Y_PERCENT_MAX_VARS, Y_PERCENT_AVAILABLE_VARS))
        return None if share is None else share * self.ratings.vars

    @property
    def connected(self) -> bool:
        """Whether the connect switch is closed."""
        return self._on(BO_CONNECT)

    @property
    def starting(self) -> bool:
        """Whether a start has been commanded and is not yet complete."""
        return self._transition is not None and self._transition[0]

    @property
    def stopping(self) -> bool:
        """Whether a stop has been commanded and is not yet complete."""
        return self._transition is not None and not self._transition[0]

    @property
    def energized(self) -> bool:
        """Whether the DER is started and connected, and so exchanging power."""
        return self.started and self.connected

    def available_watts(self) -> float:
        """What the primary source could deliver now: a slow swell around 75%."""
        return self.ratings.watts * (0.75 + 0.2 * math.sin(2 * math.pi * self.elapsed / 600.0))

    def _targets(self) -> tuple[float, float]:
        """The active and reactive power the enabled functions call for."""
        ratings = self.ratings
        if not self.energized:
            return 0.0, 0.0
        watts = self.available_watts()
        if self._on(BO_ENABLE_CHARGE_DISCHARGE):
            watts = self._setting(AO_CHARGE_DISCHARGE_TARGET) / 100.0 * ratings.watts
            full = self.state_of_charge >= 100.0
            empty = self.state_of_charge <= 0.0
            if (watts < 0 and full) or (watts > 0 and empty):
                watts = 0.0
        if self._on(BO_ENABLE_POWER_LIMIT):
            # The profile takes the two percentages of different things: of
            # maximum generation power, and of maximum charging power. This DER
            # has one rating for both. Give it a charging rating of its own and
            # the second line has to use it.
            generating = self._setting(AO_POWER_LIMIT_GENERATION) / 100.0 * ratings.watts
            charging = self._setting(AO_POWER_LIMIT_CHARGING) / 100.0 * ratings.watts
            watts = max(-charging, min(watts, generating))
        limit = self.volt_watt_limit()
        if limit is not None:
            watts = min(watts, limit)
        watts = max(-ratings.watts, min(ratings.watts, watts))

        reactive = 0.0
        if self._on(BO_ENABLE_CONSTANT_PF):
            charging = watts < 0
            factor = self._setting(
                AO_CONSTANT_PF_CHARGING if charging else AO_CONSTANT_PF_GENERATING
            )
            factor = min(1.0, max(0.01, abs(factor)))
            absorbing = self._on(BO_PF_ABSORB_CHARGING if charging else BO_PF_ABSORB_GENERATING)
            reactive = abs(watts) * math.tan(math.acos(factor)) * (-1.0 if absorbing else 1.0)
        elif self._on(BO_ENABLE_CONSTANT_VARS):
            reactive = self._setting(AO_CONSTANT_VARS_TARGET) / 100.0 * ratings.vars
        else:
            # Volt-var gives way to either fixed reactive function.
            reactive = self.volt_var_target() or 0.0
        reactive = max(-ratings.vars, min(ratings.vars, reactive))
        # Active power has priority: reactive gives way at the apparent limit.
        headroom = math.sqrt(max(0.0, ratings.volt_amperes**2 - watts**2))
        return watts, max(-headroom, min(headroom, reactive))

    def step(self, seconds: float) -> None:
        """Move the simulation forward."""
        if seconds <= 0:
            return
        self.elapsed += seconds
        if self._transition is not None:
            target, remaining = self._transition
            if remaining <= seconds:
                self.started, self._transition = target, None
            else:
                self._transition = (target, remaining - seconds)
        target_watts, target_vars = self._targets()
        share = 1.0 - math.exp(-seconds / _RESPONSE_SECONDS)
        self.watts += (target_watts - self.watts) * share
        self.vars += (target_vars - self.vars) * share

        ratings = self.ratings
        self.phase_volts = ratings.phase_volts * (
            1.0 + 0.004 * math.sin(self.elapsed / 37.0) + self._random.gauss(0.0, 0.0004)
        )
        self.hertz = (
            ratings.hertz + 0.01 * math.sin(self.elapsed / 53.0) + self._random.gauss(0.0, 0.002)
        )

        hours = seconds / 3600.0
        self.energy[0] += max(self.watts, 0.0) * hours
        self.energy[1] += max(-self.watts, 0.0) * hours
        self.energy[2] += max(self.vars, 0.0) * hours
        self.energy[3] += max(-self.vars, 0.0) * hours
        if self._on(BO_ENABLE_CHARGE_DISCHARGE):
            moved = -self.watts * hours / ratings.storage_watt_hours * 100.0
            self.state_of_charge = max(0.0, min(100.0, self.state_of_charge + moved))

    # ------------------------------------------------------- derived values

    @property
    def volt_amperes(self) -> float:
        """Apparent power at the connection point."""
        return math.hypot(self.watts, self.vars)

    @property
    def power_factor(self) -> float:
        """Power factor magnitude; unity when nothing is flowing."""
        apparent = self.volt_amperes
        return 1.0 if apparent < 1.0 else abs(self.watts) / apparent

    @property
    def amperes(self) -> float:
        """Current per phase."""
        return self.volt_amperes / (3.0 * self.phase_volts)

    def freeze_interval_seconds(self) -> float | None:
        """How often the counters freeze, or None when they are not to repeat."""
        unit = _INTERVAL_SECONDS.get(round(self._setting(AO_FREEZE_INTERVAL_UNITS)))
        interval = self._setting(AO_FREEZE_INTERVAL)
        if unit is None or interval <= 0:
            return None
        return interval * unit

    # ---------------------------------------------------------------- binding

    def bind(self, point_map: PointMap) -> Binding:
        """This DER's binding: every point it serves, and how."""
        binding = Binding()
        self._bind_state(binding)
        self._bind_settings(binding)
        self._bind_functions(binding)
        self._bind_curves(binding)
        self._bind_nameplate(binding, point_map)
        self._bind_meter(binding)
        return binding

    def _blocked(self, _value: float) -> CommandStatus | None:
        """Refuse a command while the system is locked out."""
        return CommandStatus.BLOCKED if self._on(BO_LOCKOUT) else None

    def _output(
        self,
        binding: Binding,
        kind: Kind,
        index: int,
        initial: float | bool,
        apply: Callable[[float], CommandStatus | None] | None = None,
        *,
        lockable: bool = True,
    ) -> None:
        """Bind one output as a stored setting, optionally with a side effect."""
        address = (kind, index)
        self.settings[address] = initial

        def store(value: float) -> CommandStatus | None:
            if apply is not None:
                answer = apply(value)
                if answer is not None and answer is not CommandStatus.SUCCESS:
                    return answer
            self.settings[address] = value
            return None

        binding.output(
            kind, index, store, initial=initial, check=self._blocked if lockable else None
        )

    def _start(self, value: float) -> CommandStatus | None:
        if not value:
            return None
        if not self._on(BO_PERMIT_START):
            return CommandStatus.BLOCKED
        if not self.started or self.stopping:
            self._transition = (True, _TRANSITION_SECONDS)
        return None

    def _stop(self, value: float) -> CommandStatus | None:
        if not value:
            return None
        if not self._on(BO_PERMIT_STOP):
            return CommandStatus.BLOCKED
        if self.started or self.starting:
            self._transition = (False, _TRANSITION_SECONDS)
        return None

    def _bind_state(self, binding: Binding) -> None:
        # The lockout itself stays commandable while locked out: it is how a
        # controlling station releases it.
        self._output(binding, BO, BO_LOCKOUT, False, lockable=False)
        self._output(binding, BO, BO_START, False, self._start)
        self._output(binding, BO, BO_STOP, False, self._stop)
        self._output(binding, BO, BO_PERMIT_START, True)
        self._output(binding, BO, BO_PERMIT_STOP, True)
        self._output(binding, BO, BO_CONNECT, True)
        self._output(binding, BO, BO_PF_ABSORB_GENERATING, False)
        self._output(binding, BO, BO_PF_ABSORB_CHARGING, False)

        def idle() -> bool:
            return self.energized and abs(self.watts) < 0.01 * self.ratings.watts

        states: dict[int, Reader] = {
            # Alarms: a healthy simulated system raises none but the storage ones.
            0: lambda: False,
            1: lambda: False,
            2: lambda: False,
            3: lambda: False,
            4: lambda: self.state_of_charge >= 100.0,
            5: lambda: self.state_of_charge > 95.0,
            6: lambda: self.state_of_charge < 5.0,
            7: lambda: self.state_of_charge <= 0.0,
            8: lambda: False,
            9: lambda: False,
            # State.
            10: lambda: False,
            12: lambda: self.starting,
            13: lambda: self.stopping,
            14: lambda: self.started,
            15: lambda: not self.started,
            18: idle,
            19: lambda: self.energized and not idle() and self.watts > 0,
            20: lambda: self.energized and not idle() and self.watts < 0,
            21: lambda: not self.started and self._on(BO_PERMIT_START),
            22: lambda: not self.started and not self._on(BO_PERMIT_START),
            23: lambda: self.connected,
            24: lambda: False,
            # Storage capacity is stated in watt-hours.
            27: lambda: True,
        }
        # The disconnect protections: none blocked, started or operated.
        states.update({index: (lambda: False) for index in range(52, 64)})
        for index, reader in states.items():
            binding.read(BI, index, reader)

    def _bind_settings(self, binding: Binding) -> None:
        ratings = self.ratings
        defaults = {
            AO_REFERENCE_VOLTAGE: ratings.volts,
            AO_REFERENCE_VOLTAGE_OFFSET: 0.0,
            AO_NOMINAL_FREQUENCY: ratings.hertz,
            AO_RESPONSE_TIME_PERCENT: 90.0,
            AO_PF_SIGN_CONVENTION: 2.0,
            AO_REACTIVE_REFERENCE: 2.0,
            # Return to service: voltage window in percent of reference, the
            # frequency window in hertz, then delays and ramps in seconds.
            6: 105.0,
            7: 91.7,
            8: ratings.hertz + 0.1,
            9: ratings.hertz - 0.5,
            10: 300.0,
            11: 0.0,
            12: 300.0,
            13: 0.0,
            14: 0.0,
            15: 0.0,
            16: 0.0,
            17: 0.0,
            # Counters freeze every five minutes until told otherwise.
            AO_FREEZE_INTERVAL: 300_000.0,
        }
        for index, value in defaults.items():
            self._output(binding, AO, index, value)

        def interval_units(value: float) -> CommandStatus | None:
            if round(value) != 0 and round(value) not in _INTERVAL_SECONDS:
                return CommandStatus.NOT_SUPPORTED
            return None

        self._output(binding, AO, AO_FREEZE_INTERVAL_UNITS, 1.0, interval_units)

        def counter_units(value: float) -> CommandStatus | None:
            # The counters count watt-hours and var-hours and nothing coarser.
            return None if round(value) == 1 else CommandStatus.NOT_SUPPORTED

        self._output(binding, AO, AO_COUNTER_UNITS, 1.0, counter_units)

        # Two settings whose inputs the tables do not pair with their outputs.
        binding.read(AI, AI_RESPONSE_TIME_PERCENT, lambda: self._setting(AO_RESPONSE_TIME_PERCENT))
        binding.read(AI, AI_FREEZE_INTERVAL, lambda: self._setting(AO_FREEZE_INTERVAL))

        # System meter alarm thresholds: active and reactive power, power
        # factor, then each phase voltage, high before low. A low threshold of
        # zero is one that is not set.
        phase = ratings.phase_volts
        thresholds = [
            1.1 * ratings.watts,
            0.0,
            1.1 * ratings.vars,
            0.0,
            1.0,
            0.0,
            *([1.1 * phase, 0.88 * phase] * 3),
        ]
        for offset, value in enumerate(thresholds):
            self._output(binding, AO, AO_METER_THRESHOLD_FIRST + offset, value)

    def _bind_functions(self, binding: Binding) -> None:
        self._output(binding, BO, BO_ENABLE_POWER_LIMIT, False)
        self._output(binding, BO, BO_ENABLE_CHARGE_DISCHARGE, False)
        self._output(binding, BO, BO_ENABLE_CONSTANT_VARS, False)
        self._output(binding, BO, BO_ENABLE_CONSTANT_PF, False)
        self._output(binding, BO, BO_ENABLE_VOLT_WATT, False)
        self._output(binding, BO, BO_ENABLE_VOLT_VAR, False)

        defaults = {
            AO_POWER_LIMIT_CHARGING: 100.0,
            AO_POWER_LIMIT_GENERATION: 100.0,
            AO_CONSTANT_PF_GENERATING: 1.0,
            AO_CONSTANT_PF_CHARGING: 1.0,
        }
        for function in FUNCTIONS:
            first, last = function.settings
            for index in range(first, last + 1):
                if index != function.curve:
                    self._output(binding, AO, index, defaults.get(index, 0.0))
        binding.read(AI, AI_POWER_LIMIT_REFERENCE, lambda: self.watts)

        # The two functions that follow a curve report the voltage they act
        # on and what they are asking for; with nothing asked, that is what
        # the DER is doing anyway.
        binding.read(AI, AI_VOLT_WATT_VOLTAGE, lambda: self.line_volts, deadband=10)
        binding.read(AI, AI_VOLT_VAR_VOLTAGE, lambda: self.line_volts, deadband=10)
        binding.read(AI, AI_VOLT_VAR_REFERENCE, lambda: self.nominal_volts)

        def volt_watt_output() -> float:
            limit = self.volt_watt_limit()
            return self.ratings.watts if limit is None else limit

        binding.read(AI, AI_VOLT_WATT_OUTPUT, volt_watt_output)
        binding.read(AI, AI_VOLT_VAR_OUTPUT, lambda: self.volt_var_target() or 0.0)

    def _bind_curves(self, binding: Binding) -> None:
        """The curve block, and the functions that name a curve from it."""
        self.curves.bind(
            binding,
            output=AO_CURVE_SELECTOR,
            readback=AI_CURVE_SELECTOR,
            referenced=BI_CURVE_REFERENCED,
            check=self._blocked,
        )
        for function in FUNCTIONS:
            if function.curve is None:
                continue
            enable = function.enable
            self._curve_of[function.curve] = self.curves.reference(
                binding,
                function.curve,
                types=function.curve_types,
                enabled=lambda enable=enable: self._on(enable),  # type: ignore[misc]
                check=self._blocked,
            )

    def _bind_nameplate(self, binding: Binding, point_map: PointMap) -> None:
        ratings = self.ratings
        composition = point_map.composition
        fixed = {
            # The implementation level is left at "ignore": this is a simulation.
            1: 0.0,
            2: 0.88 * ratings.volts,
            3: 1.1 * ratings.volts,
            4: ratings.watts,
            5: -ratings.watts,
            6: ratings.watts,
            7: -ratings.watts,
            8: 0.9,
            9: ratings.watts,
            10: -ratings.watts,
            11: 0.9,
            12: ratings.vars,
            13: -ratings.vars,
            14: ratings.volt_amperes,
            15: -ratings.volt_amperes,
            16: ratings.storage_watt_hours,
            17: ratings.storage_watt_hours,
            18: ratings.storage_watt_hours,
            19: ratings.amperes,
            20: -ratings.amperes,
            # IEEE 1547 performance categories: B, and III.
            22: 2.0,
            23: 3.0,
            # No schedules; equipment counts from the composition served.
            24: 0.0,
            25: float(composition.meters),
            26: float(composition.inverters),
            27: float(composition.batteries),
            28: float(composition.der_units),
            32: ratings.watts,
            33: -ratings.watts,
            34: ratings.vars,
            35: -ratings.vars,
            36: ratings.volt_amperes,
            37: -ratings.volt_amperes,
            38: 0.88 * ratings.volts,
            39: 1.1 * ratings.volts,
            # Ramp rate limits, in percent of rating per second.
            62: 100.0,
            63: 100.0,
            64: 100.0,
            65: 100.0,
        }
        for index, value in fixed.items():
            binding.read(AI, index, _fixed(value))

        available_vars = self.ratings.vars
        binding.read(
            AI, 43, lambda: self.available_watts() if self.energized else 0.0, deadband=500
        )
        binding.read(
            AI,
            44,
            lambda: -ratings.watts if self.energized and self.state_of_charge < 100.0 else 0.0,
        )
        binding.read(AI, 45, lambda: available_vars if self.energized else 0.0)
        binding.read(AI, 46, lambda: -available_vars if self.energized else 0.0)
        binding.read(AI, 47, lambda: self.state_of_charge, deadband=5)
        binding.read(AI, 48, lambda: self.state_of_charge, deadband=5)
        # Start-up status: zero while stopped, complete once started.
        binding.read(AI, 49, lambda: 99.0 if self.started else 0.0)

    def _bind_meter(self, binding: Binding) -> None:
        """The system meter at the connection point, alarms and counters."""
        ratings = self.ratings
        power_band = 0.01 * ratings.watts
        meter: list[tuple[Reader, float | None]] = [
            (lambda: 1.0, None),  # connection point: the DER to the local system
            (lambda: 6.0, None),  # three-phase wye, grounded
            (lambda: 1.0, None),  # apparent power as the vector sum
            (lambda: self.hertz, 20),
            (lambda: self.watts, power_band),
            (lambda: self.watts / 3.0, power_band),
            (lambda: self.watts / 3.0, power_band),
            (lambda: self.watts / 3.0, power_band),
            (lambda: self.vars, power_band),
            (lambda: self.vars / 3.0, power_band),
            (lambda: self.vars / 3.0, power_band),
            (lambda: self.vars / 3.0, power_band),
            (lambda: self.power_factor, 2),
            (lambda: self.volt_amperes, power_band),
            (lambda: self.phase_volts, 10),
            (lambda: 0.0, None),
            (lambda: self.phase_volts, 10),
            # Angles are reported from zero to a full turn: phase B lags A by a third.
            (lambda: 240.0, None),
            (lambda: self.phase_volts, 10),
            (lambda: 120.0, None),
            (lambda: self.phase_volts * math.sqrt(3), 10),
            (lambda: self.amperes, 10),
            (lambda: self.amperes, 10),
            (lambda: self.amperes, 10),
        ]
        for offset, (reader, deadband) in enumerate(meter):
            binding.read(AI, AI_METER_FIRST + offset, reader, deadband=deadband)

        def threshold(offset: int) -> float:
            return self._setting(AO_METER_THRESHOLD_FIRST + offset)

        def high(value: Callable[[], float], offset: int) -> Reader:
            return lambda: value() > threshold(offset)

        def low(value: Callable[[], float], offset: int) -> Reader:
            return lambda: threshold(offset) > 0.0 and value() < threshold(offset)

        measured: list[Callable[[], float]] = [
            lambda: self.watts,
            lambda: self.vars,
            lambda: self.power_factor,
            lambda: self.phase_volts,
            lambda: self.phase_volts,
            lambda: self.phase_volts,
        ]
        for position, value in enumerate(measured):
            binding.read(BI, 94 + 2 * position, high(value, 2 * position))
            binding.read(BI, 95 + 2 * position, low(value, 2 * position + 1))
        binding.read(BI, 106, lambda: False)

        for index in range(4):
            binding.read(CTR, index, _energy(self, index))
selected_curve property 
selected_curve: int

The number of the curve the edit window is showing.

line_volts property 
line_volts: float

Measured line-to-line voltage at the connection point.

nominal_volts property 
nominal_volts: float

What a curve's percent voltage is a percentage of: the reference plus its offset.

connected property 
connected: bool

Whether the connect switch is closed.

starting property 
starting: bool

Whether a start has been commanded and is not yet complete.

stopping property 
stopping: bool

Whether a stop has been commanded and is not yet complete.

energized property 
energized: bool

Whether the DER is started and connected, and so exchanging power.

volt_amperes property 
volt_amperes: float

Apparent power at the connection point.

power_factor property 
power_factor: float

Power factor magnitude; unity when nothing is flowing.

amperes property 
amperes: float

Current per phase.

volt_watt_limit 
volt_watt_limit() -> float | None

The active power volt-watt allows at this voltage, or None when it sets no limit.

Source code in src/py1815/profile/der.py 
286
287
288
289
290
291
def volt_watt_limit(self) -> float | None:
    """The active power volt-watt allows at this voltage, or None when it sets no limit."""
    if not self._on(BO_ENABLE_VOLT_WATT):
        return None
    share = self._along(AO_VOLT_WATT_CURVE, (Y_PERCENT_MAX_WATTS,))
    return None if share is None else share * self.ratings.watts
volt_var_target 
volt_var_target() -> float | None

The reactive power volt-var calls for at this voltage, or None when it calls for none.

Source code in src/py1815/profile/der.py 
293
294
295
296
297
298
def volt_var_target(self) -> float | None:
    """The reactive power volt-var calls for at this voltage, or None when it calls for none."""
    if not self._on(BO_ENABLE_VOLT_VAR):
        return None
    share = self._along(AO_VOLT_VAR_CURVE, (Y_PERCENT_MAX_VARS, Y_PERCENT_AVAILABLE_VARS))
    return None if share is None else share * self.ratings.vars
available_watts 
available_watts() -> float

What the primary source could deliver now: a slow swell around 75%.

Source code in src/py1815/profile/der.py 
320
321
322
def available_watts(self) -> float:
    """What the primary source could deliver now: a slow swell around 75%."""
    return self.ratings.watts * (0.75 + 0.2 * math.sin(2 * math.pi * self.elapsed / 600.0))
step 
step(seconds: float) -> None

Move the simulation forward.

Source code in src/py1815/profile/der.py 
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
def step(self, seconds: float) -> None:
    """Move the simulation forward."""
    if seconds <= 0:
        return
    self.elapsed += seconds
    if self._transition is not None:
        target, remaining = self._transition
        if remaining <= seconds:
            self.started, self._transition = target, None
        else:
            self._transition = (target, remaining - seconds)
    target_watts, target_vars = self._targets()
    share = 1.0 - math.exp(-seconds / _RESPONSE_SECONDS)
    self.watts += (target_watts - self.watts) * share
    self.vars += (target_vars - self.vars) * share

    ratings = self.ratings
    self.phase_volts = ratings.phase_volts * (
        1.0 + 0.004 * math.sin(self.elapsed / 37.0) + self._random.gauss(0.0, 0.0004)
    )
    self.hertz = (
        ratings.hertz + 0.01 * math.sin(self.elapsed / 53.0) + self._random.gauss(0.0, 0.002)
    )

    hours = seconds / 3600.0
    self.energy[0] += max(self.watts, 0.0) * hours
    self.energy[1] += max(-self.watts, 0.0) * hours
    self.energy[2] += max(self.vars, 0.0) * hours
    self.energy[3] += max(-self.vars, 0.0) * hours
    if self._on(BO_ENABLE_CHARGE_DISCHARGE):
        moved = -self.watts * hours / ratings.storage_watt_hours * 100.0
        self.state_of_charge = max(0.0, min(100.0, self.state_of_charge + moved))
freeze_interval_seconds 
freeze_interval_seconds() -> float | None

How often the counters freeze, or None when they are not to repeat.

Source code in src/py1815/profile/der.py 
419
420
421
422
423
424
425
def freeze_interval_seconds(self) -> float | None:
    """How often the counters freeze, or None when they are not to repeat."""
    unit = _INTERVAL_SECONDS.get(round(self._setting(AO_FREEZE_INTERVAL_UNITS)))
    interval = self._setting(AO_FREEZE_INTERVAL)
    if unit is None or interval <= 0:
        return None
    return interval * unit
bind 
bind(point_map: PointMap) -> Binding

This DER's binding: every point it serves, and how.

Source code in src/py1815/profile/der.py 
429
430
431
432
433
434
435
436
437
438
def bind(self, point_map: PointMap) -> Binding:
    """This DER's binding: every point it serves, and how."""
    binding = Binding()
    self._bind_state(binding)
    self._bind_settings(binding)
    self._bind_functions(binding)
    self._bind_curves(binding)
    self._bind_nameplate(binding, point_map)
    self._bind_meter(binding)
    return binding

Simulation

A simulated DER and its outstation, advanced together.

Kept apart from any event loop so it can be stepped by a test as easily as by a timer: each step moves the device, freezes the counters when their interval has passed, and buffers whatever changed.

Source code in src/py1815/profile/der.py 
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
class Simulation:
    """A simulated DER and its outstation, advanced together.

    Kept apart from any event loop so it can be stepped by a test as easily as
    by a timer: each step moves the device, freezes the counters when their
    interval has passed, and buffers whatever changed.
    """

    def __init__(self, der: ReferenceDer, outstation: DerOutstation) -> None:
        self.der = der
        self.outstation = outstation
        self._since_freeze = 0.0
        # The profile has the counters freeze "beginning at startup", which is
        # also what gives every frozen counter a value to report from the
        # first integrity poll.
        outstation.freeze_all()
        outstation.poll()

    def advance(self, seconds: float) -> None:
        """Move the device forward, freeze if the interval has passed, and poll."""
        self.der.step(seconds)
        self._since_freeze += seconds
        interval = self.der.freeze_interval_seconds()
        if interval is not None and self._since_freeze >= interval:
            self.outstation.freeze_all()
            self._since_freeze = 0.0
        self.outstation.poll()
advance 
advance(seconds: float) -> None

Move the device forward, freeze if the interval has passed, and poll.

Source code in src/py1815/profile/der.py 
807
808
809
810
811
812
813
814
815
def advance(self, seconds: float) -> None:
    """Move the device forward, freeze if the interval has passed, and poll."""
    self.der.step(seconds)
    self._since_freeze += seconds
    interval = self.der.freeze_interval_seconds()
    if interval is not None and self._since_freeze >= interval:
        self.outstation.freeze_all()
        self._since_freeze = 0.0
    self.outstation.poll()

build 

build(
    point_map: PointMap, *, seed: int = 0, **options: Any
) -> Simulation

Build a simulation: the reference DER bound to an outstation for point_map.

options are passed to :class:~py1815.profile.outstation.DerOutstation, for example read_only=True or event_capacity=500.

Source code in src/py1815/profile/der.py 
818
819
820
821
822
823
824
825
826
def build(point_map: PointMap, *, seed: int = 0, **options: Any) -> Simulation:
    """Build a simulation: the reference DER bound to an outstation for *point_map*.

    ``options`` are passed to :class:`~py1815.profile.outstation.DerOutstation`,
    for example ``read_only=True`` or ``event_capacity=500``.
    """
    der = ReferenceDer(seed=seed)
    outstation = DerOutstation(point_map, der.bind(point_map), **options)
    return Simulation(der, outstation)

Probe

py1815.profile.probe

A short reader of the objects in an IEEE 1815.2 outstation's responses.

It was the parser of a probe that polled an outstation once; the poll is now the master's (:mod:py1815.master.poll). The parser stays because tests of the outstation read its responses with it: it is a second reading of the same octets, written apart from :mod:py1815.decode, with its own table of sizes.

It reads the objects the IEEE 1815.2 profile selects and stops, saying so, at the first it does not know: an object's width is the only way to find the one after it.

Copyright 2026 DER Security Corp. Licensed under the Apache License, Version 2.0.

ProbeError

Bases: RuntimeError

A response that could not be read.

Source code in src/py1815/profile/probe.py 
53
54
class ProbeError(RuntimeError):
    """A response that could not be read."""

Value dataclass

One object as it was read.

Source code in src/py1815/profile/probe.py 
57
58
59
60
61
62
63
64
65
@dataclass(frozen=True)
class Value:
    """One object as it was read."""

    group: int
    variation: int
    index: int
    value: float
    flags: int | None

parse_objects 

parse_objects(
    body: bytes,
) -> tuple[list[Value], list[Value]]

The static and event objects of a response body, in the order sent.

Source code in src/py1815/profile/probe.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def parse_objects(body: bytes) -> tuple[list[Value], list[Value]]:
    """The static and event objects of a response body, in the order sent."""
    static: list[Value] = []
    events: list[Value] = []
    offset = 0
    while offset < len(body):
        if offset + 3 > len(body):
            raise ProbeError("a response ends inside an object header")
        group, variation, qualifier = body[offset : offset + 3]
        offset += 3
        if group == _COMMON_TIME and qualifier == QualifierCode.UINT8_COUNT:
            # The time a run of relative-time events counts from. Read past:
            # this probe reports what changed and not when.
            count = body[offset] if offset < len(body) else 0
            offset += 1 + count * _TIME_SIZE
            if offset > len(body):
                raise ProbeError("a response ends inside a common time of occurrence")
            continue
        size = _SIZES.get((group, variation))
        if size is None:
            raise ProbeError(f"group {group} variation {variation} is not one this probe reads")
        if qualifier in (QualifierCode.UINT8_START_STOP, QualifierCode.UINT16_START_STOP):
            wide = qualifier == QualifierCode.UINT16_START_STOP
            if offset + (4 if wide else 2) > len(body):
                raise ProbeError("a response ends inside an object range")
            start, stop = struct.unpack_from("<HH" if wide else "<BB", body, offset)
            offset += 4 if wide else 2
            for index in range(start, stop + 1):
                static.append(_value(group, variation, index, body[offset : offset + size]))
                offset += size
        elif qualifier in (
            QualifierCode.UINT8_COUNT_UINT8_INDEX,
            QualifierCode.UINT16_COUNT_UINT16_INDEX,
        ):
            wide = qualifier == QualifierCode.UINT16_COUNT_UINT16_INDEX
            width = 2 if wide else 1
            if offset + width > len(body):
                raise ProbeError("a response ends inside an object count")
            count = int.from_bytes(body[offset : offset + width], "little")
            offset += width
            for _ in range(count):
                if offset + width > len(body):
                    raise ProbeError("a response ends inside an object index")
                index = int.from_bytes(body[offset : offset + width], "little")
                offset += width
                events.append(_value(group, variation, index, body[offset : offset + size]))
                offset += size
        else:
            raise ProbeError(f"qualifier 0x{qualifier:02X} is not one this probe reads")
        if offset > len(body):
            raise ProbeError("a response ends inside an object")
    return static, events