Events

Class buffers, deadbands, and the overflow they report. What changed, when, and which class it belongs to.

py1815.events

Event buffers: what changed, when, and which class it belongs to.

A static read answers "what is the value now". An event answers "it changed, and here is what it was at the time" -- which is the question a master asks when it wants a sequence rather than a snapshot, and it is why events are held rather than computed.

Three properties are the whole of this module, and each exists because of a way event reporting goes wrong in practice.

An event is generated by a deadband, not by a difference. Reporting every read of a measured value floods the buffer with noise no operator asked for. The deadband is the operator's statement of what counts as a change.

A quality change always generates an event, whatever the magnitude. A point that goes comm-lost while holding the same number has not changed value and has changed in the way that matters most. A deadband applied to quality would hide exactly the transitions someone is watching for.

A full buffer drops its oldest event and says so. The alternative -- dropping the newest -- keeps a master reading history while the present goes unreported, and silence about either is worse than the loss.

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

EventClass

Bases: IntEnum

The class an event is reported in.

Class 0 is not here: it is the static data, which is a read rather than a buffer. An event assigned class 0 is an event that is never reported, so assigning one is refused rather than quietly honored.

Source code in src/py1815/events.py

class EventClass(IntEnum):
    """The class an event is reported in.

    Class 0 is not here: it is the static data, which is a read rather than a
    buffer. An event assigned class 0 is an event that is never reported, so
    assigning one is refused rather than quietly honored.
    """

    CLASS_1 = 1
    CLASS_2 = 2
    CLASS_3 = 3

AnalogEvent dataclass

One analog input change.

Source code in src/py1815/events.py

@dataclass(frozen=True)
class AnalogEvent:
    """One analog input change."""

    index: int
    point: AnalogPoint
    timestamp_ms: int
    #: Where this event falls among every event recorded, across classes.
    #: Not part of what the event says, so two events are still equal
    #: without it; it is what puts a response in the order things happened.
    order: int = field(default=0, compare=False)

BinaryEvent dataclass

One binary input change.

Source code in src/py1815/events.py

@dataclass(frozen=True)
class BinaryEvent:
    """One binary input change."""

    index: int
    point: BinaryPoint
    timestamp_ms: int
    #: Where this event falls among every event recorded, across classes.
    #: Not part of what the event says, so two events are still equal
    #: without it; it is what puts a response in the order things happened.
    order: int = field(default=0, compare=False)
    #: Whether the clock that stamped it had been set. An event stamped
    #: by a clock that had not is reported with relative time, behind a
    #: common time that says so.
    synchronized: bool = field(default=True, compare=False)

FrozenCounterEvent dataclass

One counter, as it stood at one freeze.

Source code in src/py1815/events.py

@dataclass(frozen=True)
class FrozenCounterEvent:
    """One counter, as it stood at one freeze."""

    index: int
    point: CounterPoint
    timestamp_ms: int
    #: Where this event falls among every event recorded, across classes.
    #: Not part of what the event says, so two events are still equal
    #: without it; it is what puts a response in the order things happened.
    order: int = field(default=0, compare=False)

EventBuffers

Per-class event storage, and the deadband rules that fill it.

Holds the last reported value per point so a deadband can be applied against what the master was actually told, rather than against the previous reading. Comparing against the previous reading lets a value drift past the deadband in steps that individually fall inside it, and the master never hears about a change that has already happened.

Source code in src/py1815/events.py

