pamoja.gpio

Idiomatic on-board bus facade.

Before a node reaches any network it talks to the chips wired to its own board. Three interfaces cover almost everything cheap hardware uses, and each carries one small piece of logic that is a classic field bug when it is wrong: the I2C address byte, the SPI clock mode, and whether a relay is active high or active low.

View Source

  1"""Idiomatic on-board bus facade.
  2
  3Before a node reaches any network it talks to the chips wired to its own board.
  4Three interfaces cover almost everything cheap hardware uses, and each carries one
  5small piece of logic that is a classic field bug when it is wrong: the I2C address
  6byte, the SPI clock mode, and whether a relay is active high or active low.
  7"""
  8
  9from __future__ import annotations
 10
 11import enum
 12from collections import deque
 13from typing import Generic, Iterable, Protocol, TypeVar
 14
 15from pamoja._native import GpioLine as _NativeLine
 16from pamoja._native import I2C_RESERVED_BELOW as _RESERVED_BELOW
 17from pamoja._native import I2C_RESERVED_FROM as _RESERVED_FROM
 18from pamoja._native import SpiClock
 19from pamoja._native import i2c_address_frame as _address_frame
 20from pamoja._native import i2c_address_frame_len as _address_frame_len
 21from pamoja._native import i2c_address_is_general_call as _is_general_call
 22from pamoja._native import i2c_address_is_reserved as _is_reserved
 23from pamoja._native import pin_edge_triggered_by as _edge_triggered_by
 24from pamoja._native import pin_level_from_bool as _level_from_bool
 25from pamoja._native import pin_level_inverted as _level_inverted
 26from pamoja._native import pin_polarity_is_asserted as _polarity_is_asserted
 27from pamoja._native import pin_polarity_level as _polarity_level
 28from pamoja._native import spi_mode_clock as _mode_clock
 29from pamoja._native import spi_mode_from_clock as _mode_from_clock
 30
 31__all__ = [
 32    "Contact",
 33    "Edge",
 34    "GpioLine",
 35    "InputLine",
 36    "Level",
 37    "OutputLine",
 38    "PinScript",
 39    "Polarity",
 40    "SpiClock",
 41    "Switch",
 42    "i2c",
 43    "pin",
 44    "spi",
 45]
 46
 47
 48class Level(str, enum.Enum):
 49    """The physical voltage level on a pin."""
 50
 51    #: A low level, near ground.
 52    LOW = "Low"
 53    #: A high level, near the supply voltage.
 54    HIGH = "High"
 55
 56
 57class Edge(str, enum.Enum):
 58    """The signal transition that triggers a pin interrupt."""
 59
 60    #: A low-to-high transition.
 61    RISING = "Rising"
 62    #: A high-to-low transition.
 63    FALLING = "Falling"
 64    #: Either transition.
 65    BOTH = "Both"
 66
 67
 68class Polarity(str, enum.Enum):
 69    """Whether a signal is asserted by a high or a low physical level."""
 70
 71    #: A high level means asserted.
 72    ACTIVE_HIGH = "ActiveHigh"
 73    #: A low level means asserted, the wiring of most buttons and relay boards.
 74    ACTIVE_LOW = "ActiveLow"
 75
 76
 77class _I2c:
 78    """I2C addressing per the NXP I2C-bus specification (UM10204)."""
 79
 80    __slots__ = ()
 81
 82    #: The lowest 7-bit address the specification keeps for itself.
 83    RESERVED_FROM = _RESERVED_FROM
 84
 85    #: The first 7-bit address above the reserved block at the bottom of the range.
 86    RESERVED_BELOW = _RESERVED_BELOW
 87
 88    def address_frame(
 89        self, address: int, *, read: bool = False, ten_bit: bool = False
 90    ) -> bytes:
 91        """Return the address bytes a controller puts on the bus for a transfer.
 92
 93        A 7-bit address frames as the single byte ``(address << 1) | r/w``; a
 94        10-bit one frames as two, the reserved ``11110`` prefix carrying the top
 95        two bits and the read/write bit, then the low eight.
 96
 97        :param address: The device address.
 98        :param read: Whether the transfer reads rather than writes.
 99        :param ten_bit: Whether this is a 10-bit address.
100        :returns: One byte for a 7-bit address, two for a 10-bit one.
101        :raises PamojaError: If the address is outside its width's range.
102        """
103        return _address_frame(address, ten_bit, read)
104
105    def frame_len(self, address: int, ten_bit: bool = False) -> int:
106        """Return how many bytes an address frame occupies.
107
108        :param address: The device address.
109        :param ten_bit: Whether this is a 10-bit address.
110        :returns: ``1`` for a 7-bit address, ``2`` for a 10-bit one.
111        :raises PamojaError: If the address is outside its width's range.
112        """
113        return _address_frame_len(address, ten_bit)
114
115    def is_reserved(self, address: int, ten_bit: bool = False) -> bool:
116        """Report whether an address falls in a range the specification reserves.
117
118        UM10204 reserves ``0x00..=0x07`` and ``0x78..=0x7F``, leaving
119        ``0x08..=0x77`` for ordinary devices.
120
121        :param address: The device address.
122        :param ten_bit: Whether this is a 10-bit address, never reserved in this
123            sense.
124        :returns: Whether the address is reserved.
125        :raises PamojaError: If the address is outside its width's range.
126        """
127        return _is_reserved(address, ten_bit)
128
129    def is_general_call(self, address: int, ten_bit: bool = False) -> bool:
130        """Report whether an address is the general call address ``0x00``.
131
132        :param address: The device address.
133        :param ten_bit: Whether this is a 10-bit address.
134        :returns: Whether this is the broadcast every device listens to.
135        :raises PamojaError: If the address is outside its width's range.
136        """
137        return _is_general_call(address, ten_bit)
138
139
140class _Spi:
141    """The four SPI clock modes, as the ``(CPOL, CPHA)`` pair datasheets quote."""
142
143    __slots__ = ()
144
145    def clock_for(self, mode: int) -> SpiClock:
146        """Return the clock polarity and phase a mode number names.
147
148        :param mode: The mode number, 0 to 3.
149        :returns: The pair.
150        :raises ValueError: If the mode number is above 3.
151        """
152        return _mode_clock(mode)
153
154    def mode_for(self, cpol: bool, cpha: bool) -> int:
155        """Return the mode number a clock polarity and phase name.
156
157        :param cpol: Whether the clock idles high.
158        :param cpha: Whether data is sampled on the trailing edge.
159        :returns: The mode number, 0 to 3. Every pair names a mode.
160        """
161        return _mode_from_clock(cpol, cpha)
162
163
164class _Pin:
165    """The GPIO pin model: levels, interrupt edges, and active polarity."""
166
167    __slots__ = ()
168
169    def level_from(self, high: bool) -> Level:
170        """Return the level a boolean names.
171
172        :param high: ``True`` for high, ``False`` for low.
173        :returns: The level.
174        """
175        return Level(_level_from_bool(high))
176
177    def invert(self, level: Level) -> Level:
178        """Return the opposite level.
179
180        :param level: The level to invert.
181        :returns: The other level.
182        """
183        return Level(_level_inverted(Level(level).value))
184
185    def triggers(self, edge: Edge, from_level: Level, to_level: Level) -> bool:
186        """Report whether a transition fires an interrupt trigger.
187
188        :param edge: The trigger configured on the pin.
189        :param from_level: The level before the change.
190        :param to_level: The level after it.
191        :returns: Whether the trigger fires.
192        """
193        return _edge_triggered_by(
194            Edge(edge).value, Level(from_level).value, Level(to_level).value
195        )
196
197    def level_for(self, polarity: Polarity, asserted: bool) -> Level:
198        """Return the physical level that represents a logical state.
199
200        :param polarity: How the signal is wired.
201        :param asserted: Whether the signal should be asserted.
202        :returns: The level to drive, inverted for active-low wiring.
203        """
204        return Level(_polarity_level(Polarity(polarity).value, asserted))
205
206    def is_asserted(self, polarity: Polarity, level: Level) -> bool:
207        """Report whether a physical level means the signal is asserted.
208
209        :param polarity: How the signal is wired.
210        :param level: The level read on the pin.
211        :returns: Whether the signal is asserted.
212        """
213        return _polarity_is_asserted(Polarity(polarity).value, Level(level).value)
214
215
216#: I2C addressing, validated before anything reaches the bus.
217i2c = _I2c()
218
219#: SPI clock modes, as a checked value rather than two transposable booleans.
220spi = _Spi()
221
222#: The GPIO pin model, so an active-low relay is handled by the type.
223pin = _Pin()
224
225
226class OutputLine(Protocol):
227    """A line a pin library drives: ``gpiozero`` or ``lgpio`` on a Raspberry Pi, a
228    vendor SDK on a microcontroller, or a :class:`PinScript` in a test. Anything with
229    this one method can sit under a :class:`Switch`."""
230
231    def drive(self, level: Level) -> None:
232        """Drive the line to a level.
233
234        :param level: The level to drive.
235        """
236        ...
237
238
239class InputLine(Protocol):
240    """A line a pin library reads, which a :class:`Contact` sits over."""
241
242    def read(self) -> Level:
243        """Read the line's level now.
244
245        :returns: The level on the line.
246        """
247        ...
248
249
250_Out = TypeVar("_Out", bound=OutputLine)
251_In = TypeVar("_In", bound=InputLine)
252
253
254class Switch(Generic[_Out]):
255    """A two-state output over any line, with its polarity said once: a relay, an
256    LED, a solenoid valve, a buzzer. ``set(True)`` asserts it, which drives the line
257    low for an active-low part, so no call site inverts a level by hand.
258
259    >>> pump = Switch.active_low(PinScript())
260    >>> pump.set(True)
261    >>> pump.is_asserted, pump.release().driven
262    (True, [<Level.LOW: 'Low'>])
263    """
264
265    __slots__ = ("_asserted", "_line", "_polarity")
266
267    def __init__(self, line: _Out, polarity: Polarity) -> None:
268        """Wrap a line, starting deasserted. Nothing is driven until :meth:`set`.
269
270        :param line: The line the part is wired to.
271        :param polarity: How the part is wired.
272        """
273        self._line = line
274        self._polarity = Polarity(polarity)
275        self._asserted = False
276
277    @classmethod
278    def active_high(cls, line: _Out) -> Switch[_Out]:
279        """A switch whose part is asserted by a high level.
280
281        :param line: The line the part is wired to.
282        :returns: The switch.
283        """
284        return cls(line, Polarity.ACTIVE_HIGH)
285
286    @classmethod
287    def active_low(cls, line: _Out) -> Switch[_Out]:
288        """A switch whose part is asserted by a low level, the wiring of most relay
289        boards.
290
291        :param line: The line the part is wired to.
292        :returns: The switch.
293        """
294        return cls(line, Polarity.ACTIVE_LOW)
295
296    @property
297    def polarity(self) -> Polarity:
298        """How the part is wired."""
299        return self._polarity
300
301    @property
302    def is_asserted(self) -> bool:
303        """Whether the part was last set on."""
304        return self._asserted
305
306    def set(self, asserted: bool) -> None:
307        """Turn the part on or off, driving whichever level that means for its wiring.
308
309        :param asserted: ``True`` to turn it on.
310        :raises Exception: Whatever the line raises when it cannot be driven.
311        """
312        self._line.drive(pin.level_for(self._polarity, asserted))
313        self._asserted = asserted
314
315    def release(self) -> _Out:
316        """Hand the line back, for a test to read what was driven or a program to
317        reuse it.
318
319        :returns: The line.
320        """
321        return self._line
322
323
324class Contact(Generic[_In]):
325    """A two-state input over any line, with its polarity said once: a button, a
326    float switch, a reed switch, a limit switch. :meth:`is_asserted` answers whether
327    it is closed, pressed, or tripped, whatever level that takes on the wire.
328
329    >>> float_switch = Contact.active_low(PinScript([Level.LOW]))
330    >>> float_switch.is_asserted()
331    True
332    """
333
334    __slots__ = ("_line", "_polarity")
335
336    def __init__(self, line: _In, polarity: Polarity) -> None:
337        """Wrap a line.
338
339        :param line: The line the part is wired to.
340        :param polarity: How the part is wired.
341        """
342        self._line = line
343        self._polarity = Polarity(polarity)
344
345    @classmethod
346    def active_high(cls, line: _In) -> Contact[_In]:
347        """A contact that reads high when asserted.
348
349        :param line: The line the part is wired to.
350        :returns: The contact.
351        """
352        return cls(line, Polarity.ACTIVE_HIGH)
353
354    @classmethod
355    def active_low(cls, line: _In) -> Contact[_In]:
356        """A contact that reads low when asserted, the wiring of a switch to ground
357        with a pull-up.
358
359        :param line: The line the part is wired to.
360        :returns: The contact.
361        """
362        return cls(line, Polarity.ACTIVE_LOW)
363
364    @property
365    def polarity(self) -> Polarity:
366        """How the part is wired."""
367        return self._polarity
368
369    def level(self) -> Level:
370        """Read the raw level on the line.
371
372        :returns: The level.
373        :raises Exception: Whatever the line raises when it cannot be read.
374        """
375        return Level(self._line.read())
376
377    def is_asserted(self) -> bool:
378        """Read the line and report whether the part is asserted.
379
380        :returns: Whether it is closed, pressed, or tripped.
381        :raises Exception: Whatever the line raises when it cannot be read.
382        """
383        return pin.is_asserted(self._polarity, self.level())
384
385    def release(self) -> _In:
386        """Hand the line back.
387
388        :returns: The line.
389        """
390        return self._line
391
392
393class GpioLine:
394    """A GPIO line opened on a Linux board, through the kernel's GPIO character device:
395    the line a :class:`Switch` or a :class:`Contact` sits over on a Raspberry Pi or any
396    Linux board.
397
398    :meth:`open_output` drives its initial level from the moment the line is taken, so an
399    active-low relay is opened ``Level.HIGH`` and stays off; :meth:`open_input` opens one
400    to read. The chip is a device file such as ``/dev/gpiochip0``, and the line is the GPIO
401    or BCM number a Raspberry Pi pinout gives. A line is held by one process at a time;
402    :meth:`close`, or leaving a ``with`` block, hands it back. On Raspberry Pi OS a user in
403    the ``gpio`` group opens lines without root. Opening raises ``PamojaError`` on any
404    platform but Linux, and names the chip and the line when either cannot be opened.
405    """
406
407    __slots__ = ("_native",)
408
409    def __init__(self, native: _NativeLine) -> None:
410        """Wrap an opened native line. Use :meth:`open_output` or :meth:`open_input`."""
411        self._native = native
412
413    @classmethod
414    def open_output(cls, chip: str, line: int, initial: Level) -> GpioLine:
415        """Open a line as an output, driving ``initial`` from the moment it is taken.
416
417        :param chip: The GPIO chip's device file, ``/dev/gpiochip0`` on most boards.
418        :param line: The line's number on that chip, the GPIO or BCM number on a
419            Raspberry Pi.
420        :param initial: The level to drive as soon as the line is taken.
421        :returns: The line.
422        :raises PamojaError: If the platform is not Linux, or the chip or the line
423            cannot be opened.
424        """
425        return cls(_NativeLine.open_output(chip, line, Level(initial).value))
426
427    @classmethod
428    def open_input(cls, chip: str, line: int) -> GpioLine:
429        """Open a line as an input.
430
431        :param chip: The GPIO chip's device file.
432        :param line: The line's number on that chip.
433        :returns: The line.
434        :raises PamojaError: If the platform is not Linux, or the chip or the line
435            cannot be opened.
436        """
437        return cls(_NativeLine.open_input(chip, line))
438
439    @property
440    def chip(self) -> str:
441        """The GPIO chip's device file."""
442        return self._native.chip
443
444    @property
445    def offset(self) -> int:
446        """The line's number on its chip."""
447        return self._native.offset
448
449    def drive(self, level: Level) -> None:
450        """Drive the line to a level. The line must have been opened as an output.
451
452        :param level: The level to drive.
453        :raises PamojaError: If the kernel refuses the write, or the line is closed.
454        """
455        self._native.drive(Level(level).value)
456
457    def read(self) -> Level:
458        """Read the level on the line now.
459
460        :returns: The level.
461        :raises PamojaError: If the kernel refuses the read, or the line is closed.
462        """
463        return Level(self._native.read())
464
465    def close(self) -> None:
466        """Hand the line back to the kernel. Calls after this raise ``PamojaError``."""
467        self._native.close()
468
469    def __enter__(self) -> GpioLine:
470        """Return the line, so it can be opened in a ``with`` block."""
471        return self
472
473    def __exit__(self, *exception: object) -> None:
474        """Close the line at the end of a ``with`` block, letting any exception through."""
475        self.close()
476
477
478class PinScript:
479    """A line for running with nothing plugged in: it answers the reads it was given,
480    in order, and records every level it is driven to. It is what the examples and
481    tests put under a :class:`Switch` or a :class:`Contact`, and the one thing a real
482    node replaces with its board's pin library. It behaves as
483    ``pamoja_hal::script::PinScript`` does in Rust: it starts released, high, and once
484    its reads run out a read answers the level it was last driven to.
485
486    >>> line = PinScript([Level.LOW])
487    >>> line.read(), line.read()
488    (<Level.LOW: 'Low'>, <Level.HIGH: 'High'>)
489    """
490
491    __slots__ = ("_driven", "_inputs", "_level")
492
493    def __init__(self, inputs: Iterable[Level] = ()) -> None:
494        """Create a released line.
495
496        :param inputs: The levels to answer reads with, in order.
497        """
498        self._inputs: deque[Level] = deque(Level(level) for level in inputs)
499        self._driven: list[Level] = []
500        self._level = Level.HIGH
501
502    @property
503    def driven(self) -> list[Level]:
504        """Every level the line was driven to, oldest first."""
505        return list(self._driven)
506
507    @property
508    def level(self) -> Level:
509        """The level the line was last driven to, high while it has never been
510        driven."""
511        return self._level
512
513    @property
514    def remaining(self) -> int:
515        """How many scripted reads are left."""
516        return len(self._inputs)
517
518    def drive(self, level: Level) -> None:
519        """Record a driven level.
520
521        :param level: The level driven.
522        """
523        self._level = Level(level)
524        self._driven.append(self._level)
525
526    def read(self) -> Level:
527        """Answer the next scripted level, or the driven level once the script runs
528        out.
529
530        :returns: The level.
531        """
532        return self._inputs.popleft() if self._inputs else self._level

