import time
from machine import Pin

from nec import NEC_8  # NEC remote, 8 bit addresses



def callback(data, addr, ctrl):
    if data < 0:  # NEC protocol sends repeat codes.
        print('Repeat code.')
    else:
        print('Data {} Addr {:04x}'.format(data, addr))

ir = NEC_8(Pin(13, Pin.IN), callback)
while True:
    time.sleep_ms(500)
    

main.py

diagram.json

# acquire.py Acquire a pulse train from an IR remote
# Supports NEC protocol.
# For a remote using NEC see https://www.adafruit.com/products/389

# Author: Peter Hinch
# Copyright Peter Hinch 2020 Released under the MIT license

from machine import Pin, freq
from sys import platform

from utime import sleep_ms, ticks_us, ticks_diff
from __init__ import IR_RX


class IR_GET(IR_RX):
    def __init__(self, pin, nedges=100, twait=100, display=True):
        self.display = display
        super().__init__(pin, nedges, twait, lambda *_ : None)
        self.data = None

    def decode(self, _):
        def near(v, target):
            return target * 0.8 < v < target * 1.2
        lb = self.edge - 1  # Possible length of burst
        if lb < 3:
            return  # Noise
        burst = []
        for x in range(lb):
            dt = ticks_diff(self._times[x + 1], self._times[x])
            if x > 0 and dt > 10000:  # Reached gap between repeats
                break
            burst.append(dt)
        lb = len(burst)  # Actual length
        # Duration of pulse train 24892 for RC-5 22205 for RC-6
        duration = ticks_diff(self._times[lb - 1], self._times[0])

        if self.display:
            for x, e in enumerate(burst):
                print('{:03d} {:5d}'.format(x, e))
            print()
            # Attempt to determine protocol
            ok = False  # Protocol not yet found
            if near(burst[0], 9000) and lb == 67:
                print('NEC')
                ok = True

            if not ok and near(burst[0], 2400) and near(burst[1], 600):  # Maybe Sony
                try:
                    nbits = {25:12, 31:15, 41:20}[lb]
                except KeyError:
                    pass
                else:
                    ok = True
                    print('Sony {}bit'.format(nbits))

            if not ok and near(burst[0], 889):  # Maybe RC-5
                if near(duration, 24892) and near(max(burst), 1778):
                    print('Philps RC-5')
                    ok = True

            if not ok and near(burst[0], 2666) and near(burst[1], 889):  # RC-6?
                if near(duration, 22205) and near(burst[1], 889) and near(burst[2], 444):
                    print('Philips RC-6 mode 0')
                    ok = True

            if not ok and near(burst[0], 2000) and near(burst[1], 1000):
                if near(duration, 19000):
                    print('Microsoft MCE edition protocol.')
                    # Constant duration, variable burst length, presumably bi-phase
                    print('Protocol start {} {} Burst length {} duration {}'.format(burst[0], burst[1], lb, duration))
                    ok = True

            if not ok and near(burst[0], 4500) and near(burst[1], 4500) and lb == 67:  # Samsung
                print('Samsung')
                ok = True

            if not ok and near(burst[0], 3500) and near(burst[1], 1680):  # Panasonic?
                print('Unsupported protocol. Panasonic?')
                ok = True

            if not ok:
                print('Unknown protocol start {} {} Burst length {} duration {}'.format(burst[0], burst[1], lb, duration))

            print()
        self.data = burst
        # Set up for new data burst. Run null callback
        self.do_callback(0, 0, 0)

    def acquire(self):
        while self.data is None:
            sleep_ms(5)
        self.close()
        return self.data

def test():
    # Define pin according to platform
    if platform == 'pyboard':
        pin = Pin('X3', Pin.IN)
    elif platform == 'esp8266':
        freq(160000000)
        pin = Pin(13, Pin.IN)
    elif platform == 'esp32' or platform == 'esp32_LoBo':
        pin = Pin(23, Pin.IN)
    elif platform == 'rp2':
        pin = Pin(16, Pin.IN)
    irg = IR_GET(pin)
    print('Waiting for IR data...')
    return irg.acquire()


acquire.py

# ir_rx __init__.py Decoder for IR remote control using synchronous code
# IR_RX abstract base class for IR receivers.

# Author: Peter Hinch
# Copyright Peter Hinch 2020-2021 Released under the MIT license

from machine import Timer, Pin
from array import array
from utime import ticks_us

# Save RAM
# from micropython import alloc_emergency_exception_buf
# alloc_emergency_exception_buf(100)


