Update syntax and improve importations

This commit is contained in:
2024-06-04 07:35:03 +00:00
parent 83fcd383b7
commit e9246bb376
8 changed files with 574 additions and 317 deletions
+37 -33
View File
@@ -7,10 +7,10 @@ from mfrc522 import MFRC522
from ssd1306 import SSD1306_I2C
# Global variables
DOOR_ID = '[Your door ID]'
WLAN_SSID = '[Your SSID]'
WLAN_PASS = '[Your password]'
SERVER_IP = '[Your server IP]'
DOOR_ID = "[Your door ID]"
WLAN_SSID = "[Your SSID]"
WLAN_PASS = "[Your password]"
SERVER_IP = "[Your server IP]"
SERVER_PORT = 5000
# Initialize RFID reader
@@ -31,23 +31,26 @@ 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.text("Initializing...", 0, 0)
oled.show()
except Exception as e:
print("display error:", e)
#init_oled()
# 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(f"Door ID: {DOOR_ID}", 0, 0) # Display Door ID at the top
oled.text("___________________", 0, 3)
lines = message.split('\n')
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)
@@ -58,7 +61,8 @@ def display_message(message, ip_address):
redled.off()
print("display error:", e)
init_oled()
# Connect to WiFi
def connect_wifi(ssid, password):
wlan = network.WLAN(network.STA_IF)
@@ -69,28 +73,30 @@ def connect_wifi(ssid, password):
print("Connecting to WiFi...")
ip_address = wlan.ifconfig()[0]
print("Connected to WiFi:", ip_address)
display_message('WiFi Connected', 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)
display_message("Server Connected", ip_address)
else:
print("Server connection failed")
display_message('Server Fail', ip_address)
display_message("Server Fail", ip_address)
time.sleep(5)
while response.status_code != 200 :
while response.status_code != 200:
wlan.connect(ssid, password)
response = requests.get(f"http://{SERVER_IP}:{SERVER_PORT}/")
display_message('Reconnecting ...', ip_address)
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)
display_message(f"Server Error \n {e}", ip_address)
time.sleep(5)
# while response.status_code != 200 :
# wlan.connect(ssid, password)
# try :
@@ -102,14 +108,12 @@ def connect_wifi(ssid, password):
# 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
}
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
@@ -120,39 +124,39 @@ def main():
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)
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])
rfid_uid_decimal = "".join([str(i) for i in uid])
print("RFID UID:", rfid_uid_decimal)
display_message('Checking...', ip_address)
display_message("Checking...", ip_address)
response = send_rfid_to_server(rfid_uid_decimal)
if response.get('access_granted'):
user_upn = response.get('upn')
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)
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)
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)
display_message("Scan your tag", ip_address)
if __name__ == "__main__":
main()
+163 -161
View File
@@ -1,111 +1,116 @@
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):
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':
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(
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)
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.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)))
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:
@@ -115,40 +120,38 @@ class MFRC522:
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)
@@ -157,35 +160,32 @@ class MFRC522:
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):
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):
@@ -194,134 +194,128 @@ class MFRC522:
stat = self.ERR
else:
stat = self.ERR
return stat, recv
def PcdSelect(self, serNum,anticolN):
def PcdSelect(self, serNum, anticolN):
backData = []
buf = []
buf.append(anticolN)
buf.append(0x70)
#i = 0
# i = 0
###xorsum=0;
for i in serNum:
buf.append(i)
#while i<5:
# 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)
(status, backData, backLen) = self._tocard(0x0C, buf)
if (status == self.OK) and (backLen == 0x18):
return 1
return 1
else:
return 0
def SelectTag(self, uid):
byte5 = 0
#(status,puid)= self.anticoll(self.PICC_ANTICOLL1)
#print("uid",uid,"puid",puid)
# (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="["
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+ "]"
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))
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
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))
(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))
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 :
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))
(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))
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)
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):
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:
@@ -330,51 +324,59 @@ class MFRC522:
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):
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 :
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 :
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 :
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 :
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)
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:
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="")
print("{:02X} ".format(value), end="")
print(" ", end="")
for value in block:
if (value > 0x20) and (value < 0x7f):
print(chr(value),end="")
if (value > 0x20) and (value < 0x7F):
print(chr(value), end="")
else:
print('.',end="")
print(".", end="")
print("")
else:
break
if status == self.ERR:
print("Authentication error")
return self.ERR
return self.OK
return self.OK