class Contact(typing.Generic[~_In]): View Source

325class Contact(Generic[_In]):
326    """A two-state input over any line, with its polarity said once: a button, a
327    float switch, a reed switch, a limit switch. :meth:`is_asserted` answers whether
328    it is closed, pressed, or tripped, whatever level that takes on the wire.
329
330    >>> float_switch = Contact.active_low(PinScript([Level.LOW]))
331    >>> float_switch.is_asserted()
332    True
333    """
334
335    __slots__ = ("_line", "_polarity")
336
337    def __init__(self, line: _In, polarity: Polarity) -> None:
338        """Wrap a line.
339
340        :param line: The line the part is wired to.
341        :param polarity: How the part is wired.
342        """
343        self._line = line
344        self._polarity = Polarity(polarity)
345
346    @classmethod
347    def active_high(cls, line: _In) -> Contact[_In]:
348        """A contact that reads high when asserted.
349
350        :param line: The line the part is wired to.
351        :returns: The contact.
352        """
353        return cls(line, Polarity.ACTIVE_HIGH)
354
355    @classmethod
356    def active_low(cls, line: _In) -> Contact[_In]:
357        """A contact that reads low when asserted, the wiring of a switch to ground
358        with a pull-up.
359
360        :param line: The line the part is wired to.
361        :returns: The contact.
362        """
363        return cls(line, Polarity.ACTIVE_LOW)
364
365    @property
366    def polarity(self) -> Polarity:
367        """How the part is wired."""
368        return self._polarity
369
370    def level(self) -> Level:
371        """Read the raw level on the line.
372
373        :returns: The level.
374        :raises Exception: Whatever the line raises when it cannot be read.
375        """
376        return Level(self._line.read())
377
378    def is_asserted(self) -> bool:
379        """Read the line and report whether the part is asserted.
380
381        :returns: Whether it is closed, pressed, or tripped.
382        :raises Exception: Whatever the line raises when it cannot be read.
383        """
384        return pin.is_asserted(self._polarity, self.level())
385
386    def release(self) -> _In:
387        """Hand the line back.
388
389        :returns: The line.
390        """
391        return self._line