# On 1st edge start a block timer. While the timer is running, record the time
# of each edge. When the timer times out decode the data. Duration must exceed
# the worst case block transmission time, but be less than the interval between
# a block start and a repeat code start (~108ms depending on protocol)

class IR_RX():
    # Result/error codes
    # Repeat button code
    REPEAT = -1
    # Error codes
    BADSTART = -2
    BADBLOCK = -3
    BADREP = -4
    OVERRUN = -5
    BADDATA = -6
    BADADDR = -7

    def __init__(self, pin, nedges, tblock, callback, *args):  # Optional args for callback
        self._pin = pin
        self._nedges = nedges
        self._tblock = tblock
        self.callback = callback
        self.args = args
        self._errf = lambda _ : None
        self.verbose = False

        self._times = array('i',  (0 for _ in range(nedges + 1)))  # +1 for overrun
        pin.irq(handler = self._cb_pin, trigger = (Pin.IRQ_FALLING | Pin.IRQ_RISING))
        self.edge = 0
        self.tim = Timer(-1)  # Sofware timer
        self.cb = self.decode

    # Pin interrupt. Save time of each edge for later decode.
    def _cb_pin(self, line):
        t = ticks_us()
        # On overrun ignore pulses until software timer times out
        if self.edge <= self._nedges:  # Allow 1 extra pulse to record overrun
            if not self.edge:  # First edge received
                self.tim.init(period=self._tblock , mode=Timer.ONE_SHOT, callback=self.cb)
            self._times[self.edge] = t
            self.edge += 1

    def do_callback(self, cmd, addr, ext, thresh=0):
        self.edge = 0
        if cmd >= thresh:
            self.callback(cmd, addr, ext, *self.args)
        else:
            self._errf(cmd)

    def error_function(self, func):
        self._errf = func

    def close(self):
        self._pin.irq(handler = None)
        self.tim.deinit()


__init__.py

# mce.py Decoder for IR remote control using synchronous code
# Supports Microsoft MCE edition remote protocol.

# Author: Peter Hinch
# Copyright Peter Hinch 2020 Released under the MIT license

# WARNING: This is experimental and subject to change.

from utime import ticks_us, ticks_diff
from __init__ import IR_RX

class MCE(IR_RX):
    init_cs = 4  # http://www.hifi-remote.com/johnsfine/DecodeIR.html#OrtekMCE says 3
    def __init__(self, pin, callback, *args):
        # Block lasts ~19ms and has <= 34 edges
        super().__init__(pin, 34, 25, callback, *args)

    def decode(self, _):
        def check(v):
            if self.init_cs == -1:
                return True
            csum = v >> 12
            cs = self.init_cs
            for _ in range(12):
                if v & 1:
                    cs += 1
                v >>= 1
            return cs == csum

        try:
            t0 = ticks_diff(self._times[1], self._times[0])  # 2000μs mark
            t1 = ticks_diff(self._times[2], self._times[1])  # 1000μs space
            if not ((1800 < t0 < 2200) and (800 < t1 < 1200)):
                raise RuntimeError(self.BADSTART)
            nedges = self.edge  # No. of edges detected
            if not 14 <= nedges <= 34:
                raise RuntimeError(self.OVERRUN if nedges > 28 else self.BADSTART)
            # Manchester decode
            mask = 1
            bit = 1
            v = 0
            x = 2
            for _ in range(16):
                # -1 convert count to index, -1 because we look ahead
                if x > nedges - 2:
                    raise RuntimeError(self.BADBLOCK)
                # width is 500/1000 nominal
                width = ticks_diff(self._times[x + 1], self._times[x])
                if not 250 < width < 1350:
                    self.verbose and print('Bad block 3 Width', width, 'x', x)
                    raise RuntimeError(self.BADBLOCK)
                short = int(width < 750)
                bit ^= short ^ 1
                v |= mask if bit else 0
                mask <<= 1
                x += 1 + short

            self.verbose and print(bin(v))
            if not check(v):
                raise RuntimeError(self.BADDATA)
            val = (v >> 6) & 0x3f
            addr = v & 0xf  # Constant for all buttons on my remote
            ctrl = (v >> 4) & 3

        except RuntimeError as e:
            val, addr, ctrl = e.args[0], 0, 0
        # Set up for new data burst and run user callback/error function
        self.do_callback(val, addr, ctrl)


mce.py

# nec.py Decoder for IR remote control using synchronous code
# Supports NEC and Samsung protocols.
# With thanks to J.E. Tannenbaum for information re Samsung protocol

