add all files

2024-05-22 12:32:30 +02:00
parent 3b7e9acd32
commit 646033244f
23 changed files with 1747 additions and 0 deletions
@@ -0,0 +1,158 @@
+import network
+import urequests as requests
+import ujson as json
+import time
+from machine import Pin, SPI, I2C
+from mfrc522 import MFRC522
+from ssd1306 import SSD1306_I2C
+
+# Global variables
+DOOR_ID = 1
+WLAN_SSID = '[Your SSID]'
+WLAN_PASS = '[Your password]'
+SERVER_IP = '[Your server IP]'
+SERVER_PORT = 5000
+
+# Initialize RFID reader
+reader = MFRC522(spi_id=0, sck=6, miso=4, mosi=7, cs=5, rst=22)
+
+# Initialize I2C for the OLED display
+i2c = I2C(id=0, scl=Pin(1), sda=Pin(0), freq=200000)
+oled = None
+
+# Initialize greenLED
+greenled = Pin(16, Pin.OUT)
+greenled.on()
+time.sleep(0.5)
+greenled.off()
+# Initialize redLED
+redled = Pin(21, Pin.OUT)
+redled.on()
+time.sleep(0.5)
+redled.off()
+
+def init_oled():
+    global oled
+    try:
+        oled = SSD1306_I2C(128, 64, i2c)
+        oled.fill(0)
+        oled.text('Initializing...', 0, 0)
+        oled.show()
+    except Exception as e:
+        print("display error:", e)
+        #init_oled()
+def display_message(message, ip_address):
+    try:
+        oled.fill(0)
+        oled.text(f'Door ID: {DOOR_ID}', 0, 0)  # Display Door ID at the top
+        oled.text("___________________", 0, 3)
+
+        lines = message.split('\n')
+        for i, line in enumerate(lines):
+            oled.text(line, 0, 20 + i * 10)  # Adjust the y position for each line
+        oled.text("__________________", 0, 47)
+        oled.text(ip_address, 0, 57)  # Display IP address at the bottom
+        oled.show()
+    except Exception as e:
+        greenled.off()
+        redled.off()
+        print("display error:", e)
+        init_oled()
+        
+# Connect to WiFi
+def connect_wifi(ssid, password):
+    wlan = network.WLAN(network.STA_IF)
+    wlan.active(True)
+    wlan.connect(ssid, password)
+    while not wlan.isconnected():
+        time.sleep(0.5)
+        print("Connecting to WiFi...")
+    ip_address = wlan.ifconfig()[0]
+    print("Connected to WiFi:", ip_address)
+    display_message('WiFi Connected', ip_address)
+
+    # Test connection to the server
+    
+    try:
+        response = requests.get(f"http://{SERVER_IP}:{SERVER_PORT}/")
+        if response.status_code == 200:
+            print("Server connection successful")
+            display_message('Server Connected', ip_address)
+        else:
+            print("Server connection failed")
+            display_message('Server Fail', ip_address)
+            time.sleep(5)
+            while response.status_code != 200 :
+                wlan.connect(ssid, password)
+                response = requests.get(f"http://{SERVER_IP}:{SERVER_PORT}/")
+                display_message('Reconnecting ...', ip_address)
+                time.sleep(1)
+    except Exception as e:
+        print("Server connection error:", e)
+        display_message(f'Server Error \n {e}', ip_address)
+        time.sleep(5)
+#         while response.status_code != 200 :
+#             wlan.connect(ssid, password)
+#             try :
+#                 response = requests.get(f"http://{SERVER_IP}:{SERVER_PORT}/")
+#                 display_message('Reconnecting ...', ip_address)
+#                 time.sleep(1)
+#             except:
+#                 pass
+# Function to send RFID UID to the server
+def send_rfid_to_server(rfid_uid):
+    url = f"http://{SERVER_IP}:{SERVER_PORT}/access"
+    headers = {'Content-Type': 'application/json'}
+    data = {
+        'rfid_uid': rfid_uid,
+        'door_id': DOOR_ID
+    }
+    response = requests.post(url, headers=headers, data=json.dumps(data))
+    return response.json()
+
+# Main loop to scan RFID tags
+def main():
+    # Retry mechanism for OLED initialization
+    for _ in range(3):
+        try:
+            init_oled()
+            break
+        except Exception as e:
+            print("OLED init error:", e)
+            time.sleep(1)
+    
+    connect_wifi(WLAN_SSID, WLAN_PASS)
+    ip_address = network.WLAN(network.STA_IF).ifconfig()[0]
+    display_message('Scan your tag', ip_address)
+
+    while True:
+        
+        (status, tag_type) = reader.request(reader.REQIDL)
+        if status == reader.OK:
+            (status, uid) = reader.SelectTagSN()
+            if status == reader.OK:
+                rfid_uid_decimal = ''.join([str(i) for i in uid])
+                print("RFID UID:", rfid_uid_decimal)
+                display_message('Checking...', ip_address)
+                
+                response = send_rfid_to_server(rfid_uid_decimal)
+                
+                if response.get('access_granted'):
+                    user_upn = response.get('upn')
+                    print("Access Granted:", user_upn)
+                    display_message(f'Access Granted\n{user_upn}', ip_address)
+                    # Turn on the LED to indicate door open
+                    greenled.on()
+                    # Add code here to open the door (e.g., trigger a relay)
+                else:
+                    print("Access Denied")
+                    display_message('Access Denied', ip_address)
+                    redled.on()
+
+                time.sleep(2)  # Delay to avoid rapid repeated reads
+                greenled.off()
+                redled.off()  # Turn off the LED
+                display_message('Scan your tag', ip_address)
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,380 @@
+from machine import Pin, SPI
+from os import uname
+ 
+ 
+class MFRC522:
+ 
+    DEBUG = False
+    OK = 0
+    NOTAGERR = 1
+    ERR = 2
+ 
+    REQIDL = 0x26
+    REQALL = 0x52
+    AUTHENT1A = 0x60
+    AUTHENT1B = 0x61
+  
+    PICC_ANTICOLL1 = 0x93
+    PICC_ANTICOLL2 = 0x95
+    PICC_ANTICOLL3 = 0x97
+  
+ 
+    def __init__(self, sck, mosi, miso, rst, cs,baudrate=1000000,spi_id=0):
+ 
+        self.sck = Pin(sck, Pin.OUT)
+        self.mosi = Pin(mosi, Pin.OUT)
+        self.miso = Pin(miso)
+        self.rst = Pin(rst, Pin.OUT)
+        self.cs = Pin(cs, Pin.OUT)
+ 
+        self.rst.value(0)
+        self.cs.value(1)
+        
+        board = uname()[0]
+ 
+        if board == 'WiPy' or board == 'LoPy' or board == 'FiPy':
+            self.spi = SPI(0)
+            self.spi.init(SPI.MASTER, baudrate=1000000, pins=(self.sck, self.mosi, self.miso))
+        elif (board == 'esp8266') or (board == 'esp32'):
+            self.spi = SPI(baudrate=100000, polarity=0, phase=0, sck=self.sck, mosi=self.mosi, miso=self.miso)
+            self.spi.init()
+        elif board == 'rp2':
+            self.spi = SPI(spi_id,baudrate=baudrate,sck=self.sck, mosi= self.mosi, miso= self.miso)
+        else:
+            raise RuntimeError("Unsupported platform")
+ 
+        self.rst.value(1)
+        self.init()
+ 
+    def _wreg(self, reg, val):
+ 
+        self.cs.value(0)
+        self.spi.write(b'%c' % int(0xff & ((reg << 1) & 0x7e)))
+        self.spi.write(b'%c' % int(0xff & val))
+        self.cs.value(1)
+ 
+    def _rreg(self, reg):
+ 
+        self.cs.value(0)
+        self.spi.write(b'%c' % int(0xff & (((reg << 1) & 0x7e) | 0x80)))
+        val = self.spi.read(1)
+        self.cs.value(1)
+ 
+        return val[0]
+ 
+    def _sflags(self, reg, mask):
+        self._wreg(reg, self._rreg(reg) | mask)
+ 
+    def _cflags(self, reg, mask):
+        self._wreg(reg, self._rreg(reg) & (~mask))
+ 
+    def _tocard(self, cmd, send):
+ 
+        recv = []
+        bits = irq_en = wait_irq = n = 0
+        stat = self.ERR
+ 
+        if cmd == 0x0E:
+            irq_en = 0x12
+            wait_irq = 0x10
+        elif cmd == 0x0C:
+            irq_en = 0x77
+            wait_irq = 0x30
+ 
+        self._wreg(0x02, irq_en | 0x80)
+        self._cflags(0x04, 0x80)
+        self._sflags(0x0A, 0x80)
+        self._wreg(0x01, 0x00)
+ 
+        for c in send:
+            self._wreg(0x09, c)
+        self._wreg(0x01, cmd)
+ 
+        if cmd == 0x0C:
+            self._sflags(0x0D, 0x80)
+ 
+        i = 2000
+        while True:
+            n = self._rreg(0x04)
+            i -= 1
+            if ~((i != 0) and ~(n & 0x01) and ~(n & wait_irq)):
+                break
+ 
+        self._cflags(0x0D, 0x80)
+ 
+        if i:
+            if (self._rreg(0x06) & 0x1B) == 0x00:
+                stat = self.OK
+ 
+                if n & irq_en & 0x01:
+                    stat = self.NOTAGERR
+                elif cmd == 0x0C:
+                    n = self._rreg(0x0A)
+                    lbits = self._rreg(0x0C) & 0x07
+                    if lbits != 0:
+                        bits = (n - 1) * 8 + lbits
+                    else:
+                        bits = n * 8
+ 
+                    if n == 0:
+                        n = 1
+                    elif n > 16:
+                        n = 16
+ 
+                    for _ in range(n):
+                        recv.append(self._rreg(0x09))
+            else:
+                stat = self.ERR
+ 
+        return stat, recv, bits
+ 
+    def _crc(self, data):
+ 
+        self._cflags(0x05, 0x04)
+        self._sflags(0x0A, 0x80)
+ 
+        for c in data:
+            self._wreg(0x09, c)
+ 
+        self._wreg(0x01, 0x03)
+ 
+        i = 0xFF
+        while True:
+            n = self._rreg(0x05)
+            i -= 1
+            if not ((i != 0) and not (n & 0x04)):
+                break
+ 
+        return [self._rreg(0x22), self._rreg(0x21)]
+ 
+    def init(self):
+ 
+        self.reset()
+        self._wreg(0x2A, 0x8D)
+        self._wreg(0x2B, 0x3E)
+        self._wreg(0x2D, 30)
+        self._wreg(0x2C, 0)
+        self._wreg(0x15, 0x40)
+        self._wreg(0x11, 0x3D)
+        self.antenna_on()
+ 
+    def reset(self):
+        self._wreg(0x01, 0x0F)
+ 
+    def antenna_on(self, on=True):
+ 
+        if on and ~(self._rreg(0x14) & 0x03):
+            self._sflags(0x14, 0x03)
+        else:
+            self._cflags(0x14, 0x03)
+ 
+    def request(self, mode):
+ 
+        self._wreg(0x0D, 0x07)
+        (stat, recv, bits) = self._tocard(0x0C, [mode])
+ 
+        if (stat != self.OK) | (bits != 0x10):
+            stat = self.ERR
+ 
+        return stat, bits
+  
+    def anticoll(self,anticolN):
+ 
+        ser_chk = 0
+        ser = [anticolN, 0x20]
+ 
+        self._wreg(0x0D, 0x00)
+        (stat, recv, bits) = self._tocard(0x0C, ser)
+ 
+        if stat == self.OK:
+            if len(recv) == 5:
+                for i in range(4):
+                    ser_chk = ser_chk ^ recv[i]
+                if ser_chk != recv[4]:
+                    stat = self.ERR
+            else:
+                stat = self.ERR
+ 
+        return stat, recv
+ 
+    
+    def PcdSelect(self, serNum,anticolN):
+        backData = []
+        buf = []
+        buf.append(anticolN)
+        buf.append(0x70)
+        #i = 0
+        ###xorsum=0;
+        for i in serNum:
+            buf.append(i)
+        #while i<5:
+        #    buf.append(serNum[i])
+        #    i = i + 1
+        pOut = self._crc(buf)
+        buf.append(pOut[0])
+        buf.append(pOut[1])
+        (status, backData, backLen) = self._tocard( 0x0C, buf)
+        if (status == self.OK) and (backLen == 0x18):
+            return  1
+        else:
+            return 0
+    
+    
+    def SelectTag(self, uid):
+        byte5 = 0
+        
+        #(status,puid)= self.anticoll(self.PICC_ANTICOLL1)
+        #print("uid",uid,"puid",puid)
+        for i in uid:
+            byte5 = byte5 ^ i
+        puid = uid + [byte5]
+        
+        if self.PcdSelect(puid,self.PICC_ANTICOLL1) == 0:
+            return (self.ERR,[])
+        return (self.OK , uid)
+        
+    def tohexstring(self,v):
+        s="["
+        for i in v:
+            if i != v[0]:
+                s = s+ ", "
+            s=s+ "0x{:02X}".format(i)
+        s= s+ "]"
+        return s
+        
+  
+            
+    
+    def SelectTagSN(self):
+        valid_uid=[]
+        (status,uid)= self.anticoll(self.PICC_ANTICOLL1)
+        #print("Select Tag 1:",self.tohexstring(uid))
+        if status != self.OK:
+            return  (self.ERR,[])
+        
+        if self.DEBUG:   print("anticol(1) {}".format(uid))
+        if self.PcdSelect(uid,self.PICC_ANTICOLL1) == 0:
+            return (self.ERR,[])
+        if self.DEBUG:   print("pcdSelect(1) {}".format(uid))
+        
+        #check if first byte is 0x88
+        if uid[0] == 0x88 :
+            #ok we have another type of card
+            valid_uid.extend(uid[1:4])
+            (status,uid)=self.anticoll(self.PICC_ANTICOLL2)
+            #print("Select Tag 2:",self.tohexstring(uid))
+            if status != self.OK:
+                return (self.ERR,[])
+            if self.DEBUG: print("Anticol(2) {}".format(uid))
+            rtn =  self.PcdSelect(uid,self.PICC_ANTICOLL2)
+            if self.DEBUG: print("pcdSelect(2) return={} uid={}".format(rtn,uid))
+            if rtn == 0:
+                return (self.ERR,[])
+            if self.DEBUG: print("PcdSelect2() {}".format(uid))
+            #now check again if uid[0] is 0x88
+            if uid[0] == 0x88 :
+                valid_uid.extend(uid[1:4])
+                (status , uid) = self.anticoll(self.PICC_ANTICOLL3)
+                #print("Select Tag 3:",self.tohexstring(uid))
+                if status != self.OK:
+                    return (self.ERR,[])
+                if self.DEBUG: print("Anticol(3) {}".format(uid))
+                if self.MFRC522_PcdSelect(uid,self.PICC_ANTICOLL3) == 0:
+                    return (self.ERR,[])
+                if self.DEBUG: print("PcdSelect(3) {}".format(uid))
+        valid_uid.extend(uid[0:5])
+        # if we are here than the uid is ok
+        # let's remove the last BYTE whic is the XOR sum
+        
+        return (self.OK , valid_uid[:len(valid_uid)-1])
+        #return (self.OK , valid_uid)
+    
+    
+   
+       
+    
+ 
+    def auth(self, mode, addr, sect, ser):
+        return self._tocard(0x0E, [mode, addr] + sect + ser[:4])[0]
+    
+    def authKeys(self,uid,addr,keyA=None, keyB=None):
+        status = self.ERR
+        if keyA is not None:
+            status = self.auth(self.AUTHENT1A, addr, keyA, uid)
+        elif keyB is not None:
+            status = self.auth(self.AUTHENT1B, addr, keyB, uid)
+        return status
+       
+ 
+    def stop_crypto1(self):
+        self._cflags(0x08, 0x08)
+ 
+    def read(self, addr):
+ 
+        data = [0x30, addr]
+        data += self._crc(data)
+        (stat, recv, _) = self._tocard(0x0C, data)
+        return stat, recv
+ 
+    def write(self, addr, data):
+ 
+        buf = [0xA0, addr]
+        buf += self._crc(buf)
+        (stat, recv, bits) = self._tocard(0x0C, buf)
+ 
+        if not (stat == self.OK) or not (bits == 4) or not ((recv[0] & 0x0F) == 0x0A):
+            stat = self.ERR
+        else:
+            buf = []
+            for i in range(16):
+                buf.append(data[i])
+            buf += self._crc(buf)
+            (stat, recv, bits) = self._tocard(0x0C, buf)
+            if not (stat == self.OK) or not (bits == 4) or not ((recv[0] & 0x0F) == 0x0A):
+                stat = self.ERR
+        return stat
+ 
+ 
+    def writeSectorBlock(self,uid, sector, block, data, keyA=None, keyB = None):
+        absoluteBlock =  sector * 4 + (block % 4)
+        if absoluteBlock > 63 :
+            return self.ERR
+        if len(data) != 16:
+            return self.ERR
+        if self.authKeys(uid,absoluteBlock,keyA,keyB) != self.ERR :
+            return self.write(absoluteBlock, data)
+        return self.ERR
+ 
+    def readSectorBlock(self,uid ,sector, block, keyA=None, keyB = None):
+        absoluteBlock =  sector * 4 + (block % 4)
+        if absoluteBlock > 63 :
+            return self.ERR, None
+        if self.authKeys(uid,absoluteBlock,keyA,keyB) != self.ERR :
+            return self.read(absoluteBlock)
+        return self.ERR, None
+ 
+    def MFRC522_DumpClassic1K(self,uid, Start=0, End=64, keyA=None, keyB=None):
+        for absoluteBlock in range(Start,End):
+            status = self.authKeys(uid,absoluteBlock,keyA,keyB)
+            # Check if authenticated
+            print("{:02d} S{:02d} B{:1d}: ".format(absoluteBlock, absoluteBlock//4 , absoluteBlock % 4),end="")
+            if status == self.OK:                    
+                status, block = self.read(absoluteBlock)
+                if status == self.ERR:
+                    break
+                else:
+                    for value in block:
+                        print("{:02X} ".format(value),end="")
+                    print("  ",end="")
+                    for value in block:
+                        if (value > 0x20) and (value < 0x7f):
+                            print(chr(value),end="")
+                        else:
+                            print('.',end="")
+                    print("")
+            else:
+                break
+        if status == self.ERR:
+            print("Authentication error")
+            return self.ERR
+        return self.OK
@@ -0,0 +1,164 @@
+# MicroPython SSD1306 OLED driver, I2C and SPI interfaces
+
+from micropython import const
+import framebuf
+
+
+# register definitions
+SET_CONTRAST = const(0x81)
+SET_ENTIRE_ON = const(0xA4)
+SET_NORM_INV = const(0xA6)
+SET_DISP = const(0xAE)
+SET_MEM_ADDR = const(0x20)
+SET_COL_ADDR = const(0x21)
+SET_PAGE_ADDR = const(0x22)
+SET_DISP_START_LINE = const(0x40)
+SET_SEG_REMAP = const(0xA0)
+SET_MUX_RATIO = const(0xA8)
+SET_IREF_SELECT = const(0xAD)
+SET_COM_OUT_DIR = const(0xC0)
+SET_DISP_OFFSET = const(0xD3)
+SET_COM_PIN_CFG = const(0xDA)
+SET_DISP_CLK_DIV = const(0xD5)
+SET_PRECHARGE = const(0xD9)
+SET_VCOM_DESEL = const(0xDB)
+SET_CHARGE_PUMP = const(0x8D)
+
+
+# Subclassing FrameBuffer provides support for graphics primitives
+# http://docs.micropython.org/en/latest/pyboard/library/framebuf.html
+class SSD1306(framebuf.FrameBuffer):
+    def __init__(self, width, height, external_vcc):
+        self.width = width
+        self.height = height
+        self.external_vcc = external_vcc
+        self.pages = self.height // 8
+        self.buffer = bytearray(self.pages * self.width)
+        super().__init__(self.buffer, self.width, self.height, framebuf.MONO_VLSB)
+        self.init_display()
+
+    def init_display(self):
+        for cmd in (
+            SET_DISP,  # display off
+            # address setting
+            SET_MEM_ADDR,
+            0x00,  # horizontal
+            # resolution and layout
+            SET_DISP_START_LINE,  # start at line 0
+            SET_SEG_REMAP | 0x01,  # column addr 127 mapped to SEG0
+            SET_MUX_RATIO,
+            self.height - 1,
+            SET_COM_OUT_DIR | 0x08,  # scan from COM[N] to COM0
+            SET_DISP_OFFSET,
+            0x00,
+            SET_COM_PIN_CFG,
+            0x02 if self.width > 2 * self.height else 0x12,
+            # timing and driving scheme
+            SET_DISP_CLK_DIV,
+            0x80,
+            SET_PRECHARGE,
+            0x22 if self.external_vcc else 0xF1,
+            SET_VCOM_DESEL,
+            0x30,  # 0.83*Vcc
+            # display
+            SET_CONTRAST,
+            0xFF,  # maximum
+            SET_ENTIRE_ON,  # output follows RAM contents
+            SET_NORM_INV,  # not inverted
+            SET_IREF_SELECT,
+            0x30,  # enable internal IREF during display on
+            # charge pump
+            SET_CHARGE_PUMP,
+            0x10 if self.external_vcc else 0x14,
+            SET_DISP | 0x01,  # display on
+        ):  # on
+            self.write_cmd(cmd)
+        self.fill(0)
+        self.show()
+
+    def poweroff(self):
+        self.write_cmd(SET_DISP)
+
+    def poweron(self):
+        self.write_cmd(SET_DISP | 0x01)
+
+    def contrast(self, contrast):
+        self.write_cmd(SET_CONTRAST)
+        self.write_cmd(contrast)
+
+    def invert(self, invert):
+        self.write_cmd(SET_NORM_INV | (invert & 1))
+
+    def rotate(self, rotate):
+        self.write_cmd(SET_COM_OUT_DIR | ((rotate & 1) << 3))
+        self.write_cmd(SET_SEG_REMAP | (rotate & 1))
+
+    def show(self):
+        x0 = 0
+        x1 = self.width - 1
+        if self.width != 128:
+            # narrow displays use centred columns
+            col_offset = (128 - self.width) // 2
+            x0 += col_offset
+            x1 += col_offset
+        self.write_cmd(SET_COL_ADDR)
+        self.write_cmd(x0)
+        self.write_cmd(x1)
+        self.write_cmd(SET_PAGE_ADDR)
+        self.write_cmd(0)
+        self.write_cmd(self.pages - 1)
+        self.write_data(self.buffer)
+
+
+class SSD1306_I2C(SSD1306):
+    def __init__(self, width, height, i2c, addr=0x3C, external_vcc=False):
+        self.i2c = i2c
+        self.addr = addr
+        self.temp = bytearray(2)
+        self.write_list = [b"\x40", None]  # Co=0, D/C#=1
+        super().__init__(width, height, external_vcc)
+
+    def write_cmd(self, cmd):
+        self.temp[0] = 0x80  # Co=1, D/C#=0
+        self.temp[1] = cmd
+        self.i2c.writeto(self.addr, self.temp)
+
+    def write_data(self, buf):
+        self.write_list[1] = buf
+        self.i2c.writevto(self.addr, self.write_list)
+
+
+class SSD1306_SPI(SSD1306):
+    def __init__(self, width, height, spi, dc, res, cs, external_vcc=False):
+        self.rate = 10 * 1024 * 1024
+        dc.init(dc.OUT, value=0)
+        res.init(res.OUT, value=0)
+        cs.init(cs.OUT, value=1)
+        self.spi = spi
+        self.dc = dc
+        self.res = res
+        self.cs = cs
+        import time
+
+        self.res(1)
+        time.sleep_ms(1)
+        self.res(0)
+        time.sleep_ms(10)
+        self.res(1)
+        super().__init__(width, height, external_vcc)
+
+    def write_cmd(self, cmd):
+        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
+        self.cs(1)
+        self.dc(0)
+        self.cs(0)
+        self.spi.write(bytearray([cmd]))
+        self.cs(1)
+
+    def write_data(self, buf):
+        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
+        self.cs(1)
+        self.dc(1)
+        self.cs(0)
+        self.spi.write(buf)
+        self.cs(1)