A two-state input over any line, with its polarity said once: a button, a float switch, a reed switch, a limit switch. is_asserted() answers whether it is closed, pressed, or tripped, whatever level that takes on the wire.

>>> float_switch = Contact.active_low(PinScript([Level.LOW]))
>>> float_switch.is_asserted()
True

Contact(line: ~_In, polarity: Polarity) View Source

337    def __init__(self, line: _In, polarity: Polarity) -> None:
338        """Wrap a line.
339
340        :param line: The line the part is wired to.
341        :param polarity: How the part is wired.
342        """
343        self._line = line
344        self._polarity = Polarity(polarity)

Wrap a line.

Parameters

line: The line the part is wired to.
polarity: How the part is wired.

@classmethod

def active_high(cls, line: ~_In) -> Contact[~_In]: View Source

346    @classmethod
347    def active_high(cls, line: _In) -> Contact[_In]:
348        """A contact that reads high when asserted.
349
350        :param line: The line the part is wired to.
351        :returns: The contact.
352        """
353        return cls(line, Polarity.ACTIVE_HIGH)

A contact that reads high when asserted.

Parameters

line: The line the part is wired to. :returns: The contact.

@classmethod

def active_low(cls, line: ~_In) -> Contact[~_In]: View Source

355    @classmethod
356    def active_low(cls, line: _In) -> Contact[_In]:
357        """A contact that reads low when asserted, the wiring of a switch to ground
358        with a pull-up.
359
360        :param line: The line the part is wired to.
361        :returns: The contact.
362        """
363        return cls(line, Polarity.ACTIVE_LOW)

