143 lines
3.8 KiB
Python
143 lines
3.8 KiB
Python
"""Flask application factory and DB setup."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import os
|
|
import secrets
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, g
|
|
|
|
from physcom.db.schema import init_db
|
|
from physcom.db.repository import Repository
|
|
|
|
|
|
DEFAULT_DB = Path("data/physcom.db")
|
|
_ENV_FILE = Path(".env")
|
|
|
|
|
|
def _load_or_generate_secret_key() -> str:
|
|
"""Return FLASK_SECRET_KEY from env, .env file, or auto-generate and persist it."""
|
|
key = os.environ.get("FLASK_SECRET_KEY", "").strip()
|
|
if key:
|
|
return key
|
|
|
|
# Try to load from .env
|
|
if _ENV_FILE.exists():
|
|
for line in _ENV_FILE.read_text().splitlines():
|
|
line = line.strip()
|
|
if line.startswith("FLASK_SECRET_KEY="):
|
|
key = line[len("FLASK_SECRET_KEY="):].strip()
|
|
if key:
|
|
return key
|
|
|
|
# Generate, persist, and return
|
|
key = secrets.token_hex(32)
|
|
with _ENV_FILE.open("a") as f:
|
|
f.write(f"FLASK_SECRET_KEY={key}\n")
|
|
return key
|
|
|
|
|
|
def get_repo() -> Repository:
|
|
"""Return a Repository scoped to the current request."""
|
|
if "repo" not in g:
|
|
db_path = Path(os.environ.get("PHYSCOM_DB", str(DEFAULT_DB)))
|
|
conn = init_db(db_path)
|
|
g.repo = Repository(conn)
|
|
return g.repo
|
|
|
|
|
|
def close_db(exc: BaseException | None = None) -> None:
|
|
repo: Repository | None = g.pop("repo", None)
|
|
if repo is not None:
|
|
repo.conn.close()
|
|
|
|
|
|
_SI_PREFIXES = [
|
|
(1e12, "T"),
|
|
(1e9, "G"),
|
|
(1e6, "M"),
|
|
(1e3, "k"),
|
|
]
|
|
|
|
|
|
def _si_format(value: object) -> str:
|
|
"""Format a number with SI prefixes for readability.
|
|
|
|
Handles string inputs (like dep.value) by trying float conversion first.
|
|
Non-numeric values are returned as-is.
|
|
"""
|
|
if isinstance(value, str):
|
|
try:
|
|
num = float(value)
|
|
except (ValueError, TypeError):
|
|
return value
|
|
elif isinstance(value, (int, float)):
|
|
num = float(value)
|
|
else:
|
|
return str(value)
|
|
|
|
if math.isnan(num) or math.isinf(num):
|
|
return str(value)
|
|
|
|
abs_num = abs(num)
|
|
if abs_num < 1000:
|
|
# Small numbers: drop trailing zeros, cap at 4 significant figures
|
|
if num == int(num) and abs_num < 100:
|
|
return str(int(num))
|
|
return f"{num:.4g}"
|
|
|
|
for threshold, prefix in _SI_PREFIXES:
|
|
if abs_num >= threshold:
|
|
scaled = num / threshold
|
|
return f"{scaled:.4g}{prefix}"
|
|
|
|
return f"{num:.4g}"
|
|
|
|
|
|
def create_app() -> Flask:
|
|
app = Flask(__name__)
|
|
app.secret_key = _load_or_generate_secret_key()
|
|
|
|
app.jinja_env.filters["si"] = _si_format
|
|
|
|
app.teardown_appcontext(close_db)
|
|
|
|
# Register blueprints
|
|
from physcom_web.routes.entities import bp as entities_bp
|
|
from physcom_web.routes.domains import bp as domains_bp
|
|
from physcom_web.routes.pipeline import bp as pipeline_bp
|
|
from physcom_web.routes.results import bp as results_bp
|
|
from physcom_web.routes.admin import bp as admin_bp
|
|
|
|
app.register_blueprint(entities_bp)
|
|
app.register_blueprint(domains_bp)
|
|
app.register_blueprint(pipeline_bp)
|
|
app.register_blueprint(results_bp)
|
|
app.register_blueprint(admin_bp)
|
|
|
|
@app.route("/")
|
|
def index():
|
|
from flask import render_template
|
|
repo = get_repo()
|
|
entities = repo.list_entities()
|
|
dims = {e.dimension for e in entities}
|
|
domains = repo.list_domains()
|
|
status_counts = repo.count_combinations_by_status()
|
|
stats = {
|
|
"entities": len(entities),
|
|
"dimensions": len(dims),
|
|
"domains": len(domains),
|
|
"combinations": sum(status_counts.values()),
|
|
}
|
|
return render_template("home.html", stats=stats)
|
|
|
|
return app
|
|
|
|
|
|
def run() -> None:
|
|
"""Entry point for `physcom-web` script."""
|
|
app = create_app()
|
|
app.run(host="0.0.0.0", debug=True, port=int(os.environ.get("PORT", "5000")))
|