Adriano/TieMeasureFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f71bbf398e2fcb1ad929041f22901b39c5d018d9
TieMeasureFlow/server/routers/setup.py
T
Adriano 429c94da94 fix: improve SPC dashboard UX — hover toolbar, visible report errors 
- Change Plotly modebar to hover-only mode to avoid overlapping chart content
- Remove redundant Plotly titles (container headers already provide them)
- Add Content-Type check and inline error display for report downloads
- Fix setup login validation and seed idempotency for existing data

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-24 16:42:47 +01:00

629 lines
22 KiB
Python
Raw Blame History

"""Setup page router - database init, admin creation, demo seed.
Protected by SETUP_PASSWORD env var. When empty/None, all endpoints return 404.
NOT protected by API Key auth (standalone access).
"""
import random
import secrets
from datetime import datetime, timedelta
from pathlib import Path
from fastapi import APIRouter, HTTPException, Request, status
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
from sqlalchemy import inspect as sa_inspect, select
from config import settings
from database import Base, engine, async_session_factory
from models import (
User, Recipe, RecipeVersion, RecipeTask, RecipeSubtask, Measurement,
)
from services.auth_service import hash_password
from services.measurement_service import calculate_pass_fail
router = APIRouter(prefix="/api/setup", tags=["setup"])
_templates_dir = Path(__file__).resolve().parent.parent / "templates"
templates = Jinja2Templates(directory=str(_templates_dir))
# ---------------------------------------------------------------------------
# Request schemas
# ---------------------------------------------------------------------------
class PasswordBody(BaseModel):
password: str
class CreateAdminBody(BaseModel):
password: str
admin_username: str
admin_password: str
admin_email: str
admin_display_name: str
class ResetDbBody(BaseModel):
password: str
confirm: bool
class ManageUserBody(BaseModel):
password: str # setup password
user_id: int | None = None # None = create, int = update
username: str
user_password: str | None = None # Required for create, optional for update
display_name: str
email: str | None = None
roles: list[str] = []
is_admin: bool = False
class ChangeUserPasswordBody(BaseModel):
password: str # setup password
user_id: int
new_password: str
class ToggleUserActiveBody(BaseModel):
password: str # setup password
user_id: int
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _check_setup_enabled() -> None:
"""Raise 404 if setup password is not configured."""
if not settings.setup_password:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
def _check_password(password: str) -> None:
"""Verify the provided password matches the setup password."""
_check_setup_enabled()
if password != settings.setup_password:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid setup password",
)
def _generate_measurements_for_subtask(
subtask: RecipeSubtask,
version_id: int,
measured_by: int,
lot_numbers: list[str],
serial_numbers: list[str],
base_date: datetime,
count: int = 30,
) -> list[Measurement]:
"""Generate realistic demo measurements with Gaussian distribution."""
nominal = float(subtask.nominal)
uwl = float(subtask.uwl)
ltl = float(subtask.ltl)
utl = float(subtask.utl)
lwl = float(subtask.lwl)
sigma = (uwl - nominal) / 2.5
measurements = []
for i in range(count):
# ~80% pass, ~13% warning, ~7% fail
r = random.random()
if r < 0.07:
# Fail: outside tolerance limits
if random.random() < 0.5:
value = random.uniform(utl, utl + (utl - nominal))
else:
value = random.uniform(ltl - (nominal - ltl), ltl)
elif r < 0.20:
# Warning: between warning and tolerance
if random.random() < 0.5:
value = random.uniform(uwl, utl)
else:
value = random.uniform(ltl, lwl)
else:
# Pass: Gaussian around nominal
value = random.gauss(nominal, sigma * 0.7)
# Clamp to within warning limits
value = max(lwl, min(uwl, value))
value = round(value, 6)
pass_fail, deviation = calculate_pass_fail(value, subtask)
# Distribute across time, lots, serials
days_offset = random.uniform(0, 30)
hours_offset = random.uniform(8, 17) # working hours
measured_at = base_date - timedelta(days=days_offset) + timedelta(hours=hours_offset)
lot = lot_numbers[i % len(lot_numbers)]
serial = serial_numbers[i % len(serial_numbers)]
input_method = "usb_caliper" if random.random() < 0.7 else "manual"
m = Measurement(
subtask_id=subtask.id,
version_id=version_id,
measured_by=measured_by,
value=value,
pass_fail=pass_fail,
deviation=deviation,
lot_number=lot,
serial_number=serial,
input_method=input_method,
measured_at=measured_at,
)
measurements.append(m)
return measurements
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@router.get("", response_class=HTMLResponse)
async def setup_page(request: Request):
"""Serve the setup page HTML."""
_check_setup_enabled()
return templates.TemplateResponse("setup/setup.html", {"request": request})
@router.get("/status")
async def setup_status():
"""Return database table status (no password required, but setup must be enabled)."""
_check_setup_enabled()
async with engine.connect() as conn:
table_names: list[str] = await conn.run_sync(
lambda sync_conn: sa_inspect(sync_conn).get_table_names()
)
return {
"tables_exist": len(table_names) > 0,
"table_count": len(table_names),
"tables": table_names,
}
@router.post("/init-db")
async def init_db(body: PasswordBody):
"""Create all database tables."""
_check_password(body.password)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
return {"status": "ok", "message": "Database tables created successfully"}
@router.post("/create-admin")
async def create_admin(body: CreateAdminBody):
"""Create an admin user with all roles."""
_check_password(body.password)
api_key = secrets.token_hex(32)
async with async_session_factory() as session:
async with session.begin():
user = User(
username=body.admin_username,
password_hash=hash_password(body.admin_password),
email=body.admin_email,
display_name=body.admin_display_name,
roles=["Maker", "MeasurementTec", "Metrologist"],
is_admin=True,
api_key=api_key,
)
session.add(user)
return {
"status": "ok",
"message": f"Admin user '{body.admin_username}' created",
"api_key": api_key,
}
@router.post("/seed")
async def seed_demo_data(body: PasswordBody):
"""Seed database with demo users and a sample recipe."""
_check_password(body.password)
async with async_session_factory() as session:
async with session.begin():
# ---- Users --------------------------------------------------------
users_data = [
{
"username": "admin",
"password": "admin123",
"display_name": "Administrator",
"email": "admin@tiemeasureflow.local",
"roles": ["Maker", "MeasurementTec", "Metrologist"],
"is_admin": True,
},
{
"username": "maker1",
"password": "maker123",
"display_name": "Mario Rossi",
"email": "maker1@tiemeasureflow.local",
"roles": ["Maker"],
"is_admin": False,
},
{
"username": "tec1",
"password": "tec123",
"display_name": "Luca Bianchi",
"email": "tec1@tiemeasureflow.local",
"roles": ["MeasurementTec"],
"is_admin": False,
},
{
"username": "metro1",
"password": "metro123",
"display_name": "Anna Verdi",
"email": "metro1@tiemeasureflow.local",
"roles": ["Metrologist"],
"is_admin": False,
},
]
created_users: list[User] = []
for u in users_data:
# Skip users that already exist
existing = await session.execute(
select(User).where(User.username == u["username"])
)
existing_user = existing.scalar_one_or_none()
if existing_user:
created_users.append(existing_user)
continue
user = User(
username=u["username"],
password_hash=hash_password(u["password"]),
display_name=u["display_name"],
email=u["email"],
roles=u["roles"],
is_admin=u["is_admin"],
api_key=secrets.token_hex(32),
)
session.add(user)
created_users.append(user)
# Flush to get auto-generated user IDs
await session.flush()
admin_user = created_users[0]
# ---- Recipe -------------------------------------------------------
# Skip if recipe already exists
existing_recipe = await session.execute(
select(Recipe).where(Recipe.code == "DEMO-001")
)
if existing_recipe.scalar_one_or_none():
return {
"status": "ok",
"message": "Demo data already exists, skipped",
"users_created": [u["username"] for u in users_data],
"recipe_created": "DEMO-001",
"measurements_created": 0,
}
recipe = Recipe(
code="DEMO-001",
name="Demo Measurement Recipe",
description="Sample recipe with shaft and surface measurements for demonstration purposes.",
created_by=admin_user.id,
)
session.add(recipe)
await session.flush()
version = RecipeVersion(
recipe_id=recipe.id,
version_number=1,
is_current=True,
created_by=admin_user.id,
change_notes="Initial demo version",
)
session.add(version)
await session.flush()
# ---- Task 1: Shaft Diameter ----------------------------------------
task1 = RecipeTask(
version_id=version.id,
order_index=0,
title="Shaft Diameter Measurement",
directive="Measure the main shaft diameters at the indicated positions.",
description="Use the caliper to measure each diameter. Ensure the shaft is clean and at room temperature.",
)
session.add(task1)
await session.flush()
subtasks_1 = [
RecipeSubtask(
task_id=task1.id,
marker_number=1,
description="D1 - Main shaft diameter",
measurement_type="diameter",
nominal=25.000,
utl=25.050,
uwl=25.030,
lwl=24.970,
ltl=24.950,
unit="mm",
),
RecipeSubtask(
task_id=task1.id,
marker_number=2,
description="D2 - Bearing seat diameter",
measurement_type="diameter",
nominal=30.000,
utl=30.021,
uwl=30.015,
lwl=29.985,
ltl=29.979,
unit="mm",
),
RecipeSubtask(
task_id=task1.id,
marker_number=3,
description="D3 - Shoulder diameter",
measurement_type="diameter",
nominal=35.000,
utl=35.039,
uwl=35.025,
lwl=34.975,
ltl=34.961,
unit="mm",
),
]
session.add_all(subtasks_1)
# ---- Task 2: Surface Flatness --------------------------------------
task2 = RecipeTask(
version_id=version.id,
order_index=1,
title="Surface Flatness Measurement",
directive="Measure surface heights at the marked reference points.",
description="Place the part on the granite surface plate. Zero the indicator at the reference point, then measure each position.",
)
session.add(task2)
await session.flush()
subtasks_2 = [
RecipeSubtask(
task_id=task2.id,
marker_number=1,
description="F1 - Center reference height",
measurement_type="height",
nominal=0.000,
utl=0.020,
uwl=0.010,
lwl=-0.010,
ltl=-0.020,
unit="mm",
),
RecipeSubtask(
task_id=task2.id,
marker_number=2,
description="F2 - Edge height (left)",
measurement_type="height",
nominal=0.000,
utl=0.030,
uwl=0.015,
lwl=-0.015,
ltl=-0.030,
unit="mm",
),
RecipeSubtask(
task_id=task2.id,
marker_number=3,
description="F3 - Edge height (right)",
measurement_type="height",
nominal=0.000,
utl=0.030,
uwl=0.015,
lwl=-0.015,
ltl=-0.030,
unit="mm",
),
]
session.add_all(subtasks_2)
# ---- Measurements -------------------------------------------------
await session.flush() # Get subtask IDs
random.seed(42) # Reproducible demo data
tec_user = created_users[2] # tec1
lot_numbers = ["LOT-2025-001", "LOT-2025-002", "LOT-2025-003", "LOT-2025-004"]
serial_numbers = [f"SN-{i:04d}" for i in range(1, 13)]
base_date = datetime.now()
all_subtasks = subtasks_1 + subtasks_2
measurements_created = 0
for st in all_subtasks:
ms = _generate_measurements_for_subtask(
subtask=st,
version_id=version.id,
measured_by=tec_user.id,
lot_numbers=lot_numbers,
serial_numbers=serial_numbers,
base_date=base_date,
count=30,
)
session.add_all(ms)
measurements_created += len(ms)
return {
"status": "ok",
"message": "Demo data seeded successfully",
"users_created": [u["username"] for u in users_data],
"recipe_created": "DEMO-001",
"measurements_created": measurements_created,
}
@router.post("/list-users")
async def list_users(body: PasswordBody):
"""List all users."""
_check_password(body.password)
async with async_session_factory() as session:
result = await session.execute(select(User).order_by(User.id))
users = result.scalars().all()
return {
"status": "ok",
"users": [
{
"id": u.id,
"username": u.username,
"display_name": u.display_name,
"email": u.email,
"roles": u.roles or [],
"is_admin": u.is_admin,
"active": u.active,
"created_at": u.created_at.isoformat() if u.created_at else None,
"last_login": u.last_login.isoformat() if u.last_login else None,
}
for u in users
],
}
@router.post("/manage-user")
async def manage_user(body: ManageUserBody):
"""Create or update a user. user_id=null creates new, user_id=int updates."""
_check_password(body.password)
VALID_ROLES = {"Maker", "MeasurementTec", "Metrologist"}
invalid = set(body.roles) - VALID_ROLES
if invalid:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid roles: {', '.join(invalid)}",
)
async with async_session_factory() as session:
async with session.begin():
if body.user_id is None:
# Create
if not body.user_password:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Password is required when creating a new user",
)
# Check username uniqueness
existing = await session.execute(
select(User).where(User.username == body.username)
)
if existing.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Username '{body.username}' already exists",
)
user = User(
username=body.username,
password_hash=hash_password(body.user_password),
display_name=body.display_name,
email=body.email,
roles=body.roles,
is_admin=body.is_admin,
api_key=secrets.token_hex(32),
)
session.add(user)
await session.flush()
return {"status": "ok", "message": f"User '{body.username}' created", "user_id": user.id}
else:
# Update
result = await session.execute(
select(User).where(User.id == body.user_id)
)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"User ID {body.user_id} not found",
)
# Check username uniqueness (exclude self)
existing = await session.execute(
select(User).where(
User.username == body.username, User.id != body.user_id
)
)
if existing.scalar_one_or_none():
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Username '{body.username}' already taken",
)
user.username = body.username
user.display_name = body.display_name
user.email = body.email
user.roles = body.roles
user.is_admin = body.is_admin
if body.user_password:
user.password_hash = hash_password(body.user_password)
return {"status": "ok", "message": f"User '{body.username}' updated", "user_id": user.id}
@router.post("/change-user-password")
async def change_user_password(body: ChangeUserPasswordBody):
"""Change a user's password and invalidate their API key."""
_check_password(body.password)
async with async_session_factory() as session:
async with session.begin():
result = await session.execute(
select(User).where(User.id == body.user_id)
)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"User ID {body.user_id} not found",
)
user.password_hash = hash_password(body.new_password)
user.api_key = None # Invalidate sessions
return {"status": "ok", "message": f"Password changed for '{user.username}'"}
@router.post("/toggle-user-active")
async def toggle_user_active(body: ToggleUserActiveBody):
"""Toggle a user's active status. Invalidates API key when deactivated."""
_check_password(body.password)
async with async_session_factory() as session:
async with session.begin():
result = await session.execute(
select(User).where(User.id == body.user_id)
)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"User ID {body.user_id} not found",
)
user.active = not user.active
if not user.active:
user.api_key = None # Invalidate sessions on deactivation
new_status = "active" if user.active else "inactive"
return {"status": "ok", "message": f"User '{user.username}' is now {new_status}", "active": user.active}
@router.post("/reset-db")
async def reset_db(body: ResetDbBody):
"""Drop all tables and recreate them. Requires confirm=true."""
_check_password(body.password)
if not body.confirm:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="You must set confirm=true to reset the database",
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
return {"status": "ok", "message": "Database reset successfully"}
Reference in New Issue View Git Blame Copy Permalink