A contact that reads low when asserted, the wiring of a switch to ground with a pull-up.

Parameters

line: The line the part is wired to. :returns: The contact.

polarity: Polarity View Source

365    @property
366    def polarity(self) -> Polarity:
367        """How the part is wired."""
368        return self._polarity

How the part is wired.

def level(self) -> Level: View Source

370    def level(self) -> Level:
371        """Read the raw level on the line.
372
373        :returns: The level.
374        :raises Exception: Whatever the line raises when it cannot be read.
375        """
376        return Level(self._line.read())

Read the raw level on the line.

:returns: The level.

Raises

Exception: Whatever the line raises when it cannot be read.

def is_asserted(self) -> bool: View Source

378    def is_asserted(self) -> bool:
379        """Read the line and report whether the part is asserted.
380
381        :returns: Whether it is closed, pressed, or tripped.
382        :raises Exception: Whatever the line raises when it cannot be read.
383        """
384        return pin.is_asserted(self._polarity, self.level())

Read the line and report whether the part is asserted.

:returns: Whether it is closed, pressed, or tripped.

Raises

Exception: Whatever the line raises when it cannot be read.

def release(self) -> ~_In: View Source

386    def release(self) -> _In:
387        """Hand the line back.
388
389        :returns: The line.
390        """
391        return self._line

