# 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-07-10  Serial added
# rev 0.19  2026-07-14  parameter order leds64 changed, WifiAP and BtnDebounce added

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

# Implemented:
# BtnDebounce(pin_nr=33)
#    BtnDebounce.pressed()				-->  bool
# 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_mosi=15, pin_cs=2, pin_sck=0, 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.writeLn(obj)
#    Serial.readLn()					-->  obj | None
# WifiAP(ssid='Esp32')
#    WifiAP.open()
#    WifiAP.close()

# =========================== BtnDebounce ============================

# Debounce for button to GND
class BtnDebounce:
    
    def __init__(self, pin_nr=33):
        self.btn = Pin(pin_nr, Pin.IN, Pin.PULL_UP)
        self.count = 0
        
        
    # return True if pressed after 3rd call
    def pressed(self):
        if self.btn(): self.count = 0			# if not pressed
        elif self.count < 3:
            self.count += 1
            return self.count == 3
        
# ========================= 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_mosi=15, pin_cs=2, pin_sck=0, 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


    # bytes[0] right column, bit[0] is bottom led
    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 port through UART.
    def __init__(self, id=1, baud=115200, rx=27, tx=26):
        self.uart = UART(id, baudrate=baud, rx=rx, tx=tx, stop=2, timeout=0)
        self.buff_rx = b''


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


    # Write <CR><LF> terminated string or obj
    def writeLn(self, obj):
        try:
            if not isinstance(obj, str): obj = json.dumps(obj)
            self.uart.write(f'{obj}\r\n'.encode())
        except: print('Serial.write: exception')
        

    # return string or python object when available
    def readLn(self):
        try:
            while ch := self.uart.read(1):
                if ch in b'\r\n':
                    obj, self.buff_rx = self.buff_rx, b''
                    obj = obj.decode()
                    try   : obj = json.loads(obj)
                    except: pass
                    return obj
                self.buff_rx += ch
        except: print('Serial.read: exception')
        
# ============================ WifiAP =============================

# Class to operate on-board wifi in AP mode
class WifiAP:
    
    def __init__(self, ssid='Esp32'):
        self.ssid = ssid
        self.close()
        
        
    # open Wifi AP, no password on 192.168.4.1
    def open(self):
        nic = network.WLAN(network.AP_IF)
        nic.config(ssid=self.ssid)
        nic.active(True)
        print(f'WiFi AP: {self.ssid} open on 192.168.4.1') 


    # close wifi AP
    def close(self):
        nic = network.WLAN(network.STA_IF)
        nic.active(False)
        nic = network.WLAN(network.AP_IF)
        nic.active(False)

# ============================ 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)
            
            
    # send everey second 'hello' and print received strings. rx=17, tx=17
    # $ picocom -b 115200 /dev/ttyUSB1
    def testSerial(rx=16, tx=17):
        serial = Serial(rx=rx, tx=tx)
        try:
            while True:
                for _ in range(20):
                    if obj := serial.readLn():
                        print(obj)
                    time.sleep_ms(100)
                serial.writeLn('hello')			# @ 2 Sec
        finally: serial.close()
        
        
    # tcp/ip network setup through Wifi Access Point
    def testWifiAP():
        wifi_ap = WifiAP()
        wifi_ap.open()
        try:
            while True:
                print('*', end='')
                time.sleep(1)
        finally: wifi_ap.close()


    testSerial()
    
# ===================== End =========================
