Skip to content

Protocol layers

The layers below the point map. Each is testable without the ones above it.

Checksum

py1815.crc

CRC-16/DNP, the checksum every DNP3 frame carries several of.

One function over bytes, with no notion of frames or blocks. The link layer decides what gets covered; this decides what the answer is.

The parameters are CRC-16/DNP: polynomial 0x3D65, initial value zero, reflected input and output, final XOR 0xFFFF. Reflection is why the table is built from 0xA6BC -- 0x3D65 with its bits reversed -- rather than from the polynomial as written. The catalogue check value for this parameter set is 0xEA82 over the ASCII digits 123456789, and the test suite pins it: a CRC that is wrong in a way the implementation agrees with is invisible until a real master rejects every frame.

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

compute 

compute(data: bytes) -> int

The CRC over data.

Source code in src/py1815/crc.py 
45
46
47
48
49
50
def compute(data: bytes) -> int:
    """The CRC over ``data``."""
    remainder = 0x0000
    for byte in data:
        remainder = (remainder >> 8) ^ _TABLE[(remainder ^ byte) & 0xFF]
    return remainder ^ _XOR_OUT

encode 

encode(data: bytes) -> bytes

data followed by its CRC, in the order it goes on the wire.

Little-endian, which is the one thing about this that cannot be derived from the polynomial: DNP3 transmits the low octet first.

Source code in src/py1815/crc.py 
53
54
55
56
57
58
59
def encode(data: bytes) -> bytes:
    """``data`` followed by its CRC, in the order it goes on the wire.

    Little-endian, which is the one thing about this that cannot be derived from
    the polynomial: DNP3 transmits the low octet first.
    """
    return data + struct.pack("<H", compute(data))

verify 

verify(block: bytes) -> bool

Whether block ends in a CRC that matches the rest of it.

Source code in src/py1815/crc.py 
62
63
64
65
66
67
68
def verify(block: bytes) -> bool:
    """Whether ``block`` ends in a CRC that matches the rest of it."""
    if len(block) <= SIZE:
        return False
    body, checksum = block[:-SIZE], block[-SIZE:]
    carried: int = struct.unpack("<H", checksum)[0]
    return carried == compute(body)

The DNP3 data link layer: frames in, frames out, and nothing above them.

This module knows about start octets, the control byte, link addresses and where the CRCs go. It does not know what a transport segment is, what an outstation does, or where the values come from. Kept separate because these rules are testable without a socket, and they are stable in a way the layers above them are not.

The frame is IEC 60870-5 FT3 as DNP3 profiles it: a ten-octet header whose last two octets are a CRC over the other eight, followed by user data in blocks of sixteen octets, each block followed by its own CRC. That shape is why a frame carrying 250 octets of user data occupies 292 on the wire.

Nothing here trusts a length field before its CRC has been checked. A corrupt or hostile header would otherwise size a read, and the reader would wait for octets that are never coming while holding what it already has.

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

Broadcast

Bases: IntEnum

Destination addresses that mean "every outstation".

Source code in src/py1815/link.py 
61
62
63
64
65
66
class Broadcast(IntEnum):
    """Destination addresses that mean "every outstation"."""

    NO_CONFIRM = 0xFFFD
    SHALL_CONFIRM = 0xFFFE
    OPTIONAL_CONFIRM = 0xFFFF

PrimaryFunction

Bases: IntEnum

Function codes a primary station sends.

Source code in src/py1815/link.py 
69
70
71
72
73
74
75
76
class PrimaryFunction(IntEnum):
    """Function codes a primary station sends."""

    RESET_LINK_STATES = 0
    TEST_LINK_STATES = 2
    CONFIRMED_USER_DATA = 3
    UNCONFIRMED_USER_DATA = 4
    REQUEST_LINK_STATUS = 9

SecondaryFunction

Bases: IntEnum

Function codes a secondary station answers with.

Source code in src/py1815/link.py 
79
80
81
82
83
84
85
class SecondaryFunction(IntEnum):
    """Function codes a secondary station answers with."""

    ACK = 0
    NACK = 1
    LINK_STATUS = 11
    NOT_SUPPORTED = 15

LinkFrameError

Bases: Exception

A frame that cannot be parsed, with the reason it could not be.

The reason is carried rather than logged here: this module decides what is malformed, and the session decides how a malformed frame is reported.

Source code in src/py1815/link.py 
88
89
90
91
92
93
class LinkFrameError(Exception):
    """A frame that cannot be parsed, with the reason it could not be.

    The reason is carried rather than logged here: this module decides what is
    malformed, and the session decides how a malformed frame is reported.
    """

LinkFrame dataclass

One parsed frame.

Source code in src/py1815/link.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
@dataclass(frozen=True)
class LinkFrame:
    """One parsed frame."""

    control: int
    destination: int
    source: int
    payload: bytes

    @property
    def from_master(self) -> bool:
        """The DIR bit: set on a frame a master sent."""
        return bool(self.control & MASK_DIR)

    @property
    def is_primary(self) -> bool:
        """The PRM bit: set on a request, clear on the answer to one."""
        return bool(self.control & MASK_PRM)

    @property
    def fcb(self) -> bool:
        return bool(self.control & MASK_FCB)

    @property
    def fcv(self) -> bool:
        """FCV on a primary frame. On a secondary frame this bit is DFC."""
        return bool(self.control & MASK_FCV)

    @property
    def dfc(self) -> bool:
        """DFC on a secondary frame. On a primary frame this bit is FCV."""
        return bool(self.control & MASK_FCV)

    @property
    def function(self) -> int:
        return self.control & MASK_FUNC

    @property
    def is_broadcast(self) -> bool:
        return self.destination in tuple(Broadcast)
from_master property 
from_master: bool