Hand the line back.

:returns: The line.

class Edge(builtins.str, enum.Enum): View Source

58class Edge(str, enum.Enum):
59    """The signal transition that triggers a pin interrupt."""
60
61    #: A low-to-high transition.
62    RISING = "Rising"
63    #: A high-to-low transition.
64    FALLING = "Falling"
65    #: Either transition.
66    BOTH = "Both"

The signal transition that triggers a pin interrupt.

RISING = <Edge.RISING: 'Rising'>

FALLING = <Edge.FALLING: 'Falling'>

BOTH = <Edge.BOTH: 'Both'>

class GpioLine: View Source

394class GpioLine:
395    """A GPIO line opened on a Linux board, through the kernel's GPIO character device:
396    the line a :class:`Switch` or a :class:`Contact` sits over on a Raspberry Pi or any
397    Linux board.
398
399    :meth:`open_output` drives its initial level from the moment the line is taken, so an
400    active-low relay is opened ``Level.HIGH`` and stays off; :meth:`open_input` opens one
401    to read. The chip is a device file such as ``/dev/gpiochip0``, and the line is the GPIO
402    or BCM number a Raspberry Pi pinout gives. A line is held by one process at a time;
403    :meth:`close`, or leaving a ``with`` block, hands it back. On Raspberry Pi OS a user in
404    the ``gpio`` group opens lines without root. Opening raises ``PamojaError`` on any
405    platform but Linux, and names the chip and the line when either cannot be opened.
406    """
407
408    __slots__ = ("_native",)
409
410    def __init__(self, native: _NativeLine) -> None:
411        """Wrap an opened native line. Use :meth:`open_output` or :meth:`open_input`."""
412        self._native = native
413
414    @classmethod
415    def open_output(cls, chip: str, line: int, initial: Level) -> GpioLine:
416        """Open a line as an output, driving ``initial`` from the moment it is taken.
417
418        :param chip: The GPIO chip's device file, ``/dev/gpiochip0`` on most boards.
419        :param line: The line's number on that chip, the GPIO or BCM number on a
420            Raspberry Pi.
421        :param initial: The level to drive as soon as the line is taken.
422        :returns: The line.
423        :raises PamojaError: If the platform is not Linux, or the chip or the line
424            cannot be opened.
425        """
426        return cls(_NativeLine.open_output(chip, line, Level(initial).value))
427
428    @classmethod
429    def open_input(cls, chip: str, line: int) -> GpioLine:
430        """Open a line as an input.
431
432        :param chip: The GPIO chip's device file.
433        :param line: The line's number on that chip.
434        :returns: The line.
435        :raises PamojaError: If the platform is not Linux, or the chip or the line
436            cannot be opened.
437        """
438        return cls(_NativeLine.open_input(chip, line))
439
440    @property
441    def chip(self) -> str:
442        """The GPIO chip's device file."""
443        return self._native.chip
444
445    @property
446    def offset(self) -> int:
447        """The line's number on its chip."""
448        return self._native.offset
449
450    def drive(self, level: Level) -> None:
451        """Drive the line to a level. The line must have been opened as an output.
452
453        :param level: The level to drive.
454        :raises PamojaError: If the kernel refuses the write, or the line is closed.
455        """
456        self._native.drive(Level(level).value)
457
458    def read(self) -> Level:
459        """Read the level on the line now.
460
461        :returns: The level.
462        :raises PamojaError: If the kernel refuses the read, or the line is closed.
463        """
464        return Level(self._native.read())
465
466    def close(self) -> None:
467        """Hand the line back to the kernel. Calls after this raise ``PamojaError``."""
468        self._native.close()
469
470    def __enter__(self) -> GpioLine:
471        """Return the line, so it can be opened in a ``with`` block."""
472        return self
473
474    def __exit__(self, *exception: object) -> None:
475        """Close the line at the end of a ``with`` block, letting any exception through."""
476        self.close()

A GPIO line opened on a Linux board, through the kernel's GPIO character device: the line a Switch or a Contact sits over on a Raspberry Pi or any Linux board.

open_output() drives its initial level from the moment the line is taken, so an active-low relay is opened Level.HIGH and stays off; open_input() opens one to read. The chip is a device file such as /dev/gpiochip0, and the line is the GPIO or BCM number a Raspberry Pi pinout gives. A line is held by one process at a time; close(), or leaving a with block, hands it back. On Raspberry Pi OS a user in the gpio group opens lines without root. Opening raises PamojaError on any platform but Linux, and names the chip and the line when either cannot be opened.

GpioLine(native: GpioLine) View Source

410    def __init__(self, native: _NativeLine) -> None:
411        """Wrap an opened native line. Use :meth:`open_output` or :meth:`open_input`."""
412        self._native = native

Wrap an opened native line. Use open_output() or open_input().

@classmethod

def open_output( cls, chip: str, line: int, initial: Level) -> GpioLine: View Source

