# Esp32 - Kit FKS0001 drivers for 4 digit 7-segm display, NeoPixel leds and Potentiometer
# provided by Anthony Vogelaar, Perth Scotland
# rev 0.10  2025-06-19  initial release
# rev 0.11  2025-06-20  converted to classes
# rev 0.12  2025-07-04  UsDistance added
# rev 0.13  2025-07-18  Leds64 added
# rev 0.14  2025-08-26  IrRec  added
# rev 0.15  2025-08-26  DHT11 added
# rev 0.16  2025-09-05  PIR and Buzzer added
# rev 0.17  2025-10-25  nbInput added
# rev 0.18  2026-02-07  Serial added

import	sys, select
import	time
from 	machine 	import 	Pin, I2C, ADC, SPI
from	collections	import	deque
import	neopixel	as		npx

# Implemented:
# Npx(pin=25)
# 	Npx.show(l0=OFF, l1=OFF, l2=OFF, l3=OFF, l4=OFF, l5=OFF)
# Led7(pin_scl=22, pin_sda=21, i2c_bus=0)
# 	Led7.show(value=0, dp=None, hex=False)
# Potm(pin=26)
# 	Potm.get()         					-->  float
# UsDistance(pin_trig=32, pin_echo=35)
#    UsDistance.get()					-->  float
# Leds64(pin_cs=2, pin_sck=0, pin_mosi=15, pin_miso=4)
#    Leds64.on(intensity=3)
#    Leds64.off()
#    Leds64.write(data=b'\x01\x80\x80\x80\x80\x80\x80\x02')
#    Leds64.movingText(txt=' Hello World ', speed=250)
# IrRec(pin_ir=35)
#     IrRec.getKey()					-->  int
# DHT11(pin_rd=34, pin_wr=23)
#    DHT11.getHT()						-->  float, float
# Pir(pin_nr=35)
#    Pir.get()							-->  bool
# Buzzer(pin_nr=32)
#    Buzzer.sound(active=False)
# nbInput()								-->  str | None
# Serial(id=1, baud=115200, rx=27, tx=26)
#    Serial.close()
#    Serial.write(by)
#    Serial.b_rx						-->  bytes | None

# ========================= Neo Pixel Leds ===========================

class Npx:
    # Basic colours for Neopixel LED's WS2812
    OFF    = bytes.fromhex('00 00 00')
    GREEN  = bytes.fromhex('FF 00 00')
    RED    = bytes.fromhex('00 FF 00')
    BLUE   = bytes.fromhex('00 00 FF')
    YELLOW = bytes.fromhex('FF FF 00')
    WHITE  = bytes.fromhex('FF FF FF')


    # Setup of NeoPixel LED output
    def __init__(self, pin=25):
        self.leds = npx.NeoPixel(Pin(pin), 6)		# Setup for 6 leds on pin


    # Set the colours of the 6 NeoPixel LED's
    def show(self, l0=OFF, l1=OFF, l2=OFF, l3=OFF, l4=OFF, l5=OFF):
        self.leds[0] = l0; self.leds[1] = l1; self.leds[2] = l2
        self.leds[3] = l3; self.leds[4] = l4; self.leds[5] = l5
        self.leds.write()

# ===================== 4 digit 7 segment LED ==========================

class  Led7:
    
    # Setup I2C channel for 4 digit 7-segment LED display
    def __init__(self, pin_scl=22, pin_sda=21, i2c_bus=0):
        self.i2c = I2C(i2c_bus, scl=Pin(pin_scl), sda=Pin(pin_sda), freq=400000)
    
       
    # Display value on 4 digit 7-segment LED display
    def show(self, value=0, dp=None, hex=False):
        CHARS    = b' 0123456789ABCDEF'
        PATTERNS = bytes.fromhex('00 3F 06 5B 4F 66 6D 7D 07 7F 6F 77 7C 39 5E 79 71')

        if value is None: b4=b'    '
        elif hex: b4 = bytes(f'{value:04X}', 'ascii')
        else    : b4 = bytes(f'{value:4}', 'ascii')
        dig0 = PATTERNS[CHARS.index(b4[0].to_bytes())];
        dig1 = PATTERNS[CHARS.index(b4[1].to_bytes())]
        dig2 = PATTERNS[CHARS.index(b4[2].to_bytes())]
        dig3 = PATTERNS[CHARS.index(b4[3].to_bytes())]
        if   dp == 4: dig0 += 0x80 
        elif dp == 3: dig1 += 0x80 
        elif dp == 2: dig2 += 0x80 
        elif dp == 1: dig3 += 0x80 
        self.i2c.writeto(36, b'\x31')				# 0x24
        self.i2c.writeto(52, dig0.to_bytes())		# 0x34
        self.i2c.writeto(53, dig1.to_bytes())		# 0x35
        self.i2c.writeto(54, dig2.to_bytes())		# 0x36
        self.i2c.writeto(55, dig3.to_bytes())		# 0x37