# For a remote using NEC see https://www.adafruit.com/products/389

# Author: Peter Hinch
# Copyright Peter Hinch 2020-2022 Released under the MIT license

from utime import ticks_us, ticks_diff
from __init__ import IR_RX

class NEC_ABC(IR_RX):
    def __init__(self, pin, extended, samsung, callback, *args):
        # Block lasts <= 80ms (extended mode) and has 68 edges
        super().__init__(pin, 68, 80, callback, *args)
        self._extended = extended
        self._addr = 0
        self._leader = 2500 if samsung else 4000  # 4.5ms for Samsung else 9ms

    def decode(self, _):
        try:
            if self.edge > 68:
                raise RuntimeError(self.OVERRUN)
            width = ticks_diff(self._times[1], self._times[0])
            if width < self._leader:  # 9ms leading mark for all valid data
                raise RuntimeError(self.BADSTART)
            width = ticks_diff(self._times[2], self._times[1])
            if width > 3000:  # 4.5ms space for normal data
                if self.edge < 68:  # Haven't received the correct number of edges
                    raise RuntimeError(self.BADBLOCK)
                # Time spaces only (marks are always 562.5µs)
                # Space is 1.6875ms (1) or 562.5µs (0)
                # Skip last bit which is always 1
                val = 0
                for edge in range(3, 68 - 2, 2):
                    val >>= 1
                    if ticks_diff(self._times[edge + 1], self._times[edge]) > 1120:
                        val |= 0x80000000
            elif width > 1700: # 2.5ms space for a repeat code. Should have exactly 4 edges.
                raise RuntimeError(self.REPEAT if self.edge == 4 else self.BADREP)  # Treat REPEAT as error.
            else:
                raise RuntimeError(self.BADSTART)
            addr = val & 0xff  # 8 bit addr
            cmd = (val >> 16) & 0xff
            if cmd != (val >> 24) ^ 0xff:
                raise RuntimeError(self.BADDATA)
            if addr != ((val >> 8) ^ 0xff) & 0xff:  # 8 bit addr doesn't match check
                if not self._extended:
                    raise RuntimeError(self.BADADDR)
                addr |= val & 0xff00  # pass assumed 16 bit address to callback
            self._addr = addr
        except RuntimeError as e:
            cmd = e.args[0]
            addr = self._addr if cmd == self.REPEAT else 0  # REPEAT uses last address
        # Set up for new data burst and run user callback
        self.do_callback(cmd, addr, 0, self.REPEAT)

class NEC_8(NEC_ABC):
    def __init__(self, pin, callback, *args):
        super().__init__(pin, False, False, callback, *args)

class NEC_16(NEC_ABC):
    def __init__(self, pin, callback, *args):
        super().__init__(pin, True, False, callback, *args)

class SAMSUNG(NEC_ABC):
    def __init__(self, pin, callback, *args):
        super().__init__(pin, True, True, callback, *args)


nec.py

# philips.py Decoder for IR remote control using synchronous code
# Supports Philips RC-5 RC-6 mode 0 protocols.

# Author: Peter Hinch
# Copyright Peter Hinch 2020 Released under the MIT license

from utime import ticks_us, ticks_diff
from __init__ import IR_RX

class RC5_IR(IR_RX):
    def __init__(self, pin, callback, *args):
        # Block lasts <= 30ms and has <= 28 edges
        super().__init__(pin, 28, 30, callback, *args)

    def decode(self, _):
        try:
            nedges = self.edge  # No. of edges detected
            if not 14 <= nedges <= 28:
                raise RuntimeError(self.OVERRUN if nedges > 28 else self.BADSTART)
            # Regenerate bitstream
            bits = 1
            bit = 1
            v = 1  # 14 bit bitstream, MSB always 1
            x = 0
            while bits < 14:
                # -1 convert count to index, -1 because we look ahead
                if x > nedges - 2:
                    print('Bad block 1 edges', nedges, 'x', x)
                    raise RuntimeError(self.BADBLOCK)
                # width is 889/1778 nominal
                width = ticks_diff(self._times[x + 1], self._times[x])
                if not 500 < width < 2100:
                    self.verbose and print('Bad block 3 Width', width, 'x', x)
                    raise RuntimeError(self.BADBLOCK)
                short = width < 1334
                if not short:
                    bit ^= 1
                v <<= 1
                v |= bit
                bits += 1
                x += 1 + int(short)
            self.verbose and print(bin(v))
            # Split into fields (val, addr, ctrl)
            val = (v & 0x3f) | (0 if ((v >> 12) & 1) else 0x40)  # Correct the polarity of S2
            addr = (v >> 6) & 0x1f
            ctrl = (v >> 11) & 1

        except RuntimeError as e:
            val, addr, ctrl = e.args[0], 0, 0
        # Set up for new data burst and run user callback
        self.do_callback(val, addr, ctrl)