414    @classmethod
415    def open_output(cls, chip: str, line: int, initial: Level) -> GpioLine:
416        """Open a line as an output, driving ``initial`` from the moment it is taken.
417
418        :param chip: The GPIO chip's device file, ``/dev/gpiochip0`` on most boards.
419        :param line: The line's number on that chip, the GPIO or BCM number on a
420            Raspberry Pi.
421        :param initial: The level to drive as soon as the line is taken.
422        :returns: The line.
423        :raises PamojaError: If the platform is not Linux, or the chip or the line
424            cannot be opened.
425        """
426        return cls(_NativeLine.open_output(chip, line, Level(initial).value))

Open a line as an output, driving initial from the moment it is taken.

Parameters

chip: The GPIO chip's device file, /dev/gpiochip0 on most boards.
line: The line's number on that chip, the GPIO or BCM number on a Raspberry Pi.
initial: The level to drive as soon as the line is taken. :returns: The line.

Raises

PamojaError: If the platform is not Linux, or the chip or the line cannot be opened.

@classmethod

def open_input(cls, chip: str, line: int) -> GpioLine: View Source

428    @classmethod
429    def open_input(cls, chip: str, line: int) -> GpioLine:
430        """Open a line as an input.
431
432        :param chip: The GPIO chip's device file.
433        :param line: The line's number on that chip.
434        :returns: The line.
435        :raises PamojaError: If the platform is not Linux, or the chip or the line
436            cannot be opened.
437        """
438        return cls(_NativeLine.open_input(chip, line))

Open a line as an input.

Parameters

chip: The GPIO chip's device file.
line: The line's number on that chip. :returns: The line.

Raises

PamojaError: If the platform is not Linux, or the chip or the line cannot be opened.

chip: str View Source

440    @property
441    def chip(self) -> str:
442        """The GPIO chip's device file."""
443        return self._native.chip

The GPIO chip's device file.

offset: int View Source

445    @property
446    def offset(self) -> int:
447        """The line's number on its chip."""
448        return self._native.offset

The line's number on its chip.

def drive(self, level: Level) -> None: View Source

450    def drive(self, level: Level) -> None:
451        """Drive the line to a level. The line must have been opened as an output.
452
453        :param level: The level to drive.
454        :raises PamojaError: If the kernel refuses the write, or the line is closed.
455        """
456        self._native.drive(Level(level).value)

Drive the line to a level. The line must have been opened as an output.

Parameters

level: The level to drive.

Raises

PamojaError: If the kernel refuses the write, or the line is closed.

def read(self) -> Level: View Source

458    def read(self) -> Level:
459        """Read the level on the line now.
460
461        :returns: The level.
462        :raises PamojaError: If the kernel refuses the read, or the line is closed.
463        """
464        return Level(self._native.read())

Read the level on the line now.

:returns: The level.

Raises

PamojaError: If the kernel refuses the read, or the line is closed.

def close(self) -> None: View Source

466    def close(self) -> None:
467        """Hand the line back to the kernel. Calls after this raise ``PamojaError``."""
468        self._native.close()

Hand the line back to the kernel. Calls after this raise PamojaError.

class InputLine(typing.Protocol): View Source

240class InputLine(Protocol):
241    """A line a pin library reads, which a :class:`Contact` sits over."""
242
243    def read(self) -> Level:
244        """Read the line's level now.
245
246        :returns: The level on the line.
247        """
248        ...

A line a pin library reads, which a Contact sits over.

InputLine(*args, **kwargs) View Source

1968def _no_init_or_replace_init(self, *args, **kwargs):
1969    cls = type(self)
1970
1971    if cls._is_protocol:
1972        raise TypeError('Protocols cannot be instantiated')
1973
1974    # Already using a custom `__init__`. No need to calculate correct
1975    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1976    if cls.__init__ is not _no_init_or_replace_init:
1977        return
1978
1979    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1980    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1981    # searches for a proper new `__init__` in the MRO. The new `__init__`
1982    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1983    # instantiation of the protocol subclass will thus use the new
1984    # `__init__` and no longer call `_no_init_or_replace_init`.
1985    for base in cls.__mro__:
1986        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1987        if init is not _no_init_or_replace_init:
1988            cls.__init__ = init
1989            break
1990    else:
1991        # should not happen
1992        cls.__init__ = object.__init__
1993
1994    cls.__init__(self, *args, **kwargs)

def read(self) -> Level: View Source

243    def read(self) -> Level:
244        """Read the line's level now.
245
246        :returns: The level on the line.
247        """
248        ...

Read the line's level now.

:returns: The level on the line.

class Level(builtins.str, enum.Enum): View Source

49class Level(str, enum.Enum):
50    """The physical voltage level on a pin."""
51
52    #: A low level, near ground.
53    LOW = "Low"
54    #: A high level, near the supply voltage.
55    HIGH = "High"

The physical voltage level on a pin.

LOW = <Level.LOW: 'Low'>

HIGH = <Level.HIGH: 'High'>

class OutputLine(typing.Protocol): View Source

227class OutputLine(Protocol):
228    """A line a pin library drives: ``gpiozero`` or ``lgpio`` on a Raspberry Pi, a
229    vendor SDK on a microcontroller, or a :class:`PinScript` in a test. Anything with
230    this one method can sit under a :class:`Switch`."""
231
232    def drive(self, level: Level) -> None:
233        """Drive the line to a level.
234
235        :param level: The level to drive.
236        """
237        ...

A line a pin library drives: gpiozero or lgpio on a Raspberry Pi, a vendor SDK on a microcontroller, or a PinScript in a test. Anything with this one method can sit under a Switch.

OutputLine(*args, **kwargs) View Source