class EventBuffers:
    """Per-class event storage, and the deadband rules that fill it.

    Holds the last reported value per point so a deadband can be applied against
    what the master was actually told, rather than against the previous reading.
    Comparing against the previous reading lets a value drift past the deadband
    in steps that individually fall inside it, and the master never hears about
    a change that has already happened.
    """

    def __init__(
        self, *, capacity: int = DEFAULT_CAPACITY, analog_latest_only: bool = False
    ) -> None:
        """
        Args:
            capacity: Events each class holds before its oldest is dropped.
            analog_latest_only: Keep one event per analog point, the newest.
                The default keeps every change, which is the sequence of
                record. The alternative is what the IEEE 1815.2 profile
                selects for analog inputs: a measurement that moved five times
                between polls is reported once, at its latest value, and a
                slow master is not handed a backlog of readings it has no use
                for. Binary and frozen counter events are never collapsed,
                because for those the sequence is the information.
        """
        if capacity < 1:
            # Otherwise the first event drops one from an empty buffer, and the
            # IndexError names a deque rather than the configuration.
            raise ValueError(f"capacity is {capacity}; a buffer holds at least one event")
        self._buffers = {cls: _ClassBuffer(capacity) for cls in EventClass}
        self._last_analog: dict[int, AnalogPoint] = {}
        self._last_binary: dict[int, BinaryPoint] = {}
        self._overflow_generation = 0
        self._recorded = 0
        self._analog_latest_only = analog_latest_only
        self._capacity = capacity
        #: The analog event each point currently has buffered, when only the
        #: newest is kept. What makes superseding one a lookup rather than a
        #: scan of every class on every reading.
        self._pending_analog: dict[int, AnalogEvent] = {}
        #: Whether the clock stamping events has been set. A session
        #: that asks its master for the time clears this until the
        #: time is written; binary events recorded meanwhile carry it.
        self.synchronized = True
        self._unsynchronized = 0

    @property
    def holds_unsynchronized(self) -> bool:
        """Whether any buffered binary event was stamped by a clock that had not been set."""
        return self._unsynchronized > 0

    def _forget(self, event: Event) -> None:
        """Stop counting an event that has left the buffers."""
        if isinstance(event, BinaryEvent) and not event.synchronized:
            self._unsynchronized -= 1

    def count(self, event_class: EventClass) -> int:
        return len(self._buffers[event_class].events)

    @property
    def total(self) -> int:
        return sum(len(buffer.events) for buffer in self._buffers.values())

    def overflowed(self) -> bool:
        """Whether any class has dropped an event since it was last reported."""
        return any(buffer.overflowed for buffer in self._buffers.values())

    @property
    def overflow_generation(self) -> int:
        """How many events have been lost, counted from the first.

        Monotonic, and deliberately not reset by ``clear_overflow``. A
        confirmation acknowledges the loss a particular response reported, and
        the flag alone cannot say whether more has been lost since that response
        was built -- it reads the same after one eviction and after a hundred.
        Clearing on the flag would retire a report of data loss the master has
        never been sent.
        """
        return self._overflow_generation

    @property
    def recorded(self) -> int:
        """How many events have been recorded, counted from the first.

        Monotonic, like :attr:`overflow_generation`, and for a similar reason:
        the count of events held cannot say whether anything new has happened,
        since one may have been retired or evicted for each one recorded. A
        session that has stopped retrying an unsolicited response reads this
        to learn that there is news worth starting again for.
        """
        return self._recorded

    def classes_with_events(self, excluding: Iterable[Event] = ()) -> set[EventClass]:
        """Which classes have something to report, for the indication bits.

        ``excluding`` names events a response is already carrying. A class
        whose every event is in the response being built has nothing *more*
        to report, and saying it has would send the master back for events
        it is holding.
        """
        carried = set(map(id, excluding))
        if not carried:
            return {cls for cls, buffer in self._buffers.items() if buffer.events}
        return {
            cls
            for cls, buffer in self._buffers.items()
            if any(id(event) not in carried for event in buffer.events)
        }

    def _next_order(self) -> int:
        self._recorded += 1
        return self._recorded

    def peek(self, event_class: EventClass, limit: int | None = None) -> list[Event]:
        """The oldest events of a class, without removing them.

        Reads do not consume. An event leaves the buffer when the master
        confirms the response carrying it, because a response that never arrives
        must not have taken the only copy with it.

        ``limit`` is taken from the front, and taken while the deque is walked
        rather than after. The difference is the whole point of the parameter:
        materializing the class and slicing it afterwards costs the buffer on
        every call, and ``capacity`` has no upper bound, so a caller asking for
        the few events that fit a response would still pay for all of them.
        """
        events = self._buffers[event_class].events
        if limit is None:
            return list(events)
        # Clamped to the length as well as to zero. ``islice`` refuses a stop
        # above ``sys.maxsize``, where a plain slice simply returns everything,
        # and a limit is an upper bound rather than a promise -- a caller that
        # names a number larger than the buffer is asking for the buffer.
        return list(islice(events, min(max(0, limit), len(events))))

    def drop(self, events: Iterable[Event]) -> None:
        """Remove events a master has confirmed.

        Matched on identity, not equality, and the difference matters. Events
        are frozen dataclasses, so two genuinely distinct events carrying the
        same index, value and timestamp compare equal -- a master confirming
        one of them would silently drop both. ``id()`` is what tells them
        apart, and tidying this into ``==`` would reintroduce that.

        It is load-bearing that the events passed in are still held by a
        buffer, since that is what keeps them alive and their ids stable for
        the length of this call. True of the ``peek`` then ``drop`` lifecycle
        above; not true of ids kept across a confirm cycle.
        """
        confirmed = set(map(id, events))
        for buffer in self._buffers.values():
            remaining: deque[Event] = deque()
            for event in buffer.events:
                if id(event) in confirmed:
                    self._forget(event)
                else:
                    remaining.append(event)
            buffer.events = remaining
        for index, pending in list(self._pending_analog.items()):
            if id(pending) in confirmed:
                del self._pending_analog[index]

    def _discard(self, event: Event) -> None:
        """Remove one event, found by identity, without rebuilding a buffer.

        ``drop`` rebuilds every class to retire a batch, which is the right
        cost once per confirmation and the wrong one once per reading.
        """
        for buffer in self._buffers.values():
            for position, held in enumerate(buffer.events):
                if held is event:
                    del buffer.events[position]
                    return

    def clear_overflow(self) -> None:
        """Forget the overflow, once it has been reported to a master."""
        for buffer in self._buffers.values():
            buffer.overflowed = False

    def _buffer(self, event_class: EventClass, event: Event) -> None:
        """Hold an event, counting it against the generation if one was lost."""
        buffer = self._buffers[event_class]
        evicted = buffer.events[0] if len(buffer.events) >= buffer.capacity else None
        if buffer.add(event):
            self._overflow_generation += 1
        if evicted is not None:
            self._forget(evicted)
        if isinstance(evicted, AnalogEvent) and self._pending_analog.get(evicted.index) is evicted:
            del self._pending_analog[evicted.index]

    @staticmethod
    def _checked_index(index: int) -> int:
        """An index an event block can actually carry.

        Checked when the event is recorded rather than when it is read. The
        encoder refuses a negative or oversized index, and that refusal arriving
        at read time would take out every read of the class -- a caller's
        mistake surfacing as a protocol failure, far from the line that made it
        and with the connection as collateral. Refusing here puts the error
        where the mistake is and leaves the buffer incapable of holding one.
        """
        if not 0 <= index <= MAX_INDEX:
            raise ValueError(
                f"index {index} is outside 0..{MAX_INDEX}, which is what an event block can carry"
            )
        return index

    def record_analog(
        self,
        index: int,
        point: AnalogPoint,
        *,
        event_class: EventClass,
        deadband: float = 0.0,
        timestamp_ms: int | None = None,
    ) -> AnalogEvent | None:
        """Record an analog reading, returning the event it generated, if any."""
        self._checked_index(index)
        previous = self._last_analog.get(index)
        if previous is not None and not self._analog_changed(previous, point, deadband):
            return None

        event = AnalogEvent(
            index,
            point,
            timestamp_ms if timestamp_ms is not None else now_ms(),
            order=self._next_order(),
        )
        self._last_analog[index] = point
        if self._analog_latest_only:
            superseded = self._pending_analog.get(index)
            if superseded is not None:
                # Removed rather than overwritten in place: the event moves to
                # the back of its class, because it now reports a change that
                # happened after everything buffered ahead of it.
                self._discard(superseded)
            self._pending_analog[index] = event
        self._buffer(event_class, event)
        return event

    @property
    def capacity(self) -> int:
        """Events each class holds before its oldest is dropped."""
        return self._capacity

    @property
    def analog_latest_only(self) -> bool:
        """Whether only the newest event per analog point is kept."""
        return self._analog_latest_only

    def prime_analog(self, index: int, point: AnalogPoint) -> None:
        """Note where an analog point stands, without reporting it.

        For the first reading after start. A change is measured against the
        last value a master was told, and at start it has been told nothing --
        it learns every point from its integrity poll, so an event per point
        here would report a few hundred changes that did not happen.
        """
        self._last_analog[self._checked_index(index)] = point

    def prime_binary(self, index: int, point: BinaryPoint) -> None:
        """Note where a binary point stands, without reporting it."""
        self._last_binary[self._checked_index(index)] = point

    def record_frozen_counter(
        self,
        index: int,
        point: CounterPoint,
        *,
        event_class: EventClass,
        timestamp_ms: int | None = None,
    ) -> FrozenCounterEvent:
        """Record one counter's value at a freeze. Always an event.

        No comparison against the previous freeze, unlike the other two. A
        frozen counter event is an entry in a log -- the value at this moment
        -- and two freezes of a counter that did not move are two entries the
        master is owed, since their absence would read as freezes that never
        happened.
        """
        self._checked_index(index)
        event = FrozenCounterEvent(
            index,
            point,
            timestamp_ms if timestamp_ms is not None else now_ms(),
            order=self._next_order(),
        )
        self._buffer(event_class, event)
        return event

    def peek_kind(self, kind: type, limit: int | None = None) -> list[Event]:
        """The oldest events of one type, across every class, without removing them.

        For a master that reads an event group by name rather than a class.
        Oldest first whichever class each is in: a master that asks for the
        next three binary changes is asking for the three that happened first,
        and taking them a class at a time would hand it a later change ahead
        of an earlier one.

        Each class is already in order, so the oldest ``limit`` overall are
        among the oldest ``limit`` of each, and no class is walked further.
        """
        found: list[Event] = []
        for event_class in EventClass:
            taken = 0
            for event in self._buffers[event_class].events:
                if limit is not None and taken >= limit:
                    break
                if isinstance(event, kind):
                    found.append(event)
                    taken += 1
        found.sort(key=lambda event: event.order)
        return found if limit is None else found[: max(0, limit)]

    def record_binary(
        self,
        index: int,
        point: BinaryPoint,
        *,
        event_class: EventClass,
        timestamp_ms: int | None = None,
        synchronized: bool | None = None,
    ) -> BinaryEvent | None:
        """Record a binary reading, returning the event it generated, if any.

        No deadband: a binary point has two values, so any change is the whole
        of its range.

        ``synchronized`` says whether the clock behind the timestamp had
        been set, and defaults to what these buffers were last told.
        """
        self._checked_index(index)
        previous = self._last_binary.get(index)
        if previous is not None and not self._binary_changed(previous, point):
            return None

        event = BinaryEvent(
            index,
            point,
            timestamp_ms if timestamp_ms is not None else now_ms(),
            order=self._next_order(),
            synchronized=self.synchronized if synchronized is None else synchronized,
        )
        if not event.synchronized:
            self._unsynchronized += 1
        self._last_binary[index] = point
        self._buffer(event_class, event)
        return event

    @staticmethod
    def _analog_changed(previous: AnalogPoint, current: AnalogPoint, deadband: float) -> bool:
        if _analog_quality_of(previous.flags) != _analog_quality_of(current.flags):
            # Quality first and unconditionally. A point that goes comm-lost
            # holding the same number has not moved and has changed in the way
            # that matters, and a deadband applied here would hide it.
            return True

        # NaN before the deadband, because every comparison involving it is
        # false -- so a reading that went NaN would look unchanged, the master
        # would never hear about it, and the encoder's REFERENCE_ERR would never
        # reach the wire. The recovery back to a real number would be lost the
        # same way. NaN to NaN is genuinely unchanged.
        was_nan, is_nan = math.isnan(previous.value), math.isnan(current.value)
        if was_nan or is_nan:
            return was_nan != is_nan

        return abs(current.value - previous.value) > deadband

    @staticmethod
    def _binary_changed(previous: BinaryPoint, current: BinaryPoint) -> bool:
        if previous.state != current.state:
            return True
        return _binary_quality_of(previous.flags) != _binary_quality_of(current.flags)

