podbay/validate.py

92 lines
2.2 KiB
Python
Raw Normal View History

2018-01-10 07:36:32 +00:00
#!/usr/bin/python
import sys
from datetime import datetime
2018-01-10 21:19:31 +00:00
import time
2018-01-27 23:45:16 +00:00
import json
2018-01-10 21:19:31 +00:00
import RPi.GPIO as IO
import zmq
2018-01-10 21:19:31 +00:00
# Magic numbers for PWM activation
2018-01-10 21:19:31 +00:00
MOTOR_PIN = 19
FREQUENCY = 100
DUTYCYCLE = 100
DURATION = 1.0
# Address of the scanning service
SCANSERVADDRESS = "tcp://localhost:5000"
2018-01-10 07:36:32 +00:00
def timestamped(s):
"""
Prepends a timestamp to a string.
"""
return "[{:%Y-%m-%d %H:%M:%S}] {}".format(datetime.now(), s)
2018-01-10 21:19:31 +00:00
def setup_pwm():
"""
Performs initialization for pulse width modulation.
Returns the PWM object for the motor pin.
"""
IO.setwarnings(False)
IO.setmode(IO.BCM)
IO.setup(MOTOR_PIN, IO.OUT)
p = IO.PWM(MOTOR_PIN, FREQUENCY)
2018-01-10 21:19:31 +00:00
return p
def unlock_door(p):
"""
Uses PWM to turn the motor connected to the door handle.
"""
try:
p.start(DUTYCYCLE)
xtime.sleep(DURATION)
finally:
p.stop()
2018-01-10 21:19:31 +00:00
2018-01-10 07:36:32 +00:00
def read_loop():
"""
Simple REPL skeleton for validating scanned IDs.
"""
# Initialize zmq socket for reading in scanned barcodes
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.setsockopt(zmq.SUBSCRIBE, "")
socket.connect(SCANSERVADDRESS)
2018-01-27 23:45:16 +00:00
# Open the access database
access_raw = open("access.json").read()
try:
access = json.loads(access_raw)
except:
raise SystemExit("Could not load access file!")
# Initialize the PWM pin for opening the door
pwm_pin = setup_pwm()
2018-01-10 07:36:32 +00:00
# Open the log file in append mode
log_file = open("./access.log", "a")
log_file.write(timestamped("=== Door security initiated ===\n"))
# Begin the loop
while True:
# Read in the ID
scanned_id = socket.recv()
2018-01-27 23:45:16 +00:00
# Determine ID authorization
authorized = id in access and "authorized" in access[id] and access[id]["authorized"]
user = access[id]["user"] if id in access and "user" in access[id] else "unknown id"
2018-01-10 07:36:32 +00:00
# If the user is not authorized, deny access
if not authorized:
2018-01-27 23:45:16 +00:00
s = timestamped("Denied {} ({})\n".format(id, user))
2018-01-10 07:36:32 +00:00
log_file.write(s)
2018-01-27 23:45:16 +00:00
print s,
2018-01-10 07:36:32 +00:00
# If the user is authorized, perform the unlock procedure
else:
2018-01-27 23:45:16 +00:00
s = timestamped("Validated {} {}\n".format(id, user))
2018-01-10 07:36:32 +00:00
log_file.write(s)
print s,
2018-01-10 07:36:32 +00:00
# TODO: Play open tone
2018-01-27 23:45:16 +00:00
if "sound" in access[id]:
pass
# Run unlock procedure
2018-01-10 21:19:31 +00:00
unlock_door(pwm_pin)
2018-01-10 07:36:32 +00:00
if __name__ == "__main__":
read_loop()