class RC6_M0(IR_RX):
    # Even on Pyboard D the 444μs nominal pulses can be recorded as up to 705μs
    # Scope shows 360-520 μs (-84μs +76μs relative to nominal)
    # Header nominal 2666, 889, 444, 889, 444, 444, 444, 444 carrier ON at end
    hdr = ((1800, 4000), (593, 1333), (222, 750), (593, 1333), (222, 750), (222, 750), (222, 750), (222, 750))
    def __init__(self, pin, callback, *args):
        # Block lasts 23ms nominal and has <=44 edges
        super().__init__(pin, 44, 30, callback, *args)

    def decode(self, _):
        try:
            nedges = self.edge  # No. of edges detected
            if not 22 <= nedges <= 44:
                raise RuntimeError(self.OVERRUN if nedges > 28 else self.BADSTART)
            for x, lims in enumerate(self.hdr):
                width = ticks_diff(self._times[x + 1], self._times[x])
                if not (lims[0] < width < lims[1]):
                    self.verbose and print('Bad start', x, width, lims)
                    raise RuntimeError(self.BADSTART)
            x += 1
            width = ticks_diff(self._times[x + 1], self._times[x])
            # 2nd bit of last 0 is 444μs (0) or 1333μs (1)
            if not 222 < width < 1555:
                self.verbose and print('Bad block 1 Width', width, 'x', x)
                raise RuntimeError(self.BADBLOCK)
            short = width < 889
            v = int(not short)
            bit = v
            bits = 1  # Bits decoded
            x += 1 + int(short)
            width = ticks_diff(self._times[x + 1], self._times[x])
            if not 222 < width < 1555:
                self.verbose and print('Bad block 2 Width', width, 'x', x)
                raise RuntimeError(self.BADBLOCK)
            short = width < 1111
            if not short:
                bit ^= 1
            x += 1 + int(short)  # If it's short, we know width of next
            v <<= 1
            v |= bit  # MSB of result
            bits += 1
            # Decode bitstream
            while bits < 17:
                # -1 convert count to index, -1 because we look ahead
                if x > nedges - 2:
                    raise RuntimeError(self.BADBLOCK)
                # width is 444/889 nominal
                width = ticks_diff(self._times[x + 1], self._times[x])
                if not 222 < width < 1111:
                    self.verbose and print('Bad block 3 Width', width, 'x', x)
                    raise RuntimeError(self.BADBLOCK)
                short = width < 666
                if not short:
                    bit ^= 1
                v <<= 1
                v |= bit
                bits += 1
                x += 1 + int(short)

            if self.verbose:
                 ss = '20-bit format {:020b} x={} nedges={} bits={}'
                 print(ss.format(v, x, nedges, bits))

            val = v & 0xff
            addr = (v >> 8) & 0xff
            ctrl = (v >> 16) & 1
        except RuntimeError as e:
            val, addr, ctrl = e.args[0], 0, 0
        # Set up for new data burst and run user callback
        self.do_callback(val, addr, ctrl)


philips.py

# print_error.py Error print for IR receiver

# Author: Peter Hinch
# Copyright Peter Hinch 2020 Released under the MIT license

from __init__ import IR_RX

_errors = {IR_RX.BADSTART : 'Invalid start pulse',
           IR_RX.BADBLOCK : 'Error: bad block',
           IR_RX.BADREP : 'Error: repeat',
           IR_RX.OVERRUN : 'Error: overrun',
           IR_RX.BADDATA : 'Error: invalid data',
           IR_RX.BADADDR : 'Error: invalid address'}

def print_error(data):
    if data in _errors:
        print(_errors[data])
    else:
        print('Unknown error code:', data)


print_error.py

# sony.py Decoder for IR remote control using synchronous code
# Sony SIRC protocol.

# Author: Peter Hinch
# Copyright Peter Hinch 2020 Released under the MIT license

from utime import ticks_us, ticks_diff

from __init__ import IR_RX