holds_unsynchronized property

holds_unsynchronized: bool

Whether any buffered binary event was stamped by a clock that had not been set.

overflow_generation property

overflow_generation: int

How many events have been lost, counted from the first.

Monotonic, and deliberately not reset by clear_overflow. A confirmation acknowledges the loss a particular response reported, and the flag alone cannot say whether more has been lost since that response was built -- it reads the same after one eviction and after a hundred. Clearing on the flag would retire a report of data loss the master has never been sent.

recorded property

recorded: int

How many events have been recorded, counted from the first.

Monotonic, like :attr:overflow_generation, and for a similar reason: the count of events held cannot say whether anything new has happened, since one may have been retired or evicted for each one recorded. A session that has stopped retrying an unsolicited response reads this to learn that there is news worth starting again for.

capacity property

capacity: int

Events each class holds before its oldest is dropped.

analog_latest_only property

analog_latest_only: bool

Whether only the newest event per analog point is kept.

overflowed

overflowed() -> bool

Whether any class has dropped an event since it was last reported.

Source code in src/py1815/events.py

def overflowed(self) -> bool:
    """Whether any class has dropped an event since it was last reported."""
    return any(buffer.overflowed for buffer in self._buffers.values())

classes_with_events

classes_with_events(
    excluding: Iterable[Event] = (),
) -> set[EventClass]