The DIR bit: set on a frame a master sent.

is_primary property 
is_primary: bool

The PRM bit: set on a request, clear on the answer to one.

fcv property 
fcv: bool

FCV on a primary frame. On a secondary frame this bit is DFC.

dfc property 
dfc: bool

DFC on a secondary frame. On a primary frame this bit is FCV.

FrameReader

Turns a byte stream into frames, and resynchronizes when it has to.

TCP delivers a stream, not frames: one read can carry half a frame, three frames, or a frame followed by noise. This holds what it has and yields what is complete.

Resynchronization discards two octets, not the frame the header claims. A header whose CRC fails may not be a header at all -- the start octets can occur inside user data of a frame whose beginning was lost -- so its length field is not evidence of anything. Skipping past the start octets and searching again recovers at the next real frame; trusting the length would skip into the middle of one and stay out of step.

Source code in src/py1815/link.py 
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
class FrameReader:
    """Turns a byte stream into frames, and resynchronizes when it has to.

    TCP delivers a stream, not frames: one read can carry half a frame, three
    frames, or a frame followed by noise. This holds what it has and yields what
    is complete.

    **Resynchronization discards two octets, not the frame the header claims.**
    A header whose CRC fails may not be a header at all -- the start octets can
    occur inside user data of a frame whose beginning was lost -- so its length
    field is not evidence of anything. Skipping past the start octets and
    searching again recovers at the next real frame; trusting the length would
    skip into the middle of one and stay out of step.
    """

    def __init__(self) -> None:
        self._buffer = bytearray()

    def feed(self, data: bytes) -> list[LinkFrame]:
        """Add received octets and return every frame they completed."""
        self._buffer += data
        frames: list[LinkFrame] = []
        while True:
            frame = self._take()
            if frame is None:
                return frames
            frames.append(frame)

    def _take(self) -> LinkFrame | None:
        while True:
            start = self._buffer.find(START)
            if start == -1:
                # Keep a single trailing octet: it may be the first half of the
                # start sequence, with its partner in the next read.
                self._buffer = self._buffer[-1:] if self._buffer[-1:] == START[:1] else bytearray()
                return None
            del self._buffer[:start]

            if len(self._buffer) < HEADER_SIZE:
                return None
            if not crc.verify(bytes(self._buffer[:HEADER_SIZE])):
                del self._buffer[: len(START)]
                continue

            length = self._buffer[2]
            if length < MIN_LENGTH:
                del self._buffer[: len(START)]
                continue

            total = frame_size(length - _LENGTH_OVERHEAD)
            if len(self._buffer) < total:
                return None

            candidate = bytes(self._buffer[:total])
            try:
                frame = parse(candidate)
            except LinkFrameError:
                # The header verified, so the frame's extent is known and a bad
                # data block only costs this frame. Dropping the whole of it
                # keeps the reader aligned on the next one.
                del self._buffer[:total]
                continue
            del self._buffer[:total]
            return frame
feed 
feed(data: bytes) -> list[LinkFrame]

Add received octets and return every frame they completed.

Source code in src/py1815/link.py 
249
250
251
252
253
254
255
256
257
def feed(self, data: bytes) -> list[LinkFrame]:
    """Add received octets and return every frame they completed."""
    self._buffer += data
    frames: list[LinkFrame] = []
    while True:
        frame = self._take()
        if frame is None:
            return frames
        frames.append(frame)

control_byte 

control_byte(
    *,
    from_master: bool,
    primary: bool,
    function: int,
    fcb: bool = False,
    fcv: bool = False,
) -> int

Assemble a control octet from what each bit means.

Source code in src/py1815/link.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def control_byte(
    *,
    from_master: bool,
    primary: bool,
    function: int,
    fcb: bool = False,
    fcv: bool = False,
) -> int:
    """Assemble a control octet from what each bit means."""
    control = function & MASK_FUNC
    if from_master:
        control |= MASK_DIR
    if primary:
        control |= MASK_PRM
    if fcb:
        control |= MASK_FCB
    if fcv:
        control |= MASK_FCV
    return control

block_count 

block_count(user_data_length: int) -> int

How many CRC-carrying blocks user_data_length octets occupy.

