ir_rx.py

   1 # ir_rx.py Decoder for IR remote control using synchronous code
   2 # Supports RC-5 RC-6 mode 0 and NEC protocols.
   3 # For a remote using NEC see https://www.adafruit.com/products/389
   4
   5 # Author: Peter Hinch
   6 # Copyright Peter Hinch 2020 Released under the MIT license
   7
   8 from sys import platform
   9 from micropython import const
  10 from machine import Timer
  11 from array import array
  12 from utime import ticks_us, ticks_diff
  13
  14 if platform == 'pyboard':
  15     from pyb import Pin, ExtInt
  16 else:
  17     from machine import Pin
  18
  19 ESP32 = platform == 'esp32' or platform == 'esp32_LoBo'
  20
  21 # Save RAM
  22 # from micropython import alloc_emergency_exception_buf
  23 # alloc_emergency_exception_buf(100)
  24
  25 # Result codes (accessible to application)
  26 # Repeat button code
  27 REPEAT = -1
  28 # Error codes
  29 BADSTART = -2
  30 BADBLOCK = -3
  31 BADREP = -4
  32 OVERRUN = -5
  33 BADDATA = -6
  34 BADADDR = -7
  35
  36
  37 # On 1st edge start a block timer. While the timer is running, record the time
  38 # of each edge. When the timer times out decode the data. Duration must exceed
  39 # the worst case block transmission time, but be less than the interval between
  40 # a block start and a repeat code start (~108ms depending on protocol)
  41
  42 class IR_RX():
  43     verbose = False
  44     def __init__(self, pin, nedges, tblock, callback, *args):  # Optional args for callback
  45         self._nedges = nedges
  46         self._tblock = tblock
  47         self.callback = callback
  48         self.args = args
  49
  50         self._times = array('i',  (0 for _ in range(nedges + 1)))  # +1 for overrun
  51         if platform == 'pyboard':
  52             ei = ExtInt(pin, ExtInt.IRQ_RISING_FALLING, Pin.PULL_NONE, self._cb_pin)
  53         elif ESP32:
  54             ei = pin.irq(handler = self._cb_pin, trigger = (Pin.IRQ_FALLING | Pin.IRQ_RISING))
  55         else:
  56             ei = pin.irq(handler = self._cb_pin, trigger = (Pin.IRQ_FALLING | Pin.IRQ_RISING), hard = True)
  57         self._eint = ei  # Keep reference??
  58         self.edge = 0
  59         self.tim = Timer(-1)  # Sofware timer
  60         self.cb = self.decode
  61
  62
  63     # Pin interrupt. Save time of each edge for later decode.
  64     def _cb_pin(self, line):
  65         t = ticks_us()
  66         # On overrun ignore pulses until software timer times out
  67         if self.edge <= self._nedges:  # Allow 1 extra pulse to record overrun
  68             if not self.edge:  # First edge received
  69                 self.tim.init(period=self._tblock , mode=Timer.ONE_SHOT, callback=self.cb)
  70             self._times[self.edge] = t
  71             self.edge += 1
  72
  73 class NEC_IR(IR_RX):
  74     def __init__(self, pin, callback, extended=True, *args):
  75         # Block lasts <= 80ms and has 68 edges
  76         tblock = 80 if extended else 73  # Allow for some tx tolerance (?)
  77         super().__init__(pin, 68, tblock, callback, *args)
  78         self._extended = extended
  79         self._addr = 0
  80
  81     def decode(self, _):
  82         try:
  83             if self.edge > 68:
  84                 raise RuntimeError(OVERRUN)
  85             width = ticks_diff(self._times[1], self._times[0])
  86             if width < 4000:  # 9ms leading mark for all valid data
  87                 raise RuntimeError(BADSTART)
  88             width = ticks_diff(self._times[2], self._times[1])
  89             if width > 3000:  # 4.5ms space for normal data
  90                 if self.edge < 68:  # Haven't received the correct number of edges
  91                     raise RuntimeError(BADBLOCK)
  92                 # Time spaces only (marks are always 562.5µs)
  93                 # Space is 1.6875ms (1) or 562.5µs (0)
  94                 # Skip last bit which is always 1
  95                 val = 0
  96                 for edge in range(3, 68 - 2, 2):
  97                     val >>= 1
  98                     if ticks_diff(self._times[edge + 1], self._times[edge]) > 1120:
  99                         val |= 0x80000000
 100             elif width > 1700: # 2.5ms space for a repeat code. Should have exactly 4 edges.
 101                 raise RuntimeError(REPEAT if self.edge == 4 else BADREP)  # Treat REPEAT as error.
 102             else:
 103                 raise RuntimeError(BADSTART)
 104             addr = val & 0xff  # 8 bit addr
 105             cmd = (val >> 16) & 0xff
 106             if cmd != (val >> 24) ^ 0xff:
 107                 raise RuntimeError(BADDATA)
 108             if addr != ((val >> 8) ^ 0xff) & 0xff:  # 8 bit addr doesn't match check
 109                 if not self._extended:
 110                     raise RuntimeError(BADADDR)
 111                 addr |= val & 0xff00  # pass assumed 16 bit address to callback
 112             self._addr = addr
 113         except RuntimeError as e:
 114             cmd = e.args[0]
 115             addr = self._addr if cmd == REPEAT else 0  # REPEAT uses last address
 116         self.edge = 0  # Set up for new data burst and run user callback
 117         self.callback(cmd, addr, 0, *self.args)
 118
 119
 120 class SONY_IR(IR_RX):
 121     def __init__(self, pin, callback, bits=20, *args):
 122         # 20 bit block has 42 edges and lasts <= 39ms nominal. Add 4ms to time
 123         # for tolerances except in 20 bit case where timing is tight with a
 124         # repeat period of 45ms.
 125         t = int(3 + bits * 1.8) + (1 if bits == 20 else 4)
 126         super().__init__(pin, 2 + bits * 2, t, callback, *args)
 127         self._addr = 0
 128         self._bits = bits
 129
 130     def decode(self, _):
 131         try:
 132             nedges = self.edge  # No. of edges detected
 133             self.verbose and print('nedges', nedges)
 134             if nedges > 42:
 135                 raise RuntimeError(OVERRUN)
 136             bits = (nedges - 2) // 2
 137             if nedges not in (26, 32, 42) or bits > self._bits:
 138                 raise RuntimeError(BADBLOCK)
 139             self.verbose and print('SIRC {}bit'.format(bits))
 140             width = ticks_diff(self._times[1], self._times[0])
 141             if not 1800 < width < 3000:  # 2.4ms leading mark for all valid data
 142                 raise RuntimeError(BADSTART)
 143             width = ticks_diff(self._times[2], self._times[1])
 144             if not 350 < width < 1000:  # 600μs space
 145                 raise RuntimeError(BADSTART)
 146
 147             val = 0  # Data received, LSB 1st
 148             x = 2
 149             bit = 1
 150             while x < nedges - 2:
 151                 if ticks_diff(self._times[x + 1], self._times[x]) > 900:
 152                     val |= bit
 153                 bit <<= 1
 154                 x += 2
 155
 156             cmd = val & 0x7f  # 7 bit command
 157             val >>= 7
 158             if nedges < 42:
 159                 addr = val & 0xff  # 5 or 8 bit addr
 160                 val = 0
 161             else:
 162                 addr = val & 0x1f  # 5 bit addr
 163                 val >>= 5  # 8 bit extended
 164         except RuntimeError as e:
 165             cmd = e.args[0]
 166             addr = 0
 167             val = 0
 168         self.edge = 0  # Set up for new data burst and run user callback
 169         self.callback(cmd, addr, val, *self.args)
 170
 171 class RC5_IR(IR_RX):
 172     def __init__(self, pin, callback, *args):
 173         # Block lasts <= 30ms and has <= 28 edges
 174         super().__init__(pin, 28, 30, callback, *args)
 175
 176     def decode(self, _):
 177         try:
 178             nedges = self.edge  # No. of edges detected
 179             if not 14 <= nedges <= 28:
 180                 raise RuntimeError(OVERRUN if nedges > 28 else BADSTART)
 181             # Regenerate bitstream
 182             bits = 0
 183             bit = 1
 184             for x in range(1, nedges):
 185                 width = ticks_diff(self._times[x], self._times[x - 1])
 186                 if not 500 < width < 2000:
 187                     raise RuntimeError(BADBLOCK)
 188                 for _ in range(1 if width < 1334 else 2):
 189                     bits <<= 1
 190                     bits |= bit
 191                 bit ^= 1
 192             self.verbose and print(bin(bits))  # Matches inverted scope waveform
 193             # Decode Manchester code
 194             x = 30
 195             while not bits >> x:
 196                 x -= 1
 197             m0 = 1 << x  # Mask MS two bits (always 01)
 198             m1 = m0 << 1
 199             v = 0  # 14 bit bitstream
 200             for _ in range(14):
 201                 v <<= 1
 202                 b0 = (bits & m0) > 0
 203                 b1 = (bits & m1) > 0
 204                 if b0 == b1:
 205                     raise RuntimeError(BADBLOCK)
 206                 v |= b0
 207                 m0 >>= 2
 208                 m1 >>= 2
 209             # Split into fields (val, addr, ctrl)
 210             val = (v & 0x3f) | (0x40 if ((v >> 12) & 1) else 0)
 211             addr = (v >> 6) & 0x1f
 212             ctrl = (v >> 11) & 1
 213
 214         except RuntimeError as e:
 215             val, addr, ctrl = e.args[0], 0, 0
 216         self.edge = 0  # Set up for new data burst and run user callback
 217         self.callback(val, addr, ctrl, *self.args)
 218
 219 class RC6_M0(IR_RX):
 220     # Even on Pyboard D the 444μs nominal pulses can be recorded as up to 705μs
 221     # Scope shows 360-520 μs (-84μs +76μs relative to nominal)
 222     # Header nominal 2666, 889, 444, 889, 444, 444, 444, 444 carrier ON at end
 223     hdr = ((1800, 4000), (593, 1333), (222, 750), (593, 1333), (222, 750), (222, 750), (222, 750), (222, 750))
 224     def __init__(self, pin, callback, *args):
 225         # Block lasts 23ms nominal and has <=44 edges
 226         super().__init__(pin, 44, 30, callback, *args)
 227
 228     def decode(self, _):
 229         try:
 230             nedges = self.edge  # No. of edges detected
 231             if not 22 <= nedges <= 44:
 232                 raise RuntimeError(OVERRUN if nedges > 28 else BADSTART)
 233             for x, lims in enumerate(self.hdr):
 234                 width = ticks_diff(self._times[x + 1], self._times[x])
 235                 if not (lims[0] < width < lims[1]):
 236                     self.verbose and print('Bad start', x, width, lims)
 237                     raise RuntimeError(BADSTART)
 238             x += 1
 239             width = ticks_diff(self._times[x + 1], self._times[x])
 240             # 2nd bit of last 0 is 444μs (0) or 1333μs (1)
 241             if not 222 < width < 1555:
 242                 self.verbose and print('Bad block 1 Width', width, 'x', x)
 243                 raise RuntimeError(BADBLOCK)
 244             short = width < 889
 245             v = int(not short)
 246             bit = v
 247             bits = 1  # Bits decoded
 248             x += 1 + int(short)
 249             width = ticks_diff(self._times[x + 1], self._times[x])
 250             if not 222 < width < 1555:
 251                 self.verbose and print('Bad block 2 Width', width, 'x', x)
 252                 raise RuntimeError(BADBLOCK)
 253             short = width < 1111
 254             if not short:
 255                 bit ^= 1
 256             x += 1 + int(short)  # If it's short, we know width of next
 257             v <<= 1
 258             v |= bit  # MSB of result
 259             bits += 1
 260             # Decode bitstream
 261             while bits < 17:
 262                 # -1 convert count to index, -1 because we look ahead
 263                 if x > nedges - 2:
 264                     raise RuntimeError(BADBLOCK)
 265                 # width is 444/889 nominal
 266                 width = ticks_diff(self._times[x + 1], self._times[x])
 267                 if not 222 < width < 1111:
 268                     self.verbose and print('Bad block 3 Width', width, 'x', x)
 269                     raise RuntimeError(BADBLOCK)
 270                 short = width < 666
 271                 if not short:
 272                     bit ^= 1
 273                 v <<= 1
 274                 v |= bit
 275                 bits += 1
 276                 x += 1 + int(short)
 277
 278             if self.verbose:
 279                  ss = '20-bit format {:020b} x={} nedges={} bits={}'
 280                  print(ss.format(v, x, nedges, bits))
 281
 282             val = v & 0xff
 283             addr = (v >> 8) & 0xff
 284             ctrl = (v >> 16) & 1
 285         except RuntimeError as e:
 286             val, addr, ctrl = e.args[0], 0, 0
 287         self.edge = 0  # Set up for new data burst and run user callback
 288         self.callback(val, addr, ctrl, *self.args)