Which classes have something to report, for the indication bits.

excluding names events a response is already carrying. A class whose every event is in the response being built has nothing more to report, and saying it has would send the master back for events it is holding.

Source code in src/py1815/events.py

def classes_with_events(self, excluding: Iterable[Event] = ()) -> set[EventClass]:
    """Which classes have something to report, for the indication bits.

    ``excluding`` names events a response is already carrying. A class
    whose every event is in the response being built has nothing *more*
    to report, and saying it has would send the master back for events
    it is holding.
    """
    carried = set(map(id, excluding))
    if not carried:
        return {cls for cls, buffer in self._buffers.items() if buffer.events}
    return {
        cls
        for cls, buffer in self._buffers.items()
        if any(id(event) not in carried for event in buffer.events)
    }

peek

peek(
    event_class: EventClass, limit: int | None = None
) -> list[Event]

The oldest events of a class, without removing them.

Reads do not consume. An event leaves the buffer when the master confirms the response carrying it, because a response that never arrives must not have taken the only copy with it.

limit is taken from the front, and taken while the deque is walked rather than after. The difference is the whole point of the parameter: materializing the class and slicing it afterwards costs the buffer on every call, and capacity has no upper bound, so a caller asking for the few events that fit a response would still pay for all of them.

Source code in src/py1815/events.py