class SONY_ABC(IR_RX):  # Abstract base class
    def __init__(self, pin, bits, callback, *args):
        # 20 bit block has 42 edges and lasts <= 39ms nominal. Add 4ms to time
        # for tolerances except in 20 bit case where timing is tight with a
        # repeat period of 45ms.
        t = int(3 + bits * 1.8) + (1 if bits == 20 else 4)
        super().__init__(pin, 2 + bits * 2, t, callback, *args)
        self._addr = 0
        self._bits = 20

    def decode(self, _):
        try:
            nedges = self.edge  # No. of edges detected
            self.verbose and print('nedges', nedges)
            if nedges > 42:
                raise RuntimeError(self.OVERRUN)
            bits = (nedges - 2) // 2
            if nedges not in (26, 32, 42) or bits > self._bits:
                raise RuntimeError(self.BADBLOCK)
            self.verbose and print('SIRC {}bit'.format(bits))
            width = ticks_diff(self._times[1], self._times[0])
            if not 1800 < width < 3000:  # 2.4ms leading mark for all valid data
                raise RuntimeError(self.BADSTART)
            width = ticks_diff(self._times[2], self._times[1])
            if not 350 < width < 1000:  # 600μs space
                raise RuntimeError(self.BADSTART)

            val = 0  # Data received, LSB 1st
            x = 2
            bit = 1
            while x <= nedges - 2:
                if ticks_diff(self._times[x + 1], self._times[x]) > 900:
                    val |= bit
                bit <<= 1
                x += 2
            cmd = val & 0x7f  # 7 bit command
            val >>= 7
            if nedges < 42:
                addr = val & 0xff  # 5 or 8 bit addr
                val = 0
            else:
                addr = val & 0x1f  # 5 bit addr
                val >>= 5  # 8 bit extended
        except RuntimeError as e:
            cmd = e.args[0]
            addr = 0
            val = 0
        self.do_callback(cmd, addr, val)

class SONY_12(SONY_ABC):
    def __init__(self, pin, callback, *args):
        super().__init__(pin, 12, callback, *args)

class SONY_15(SONY_ABC):
    def __init__(self, pin, callback, *args):
        super().__init__(pin, 15, callback, *args)

class SONY_20(SONY_ABC):
    def __init__(self, pin, callback, *args):
        super().__init__(pin, 20, callback, *args)



sony.py

# test.py Test program for IR remote control decoder
# Supports Pyboard, ESP32 and ESP8266

# Author: Peter Hinch
# Copyright Peter Hinch 2020-2022 Released under the MIT license

# Run this to characterise a remote.

from sys import platform
import time
import gc
from machine import Pin, freq
from print_error import print_error  # Optional print of error codes

# Import all implemented classes
from nec import NEC_8, NEC_16, SAMSUNG
from sony import SONY_12, SONY_15, SONY_20
from philips import RC5_IR, RC6_M0
from mce import MCE

# Define pin according to platform
if platform == "pyboard":
    p = Pin("X3", Pin.IN)
elif platform == "esp8266":
    freq(160000000)
    p = Pin(13, Pin.IN)
elif platform == "esp32" or platform == "esp32_LoBo":
    p = Pin(23, Pin.IN)
elif platform == "rp2":
    p = Pin(16, Pin.IN)

# User callback
def cb(data, addr, ctrl):
    if data < 0:  # NEC protocol sends repeat codes.
        print("Repeat code.")
    else:
        print(f"Data 0x{data:02x} Addr 0x{addr:04x} Ctrl 0x{ctrl:02x}")


def test(proto=0):
    classes = (NEC_8, NEC_16, SONY_12, SONY_15, SONY_20, RC5_IR, RC6_M0, MCE, SAMSUNG)
    ir = classes[proto](p, cb)  # Instantiate receiver
    ir.error_function(print_error)  # Show debug information
    # ir.verbose = True
    # A real application would do something here...
    try:
        while True:
            print("running")
            time.sleep(5)
            gc.collect()
    except KeyboardInterrupt:
        ir.close()


# **** DISPLAY GREETING ****
s = """Test for IR receiver. Run:
from ir_rx.test import test
test() for NEC 8 bit protocol,
test(1) for NEC 16 bit,
test(2) for Sony SIRC 12 bit,
test(3) for Sony SIRC 15 bit,
test(4) for Sony SIRC 20 bit,
test(5) for Philips RC-5 protocol,
test(6) for RC6 mode 0.
test(7) for Microsoft Vista MCE.
test(8) for Samsung.

Hit ctrl-c to stop, then ctrl-d to soft reset."""

print(s)