# ====================== Potmeter ADC ==========================

class Potm:
    
    def __init__(self, pin=26):
        self.potm = ADC(pin, atten=ADC.ATTN_11DB)
    

    # get potentiometer voltage 0.128 .. 3.126 Volt
    def get(self):
        return round(self.potm.read_uv()/1_000_000, 3)

# ===================== Ultra-sonic disdtance =============================

class UsDistance:
    
    def __init__(self, pin_trig=32, pin_echo=35):
        self.trig = Pin(pin_trig, Pin.OUT, value=0)
        self.echo = Pin(pin_echo, Pin.IN, Pin.PULL_UP)
        

    # return the measured gap in cm
    def get(self):
        self.trig(1)								# send impulse
        time.sleep_us(10)
        self.trig(0)
        while not self.echo(): pass					# wait until echo high
        t1 = time.ticks_us()
        while self.echo(): pass						# wait until echo low
        dt = time.ticks_diff(time.ticks_us(), t1)	# pulse width in uSec
        return round(dt / 59, 1)					# gap in cm @ 340 m/sec

# ============================= Leds64 =========================

# character table for 8x8 led panel
ctbl = dict(
    A = b'\x00\x1e\xf0\x90\x90\xfe\x1e\x00',
    B = b'\x00\x1e\xf2\x92\x9e\xfe\x82\x00',
    C = b'\x00\xc6\x82\x82\x82\xfe\x1e\x00',
    D = b'\x00\xfe\x82\x82\x9e\xfe\x82\x00',
    E = b'\x00\x82\x82\x92\x92\xfe\x1e\x00',
    F = b'\x00\x80\x80\x90\x90\xfe\x1e\x00',
    G = b'\x00\xde\x92\x92\x82\xfe\x1e\x00',
    H = b'\x00\x1e\xf0\x10\x10\xfe\x1e\x00',
    I = b'\x00\x82\x82\xfe\x9e\x82\x00\x00',
    J = b'\x00\x00\x9e\xfe\x82\x02\x06\x00',
    K = b'\x00\x02\xc2\x2c\x10\xfe\x1e\x00',
    L = b'\x00\x06\x02\x02\x02\xfe\x1e\x00',
    M = b'\x00\xfe\xc0\x60\xc0\xfe\x1e\x00',
    N = b'\x00\xfe\x04\x78\x80\xfe\x1e\x00',
    O = b'\x00\xfe\x82\x82\x82\xfe\x1e\x00',
    P = b'\x00\xf0\x90\x90\x90\xfe\x1e\x00',
    Q = b'\x00\xfa\x84\x8a\x82\xfe\x1e\x00',
    R = b'\x00\xf2\x96\x98\x90\xfe\x1e\x00',
    S = b'\x00\x00\xde\x9e\x92\xf2\x06\x00',
    T = b'\x00\x80\x80\xfe\x9e\x80\x00\x00',
    U = b'\x00\xfe\x02\x02\x02\xfe\x1e\x00',
    V = b'\x00\xe0\x10\x08\x04\xfe\x1e\x00',
    W = b'\x00\xfe\x06\x1c\x06\xfe\x1e\x00',
    X = b'\x00\x82\x5e\x30\x30\x5e\x8e\x00',
    Y = b'\x00\x80\xf8\x1c\x26\xc2\x80\x00',
    Z = b'\x00\xc2\xe2\xb2\x9a\x8e\x86\x00')
ctbl.update({
    '0': b'\x3c\x42\xa1\x91\x89\x85\x42\x3c',
    '1': b'\x00\x01\x01\xff\x81\x41\x21\x00',
    '2': b'\x00\x61\x91\x89\x85\x83\x41\x00',
    '3': b'\x00\x6e\x91\x91\x91\x81\x42\x00',
    '4': b'\x00\x04\xff\x44\x24\x14\x0c\x00',
    '5': b'\x00\x8e\x91\x91\x91\x91\xf2\x00',
    '6': b'\x00\x0e\x91\x91\x91\x51\x3e\x00',
    '7': b'\x00\xc0\xa0\x90\x88\x87\x80\x00',
    '8': b'\x00\x6e\x91\x91\x91\x91\x6e\x00',
    '9': b'\x00\x7c\x8a\x89\x89\x89\x70\x00',
    ' ': bytes(8)})


