From db5d686938c26d2079524996899407f7a312d5e7 Mon Sep 17 00:00:00 2001 From: "Adriano (daitarn)" Date: Wed, 7 Oct 2026 12:25:16 +0200 Subject: [PATCH] Monitor stazioni: GET /api/production-runs/open (#57) Co-Authored-By: Claude Sonnet 5.5 --- src/backend/api/routers/production.py | 24 ++++- src/backend/models/api/monitor.py | 49 +++++++++ src/backend/services/monitor_service.py | 133 ++++++++++++++++++++++++ src/backend/tests/test_monitor.py | 83 +++++++++++++++ 4 files changed, 288 insertions(+), 1 deletion(-) create mode 100644 src/backend/models/api/monitor.py create mode 100644 src/backend/services/monitor_service.py create mode 100644 src/backend/tests/test_monitor.py diff --git a/src/backend/api/routers/production.py b/src/backend/api/routers/production.py index 9fdb5f3..2a4f8c6 100644 --- a/src/backend/api/routers/production.py +++ b/src/backend/api/routers/production.py @@ -20,7 +20,8 @@ from src.backend.models.api.production import ( ) from src.backend.models.orm.production import ProductionRun from src.backend.models.orm.user import User -from src.backend.services import auth_service, production_service +from src.backend.models.api.monitor import MonitorRun +from src.backend.services import auth_service, monitor_service, production_service router = APIRouter(prefix="/api/production-runs", tags=["production"]) @@ -81,6 +82,27 @@ async def get_current_production_run( return await _as_response(db, run) if run is not None else None +async def _require_high_authority(user: User = Depends(get_current_user)) -> User: + """Monitor of all stations (#57): admin or Supervisor (capoturno) only.""" + if user.is_admin or user.has_role("Supervisor"): + return user + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, detail="Supervisor role required", + ) + + +@router.get("/open", response_model=list[MonitorRun]) +async def list_open_production_runs( + user: User = Depends(_require_high_authority), + db: AsyncSession = Depends(get_db), +): + """Every open run with timer, last event and recent measurements (#57). + + Declared before /{run_id} so "open" is not parsed as a run id. + """ + return await monitor_service.list_open_runs(db) + + @router.get("/{run_id}", response_model=ProductionRunWithEventsResponse) async def get_production_run( run_id: int, diff --git a/src/backend/models/api/monitor.py b/src/backend/models/api/monitor.py new file mode 100644 index 0000000..bca53bb --- /dev/null +++ b/src/backend/models/api/monitor.py @@ -0,0 +1,49 @@ +"""API models for the live station monitor (#57).""" +from datetime import datetime +from typing import Optional + +from pydantic import BaseModel + + +class MonitorSubtask(BaseModel): + """Mean of the recent measurements of one quote, against its tolerance.""" + + subtask_id: int + description: str + unit: str + count: int + mean: float + ltl: Optional[float] = None + utl: Optional[float] = None + # True when ltl <= mean <= utl (a missing limit does not constrain). + in_tolerance: bool + + +class MonitorEvent(BaseModel): + event_type: str + created_at: datetime + note: Optional[str] = None + + +class MonitorRun(BaseModel): + id: int + station_code: str + station_name: str + recipe_code: str + recipe_name: str + version_number: int + operator: str + status: str + started_at: datetime + duration_seconds: int + cycle_count: int + seconds_to_next_measurement: Optional[int] = None + overdue: bool + last_event: Optional[MonitorEvent] = None + # Counts over the last MONITOR_WINDOW measurements of the run. + measured: int + pass_count: int + warning_count: int + fail_count: int + subtasks: list[MonitorSubtask] + server_time: datetime diff --git a/src/backend/services/monitor_service.py b/src/backend/services/monitor_service.py new file mode 100644 index 0000000..81533b0 --- /dev/null +++ b/src/backend/services/monitor_service.py @@ -0,0 +1,133 @@ +"""Live monitor of the open production runs (#57). + +One query per table, whatever the number of runs: measurements and last events are +ranked per run with a window function and cut in the database, not in a loop. +""" +from collections import defaultdict + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from src.backend.models.api.monitor import MonitorEvent, MonitorRun, MonitorSubtask +from src.backend.models.orm.measurement import Measurement +from src.backend.models.orm.production import ProductionEvent, ProductionRun +from src.backend.models.orm.task import RecipeSubtask +from src.backend.models.orm.user import User +from src.backend.services import production_service + +# Recent measurements per run that feed counts and means (#57: last N). +MONITOR_WINDOW = 20 + + +def _in_tolerance(mean: float, ltl, utl) -> bool: + # A missing limit does not constrain. + if ltl is not None and mean < float(ltl): + return False + if utl is not None and mean > float(utl): + return False + return True + + +async def list_open_runs(db: AsyncSession) -> list[MonitorRun]: + runs = (await db.execute( + select(ProductionRun) + .where(ProductionRun.status.in_(("running", "paused"))) + .order_by(ProductionRun.started_at) + )).scalars().all() + if not runs: + return [] + run_ids = [r.id for r in runs] + + # Operators: one query. + users = { + u.id: u for u in (await db.execute( + select(User).where(User.id.in_({r.operator_id for r in runs})) + )).scalars().all() + } + + # Last event per run: rank desc inside the run, keep rank 1. + ev_rank = func.row_number().over( + partition_by=ProductionEvent.run_id, + order_by=(ProductionEvent.created_at.desc(), ProductionEvent.id.desc()), + ).label("rn") + ev_sub = select( + ProductionEvent.run_id.label("run_id"), + ProductionEvent.event_type.label("event_type"), + ProductionEvent.created_at.label("created_at"), + ProductionEvent.note.label("note"), + ev_rank, + ).where(ProductionEvent.run_id.in_(run_ids)).subquery() + ev_rows = (await db.execute( + select(ev_sub.c.run_id, ev_sub.c.event_type, ev_sub.c.created_at, ev_sub.c.note) + .where(ev_sub.c.rn == 1) + )).all() + last_event = {r.run_id: r for r in ev_rows} + + # Last N measurements per run, joined to their subtask for the limits. + m_rank = func.row_number().over( + partition_by=Measurement.production_run_id, + order_by=(Measurement.measured_at.desc(), Measurement.id.desc()), + ).label("rn") + m_sub = select( + Measurement.production_run_id.label("run_id"), + Measurement.subtask_id.label("subtask_id"), + Measurement.value.label("value"), + Measurement.pass_fail.label("pass_fail"), + m_rank, + ).where(Measurement.production_run_id.in_(run_ids)).subquery() + m_rows = (await db.execute( + select( + m_sub.c.run_id, m_sub.c.subtask_id, m_sub.c.value, m_sub.c.pass_fail, + RecipeSubtask.description, RecipeSubtask.unit, + RecipeSubtask.ltl, RecipeSubtask.utl, + ) + .join(RecipeSubtask, RecipeSubtask.id == m_sub.c.subtask_id) + .where(m_sub.c.rn <= MONITOR_WINDOW) + )).all() + by_run = defaultdict(list) + for row in m_rows: + by_run[row.run_id].append(row) + + result = [] + for run in runs: + rows = by_run.get(run.id, []) + per_sub: dict[int, list] = defaultdict(list) + for row in rows: + per_sub[row.subtask_id].append(row) + subtasks = [] + for sid, group in sorted(per_sub.items()): + mean = sum(float(g.value) for g in group) / len(group) + first = group[0] + subtasks.append(MonitorSubtask( + subtask_id=sid, description=first.description, unit=first.unit, + count=len(group), mean=round(mean, 6), + ltl=float(first.ltl) if first.ltl is not None else None, + utl=float(first.utl) if first.utl is not None else None, + in_tolerance=_in_tolerance(mean, first.ltl, first.utl), + )) + timer = production_service.describe(run) + ev = last_event.get(run.id) + operator = users.get(run.operator_id) + result.append(MonitorRun( + id=run.id, + station_code=run.station.code, station_name=run.station.name, + recipe_code=run.recipe.code, recipe_name=run.recipe.name, + version_number=run.version.version_number, + operator=(operator.display_name or operator.username) if operator else "", + status=run.status, + started_at=run.started_at, + duration_seconds=max(0, int((timer["server_time"] - run.started_at).total_seconds())), + cycle_count=run.cycle_count, + seconds_to_next_measurement=timer["seconds_to_next_measurement"], + overdue=timer["overdue"], + last_event=MonitorEvent( + event_type=ev.event_type, created_at=ev.created_at, note=ev.note, + ) if ev else None, + measured=len(rows), + pass_count=sum(1 for g in rows if g.pass_fail == "pass"), + warning_count=sum(1 for g in rows if g.pass_fail == "warning"), + fail_count=sum(1 for g in rows if g.pass_fail == "fail"), + subtasks=subtasks, + server_time=timer["server_time"], + )) + return result diff --git a/src/backend/tests/test_monitor.py b/src/backend/tests/test_monitor.py new file mode 100644 index 0000000..2d63e52 --- /dev/null +++ b/src/backend/tests/test_monitor.py @@ -0,0 +1,83 @@ +"""GET /api/production-runs/open - live monitor of the open runs (#57).""" +from datetime import datetime, timedelta + +from sqlalchemy import select + +from src.backend.models.orm.measurement import Measurement +from src.backend.models.orm.production import ProductionEvent, ProductionRun +from src.backend.models.orm.station import Station +from src.backend.models.orm.task import RecipeSubtask, RecipeTask +from src.backend.tests.conftest import _create_user, auth_headers, create_test_recipe + + +async def _run(db, user, code, status="running", overdue=False, open_=True): + """A station plus a run on a fresh recipe (limits ltl 9.5 / utl 10.5).""" + station = Station(code=f"ST-{code}", name=f"Linea {code}", active=True, created_by=user.id) + db.add(station) + recipe = await create_test_recipe(db, user_id=user.id, code=f"REC-{code}") + await db.flush() + version = recipe.versions[0] + now = datetime.utcnow() + run = ProductionRun( + station_id=station.id, recipe_id=recipe.id, version_id=version.id, + operator_id=user.id, status=status if open_ else "closed", + started_at=now - timedelta(hours=1), + measurement_interval_minutes=30, + next_measurement_at=now + (timedelta(minutes=-5) if overdue else timedelta(minutes=20)), + cycle_count=3, + active_station_id=station.id if open_ else None, + ) + db.add(run) + await db.flush() + db.add(ProductionEvent(run_id=run.id, event_type="start", user_id=user.id)) + sub = (await db.execute( + select(RecipeSubtask).join(RecipeTask, RecipeTask.id == RecipeSubtask.task_id) + .where(RecipeTask.version_id == version.id) + )).scalars().first() + return run, sub, version + + +def _m(run, sub, version, user, value, pf, minutes=0): + return Measurement( + subtask_id=sub.id, version_id=version.id, measured_by=user.id, value=value, + pass_fail=pf, production_run_id=run.id, + measured_at=datetime.utcnow() - timedelta(minutes=minutes), + ) + + +async def test_open_requires_high_authority(client, db_session, measurement_tec_user): + resp = await client.get("/api/production-runs/open", headers=auth_headers(measurement_tec_user)) + assert resp.status_code == 403 + + +async def test_open_lists_only_open_runs_with_means(client, db_session, admin_user): + sup = await _create_user(db_session, "capo", roles=["Supervisor"]) + r1, s1, v1 = await _run(db_session, admin_user, "A", overdue=True) + r2, s2, v2 = await _run(db_session, admin_user, "B", status="paused") + await _run(db_session, admin_user, "C", open_=False) + # Run A: mean 10.0 (inside), run B: mean 10.8 (outside, one fail). + for i, (val, pf) in enumerate([(9.9, "pass"), (10.1, "pass")]): + db_session.add(_m(r1, s1, v1, admin_user, val, pf, i)) + for i, (val, pf) in enumerate([(10.6, "fail"), (11.0, "fail"), (10.8, "warning")]): + db_session.add(_m(r2, s2, v2, admin_user, val, pf, i)) + await db_session.commit() + + resp = await client.get("/api/production-runs/open", headers=auth_headers(sup)) + assert resp.status_code == 200 + body = {r["station_code"]: r for r in resp.json()} + assert set(body) == {"ST-A", "ST-B"} # closed run C is absent + a, b = body["ST-A"], body["ST-B"] + assert a["overdue"] is True and a["seconds_to_next_measurement"] < 0 + assert b["overdue"] is False and b["status"] == "paused" + assert a["duration_seconds"] >= 3600 and a["cycle_count"] == 3 + assert a["last_event"]["event_type"] == "start" + assert a["pass_count"] == 2 and a["fail_count"] == 0 + assert a["subtasks"][0]["mean"] == 10.0 and a["subtasks"][0]["in_tolerance"] is True + assert b["fail_count"] == 2 and b["warning_count"] == 1 + assert round(b["subtasks"][0]["mean"], 3) == 10.8 + assert b["subtasks"][0]["in_tolerance"] is False + + +async def test_open_empty_for_admin(client, admin_user): + resp = await client.get("/api/production-runs/open", headers=auth_headers(admin_user)) + assert resp.status_code == 200 and resp.json() == []