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)