# Esp32 driver for Max7218 8x8 Led  panel
class Leds64:
    
    # miso is required for spi driver, not for Leds64 (mosi = DIN)
    def __init__(self, pin_cs=2, pin_sck=0, pin_mosi=15, pin_miso=4):
        self.cs   = Pin(pin_cs, Pin.OUT, value=1)
        self.spi  = SPI(1, 1_000_000, polarity=0, phase=0, sck=Pin(pin_sck),
                        mosi=Pin(pin_mosi), miso=Pin(pin_miso))
        
        
    def on(self, intensity=3):
        self.cs(0); self.spi.write(b'\x0C\x01'); self.cs(1)			# mode = normal
        par = bytearray(b'\x0A\x03'); par[1] = intensity
        self.cs(0); self.spi.write(par);         self.cs(1)			# intensity 3/15
        self.cs(0); self.spi.write(b'\x0B\xFF'); self.cs(1)			# scan limit
        for idx in range(1, 9):
            self.cs(0); self.spi.write(bytes([idx, 0])); self.cs(1)	# reset 8x8 ram
        

    def off(self):
        self.cs(0); self.spi.write(b'\x0C\x00'); self.cs(1)			# mode = off


    def write(self, data=b'\x01\x80\x80\x80\x80\x80\x80\x02'):
        for idx in range(8):
            word = bytes([idx+1, data[idx]])
            self.cs(0); self.spi.write(word); self.cs(1)			# mode = off

    # show txt on 8x8 led panel as left moving text
    def movingText(self, txt=' Hello World ', speed=250):
        res = bytearray(0)
        for ch in txt.upper():
            if ch in ctbl:
                res += bytes(reversed(ctbl[ch]))
            else:
                res += bytearray(8)
        for i in range(len(res) - 8):
            pattern = res[i:i+8]
            self.write(bytes(reversed(pattern)))
            time.sleep_ms(speed)

# ============================= IrRec ================================

class IrRec:

    # receive the IR dode from remote controls
    def __init__(self, pin_ir=35):
        self.count  = 0							# number of received bits
        self.data   = 0							# received bit accumulator
        self.t_last = time.ticks_us()
        self.que    = deque((), 100)
        ir_rx = Pin(pin_ir, Pin.IN, Pin.PULL_UP)
        ir_rx.irq(self._onIrq, trigger=Pin.IRQ_RISING)


    # save time of rising edge in queue
    def _onIrq(self, sender):
        self.que.append(time.ticks_us())
        
        
    # decode collected rising edge times to bits, return integer or None
    def getKey(self):
        while(self.que):
            self.t_last, t = self.que.popleft(), self.t_last
            dt = time.ticks_diff(self.t_last, t)
            if 900 < dt < 1500:							# if '0'
                self.data   = self.data << 1
                self.count += 1
            elif 2100 < dt < 2700:						# if '1'
                self.data   = (self.data << 1) + 1
                self.count += 1
            else:										# if error
                self.data  = 0
                self.count = 1
            if self.count == 33:						# if word complete
                res = self.data
                self.data  = 0
                self.count = 0
                return res								# return decoded word

# ============================ DHT11 ===============================

# Driver for DHT11 humidity and temperature sensor
class DHT11:
    
    # One wire DHT11 I/O is connected to pin_rd and pin_wr
    def __init__(self, pin_rd=34, pin_wr=23):
        self.wr = Pin(pin_wr, Pin.OPEN_DRAIN, value=1)		# writing to DHT11
        self.rd = Pin(pin_rd, Pin.IN, Pin.PULL_UP)			# reading from DHT11
        self.rd.irq(self._onIrq, trigger=Pin.IRQ_RISING)
        self.fifo	= deque([], 50)							# collect time of up to 50 positive rd edges
        
    
    # return humidity and temperature. Allow for a 2 second init after power-up
    def getHT(self):
        
        # decode fifo content into humidity and time
        def decode():
            res = 0											# bit accumulator
            t0 = self.fifo.popleft()						# time firs positive edge
            while self.fifo:
                t0, t1 = self.fifo.popleft(), t0			# delta time
                res = res << 1
                if time.ticks_diff(t0, t1) > 100: res += 1	# shift in LSB
            humidity    = (res >> 32 & 0xFF) + ((res >> 24 & 0xFF) / 10)	# res[1] + res[2] / 10
            temperature = (res >> 16 & 0xFF) + ((res >>  8 & 0xFF) / 10)	# res[3] + res[1] / 10
            return humidity, temperature
            
            
        self.wr(0)											# low for 19 mSec = start
        while self.fifo: self.fifo.popleft()				# empty fifo
        time.sleep_ms(19)
        self.wr(1)											# end of start
        time.sleep_ms(6)									# wait to collect incoming data
        return decode()										# return humidity and temperature as evaluated
        
        
    # save time of rising edge in fifo
    def _onIrq(self, sender):
        self.fifo.append(time.ticks_us())        

# ============================= Pir ===============================

