pamoja.actuators

  1"""Idiomatic actuator-driver facade.
  2
  3Two parts that move something: a PCA9685 driving up to sixteen servos, LEDs, or valves,
  4and a stepper motor walked one coil pattern at a time. :class:`Pca9685` drives the part
  5over an :class:`~pamoja.hal.I2cBus`, and ``pca9685.sim.part`` stands one up on a simulated
  6bus that keeps the datasheet's rules. The ``pwm`` builders make the four register bytes of
  7one channel's setting. :class:`FourWire` and :class:`StepDir` drive a stepper through its
  8pins, over any :class:`~pamoja.gpio.OutputLine`.
  9"""
 10
 11from __future__ import annotations
 12
 13import enum
 14from typing import Generic, NamedTuple, Optional, Tuple, TypeVar
 15
 16from pamoja._native import Pca9685 as _NativePca9685
 17from pamoja._native import Stepper as _NativeStepper
 18from pamoja._native import pca9685_sim_part as _pca9685_sim_part
 19from pamoja._native import pca9685_channel_register as _channel_register
 20from pamoja._native import pca9685_frequency_for_prescale as _frequency_for_prescale
 21from pamoja._native import pca9685_limits as _limits
 22from pamoja._native import pca9685_prescale_for_frequency as _prescale_for_frequency
 23from pamoja._native import pwm_counts as _pwm_counts
 24from pamoja._native import pwm_duty as _pwm_duty
 25from pamoja._native import pwm_from_counts as _pwm_from_counts
 26from pamoja._native import pwm_full_off as _pwm_full_off
 27from pamoja._native import pwm_full_on as _pwm_full_on
 28from pamoja._native import pwm_servo as _pwm_servo
 29from pamoja._native import stepper_step_count as _step_count
 30from pamoja._native import stepper_steps_for_degrees as _steps_for_degrees
 31from pamoja._native import stepper_timing as _stepper_timing
 32from pamoja.gpio import Level, OutputLine
 33from pamoja.hal import Delay, I2cBus, I2cPart, SleepDelay
 34
 35__all__ = [
 36    "Direction",
 37    "Drive",
 38    "FourWire",
 39    "Pca9685",
 40    "Pwm",
 41    "StepDir",
 42    "Stepper",
 43    "pca9685",
 44    "pwm",
 45    "stepper",
 46    "steps_for_degrees",
 47]
 48
 49_INTERNAL_OSC_HZ, _CHANNELS, _COUNTS = _limits()
 50_DEFAULT_STEP_MICROS, _DEFAULT_PULSE_MICROS = _stepper_timing()
 51
 52_Out = TypeVar("_Out", bound=OutputLine)
 53_Step = TypeVar("_Step", bound=OutputLine)
 54_Dir = TypeVar("_Dir", bound=OutputLine)
 55
 56
 57class Drive(str, enum.Enum):
 58    """A stepper drive pattern, trading torque, smoothness, and resolution."""
 59
 60    #: One coil energized at a time: four steps, least torque and least power.
 61    WAVE = "Wave"
 62    #: Two adjacent coils at a time: four steps, most torque.
 63    FULL_STEP = "FullStep"
 64    #: Alternating one and two coils: eight steps, double resolution.
 65    HALF_STEP = "HalfStep"
 66
 67    @property
 68    def step_count(self) -> int:
 69        """How many steps make up one electrical cycle of this pattern."""
 70        return _step_count(self.value)
 71
 72
 73class Direction(str, enum.Enum):
 74    """Which way to step a motor."""
 75
 76    #: Advance the sequence, turning the shaft one way.
 77    FORWARD = "Forward"
 78    #: Reverse the sequence, turning the shaft the other way.
 79    BACKWARD = "Backward"
 80
 81
 82class Stepper:
 83    """A stepper motor's place in its drive sequence, and how far it has turned.
 84
 85    Example::
 86
 87        motor = Stepper(Drive.HALF_STEP)
 88        coils = motor.step(Direction.FORWARD)
 89    """
 90
 91    __slots__ = ("_native",)
 92
 93    def __init__(self, drive: Drive) -> None:
 94        """Create a stepper at the start of a pattern, with its position at zero.
 95
 96        :param drive: The coil pattern to walk.
 97        """
 98        self._native = _NativeStepper(Drive(drive).value)
 99