Source code in src/py1815/link.py 
159
160
161
def block_count(user_data_length: int) -> int:
    """How many CRC-carrying blocks ``user_data_length`` octets occupy."""
    return -(-user_data_length // DATA_BLOCK_SIZE)

frame_size 

frame_size(user_data_length: int) -> int

The octets a frame carrying this much user data occupies on the wire.

Source code in src/py1815/link.py 
164
165
166
def frame_size(user_data_length: int) -> int:
    """The octets a frame carrying this much user data occupies on the wire."""
    return HEADER_SIZE + user_data_length + block_count(user_data_length) * crc.SIZE

build 

build(
    control: int,
    destination: int,
    source: int,
    payload: bytes = b"",
) -> bytes

One frame, ready to send.

Source code in src/py1815/link.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def build(control: int, destination: int, source: int, payload: bytes = b"") -> bytes:
    """One frame, ready to send."""
    if len(payload) > MAX_USER_DATA:
        raise LinkFrameError(
            f"user data is {len(payload)} octets, and a frame carries at most {MAX_USER_DATA}"
        )
    for name, address in (("destination", destination), ("source", source)):
        if not 0 <= address <= 0xFFFF:
            raise LinkFrameError(f"{name} address {address} does not fit two octets")
    header = struct.pack(
        "<2sBBHH", START, _LENGTH_OVERHEAD + len(payload), control, destination, source
    )
    frame = bytearray(crc.encode(header))
    for start in range(0, len(payload), DATA_BLOCK_SIZE):
        frame += crc.encode(payload[start : start + DATA_BLOCK_SIZE])
    return bytes(frame)

parse 

parse(frame: bytes) -> LinkFrame

Parse one complete frame, or say why it is not one.

Source code in src/py1815/link.py 
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 parse(frame: bytes) -> LinkFrame:
    """Parse one complete frame, or say why it is not one."""
    if len(frame) < HEADER_SIZE:
        raise LinkFrameError(f"frame is {len(frame)} octets, shorter than a header")
    if not frame.startswith(START):
        raise LinkFrameError("frame does not begin with the start octets")
    if not crc.verify(frame[:HEADER_SIZE]):
        raise LinkFrameError("header CRC does not match")

    length = frame[2]
    if length < MIN_LENGTH:
        raise LinkFrameError(f"length octet is {length}, below the minimum of {MIN_LENGTH}")
    user_data_length = length - _LENGTH_OVERHEAD
    expected = frame_size(user_data_length)
    if len(frame) != expected:
        raise LinkFrameError(
            f"frame is {len(frame)} octets, and its length octet describes {expected}"
        )

    control = frame[3]
    destination, source = struct.unpack("<HH", frame[4:_HEADER_BODY])

    payload = bytearray()
    offset = HEADER_SIZE
    remaining = user_data_length
    while remaining:
        size = min(remaining, DATA_BLOCK_SIZE)
        block = frame[offset : offset + size + crc.SIZE]
        if not crc.verify(block):
            raise LinkFrameError(f"data block CRC at octet {offset} does not match")
        payload += block[:size]
        offset += size + crc.SIZE
        remaining -= size

    return LinkFrame(
        control=control, destination=destination, source=source, payload=bytes(payload)
    )

is_valid_address 

is_valid_address(address: int) -> bool

Whether an address may identify a station, rather than being reserved.

Source code in src/py1815/link.py 
226
227
228
def is_valid_address(address: int) -> bool:
    """Whether an address may identify a station, rather than being reserved."""
    return 0 <= address <= MAX_ADDRESS

Transport

py1815.transport

The DNP3 transport function: one application fragment across several frames.

The smallest layer in the protocol -- a single octet of header carrying FIR, FIN and a six-bit sequence number -- and the one with the most ways to be subtly wrong, because every rule it enforces is about what to do with a series that did not arrive as promised.

Segmentation is deterministic and reassembly is suspicious. A receiver that accepted segments in any order, or kept accumulating them, would hand the application layer fragments assembled out of two different requests, or hold memory for a series whose sender has gone away.

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

TransportError

Bases: Exception

A segment that breaks a rule of the transport function.

Raised rather than swallowed so the session can report it. Every case leaves the reassembler empty: a series that broke its own sequence cannot be continued, and continuing it is how two requests become one fragment.

Source code in src/py1815/transport.py 
37
38
39
40
41
42
43
class TransportError(Exception):
    """A segment that breaks a rule of the transport function.

    Raised rather than swallowed so the session can report it. Every case leaves
    the reassembler empty: a series that broke its own sequence cannot be
    continued, and continuing it is how two requests become one fragment.
    """

Reassembler

Collects segments into application fragments.

Source code in src/py1815/transport.py 
 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
class Reassembler:
    """Collects segments into application fragments."""

    def __init__(self, *, max_fragment: int = DEFAULT_MAX_FRAGMENT) -> None:
        self._max_fragment = max_fragment
        self._buffer = bytearray()
        self._expected: int | None = None
        #: The segment accepted last, for telling a repeat of it from a
        #: different segment under the same sequence number.
        self._previous: bytes | None = None

    @property
    def in_progress(self) -> bool:
        return self._expected is not None

    def reset(self) -> None:
        """Abandon any partial fragment.

        The session calls this when the association restarts: segments from
        before a link reset must not be completed by segments from after one.
        """
        self._buffer = bytearray()
        self._expected = None
        self._previous = None

    def add(self, tpdu: bytes) -> bytes | None:
        """Accept one segment, returning the fragment it completed, if any.

        A fragment with nothing in it is not returned: a series that
        carried no application data has nothing to hand to the
        application layer.
        """
        # Reset before refusing, not after. These two checks fire before any
        # state is touched, and leaving a partial series in place let the next
        # segment carrying the expected sequence append to data from before the
        # malformed one -- splicing two requests into one fragment, which is
        # exactly what the rest of this class exists to prevent.
        if not tpdu:
            self.reset()
            raise TransportError("segment is empty")
        if len(tpdu) > MAX_SEGMENT:
            self.reset()
            raise TransportError(f"segment is {len(tpdu)} octets, and the maximum is {MAX_SEGMENT}")

        header = tpdu[0]
        first = bool(header & FIR_MASK)
        final = bool(header & FIN_MASK)
        sequence = header & SEQ_MASK
        payload = tpdu[1:]

        if first:
            # A FIR arriving mid-series is not an error: it is the sender
            # starting again, which is exactly what it does after a timeout.
            # Anything already held belongs to the abandoned series.
            self._buffer = bytearray()
        else:
            if self._expected is None:
                self.reset()
                raise TransportError("segment continues a series that was never started")
            if self._previous is not None and sequence == self._previous[0] & SEQ_MASK:
                # The sequence number of the segment just accepted. Sent
                # again octet for octet, it is a link-layer repeat and is
                # dropped on its own; anything else under that number means
                # the series cannot be trusted.
                if tpdu == self._previous:
                    return None
                self.reset()
                raise TransportError(f"segment {sequence} arrived twice with different contents")
            if sequence != self._expected:
                expected = self._expected
                self.reset()
                raise TransportError(f"segment sequence is {sequence}, and {expected} was expected")

        if len(self._buffer) + len(payload) > self._max_fragment:
            self.reset()
            raise TransportError(f"fragment exceeds {self._max_fragment} octets and was abandoned")

        self._buffer += payload
        if final:
            fragment = bytes(self._buffer)
            self.reset()
            return fragment or None

        self._expected = (sequence + 1) % SEQUENCE_MODULUS
        self._previous = bytes(tpdu)
        return None
reset 
reset() -> None

Abandon any partial fragment.

The session calls this when the association restarts: segments from before a link reset must not be completed by segments from after one.

Source code in src/py1815/transport.py 
90
91
92
93
94
95
96
97
98
def reset(self) -> None:
    """Abandon any partial fragment.

    The session calls this when the association restarts: segments from
    before a link reset must not be completed by segments from after one.
    """
    self._buffer = bytearray()
    self._expected = None
    self._previous = None
add 
add(tpdu: bytes) -> bytes | None

Accept one segment, returning the fragment it completed, if any.

A fragment with nothing in it is not returned: a series that carried no application data has nothing to hand to the application layer.

Source code in src/py1815/transport.py 
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
def add(self, tpdu: bytes) -> bytes | None:
    """Accept one segment, returning the fragment it completed, if any.

    A fragment with nothing in it is not returned: a series that
    carried no application data has nothing to hand to the
    application layer.
    """
    # Reset before refusing, not after. These two checks fire before any
    # state is touched, and leaving a partial series in place let the next
    # segment carrying the expected sequence append to data from before the
    # malformed one -- splicing two requests into one fragment, which is
    # exactly what the rest of this class exists to prevent.
    if not tpdu:
        self.reset()
        raise TransportError("segment is empty")
    if len(tpdu) > MAX_SEGMENT:
        self.reset()
        raise TransportError(f"segment is {len(tpdu)} octets, and the maximum is {MAX_SEGMENT}")

    header = tpdu[0]
    first = bool(header & FIR_MASK)
    final = bool(header & FIN_MASK)
    sequence = header & SEQ_MASK
    payload = tpdu[1:]

    if first:
        # A FIR arriving mid-series is not an error: it is the sender
        # starting again, which is exactly what it does after a timeout.
        # Anything already held belongs to the abandoned series.
        self._buffer = bytearray()
    else:
        if self._expected is None:
            self.reset()
            raise TransportError("segment continues a series that was never started")
        if self._previous is not None and sequence == self._previous[0] & SEQ_MASK:
            # The sequence number of the segment just accepted. Sent
            # again octet for octet, it is a link-layer repeat and is
            # dropped on its own; anything else under that number means
            # the series cannot be trusted.
            if tpdu == self._previous:
                return None
            self.reset()
            raise TransportError(f"segment {sequence} arrived twice with different contents")
        if sequence != self._expected:
            expected = self._expected
            self.reset()
            raise TransportError(f"segment sequence is {sequence}, and {expected} was expected")

    if len(self._buffer) + len(payload) > self._max_fragment:
        self.reset()
        raise TransportError(f"fragment exceeds {self._max_fragment} octets and was abandoned")

    self._buffer += payload
    if final:
        fragment = bytes(self._buffer)
        self.reset()
        return fragment or None

    self._expected = (sequence + 1) % SEQUENCE_MODULUS
    self._previous = bytes(tpdu)
    return None

header_byte 

header_byte(
    *, first: bool, final: bool, sequence: int
) -> int

Assemble a transport header octet.

Source code in src/py1815/transport.py 
46
47
48
49
50
51
52
53
def header_byte(*, first: bool, final: bool, sequence: int) -> int:
    """Assemble a transport header octet."""
    header = sequence % SEQUENCE_MODULUS
    if first:
        header |= FIR_MASK
    if final:
        header |= FIN_MASK
    return header

segment 

segment(
    fragment: bytes, *, first_sequence: int = 0
) -> list[bytes]

Split an application fragment into segments, in the order they are sent.

An empty fragment still produces one segment. It is a legal thing to send -- a response can carry a header and no objects -- and returning nothing would silently drop it.

Source code in src/py1815/transport.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def segment(fragment: bytes, *, first_sequence: int = 0) -> list[bytes]:
    """Split an application fragment into segments, in the order they are sent.

    An empty fragment still produces one segment. It is a legal thing to send --
    a response can carry a header and no objects -- and returning nothing would
    silently drop it.
    """
    chunks = [fragment[i : i + MAX_PAYLOAD] for i in range(0, len(fragment), MAX_PAYLOAD)] or [b""]
    segments = []
    for index, chunk in enumerate(chunks):
        header = header_byte(
            first=index == 0,
            final=index == len(chunks) - 1,
            sequence=first_sequence + index,
        )
        segments.append(bytes([header]) + chunk)
    return segments

Application

py1815.application

The DNP3 application layer: request headers in, response headers out.

This module reads the shape of a request -- the application control octet, the function code, and the object headers that say what is being asked for -- and writes the shape of a response. It knows nothing about what any group means, how wide a point is, or where the values come from. Object semantics belong to the module that owns the point map; this one owns the envelope.

Where it stops is deliberate. A request that carries object data is parsed as far as its first object header. Walking past that means knowing how many octets each group and variation occupies, which is the point map's knowledge, and a parser that guessed would misread the second header of every write. Requests that carry no object data -- a READ, an unsolicited enable, an assign-class -- are parsed in full, and those are the whole of the read path.

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

FunctionCode

Bases: IntEnum

Function codes, including the ones this outstation refuses.

The refused ones are named rather than left as integers because answering "function not supported" is a different thing from failing to recognize the request at all, and the log line should be able to say which.

Source code in src/py1815/application.py 
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
class FunctionCode(IntEnum):
    """Function codes, including the ones this outstation refuses.

    The refused ones are named rather than left as integers because answering
    "function not supported" is a different thing from failing to recognize the
    request at all, and the log line should be able to say which.
    """

    CONFIRM = 0x00
    READ = 0x01
    WRITE = 0x02
    SELECT = 0x03
    OPERATE = 0x04
    DIRECT_OPERATE = 0x05
    DIRECT_OPERATE_NR = 0x06
    IMMED_FREEZE = 0x07
    IMMED_FREEZE_NR = 0x08
    FREEZE_CLEAR = 0x09
    FREEZE_CLEAR_NR = 0x0A
    FREEZE_AT_TIME = 0x0B
    FREEZE_AT_TIME_NR = 0x0C
    COLD_RESTART = 0x0D
    WARM_RESTART = 0x0E
    INITIALIZE_DATA = 0x0F
    INITIALIZE_APPLICATION = 0x10
    START_APPLICATION = 0x11
    STOP_APPLICATION = 0x12
    SAVE_CONFIGURATION = 0x13
    ENABLE_UNSOLICITED = 0x14
    DISABLE_UNSOLICITED = 0x15
    ASSIGN_CLASS = 0x16
    DELAY_MEASURE = 0x17
    RECORD_CURRENT_TIME = 0x18
    OPEN_FILE = 0x19
    CLOSE_FILE = 0x1A
    DELETE_FILE = 0x1B
    GET_FILE_INFO = 0x1C
    AUTHENTICATE_FILE = 0x1D
    ABORT_FILE = 0x1E
    ACTIVATE_CONFIG = 0x1F
    AUTH_REQUEST = 0x20
    AUTH_REQUEST_NO_ACK = 0x21
    RESPONSE = 0x81
    UNSOLICITED_RESPONSE = 0x82
    AUTH_RESPONSE = 0x83

QualifierCode

Bases: IntEnum

Qualifiers this outstation accepts in a request.

The high nibble is the index or size prefix and the low nibble is the range specifier, which is why these values are not contiguous.

Source code in src/py1815/application.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class QualifierCode(IntEnum):
    """Qualifiers this outstation accepts in a request.

    The high nibble is the index or size prefix and the low nibble is the range
    specifier, which is why these values are not contiguous.
    """

    UINT8_START_STOP = 0x00
    UINT16_START_STOP = 0x01
    ALL_OBJECTS = 0x06
    UINT8_COUNT = 0x07
    UINT16_COUNT = 0x08
    UINT8_COUNT_UINT8_INDEX = 0x17
    UINT16_COUNT_UINT16_INDEX = 0x28

IINBit

Bases: IntEnum

Internal indication bits, by the octet each lives in.

Source code in src/py1815/application.py 
148
149
150
151
152
153
154
155
156
157
158
class IINBit(IntEnum):
    """Internal indication bits, by the octet each lives in."""

    BROADCAST = 0x01
    CLASS_1_EVENTS = 0x02
    CLASS_2_EVENTS = 0x04
    CLASS_3_EVENTS = 0x08
    NEED_TIME = 0x10
    LOCAL_CONTROL = 0x20
    DEVICE_TROUBLE = 0x40
    DEVICE_RESTART = 0x80

IIN2Bit

Bases: IntEnum

The second internal indication octet: what went wrong with this request.

Source code in src/py1815/application.py 
161
162
163
164
165
166
167
168
169
class IIN2Bit(IntEnum):
    """The second internal indication octet: what went wrong with this request."""

    FUNC_NOT_SUPPORTED = 0x01
    OBJECT_UNKNOWN = 0x02
    PARAM_ERROR = 0x04
    EVENT_BUFFER_OVERFLOW = 0x08
    ALREADY_EXECUTING = 0x10
    CONFIG_CORRUPT = 0x20

IIN dataclass

The two internal indication octets a response carries.

Source code in src/py1815/application.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@dataclass(frozen=True)
class IIN:
    """The two internal indication octets a response carries."""

    first: int = 0
    second: int = 0

    def __or__(self, other: IIN) -> IIN:
        return IIN(self.first | other.first, self.second | other.second)

    def is_set(self, bit: IINBit | IIN2Bit) -> bool:
        octet = self.first if isinstance(bit, IINBit) else self.second
        return bool(octet & bit)

    def to_bytes(self) -> bytes:
        return bytes([self.first & 0xFF, self.second & 0xFF])

AppControl dataclass

The application control octet.

Source code in src/py1815/application.py 
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
@dataclass(frozen=True)
class AppControl:
    """The application control octet."""

    fir: bool = True
    fin: bool = True
    con: bool = False
    uns: bool = False
    sequence: int = 0

    @classmethod
    def from_byte(cls, octet: int) -> AppControl:
        return cls(
            fir=bool(octet & FIR_MASK),
            fin=bool(octet & FIN_MASK),
            con=bool(octet & CON_MASK),
            uns=bool(octet & UNS_MASK),
            sequence=octet & SEQ_MASK,
        )

    def to_byte(self) -> int:
        octet = self.sequence % SEQUENCE_MODULUS
        if self.fir:
            octet |= FIR_MASK
        if self.fin:
            octet |= FIN_MASK
        if self.con:
            octet |= CON_MASK
        if self.uns:
            octet |= UNS_MASK
        return octet

ObjectHeader dataclass

One object header: what is being asked for, and over what range.

Source code in src/py1815/application.py 
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
@dataclass(frozen=True)
class ObjectHeader:
    """One object header: what is being asked for, and over what range."""

    group: int
    variation: int
    qualifier: QualifierCode
    start: int | None = None
    stop: int | None = None
    count: int | None = None
    #: The indices an index-prefixed qualifier names, when they are contiguous.
    #: A request carrying object data interleaves them with the objects, so they
    #: are read only where nothing follows them -- see :func:`parse_request`.
    indices: tuple[int, ...] | None = None

    @property
    def is_class_request(self) -> bool:
        return self.group == CLASS_GROUP and self.variation in CLASS_VARIATIONS

    @property
    def event_class(self) -> int | None:
        """The class this header names, or None if it does not name one."""
        if not self.is_class_request:
            return None
        return CLASS_VARIATIONS[self.variation]
event_class property 
event_class: int | None

The class this header names, or None if it does not name one.

RequestError

Bases: Exception

A request that cannot be parsed, and the indication bit that says so.

Carrying the bit is what lets the session answer a malformed request correctly without re-deriving why it was malformed. A qualifier this outstation does not implement is a parameter error; a group it does not know is an unknown object. Those are different answers, and a master acts on the difference.

Source code in src/py1815/application.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
class RequestError(Exception):
    """A request that cannot be parsed, and the indication bit that says so.

    Carrying the bit is what lets the session answer a malformed request
    correctly without re-deriving why it was malformed. A qualifier this
    outstation does not implement is a parameter error; a group it does not know
    is an unknown object. Those are different answers, and a master acts on the
    difference.
    """

    def __init__(self, reason: str, bit: IIN2Bit = IIN2Bit.PARAM_ERROR) -> None:
        super().__init__(reason)
        self.bit = bit

Request dataclass

One parsed request fragment.

Source code in src/py1815/application.py 
265
266
267
268
269
270
271
272
273
274
@dataclass(frozen=True)
class Request:
    """One parsed request fragment."""

    control: AppControl
    function: int
    headers: tuple[ObjectHeader, ...] = ()
    #: Object data this module did not consume, for a function that carries some.
    body: bytes = b""
    known_function: FunctionCode | None = field(default=None, compare=False)

ObjectBlock dataclass

One object header and the indexed objects that followed it.

Source code in src/py1815/application.py 
398
399
400
401
402
403
404
405
406
@dataclass(frozen=True)
class ObjectBlock:
    """One object header and the indexed objects that followed it."""

    header: ObjectHeader
    #: ``(index, octets)`` in the order they arrived. An index may repeat: a
    #: request naming the same point twice is a request naming it twice, and
    #: collapsing that here would answer fewer objects than were asked about.
    items: tuple[tuple[int, bytes], ...]

ResponseError

Bases: Exception

A fragment from an outstation that is not a response.

Source code in src/py1815/application.py 
532
533
class ResponseError(Exception):
    """A fragment from an outstation that is not a response."""

Response dataclass

One parsed response fragment: the envelope, and the objects unread.

Source code in src/py1815/application.py 
536
537
538
539
540
541
542
543
544
545
546
547
548
@dataclass(frozen=True)
class Response:
    """One parsed response fragment: the envelope, and the objects unread."""

    control: AppControl
    function: FunctionCode
    iin: IIN
    #: The object headers and objects, exactly as sent.
    body: bytes = b""

    @property
    def unsolicited(self) -> bool:
        return self.function is FunctionCode.UNSOLICITED_RESPONSE

parse_request 

parse_request(fragment: bytes) -> Request

Parse a request fragment, or say why it is not one.

Source code in src/py1815/application.py 
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
def parse_request(fragment: bytes) -> Request:
    """Parse a request fragment, or say why it is not one."""
    if len(fragment) < REQUEST_HEADER_SIZE:
        raise RequestError("fragment is shorter than an application header")

    control = AppControl.from_byte(fragment[0])
    function = fragment[1]
    try:
        known: FunctionCode | None = FunctionCode(function)
    except ValueError:
        known = None

    if known is FunctionCode.CONFIRM:
        if len(fragment) > REQUEST_HEADER_SIZE:
            raise RequestError("a confirmation carries no objects")
        return Request(control, function, known_function=known)

    if known is None or known not in _PARSED_FUNCTIONS:
        # Two cases, one answer. Nothing here knows an unrecognized function's
        # body layout, and a recognized function this module does not parse is
        # in the same position. Reading either's first octets as an object
        # header invents a structure and then refuses the request for failing to
        # have it. The answer a master is owed is "function not supported", and
        # producing that needs the request parsed rather than rejected. The
        # payload travels untouched in ``body``, and ``known_function`` travels
        # with it so a refusal can name what it refused.
        return Request(control, function, body=fragment[REQUEST_HEADER_SIZE:], known_function=known)

    headers: list[ObjectHeader] = []
    offset = REQUEST_HEADER_SIZE
    header_only = known in HEADER_ONLY_FUNCTIONS
    while offset < len(fragment):
        # In a request carrying no object data, an index-prefixed qualifier is
        # followed by its index list and nothing else, so the list is read here.
        # In one that carries objects, each index prefixes its own object and
        # the two interleave -- which is why parsing stops at the first header
        # there rather than walking a list that is not contiguous.
        header, offset = _parse_header(fragment, offset, indices_follow=header_only)
        headers.append(header)
        if not header_only:
            break

    return Request(
        control=control,
        function=function,
        headers=tuple(headers),
        body=fragment[offset:],
        known_function=known,
    )

parse_header_list 

parse_header_list(data: bytes) -> tuple[ObjectHeader, ...]

Object headers that follow one another with no object data between them.

For a function parse_request hands back unparsed, once the caller has decided it will answer it. A freeze names the counters it applies to the way a read names what it wants, so the same walk reads both; the session asks for it only when something is there to freeze, which keeps a refused freeze answered as an unsupported function rather than as a parse error.

Source code in src/py1815/application.py 
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
def parse_header_list(data: bytes) -> tuple[ObjectHeader, ...]:
    """Object headers that follow one another with no object data between them.

    For a function ``parse_request`` hands back unparsed, once the caller has
    decided it will answer it. A freeze names the counters it applies to the
    way a read names what it wants, so the same walk reads both; the session
    asks for it only when something is there to freeze, which keeps a refused
    freeze answered as an unsupported function rather than as a parse error.
    """
    headers: list[ObjectHeader] = []
    offset = 0
    while offset < len(data):
        header, offset = _parse_header(data, offset, indices_follow=True)
        headers.append(header)
    return tuple(headers)

parse_object_blocks 

parse_object_blocks(
    request: Request,
    size_of: Callable[[int, int], int | None],
) -> tuple[ObjectBlock, ...]

Walk a request whose objects interleave with their indices.

parse_request stops at the first header for such a request and leaves the remainder in body, because the indices do not form a list that can be read ahead of the objects. This walks that remainder: for each header, count objects each prefixed by its own index, then the next header.

size_of answers how wide one object of a group and variation is, and None where it knows of none. The layer is kept ignorant of object formats deliberately -- it parses headers and qualifiers and nothing below them -- so the caller supplies the only thing it cannot know.

Every failure here is a failure of the fragment rather than of one object, and that is not a stylistic choice. A truncated body, a count that disagrees with the octets present, or a width nothing recognizes may leave no complete object at all, and a status has to be attached to something. Raising :class:RequestError is what tells the caller to refuse the fragment rather than to answer per index.

No object count is imposed. A fragment is already bounded, and eleven octets plus an index is the smallest control a request can carry, so the ceiling is the one the transport already sets.

Source code in src/py1815/application.py 
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
def parse_object_blocks(
    request: Request, size_of: Callable[[int, int], int | None]
) -> tuple[ObjectBlock, ...]:
    """Walk a request whose objects interleave with their indices.

    ``parse_request`` stops at the first header for such a request and leaves
    the remainder in ``body``, because the indices do not form a list that can
    be read ahead of the objects. This walks that remainder: for each header,
    ``count`` objects each prefixed by its own index, then the next header.

    ``size_of`` answers how wide one object of a group and variation is, and
    ``None`` where it knows of none. The layer is kept ignorant of object
    formats deliberately -- it parses headers and qualifiers and nothing below
    them -- so the caller supplies the only thing it cannot know.

    Every failure here is a failure of the fragment rather than of one object,
    and that is not a stylistic choice. A truncated body, a count that disagrees
    with the octets present, or a width nothing recognizes may leave no complete
    object at all, and a status has to be attached to something. Raising
    :class:`RequestError` is what tells the caller to refuse the fragment rather
    than to answer per index.

    No object count is imposed. A fragment is already bounded, and eleven octets
    plus an index is the smallest control a request can carry, so the ceiling is
    the one the transport already sets.
    """
    if not request.headers:
        raise RequestError("a request carrying objects has no object header")

    blocks: list[ObjectBlock] = []
    header = request.headers[0]
    body = request.body
    offset = 0

    while True:
        width = _INDEX_WIDTHS.get(header.qualifier)
        if width is None:
            raise RequestError(
                f"qualifier 0x{int(header.qualifier):02X} prefixes no index, so the "
                "objects after it cannot be told apart"
            )
        if header.count is None:
            raise RequestError("an index-prefixed qualifier carries no count")

        size = size_of(header.group, header.variation)
        if size is None:
            raise RequestError(
                f"no object of group {header.group} variation {header.variation} is known, "
                "so its width cannot be read"
            )

        items: list[tuple[int, bytes]] = []
        for _ in range(header.count):
            end = offset + width + size
            if end > len(body):
                raise RequestError(
                    f"group {header.group} variation {header.variation} declares "
                    f"{header.count} objects and the body holds fewer"
                )
            items.append(
                (
                    int.from_bytes(body[offset : offset + width], "little"),
                    bytes(body[offset + width : end]),
                )
            )
            offset = end

        blocks.append(ObjectBlock(header=header, items=tuple(items)))

        if offset == len(body):
            return tuple(blocks)
        header, offset = _parse_header(body, offset, indices_follow=False)

object_header 

object_header(
    group: int, variation: int, *, start: int, stop: int
) -> bytes

An object header covering a contiguous range, in the narrowest qualifier.

Narrowest rather than always 16-bit because a master reads the qualifier to size the range field, and a one-octet range is two octets cheaper in every response that fits it -- which, for a fleet served in blocks, is most of them.

Source code in src/py1815/application.py 
483
484
485
486
487
488
489
490
491
492
493
494
495
496
def object_header(group: int, variation: int, *, start: int, stop: int) -> bytes:
    """An object header covering a contiguous range, in the narrowest qualifier.

    Narrowest rather than always 16-bit because a master reads the qualifier to
    size the range field, and a one-octet range is two octets cheaper in every
    response that fits it -- which, for a fleet served in blocks, is most of them.
    """
    if stop < start:
        raise ValueError(f"range {start}..{stop} ends before it begins")
    if start < 0 or stop > 0xFFFF:
        raise ValueError(f"range {start}..{stop} does not fit a 16-bit index")
    if stop <= 0xFF:
        return bytes([group, variation, QualifierCode.UINT8_START_STOP, start, stop])
    return struct.pack("<BBBHH", group, variation, QualifierCode.UINT16_START_STOP, start, stop)

build_response 

build_response(
    *,
    control: AppControl,
    iin: IIN,
    body: bytes = b"",
    function: FunctionCode = FunctionCode.RESPONSE,
) -> bytes

One response fragment, ready for the transport function.

Source code in src/py1815/application.py 
499
500
501
502
503
504
505
506
507
def build_response(
    *,
    control: AppControl,
    iin: IIN,
    body: bytes = b"",
    function: FunctionCode = FunctionCode.RESPONSE,
) -> bytes:
    """One response fragment, ready for the transport function."""
    return bytes([control.to_byte(), function, *iin.to_bytes()]) + body

null_response 

null_response(*, sequence: int, iin: IIN) -> bytes

A response carrying indications and no objects.

The answer to a request that was understood and produced nothing to say -- a refusal, an acknowledged write, an empty class poll.

Source code in src/py1815/application.py 
510
511
512
513
514
515
516
517
518
519
def null_response(*, sequence: int, iin: IIN) -> bytes:
    """A response carrying indications and no objects.

    The answer to a request that was understood and produced nothing to say --
    a refusal, an acknowledged write, an empty class poll.
    """
    return build_response(
        control=AppControl(fir=True, fin=True, sequence=sequence),
        iin=iin,
    )

parse_response 

parse_response(fragment: bytes) -> Response

Parse a response fragment, or say why it is not one.

A fragment whose function code and unsolicited bit disagree is refused: a master confirms the two kinds in different sequence series, and one that could be read either way would be confirmed in the wrong one.

Source code in src/py1815/application.py 
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
def parse_response(fragment: bytes) -> Response:
    """Parse a response fragment, or say why it is not one.

    A fragment whose function code and unsolicited bit disagree is refused: a
    master confirms the two kinds in different sequence series, and one that
    could be read either way would be confirmed in the wrong one.
    """
    if len(fragment) < RESPONSE_HEADER_SIZE:
        raise ResponseError("fragment is shorter than a response header")
    control = AppControl.from_byte(fragment[0])
    try:
        function = FunctionCode(fragment[1])
    except ValueError as exc:
        raise ResponseError(f"function code 0x{fragment[1]:02X} is not a response") from exc
    if function not in RESPONSE_FUNCTIONS:
        raise ResponseError(f"{function.name} is a request, and an outstation sends none")
    if control.uns != (function is FunctionCode.UNSOLICITED_RESPONSE):
        raise ResponseError(
            f"{function.name} arrived with the unsolicited bit {'set' if control.uns else 'clear'}"
        )
    return Response(
        control=control,
        function=function,
        iin=IIN(fragment[2], fragment[3]),
        body=bytes(fragment[RESPONSE_HEADER_SIZE:]),
    )

build_request 

build_request(
    function: FunctionCode,
    *,
    sequence: int,
    body: bytes = b"",
) -> bytes

One request fragment, ready for the transport function.

A request is always a single fragment, first and final, and never asks to be confirmed: the response is the acknowledgment.

Source code in src/py1815/application.py 
579
580
581
582
583
584
585
586
def build_request(function: FunctionCode, *, sequence: int, body: bytes = b"") -> bytes:
    """One request fragment, ready for the transport function.

    A request is always a single fragment, first and final, and never asks to
    be confirmed: the response is the acknowledgment.
    """
    control = AppControl(fir=True, fin=True, sequence=sequence)
    return bytes([control.to_byte(), function]) + body

build_confirm 

build_confirm(
    *, sequence: int, unsolicited: bool = False
) -> bytes

The confirmation of a response fragment that asked for one.

It carries the sequence number of the fragment it confirms and that fragment's unsolicited bit, which is what tells an outstation which of its two exchanges is being confirmed.

Source code in src/py1815/application.py 
589
590
591
592
593
594
595
596
597
def build_confirm(*, sequence: int, unsolicited: bool = False) -> bytes:
    """The confirmation of a response fragment that asked for one.

    It carries the sequence number of the fragment it confirms and that
    fragment's unsolicited bit, which is what tells an outstation which of
    its two exchanges is being confirmed.
    """
    control = AppControl(fir=True, fin=True, uns=unsolicited, sequence=sequence)
    return bytes([control.to_byte(), FunctionCode.CONFIRM])

all_objects_header 

all_objects_header(group: int, variation: int) -> bytes

An object header asking for every object of a group and variation.

Source code in src/py1815/application.py 
600
601
602
def all_objects_header(group: int, variation: int) -> bytes:
    """An object header asking for every object of a group and variation."""
    return bytes([group, variation, QualifierCode.ALL_OBJECTS])

class_header 

class_header(event_class: int) -> bytes

The header that asks for a class: 0 for static data, 1 to 3 for events.

Source code in src/py1815/application.py 
605
606
607
608
609
610
def class_header(event_class: int) -> bytes:
    """The header that asks for a class: 0 for static data, 1 to 3 for events."""
    variations = {number: variation for variation, number in CLASS_VARIATIONS.items()}
    if event_class not in variations:
        raise ValueError(f"there is no class {event_class}; a class is 0 to 3")
    return all_objects_header(CLASS_GROUP, variations[event_class])

index_list_header 

index_list_header(
    group: int, variation: int, indices: Sequence[int]
) -> bytes

An object header naming points one at a time, in the narrowest qualifier.

For a request that carries no object data, where the indices follow the count as a list. The order given is the order asked, and an index may repeat.

Source code in src/py1815/application.py 
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
def index_list_header(group: int, variation: int, indices: Sequence[int]) -> bytes:
    """An object header naming points one at a time, in the narrowest qualifier.

    For a request that carries no object data, where the indices follow the
    count as a list. The order given is the order asked, and an index may
    repeat.
    """
    if not indices:
        raise ValueError("an index list names at least one point")
    if min(indices) < 0 or max(indices) > 0xFFFF:
        raise ValueError("an index is 0 to 65535")
    if max(indices) <= 0xFF and len(indices) <= 0xFF:
        return bytes(
            [group, variation, QualifierCode.UINT8_COUNT_UINT8_INDEX, len(indices), *indices]
        )
    return struct.pack(
        f"<BBBH{len(indices)}H",
        group,
        variation,
        QualifierCode.UINT16_COUNT_UINT16_INDEX,
        len(indices),
        *indices,
    )