class Pir:
    
    def __init__(self, pin_nr=35):
        self.pir = Pin(pin_nr, Pin.IN, Pin.PULL_UP)


    # return True if PIR active
    def get(self):
        return bool(self.pir())

# ============================ Buzzer =============================

class Buzzer:
    
    def __init__(self, pin_nr=32):
        self.bzr = Pin(pin_nr, Pin.OUT, value=0)


    # active = True for sound
    def sound(self, active=False):
        self.bzr(active)

# ============================ nbKeyb ===========================

# Non blocking keyboard / USB input of lines. Returns line or None.
# As Thonny is line based it works on this platform (a line after b'\r\n') 
# Working on character base when connected to a terminal
#    $ picocom -b 115200 --imap lfcrlf /dev/ttyACM0 [/dev/ttyUSB0]
def nbInput(statics=dict(buff=bytearray(0), poll=None)):
    if not statics['poll']:
        statics['poll'] = select.poll() 		            	# Set up serial input polling object
        statics['poll'].register(sys.stdin, select.POLLIN)		# Register polling object
    while statics['poll'].poll(0) and (ch := sys.stdin.buffer.read(1)):
        if ch in b'\r\n':
            res = statics['buff']
            statics['buff'] = bytearray(0)
            if res: return res.decode()
        else:
            statics['buff'] += ch

# =========================== Serial ================================

class Serial:

    # Open serial UART port.
    def __init__(self, id=1, baud=115200, rx=27, tx=26):
        self.b_rx
        self.uart  = UART(id, baudrate=baud, rx=rx, tx=tx, stop=2)
        self.uart.irq(self.onRx, trigger=UART.IRQ_RXIDLE)


    # close the serial port
    def close(self):
        if self.uart:
            self.uart.deinit()
            self.uart = None


    # Write bytes or bytearray
    def write(self, by):
        self.uart.write(by)


    # return read data when available
    def read(self):
        sts = machine.disable_irq()
        res, self.b_rx = self.b_rx, None
        machine.enable_irq(sts)
        return res
    
    
    # Irq handler on data receive gap, using timer 0
    #    save received bytes in self.b_rx
    def onRx(self, sender):
        if b_rx := self.uart.read():			# return b'\x00 on no data
            self.b_rx = b_rx

# ============================ Tests ==============================

if __name__ == '__main__':
    
    # Testing Neopixel LED's on FKS0001
    def testNpx():
        npx = Npx()
        while True:
            npx.show(Npx.GREEN);   time.sleep(1)
            npx.show(l1=Npx.RED);  time.sleep(1)
            npx.show(l2=Npx.BLUE); time.sleep(1)


    # Testing 4 digit 7-segment LED's on FKS0001
    def testLed7():
        led7 = Led7()
        while True:
            led7.show(1000, 3);        time.sleep(2)		# 10.00
            led7.show(1000, hex=True); time.sleep(2)
            led7.show(None);           time.sleep(2)


    # testing potentiometer adc
    def testPotm():
        potm = Potm()
        while True:
            print(potm.get(), 'Volt')
            time.sleep(1)


    # Print ultra-sonic distance to shell at 0.5 Sec
    def testUS():
        dist = UsDistance()
        while True:
            gap = dist.get()
            print(gap)
            time.sleep(0.5)


    # test 8 x 8 leds panel
    def testLeds64():
        leds64 = Leds64()
        leds64.on()
        leds64.movingText()
        leds64.movingText(' 123 ')


    # test the IR receiver
    def testIR():
        ir_rec = IrRec()
        print('IR receiver ready ...')
        while True:
            time.sleep_ms(10)
            if res := ir_rec.getKey():
                print(hex(res))


    # Test DHT11 humidity and temperature sensor
    def testHuTemp():
        dht11 = DHT11()
        h, t = dht11.getHT()
        print('humidity    =', h)
        print('temperature =', t)


    # Test Pir and Buzzer
    def testPir():
        pir = Pir()
        buzzer = Buzzer()
        try:
            while True:
                buzzer.sound(pir.get())
        finally:
            buzzer.sound(False)
            
            
    # test non blocking keyboard input
    def testNbKeyb():
        while True:
            print('*', end='')								# print @ 1 Sec
            for _ in range(10):
                if s := nbInput(): print('\nkeyb:', s)		# test input @ 100 mSec
                time.sleep_ms(100)
            
            
    # NOT TESTED. send everey second b'hello' through serial UART port and print received bytes
    def testSerial():
        
        uart = Serial(onRx)									# rx=27, tx=26
        try:
            while True:
                for _ in range(10):
                    if b_rx := uart.read():				# @ 100 mSec
                        print(b_rx)
                    time.sleep_ms(100)
                serial.write(b'hello\n')			# @ 1 Sec
        finally: uart.close()
        
        
    testSerial()
    
# ===================== End =========================