def peek(self, event_class: EventClass, limit: int | None = None) -> list[Event]:
    """The oldest events of a class, without removing them.

    Reads do not consume. An event leaves the buffer when the master
    confirms the response carrying it, because a response that never arrives
    must not have taken the only copy with it.

    ``limit`` is taken from the front, and taken while the deque is walked
    rather than after. The difference is the whole point of the parameter:
    materializing the class and slicing it afterwards costs the buffer on
    every call, and ``capacity`` has no upper bound, so a caller asking for
    the few events that fit a response would still pay for all of them.
    """
    events = self._buffers[event_class].events
    if limit is None:
        return list(events)
    # Clamped to the length as well as to zero. ``islice`` refuses a stop
    # above ``sys.maxsize``, where a plain slice simply returns everything,
    # and a limit is an upper bound rather than a promise -- a caller that
    # names a number larger than the buffer is asking for the buffer.
    return list(islice(events, min(max(0, limit), len(events))))

drop

drop(events: Iterable[Event]) -> None

Remove events a master has confirmed.

Matched on identity, not equality, and the difference matters. Events are frozen dataclasses, so two genuinely distinct events carrying the same index, value and timestamp compare equal -- a master confirming one of them would silently drop both. id() is what tells them apart, and tidying this into == would reintroduce that.

It is load-bearing that the events passed in are still held by a buffer, since that is what keeps them alive and their ids stable for the length of this call. True of the peek then drop lifecycle above; not true of ids kept across a confirm cycle.

Source code in src/py1815/events.py