100    def step(self, direction: Direction) -> int:
101        """Advance one step and return the four-bit coil pattern to apply.
102
103        The most significant of the four bits is the first coil.
104
105        :param direction: Which way to turn.
106        :returns: The coil pattern.
107        """
108        return self._native.step(Direction(direction).value)
109
110    @property
111    def coils(self) -> int:
112        """The coil pattern currently held, without advancing."""
113        return self._native.coils
114
115    @property
116    def steps(self) -> int:
117        """How many steps have been taken, signed by direction."""
118        return self._native.steps
119
120
121def _direction_of(count: int) -> Direction:
122    return Direction.BACKWARD if count < 0 else Direction.FORWARD
123
124
125class FourWire(Generic[_Out]):
126    """A four-wire stepper driven coil by coil, through a transistor array such as a
127    ULN2003 or an H-bridge, as ``pamoja_actuators::stepper::FourWire`` drives one in Rust.
128
129    Each step energizes the coils the drive pattern names, coil A for the pattern's high
130    bit down to coil D for its low bit, then waits the step interval. The position counts
131    steps from where the motor was when the driver was built, forward positive. A coil
132    line is any :class:`~pamoja.gpio.OutputLine`: a :class:`~pamoja.gpio.GpioLine` on a
133    board, or a :class:`~pamoja.gpio.PinScript` that records every level.
134
135    >>> from pamoja.gpio import PinScript
136    >>> from pamoja.hal import DelayLog
137    >>> coils = (PinScript(), PinScript(), PinScript(), PinScript())
138    >>> motor = FourWire(coils, Drive.FULL_STEP, delay=DelayLog())
139    >>> motor.steps(2)
140    >>> motor.position
141    2
142    """
143
144    __slots__ = ("_coils", "_delay", "_drive", "_position", "_sequence", "_step_micros")
145
146    def __init__(
147        self,
148        coils: Tuple[_Out, _Out, _Out, _Out],
149        drive: Drive,
150        *,
151        step_micros: int = _DEFAULT_STEP_MICROS,
152        delay: Optional[Delay] = None,
153    ) -> None:
154        """Wrap the four coil lines, without driving them.
155
156        :param coils: The lines for coils A, B, C, and D, in the motor's phase order.
157        :param drive: The coil pattern to step through.
158        :param step_micros: The pause after each step, which sets the speed.
159        :param delay: What waits: a :class:`~pamoja.hal.SleepDelay` unless given, or a
160            :class:`~pamoja.hal.DelayLog` to run with nothing plugged in.
161        :raises ValueError: If there are not four coil lines.
162        """
163        lines = tuple(coils)
164        if len(lines) != 4:
165            raise ValueError(f"a four-wire stepper has four coil lines, not {len(lines)}")
166        a, b, c, d = lines
167        self._coils = (a, b, c, d)
168        self._drive = Drive(drive)
169        self._sequence = _NativeStepper(self._drive.value)
170        self._position = 0
171        self._step_micros = step_micros
172        self._delay: Delay = SleepDelay() if delay is None else delay
173
174    @property
175    def position(self) -> int:
176        """The step count since the driver was built, forward positive."""
177        return self._position
178
179    @property
180    def drive(self) -> Drive:
181        """The coil pattern in use."""
182        return self._drive
183
184    @property
185    def step_micros(self) -> int:
186        """The pause after each step, in microseconds."""
187        return self._step_micros
188
189    def step(self, direction: Direction) -> None:
190        """Take one step and wait the step interval.
191
192        :param direction: Which way to step.
193        :raises Exception: Whatever a coil line raises when it cannot be driven.
194        """
195        direction = Direction(direction)
196        self._energize(self._sequence.step(direction.value))
197        self._position += 1 if direction is Direction.FORWARD else -1
198        self._delay.delay_micros(self._step_micros)
199
200    def steps(self, count: int) -> None:
201        """Take ``count`` steps, backward when negative.
202
203        :param count: The signed number of steps.
204        :raises Exception: Whatever a coil line raises; the steps already taken stay
205            counted.
206        """
207        direction = _direction_of(count)
208        for _ in range(abs(count)):
209            self.step(direction)
210
211    def idle(self) -> None:
212        """Drop every coil, so the motor holds nothing and draws nothing.
213
214        :raises Exception: Whatever a coil line raises when it cannot be driven.
215        """
216        self._energize(0)
217
218    def release(self) -> Tuple[_Out, _Out, _Out, _Out]:
219        """Hand the coil lines back, for a test to read what was driven or a program to
220        reuse them.
221
222        :returns: The lines for coils A, B, C, and D.
223        """
224        return self._coils
225
226    def _energize(self, coils: int) -> None:
227        for index, line in enumerate(self._coils):
228            line.drive(Level.HIGH if coils & (0b1000 >> index) else Level.LOW)
229
230
231class StepDir(Generic[_Step, _Dir]):
232    """A stepper behind a step and direction driver chip such as an A4988 or a DRV8825,
233    as ``pamoja_actuators::stepper::StepDir`` drives one in Rust.
234
235    Each step sets the direction line, high for forward, and holds it for the pulse
236    width, since the chip reads the direction on the step line's rising edge and needs it
237    settled first. Then it pulses the step line high for the pulse width, brings it low,
238    and waits the step interval. Microstepping, current limiting, and enable are the
239    chip's own pins and settings, outside this driver.
240
241    >>> from pamoja.gpio import PinScript
242    >>> from pamoja.hal import DelayLog
243    >>> motor = StepDir(PinScript(), PinScript(), delay=DelayLog())
244    >>> motor.steps(-1)
245    >>> motor.position
246    -1
247    """
248
249    __slots__ = (
250        "_delay",
251        "_direction",
252        "_position",
253        "_pulse_micros",
254        "_step",
255        "_step_micros",
256    )
257
258    def __init__(
259        self,
260        step: _Step,
261        direction: _Dir,
262        *,
263        pulse_micros: int = _DEFAULT_PULSE_MICROS,
264        step_micros: int = _DEFAULT_STEP_MICROS,
265        delay: Optional[Delay] = None,
266    ) -> None:
267        """Wrap the step and direction lines, without driving them.
268
269        :param step: The line the chip counts rising edges on.
270        :param direction: The line the chip reads the direction from; high is forward.
271        :param pulse_micros: How long the direction line is held before the step line
272            rises, and the step line is then held high.
273        :param step_micros: The pause after each step, which sets the speed.
274        :param delay: What waits: a :class:`~pamoja.hal.SleepDelay` unless given, or a
275            :class:`~pamoja.hal.DelayLog` to run with nothing plugged in.
276        """
277        self._step = step
278        self._direction = direction
279        self._position = 0
280        self._pulse_micros = pulse_micros
281        self._step_micros = step_micros
282        self._delay: Delay = SleepDelay() if delay is None else delay
283
284    @property
285    def position(self) -> int:
286        """The step count since the driver was built, forward positive."""
287        return self._position
288
289    @property
290    def pulse_micros(self) -> int:
291        """How long the direction is held before each step pulse, and the pulse itself, in
292        microseconds."""
293        return self._pulse_micros
294
295    @property
296    def step_micros(self) -> int:
297        """The pause after each step, in microseconds."""
298        return self._step_micros
299
300    def step(self, direction: Direction) -> None:
301        """Take one step and wait the step interval.
302
303        :param direction: Which way to step.
304        :raises Exception: Whatever a line raises when it cannot be driven.
305        """
306        forward = Direction(direction) is Direction.FORWARD
307        self._direction.drive(Level.HIGH if forward else Level.LOW)
308        self._delay.delay_micros(self._pulse_micros)
309        self._step.drive(Level.HIGH)
310        self._delay.delay_micros(self._pulse_micros)
311        self._step.drive(Level.LOW)
312        self._position += 1 if forward else -1
313        self._delay.delay_micros(self._step_micros)
314
315    def steps(self, count: int) -> None:
316        """Take ``count`` steps, backward when negative.
317
318        :param count: The signed number of steps.
319        :raises Exception: Whatever a line raises; the steps already taken stay counted.
320        """
321        direction = _direction_of(count)
322        for _ in range(abs(count)):
323            self.step(direction)
324
325    def release(self) -> Tuple[_Step, _Dir]:
326        """Hand the lines back.
327
328        :returns: The step line, then the direction line.
329        """
330        return self._step, self._direction
331
332
333class Pwm(NamedTuple):
334    """When in a period a PWM output goes high and low, in counts."""
335
336    on: int
337    """The count at which the output goes high."""
338
339    off: int
340    """The count at which it goes low, or the full-off flag."""
341
342
343class Pca9685:
344    """An NXP PCA9685 on an I2C bus, driving sixteen PWM outputs.
345
346    The driver programs the prescale for its frequency with the oscillator asleep, because
347    the part only takes a prescale then, wakes it, waits the 500 us it needs, and loads a
348    channel's four registers in one transfer. Every call releases the interpreter while the
349    bus is busy.
350
351    Example::
352
353        board = Pca9685(bus, pca9685.DEFAULT_ADDRESS, frequency_hz=50)
354        board.set_channel(0, pwm.servo(1500))
355    """
356
357    __slots__ = ("_native",)
358
359    def __init__(
360        self,
361        bus: I2cBus,
362        address: int = 0x40,
363        *,
364        frequency_hz: int = 200,
365        oscillator_hz: int = _INTERNAL_OSC_HZ,
366        totem_pole: bool = True,
367        inverted: bool = False,
368        change_on_ack: bool = False,
369    ) -> None:
370        """Make a driver for the part at ``address`` on ``bus``. Nothing is sent until
371        :meth:`init` or the first channel is loaded.
372
373        :param bus: The bus the part is on.
374        :param address: The address its A5 to A0 pins select, ``0x40`` with all six low.
375        :param frequency_hz: The PWM frequency every channel shares; 50 for hobby servos.
376        :param oscillator_hz: The clock the prescaler divides, for a board driving EXTCLK.
377        :param totem_pole: Totem-pole outputs when true, open-drain when false.
378        :param inverted: Invert the output logic, for a board with no external driver.
379        :param change_on_ack: Change the outputs on each write's acknowledge, not the stop.
380        """
381        self._native = _NativePca9685(
382            bus._native,
383            address,
384            frequency_hz,
385            oscillator_hz,
386            totem_pole,
387            inverted,
388            change_on_ack,
389        )
390
391    def init(self) -> None:
392        """Program the prescale and the output wiring, wake the oscillator, and restart.
393
394        :raises PamojaError: If the bus fails.
395        """
396        self._native.init()
397
398    def set_channel(self, channel: int, setting: bytes) -> None:
399        """Load one channel, initializing the part first if :meth:`init` has not run.
400
401        :param channel: The output, 0 to 15.
402        :param setting: The four register bytes the ``pwm`` builders make.
403        :raises ValueError: For a channel past 15, or a setting that is not four bytes.
404        :raises PamojaError: If the bus fails.
405        """
406        self._native.set_channel(channel, bytes(setting))
407
408    def channel(self, channel: int) -> bytes:
409        """Read one channel's four register bytes back from the part.
410
411        The registers are read one at a time, so the read works whether or not MODE1's
412        auto-increment bit is set, and it changes nothing on the part: a program that
413        restarts can ask a running part what it holds. ``pwm.counts`` names the counts.
414
415        :param channel: The output, 0 to 15.
416        :returns: The four register bytes the channel holds.
417        :raises ValueError: For a channel past 15.
418        :raises PamojaError: If the bus fails.
419        """
420        return bytes(self._native.channel(channel))
421
422    def set_all(self, setting: bytes) -> None:
423        """Load every channel with the same four bytes in one transfer.
424
425        :param setting: The four register bytes the ``pwm`` builders make.
426        :raises PamojaError: If the bus fails.
427        """
428        self._native.set_all(bytes(setting))
429
430    def sleep(self) -> None:
431        """Stop the oscillator; the channels keep their settings.
432
433        :raises PamojaError: If the bus fails.
434        """
435        self._native.sleep()
436
437    def wake(self) -> None:
438        """Wake the oscillator, wait the 500 us it needs, and restart the channels.
439
440        :raises PamojaError: If the bus fails.
441        """
442        self._native.wake()
443
444    def software_reset(self) -> None:
445        """Send the general-call software reset to every PCA9685 on the bus.
446
447        It goes to address ``0x00``, which nothing on a simulated bus answers.
448
449        :raises PamojaError: If the bus fails, or nothing acknowledges the general call.
450        """
451        self._native.software_reset()
452
453    @property
454    def prescale(self) -> int:
455        """The prescale value the driver writes for its frequency."""
456        return self._native.prescale
457
458    @property
459    def frequency(self) -> float:
460        """The frequency the part runs at once the prescaler has rounded, in hertz."""
461        return self._native.frequency
462
463
464class _Pca9685Sim:
465    """A PCA9685 that is not there, keeping its datasheet's rules: PRE_SCALE takes a write
466    only while the part sleeps, the register pointer moves on only with auto-increment set,
467    and one write to the ALL_LED registers loads every channel."""
468
469    __slots__ = ()
470
471    def part(self, address: int = 0x40) -> I2cPart:
472        """A simulated PCA9685 as it powers up: asleep at 200 Hz with every output off.
473
474        :param address: The address it answers to.
475        :returns: A part to put on a simulated bus.
476        """
477        return I2cPart._wrap(_pca9685_sim_part(address))
478
479
480class _Pca9685:
481    """An NXP PCA9685 16-channel PWM controller, for servos, LEDs, and valves."""
482
483    __slots__ = ()
484
485    #: The address it answers at with its six address pins low.
486    DEFAULT_ADDRESS = 0x40
487    #: The part's internal oscillator frequency, in hertz.
488    INTERNAL_OSC_HZ = _INTERNAL_OSC_HZ
489    #: How many channels it drives.
490    CHANNELS = _CHANNELS
491    #: How many counts each period is divided into.
492    COUNTS = _COUNTS
493    #: How long the oscillator takes to run once woken, in microseconds.
494    OSCILLATOR_STARTUP_MICROS = 500
495    #: Mode register 1: restart, clock, auto-increment, sleep, and the addresses answered.
496    REGISTER_MODE1 = 0x00
497    #: Mode register 2: how the outputs are wired and when they change.
498    REGISTER_MODE2 = 0x01
499    #: The first of channel 0's four registers; :meth:`channel_register` gives the rest.
500    REGISTER_LED0_ON_L = 0x06
501    #: The first of the four registers that load every channel at once.
502    REGISTER_ALL_LED_ON_L = 0xFA
503    #: The prescaler, writable only while the part sleeps.
504    REGISTER_PRE_SCALE = 0xFE
505    #: MODE1's RESTART bit: set when the part slept with a channel running.
506    MODE1_RESTART = 0x80
507    #: MODE1's EXTCLK bit: the prescaler divides the EXTCLK pin.
508    MODE1_EXTCLK = 0x40
509    #: MODE1's auto-increment bit: the register pointer moves on after each byte.
510    MODE1_AUTO_INCREMENT = 0x20
511    #: MODE1's SLEEP bit: the oscillator is off and PRE_SCALE takes a write.
512    MODE1_SLEEP = 0x10
513    #: MODE1 at power-up: asleep, answering the All Call address.
514    MODE1_RESET = 0x11
515    #: MODE2 at power-up: totem-pole outputs.
516    MODE2_RESET = 0x04
517    #: PRE_SCALE at power-up: 200 Hz on the internal oscillator.
518    PRE_SCALE_RESET = 0x1E
519    #: The smallest value the part loads into PRE_SCALE.
520    PRE_SCALE_MIN = 3
521    #: A PCA9685 that is not there.
522    sim = _Pca9685Sim()
523
524    def channel_register(self, channel: int) -> int:
525        """Return the first of a channel's four consecutive registers.
526
527        :param channel: The channel, 0 to 15.
528        :returns: The register address.
529        :raises ValueError: If the channel is beyond the part.
530        """
531        return _channel_register(channel)
532
533    def prescale_for_frequency(
534        self, update_rate_hz: int, osc_hz: int = _INTERNAL_OSC_HZ
535    ) -> int:
536        """Return the prescale value that sets an update rate.
537
538        :param update_rate_hz: The PWM frequency wanted.
539        :param osc_hz: The oscillator frequency, usually the internal one.
540        :returns: The prescale register value.
541        """
542        return _prescale_for_frequency(update_rate_hz, osc_hz)
543
544    def frequency_for_prescale(
545        self, prescale: int, osc_hz: int = _INTERNAL_OSC_HZ
546    ) -> float:
547        """Return the update rate a prescale value produces.
548
549        :param prescale: The prescale register value.
550        :param osc_hz: The oscillator frequency, usually the internal one.
551        :returns: The frequency in hertz.
552        """
553        return _frequency_for_prescale(prescale, osc_hz)
554
555
556class _Pwm:
557    """The four register bytes for one PCA9685 channel.
558
559    Each call returns them in the channel's own register order, so they can be
560    written in a single bus transaction.
561    """
562
563    __slots__ = ()
564
565    def from_counts(self, on: int, off: int) -> bytes:
566        """Build a setting from explicit on and off counts.
567
568        :param on: The count at which the output goes high.
569        :param off: The count at which it goes low.
570        :returns: The four register bytes; counts are masked to 12 bits.
571        """
572        return _pwm_from_counts(on, off)
573
574    def duty(self, off: int) -> bytes:
575        """Build a setting with no phase delay: on at count 0, off at ``off``.
576
577        The datasheet rules out the same count in on and off, so 0 is the full-off
578        setting and 4096 or more the full-on one.
579
580        :param off: The count at which the output goes low, which sets the duty.
581        :returns: The four register bytes.
582        """
583        return _pwm_duty(off)
584
585    def servo(self, pulse_micros: int, update_rate_hz: int = 50) -> bytes:
586        """Build the setting that drives a hobby servo to a pulse width.
587
588        :param pulse_micros: The high-pulse width in microseconds. Typical travel
589            is about 1000 to 2000 microseconds.
590        :param update_rate_hz: The PWM frequency the controller is set to.
591        :returns: The four register bytes.
592        """
593        return _pwm_servo(pulse_micros, update_rate_hz)
594
595    def counts(self, data: bytes) -> Pwm:
596        """Read a setting back from the four register bytes a channel holds.
597
598        The inverse of the builders above, so a caller can read a channel off the bus
599        and see what it is set to rather than decoding the registers by hand.
600
601        :param data: The four channel registers.
602        :returns: The counts at which the output goes high and low, named.
603        :raises ValueError: If ``data`` is not four bytes.
604        """
605        on, off = _pwm_counts(bytes(data))
606        return Pwm(on, off)
607
608    def full_on(self) -> bytes:
609        """Return the setting that holds a channel continuously high.
610
611        :returns: The four register bytes.
612        """
613        return _pwm_full_on()
614
615    def full_off(self) -> bytes:
616        """Return the setting that holds a channel continuously low.
617
618        This is the power-on state, and its flag takes precedence over the full-on flag
619        when both are set.
620
621        :returns: The four register bytes.
622        """
623        return _pwm_full_off()
624
625
626def steps_for_degrees(degrees: float, steps_per_revolution: int) -> int:
627    """Return how many steps a rotation of ``degrees`` takes on a given motor.
628
629    :param degrees: The angle to turn through.
630    :param steps_per_revolution: The motor's steps per full revolution.
631    :returns: The step count, negative for a negative angle.
632    """
633    return _steps_for_degrees(degrees, steps_per_revolution)
634
635
636class _StepperTiming:
637    """Stepper motor timing, as ``pamoja_actuators::stepper`` starts a driver with."""
638
639    __slots__ = ()
640
641    #: The pause a driver takes after each step unless given another, in microseconds.
642    DEFAULT_STEP_MICROS = _DEFAULT_STEP_MICROS
643    #: How long a step and direction driver holds the direction before a step pulse, and
644    #: the pulse itself, in microseconds.
645    DEFAULT_PULSE_MICROS = _DEFAULT_PULSE_MICROS
646
647
648#: Stepper motor timing.
649stepper = _StepperTiming()
650
651#: An NXP PCA9685 16-channel PWM controller.
652pca9685 = _Pca9685()
653
654#: The four register bytes for one PCA9685 channel.
655pwm = _Pwm()