1968def _no_init_or_replace_init(self, *args, **kwargs):
1969    cls = type(self)
1970
1971    if cls._is_protocol:
1972        raise TypeError('Protocols cannot be instantiated')
1973
1974    # Already using a custom `__init__`. No need to calculate correct
1975    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1976    if cls.__init__ is not _no_init_or_replace_init:
1977        return
1978
1979    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1980    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1981    # searches for a proper new `__init__` in the MRO. The new `__init__`
1982    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1983    # instantiation of the protocol subclass will thus use the new
1984    # `__init__` and no longer call `_no_init_or_replace_init`.
1985    for base in cls.__mro__:
1986        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1987        if init is not _no_init_or_replace_init:
1988            cls.__init__ = init
1989            break
1990    else:
1991        # should not happen
1992        cls.__init__ = object.__init__
1993
1994    cls.__init__(self, *args, **kwargs)

def drive(self, level: Level) -> None: View Source

232    def drive(self, level: Level) -> None:
233        """Drive the line to a level.
234
235        :param level: The level to drive.
236        """
237        ...

Drive the line to a level.

Parameters

level: The level to drive.

class PinScript: View Source

479class PinScript:
480    """A line for running with nothing plugged in: it answers the reads it was given,
481    in order, and records every level it is driven to. It is what the examples and
482    tests put under a :class:`Switch` or a :class:`Contact`, and the one thing a real
483    node replaces with its board's pin library. It behaves as
484    ``pamoja_hal::script::PinScript`` does in Rust: it starts released, high, and once
485    its reads run out a read answers the level it was last driven to.
486
487    >>> line = PinScript([Level.LOW])
488    >>> line.read(), line.read()
489    (<Level.LOW: 'Low'>, <Level.HIGH: 'High'>)
490    """
491
492    __slots__ = ("_driven", "_inputs", "_level")
493
494    def __init__(self, inputs: Iterable[Level] = ()) -> None:
495        """Create a released line.
496
497        :param inputs: The levels to answer reads with, in order.
498        """
499        self._inputs: deque[Level] = deque(Level(level) for level in inputs)
500        self._driven: list[Level] = []
501        self._level = Level.HIGH
502
503    @property
504    def driven(self) -> list[Level]:
505        """Every level the line was driven to, oldest first."""
506        return list(self._driven)
507
508    @property
509    def level(self) -> Level:
510        """The level the line was last driven to, high while it has never been
511        driven."""
512        return self._level
513
514    @property
515    def remaining(self) -> int:
516        """How many scripted reads are left."""
517        return len(self._inputs)
518
519    def drive(self, level: Level) -> None:
520        """Record a driven level.
521
522        :param level: The level driven.
523        """
524        self._level = Level(level)
525        self._driven.append(self._level)
526
527    def read(self) -> Level:
528        """Answer the next scripted level, or the driven level once the script runs
529        out.
530
531        :returns: The level.
532        """
533        return self._inputs.popleft() if self._inputs else self._level

A line for running with nothing plugged in: it answers the reads it was given, in order, and records every level it is driven to. It is what the examples and tests put under a Switch or a Contact, and the one thing a real node replaces with its board's pin library. It behaves as pamoja_hal::script::PinScript does in Rust: it starts released, high, and once its reads run out a read answers the level it was last driven to.

>>> line = PinScript([Level.LOW])
>>> line.read(), line.read()
(<Level.LOW: 'Low'>, <Level.HIGH: 'High'>)

PinScript(inputs: Iterable[Level] = ()) View Source

494    def __init__(self, inputs: Iterable[Level] = ()) -> None:
495        """Create a released line.
496
497        :param inputs: The levels to answer reads with, in order.
498        """
499        self._inputs: deque[Level] = deque(Level(level) for level in inputs)
500        self._driven: list[Level] = []
501        self._level = Level.HIGH

Create a released line.

Parameters

inputs: The levels to answer reads with, in order.

driven: list[Level] View Source

503    @property
504    def driven(self) -> list[Level]:
505        """Every level the line was driven to, oldest first."""
506        return list(self._driven)

Every level the line was driven to, oldest first.

level: Level View Source

508    @property
509    def level(self) -> Level:
510        """The level the line was last driven to, high while it has never been
511        driven."""
512        return self._level

The level the line was last driven to, high while it has never been driven.

remaining: int View Source

514    @property
515    def remaining(self) -> int:
516        """How many scripted reads are left."""
517        return len(self._inputs)

How many scripted reads are left.

def drive(self, level: Level) -> None: View Source

519    def drive(self, level: Level) -> None:
520        """Record a driven level.
521
522        :param level: The level driven.
523        """
524        self._level = Level(level)
525        self._driven.append(self._level)

Record a driven level.

Parameters

level: The level driven.

def read(self) -> Level: View Source

527    def read(self) -> Level:
528        """Answer the next scripted level, or the driven level once the script runs
529        out.
530
531        :returns: The level.
532        """
533        return self._inputs.popleft() if self._inputs else self._level

Answer the next scripted level, or the driven level once the script runs out.

:returns: The level.

class Polarity(builtins.str, enum.Enum): View Source

69class Polarity(str, enum.Enum):
70    """Whether a signal is asserted by a high or a low physical level."""
71
72    #: A high level means asserted.
73    ACTIVE_HIGH = "ActiveHigh"
74    #: A low level means asserted, the wiring of most buttons and relay boards.
75    ACTIVE_LOW = "ActiveLow"

Whether a signal is asserted by a high or a low physical level.

ACTIVE_HIGH = <Polarity.ACTIVE_HIGH: 'ActiveHigh'>

ACTIVE_LOW = <Polarity.ACTIVE_LOW: 'ActiveLow'>

class SpiClock:

The clock polarity and phase pair an SPI mode number names.

cpha

