cleanup
This commit is contained in:
@@ -1,100 +0,0 @@
|
||||
{
|
||||
"1": {
|
||||
"name": "Michal_Dolezal",
|
||||
"time": 59686.214496,
|
||||
"laps": 3,
|
||||
"lap_time": {
|
||||
"0": 1570.6283019999973,
|
||||
"1": 1714.182087000001,
|
||||
"2": 1446.9299450000035
|
||||
}
|
||||
},
|
||||
"2": {
|
||||
"name": "David_Hubalek",
|
||||
"time": 61285.385381,
|
||||
"laps": 3,
|
||||
"lap_time": {
|
||||
"0": 1378.498722999997,
|
||||
"1": 1902.1900959999984,
|
||||
"2": 2436.201885000002
|
||||
}
|
||||
},
|
||||
"9": {
|
||||
"name": "Lucie_Ponocna",
|
||||
"time": 60148.765882,
|
||||
"laps": 3,
|
||||
"lap_time": {
|
||||
"0": 1481.6570529999954,
|
||||
"1": 1371.9764339999965,
|
||||
"2": 1392.5671650000004
|
||||
}
|
||||
},
|
||||
"8": {
|
||||
"name": "Jan_Cada",
|
||||
"time": 60218.945151,
|
||||
"laps": 3,
|
||||
"lap_time": {
|
||||
"0": 1293.8216450000036,
|
||||
"1": 1341.646053000004,
|
||||
"2": 1402.6441269999996
|
||||
}
|
||||
},
|
||||
"10": {
|
||||
"name": "Jenda",
|
||||
"time": 61880.872883,
|
||||
"laps": 3,
|
||||
"lap_time": {
|
||||
"0": 1365.9300070000027,
|
||||
"1": 1959.2283549999993,
|
||||
"2": 1957.2417010000063
|
||||
}
|
||||
},
|
||||
"4": {
|
||||
"name": "Vojta_Flekr",
|
||||
"time": 61567.55196,
|
||||
"laps": 3,
|
||||
"lap_time": {
|
||||
"0": 1447.2361820000006,
|
||||
"1": 1654.6209919999965,
|
||||
"2": 1696.5821589999978
|
||||
}
|
||||
},
|
||||
"7": {
|
||||
"name": "Lukas_Bouchal",
|
||||
"time": 61356.658114,
|
||||
"laps": 3,
|
||||
"lap_time": {
|
||||
"0": 1456.207137999998,
|
||||
"1": 1405.8979709999985,
|
||||
"2": 1365.8517150000043
|
||||
}
|
||||
},
|
||||
"3": {
|
||||
"name": "Adam_Krejsa",
|
||||
"time": 61475.209162,
|
||||
"laps": 3,
|
||||
"lap_time": {
|
||||
"0": 1313.4791200000036,
|
||||
"1": 1358.869869999995,
|
||||
"2": 1402.3635040000008
|
||||
}
|
||||
},
|
||||
"6": {
|
||||
"name": "David_Bedrnicek",
|
||||
"time": 62894.44809,
|
||||
"laps": 3,
|
||||
"lap_time": {
|
||||
"0": 1480.398721999998,
|
||||
"1": 1721.0690019999965,
|
||||
"2": 1939.8222899999964
|
||||
}
|
||||
},
|
||||
"11": {
|
||||
"name": "Nacelnik",
|
||||
"time": 64679.27699,
|
||||
"laps": 1,
|
||||
"lap_time": {
|
||||
"0": 2533.054164000001
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,32 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Created on Sat Oct 21 17:19:11 2023
|
||||
|
||||
@author: angoosh
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
FILE = './Alkator.json'
|
||||
|
||||
data_dict = ''
|
||||
results = {}
|
||||
|
||||
with open(FILE,'r') as f:
|
||||
file_content = f.read()
|
||||
data_dict = json.loads(file_content)
|
||||
|
||||
for id in data_dict:
|
||||
laps = int(data_dict[id]['laps'])
|
||||
time = 0
|
||||
for i in range(0,laps):
|
||||
time += float(data_dict[id]['lap_time'][str(i)])
|
||||
time = (time - (600*laps)) / 60
|
||||
print(data_dict[id]['name'],time)
|
||||
data_dict[id]['total_time'] = time
|
||||
results[data_dict[id]['name']] = time
|
||||
|
||||
print(results)
|
||||
dict(sorted(results.items(), key=lambda item: item[0]))
|
||||
print(results)
|
||||
@@ -1,7 +0,0 @@
|
||||
{
|
||||
"station_id": 2,
|
||||
"host": "https://beta.alkator.cz",
|
||||
"login": "station_register@alkator.cz",
|
||||
"password": "password_heslo",
|
||||
"countdown_seconds": 60
|
||||
}
|
||||
@@ -1 +0,0 @@
|
||||
[]
|
||||
@@ -1,205 +0,0 @@
|
||||
import gi
|
||||
import threading
|
||||
import time
|
||||
import datetime
|
||||
import requests
|
||||
import json
|
||||
|
||||
gi.require_version('Gtk', '4.0')
|
||||
from queue import Queue
|
||||
from gi.repository import Gtk, Gio, Gdk
|
||||
from requests.adapters import HTTPAdapter, Retry
|
||||
from usbcardreader import UsbCardReader
|
||||
|
||||
|
||||
with open('config.json', 'r') as f:
|
||||
config = json.load(f)
|
||||
|
||||
|
||||
activeCards = []
|
||||
|
||||
TIME = config['countdown_seconds']
|
||||
|
||||
db = {}
|
||||
|
||||
queue_to_send = []
|
||||
|
||||
|
||||
session = requests.Session()
|
||||
|
||||
|
||||
retries = Retry(total=10,
|
||||
backoff_factor=1,
|
||||
status_forcelist=[ 500, 502, 503, 504 ])
|
||||
|
||||
session.mount('https://', HTTPAdapter(max_retries=retries))
|
||||
|
||||
|
||||
session.post(config['host'] + '/api/login', {'email':config['login'], 'password': config['password']})
|
||||
|
||||
def refreshDb():
|
||||
global db
|
||||
resp = session.get(config['host'] + '/api/racers')
|
||||
if resp.status_code == 200:
|
||||
json = resp.json()
|
||||
db = {}
|
||||
for racer in json:
|
||||
if racer["card_id"]:
|
||||
db[racer["card_id"]] = racer
|
||||
|
||||
|
||||
def refreshDbEvery3mins():
|
||||
while True:
|
||||
refreshDb()
|
||||
time.sleep(3 * 60)
|
||||
|
||||
|
||||
threading.Thread(target=refreshDbEvery3mins).start()
|
||||
|
||||
log = []
|
||||
|
||||
|
||||
with open('log.json', 'r') as f:
|
||||
log = json.loads(f.read())
|
||||
|
||||
|
||||
queue = Queue()
|
||||
|
||||
|
||||
def worker():
|
||||
while True:
|
||||
item = queue.get()
|
||||
log.append(item)
|
||||
with open('log.json', 'w') as f:
|
||||
f.write(json.dumps(log))
|
||||
response = session.post(item['url'], json=item['json'])
|
||||
print(item, response)
|
||||
|
||||
|
||||
threading.Thread(target=worker).start()
|
||||
|
||||
|
||||
def callbackOnCard(card_id):
|
||||
global win, activeCards, db
|
||||
if card_id not in db:
|
||||
refreshDb()
|
||||
if card_id in db:
|
||||
if card_id not in activeCards:
|
||||
activeCards.append(card_id)
|
||||
win.arb(db[card_id])
|
||||
print('Card is active now')
|
||||
else:
|
||||
print('Card is already active')
|
||||
else:
|
||||
print(f'Card {card_id} not found in db')
|
||||
|
||||
|
||||
class GridRow():
|
||||
def __init__(self, parent, racer, index):
|
||||
global queue_to_send
|
||||
self.id = racer['starting_number']
|
||||
self.parent = parent
|
||||
|
||||
self.grid = Gtk.Grid()
|
||||
|
||||
if index % 2 == 0:
|
||||
self.grid.set_name("lightgrid")
|
||||
else:
|
||||
self.grid.set_name("darkgrid")
|
||||
|
||||
self.runid = Gtk.Label(label=str(racer["starting_number"]), name="gridlabel")
|
||||
self.runid.set_justify(2)
|
||||
self.runid.set_width_chars(10)
|
||||
|
||||
self.runb = Gtk.Label(label='00:00', name="gridlabel")
|
||||
self.runb.set_hexpand(1)
|
||||
|
||||
self.grid.attach(self.runid, 0, 0, 1, 1)
|
||||
self.grid.attach(self.runb, 1, 0, 1, 1)
|
||||
|
||||
parent.grid.attach(self.grid, 0, index, 2, 1)
|
||||
|
||||
countdown=threading.Thread(target=self.updateTime)
|
||||
countdown.start()
|
||||
|
||||
queue.put({'url': config['host'] + '/api/station/register', 'json': {
|
||||
'card_id': racer['card_id'],
|
||||
'station_id': config['station_id'],
|
||||
'time': datetime.datetime.now().strftime('%d.%m.%Y %H:%M:%S.%f')
|
||||
}})
|
||||
|
||||
def printIndex(self, button):
|
||||
print(self.index)
|
||||
|
||||
def updateTime(self):
|
||||
t = TIME
|
||||
while t >= -10:
|
||||
if t < 0:
|
||||
mins, secs = divmod((t * -1), 60)
|
||||
timer = '-{:02d}:{:02d}'.format(mins, secs)
|
||||
else:
|
||||
mins, secs = divmod(t, 60)
|
||||
timer = '{:02d}:{:02d}'.format(mins, secs)
|
||||
self.runb.set_label(timer)
|
||||
time.sleep(1)
|
||||
t -= 1
|
||||
if t == 10:
|
||||
if self.grid.get_name() == "lightgrid":
|
||||
self.grid.set_name("lredgrid")
|
||||
else:
|
||||
self.grid.set_name("dredgrid")
|
||||
|
||||
GridWindow.remRow(self.parent)
|
||||
|
||||
class GridWindow(Gtk.ApplicationWindow):
|
||||
def __init__(self, **kargs):
|
||||
super().__init__(**kargs, title='Alkator Clock')
|
||||
|
||||
css_provider = Gtk.CssProvider()
|
||||
css_provider.load_from_file(Gio.File.new_for_path("style.css"))
|
||||
Gtk.StyleContext.add_provider_for_display(Gdk.Display.get_default(), css_provider, Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION)
|
||||
|
||||
runlabel = Gtk.Label(label="ID", name="gridlabel")
|
||||
runlabel.set_justify(2)
|
||||
cntlabel = Gtk.Label(label="Countdown", name="gridlabel")
|
||||
cntlabel.set_justify(2)
|
||||
|
||||
self.runners = []
|
||||
|
||||
self.grid = Gtk.Grid()
|
||||
self.grid.attach(runlabel, 0, 0, 1, 1)
|
||||
self.grid.attach(cntlabel, 1, 0, 2, 1)
|
||||
|
||||
self.set_child(self.grid)
|
||||
|
||||
def arb(self, racer):
|
||||
self.runners.append(GridRow(self, racer, len(activeCards)))
|
||||
|
||||
def rrb(self, button):
|
||||
self.remRow()
|
||||
|
||||
def addRow(self, runner, index):
|
||||
runid = Gtk.Label(label=str(runner))
|
||||
runb = Gtk.Button(label='time_placeholder')
|
||||
|
||||
self.grid.attach(runid, 0, index, 1, 1)
|
||||
self.grid.attach(runb, 1, index, 1, 1)
|
||||
|
||||
def remRow(self):
|
||||
global activeCards
|
||||
if len(activeCards) > 0:
|
||||
self.grid.remove_row(1)
|
||||
del activeCards[0]
|
||||
|
||||
|
||||
def on_activate(app):
|
||||
# Create window
|
||||
global win
|
||||
win = GridWindow(application=app)
|
||||
win.present()
|
||||
|
||||
UsbCardReader(callbackOnCard)
|
||||
app = Gtk.Application(application_id='com.example.App')
|
||||
app.connect('activate', on_activate)
|
||||
|
||||
app.run(None)
|
||||
@@ -1,4 +0,0 @@
|
||||
{
|
||||
"1": 3619716179,
|
||||
"2": 3620062547
|
||||
}
|
||||
@@ -1,184 +0,0 @@
|
||||
import sys
|
||||
import datetime
|
||||
import requests
|
||||
import threading
|
||||
import json
|
||||
from PyQt6 import QtWidgets, uic, QtGui, QtCore
|
||||
from PyQt6.QtCore import pyqtSignal as Signal, pyqtSlot as Slot
|
||||
from queue import Queue
|
||||
|
||||
from requests.adapters import HTTPAdapter, Retry
|
||||
from usbcardreader import UsbCardReader
|
||||
|
||||
app = QtWidgets.QApplication(sys.argv)
|
||||
|
||||
with open('config.json', 'r') as f:
|
||||
config = json.load(f)
|
||||
|
||||
session = requests.Session()
|
||||
session.post(config['host'] + '/api/login', {'email':config['login'], 'password': config['password']})
|
||||
|
||||
|
||||
retries = Retry(total=10,
|
||||
backoff_factor=1,
|
||||
status_forcelist=[ 500, 502, 503, 504 ],)
|
||||
|
||||
session.mount('https://', HTTPAdapter(max_retries=retries))
|
||||
|
||||
|
||||
last_card = 0
|
||||
last_time = datetime.datetime.now()
|
||||
|
||||
racers = []
|
||||
|
||||
log = []
|
||||
|
||||
|
||||
with open('log.json', 'r') as f:
|
||||
log = json.loads(f.read())
|
||||
|
||||
|
||||
queue = Queue()
|
||||
|
||||
|
||||
class Worker(QtCore.QThread):
|
||||
returned_state = Signal(str)
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.returned_state.connect(self.handle_returned_state)
|
||||
self.worker_thread = threading.Thread(target=self.worker)
|
||||
self.worker_thread.start()
|
||||
|
||||
def worker(self):
|
||||
while True:
|
||||
item = queue.get()
|
||||
log.append({
|
||||
'url': item['url'],
|
||||
'json': item['json'],
|
||||
})
|
||||
with open('log.json', 'w') as f:
|
||||
f.write(json.dumps(log))
|
||||
response = session.post(item['url'], json=item['json'])
|
||||
if response.status_code == 200:
|
||||
if item.get('on_success'):
|
||||
item['on_success']()
|
||||
if item.get('message'):
|
||||
self.returned_state.emit(f'{item['message']}')
|
||||
else:
|
||||
if item.get('failed'):
|
||||
self.returned_state.emit(f'{item['failed']}')
|
||||
|
||||
print(item, response)
|
||||
def handle_returned_state(self, state):
|
||||
mb = QtWidgets.QMessageBox(text=f"{state}")
|
||||
mb.exec()
|
||||
|
||||
|
||||
worker = Worker()
|
||||
|
||||
|
||||
def register_racer():
|
||||
card_id = last_card
|
||||
index = window.racers.currentIndex()
|
||||
if not index.data():
|
||||
mb = QtWidgets.QMessageBox(text='Prosím vyberte závodníka!')
|
||||
mb.exec()
|
||||
return
|
||||
racer_id = int(index.data().split(' ')[-1])
|
||||
try:
|
||||
starting_number = max(filter(lambda r: r['starting_number'], racers), key=lambda racer: racer['starting_number'])['starting_number'] + 1
|
||||
except ValueError:
|
||||
starting_number = 1
|
||||
racer = list(filter(lambda x: x['racer_id'] == racer_id, racers))[0]
|
||||
|
||||
def register_update_racer():
|
||||
racer['card_id'] = card_id
|
||||
racer['starting_number'] = starting_number
|
||||
racer['started'] = False
|
||||
updateRacers()
|
||||
|
||||
queue.put({
|
||||
'url': config['host'] + '/api/card/register',
|
||||
'json': {'racer_id': racer_id, 'starting_number': starting_number, 'card_id': card_id, 'time': datetime.datetime.now().strftime('%d.%m.%Y %H:%M:%S.%f')},
|
||||
'message': f"Úspěšné zaregistrování závodníka {racer['first_name']} {racer['last_name']} se startovním číslem {starting_number}!",
|
||||
'failed': f"Neúspěšná registrace závodníka {racer['first_name']} {racer['last_name']}",
|
||||
'on_success': register_update_racer,
|
||||
})
|
||||
|
||||
|
||||
model = QtGui.QStandardItemModel()
|
||||
|
||||
def updateRacers():
|
||||
model.clear()
|
||||
for racer in racers:
|
||||
if racer['starting_number'] or racer['card_id']:
|
||||
continue
|
||||
item = QtGui.QStandardItem(f"{racer['last_name']} {racer['first_name']} {racer['date_of_birth']} {racer['racer_id']}")
|
||||
model.appendRow(item)
|
||||
|
||||
|
||||
updateRacers()
|
||||
|
||||
|
||||
def reload_racers():
|
||||
global racers
|
||||
response = session.get(config['host'] + '/api/racers')
|
||||
|
||||
racers = response.json()
|
||||
updateRacers()
|
||||
|
||||
|
||||
reload_racers()
|
||||
|
||||
|
||||
def findByCard(card_id):
|
||||
for racer in racers:
|
||||
if racer['card_id'] == card_id:
|
||||
return racer
|
||||
|
||||
|
||||
def updateLastCard(card_id):
|
||||
global last_card, last_time
|
||||
if card_id == last_card and datetime.datetime.now() - last_time < datetime.timedelta(seconds=15):
|
||||
return
|
||||
time = last_time = datetime.datetime.now()
|
||||
last_card = card_id
|
||||
window.lastCard.setText(str(card_id))
|
||||
|
||||
racer = findByCard(card_id)
|
||||
if racer:
|
||||
if racer['started']:
|
||||
def finish_update_racer():
|
||||
racer['card_id'] = None
|
||||
queue.put({
|
||||
'url': config['host'] + '/api/card/unregister',
|
||||
'json': {'racer_id': racer['racer_id'], 'starting_number': racer['starting_number'], 'card_id': card_id, 'time': time.strftime('%d.%m.%Y %H:%M:%S.%f')},
|
||||
'message': f"Úspěšné odhlášení závodníka {racer['starting_number']}!",
|
||||
'failed': f"Neúspěšné odhlášení karty závodníka {racer['starting_number']}",
|
||||
'on_success': finish_update_racer,
|
||||
})
|
||||
else:
|
||||
def start_update_racer():
|
||||
racer['started'] = True
|
||||
queue.put({
|
||||
'url': config['host'] + '/api/station/register',
|
||||
'json': {'card_id': card_id, 'time': time.strftime('%d.%m.%Y %H:%M:%S.%f'), 'station_id': 1},
|
||||
'message': f"Úspěšné odstartování závodníka {racer['starting_number']}!",
|
||||
'failed': f"Núspěšné odstartování závodníka {racer['starting_number']}, prosím registraci opakujte!",
|
||||
'on_success': start_update_racer,
|
||||
})
|
||||
|
||||
|
||||
UsbCardReader(updateLastCard)
|
||||
|
||||
|
||||
|
||||
|
||||
window = uic.loadUi("registrace.ui")
|
||||
window.racers.setModel(model)
|
||||
window.register_racer.clicked.connect(register_racer)
|
||||
window.reload_racers.clicked.connect(reload_racers)
|
||||
window.show()
|
||||
|
||||
|
||||
app.exec()
|
||||
@@ -1,61 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<ui version="4.0">
|
||||
<class>Frame</class>
|
||||
<widget class="QFrame" name="Frame">
|
||||
<property name="geometry">
|
||||
<rect>
|
||||
<x>0</x>
|
||||
<y>0</y>
|
||||
<width>315</width>
|
||||
<height>289</height>
|
||||
</rect>
|
||||
</property>
|
||||
<property name="windowTitle">
|
||||
<string>Registrace závodníků</string>
|
||||
</property>
|
||||
<layout class="QGridLayout" name="gridLayout">
|
||||
<item row="4" column="0">
|
||||
<spacer name="verticalSpacer">
|
||||
<property name="orientation">
|
||||
<enum>Qt::Orientation::Vertical</enum>
|
||||
</property>
|
||||
<property name="sizeHint" stdset="0">
|
||||
<size>
|
||||
<width>20</width>
|
||||
<height>40</height>
|
||||
</size>
|
||||
</property>
|
||||
</spacer>
|
||||
</item>
|
||||
<item row="0" column="0">
|
||||
<widget class="QPushButton" name="register_racer">
|
||||
<property name="text">
|
||||
<string>Registrovat</string>
|
||||
</property>
|
||||
</widget>
|
||||
</item>
|
||||
<item row="2" column="0">
|
||||
<widget class="QLineEdit" name="lastCard"/>
|
||||
</item>
|
||||
<item row="1" column="0">
|
||||
<widget class="QLabel" name="label_2">
|
||||
<property name="text">
|
||||
<string>číslo karty</string>
|
||||
</property>
|
||||
</widget>
|
||||
</item>
|
||||
<item row="0" column="1" rowspan="5">
|
||||
<widget class="QListView" name="racers"/>
|
||||
</item>
|
||||
<item row="3" column="0">
|
||||
<widget class="QPushButton" name="reload_racers">
|
||||
<property name="text">
|
||||
<string>Obnovit Závodníky</string>
|
||||
</property>
|
||||
</widget>
|
||||
</item>
|
||||
</layout>
|
||||
</widget>
|
||||
<resources/>
|
||||
<connections/>
|
||||
</ui>
|
||||
@@ -1,51 +0,0 @@
|
||||
import usb.core
|
||||
import usb.util
|
||||
import time
|
||||
import requests
|
||||
import json
|
||||
|
||||
USB_IF = 0 # Interface
|
||||
USB_TIMEOUT = 5 # Timeout in MS
|
||||
USB_VENDOR = 0xffff # Vendor-ID:
|
||||
USB_PRODUCT = 0x0035 # Product-ID
|
||||
|
||||
# Find the HID device by vender/product ID
|
||||
dev = usb.core.find(idVendor=USB_VENDOR, idProduct=USB_PRODUCT)
|
||||
|
||||
# Get and store the endpoint
|
||||
endpoint = dev[0][(0,0)][0]
|
||||
|
||||
if dev.is_kernel_driver_active(USB_IF) is True:
|
||||
dev.detach_kernel_driver(USB_IF)
|
||||
|
||||
# Claim the device
|
||||
usb.util.claim_interface(dev, USB_IF)
|
||||
|
||||
receivedNumber = 0
|
||||
while True:
|
||||
control = None
|
||||
|
||||
try:
|
||||
# Read a character from the device
|
||||
control = dev.read(endpoint.bEndpointAddress, endpoint.wMaxPacketSize, USB_TIMEOUT)
|
||||
# Here you have to analyze what's coming in.
|
||||
# In my case you had to check the first byte (command)
|
||||
if (control[2] != 40) & (control[2] != 0):
|
||||
# Convert ascii to a number, there's probably better ways to do so.
|
||||
receivedDigit = control[2] - 29
|
||||
|
||||
if receivedDigit == 10:
|
||||
receivedDigit = 0
|
||||
|
||||
# Append the digit to the number
|
||||
receivedNumber = 10 * receivedNumber + receivedDigit
|
||||
|
||||
# Check if the received character is CRLF
|
||||
if (( control[0] == 0 )) & (( control[2] == 40 )) & (( not receivedNumber == 0 )):
|
||||
print('cardNumber:', receivedNumber)
|
||||
except KeyboardInterrupt:
|
||||
exit()
|
||||
except:
|
||||
pass
|
||||
|
||||
time.sleep(0.001) # Let CTRL+C actually exit
|
||||
@@ -1,2 +0,0 @@
|
||||
#!/bin/bash
|
||||
GTK_THEME=Adwaita:light python ./main.py
|
||||
@@ -1,110 +0,0 @@
|
||||
import threading
|
||||
import time
|
||||
import usb.core
|
||||
import usb.util
|
||||
import requests
|
||||
import json
|
||||
import datetime
|
||||
|
||||
DEBUG = False
|
||||
|
||||
if DEBUG == True:
|
||||
index = 2
|
||||
else:
|
||||
index = 1
|
||||
activeCards = []
|
||||
|
||||
TIME = 600
|
||||
|
||||
USB_IF = 0 # Interface
|
||||
USB_TIMEOUT = 5 # Timeout in ms
|
||||
USB_VENDOR = 0xffff # Vendor-ID:
|
||||
USB_PRODUCT = 0x0035 # Product-ID
|
||||
|
||||
with open('people.json', 'r') as f:
|
||||
db = json.load(f)
|
||||
|
||||
try:
|
||||
times = {}
|
||||
with open('times.json', 'r') as f:
|
||||
times_json = json.load(f)
|
||||
for i in times_json:
|
||||
times[i] = {}
|
||||
times[i]['start'] = datetime.datetime.strptime(times_json[i]['start'],"%d/%m/%Y, %H:%M:%S")
|
||||
try:
|
||||
times[i]['end'] = datetime.datetime.strptime(times_json[i]['end'],"%d/%m/%Y, %H:%M:%S")
|
||||
times[i]['duration'] = times[i]['end'] - times[i]['start']
|
||||
except:
|
||||
print(i, "Not finished yet")
|
||||
print(times)
|
||||
except Exception as e: print(e)
|
||||
#except:
|
||||
# times = {}
|
||||
# times_json = {}
|
||||
|
||||
class UsbCardReader():
|
||||
def __init__(self, **kargs):
|
||||
# Find the HID device by vendor/product ID
|
||||
self.dev = usb.core.find(idVendor=USB_VENDOR, idProduct=USB_PRODUCT)
|
||||
# Get and store the endpoint
|
||||
self.endpoint = self.dev[0][(0,0)][0]
|
||||
if self.dev.is_kernel_driver_active(USB_IF) is True:
|
||||
self.dev.detach_kernel_driver(USB_IF)
|
||||
# Claim the device
|
||||
usb.util.claim_interface(self.dev, USB_IF)
|
||||
self.receivedNumber = 0
|
||||
|
||||
cardread=threading.Thread(target=self.read)
|
||||
cardread.start()
|
||||
|
||||
def read(self):
|
||||
global activeCards, db, times, times_json
|
||||
|
||||
print("Waiting for card")
|
||||
receivedNumber = 0
|
||||
|
||||
while True:
|
||||
control = None
|
||||
try:
|
||||
control = self.dev.read(self.endpoint.bEndpointAddress, self.endpoint.wMaxPacketSize, USB_TIMEOUT)
|
||||
if (control[2] != 40) & (control[2] != 0):
|
||||
receivedDigit = control[2] - 29
|
||||
if receivedDigit == 10:
|
||||
receivedDigit = 0
|
||||
receivedNumber = 10 * receivedNumber + receivedDigit
|
||||
|
||||
if (( control[0] == 0 )) & (( control[2] == 40 )) & (( not receivedNumber == 0 )):
|
||||
try:
|
||||
with open('people.json', 'r') as f:
|
||||
db = json.load(f)
|
||||
for i in db:
|
||||
if db[i] == receivedNumber:
|
||||
print("Runner:", i)
|
||||
print(db)
|
||||
runner_found = False
|
||||
for j in times:
|
||||
if j == i:
|
||||
runner_found = True
|
||||
times[j]['end'] = datetime.datetime.now()
|
||||
times[j]['duration'] = times[j]['end'] - times[j]['start']
|
||||
times_json[j]['end'] = times[j]['end'].strftime("%d/%m/%Y, %H:%M:%S")
|
||||
times_json[j]['duration'] = str(times[j]['duration'])
|
||||
if runner_found == False:
|
||||
times[i] = {}
|
||||
times[i]['start'] = datetime.datetime.now()
|
||||
times_json[i] = {}
|
||||
times_json[i]['start'] = times[i]['start'].strftime("%d/%m/%Y, %H:%M:%S")
|
||||
with open("times.json", 'w') as f:
|
||||
f.write(json.dumps(times_json, indent = 4))
|
||||
print(times)
|
||||
except Exception as e: print(e)
|
||||
receivedNumber = 0
|
||||
|
||||
except KeyboardInterrupt:
|
||||
exit()
|
||||
except:
|
||||
pass
|
||||
|
||||
time.sleep(0.001)
|
||||
|
||||
UsbCardReader()
|
||||
@@ -1,20 +0,0 @@
|
||||
#lightgrid {
|
||||
background-color: #F0F0F0;
|
||||
}
|
||||
|
||||
#darkgrid {
|
||||
background-color: #D0D0D0;
|
||||
}
|
||||
|
||||
#lredgrid {
|
||||
background-color: #FF8888;
|
||||
}
|
||||
|
||||
#dredgrid {
|
||||
background-color: #FF6666;
|
||||
}
|
||||
|
||||
#gridlabel {
|
||||
font-size: 16pt;
|
||||
font-weight: bold;
|
||||
}
|
||||
@@ -1,12 +0,0 @@
|
||||
{
|
||||
"1": {
|
||||
"start": "25/02/2025, 18:06:16",
|
||||
"end": "25/02/2025, 18:06:31",
|
||||
"duration": "0:00:14.553644"
|
||||
},
|
||||
"2": {
|
||||
"start": "25/02/2025, 18:15:08",
|
||||
"end": "25/02/2025, 18:15:27",
|
||||
"duration": "0:00:19.543765"
|
||||
}
|
||||
}
|
||||
@@ -1,52 +0,0 @@
|
||||
import usb.core
|
||||
import usb.util
|
||||
import threading
|
||||
import time
|
||||
|
||||
|
||||
USB_IF = 0 # Interface
|
||||
USB_TIMEOUT = 5 # Timeout in ms
|
||||
USB_VENDOR = 0xffff # Vendor-ID:
|
||||
USB_PRODUCT = 0x0035 # Product-ID
|
||||
|
||||
|
||||
class UsbCardReader:
|
||||
def __init__(self, callback):
|
||||
#fn called on found card
|
||||
self.callback = callback
|
||||
# Find the HID device by vendor/product ID
|
||||
self.dev = usb.core.find(idVendor=USB_VENDOR, idProduct=USB_PRODUCT)
|
||||
# Get and store the endpoint
|
||||
self.endpoint = self.dev[0][(0,0)][0]
|
||||
if self.dev.is_kernel_driver_active(USB_IF) is True:
|
||||
self.dev.detach_kernel_driver(USB_IF)
|
||||
# Claim the device
|
||||
usb.util.claim_interface(self.dev, USB_IF)
|
||||
self.receivedNumber = 0
|
||||
|
||||
cardread=threading.Thread(target=self.read)
|
||||
cardread.start()
|
||||
|
||||
def read(self):
|
||||
receivedNumber = 0
|
||||
|
||||
while True:
|
||||
control = None
|
||||
try:
|
||||
control = self.dev.read(self.endpoint.bEndpointAddress, self.endpoint.wMaxPacketSize, USB_TIMEOUT)
|
||||
if (control[2] != 40) and (control[2] != 0):
|
||||
receivedDigit = control[2] - 29
|
||||
if receivedDigit == 10:
|
||||
receivedDigit = 0
|
||||
receivedNumber = 10 * receivedNumber + receivedDigit
|
||||
|
||||
if ( control[0] == 0 ) and ( control[2] == 40 ) and ( not receivedNumber == 0 ):
|
||||
self.callback(receivedNumber)
|
||||
receivedNumber = 0
|
||||
except KeyboardInterrupt:
|
||||
exit()
|
||||
except:
|
||||
receivedNumber = 0
|
||||
pass
|
||||
|
||||
time.sleep(0.001)
|
||||
@@ -1,84 +0,0 @@
|
||||
import time
|
||||
import usb.core
|
||||
import usb.util
|
||||
import requests
|
||||
import json
|
||||
import threading
|
||||
|
||||
USB_IF = 0 # Interface
|
||||
USB_TIMEOUT = 5 # Timeout in ms
|
||||
USB_VENDOR = 0xffff # Vendor-ID:
|
||||
USB_PRODUCT = 0x0035 # Product-ID
|
||||
|
||||
activeCards = {}
|
||||
|
||||
try:
|
||||
with open('people.json', 'r') as f:
|
||||
activeCards = json.load(f)
|
||||
except:
|
||||
print("No people.json found")
|
||||
|
||||
class UsbCardReader():
|
||||
def __init__(self, **kargs):
|
||||
# Find the HID device by vendor/product ID
|
||||
self.dev = usb.core.find(idVendor=USB_VENDOR, idProduct=USB_PRODUCT)
|
||||
# Get and store the endpoint
|
||||
self.endpoint = self.dev[0][(0,0)][0]
|
||||
if self.dev.is_kernel_driver_active(USB_IF) is True:
|
||||
self.dev.detach_kernel_driver(USB_IF)
|
||||
# Claim the device
|
||||
usb.util.claim_interface(self.dev, USB_IF)
|
||||
self.receivedNumber = 0
|
||||
|
||||
def read(self):
|
||||
global activeCards
|
||||
|
||||
print("Priloz kartu")
|
||||
|
||||
while True:
|
||||
control = None
|
||||
try:
|
||||
control = self.dev.read(self.endpoint.bEndpointAddress, self.endpoint.wMaxPacketSize, USB_TIMEOUT)
|
||||
if (control[2] != 40) & (control[2] != 0):
|
||||
receivedDigit = control[2] - 29
|
||||
if receivedDigit == 10:
|
||||
receivedDigit = 0
|
||||
self.receivedNumber = 10 * self.receivedNumber + receivedDigit
|
||||
|
||||
if (( control[0] == 0 )) & (( control[2] == 40 )) & (( not self.receivedNumber == 0 )):
|
||||
try:
|
||||
cardPresent = False
|
||||
for i in activeCards:
|
||||
if activeCards[i] == self.receivedNumber:
|
||||
cardPresent = True
|
||||
break
|
||||
if cardPresent == False:
|
||||
assign=threading.Thread(target=self.assign)
|
||||
assign.start()
|
||||
else:
|
||||
print("Naramek jiz zaregistrovan")
|
||||
self.receivedNumber = 0
|
||||
print("Priloz kartu")
|
||||
except Exception as e: print(e)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
exit()
|
||||
except:
|
||||
pass
|
||||
|
||||
time.sleep(0.001)
|
||||
|
||||
def assign(self):
|
||||
global activeCards
|
||||
|
||||
n = input("Cislo zavodnika: ")
|
||||
activeCards[str(n)] = self.receivedNumber
|
||||
with open('people.json', 'w') as f:
|
||||
f.write(json.dumps(activeCards, indent = 4))
|
||||
self.receivedNumber = 0
|
||||
print("Ulozeno, priloz kartu")
|
||||
|
||||
if __name__ == "__main__":
|
||||
reader = UsbCardReader()
|
||||
print("Zapis zavodniku ready")
|
||||
reader.read()
|
||||
100
SW/PC/main.py
100
SW/PC/main.py
@@ -1,100 +0,0 @@
|
||||
#!/usr/bin/python
|
||||
|
||||
import serial
|
||||
import json
|
||||
from datetime import datetime
|
||||
|
||||
JSON_PATH = './Alkator.json'
|
||||
|
||||
data_dict = ''
|
||||
|
||||
try:
|
||||
with open(JSON_PATH, 'r') as file:
|
||||
file_content = file.read()
|
||||
data_dict = json.loads(file_content)
|
||||
print('json_exists')
|
||||
except:
|
||||
data_dict = {}
|
||||
|
||||
def get_seconds():
|
||||
current_time = datetime.now()
|
||||
hours = current_time.hour
|
||||
mins = current_time.minute
|
||||
secs = current_time.second
|
||||
us = current_time.microsecond
|
||||
seconds = hours*3600 + mins*60 + secs
|
||||
seconds = float(str(seconds) + '.' + str(us))
|
||||
return seconds
|
||||
|
||||
def read(ser):
|
||||
while True:
|
||||
mes = ser.read_until(b'\r')
|
||||
if b'Reading ntag' in mes:
|
||||
print('reading')
|
||||
if b'Data' in mes:
|
||||
data = mes.decode('utf-8')
|
||||
data = data[6:].replace('\x00', '').replace('\r', '').split(' ')
|
||||
print('data read done')
|
||||
try:
|
||||
lap_secs = get_seconds() - data_dict[data[0]]['time']
|
||||
print(lap_secs)
|
||||
lap = data_dict[data[0]]['laps']
|
||||
try:
|
||||
data_dict[data[0]]['lap_time'][lap] = lap_secs
|
||||
except:
|
||||
data_dict[data[0]]['lap_time'] = {lap: lap_secs}
|
||||
lap += 1
|
||||
data_dict[data[0]]['laps'] = lap
|
||||
data_dict[data[0]]['time'] = get_seconds()
|
||||
|
||||
except:
|
||||
data_dict[data[0]] = {'name': data[1]}
|
||||
data_dict[data[0]]['time'] = get_seconds()
|
||||
data_dict[data[0]]['laps'] = 0
|
||||
|
||||
data_json = json.dumps(data_dict, indent=4)
|
||||
print(data_json)
|
||||
with open(JSON_PATH, 'w') as file:
|
||||
file.write(data_json)
|
||||
|
||||
def write(ser):
|
||||
while True:
|
||||
data = input('ID:')
|
||||
data = data + '\r'
|
||||
ser.write(data.encode())
|
||||
while True:
|
||||
mes = ser.read_until(b'\r')
|
||||
if b'Waiting' in mes:
|
||||
print('Waiting for card')
|
||||
if b'Erasing' in mes:
|
||||
print('Erasing card')
|
||||
if b'Writing' in mes:
|
||||
print('Writing data to card')
|
||||
if b'done' in mes:
|
||||
print('Write done')
|
||||
break
|
||||
|
||||
def main():
|
||||
s = serial.Serial(port="/dev/ttyACM3", parity=serial.PARITY_EVEN, stopbits=serial.STOPBITS_ONE, timeout=1)
|
||||
s.flush()
|
||||
|
||||
rw = input('Read or Write tags? [R/w]')
|
||||
rw = rw.lower()
|
||||
|
||||
if rw == 'w':
|
||||
s.write('w\r'.encode())
|
||||
write(s)
|
||||
elif rw == 'r':
|
||||
s.write('r\r'.encode())
|
||||
read(s)
|
||||
elif rw == '':
|
||||
s.write('r\r'.encode())
|
||||
read(s)
|
||||
else:
|
||||
print('Invalid option')
|
||||
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,163 +0,0 @@
|
||||
import neopixel
|
||||
import time
|
||||
import sys
|
||||
import select
|
||||
from machine import Pin, SPI
|
||||
|
||||
#neopixel
|
||||
ws_pin = 2
|
||||
led_num = 10
|
||||
BRIGHTNESS = 0.2 # Adjust the brightness (0.0 - 1.0)
|
||||
|
||||
led_strip = neopixel.NeoPixel(Pin(ws_pin), led_num)
|
||||
|
||||
push_button = Pin(24, Pin.IN, Pin.PULL_UP)
|
||||
buzzer = Pin(3, Pin.OUT)
|
||||
led = Pin(25, Pin.OUT)
|
||||
|
||||
pushbutton_pressed = 0
|
||||
blink_interrupted = 0
|
||||
|
||||
def set_neopixel_brightness(color):
|
||||
r, g, b = color
|
||||
r = int(r * BRIGHTNESS)
|
||||
g = int(g * BRIGHTNESS)
|
||||
b = int(b * BRIGHTNESS)
|
||||
return (r, g, b)
|
||||
|
||||
def set_neopixel_color(r,g,b):
|
||||
color = (r,g,b)
|
||||
color = set_neopixel_brightness(color)
|
||||
led_strip.fill(color)
|
||||
led_strip.write()
|
||||
|
||||
set_neopixel_color(0,0,0)
|
||||
|
||||
|
||||
|
||||
def loop_neopixel():
|
||||
# Display red
|
||||
color = (255, 0, 0) # Red color
|
||||
color = set_neopixel_brightness(color)
|
||||
led_strip.fill(color)
|
||||
led_strip.write()
|
||||
time.sleep(1)
|
||||
|
||||
# Display green
|
||||
color = (0, 255, 0) # Green color
|
||||
color = set_neopixel_brightness(color)
|
||||
led_strip.fill(color)
|
||||
led_strip.write()
|
||||
time.sleep(1)
|
||||
|
||||
# Display blue
|
||||
color = (0, 0, 255) # Blue color
|
||||
color = set_neopixel_brightness(color)
|
||||
led_strip.fill(color)
|
||||
led_strip.write()
|
||||
time.sleep(1)
|
||||
|
||||
neopix_loop_dot_pos = 0
|
||||
def neopixel_loop_dot():
|
||||
global neopix_loop_dot_pos
|
||||
color = (255, 0, 0)
|
||||
black = (0, 0, 0)
|
||||
|
||||
color = set_neopixel_brightness(color)
|
||||
led_strip[neopix_loop_dot_pos] = black
|
||||
neopix_loop_dot_pos += 1
|
||||
if neopix_loop_dot_pos > 9:
|
||||
neopix_loop_dot_pos = 0
|
||||
led_strip[neopix_loop_dot_pos] = color
|
||||
led_strip.write()
|
||||
time.sleep(0.2)
|
||||
|
||||
def lightup_led(index, color):
|
||||
led_strip[index] = color
|
||||
led_strip.write()
|
||||
|
||||
def blink_and_beep(times):
|
||||
global pushbutton_pressed, blink_interrupted
|
||||
|
||||
blink_delay = 0.2
|
||||
|
||||
#insert beep
|
||||
buzzer.value(1)
|
||||
for i in range(0, times):
|
||||
set_neopixel_color(255, 0, 0)
|
||||
buzzer.value(0)
|
||||
time.sleep(blink_delay)
|
||||
set_neopixel_color(0, 0, 0)
|
||||
buzzer.value(1)
|
||||
time.sleep(blink_delay)
|
||||
if pushbutton_pressed == 1:
|
||||
pushbutton_pressed = 0
|
||||
blink_interrupted = 1
|
||||
i = times
|
||||
buzzer.value(0)
|
||||
return
|
||||
buzzer.value(0)
|
||||
|
||||
ten_min_timer = 0
|
||||
ten_min_timer_led_index = 11
|
||||
def count_10_min():
|
||||
global ten_min_timer, ten_min_timer_led_index, pushbutton_pressed, blink_interrupted
|
||||
|
||||
red = (255, 0, 0)
|
||||
red = set_neopixel_brightness(red)
|
||||
green = (0, 255, 0)
|
||||
green = set_neopixel_brightness(green)
|
||||
|
||||
if ten_min_timer_led_index == 11:
|
||||
return
|
||||
|
||||
if (ten_min_timer % 60) == 0:
|
||||
if ten_min_timer_led_index == 10:
|
||||
pushbutton_pressed = 0
|
||||
blink_and_beep(15)
|
||||
if blink_interrupted == 1:
|
||||
blink_interrupted = 0
|
||||
return
|
||||
ten_min_timer_led_index = 11
|
||||
led_strip[0] = green
|
||||
led_strip[2] = green
|
||||
led_strip[4] = green
|
||||
led_strip[6] = green
|
||||
led_strip[8] = green
|
||||
led_strip.write()
|
||||
return
|
||||
lightup_led(ten_min_timer_led_index, red)
|
||||
ten_min_timer_led_index += 1
|
||||
|
||||
ten_min_timer += 1
|
||||
|
||||
def button_callback(push_button):
|
||||
global ten_min_timer, ten_min_timer_led_index, pushbutton_pressed
|
||||
pushbutton_pressed = 1
|
||||
ten_min_timer = 0
|
||||
ten_min_timer_led_index = 0
|
||||
set_neopixel_color(0,0,0)
|
||||
count_10_min()
|
||||
|
||||
set_neopixel_color(0,0,0)
|
||||
push_button.irq(trigger=Pin.IRQ_FALLING, handler=button_callback)
|
||||
|
||||
green = (0, 255, 0)
|
||||
green = set_neopixel_brightness(green)
|
||||
led_strip[0] = green
|
||||
led_strip[2] = green
|
||||
led_strip[4] = green
|
||||
led_strip[6] = green
|
||||
led_strip[8] = green
|
||||
led_strip.write()
|
||||
|
||||
led_state = False
|
||||
while True:
|
||||
count_10_min()
|
||||
time.sleep(1)
|
||||
if led_state:
|
||||
led.low()
|
||||
led_state = False
|
||||
else:
|
||||
led.high()
|
||||
led_state = True
|
||||
@@ -1,408 +0,0 @@
|
||||
# @Author: carlosgilgonzalez
|
||||
# @Date: 2019-10-15T16:13:47+01:00
|
||||
# @Last modified by: carlosgilgonzalez
|
||||
# @Last modified time: 2019-10-15T16:56:27+01:00
|
||||
|
||||
|
||||
# Original work:
|
||||
# Adafruit PN532 NFC/RFID control library.
|
||||
# Author: Tony DiCola
|
||||
|
||||
# Partial Port to Micropython:
|
||||
# Micropython PN532 NFC/RFID control library.
|
||||
# Author: Carlos Gil Gonzalez
|
||||
"""
|
||||
Micropython PN532 NFC/RFID control library (SPI)
|
||||
https://github.com/Carglglz/NFC_PN532_SPI
|
||||
"""
|
||||
|
||||
import time
|
||||
from machine import Pin
|
||||
from micropython import const
|
||||
|
||||
# __version__ = "0.0.0-auto.0"
|
||||
# __repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_PN532.git"
|
||||
|
||||
# pylint: disable=bad-whitespace
|
||||
_PREAMBLE = const(0x00)
|
||||
_STARTCODE1 = const(0x00)
|
||||
_STARTCODE2 = const(0xFF)
|
||||
_POSTAMBLE = const(0x00)
|
||||
|
||||
_HOSTTOPN532 = const(0xD4)
|
||||
_PN532TOHOST = const(0xD5)
|
||||
|
||||
# PN532 Commands
|
||||
_COMMAND_DIAGNOSE = const(0x00)
|
||||
_COMMAND_GETFIRMWAREVERSION = const(0x02)
|
||||
_COMMAND_GETGENERALSTATUS = const(0x04)
|
||||
_COMMAND_SETSERIALBAUDRATE = const(0x10)
|
||||
_COMMAND_SETPARAMETERS = const(0x12)
|
||||
_COMMAND_SAMCONFIGURATION = const(0x14)
|
||||
|
||||
_COMMAND_INLISTPASSIVETARGET = const(0x4A)
|
||||
|
||||
_COMMAND_INDATAEXCHANGE = const(0x40)
|
||||
_COMMAND_INCOMMUNICATETHRU = const(0x42)
|
||||
|
||||
|
||||
_RESPONSE_INDATAEXCHANGE = const(0x41)
|
||||
_RESPONSE_INLISTPASSIVETARGET = const(0x4B)
|
||||
|
||||
_WAKEUP = const(0x55)
|
||||
|
||||
_MIFARE_ISO14443A = const(0x00)
|
||||
|
||||
# Mifare Commands
|
||||
MIFARE_CMD_AUTH_A = const(0x60)
|
||||
MIFARE_CMD_AUTH_B = const(0x61)
|
||||
MIFARE_CMD_READ = const(0x30)
|
||||
MIFARE_CMD_WRITE = const(0xA0)
|
||||
MIFARE_ULTRALIGHT_CMD_WRITE = const(0xA2)
|
||||
|
||||
# Known keys
|
||||
KEY_DEFAULT_B = bytes([0xFF]*6)
|
||||
|
||||
|
||||
_ACK = b'\x00\x00\xFF\x00\xFF\x00'
|
||||
_FRAME_START = b'\x00\x00\xFF'
|
||||
# pylint: enable=bad-whitespace
|
||||
_SPI_STATREAD = const(0x02)
|
||||
_SPI_DATAWRITE = const(0x01)
|
||||
_SPI_DATAREAD = const(0x03)
|
||||
_SPI_READY = const(0x01)
|
||||
|
||||
|
||||
def _reset(pin):
|
||||
"""Perform a hardware reset toggle"""
|
||||
pin.init(Pin.OUT)
|
||||
pin.value(True)
|
||||
time.sleep(0.1)
|
||||
pin.value(False)
|
||||
time.sleep(0.5)
|
||||
pin.value(True)
|
||||
time.sleep(0.1)
|
||||
|
||||
|
||||
class BusyError(Exception):
|
||||
"""Base class for exceptions in this module."""
|
||||
pass
|
||||
|
||||
|
||||
def reverse_bit(num):
|
||||
"""Turn an LSB byte to an MSB byte, and vice versa. Used for SPI as
|
||||
it is LSB for the PN532, but 99% of SPI implementations are MSB only!"""
|
||||
result = 0
|
||||
for _ in range(8):
|
||||
result <<= 1
|
||||
result += (num & 1)
|
||||
num >>= 1
|
||||
return result
|
||||
|
||||
|
||||
class PN532:
|
||||
"""Driver for the PN532 connected over SPI. Pass in a hardware or bitbang
|
||||
SPI device & chip select digitalInOut pin. Optional IRQ pin (not used),
|
||||
reset pin and debugging output."""
|
||||
|
||||
def __init__(self, spi, cs_pin, irq=None, reset=None, debug=False):
|
||||
"""Create an instance of the PN532 class using SPI"""
|
||||
self.debug = debug
|
||||
self._irq = irq
|
||||
self.CSB = cs_pin
|
||||
self._spi = spi
|
||||
self.CSB.on()
|
||||
if reset:
|
||||
if debug:
|
||||
print("Resetting")
|
||||
_reset(reset)
|
||||
|
||||
try:
|
||||
self._wakeup()
|
||||
# self.get_firmware_version() # first time often fails, try 2ce
|
||||
return
|
||||
except (BusyError, RuntimeError):
|
||||
pass
|
||||
# self.get_firmware_version()
|
||||
|
||||
def _wakeup(self):
|
||||
"""Send any special commands/data to wake up PN532"""
|
||||
time.sleep(1)
|
||||
self.CSB.off()
|
||||
time.sleep_ms(2)
|
||||
self._spi.write(bytearray([0x00]))
|
||||
time.sleep_ms(2)
|
||||
self.CSB.on() # pylint: disable=no-member
|
||||
time.sleep(1)
|
||||
|
||||
def _wait_ready(self, timeout=1000):
|
||||
"""Poll PN532 if status byte is ready, up to `timeout` milliseconds"""
|
||||
status_query = bytearray([reverse_bit(_SPI_STATREAD), 0])
|
||||
status = bytearray([0, 0])
|
||||
timestamp = time.ticks_ms()
|
||||
while time.ticks_diff(time.ticks_ms(), timestamp) < timeout:
|
||||
time.sleep(0.02) # required
|
||||
self.CSB.off()
|
||||
time.sleep_ms(2)
|
||||
self._spi.write_readinto(status_query, status)
|
||||
time.sleep_ms(2)
|
||||
self.CSB.on()
|
||||
if reverse_bit(status[1]) == 0x01: # LSB data is read in MSB
|
||||
return True # Not busy anymore!
|
||||
else:
|
||||
time.sleep(0.01) # pause a bit till we ask again
|
||||
# Timed out!
|
||||
return False
|
||||
|
||||
def _read_data(self, count):
|
||||
"""Read a specified count of bytes from the PN532."""
|
||||
# Build a read request frame.
|
||||
frame = bytearray(count+1)
|
||||
# Add the SPI data read signal byte, but LSB'ify it
|
||||
frame[0] = reverse_bit(_SPI_DATAREAD)
|
||||
time.sleep(0.02) # required
|
||||
self.CSB.off()
|
||||
time.sleep_ms(2)
|
||||
self._spi.write_readinto(frame, frame)
|
||||
time.sleep_ms(2)
|
||||
self.CSB.on()
|
||||
for i, val in enumerate(frame):
|
||||
frame[i] = reverse_bit(val) # turn LSB data to MSB
|
||||
if self.debug:
|
||||
print("DEBUG: _read_data: ", [hex(i) for i in frame[1:]])
|
||||
return frame[1:] # don't return the status byte
|
||||
|
||||
def _write_data(self, framebytes):
|
||||
"""Write a specified count of bytes to the PN532"""
|
||||
# start by making a frame with data write in front,
|
||||
# then rest of bytes, and LSBify it
|
||||
rev_frame = [reverse_bit(x)
|
||||
for x in bytes([_SPI_DATAWRITE]) + framebytes]
|
||||
if self.debug:
|
||||
print("DEBUG: _write_data: ", [hex(i) for i in rev_frame])
|
||||
time.sleep(0.02) # required
|
||||
self.CSB.off()
|
||||
time.sleep_ms(2)
|
||||
self._spi.write(bytes(rev_frame)) # pylint: disable=no-member
|
||||
time.sleep_ms(2)
|
||||
self.CSB.on()
|
||||
|
||||
def _write_frame(self, data):
|
||||
"""Write a frame to the PN532 with the specified data bytearray."""
|
||||
assert data is not None and 1 < len(
|
||||
data) < 255, 'Data must be array of 1 to 255 bytes.'
|
||||
# Build frame to send as:
|
||||
# - Preamble (0x00)
|
||||
# - Start code (0x00, 0xFF)
|
||||
# - Command length (1 byte)
|
||||
# - Command length checksum
|
||||
# - Command bytes
|
||||
# - Checksum
|
||||
# - Postamble (0x00)
|
||||
length = len(data)
|
||||
frame = bytearray(length+8)
|
||||
frame[0] = _PREAMBLE
|
||||
frame[1] = _STARTCODE1
|
||||
frame[2] = _STARTCODE2
|
||||
checksum = sum(frame[0:3])
|
||||
frame[3] = length & 0xFF
|
||||
frame[4] = (~length + 1) & 0xFF
|
||||
frame[5:-2] = data
|
||||
checksum += sum(data)
|
||||
frame[-2] = ~checksum & 0xFF
|
||||
frame[-1] = _POSTAMBLE
|
||||
# Send frame.
|
||||
if self.debug:
|
||||
print('DEBUG: _write_frame: ', [hex(i) for i in frame])
|
||||
self._write_data(bytes(frame))
|
||||
|
||||
def _read_frame(self, length):
|
||||
"""Read a response frame from the PN532 of at most length bytes in size.
|
||||
Returns the data inside the frame if found, otherwise raises an exception
|
||||
if there is an error parsing the frame. Note that less than length bytes
|
||||
might be returned!
|
||||
"""
|
||||
# Read frame with expected length of data.
|
||||
response = self._read_data(length+8)
|
||||
if self.debug:
|
||||
print('DEBUG: _read_frame:', [hex(i) for i in response])
|
||||
|
||||
# Swallow all the 0x00 values that preceed 0xFF.
|
||||
offset = 0
|
||||
while response[offset] == 0x00:
|
||||
offset += 1
|
||||
if offset >= len(response):
|
||||
raise RuntimeError(
|
||||
'Response frame preamble does not contain 0x00FF!')
|
||||
if response[offset] != 0xFF:
|
||||
raise RuntimeError(
|
||||
'Response frame preamble does not contain 0x00FF!')
|
||||
offset += 1
|
||||
if offset >= len(response):
|
||||
raise RuntimeError('Response contains no data!')
|
||||
# Check length & length checksum match.
|
||||
frame_len = response[offset]
|
||||
if (frame_len + response[offset+1]) & 0xFF != 0:
|
||||
raise RuntimeError(
|
||||
'Response length checksum did not match length!')
|
||||
# Check frame checksum value matches bytes.
|
||||
checksum = sum(response[offset+2:offset+2+frame_len+1]) & 0xFF
|
||||
if checksum != 0:
|
||||
raise RuntimeError(
|
||||
'Response checksum did not match expected value: ', checksum)
|
||||
# Return frame data.
|
||||
return response[offset+2:offset+2+frame_len]
|
||||
|
||||
def call_function(self, command, response_length=0, params=[], timeout=1000): # pylint: disable=dangerous-default-value
|
||||
"""Send specified command to the PN532 and expect up to response_length
|
||||
bytes back in a response. Note that less than the expected bytes might
|
||||
be returned! Params can optionally specify an array of bytes to send as
|
||||
parameters to the function call. Will wait up to timeout seconds
|
||||
for a response and return a bytearray of response bytes, or None if no
|
||||
response is available within the timeout.
|
||||
"""
|
||||
# Build frame data with command and parameters.
|
||||
data = bytearray(2+len(params))
|
||||
data[0] = _HOSTTOPN532
|
||||
data[1] = command & 0xFF
|
||||
for i, val in enumerate(params):
|
||||
data[2+i] = val
|
||||
# Send frame and wait for response.
|
||||
try:
|
||||
self._write_frame(data)
|
||||
except OSError:
|
||||
self._wakeup()
|
||||
return None
|
||||
if not self._wait_ready(timeout):
|
||||
if(self.debug):
|
||||
print('DEBUG: _wait_ready timed out waiting for ACK')
|
||||
return None
|
||||
# Verify ACK response and wait to be ready for function response.
|
||||
if not _ACK == self._read_data(len(_ACK)):
|
||||
raise RuntimeError('Did not receive expected ACK from PN532!')
|
||||
if not self._wait_ready(timeout):
|
||||
if(self.debug):
|
||||
print('DEBUG: _wait_ready timed out waiting for response')
|
||||
return None
|
||||
# Read response bytes.
|
||||
response = self._read_frame(response_length+2)
|
||||
if(self.debug):
|
||||
print('DEBUG: call_function response:', [hex(i) for i in response])
|
||||
# Check that response is for the called function.
|
||||
if not (response[0] == _PN532TOHOST and response[1] == (command+1)):
|
||||
raise RuntimeError('Received unexpected command response!')
|
||||
# Return response data.
|
||||
return response[2:]
|
||||
|
||||
def get_firmware_version(self):
|
||||
"""Call PN532 GetFirmwareVersion function and return a tuple with the IC,
|
||||
Ver, Rev, and Support values.
|
||||
"""
|
||||
response = self.call_function(
|
||||
_COMMAND_GETFIRMWAREVERSION, 4, timeout=500)
|
||||
if response is None:
|
||||
raise RuntimeError('Failed to detect the PN532')
|
||||
return tuple(response)
|
||||
|
||||
def SAM_configuration(self): # pylint: disable=invalid-name
|
||||
"""Configure the PN532 to read MiFare cards."""
|
||||
# Send SAM configuration command with configuration for:
|
||||
# - 0x01, normal mode
|
||||
# - 0x14, timeout 50ms * 20 = 1 second
|
||||
# - 0x01, use IRQ pin
|
||||
# Note that no other verification is necessary as call_function will
|
||||
# check the command was executed as expected.
|
||||
self.call_function(_COMMAND_SAMCONFIGURATION,
|
||||
params=[0x01, 0x14, 0x01])
|
||||
|
||||
def read_passive_target(self, card_baud=_MIFARE_ISO14443A, timeout=1000):
|
||||
"""Wait for a MiFare card to be available and return its UID when found.
|
||||
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.
|
||||
"""
|
||||
# Send passive read command for 1 card. Expect at most a 7 byte UUID.
|
||||
try:
|
||||
response = self.call_function(_COMMAND_INLISTPASSIVETARGET,
|
||||
params=[0x01, card_baud],
|
||||
response_length=19,
|
||||
timeout=timeout)
|
||||
except BusyError:
|
||||
return None # no card found!
|
||||
# If no response is available return None to indicate no card is present.
|
||||
if response is None:
|
||||
return None
|
||||
# Check only 1 card with up to a 7 byte UID is present.
|
||||
if response[0] != 0x01:
|
||||
raise RuntimeError('More than one card detected!')
|
||||
if response[5] > 7:
|
||||
raise RuntimeError('Found card with unexpectedly long UID!')
|
||||
# Return UID of card.
|
||||
return response[6:6+response[5]]
|
||||
|
||||
def ntag2xx_write_block(self, block_number, data):
|
||||
"""Write a block of data to the card. Block number should be the block
|
||||
to write and data should be a byte array of length 4 with the data to
|
||||
write. If the data is successfully written then True is returned,
|
||||
otherwise False is returned.
|
||||
"""
|
||||
assert data is not None and len(
|
||||
data) == 4, 'Data must be an array of 4 bytes!'
|
||||
# Build parameters for InDataExchange command to do NTAG203 classic write.
|
||||
params = bytearray(3+len(data))
|
||||
params[0] = 0x01 # Max card numbers
|
||||
params[1] = MIFARE_ULTRALIGHT_CMD_WRITE
|
||||
params[2] = block_number & 0xFF
|
||||
params[3:] = data
|
||||
# Send InDataExchange request.
|
||||
response = self.call_function(_COMMAND_INDATAEXCHANGE,
|
||||
params=params,
|
||||
response_length=1)
|
||||
return response[0] == 0x00
|
||||
|
||||
def ntag2xx_read_block(self, block_number):
|
||||
"""Read a block of data from the card. Block number should be the block
|
||||
to read. If the block is successfully read a bytearray of length 16 with
|
||||
data starting at the specified block will be returned. If the block is
|
||||
not read then None will be returned.
|
||||
"""
|
||||
return self.mifare_classic_read_block(block_number)[0:4] # only 4 bytes per page
|
||||
|
||||
def mifare_classic_read_block(self, block_number):
|
||||
"""Read a block of data from the card. Block number should be the block
|
||||
to read. If the block is successfully read a bytearray of length 16 with
|
||||
data starting at the specified block will be returned. If the block is
|
||||
not read then None will be returned.
|
||||
"""
|
||||
# Send InDataExchange request to read block of MiFare data.
|
||||
response = self.call_function(_COMMAND_INDATAEXCHANGE,
|
||||
params=[0x01, MIFARE_CMD_READ,
|
||||
block_number & 0xFF],
|
||||
response_length=17)
|
||||
# Check first response is 0x00 to show success.
|
||||
if response[0] != 0x00:
|
||||
return None
|
||||
# Return first 4 bytes since 16 bytes are always returned.
|
||||
return response[1:]
|
||||
|
||||
def mifare_classic_authenticate_block(self, uid, block_number, key_number=MIFARE_CMD_AUTH_B, key=KEY_DEFAULT_B): # pylint: disable=invalid-name
|
||||
"""Authenticate specified block number for a MiFare classic card. Uid
|
||||
should be a byte array with the UID of the card, block number should be
|
||||
the block to authenticate, key number should be the key type (like
|
||||
MIFARE_CMD_AUTH_A or MIFARE_CMD_AUTH_B), and key should be a byte array
|
||||
with the key data. Returns True if the block was authenticated, or False
|
||||
if not authenticated.
|
||||
"""
|
||||
# Build parameters for InDataExchange command to authenticate MiFare card.
|
||||
uidlen = len(uid)
|
||||
keylen = len(key)
|
||||
params = bytearray(3 + uidlen + keylen)
|
||||
params[0] = 0x01 # Max card numbers
|
||||
params[1] = key_number & 0xFF
|
||||
params[2] = block_number & 0xFF
|
||||
params[3 : 3 + keylen] = key
|
||||
params[3 + keylen :] = uid
|
||||
# Send InDataExchange request and verify response is 0x00.
|
||||
response = self.call_function(
|
||||
_COMMAND_INDATAEXCHANGE, params=params, response_length=1
|
||||
)
|
||||
return response[0] == 0x00
|
||||
@@ -1,214 +0,0 @@
|
||||
import NFC_PN532 as nfc
|
||||
import neopixel
|
||||
import time
|
||||
import sys
|
||||
import select
|
||||
from machine import Pin, SPI
|
||||
|
||||
#neopixel
|
||||
ws_pin = 0
|
||||
led_num = 8
|
||||
BRIGHTNESS = 0.2 # Adjust the brightness (0.0 - 1.0)
|
||||
|
||||
led_strip = neopixel.NeoPixel(Pin(ws_pin), led_num)
|
||||
|
||||
def set_neopixel_brightness(color):
|
||||
r, g, b = color
|
||||
r = int(r * BRIGHTNESS)
|
||||
g = int(g * BRIGHTNESS)
|
||||
b = int(b * BRIGHTNESS)
|
||||
return (r, g, b)
|
||||
|
||||
def set_neopixel_color(r,g,b):
|
||||
color = (r,g,b)
|
||||
color = set_neopixel_brightness(color)
|
||||
led_strip.fill(color)
|
||||
led_strip.write()
|
||||
|
||||
set_neopixel_color(0,0,0)
|
||||
|
||||
# SPI
|
||||
spi_dev = SPI(0, baudrate=1000000)
|
||||
cs = Pin(17, Pin.OUT)
|
||||
cs.on()
|
||||
|
||||
# SENSOR INIT
|
||||
pn532 = nfc.PN532(spi_dev,cs)
|
||||
ic, ver, rev, support = pn532.get_firmware_version()
|
||||
#print('Found PN532 with firmware version: {0}.{1}'.format(ver, rev))
|
||||
|
||||
# Configure PN532 to communicate with MiFare cards
|
||||
pn532.SAM_configuration()
|
||||
|
||||
# FUNCTION TO READ NFC
|
||||
def read_nfc(dev, tmot):
|
||||
"""Accepts a device and a timeout in millisecs """
|
||||
print('Reading...')
|
||||
uid = dev.read_passive_target(timeout=tmot)
|
||||
if uid is None:
|
||||
print('CARD NOT FOUND')
|
||||
else:
|
||||
numbers = [i for i in uid]
|
||||
string_ID = '{}-{}-{}-{}'.format(*numbers)
|
||||
print('Found card with UID:', [hex(i) for i in uid])
|
||||
print('Number_id: {}'.format(string_ID))
|
||||
|
||||
def loop_neopixel():
|
||||
# Display red
|
||||
color = (255, 0, 0) # Red color
|
||||
color = set_neopixel_brightness(color)
|
||||
led_strip.fill(color)
|
||||
led_strip.write()
|
||||
time.sleep(1)
|
||||
|
||||
# Display green
|
||||
color = (0, 255, 0) # Green color
|
||||
color = set_neopixel_brightness(color)
|
||||
led_strip.fill(color)
|
||||
led_strip.write()
|
||||
time.sleep(1)
|
||||
|
||||
# Display blue
|
||||
color = (0, 0, 255) # Blue color
|
||||
color = set_neopixel_brightness(color)
|
||||
led_strip.fill(color)
|
||||
led_strip.write()
|
||||
time.sleep(1)
|
||||
|
||||
def read_nfc_loop():
|
||||
sys.stdout.write('Waiting for RFID/NFC card to read!\r')
|
||||
set_neopixel_color(255,0,0)
|
||||
while True:
|
||||
# Check if a card is available to read
|
||||
uid = pn532.read_passive_target(timeout=0.5)
|
||||
# Try again if no card is available.
|
||||
if uid is not None:
|
||||
break
|
||||
|
||||
set_neopixel_color(255,255,0)
|
||||
|
||||
ntag_string = 'Data: '
|
||||
sys.stdout.write('Reading ntag \r')
|
||||
|
||||
leds_on = 0
|
||||
for page in range(20,40):
|
||||
try:
|
||||
ntag2xx_page = pn532.ntag2xx_read_block(page)
|
||||
if (page % 2) == 0:
|
||||
if page == 0:
|
||||
pass
|
||||
elif page == 2:
|
||||
pass
|
||||
else:
|
||||
led_strip[leds_on] = (0,0,255)
|
||||
leds_on += 1
|
||||
if leds_on > 7:
|
||||
leds_on = 7
|
||||
led_strip.write()
|
||||
|
||||
if ntag2xx_page is not None:
|
||||
for char in ntag2xx_page:
|
||||
ntag_string += chr(char)
|
||||
else:
|
||||
sys.stdout.write('Error reading card\r')
|
||||
return 1
|
||||
except:
|
||||
sys.stdout.write('Error reading card\r')
|
||||
return 1
|
||||
|
||||
set_neopixel_color(0,255,0)
|
||||
sys.stdout.write(ntag_string + '\r')
|
||||
time.sleep(3)
|
||||
|
||||
return 0
|
||||
#print('PAGE ',page,': ',[chr(x) for x in ntag2xx_page])
|
||||
|
||||
def write_nfc_loop(data):
|
||||
content = data.encode('utf-8')
|
||||
length = len(content)
|
||||
pages = int(length / 4)
|
||||
if (length % 4) != 0:
|
||||
pages += 1
|
||||
|
||||
sys.stdout.write("Waiting for RFID/NFC card to write to!\r")
|
||||
set_neopixel_color(255,0,0)
|
||||
while True:
|
||||
# Check if a card is available to read
|
||||
uid = pn532.read_passive_target(timeout=0.5)
|
||||
# Try again if no card is available.
|
||||
if uid is not None:
|
||||
break
|
||||
|
||||
set_neopixel_color(255,255,0)
|
||||
|
||||
sys.stdout.write('Erasing card\r')
|
||||
for page in range(20,41):
|
||||
content_erase = b'\x00\x00\x00\x00'
|
||||
pn532.ntag2xx_write_block(page, content_erase)
|
||||
|
||||
sys.stdout.write('Writing card\r')
|
||||
for page in range(20,20 + pages):
|
||||
content_slice = bytearray(4)
|
||||
for i in range(0,4):
|
||||
try:
|
||||
content_slice[i] = (content[i + 4 * (page - 20)])
|
||||
except:
|
||||
content_slice[i] = 0
|
||||
#content_slice.append(content[i + 4 * (page - 20)])
|
||||
print(content_slice)
|
||||
pn532.ntag2xx_write_block(page, content_slice)
|
||||
|
||||
sys.stdout.write('Write done\r')
|
||||
set_neopixel_color(0,255,0)
|
||||
|
||||
def loop():
|
||||
poll_results = poll_obj.poll(1)
|
||||
if poll_results:
|
||||
rw = sys.stdin.readline().strip()
|
||||
sys.stdout.write("received data: " + rw + "\r")
|
||||
rw = rw.lower()
|
||||
if rw == 'w':
|
||||
while True:
|
||||
sys.stdout.write('ID: \r')
|
||||
poll_results = poll_obj.poll(1)
|
||||
data = sys.stdin.readline().strip()
|
||||
write_nfc_loop(data)
|
||||
elif rw == 'r':
|
||||
while True:
|
||||
ret = read_nfc_loop()
|
||||
if ret == 255:
|
||||
set_neopixel_color(0,0,255)
|
||||
rw = None
|
||||
return 1
|
||||
elif rw == '':
|
||||
while True:
|
||||
ret = read_nfc_loop()
|
||||
if ret == 255:
|
||||
set_neopixel_color(0,0,255)
|
||||
rw = None
|
||||
return 1
|
||||
else:
|
||||
sys.stdout.write('Invalid option\r')
|
||||
|
||||
poll_obj = select.poll()
|
||||
poll_obj.register(sys.stdin, select.POLLIN)
|
||||
|
||||
set_neopixel_color(0,0,255)
|
||||
|
||||
|
||||
while True:
|
||||
poll_results = poll_obj.poll(1)
|
||||
if poll_results:
|
||||
while True:
|
||||
try:
|
||||
ret = loop()
|
||||
if ret == 1:
|
||||
poll_results = None
|
||||
break
|
||||
except:
|
||||
set_neopixel_color(0,0,255)
|
||||
break
|
||||
else:
|
||||
continue
|
||||
#except:
|
||||
# set_neopixel_color(0,0,0)
|
||||
Reference in New Issue
Block a user