raspberrypi: Implement rotaryio, improve StateMachine

[jepler]

Fixes #4226.

Now rotaryio is implemented in this port. For the attached example, either connect a rotary encoder at GP2/3, or connect GP2/3 and GP20/21 to use a synthesized signal.

Click to show Python code

import board
import rotaryio
import digitalio
import time

class QuadratureGenerator:
    def __init__(self, pin_a, pin_b):
        self.a = digitalio.DigitalInOut(pin_a)
        self.b = digitalio.DigitalInOut(pin_b)
        self.a.switch_to_output()
        self.b.switch_to_output()
        self.state = 0

    def __call__(self):
        direction = -1 if (time.monotonic_ns() // 1000000000 % 3) else 1
        state = self.state = (self.state + direction) % 4

        if state == 0:
            self.a.value = 0
            self.b.value = 0
        elif state == 1:
            self.a.value = 1
            self.b.value = 0
        elif state == 2:
            self.a.value = 1
            self.b.value = 1
        else:
            self.a.value = 0
            self.b.value = 1

gen = QuadratureGenerator(board.GP20, board.GP21)


r = rotaryio.IncrementalEncoder(board.GP2, board.GP3)
op = None
while True:
    gen()
    p = r.position
    if op != p:
        print(p)
        op = p

The following functionality was added to StateMachine:

• The ability to query the depth of the rx fifo
• The ability to set pull up/down/bus-keep
• A function to check if pins are sequential

Also a few other things "crept in":

• A missing deinit call in PDMIn
• StateMachine.h was not a self-sufficient header, it needed the declaration of mcu_pin_obj_t

With the Python API improvements, it's also possible to implement a form of IncrementalEncoder in pure Python:

Click to show Python code

# This example is adapted in part from micropython:
# https://github.com/micropython/micropython/pull/6894/files

import adafruit_pioasm
import board
import rp2pio
import array

class IncrementalEncoder:
    _state_look_up_table = array.array("b", [
            #Direction = 1
            0, # 00 to 00
           -1, # 00 to 01
           +1, # 00 to 10
           +2, # 00 to 11

           +1, # 01 to 00
            0, # 01 to 01
           +2, # 01 to 10
           -1, # 01 to 11

           -1, # 10 to 00
           +2, # 10 to 01
            0, # 10 to 10
           +1, # 10 to 11

           +2, # 11 to 00
           +1, # 11 to 01
           -1, # 11 to 10
            0, # 11 to 11

            #Direction = 0
            0, # 00 to 00
           -1, # 00 to 01
           +1, # 00 to 10
           -2, # 00 to 11

           +1, # 01 to 00
            0, # 01 to 01
           -2, # 01 to 10
           -1, # 01 to 11

           -1, # 10 to 00
           -2, # 10 to 01
            0, # 10 to 10
           +1, # 10 to 11

           -2, # 11 to 00
           +1, # 11 to 01
           -1, # 11 to 10
            0, # 11 to 11
        ])

    _sm_code = adafruit_pioasm.assemble("""
    again:
        in pins, 2
        mov x, isr
        jmp x!=y, push_data
        mov isr, null
        jmp again
    push_data:
        push
        ; irq wait 0 rel
        mov y, x
    """)
    _sm_init=adafruit_pioasm.assemble("set y 31")

    def __init__(self, pin_a, pin_b):
        if not rp2pio.pins_are_sequential([pin_a, pin_b]):
            raise ValueError("Pins must be sequential")

        self._sm = rp2pio.StateMachine(
            self._sm_code,
            160_000,
            init=self._sm_init,
            first_in_pin=pin_a,
            in_pin_count=2,
            pull_in_pin_up=0b11,
            in_shift_right=False
        )

        self._counter = 0
        self._direction = 0
        self._lut_index = 0 
        self._buffer = bytearray(1)

    def _update_state_machine(self, state):
        lut_index = self._lut_index | (state & 3)
        lut = self._state_look_up_table[lut_index]
        self._counter += lut
        if lut:
            self._direction = 1 if (lut > 0) else 0
        self._lut_index = ((lut_index << 2) & 0b1100) | (self._direction << 4)

    def deinit(self):
        self._sm.deinit()

    @property
    def value(self):
        while self._sm.in_waiting:
            self._sm.readinto(self._buffer)
            self._update_state_machine(self._buffer[0])
        return self._counter



encoder = IncrementalEncoder(board.GP2, board.GP3)

old_value = None
while True:
    value = encoder.value
    if old_value != value:
        print("Encoder:", value)
        old_value = value

[tannewt]

Thanks! I think you need a make translate. Code looks good otherwise!

[jepler]

@tannewt done -- if you get a chance to check again, that'd be dandy

[dhalbert]

I tested with a volume-control program I use daily on a Trinket M0. It works just fine!

Did you diagnose the "dead spot" problem from #3967? I don't see that issue immediately.

[jepler]

No, I used the same quadrature decoder as the samd port.

[dhalbert]

One style thing. Thanks for doing this!

[tannewt]

This looks good to me! Thank you! I think @DavePutz will want to integrate this into the PulseIn work too.

[tannewt]

For next time, I'd put this in ports/raspberrypi/common-hal/rp2pio/StateMachine.h because it's a port-specific API that is unrelated to the Python API. It's a bit weird since this module is itself port-specific.

Fixed