271 lines
9.4 KiB
Python
271 lines
9.4 KiB
Python
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import click
|
|
from flask import Flask, jsonify, redirect, render_template, request, url_for
|
|
from flask_login import (
|
|
LoginManager,
|
|
current_user,
|
|
login_required,
|
|
login_user,
|
|
logout_user,
|
|
)
|
|
from sqlalchemy import func
|
|
|
|
from models import Sensor, SensorReadout, User, db
|
|
|
|
login_manager = LoginManager()
|
|
|
|
|
|
def create_app():
|
|
app = Flask(__name__)
|
|
|
|
db_path = os.path.join(os.path.dirname(__file__), "data.db")
|
|
app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"
|
|
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
|
app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY", "dev-secret-key-change-me")
|
|
|
|
db.init_app(app)
|
|
login_manager.init_app(app)
|
|
login_manager.login_view = "login"
|
|
|
|
@login_manager.user_loader
|
|
def load_user(user_id):
|
|
return db.session.get(User, int(user_id))
|
|
|
|
@login_manager.unauthorized_handler
|
|
def unauthorized():
|
|
# Return JSON for API calls instead of redirecting to the login page.
|
|
if request.path.startswith("/api/"):
|
|
return jsonify({"error": "authentication required"}), 401
|
|
next_page = request.path
|
|
login_url = url_for("login")
|
|
if next_page and next_page != login_url:
|
|
login_url += f"?next={next_page}"
|
|
return redirect(login_url)
|
|
|
|
def _is_safe_next(target: str | None) -> bool:
|
|
return bool(target) and target.startswith("/") and not target.startswith("//")
|
|
|
|
# --- Pages ---
|
|
|
|
@app.route("/")
|
|
def index():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for("home"))
|
|
return redirect(url_for("login"))
|
|
|
|
@app.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for("home"))
|
|
error = None
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip()
|
|
password = request.form.get("password", "")
|
|
user = User.query.filter_by(username=username).first()
|
|
if user and user.check_password(password):
|
|
login_user(user)
|
|
next_page = request.args.get("next") or request.form.get("next")
|
|
if not _is_safe_next(next_page):
|
|
next_page = url_for("home")
|
|
return redirect(next_page)
|
|
error = "Invalid username or password."
|
|
return render_template("login.html", error=error, next=request.args.get("next", ""))
|
|
|
|
@app.route("/logout")
|
|
@login_required
|
|
def logout():
|
|
logout_user()
|
|
return redirect(url_for("login"))
|
|
|
|
@app.route("/home")
|
|
@login_required
|
|
def home():
|
|
return render_template("home.html", user=current_user)
|
|
|
|
@app.route("/sensors")
|
|
@login_required
|
|
def sensors_page():
|
|
return render_template("sensors.html", user=current_user)
|
|
|
|
@app.route("/api/docs")
|
|
@login_required
|
|
def api_docs():
|
|
return render_template("docs.html", user=current_user)
|
|
|
|
# --- API: device endpoints (intentionally public) ---
|
|
# The sensor firmware has no browser session, but it needs the sensor
|
|
# list to resolve a name to an id before pushing readouts.
|
|
|
|
@app.route("/api/sensors")
|
|
def api_sensors():
|
|
sensors = Sensor.query.order_by(Sensor.name).all()
|
|
return jsonify([{"id": s.id, "name": s.name} for s in sensors])
|
|
|
|
# --- API: dashboard reads (login required, the browser session cookie is enough) ---
|
|
|
|
@app.route("/api/sensors/<int:sensor_id>/types")
|
|
@login_required
|
|
def api_sensor_types(sensor_id):
|
|
types = (
|
|
db.session.query(SensorReadout.type)
|
|
.filter_by(sensor_id=sensor_id)
|
|
.distinct()
|
|
.order_by(SensorReadout.type)
|
|
.all()
|
|
)
|
|
return jsonify([t[0] for t in types])
|
|
|
|
@app.route("/api/sensors/<int:sensor_id>/readouts")
|
|
@login_required
|
|
def api_sensor_readouts(sensor_id):
|
|
type_ = request.args.get("type")
|
|
query = SensorReadout.query.filter_by(sensor_id=sensor_id)
|
|
if type_:
|
|
query = query.filter_by(type=type_)
|
|
readouts = query.order_by(SensorReadout.timestamp.desc()).limit(300).all()
|
|
return jsonify(
|
|
[
|
|
{"value": r.value, "timestamp": r.timestamp.isoformat() + "Z"}
|
|
for r in readouts
|
|
]
|
|
)
|
|
|
|
@app.route("/api/sensors/<int:sensor_id>/readouts/hourly")
|
|
@login_required
|
|
def api_sensor_readouts_hourly(sensor_id):
|
|
type_ = request.args.get("type")
|
|
range_ = request.args.get("range", "all")
|
|
# Timestamps are stored as naive UTC, so the cutoff must be naive UTC too.
|
|
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
if range_ == "today":
|
|
cutoff = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
elif range_ == "week":
|
|
cutoff = now - timedelta(days=7)
|
|
elif range_ == "month":
|
|
cutoff = now - timedelta(days=30)
|
|
elif range_ == "all":
|
|
cutoff = None
|
|
else:
|
|
return jsonify({"error": "range must be one of: today, week, month, all"}), 400
|
|
query = (
|
|
db.session.query(
|
|
func.strftime("%Y-%m-%dT%H:00:00", SensorReadout.timestamp).label("hour"),
|
|
func.avg(SensorReadout.value).label("avg_value"),
|
|
)
|
|
.filter(SensorReadout.sensor_id == sensor_id)
|
|
)
|
|
if cutoff is not None:
|
|
query = query.filter(SensorReadout.timestamp >= cutoff)
|
|
if type_:
|
|
query = query.filter(SensorReadout.type == type_)
|
|
rows = (
|
|
query.group_by("hour")
|
|
.order_by("hour")
|
|
.all()
|
|
)
|
|
return jsonify(
|
|
[{"timestamp": r.hour, "value": round(r.avg_value, 2)} for r in rows]
|
|
)
|
|
|
|
# --- API: device writes (intentionally public — called by devices/sensors) ---
|
|
|
|
@app.route("/api/sensors", methods=["POST"])
|
|
def api_create_sensor():
|
|
data = request.get_json(silent=True)
|
|
if not data or "name" not in data:
|
|
return jsonify({"error": "name is required"}), 400
|
|
|
|
name = data["name"].strip()
|
|
if not name:
|
|
return jsonify({"error": "name cannot be empty"}), 400
|
|
|
|
if Sensor.query.filter_by(name=name).first():
|
|
return jsonify({"error": "sensor already exists"}), 409
|
|
|
|
sensor = Sensor(name=name)
|
|
db.session.add(sensor)
|
|
db.session.commit()
|
|
return jsonify({"id": sensor.id, "name": sensor.name}), 201
|
|
|
|
@app.route("/api/sensors/<int:sensor_id>/readouts", methods=["POST"])
|
|
def api_create_readout(sensor_id):
|
|
sensor = db.session.get(Sensor, sensor_id)
|
|
if not sensor:
|
|
return jsonify({"error": "sensor not found"}), 404
|
|
|
|
data = request.get_json(silent=True)
|
|
if not data:
|
|
return jsonify({"error": "request body is required"}), 400
|
|
|
|
missing = [f for f in ("type", "value") if f not in data]
|
|
if missing:
|
|
return jsonify({"error": f"missing fields: {', '.join(missing)}"}), 400
|
|
|
|
try:
|
|
value = float(data["value"])
|
|
except (TypeError, ValueError):
|
|
return jsonify({"error": "value must be a number"}), 400
|
|
|
|
readout = SensorReadout(
|
|
sensor_id=sensor_id,
|
|
type=str(data["type"]).strip(),
|
|
value=value,
|
|
)
|
|
db.session.add(readout)
|
|
db.session.commit()
|
|
return jsonify(
|
|
{
|
|
"id": readout.id,
|
|
"sensor_id": readout.sensor_id,
|
|
"type": readout.type,
|
|
"value": readout.value,
|
|
"timestamp": readout.timestamp.isoformat() + "Z",
|
|
}
|
|
), 201
|
|
|
|
# --- Command for creating the first admin ---
|
|
# Usage:
|
|
# flask --app app create-admin --username admin --password 'secret'
|
|
# ADMIN_USERNAME=admin ADMIN_PASSWORD=secret flask --app app create-admin
|
|
# flask --app app create-admin (asks interactively)
|
|
|
|
@app.cli.command("create-admin")
|
|
@click.option("--username", default=None, help="Administrator username")
|
|
@click.option("--password", default=None, help="Administrator password")
|
|
def create_admin(username, password):
|
|
"""Create or update an administrator account (is_admin=True)."""
|
|
username = username or os.environ.get("ADMIN_USERNAME")
|
|
if not username:
|
|
username = click.prompt("Admin username")
|
|
username = username.strip()
|
|
if not username:
|
|
raise click.ClickException("Username cannot be empty.")
|
|
|
|
password = password or os.environ.get("ADMIN_PASSWORD")
|
|
if not password:
|
|
password = click.prompt("Admin password", hide_input=True, confirmation_prompt=True)
|
|
if not password:
|
|
raise click.ClickException("Password cannot be empty.")
|
|
|
|
user = User.query.filter_by(username=username).first()
|
|
if user:
|
|
user.set_password(password)
|
|
user.is_admin = True
|
|
db.session.commit()
|
|
click.echo(f"Updated existing user '{username}' (granted is_admin=True).")
|
|
else:
|
|
user = User(username=username, is_admin=True)
|
|
user.set_password(password)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
click.echo(f"Created administrator '{username}'.")
|
|
|
|
return app
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app = create_app()
|
|
app.run(debug=True, host='0.0.0.0')
|