def drop(self, events: Iterable[Event]) -> None:
    """Remove events a master has confirmed.

    Matched on identity, not equality, and the difference matters. Events
    are frozen dataclasses, so two genuinely distinct events carrying the
    same index, value and timestamp compare equal -- a master confirming
    one of them would silently drop both. ``id()`` is what tells them
    apart, and tidying this into ``==`` would reintroduce that.

    It is load-bearing that the events passed in are still held by a
    buffer, since that is what keeps them alive and their ids stable for
    the length of this call. True of the ``peek`` then ``drop`` lifecycle
    above; not true of ids kept across a confirm cycle.
    """
    confirmed = set(map(id, events))
    for buffer in self._buffers.values():
        remaining: deque[Event] = deque()
        for event in buffer.events:
            if id(event) in confirmed:
                self._forget(event)
            else:
                remaining.append(event)
        buffer.events = remaining
    for index, pending in list(self._pending_analog.items()):
        if id(pending) in confirmed:
            del self._pending_analog[index]

clear_overflow

clear_overflow() -> None

Forget the overflow, once it has been reported to a master.

Source code in src/py1815/events.py

def clear_overflow(self) -> None:
    """Forget the overflow, once it has been reported to a master."""
    for buffer in self._buffers.values():
        buffer.overflowed = False

record_analog

record_analog(
    index: int,
    point: AnalogPoint,
    *,
    event_class: EventClass,
    deadband: float = 0.0,
    timestamp_ms: int | None = None,
) -> AnalogEvent | None

Record an analog reading, returning the event it generated, if any.

Source code in src/py1815/events.py

def record_analog(
    self,
    index: int,
    point: AnalogPoint,
    *,
    event_class: EventClass,
    deadband: float = 0.0,
    timestamp_ms: int | None = None,
) -> AnalogEvent | None:
    """Record an analog reading, returning the event it generated, if any."""
    self._checked_index(index)
    previous = self._last_analog.get(index)
    if previous is not None and not self._analog_changed(previous, point, deadband):
        return None

    event = AnalogEvent(
        index,
        point,
        timestamp_ms if timestamp_ms is not None else now_ms(),
        order=self._next_order(),
    )
    self._last_analog[index] = point
    if self._analog_latest_only:
        superseded = self._pending_analog.get(index)
        if superseded is not None:
            # Removed rather than overwritten in place: the event moves to
            # the back of its class, because it now reports a change that
            # happened after everything buffered ahead of it.
            self._discard(superseded)
        self._pending_analog[index] = event
    self._buffer(event_class, event)
    return event

prime_analog

prime_analog(index: int, point: AnalogPoint) -> None

Note where an analog point stands, without reporting it.

For the first reading after start. A change is measured against the last value a master was told, and at start it has been told nothing -- it learns every point from its integrity poll, so an event per point here would report a few hundred changes that did not happen.

Source code in src/py1815/events.py

def prime_analog(self, index: int, point: AnalogPoint) -> None:
    """Note where an analog point stands, without reporting it.

    For the first reading after start. A change is measured against the
    last value a master was told, and at start it has been told nothing --
    it learns every point from its integrity poll, so an event per point
    here would report a few hundred changes that did not happen.
    """
    self._last_analog[self._checked_index(index)] = point

prime_binary

prime_binary(index: int, point: BinaryPoint) -> None

Note where a binary point stands, without reporting it.

Source code in src/py1815/events.py

def prime_binary(self, index: int, point: BinaryPoint) -> None:
    """Note where a binary point stands, without reporting it."""
    self._last_binary[self._checked_index(index)] = point

record_frozen_counter

record_frozen_counter(
    index: int,
    point: CounterPoint,
    *,
    event_class: EventClass,
    timestamp_ms: int | None = None,
) -> FrozenCounterEvent

Record one counter's value at a freeze. Always an event.

No comparison against the previous freeze, unlike the other two. A frozen counter event is an entry in a log -- the value at this moment -- and two freezes of a counter that did not move are two entries the master is owed, since their absence would read as freezes that never happened.

Source code in src/py1815/events.py

