ir_rx/__init__.py

   1 # ir_rx __init__.py Decoder for IR remote control using synchronous code
   2 # IR_RX abstract base class for IR receivers.
   3
   4 # Author: Peter Hinch
   5 # Copyright Peter Hinch 2020 Released under the MIT license
   6
   7 from sys import platform
   8 from machine import Timer, Pin
   9 from array import array
  10 from utime import ticks_us
  11
  12 # Save RAM
  13 # from micropython import alloc_emergency_exception_buf
  14 # alloc_emergency_exception_buf(100)
  15
  16
  17 # On 1st edge start a block timer. While the timer is running, record the time
  18 # of each edge. When the timer times out decode the data. Duration must exceed
  19 # the worst case block transmission time, but be less than the interval between
  20 # a block start and a repeat code start (~108ms depending on protocol)
  21
  22 class IR_RX():
  23     # Result/error codes
  24     # Repeat button code
  25     REPEAT = -1
  26     # Error codes
  27     BADSTART = -2
  28     BADBLOCK = -3
  29     BADREP = -4
  30     OVERRUN = -5
  31     BADDATA = -6
  32     BADADDR = -7
  33
  34     def __init__(self, pin, nedges, tblock, callback, *args):  # Optional args for callback
  35         self._pin = pin
  36         self._nedges = nedges
  37         self._tblock = tblock
  38         self.callback = callback
  39         self.args = args
  40         self._errf = lambda _ : None
  41         self.verbose = False
  42
  43         self._times = array('i',  (0 for _ in range(nedges + 1)))  # +1 for overrun
  44         if platform == 'esp32' or platform == 'esp32_LoBo':  # ESP32 Doesn't support hard IRQ
  45             pin.irq(handler = self._cb_pin, trigger = (Pin.IRQ_FALLING | Pin.IRQ_RISING))
  46         else:
  47             pin.irq(handler = self._cb_pin, trigger = (Pin.IRQ_FALLING | Pin.IRQ_RISING), hard = True)
  48         self.edge = 0
  49         self.tim = Timer(-1)  # Sofware timer
  50         self.cb = self.decode
  51
  52     # Pin interrupt. Save time of each edge for later decode.
  53     def _cb_pin(self, line):
  54         t = ticks_us()
  55         # On overrun ignore pulses until software timer times out
  56         if self.edge <= self._nedges:  # Allow 1 extra pulse to record overrun
  57             if not self.edge:  # First edge received
  58                 self.tim.init(period=self._tblock , mode=Timer.ONE_SHOT, callback=self.cb)
  59             self._times[self.edge] = t
  60             self.edge += 1
  61
  62     def do_callback(self, cmd, addr, ext, thresh=0):
  63         self.edge = 0
  64         if cmd >= thresh:
  65             self.callback(cmd, addr, ext, *self.args)
  66         else:
  67             self._errf(cmd)
  68
  69     def error_function(self, func):
  70         self._errf = func
  71
  72     def close(self):
  73         self._pin.irq(handler = None)
  74         self.tim.deinit()