ir_tx/rp2_rmt.py

   1 # rp2_rmt.py A RMT-like class for the RP2.
   2
   3 # Released under the MIT License (MIT). See LICENSE.
   4
   5 # Copyright (c) 2021 Peter Hinch
   6
   7 from machine import Pin, PWM
   8 import rp2
   9
  10 @rp2.asm_pio(set_init=rp2.PIO.OUT_LOW, autopull=True, pull_thresh=32)
  11 def pulsetrain():
  12     wrap_target()
  13     out(x, 32)  # No of 1MHz ticks. Block if FIFO MT at end.
  14     irq(rel(0))
  15     set(pins, 1)  # Set pin high
  16     label('loop')
  17     jmp(x_dec,'loop')
  18     irq(rel(0))
  19     set(pins, 0)  # Set pin low
  20     out(y, 32)  # Low time.
  21     label('loop_lo')
  22     jmp(y_dec,'loop_lo')
  23     wrap()
  24
  25 @rp2.asm_pio(autopull=True, pull_thresh=32)
  26 def irqtrain():
  27     wrap_target()
  28     out(x, 32)  # No of 1MHz ticks. Block if FIFO MT at end.
  29     irq(rel(0))
  30     label('loop')
  31     jmp(x_dec,'loop')
  32     wrap()
  33
  34 class DummyPWM:
  35     def duty_u16(self, _):
  36         pass
  37
  38 class RP2_RMT:
  39
  40     def __init__(self, pin_pulse=None, carrier=None, sm_no=0, sm_freq=1_000_000):
  41         if carrier is None:
  42             self.pwm = DummyPWM()
  43             self.duty = (0, 0)
  44         else:
  45             pin_car, freq, duty = carrier
  46             self.pwm = PWM(pin_car)  # Set up PWM with carrier off.
  47             self.pwm.freq(freq)
  48             self.pwm.duty_u16(0)
  49             self.duty = (int(0xffff * duty // 100), 0)
  50         if pin_pulse is None:
  51             self.sm = rp2.StateMachine(sm_no, irqtrain, freq=sm_freq)
  52         else:
  53             self.sm = rp2.StateMachine(sm_no, pulsetrain, freq=sm_freq, set_base=pin_pulse)
  54         self.apt = 0  # Array index
  55         self.arr = None  # Array
  56         self.ict = None  # Current IRQ count
  57         self.icm = 0  # End IRQ count
  58         self.reps = 0  # 0 == forever n == no. of reps
  59         rp2.PIO(0).irq(self._cb)
  60
  61     # IRQ callback. Because of FIFO IRQ's keep arriving after STOP.
  62     def _cb(self, pio):
  63         self.pwm.duty_u16(self.duty[self.ict & 1])
  64         self.ict += 1
  65         if d := self.arr[self.apt]:  # If data available feed FIFO
  66             self.sm.put(d)
  67             self.apt += 1
  68         else:
  69             if r := self.reps != 1:  # All done if reps == 1
  70                 if r:  # 0 == run forever
  71                     self.reps -= 1
  72                 self.sm.put(self.arr[0])
  73                 self.apt = 1  # Set pointer and count to state
  74                 self.ict = 1  # after 1st IRQ
  75
  76     # Arg is an array of times in μs terminated by 0.
  77     def send(self, ar, reps=1, check=True):
  78         self.sm.active(0)
  79         self.reps = reps
  80         ar[-1] = 0  # Ensure at least one STOP
  81         for x, d in enumerate(ar):  # Find 1st STOP
  82             if d == 0:
  83                 break
  84         if check:
  85             # Discard any trailing mark which would leave carrier on.
  86             if (x & 1):
  87                 x -= 1
  88                 ar[x] = 0
  89         self.icm = x  # index of 1st STOP
  90         mv = memoryview(ar)
  91         n = min(x, 4)  # Fill FIFO if there are enough data points.
  92         self.sm.put(mv[0 : n])
  93         self.arr = ar  # Initial conditions for ISR
  94         self.apt = n  # Point to next data value
  95         self.ict = 0  # IRQ count
  96         self.sm.active(1)
  97
  98     def busy(self):
  99         if self.ict is None:
 100             return False  # Just instantiated
 101         return self.ict < self.icm
 102
 103     def cancel(self):
 104         self.reps = 1