pamoja.hal

  1"""Idiomatic bus facade.
  2
  3A driver is a conversation with a part, and a bus is what carries it. An
  4:class:`I2cBus` is one bus that the program and every driver on it share, with one of
  5three things on the other end: the kernel's adapter on a Linux board, simulated parts
  6that answer from their registers, or a script of the transfers a driver is expected to
  7make. A driver runs the same way over all three, so a program is written and tested
  8with nothing plugged in and then pointed at ``/dev/i2c-1``. A :class:`SerialPort` is the
  9same idea for a UART, and :class:`DelayLog` and :class:`SleepDelay` pace a driver that
 10waits between pin changes.
 11"""
 12
 13from __future__ import annotations
 14
 15import enum
 16import time
 17from dataclasses import dataclass
 18from typing import Iterable, List, Optional, Protocol, Tuple, Union
 19
 20from pamoja._native import CommandPart as _NativeCommandPart
 21from pamoja._native import I2cBus as _NativeBus
 22from pamoja._native import I2cPart as _NativePart
 23from pamoja._native import I2cStep as _NativeStep
 24from pamoja._native import SerialPort as _NativeSerialPort
 25from pamoja._native import SerialStep as _NativeSerialStep
 26from pamoja._native import WordPart as _NativeWordPart
 27from pamoja._native import serial_bits_per_character as _serial_bits_per_character
 28from pamoja._native import serial_character_nanos as _serial_character_nanos
 29from pamoja._native import serial_transfer_micros as _serial_transfer_micros
 30
 31__all__ = [
 32    "CommandPart",
 33    "Delay",
 34    "DelayLog",
 35    "I2cBus",
 36    "I2cBusKind",
 37    "I2cFault",
 38    "I2cPart",
 39    "I2cStep",
 40    "Parity",
 41    "SerialPort",
 42    "SerialPortKind",
 43    "SerialSettings",
 44    "SerialStep",
 45    "SimulatedPart",
 46    "SleepDelay",
 47    "WordPart",
 48]
 49
 50
 51class I2cBusKind(str, enum.Enum):
 52    """What answers on a bus."""
 53
 54    #: The kernel's adapter, with real parts on real wires.
 55    ADAPTER = "Adapter"
 56    #: Simulated parts, answering from their registers.
 57    SIMULATED = "Simulated"
 58    #: A script of the transfers a driver is expected to make.
 59    SCRIPTED = "Scripted"
 60
 61
 62class I2cFault(str, enum.Enum):
 63    """How a scripted step fails the transfer that reaches it."""
 64
 65    #: Nothing acknowledged the address.
 66    NO_ACKNOWLEDGE_ADDRESS = "NoAcknowledgeAddress"
 67    #: The part did not acknowledge a data byte.
 68    NO_ACKNOWLEDGE_DATA = "NoAcknowledgeData"
 69    #: A missing acknowledge, with no telling whether of the address or the data.
 70    NO_ACKNOWLEDGE = "NoAcknowledge"
 71    #: A bus error, such as a misplaced start or stop condition.
 72    BUS = "Bus"
 73    #: Another controller won the bus.
 74    ARBITRATION_LOSS = "ArbitrationLoss"
 75    #: Data arrived faster than it was taken.
 76    OVERRUN = "Overrun"
 77    #: A failure of no more particular kind.
 78    OTHER = "Other"
 79
 80
 81class I2cPart:
 82    """A part that is not there, answering from 256 registers.
 83
 84    A write names a register and fills it and the ones after it; a read takes them
 85    back from wherever the last write left off. What a driver writes stays written, so
 86    :meth:`register` reads a part's configuration back once a driver is done with it.
 87
 88    >>> BME280, CHIP_ID_REGISTER, BME280_CHIP_ID = 0x76, 0xD0, 0x60
 89    >>> part = I2cPart(BME280).holding(CHIP_ID_REGISTER, bytes([BME280_CHIP_ID]))
 90    >>> part.register(CHIP_ID_REGISTER) == BME280_CHIP_ID
 91    True
 92    """
 93
 94    __slots__ = ("_native",)
 95
 96    def __init__(self, address: int) -> None:
 97        """Make a part answering at one address, with every register reading zero.
 98
 99        :param address: The 7-bit address it answers to.
100        """
101        self._native = _NativePart(address)
102
103    @classmethod
104    def _wrap(cls, native: _NativePart) -> I2cPart:
105        part = cls.__new__(cls)
106        part._native = native
107        return part
108
109    def holding(self, first: int, data: bytes) -> I2cPart:
110        """Put bytes in the part from a register on, and return the part.
111
112        :param first: The register the bytes start at.
113        :param data: What to put there. Past the last register it wraps to the first.
114        :returns: This part, so calls chain.
115        """
116        self._native.load(first, bytes(data))
117        return self
118
119    def register(self, register: int) -> int:
120        """Read what one register holds now.
121
122        :param register: Which register.
123        :returns: Its value, which is what a driver wrote if it wrote one.
124        """
125        return self._native.register(register)
126
127    def read(self, first: int, length: int) -> bytes:
128        """Read consecutive registers from one register on.
129
130        :param first: The first register.
131        :param length: How many registers.
132        :returns: One byte per register.
133        """
134        return bytes(self._native.read(first, length))
135
136    @property
137    def address(self) -> int:
138        """The address the part answers to."""
139        return self._native.address
140
141    @property
142    def transfers(self) -> int:
143        """How many transfers the part has served."""
144        return self._native.transfers
145
146
147class WordPart:
148    """A part that is not there, answering from 256 registers sixteen bits wide.
149
150    This is how Texas Instruments lays out parts such as the TMP117, the INA219 and
151    INA226, the OPT3001, the ADS1115, and the HDC1080. A pointer byte names a register and
152    a register travels most significant byte first. Bits the part sets for itself, such as
153    a conversion-ready flag, are marked with :meth:`read_only` and keep the part's value
154    whatever a driver writes.
155
156    >>> TMP117, DEVICE_ID_REGISTER, TMP117_DEVICE_ID = 0x48, 0x0F, 0x0117
157    >>> part = WordPart(TMP117).holding(DEVICE_ID_REGISTER, TMP117_DEVICE_ID)
158    >>> part.word(DEVICE_ID_REGISTER) == TMP117_DEVICE_ID
159    True
160    """
161
162    __slots__ = ("_native",)
163
164    def __init__(self, address: int) -> None:
165        """Make a part answering at one address, with every register reading zero.
166
167        :param address: The 7-bit address it answers to.
168        """
169        self._native = _NativeWordPart(address)
170
171    @classmethod
172    def _wrap(cls, native: _NativeWordPart) -> WordPart:
173        part = cls.__new__(cls)
174        part._native = native
175        return part
176
177    def holding(self, register: int, value: int) -> WordPart:
178        """Put a value in one register, and return the part.
179
180        :param register: The register.
181        :param value: What it holds, read-only bits included.
182        :returns: This part, so calls chain.
183        """
184        self._native.set(register, value)
185        return self
186
187    def read_only(self, register: int, mask: int) -> WordPart:
188        """Mark bits of one register as the part's to set, and return the part.
189
190        :param register: The register.
191        :param mask: The bits a driver's write leaves as the part holds them.
192        :returns: This part, so calls chain.
193        """
194        self._native.read_only(register, mask)
195        return self
196
197    def set(self, register: int, value: int) -> None:
198        """Put a value in one register, read-only bits included, as the part itself would.
199
200        :param register: The register.
201        :param value: What it holds.
202        """
203        self._native.set(register, value)
204
205    def word(self, register: int) -> int:
206        """Read what one register holds now.
207
208        :param register: Which register.
209        :returns: Its value, which is what a driver wrote apart from the read-only bits.
210        """
211        return self._native.word(register)
212
213    @property
214    def address(self) -> int:
215        """The address the part answers to."""
216        return self._native.address
217
218    @property
219    def transfers(self) -> int:
220        """How many transfers the part has served."""
221        return self._native.transfers
222
223
224class CommandPart:
225    """A part that is not there, answering commands with the replies it was given.
226
227    This is how Sensirion lays out parts such as the SHT3x and the SCD4x. A write sends a
228    command and any arguments after it; a read takes the reply that command left, once,
229    padded with ``0xFF`` the way an idle bus reads. A command given no reply leaves none,
230    and a read then is not acknowledged, which is what a real part does when asked for data
231    it does not have.
232
233    >>> SHT3X, READ_STATUS = 0x44, (0xF32D).to_bytes(2, "big")
234    >>> STATUS_AFTER_RESET = bytes([0x80, 0x10, 0xE1])  # the word 0x8010, then its CRC
235    >>> part = CommandPart(SHT3X).answering(READ_STATUS, STATUS_AFTER_RESET)
236    >>> part.address == SHT3X
237    True
238    """
239
240    __slots__ = ("_native",)
241
242    def __init__(self, address: int, width: int = 2) -> None:
243        """Make a part answering at one address that has been given no replies yet.
244
245        :param address: The 7-bit address it answers to.
246        :param width: How many bytes a command takes: two for Sensirion's commands.
247        """
248        self._native = _NativeCommandPart(address, width)
249
250    @classmethod
251    def _wrap(cls, native: _NativeCommandPart) -> CommandPart:
252        part = cls.__new__(cls)
253        part._native = native
254        return part
255
256    def answering(self, command: bytes, reply: bytes) -> CommandPart:
257        """Answer one command with a reply from now on, and return the part.
258
259        :param command: The command's bytes.
260        :param reply: What a read after it returns, in place of any reply given before.
261        :returns: This part, so calls chain.
262        """
263        self.answer(command, reply)
264        return self
265
266    def answer(self, command: bytes, reply: bytes) -> None:
267        """Answer one command with a reply from now on.
268
269        :param command: The command's bytes.
270        :param reply: What a read after it returns, in place of any reply given before.
271        """
272        self._native.answer(bytes(command), bytes(reply))
273
274    @property
275    def received(self) -> List[bytes]:
276        """Every write the part has received, oldest first: a command and any arguments."""
277        return [bytes(write) for write in self._native.received]
278
279    @property
280    def address(self) -> int:
281        """The address the part answers to."""
282        return self._native.address
283
284    @property
285    def transfers(self) -> int:
286        """How many transfers the part has served."""
287        return self._native.transfers
288
289
290#: Any simulated part: a bus takes each kind and gives each back as its own class.
291SimulatedPart = Union[I2cPart, WordPart, CommandPart]
292
293
294def _part_of(native: object) -> SimulatedPart:
295    """Wrap a native part as the class of part it is."""
296    if isinstance(native, _NativeWordPart):
297        return WordPart._wrap(native)
298    if isinstance(native, _NativeCommandPart):
299        return CommandPart._wrap(native)
300    return I2cPart._wrap(native)
301
302
303class I2cStep:
304    """One transfer a script expects, and what the part answers."""
305
306    __slots__ = ("_native",)
307
308    def __init__(self, native: _NativeStep) -> None:
309        """Wrap a native step. Use :meth:`write`, :meth:`read`, :meth:`write_read`, or
310        :meth:`fault`."""
311        self._native = native
312
313    @classmethod
314    def write(cls, address: int, data: bytes) -> I2cStep:
315        """The driver writes exactly ``data`` to the address.
316
317        :param address: The 7-bit address the write must go to.
318        :param data: The bytes the driver must send.
319        :returns: The step.
320        """
321        return cls(_NativeStep.write(address, bytes(data)))
322
323    @classmethod
324    def read(cls, address: int, reply: bytes) -> I2cStep:
325        """The driver reads from the address and receives ``reply``.
326
327        :param address: The 7-bit address the read must come from.
328        :param reply: The bytes the part answers with; the driver must ask for exactly
329            this many.
330        :returns: The step.
331        """
332        return cls(_NativeStep.read(address, bytes(reply)))
333
334    @classmethod
335    def write_read(cls, address: int, data: bytes, reply: bytes) -> I2cStep:
336        """The driver writes ``data`` and then reads ``reply`` in one transaction, the
337        shape of a register read.
338
339        :param address: The 7-bit address of the part.
340        :param data: The bytes the driver must send first, usually a register address.
341        :param reply: The bytes the part answers with.
342        :returns: The step.
343        """
344        return cls(_NativeStep.write_read(address, bytes(data), bytes(reply)))
345
346    @classmethod
347    def fault(cls, address: int, fault: I2cFault) -> I2cStep:
348        """The next transfer to the address fails, the way a missing or busy part does.
349
350        :param address: The 7-bit address the failing transfer must go to.
351        :param fault: The failure the driver sees.
352        :returns: The step.
353        """
354        return cls(_NativeStep.fault(address, I2cFault(fault).value))
355
356
357class I2cBus:
358    """One I2C bus, shared by the program and every driver built on it.
359
360    :meth:`open` opens the kernel's adapter on a Linux board; :meth:`simulated` puts
361    simulated parts of any kind on a bus, each answering at its own address;
362    :meth:`scripted` plays :class:`I2cStep` s in order and refuses any other transfer. A
363    failed transfer raises ``PamojaError`` with the reason: nothing answered at the address,
364    the script expected something else, or the kernel's own words.
365
366    >>> BME280, CHIP_ID_REGISTER, BME280_CHIP_ID = 0x76, 0xD0, 0x60
367    >>> bus = I2cBus.simulated([I2cPart(BME280).holding(CHIP_ID_REGISTER, bytes([BME280_CHIP_ID]))])
368    >>> bus.write_read(BME280, bytes([CHIP_ID_REGISTER]), 1) == bytes([BME280_CHIP_ID])
369    True
370    >>> bus.transfers
371    1
372    """
373
374    __slots__ = ("_native",)
375
376    def __init__(self, native: _NativeBus) -> None:
377        """Wrap a native bus. Use :meth:`open`, :meth:`simulated`, or :meth:`scripted`."""
378        self._native = native
379
380    @classmethod
381    def open(cls, path: str) -> I2cBus:
382        """Open the kernel's I2C adapter, such as ``/dev/i2c-1`` on a Raspberry Pi.
383
384        :param path: The adapter's device file.
385        :returns: The bus, with the real parts wired to it on the other end.
386        :raises PamojaError: Anywhere but Linux, and when the file cannot be opened as
387            an adapter: the interface is not turned on, or the process may not use it.
388        """
389        return cls(_NativeBus.open(path))
390
391    @classmethod
392    def simulated(cls, parts: Iterable[SimulatedPart] = ()) -> I2cBus:
393        """Make a bus of simulated parts, each answering at its own address.
394
395        :param parts: The parts on the bus, of any kind. A later part at an address an
396            earlier one holds takes its place.
397        :returns: The bus. A transfer to an address no part holds raises
398            ``PamojaError``, as nothing acknowledges it.
399        """
400        return cls(_NativeBus.simulated([part._native for part in parts]))
401
402    @classmethod
403    def scripted(cls, steps: Iterable[I2cStep]) -> I2cBus:
404        """Make a bus that plays the steps in order and refuses any other transfer.
405
406        :param steps: The transfers a driver is expected to make, and the replies.
407        :returns: The bus.
408        """
409        return cls(_NativeBus.scripted([step._native for step in steps]))
410
411    def attach(self, part: SimulatedPart) -> None:
412        """Put a copy of a part on a simulated bus, in place of any part at its address.
413
414        A driver keeps working across the change, which is how a test moves a reading on.
415
416        :param part: The part, of any kind.
417        :raises PamojaError: If the bus is not simulated.
418        """
419        self._native.attach(part._native)
420
421    @property
422    def kind(self) -> I2cBusKind:
423        """What answers on the bus."""
424        return I2cBusKind(self._native.kind)
425
426    def write(self, address: int, data: bytes) -> None:
427        """Write bytes to a part in one transaction: usually a register and its value.
428
429        :param address: The part's 7-bit address.
430        :param data: The bytes to write.
431        :raises PamojaError: If the transfer fails.
432        """
433        self._native.write(address, bytes(data))
434
435    def read(self, address: int, length: int) -> bytes:
436        """Read bytes from a part in one transaction.
437
438        :param address: The part's 7-bit address.
439        :param length: How many bytes to read.
440        :returns: The bytes.
441        :raises PamojaError: If the transfer fails.
442        """
443        return bytes(self._native.read(address, length))
444
445    def write_read(self, address: int, data: bytes, length: int) -> bytes:
446        """Write bytes and then read the reply in one transaction, with a repeated start
447        between them, which is how a register is read.
448
449        :param address: The part's 7-bit address.
450        :param data: What to write first, usually the register address.
451        :param length: How many bytes to read.
452        :returns: The reply.
453        :raises PamojaError: If the transfer fails.
454        """
455        return bytes(self._native.write_read(address, bytes(data), length))
456
457    def part(self, address: int) -> Optional[SimulatedPart]:
458        """Copy what a simulated part holds now, with whatever drivers wrote to it.
459
460        :param address: The part's address.
461        :returns: The copy, as the class of part it is, or ``None`` when the bus is not
462            simulated or no part holds the address.
463        """
464        native = self._native.part(address)
465        return None if native is None else _part_of(native)
466
467    @property
468    def transfers(self) -> int:
469        """How many transfers have been made on the bus, by the program and every driver
470        on it, including any that failed."""
471        return self._native.transfers
472
473    @property
474    def remaining(self) -> Optional[int]:
475        """How many steps a script has left, or ``None`` when the bus is not scripted."""
476        return self._native.remaining
477
478    @property
479    def waited_micros(self) -> int:
480        """How long the drivers on the bus have asked to wait, in microseconds, whether
481        or not the process slept through it."""
482        return self._native.waited_micros
483
484
485class Parity(str, enum.Enum):
486    """The parity bit each character on a serial line carries."""
487
488    #: No parity bit.
489    NONE = "None"
490    #: A bit that makes the count of ones even, what Modbus RTU asks for by default.
491    EVEN = "Even"
492    #: A bit that makes the count of ones odd.
493    ODD = "Odd"
494
495
496class SerialPortKind(str, enum.Enum):
497    """What is on the other end of a serial port."""
498
499    #: The kernel's serial device, with a real line on the other end.
500    DEVICE = "Device"
501    #: The port's own output, looped back to its input.
502    LOOPED = "Looped"
503    #: The other end of a null-modem pair.
504    PAIRED = "Paired"
505    #: A simulated device that answers each write.
506    SIMULATED = "Simulated"
507    #: A script of the writes a driver is expected to make.
508    SCRIPTED = "Scripted"
509
510
511@dataclass(frozen=True)
512class SerialSettings:
513    """A port's speed and character format: eight data bits, with the parity and stop bits
514    given.
515
516    >>> modbus = SerialSettings(9_600, Parity.EVEN)
517    >>> str(modbus), modbus.bits_per_character, modbus.character_nanos
518    ('9600 8E1', 11, 1145834)
519    """
520
521    #: The speed, in bits a second.
522    baud: int
523    #: The parity bit each character carries.
524    parity: Parity = Parity.NONE
525    #: 1 or 2 stop bits.
526    stop_bits: int = 1
527
528    def _values(self) -> tuple:
529        return (self.baud, Parity(self.parity).value, self.stop_bits)
530
531    @property
532    def bits_per_character(self) -> int:
533        """The bits one character takes on the wire: a start bit, eight data bits, the
534        parity bit if there is one, and the stop bits."""
535        return _serial_bits_per_character(*self._values())
536
537    @property
538    def character_nanos(self) -> int:
539        """How long one character takes on the wire, in nanoseconds, rounded up."""
540        return _serial_character_nanos(*self._values())
541
542    def transfer_micros(self, count: int) -> int:
543        """How long ``count`` bytes sent back to back take on the wire.
544
545        :param count: How many bytes.
546        :returns: The time in microseconds, rounded up.
547        """
548        return _serial_transfer_micros(*self._values(), count)
549
550    def __str__(self) -> str:
551        letter = {Parity.NONE: "N", Parity.EVEN: "E", Parity.ODD: "O"}[Parity(self.parity)]
552        return f"{self.baud} 8{letter}{self.stop_bits}"
553
554
555class SerialStep:
556    """One step of a scripted port: :meth:`write` for a write the program is expected to
557    make, and :meth:`read` for bytes the far end sends."""
558
559    __slots__ = ("_native",)
560
561    def __init__(self, native: _NativeSerialStep) -> None:
562        """Wrap a native step; use :meth:`write` or :meth:`read` instead."""
563        self._native = native
564
565    @classmethod
566    def write(cls, data: bytes) -> SerialStep:
567        """A write the program is expected to make, in one call.
568
569        :param data: The bytes of the write.
570        :returns: The step.
571        """
572        return cls(_NativeSerialStep.write(bytes(data)))
573
574    @classmethod
575    def read(cls, data: bytes) -> SerialStep:
576        """Bytes the far end sends, readable once every step before them has happened.
577
578        :param data: What arrives.
579        :returns: The step.
580        """
581        return cls(_NativeSerialStep.read(bytes(data)))
582
583
584class SerialPort:
585    """One serial port, shared by the program and every driver built on it.
586
587    :meth:`open` opens the kernel's serial device raw on a Linux board, :meth:`looped` is a
588    line with TX wired to RX, :meth:`pair` the two ends of a null-modem cable, and
589    :meth:`scripted` a port that checks each write against a script. A write returns once
590    the bytes have left the UART, and a read once bytes have arrived or its timeout has
591    passed. On anything but the kernel's device a read never waits: it returns at once, and
592    the time it would have waited is added to :attr:`waited_micros`. Every call releases the
593    interpreter while the line is busy, and a failure raises ``PamojaError``.
594
595    >>> gateway, node = SerialPort.pair(SerialSettings(115_200))
596    >>> node.write(b"t=21.5")
597    >>> gateway.read(16, timeout=0.1)
598    b't=21.5'
599    """
600
601    __slots__ = ("_native",)
602
603    def __init__(self, native: _NativeSerialPort) -> None:
604        """Wrap a native port; use one of the class methods instead."""
605        self._native = native
606
607    @classmethod
608    def open(cls, path: str, settings: SerialSettings) -> SerialPort:
609        """Open the kernel's serial device raw.
610
611        :param path: ``/dev/serial0`` for a Raspberry Pi's own UART, ``/dev/ttyUSB0`` or
612            ``/dev/ttyACM0`` for a USB adapter.
613        :param settings: The speed, a standard rate from 1200 to 921600, and the format.
614        :returns: The port.
615        :raises PamojaError: Anywhere but Linux, or when the device cannot be opened.
616        """
617        return cls(_NativeSerialPort.open(path, *settings._values()))
618
619    @classmethod
620    def looped(cls, settings: SerialSettings) -> SerialPort:
621        """A line looped back on itself: every byte written is waiting to be read.
622
623        :param settings: The speed and format the line runs at.
624        :returns: The port.
625        """
626        return cls(_NativeSerialPort.looped(*settings._values()))
627
628    @classmethod
629    def pair(cls, settings: SerialSettings) -> Tuple[SerialPort, SerialPort]:
630        """The two ends of a null-modem pair: what one end writes, the other reads.
631
632        :param settings: The speed and format both ends run at.
633        :returns: The two ends.
634        """
635        one, other = _NativeSerialPort.pair(*settings._values())
636        return cls(one), cls(other)
637
638    @classmethod
639    def scripted(cls, settings: SerialSettings, steps: Iterable[SerialStep]) -> SerialPort:
640        """A port that checks each write against the next step of a script.
641
642        :param settings: The speed and format the line runs at.
643        :param steps: The writes and reads, in order; reads at the start are there at once.
644        :returns: The port.
645        """
646        natives = [step._native for step in steps]
647        return cls(_NativeSerialPort.scripted(*settings._values(), natives))
648
649    @property
650    def kind(self) -> SerialPortKind:
651        """What is on the other end of the port."""
652        return SerialPortKind(self._native.kind)
653
654    @property
655    def settings(self) -> SerialSettings:
656        """The speed and character format the port runs at."""
657        baud, parity, stop_bits = self._native.settings
658        return SerialSettings(baud, Parity(parity), stop_bits)
659
660    def write(self, data: bytes) -> None:
661        """Write bytes, returning once they have left the UART.
662
663        :param data: The bytes, in order.
664        :raises PamojaError: When a script expected another write, or the device fails.
665        """
666        self._native.write(bytes(data))
667
668    def read(self, size: int, timeout: float) -> bytes:
669        """Read up to ``size`` bytes, waiting up to ``timeout`` for the first one when
670        nothing has arrived.
671
672        :param size: The most bytes to read.
673        :param timeout: How long to wait for the first byte, in seconds.
674        :returns: What arrived, empty when the timeout passed with nothing.
675        :raises PamojaError: When the device fails.
676        """
677        return self._native.read(size, _micros(timeout))
678
679    def discard_input(self) -> None:
680        """Drop whatever has arrived and not been read, as a client does before a request
681        so a stale reply cannot be taken for the new one.
682
683        :raises PamojaError: When the device fails.
684        """
685        self._native.discard_input()
686
687    def wait(self, seconds: float) -> None:
688        """Wait, as a protocol does to leave the line silent between frames: really on the
689        kernel's device, and anywhere else only counted.
690
691        :param seconds: How long.
692        """
693        self._native.wait(_micros(seconds))
694
695    @property
696    def written(self) -> int:
697        """How many bytes have been written through the port."""
698        return self._native.written
699
700    @property
701    def received(self) -> int:
702        """How many bytes have been read through the port."""
703        return self._native.received
704
705    @property
706    def waited_micros(self) -> int:
707        """How long reads have waited without an answer, and waits have waited, in
708        microseconds, whether or not the process slept through it."""
709        return self._native.waited_micros
710
711    @property
712    def remaining(self) -> Optional[int]:
713        """How many steps a script has left, or ``None`` when the port is not scripted."""
714        return self._native.remaining
715
716
717def _micros(seconds: float) -> int:
718    if seconds < 0:
719        raise ValueError("a time must be zero or more")
720    return round(seconds * 1_000_000)
721
722
723class Delay(Protocol):
724    """What paces a driver that has to wait between pin changes, such as a stepper
725    between steps. :class:`SleepDelay` really waits; :class:`DelayLog` counts every wait
726    and waits for none, for a program run with nothing plugged in."""
727
728    def delay_micros(self, micros: int) -> None:
729        """Wait, or count the wait.
730
731        :param micros: How long, in microseconds.
732        """
733        ...
734
735
736class DelayLog:
737    """A delay that records every wait it is asked for and sleeps through none of them,
738    as ``pamoja_hal::script::DelayLog`` does in Rust.
739
740    >>> delay = DelayLog()
741    >>> delay.delay_micros(480)
742    >>> delay.delay_micros(10_000)
743    >>> delay.total_micros, delay.total_millis
744    (10480, 10)
745    """
746
747    __slots__ = ("_total", "_waits")
748
749    def __init__(self) -> None:
750        """Create a log with nothing waited yet."""
751        self._waits: List[int] = []
752        self._total = 0
753
754    @property
755    def waits_micros(self) -> List[int]:
756        """Every wait asked for, in microseconds, oldest first."""
757        return list(self._waits)
758
759    @property
760    def total_micros(self) -> int:
761        """The waits added up, in microseconds."""
762        return self._total
763
764    @property
765    def total_millis(self) -> int:
766        """The waits added up, in whole milliseconds, rounded down."""
767        return self._total // 1_000
768
769    def delay_micros(self, micros: int) -> None:
770        """Record a wait.
771
772        :param micros: How long, in microseconds.
773        """
774        self._waits.append(micros)
775        self._total += micros
776
777    def clear(self) -> None:
778        """Forget every recorded wait."""
779        self._waits.clear()
780        self._total = 0
781
782
783class SleepDelay:
784    """A delay that really waits: :func:`time.sleep` for a millisecond or more, and a
785    spin on :func:`time.perf_counter_ns` for a shorter wait, which the scheduler cannot
786    keep. A sleep lasts at least what was asked and may run over by the scheduler's own
787    latency."""
788
789    __slots__ = ()
790
791    def delay_micros(self, micros: int) -> None:
792        """Wait.
793
794        :param micros: How long, in microseconds.
795        """
796        if micros >= 1_000:
797            time.sleep(micros / 1_000_000)
798            return
799        until = time.perf_counter_ns() + micros * 1_000
800        while time.perf_counter_ns() < until:
801            pass