Compare commits
5 commits
82693fd85a
...
b3fd037cd9
Author | SHA1 | Date | |
---|---|---|---|
b3fd037cd9 | |||
0ba4b41917 | |||
989751c102 | |||
5175af4f5a | |||
d4344b464e |
3 changed files with 349 additions and 50 deletions
327
esp32/main.py
327
esp32/main.py
|
@ -2,57 +2,290 @@ from pn532 import PN532Uart, PN532Error
|
|||
import utime
|
||||
import machine
|
||||
import hashlib
|
||||
import os
|
||||
import asyncio
|
||||
import network
|
||||
import requests
|
||||
import _thread
|
||||
from umqtt.simple import MQTTClient
|
||||
|
||||
DEBUG = True
|
||||
|
||||
try:
|
||||
door = machine.Pin(2, machine.Pin.OUT)
|
||||
door.value(0)
|
||||
|
||||
uart = machine.UART(1, tx=16, rx=17, baudrate=9600)
|
||||
uart.write("F")
|
||||
rf = PN532Uart(2, tx=22, rx=19)
|
||||
rf.SAM_configuration()
|
||||
ic, ver, rev, support = rf.get_firmware_version()
|
||||
print('Found PN532 with firmware version: {0}.{1}'.format(ver, rev))
|
||||
except Exception as e:
|
||||
rf = None
|
||||
print('No NFC reader (PN532) detected')
|
||||
class Keypad:
|
||||
CMD_RESET = 'F'
|
||||
CMD_ENABLE_FEEDBACK = 'Q'
|
||||
CMD_LED_GREEN = 'G'
|
||||
CMD_LED_RED = 'R'
|
||||
CMD_GRANTED = 'H' # AKA happy
|
||||
CMD_DENIED = 'S' # AKA sad
|
||||
|
||||
while rf is not None:
|
||||
try:
|
||||
uid = rf.read_passive_target()
|
||||
print("Card UUID: " + ''.join('{:02x}'.format(x) for x in uid))
|
||||
card=''.join('{:02x}'.format(x) for x in uid)
|
||||
uart.write("QG")
|
||||
def __init__(self, uart):
|
||||
self._uart = uart
|
||||
|
||||
def write(self, data):
|
||||
self._uart.write(data)
|
||||
self._uart.flush()
|
||||
|
||||
async def get_pin(self, timeout_ms=10000):
|
||||
uart = self._uart
|
||||
|
||||
# discard previous bytes
|
||||
uart.read(uart.any())
|
||||
|
||||
uart.write(self.CMD_ENABLE_FEEDBACK)
|
||||
uart.write(self.CMD_LED_GREEN)
|
||||
uart.flush()
|
||||
c=0
|
||||
buf=''
|
||||
success = False
|
||||
|
||||
data = bytearray()
|
||||
while len(data) < 4 and timeout_ms > 0:
|
||||
n = uart.any()
|
||||
for c in uart.read(n):
|
||||
if ord('0') <= c <= ord('9'):
|
||||
# print(f"rx: {c}")
|
||||
data.append(c)
|
||||
await asyncio.sleep(0.1)
|
||||
timeout_ms -= 100
|
||||
|
||||
return data[:4]
|
||||
|
||||
|
||||
class Net:
|
||||
def __init__(self):
|
||||
self._connected = False
|
||||
self._wlan = network.WLAN(network.STA_IF)
|
||||
|
||||
self._events = []
|
||||
# keepalive is needed due to: https://github.com/eclipse/mosquitto/issues/2462
|
||||
self._mqtt = MQTTClient("lock", "10.11.1.1", keepalive=5)
|
||||
self._mqtt.set_callback(self._mqtt_cb)
|
||||
|
||||
async def loop(self):
|
||||
self.start()
|
||||
while True:
|
||||
if uart.any():
|
||||
c=c+1
|
||||
buf+=str(chr(uart.read()[0]))
|
||||
if c == 4:
|
||||
break
|
||||
hash=hashlib.sha256(str("{0:#0{1}x}".format(int(buf),10)+":"+card[6:8]+card[4:6]+card[2:4]+card[0:2])[2:].encode()).digest().hex()
|
||||
print('Card hash: {0}'.format(hash))
|
||||
f = open("hashes")
|
||||
for user in f:
|
||||
if(user.strip() == hash):
|
||||
success = True
|
||||
if(success):
|
||||
print('Known hash, opening door')
|
||||
uart.write("FH")
|
||||
door.value(1)
|
||||
utime.sleep(2)
|
||||
door.value(0)
|
||||
self.update()
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
def _mqtt_cb(self, topic, msg):
|
||||
if topic == b'locks/internal/command':
|
||||
if msg == b'sync':
|
||||
try:
|
||||
print("starting sync")
|
||||
# TODO: add auth support to http server
|
||||
url = "http://10.11.1.1:8000/hashes/internal"
|
||||
print(f"fetching: {url}")
|
||||
rsp = requests.get(url)
|
||||
print(f"sync code: {rsp.status_code}")
|
||||
with open('hashes_new', 'wb') as f:
|
||||
if 200 <= rsp.status_code < 300:
|
||||
while True:
|
||||
chunk = rsp.raw.read(512)
|
||||
if not chunk:
|
||||
break
|
||||
f.write(chunk)
|
||||
os.rename('hashes_new', 'hashes')
|
||||
print("sync finished")
|
||||
self.send_event("sync", 'success'.encode())
|
||||
except Exception as e:
|
||||
print(f"sync error: {e}")
|
||||
self.send_event("sync", 'fail'.encode())
|
||||
else:
|
||||
print(f"uncrecognised command: {msg}")
|
||||
|
||||
|
||||
def send_event(self, name, payload):
|
||||
# TODO: limit number of events in queue
|
||||
self._events.append((name, payload))
|
||||
|
||||
|
||||
def _run_mqtt(self):
|
||||
mqtt = self._mqtt
|
||||
while True:
|
||||
try:
|
||||
while True:
|
||||
mqtt.connect()
|
||||
mqtt.publish("locks/internal/mac", self._wlan.config('mac').hex())
|
||||
mqtt.subscribe("locks/internal/command")
|
||||
|
||||
i = 0
|
||||
while True:
|
||||
try:
|
||||
while True:
|
||||
name, payload = self._events.pop()
|
||||
print("sending event")
|
||||
mqtt.sock.setblocking(True)
|
||||
mqtt.publish(f"locks/internal/events/{name}", payload)
|
||||
except IndexError:
|
||||
pass
|
||||
|
||||
if i == 10:
|
||||
i = 0
|
||||
mqtt.ping()
|
||||
else:
|
||||
i += 1
|
||||
|
||||
mqtt.check_msg()
|
||||
utime.sleep(0.25)
|
||||
|
||||
except Exception as e:
|
||||
print(f"mqtt exception: {e}")
|
||||
utime.sleep(6)
|
||||
|
||||
def start(self):
|
||||
self._wlan.active(True)
|
||||
# for net in self._wlan.scan():
|
||||
# print(f'scan: {net}')
|
||||
print("mac:", self._wlan.config('mac').hex())
|
||||
|
||||
_thread.start_new_thread(self._run_mqtt, ())
|
||||
|
||||
with open('wifi', 'r') as f:
|
||||
ssid = f.readline().strip()
|
||||
key = f.readline().strip()
|
||||
|
||||
if not self._wlan.isconnected():
|
||||
print(f'connecting to network ssid={ssid!r}')
|
||||
self._wlan.connect(ssid, key)
|
||||
|
||||
|
||||
def update(self):
|
||||
if self._wlan.isconnected():
|
||||
if not self._connected:
|
||||
print('network config:', self._wlan.ifconfig())
|
||||
self._connected = True
|
||||
else:
|
||||
print('Unknown hash, ignoring')
|
||||
uart.write("FS")
|
||||
utime.sleep(2)
|
||||
buf=''
|
||||
except PN532Error as e:
|
||||
print('PN532:', e)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
if self._connected:
|
||||
print('disconnected')
|
||||
self._connected = False
|
||||
|
||||
|
||||
class Door:
|
||||
def __init__(self, pin):
|
||||
self._pin = pin
|
||||
|
||||
def unlock(self):
|
||||
if self._pin is not None:
|
||||
self._pin.value(1)
|
||||
|
||||
def lock(self):
|
||||
if self._pin is not None:
|
||||
self._pin.value(0)
|
||||
|
||||
def generate_hash(card_uid, pin):
|
||||
card_uid = bytes(reversed(card_uid[:4])).hex()
|
||||
s = "{:08x}:{}".format(int(pin), card_uid)
|
||||
hash=hashlib.sha256(s.encode('ascii')).digest().hex()
|
||||
return hash
|
||||
|
||||
|
||||
async def handle_auth(nfc, keypad, door, net):
|
||||
while True:
|
||||
card_uid = await nfc.wait_uid()
|
||||
print("Card UUID: " + ''.join('{:02x}'.format(x) for x in card_uid))
|
||||
|
||||
pin = await keypad.get_pin()
|
||||
keypad.write(keypad.CMD_RESET)
|
||||
|
||||
if len(pin) < 4:
|
||||
print("Pin timeout")
|
||||
keypad.write(keypad.CMD_DENIED)
|
||||
await asyncio.sleep(0.5)
|
||||
keypad.write(keypad.CMD_RESET)
|
||||
continue
|
||||
|
||||
hash = generate_hash(card_uid, pin)
|
||||
print(f'Card hash: {hash}')
|
||||
|
||||
hash_found = False
|
||||
with open("hashes") as f:
|
||||
for user in f:
|
||||
if(user.strip() == hash):
|
||||
hash_found = True
|
||||
break
|
||||
|
||||
net.send_event("hash", hash.encode())
|
||||
|
||||
try:
|
||||
if hash_found:
|
||||
print('Known hash, opening door')
|
||||
keypad.write(keypad.CMD_GRANTED)
|
||||
door.unlock()
|
||||
else:
|
||||
print('Unknown hash, ignoring')
|
||||
keypad.write(keypad.CMD_DENIED)
|
||||
|
||||
await asyncio.sleep(2)
|
||||
|
||||
finally:
|
||||
door.lock()
|
||||
keypad.write(keypad.CMD_RESET)
|
||||
|
||||
|
||||
class Nfc:
|
||||
def __init__(self):
|
||||
self._uids = []
|
||||
self._flag = asyncio.Event()
|
||||
|
||||
async def wait_uid(self):
|
||||
# discard queued uids
|
||||
for _ in range(len(self._uids)):
|
||||
self._uids.pop()
|
||||
|
||||
while not self._uids:
|
||||
await self._flag.wait()
|
||||
self._flag.clear()
|
||||
|
||||
uid = None
|
||||
while self._uids:
|
||||
uid = self._uids.pop()
|
||||
return uid
|
||||
|
||||
async def loop(self):
|
||||
rf = PN532Uart(2, rx=19, tx=22)
|
||||
|
||||
try:
|
||||
rf.SAM_configuration()
|
||||
ic, ver, rev, support = rf.get_firmware_version()
|
||||
print(f'Found PN532 with firmware version: {ver}.{rev}')
|
||||
except Exception as e:
|
||||
print('No NFC reader (PN532) detected')
|
||||
raise
|
||||
|
||||
while True:
|
||||
uid = None
|
||||
try:
|
||||
uid = rf.read_passive_target()
|
||||
except PN532Error as e:
|
||||
print('PN532:', e)
|
||||
|
||||
rf.power_down()
|
||||
|
||||
if uid is not None:
|
||||
self._uids.append(uid)
|
||||
self._flag.set()
|
||||
|
||||
await asyncio.sleep(0.25)
|
||||
|
||||
|
||||
async def main():
|
||||
print("Lock app wil start in 2s")
|
||||
await asyncio.sleep(2)
|
||||
|
||||
keypad = Keypad(machine.UART(1, tx=16, rx=17, baudrate=9600))
|
||||
keypad.write(keypad.CMD_RESET)
|
||||
|
||||
nfc = Nfc()
|
||||
|
||||
door = Door(machine.Pin(2, machine.Pin.OUT))
|
||||
door.lock()
|
||||
|
||||
net = Net()
|
||||
|
||||
await asyncio.gather(
|
||||
handle_auth(nfc, keypad, door, net),
|
||||
nfc.loop(),
|
||||
net.loop()
|
||||
)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
asyncio.run(main())
|
||||
|
|
|
@ -1,3 +1,27 @@
|
|||
# repo: https://github.com/insighio/micropython-pn532-uart
|
||||
#
|
||||
# MIT License
|
||||
#
|
||||
# Copyright (c) 2021 Infinite Tree Inc
|
||||
#
|
||||
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
# of this software and associated documentation files (the "Software"), to deal
|
||||
# in the Software without restriction, including without limitation the rights
|
||||
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
# copies of the Software, and to permit persons to whom the Software is
|
||||
# furnished to do so, subject to the following conditions:
|
||||
#
|
||||
# The above copyright notice and this permission notice shall be included in all
|
||||
# copies or substantial portions of the Software.
|
||||
#
|
||||
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
# SOFTWARE.
|
||||
|
||||
from micropython import const
|
||||
import machine
|
||||
import utime
|
||||
|
@ -11,6 +35,8 @@ _COMMAND_GETFIRMWAREVERSION = const(0x02)
|
|||
_COMMAND_INLISTPASSIVETARGET = const(0x4A)
|
||||
_COMMAND_INRELEASE = const(0x52)
|
||||
_COMMAND_SAMCONFIGURATION = const(0x14)
|
||||
_COMMAND_RFCONFIGURATION = const(0x32)
|
||||
_COMMAND_POWERDOWN = const(0x16)
|
||||
|
||||
# Send Frames
|
||||
_PREAMBLE = const(0x00)
|
||||
|
@ -171,9 +197,23 @@ class PN532Uart(object):
|
|||
if self.debug:
|
||||
print("Sending SAM_CONFIGURATION")
|
||||
|
||||
response = self.call_function(_COMMAND_SAMCONFIGURATION, params=[0x01, 0x14, 0x01])
|
||||
response = self.call_function(_COMMAND_SAMCONFIGURATION, params=[0x01])
|
||||
if self.debug:
|
||||
print('SAM_configuration:', [hex(i) for i in response])
|
||||
print('SAM_configuration:', response.hex())
|
||||
|
||||
# Enable RF:
|
||||
# 0x32 - Cmd: RFConfiguration
|
||||
# 0x01 - Item: RF Field
|
||||
# 0x03 - AutoRFCA=on, RF=on
|
||||
self.call_function(_COMMAND_RFCONFIGURATION, params=[0x01, 0x03])
|
||||
|
||||
# Return imidiately aftery trying once:
|
||||
# 0x32 - Cmd: RFConfiguration
|
||||
# 0x05 - Item: MaxRetries
|
||||
# 0x00 - 0 retries (1 try)
|
||||
# 0x00 - 0 retries (1 try)
|
||||
# 0x00 - 0 retries (1 try)
|
||||
self.call_function(_COMMAND_RFCONFIGURATION, params=[0x05, 0x00, 0x00, 0x00])
|
||||
|
||||
def get_firmware_version(self):
|
||||
"""
|
||||
|
@ -194,13 +234,22 @@ class PN532Uart(object):
|
|||
Will wait up to timeout seconds and return None if no card is found,
|
||||
otherwise a bytearray with the UID of the found card is returned.
|
||||
"""
|
||||
|
||||
# Enable RF:
|
||||
# 0x32 - Cmd: RFConfiguration
|
||||
# 0x01 - Item: RF Field
|
||||
# 0x03 - AutoRFCA=on, RF=on
|
||||
self.call_function(_COMMAND_RFCONFIGURATION, params=[0x01, 0x03])
|
||||
|
||||
if self.debug:
|
||||
print("Sending INIT_PASSIVE_TARGET")
|
||||
# Send passive read command for 1 card. Expect at most a 7 byte UUID.
|
||||
response = self.call_function(_COMMAND_INLISTPASSIVETARGET, params=[0x01, card_baud])
|
||||
|
||||
# Check only 1 card with up to a 7 byte UID is present.
|
||||
if response[0] != 0x01:
|
||||
if response[0] == 0x00:
|
||||
return None
|
||||
if response[0] > 0x01:
|
||||
raise PN532Error('More than one card detected!')
|
||||
if response[5] > 7:
|
||||
raise PN532Error('Found card with unexpectedly long UID!')
|
||||
|
@ -208,6 +257,18 @@ class PN532Uart(object):
|
|||
# Return UID of card.
|
||||
return response[6:6+response[5]]
|
||||
|
||||
def power_down(self):
|
||||
# Disable RF:
|
||||
# 0x32 - Cmd: RFConfiguration
|
||||
# 0x01 - Item: RF Field
|
||||
# 0x00 - AutoRFCA=off, RF=off
|
||||
self.call_function(_COMMAND_RFCONFIGURATION, params=[0x01, 0x00])
|
||||
|
||||
# Power down
|
||||
# 0x16 - Cmd: PowerDown
|
||||
# 0x10 - WakeUpEnable: 5 bit - HSU (UART)
|
||||
self.call_function(_COMMAND_POWERDOWN, params=[0x10])
|
||||
|
||||
def release_targets(self):
|
||||
if self.debug:
|
||||
print("Release Targets")
|
||||
|
|
|
@ -5,3 +5,8 @@
|
|||
3) if re-making the hw from scratch, you need to make some HW mods to the keypad:
|
||||
- green led moved to a different pin (due to UART conflict)
|
||||
4) verify the pin IDs
|
||||
|
||||
Clock external 4 mhz
|
||||
Board: ATmega168
|
||||
Variant: 168/168A
|
||||
avrdude -p atmega168 -c usbasp -v -B 4 -U lfuse:w:0xed:m -U hfuse:w:0xdd:m -U efuse:w:0xf9:m
|
||||
|
|
Loading…
Reference in a new issue