174 lines
4.5 KiB
Python
174 lines
4.5 KiB
Python
import subprocess
|
|
import requests
|
|
import time
|
|
from flask import Flask, render_template, request, redirect, url_for, session
|
|
from authlib.integrations.flask_client import OAuth
|
|
from time import sleep
|
|
from dynaconf import Dynaconf
|
|
import logging
|
|
import shelve
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
config = Dynaconf(settings_files = ['config.yaml'])
|
|
|
|
# Setup shelf.
|
|
with shelve.open('shelf', 'c', writeback=True) as shelf:
|
|
if 'logins' not in shelf.keys():
|
|
shelf['logins'] = {}
|
|
|
|
if 'checkins' not in shelf.keys():
|
|
shelf['checkins'] = {}
|
|
|
|
if 'admissions' not in shelf.keys():
|
|
shelf['admissions'] = {}
|
|
|
|
|
|
|
|
|
|
class Pinctrl:
|
|
def __init__(self):
|
|
pass
|
|
|
|
def _run(self, verb, pin, option):
|
|
if config.dry_run is False:
|
|
subprocess.run(['pinctrl', verb, str(pin), option])
|
|
else:
|
|
logger.info(f'Simulating command `pinctrl {verb} {pin} {option}`')
|
|
|
|
def input(self, pin):
|
|
self._run('set', pin, 'ip')
|
|
def output(self, pin):
|
|
self._run('set', pin, 'op')
|
|
def on(self, pin):
|
|
self._run('set', pin, 'dh')
|
|
def off(self, pin):
|
|
self._run('set', pin, 'dl')
|
|
|
|
pinctrl = Pinctrl()
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = config.oauth.secret_key
|
|
oauth = OAuth(app)
|
|
|
|
oauth.register(
|
|
'jimlab',
|
|
**config.oauth.to_dict()
|
|
)
|
|
|
|
# Set up lights.
|
|
for led, pin in config.pinout.led.items():
|
|
pinctrl.output(pin)
|
|
pinctrl.off(pin)
|
|
|
|
|
|
@app.route('/')
|
|
def homepage():
|
|
user = session.get('user')
|
|
if not user:
|
|
return redirect(url_for('login'))
|
|
|
|
logging.info(f"User {user} is logged in.")
|
|
|
|
with shelve.open('shelf', 'r') as shelf:
|
|
|
|
return render_template(
|
|
'index.html',
|
|
user=user,
|
|
logins=shelf.get('logins', {}),
|
|
checkins=shelf.get('checkins', {}),
|
|
admissions=shelf.get('admissions', {}),
|
|
)
|
|
|
|
|
|
@app.route('/login')
|
|
def login():
|
|
logging.info("Login requested")
|
|
redirect_uri = url_for('auth', _external=True)
|
|
return oauth.jimlab.authorize_redirect(redirect_uri)
|
|
|
|
@app.route('/auth')
|
|
def auth():
|
|
token = oauth.jimlab.authorize_access_token()
|
|
logger.critical(f"Token received: {token}")
|
|
|
|
# Fetch user info
|
|
resp = oauth.jimlab.get('https://auth.jimlab.io/application/o/userinfo/') # or the correct user info endpoint
|
|
userinfo = resp.json()
|
|
logger.critical(f"Userinfo: {userinfo}")
|
|
|
|
session['user'] = userinfo
|
|
|
|
# Log the login.
|
|
with shelve.open('shelf', 'w', writeback=True) as shelf:
|
|
username = userinfo.get('preferred_username', 'unknown')
|
|
if username not in shelf['logins']:
|
|
shelf['logins'][username] = []
|
|
shelf['logins'][username].append((userinfo, time.time()))
|
|
|
|
return redirect('/')
|
|
|
|
@app.route('/checkin', methods=['POST'])
|
|
def check_in():
|
|
|
|
if not session.get('user'):
|
|
return redirect(url_for('login'))
|
|
user = session['user']
|
|
username = user.get('preferred_username', 'unknown')
|
|
logger.info(f"Check-in requested by user: {username}")
|
|
|
|
with shelve.open('shelf', 'w', writeback=True) as shelf:
|
|
if username not in shelf['checkins']:
|
|
shelf['checkins'][username] = []
|
|
shelf['checkins'][username].append(time.time())
|
|
|
|
return redirect(url_for('homepage'))
|
|
|
|
|
|
|
|
# @app.route('/')
|
|
# def index():
|
|
# logger.info(oauth.jimlab.get('user'))
|
|
# return render_template('index.html')
|
|
|
|
# @app.route('/checkin', methods=['POST'])
|
|
# def check_in():
|
|
# if can_check_in():
|
|
# pinctrl.on(config.pinout.led[config.led_map.guest])
|
|
# print("Visitor checked in")
|
|
# else:
|
|
# pinctrl.on(config.pinout.led[config.led_map.error])
|
|
# print("Exception occurred: Cannot accept visitors currently")
|
|
|
|
# return redirect(url_for('index'))
|
|
|
|
# @app.route('/status')
|
|
# def get_status_route():
|
|
# status = get_status()
|
|
# return render_template('status.html', status=status)
|
|
|
|
# def get_status():
|
|
# if is_accepting_visitors():
|
|
# return "accepting"
|
|
# else:
|
|
# return "not accepting"
|
|
|
|
# def is_accepting_visitors():
|
|
# # This would be checked against Authentik server or a configuration file
|
|
# # For simplicity, let's assume it's always true in this example
|
|
# return True
|
|
|
|
# def can_check_in():
|
|
# # Logic to check if the system can accept a visitor
|
|
# return is_accepting_visitors()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
app.run(host='0.0.0.0', port=config.server.port, debug=config.server.debug)
|
|
finally:
|
|
# Turn off lights
|
|
for led, pin in config.pinout.led.items():
|
|
pinctrl.off(pin)
|
|
|
|
|