tanner
/
airlock-fuse33
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Make specific to Fuse33 / Fabman

Browse Source
master
Tanner Collin 8 months ago
parent c4998709a9
commit 2d74cb11e5
2 changed files with 135 additions and 69 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 199
      main.py
  2. 5
      secrets.py.example

199
main.py
Unescape View File

@ -1,61 +1,90 @@
import os
import logging
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO)
TEST = os.environ.get('TEST', False)
DEBUG = os.environ.get('DEBUG', False)
from multiprocessing import Process, Queue
from queue import Empty
import RPi.GPIO as GPIO
import os
import json
import requests
import serial
import time
from signal import *
if not TEST:
import serial
import RPi.GPIO as GPIO
import secrets
DEBUG = os.environ.get('DEBUG', False)
RELAY_PIN = 17
RFID_EN_PIN = 27
CARDS_FILE = 'card_data.json'
OPEN_DURATION = 4
API_STATS = 'https://api.my.protospace.ca/stats/'
API_DOOR = 'https://api.my.protospace.ca/door/'
API_SEEN = lambda x: 'https://api.my.protospace.ca/door/{}/seen/'.format(x)
VALID_PACKAGES = [
'Maker',
'Maker Plus',
'Maker Pro',
#'Storage bin rental',
#'Backyard Rental Spot',
'IndiCity Laser Space Rental',
#'Shipping container rental',
'Day Pass Holder',
'Access to everything 24/7',
'Barter Membership',
'Loft Member',
]
TEST_PIPE = '/tmp/airlock'
os.remove(TEST_PIPE)
API_MEMBERS = 'https://fabman.io/api/v1/members?limit=1000&embed=key&embed=activePackages&includeKeyToken=true'
ser = None
def unlock_door():
GPIO.output(RELAY_PIN, GPIO.HIGH)
GPIO.output(RFID_EN_PIN, GPIO.HIGH)
logging.info('Unlocking door...')
time.sleep(OPEN_DURATION)
if not TEST:
GPIO.output(RELAY_PIN, GPIO.HIGH)
GPIO.output(RFID_EN_PIN, GPIO.HIGH)
GPIO.output(RELAY_PIN, GPIO.LOW)
GPIO.output(RFID_EN_PIN, GPIO.LOW)
time.sleep(OPEN_DURATION)
GPIO.output(RELAY_PIN, GPIO.LOW)
GPIO.output(RFID_EN_PIN, GPIO.LOW)
def lock_door_on_exit(*args):
logging.info('Exiting, locking door...')
GPIO.output(RELAY_PIN, GPIO.LOW)
GPIO.output(RFID_EN_PIN, GPIO.LOW)
if not TEST:
GPIO.output(RELAY_PIN, GPIO.LOW)
GPIO.output(RFID_EN_PIN, GPIO.LOW)
os._exit(0)
def init():
global ser, cards
GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)
GPIO.setup(RELAY_PIN, GPIO.OUT)
GPIO.output(RELAY_PIN, GPIO.LOW)
GPIO.setup(RFID_EN_PIN, GPIO.OUT)
GPIO.output(RFID_EN_PIN, GPIO.LOW)
logging.info('GPIO initialized')
ser = serial.Serial(port='/dev/ttyAMA0', baudrate=2400, timeout=0.1)
logging.info('Serial initialized')
if not TEST:
GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)
GPIO.setup(RELAY_PIN, GPIO.OUT)
GPIO.output(RELAY_PIN, GPIO.LOW)
GPIO.setup(RFID_EN_PIN, GPIO.OUT)
GPIO.output(RFID_EN_PIN, GPIO.LOW)
logging.info('GPIO initialized')
if TEST:
os.mkfifo(TEST_PIPE)
logging.info('Test pipe initialized')
else:
ser = serial.Serial(port='/dev/ttyAMA0', baudrate=2400, timeout=0.1)
logging.info('Serial initialized')
for sig in (SIGABRT, SIGILL, SIGINT, SIGSEGV, SIGTERM):
signal(sig, lock_door_on_exit)
@ -65,89 +94,127 @@ def reader_thread(card_data_queue):
recent_scans = {}
with open(CARDS_FILE, 'r') as f:
card_data = json.load(f)
logging.info('Read {} card numbers from disk'.format(str(len(card_data))))
cards = json.load(f)
logging.info('Read {} cards from disk'.format(len(cards)))
while True:
try:
card_data = card_data_queue.get_nowait()
cards = card_data_queue.get_nowait()
except Empty:
pass
card = ser.readline()
if TEST:
with open(TEST_PIPE, 'r') as pipe:
card = pipe.readline()
else:
card = ser.readline()
if not card: continue
try:
card = card.decode().strip()
except AttributeError:
card = card.strip()
except UnicodeDecodeError:
continue
if len(card) != 10: continue
if len(card) != 14: continue
# debounce card scans
now = time.time()
if card in recent_scans:
if now - recent_scans[card] < 5.0:
logging.info('Debounce skipping card scan')
continue
recent_scans[card] = now
logging.info('Read card: ' + card)
if card in card_data:
if card in cards:
logging.info('Card recognized')
else:
logging.info('Card not recognized, denying access')
continue
logging.info('DOOR ACCESS - Card: {} | Name: {}'.format(
card, card_data[card],
))
card_data = cards[card]
logging.info('Card belongs to: %s', card_data['name'])
if not any(package in card_data['packages'] for package in VALID_PACKAGES):
logging.info('No valid packages found: %s', str(card_data['packages']))
continue
logging.info('DOOR ACCESS GRANTED - Card: %s | Name: %s', card, card_data['name'])
unlock_door()
try:
res = requests.post(API_SEEN(card), timeout=2)
res.raise_for_status()
except BaseException as e:
logging.error('Problem POSTing seen: {} - {}'.format(e.__class__.__name__, str(e)))
#try:
# res = requests.post(API_SEEN(card), timeout=2)
# res.raise_for_status()
#except BaseException as e:
# logging.error('Problem POSTing seen: {} - {}'.format(e.__class__.__name__, str(e)))
# continue
def get_cards(card_data_queue):
try:
headers = {'Authorization': 'Bearer ' + secrets.FABMAN_API_KEY}
res = requests.get(API_MEMBERS, headers=headers, timeout=10)
res.raise_for_status()
res = res.json()
except BaseException as e:
logging.exception('Problem GETting Fabman API: {} - {}'.format(e.__class__.__name__, str(e)))
return
members = res
cards = {}
logging.info('Got {} members from API'.format(str(len(res))))
for member in members:
if member['state'] != 'active':
continue
def update_thread(card_data_queue):
last_card_change = None
packages = []
while True:
time.sleep(5)
for member_packages in member['_embedded']['memberPackages']:
package = member_packages['_embedded']['package']
try:
res = requests.get(API_STATS, timeout=5)
res.raise_for_status()
res = res.json()
except BaseException as e:
logging.error('Problem GETting stats: {} - {}'.format(e.__class__.__name__, str(e)))
if package['state'] != 'active':
continue
packages.append(package['name'])
key = member['_embedded']['key']
if not key:
continue
if res['last_card_change'] == last_card_change:
if key['state'] != 'active':
continue
last_card_change = res['last_card_change']
logging.info('Cards changed, pulling update from API')
token = key['token']
name = '{} {} ({})'.format(member['firstName'], member['lastName'], member['memberNumber'])
try:
headers = {'Authorization': 'Bearer ' + secrets.DOOR_API_KEY}
res = requests.get(API_DOOR, headers=headers, timeout=5)
res.raise_for_status()
res = res.json()
except BaseException as e:
logging.error('Problem GETting door: {} - {}'.format(e.__class__.__name__, str(e)))
last_card_change = None
continue
cards[token] = dict(name=name, packages=packages)
logging.info('Processed {} cards'.format(len(cards)))
card_data_queue.put(cards)
logging.info('Writing data to file')
with open(CARDS_FILE, 'w') as f:
json.dump(cards, f, indent=4)
def update_thread(card_data_queue):
if not DEBUG: time.sleep(10)
while True:
logging.info('Updating cards...')
get_cards(card_data_queue)
logging.info('Got {} cards from API'.format(str(len(res))))
card_data_queue.put(res)
time.sleep(300)
logging.info('Writing data to file')
with open(CARDS_FILE, 'w') as f:
json.dump(res, f)
def watchdog_thread():
while True:

5
secrets.py.example
Unescape View File

@ -1,3 +1,2 @@
# Door cards API key
# should be equal to the auth token value set in Spaceport
DOOR_API_KEY = ''
# Fabman API key
FABMAN_API_KEY = ''

Write Preview
Loading…
Cancel
Save