def record_frozen_counter(
    self,
    index: int,
    point: CounterPoint,
    *,
    event_class: EventClass,
    timestamp_ms: int | None = None,
) -> FrozenCounterEvent:
    """Record one counter's value at a freeze. Always an event.

    No comparison against the previous freeze, unlike the other two. A
    frozen counter event is an entry in a log -- the value at this moment
    -- and two freezes of a counter that did not move are two entries the
    master is owed, since their absence would read as freezes that never
    happened.
    """
    self._checked_index(index)
    event = FrozenCounterEvent(
        index,
        point,
        timestamp_ms if timestamp_ms is not None else now_ms(),
        order=self._next_order(),
    )
    self._buffer(event_class, event)
    return event

peek_kind

peek_kind(
    kind: type, limit: int | None = None
) -> list[Event]

The oldest events of one type, across every class, without removing them.

For a master that reads an event group by name rather than a class. Oldest first whichever class each is in: a master that asks for the next three binary changes is asking for the three that happened first, and taking them a class at a time would hand it a later change ahead of an earlier one.

Each class is already in order, so the oldest limit overall are among the oldest limit of each, and no class is walked further.

Source code in src/py1815/events.py

def peek_kind(self, kind: type, limit: int | None = None) -> list[Event]:
    """The oldest events of one type, across every class, without removing them.

    For a master that reads an event group by name rather than a class.
    Oldest first whichever class each is in: a master that asks for the
    next three binary changes is asking for the three that happened first,
    and taking them a class at a time would hand it a later change ahead
    of an earlier one.

    Each class is already in order, so the oldest ``limit`` overall are
    among the oldest ``limit`` of each, and no class is walked further.
    """
    found: list[Event] = []
    for event_class in EventClass:
        taken = 0
        for event in self._buffers[event_class].events:
            if limit is not None and taken >= limit:
                break
            if isinstance(event, kind):
                found.append(event)
                taken += 1
    found.sort(key=lambda event: event.order)
    return found if limit is None else found[: max(0, limit)]

record_binary

record_binary(
    index: int,
    point: BinaryPoint,
    *,
    event_class: EventClass,
    timestamp_ms: int | None = None,
    synchronized: bool | None = None,
) -> BinaryEvent | None

Record a binary reading, returning the event it generated, if any.

No deadband: a binary point has two values, so any change is the whole of its range.

synchronized says whether the clock behind the timestamp had been set, and defaults to what these buffers were last told.

Source code in src/py1815/events.py

def record_binary(
    self,
    index: int,
    point: BinaryPoint,
    *,
    event_class: EventClass,
    timestamp_ms: int | None = None,
    synchronized: bool | None = None,
) -> BinaryEvent | None:
    """Record a binary reading, returning the event it generated, if any.

    No deadband: a binary point has two values, so any change is the whole
    of its range.

    ``synchronized`` says whether the clock behind the timestamp had
    been set, and defaults to what these buffers were last told.
    """
    self._checked_index(index)
    previous = self._last_binary.get(index)
    if previous is not None and not self._binary_changed(previous, point):
        return None

    event = BinaryEvent(
        index,
        point,
        timestamp_ms if timestamp_ms is not None else now_ms(),
        order=self._next_order(),
        synchronized=self.synchronized if synchronized is None else synchronized,
    )
    if not event.synchronized:
        self._unsynchronized += 1
    self._last_binary[index] = point
    self._buffer(event_class, event)
    return event

now_ms

now_ms() -> int

Wall-clock milliseconds, which is what a DNP3 timestamp carries.

Wall clock rather than monotonic, deliberately: the timestamp is for the master's sequence of record, and a monotonic reading means nothing to it. A clock that steps backwards produces events out of order, which is a real hazard and the operator's to resolve -- not one to be hidden here by inventing times that look plausible.

Source code in src/py1815/events.py

def now_ms() -> int:
    """Wall-clock milliseconds, which is what a DNP3 timestamp carries.

    Wall clock rather than monotonic, deliberately: the timestamp is for the
    master's sequence of record, and a monotonic reading means nothing to it. A
    clock that steps backwards produces events out of order, which is a real
    hazard and the operator's to resolve -- not one to be hidden here by
    inventing times that look plausible.
    """
    return int(time.time() * 1000)