Whether data is sampled on the trailing edge (CPHA = 1), which is modes 1 and 3.

cpol

Whether the clock idles high (CPOL = 1), which is modes 2 and 3.

class Switch(typing.Generic[~_Out]): View Source

255class Switch(Generic[_Out]):
256    """A two-state output over any line, with its polarity said once: a relay, an
257    LED, a solenoid valve, a buzzer. ``set(True)`` asserts it, which drives the line
258    low for an active-low part, so no call site inverts a level by hand.
259
260    >>> pump = Switch.active_low(PinScript())
261    >>> pump.set(True)
262    >>> pump.is_asserted, pump.release().driven
263    (True, [<Level.LOW: 'Low'>])
264    """
265
266    __slots__ = ("_asserted", "_line", "_polarity")
267
268    def __init__(self, line: _Out, polarity: Polarity) -> None:
269        """Wrap a line, starting deasserted. Nothing is driven until :meth:`set`.
270
271        :param line: The line the part is wired to.
272        :param polarity: How the part is wired.
273        """
274        self._line = line
275        self._polarity = Polarity(polarity)
276        self._asserted = False
277
278    @classmethod
279    def active_high(cls, line: _Out) -> Switch[_Out]:
280        """A switch whose part is asserted by a high level.
281
282        :param line: The line the part is wired to.
283        :returns: The switch.
284        """
285        return cls(line, Polarity.ACTIVE_HIGH)
286
287    @classmethod
288    def active_low(cls, line: _Out) -> Switch[_Out]:
289        """A switch whose part is asserted by a low level, the wiring of most relay
290        boards.
291
292        :param line: The line the part is wired to.
293        :returns: The switch.
294        """
295        return cls(line, Polarity.ACTIVE_LOW)
296
297    @property
298    def polarity(self) -> Polarity:
299        """How the part is wired."""
300        return self._polarity
301
302    @property
303    def is_asserted(self) -> bool:
304        """Whether the part was last set on."""
305        return self._asserted
306
307    def set(self, asserted: bool) -> None:
308        """Turn the part on or off, driving whichever level that means for its wiring.
309
310        :param asserted: ``True`` to turn it on.
311        :raises Exception: Whatever the line raises when it cannot be driven.
312        """
313        self._line.drive(pin.level_for(self._polarity, asserted))
314        self._asserted = asserted
315
316    def release(self) -> _Out:
317        """Hand the line back, for a test to read what was driven or a program to
318        reuse it.
319
320        :returns: The line.
321        """
322        return self._line

A two-state output over any line, with its polarity said once: a relay, an LED, a solenoid valve, a buzzer. set(True) asserts it, which drives the line low for an active-low part, so no call site inverts a level by hand.

>>> pump = Switch.active_low(PinScript())
>>> pump.set(True)
>>> pump.is_asserted, pump.release().driven
(True, [<Level.LOW: 'Low'>])

Switch(line: ~_Out, polarity: Polarity) View Source

268    def __init__(self, line: _Out, polarity: Polarity) -> None:
269        """Wrap a line, starting deasserted. Nothing is driven until :meth:`set`.
270
271        :param line: The line the part is wired to.
272        :param polarity: How the part is wired.
273        """
274        self._line = line
275        self._polarity = Polarity(polarity)
276        self._asserted = False

Wrap a line, starting deasserted. Nothing is driven until set().

Parameters

line: The line the part is wired to.
polarity: How the part is wired.

@classmethod

def active_high(cls, line: ~_Out) -> Switch[~_Out]: View Source

278    @classmethod
279    def active_high(cls, line: _Out) -> Switch[_Out]:
280        """A switch whose part is asserted by a high level.
281
282        :param line: The line the part is wired to.
283        :returns: The switch.
284        """
285        return cls(line, Polarity.ACTIVE_HIGH)

A switch whose part is asserted by a high level.

Parameters

line: The line the part is wired to. :returns: The switch.

@classmethod

def active_low(cls, line: ~_Out) -> Switch[~_Out]: View Source

287    @classmethod
288    def active_low(cls, line: _Out) -> Switch[_Out]:
289        """A switch whose part is asserted by a low level, the wiring of most relay
290        boards.
291
292        :param line: The line the part is wired to.
293        :returns: The switch.
294        """
295        return cls(line, Polarity.ACTIVE_LOW)

A switch whose part is asserted by a low level, the wiring of most relay boards.

Parameters

line: The line the part is wired to. :returns: The switch.

polarity: Polarity View Source

297    @property
298    def polarity(self) -> Polarity:
299        """How the part is wired."""
300        return self._polarity

How the part is wired.

is_asserted: bool View Source

302    @property
303    def is_asserted(self) -> bool:
304        """Whether the part was last set on."""
305        return self._asserted

Whether the part was last set on.

def set(self, asserted: bool) -> None: View Source

307    def set(self, asserted: bool) -> None:
308        """Turn the part on or off, driving whichever level that means for its wiring.
309
310        :param asserted: ``True`` to turn it on.
311        :raises Exception: Whatever the line raises when it cannot be driven.
312        """
313        self._line.drive(pin.level_for(self._polarity, asserted))
314        self._asserted = asserted

Turn the part on or off, driving whichever level that means for its wiring.

Parameters

asserted: True to turn it on.

Raises

Exception: Whatever the line raises when it cannot be driven.

def release(self) -> ~_Out: View Source

316    def release(self) -> _Out:
317        """Hand the line back, for a test to read what was driven or a program to
318        reuse it.
319
320        :returns: The line.
321        """
322        return self._line

Hand the line back, for a test to read what was driven or a program to reuse it.

:returns: The line.

i2c = <pamoja.gpio._I2c object>

pin = <pamoja.gpio._Pin object>

spi = <pamoja.gpio._Spi object>