add app
This commit is contained in:
235
app.py
Normal file
235
app.py
Normal file
@@ -0,0 +1,235 @@
|
||||
|
||||
from flask import Flask, request, jsonify, render_template_string, redirect
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
from datetime import datetime, timezone, timedelta
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///rfid.db"
|
||||
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
||||
|
||||
db = SQLAlchemy(app)
|
||||
|
||||
# --------------------------------------------------
|
||||
# MODELS
|
||||
# --------------------------------------------------
|
||||
|
||||
class Student(db.Model):
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
uid = db.Column(db.String(64), unique=True)
|
||||
name = db.Column(db.String(128))
|
||||
|
||||
current_room = db.Column(db.String(64))
|
||||
previous_room = db.Column(db.String(64))
|
||||
expected_return = db.Column(db.String(64))
|
||||
|
||||
state = db.Column(db.String(32), default="IN_ROOM")
|
||||
|
||||
last_scan = db.Column(db.DateTime)
|
||||
last_reader = db.Column(db.String(64))
|
||||
|
||||
|
||||
class Event(db.Model):
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
uid = db.Column(db.String(64))
|
||||
room = db.Column(db.String(64))
|
||||
event_type = db.Column(db.String(64))
|
||||
timestamp = db.Column(db.DateTime)
|
||||
|
||||
|
||||
class Bathroom(db.Model):
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
room = db.Column(db.String(64), unique=True)
|
||||
count = db.Column(db.Integer, default=0)
|
||||
max = db.Column(db.Integer, default=2)
|
||||
|
||||
|
||||
class Anomaly(db.Model):
|
||||
id = db.Column(db.Integer, primary_key=True)
|
||||
uid = db.Column(db.String(64))
|
||||
type = db.Column(db.String(64))
|
||||
timestamp = db.Column(db.DateTime)
|
||||
|
||||
|
||||
# --------------------------------------------------
|
||||
# UTIL
|
||||
# --------------------------------------------------
|
||||
|
||||
COOLDOWN = 3
|
||||
|
||||
def now():
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def recent_scan(student):
|
||||
if not student.last_scan:
|
||||
return False
|
||||
return (now() - student.last_scan).total_seconds() < COOLDOWN
|
||||
|
||||
|
||||
# --------------------------------------------------
|
||||
# MOVEMENT ENGINE
|
||||
# --------------------------------------------------
|
||||
|
||||
def process_scan(student, room):
|
||||
timestamp = now()
|
||||
|
||||
# duplicate suppression
|
||||
if recent_scan(student) and student.last_reader == room:
|
||||
return None
|
||||
|
||||
student.last_scan = timestamp
|
||||
student.last_reader = room
|
||||
|
||||
event_type = "move"
|
||||
|
||||
# bathroom logic
|
||||
if room.startswith("bathroom"):
|
||||
if student.state != "IN_BATHROOM":
|
||||
student.previous_room = student.current_room
|
||||
student.current_room = room
|
||||
student.state = "IN_BATHROOM"
|
||||
student.expected_return = student.previous_room
|
||||
event_type = "bathroom_enter"
|
||||
else:
|
||||
student.current_room = "hallway"
|
||||
student.state = "IN_HALLWAY"
|
||||
event_type = "bathroom_exit"
|
||||
|
||||
else:
|
||||
# returning from bathroom
|
||||
if student.state == "IN_HALLWAY" and student.expected_return == room:
|
||||
student.expected_return = None
|
||||
|
||||
student.previous_room = student.current_room
|
||||
student.current_room = room
|
||||
student.state = "IN_ROOM"
|
||||
|
||||
db.session.add(Event(
|
||||
uid=student.uid,
|
||||
room=room,
|
||||
event_type=event_type,
|
||||
timestamp=timestamp
|
||||
))
|
||||
|
||||
# anomaly detection
|
||||
if student.expected_return and room != student.expected_return and student.state == "IN_HALLWAY":
|
||||
db.session.add(Anomaly(
|
||||
uid=student.uid,
|
||||
type="WRONG_RETURN",
|
||||
timestamp=timestamp
|
||||
))
|
||||
|
||||
db.session.commit()
|
||||
|
||||
|
||||
# --------------------------------------------------
|
||||
# ROUTES
|
||||
# --------------------------------------------------
|
||||
|
||||
@app.route("/scan", methods=["POST"])
|
||||
def scan():
|
||||
data = request.json
|
||||
|
||||
student = Student.query.filter_by(uid=data["uid"]).first()
|
||||
if not student:
|
||||
student = Student(uid=data["uid"], name="Unknown")
|
||||
db.session.add(student)
|
||||
db.session.commit()
|
||||
|
||||
process_scan(student, data["room"])
|
||||
|
||||
return jsonify({"status": "ok"})
|
||||
|
||||
|
||||
# --------------------------------------------------
|
||||
# ADMIN DASHBOARD
|
||||
# --------------------------------------------------
|
||||
|
||||
ADMIN_HTML = """
|
||||
<!doctype html>
|
||||
<html>
|
||||
<head>
|
||||
<script src="https://unpkg.com/htmx.org"></script>
|
||||
<script src="https://cdn.tailwindcss.com"></script>
|
||||
</head>
|
||||
<body class="bg-gray-900 text-white p-6">
|
||||
|
||||
<h1 class="text-2xl font-bold">Admin Dashboard</h1>
|
||||
|
||||
<div hx-get="/admin/students" hx-trigger="load, every 3s" hx-swap="innerHTML"></div>
|
||||
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
@app.route("/admin")
|
||||
def admin():
|
||||
return ADMIN_HTML
|
||||
|
||||
|
||||
@app.route("/admin/students")
|
||||
def admin_students():
|
||||
students = Student.query.all()
|
||||
|
||||
html = ""
|
||||
for s in students:
|
||||
html += f"""
|
||||
<div class='bg-gray-800 p-3 m-2 rounded'>
|
||||
<b>{s.name}</b> ({s.uid})<br>
|
||||
Room: {s.current_room}<br>
|
||||
State: {s.state}<br>
|
||||
</div>
|
||||
"""
|
||||
|
||||
return html
|
||||
|
||||
|
||||
# --------------------------------------------------
|
||||
# TEACHER DASHBOARD
|
||||
# --------------------------------------------------
|
||||
|
||||
TEACHER_HTML = """
|
||||
<!doctype html>
|
||||
<html>
|
||||
<head>
|
||||
<script src="https://unpkg.com/htmx.org"></script>
|
||||
<script src="https://cdn.tailwindcss.com"></script>
|
||||
</head>
|
||||
<body class="bg-blue-950 text-white p-6">
|
||||
|
||||
<h1 class="text-2xl font-bold">Teacher Dashboard</h1>
|
||||
|
||||
<div hx-get="/teacher/class" hx-trigger="load, every 3s" hx-swap="innerHTML"></div>
|
||||
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
@app.route("/teacher")
|
||||
def teacher():
|
||||
return TEACHER_HTML
|
||||
|
||||
|
||||
@app.route("/teacher/class")
|
||||
def teacher_class():
|
||||
students = Student.query.all()
|
||||
|
||||
html = ""
|
||||
for s in students:
|
||||
html += f"""
|
||||
<div class='bg-blue-900 p-3 m-2 rounded'>
|
||||
{s.name} - {s.current_room} - {s.state}
|
||||
</div>
|
||||
"""
|
||||
|
||||
return html
|
||||
|
||||
|
||||
# --------------------------------------------------
|
||||
# INIT
|
||||
# --------------------------------------------------
|
||||
|
||||
if __name__ == "__main__":
|
||||
with app.app_context():
|
||||
db.create_all()
|
||||
app.run(host="0.0.0.0", port=5000, debug=True)
|
||||
Reference in New Issue
Block a user