482 lines
17 KiB
Python
482 lines
17 KiB
Python
"""Business logic for production runs.
|
|
|
|
Routers must call into these functions rather than manipulating models directly.
|
|
All functions are async and accept an AsyncSession; they flush but do NOT commit
|
|
(commit is handled by the FastAPI get_db dependency).
|
|
|
|
Clock convention
|
|
----------------
|
|
Every timestamp this module writes comes from ``_now()``, so the deadline and the
|
|
comparison against it are always on the same clock. Clients are handed
|
|
``seconds_to_next_measurement`` already computed rather than the raw deadline: a
|
|
naive datetime crossing the wire would be parsed in the browser's own timezone, and
|
|
the countdown would be off by the UTC offset. The number is the contract; the
|
|
timestamp is only there for the audit trail.
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
from src.backend.utils.tempo import utcnow
|
|
from typing import Optional
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.backend.models.api.production import ProductionRunCreate
|
|
from src.backend.models.orm.production import ProductionEvent, ProductionRun
|
|
from src.backend.models.orm.recipe import Recipe, RecipeVersion
|
|
from src.backend.models.orm.station import Station, StationRecipeAssignment
|
|
from src.backend.models.orm.task import MEASURING_TASK_TYPES, RecipeTask
|
|
from src.backend.models.orm.user import User
|
|
from src.backend.services import (
|
|
auth_service, measurement_service, production_export_service,
|
|
)
|
|
|
|
|
|
def _now() -> datetime:
|
|
"""Single source of time for runs.
|
|
|
|
Naive UTC, matching what the database writes for ``func.now()`` defaults,
|
|
so a run's started_at and its deadline are comparable.
|
|
"""
|
|
return utcnow()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Derived view
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def seconds_to_next_measurement(run: ProductionRun, now: Optional[datetime] = None) -> Optional[int]:
|
|
"""Seconds until the next measurement is due; negative once it is overdue.
|
|
|
|
Letting the value go negative is deliberate: the operator has to see how long
|
|
the line has been past the interval, not merely that it elapsed.
|
|
While the run is paused the countdown is frozen at what was left when the line
|
|
stopped - a fermo linea must not eat into the measurement interval.
|
|
"""
|
|
if run.next_measurement_at is None:
|
|
return None
|
|
reference = run.paused_at if (run.status == "paused" and run.paused_at) else (now or _now())
|
|
return int(round((run.next_measurement_at - reference).total_seconds()))
|
|
|
|
|
|
def describe(run: ProductionRun) -> dict:
|
|
"""Fields the API adds on top of the stored columns."""
|
|
now = _now()
|
|
remaining = seconds_to_next_measurement(run, now)
|
|
return {
|
|
"seconds_to_next_measurement": remaining,
|
|
"overdue": remaining is not None and remaining < 0,
|
|
"server_time": now,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lookups
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def get_run(db: AsyncSession, run_id: int) -> ProductionRun:
|
|
result = await db.execute(select(ProductionRun).where(ProductionRun.id == run_id))
|
|
run = result.scalar_one_or_none()
|
|
if run is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Production run not found",
|
|
)
|
|
return run
|
|
|
|
|
|
async def get_open_run_for_station(
|
|
db: AsyncSession, station_id: int,
|
|
) -> Optional[ProductionRun]:
|
|
"""The run currently open at a station, running or paused."""
|
|
result = await db.execute(
|
|
select(ProductionRun).where(ProductionRun.active_station_id == station_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_station_by_code(db: AsyncSession, code: str) -> Station:
|
|
result = await db.execute(select(Station).where(Station.code == code))
|
|
station = result.scalar_one_or_none()
|
|
if station is None or not station.active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Station '{code}' not found or inactive",
|
|
)
|
|
return station
|
|
|
|
|
|
async def _require_recipe_assigned_to_station(
|
|
db: AsyncSession, station_id: int, recipe_id: int,
|
|
) -> None:
|
|
"""A configured station only runs what it is assigned; an unconfigured one is free.
|
|
|
|
Mirrors station_service.list_station_recipes, the read side an operator's
|
|
recipe list is already filtered by: a station with at least one assignment
|
|
restricts to those recipes, one with none has not been set up to filter at
|
|
all and stays open to any active recipe.
|
|
"""
|
|
result = await db.execute(
|
|
select(StationRecipeAssignment.recipe_id).where(
|
|
StationRecipeAssignment.station_id == station_id,
|
|
)
|
|
)
|
|
assigned_ids = set(result.scalars().all())
|
|
if assigned_ids and recipe_id not in assigned_ids:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="This recipe is not assigned to this station",
|
|
)
|
|
|
|
|
|
async def measurement_task_ids(db: AsyncSession, version_id: int) -> list[int]:
|
|
"""Ids of the measurement tasks of a version, in the order they are executed.
|
|
|
|
This is the loop the operator lives in once production starts: the client uses
|
|
the first of these to bring the measurement back on screen when the interval
|
|
expires, and the last to know which task closes a cycle.
|
|
|
|
The type is read from the declared task_type (point 2), not from whether quotes
|
|
happen to be present: a measurement task with no quotes yet is still where the
|
|
operator has to be taken.
|
|
"""
|
|
result = await db.execute(
|
|
select(RecipeTask.id)
|
|
.where(
|
|
RecipeTask.version_id == version_id,
|
|
RecipeTask.task_type.in_(MEASURING_TASK_TYPES),
|
|
)
|
|
.order_by(RecipeTask.order_index, RecipeTask.id)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def list_run_events(db: AsyncSession, run_id: int) -> list[ProductionEvent]:
|
|
result = await db.execute(
|
|
select(ProductionEvent)
|
|
.where(ProductionEvent.run_id == run_id)
|
|
.order_by(ProductionEvent.created_at, ProductionEvent.id)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Supervisor authorisation
|
|
# The capoturno check lives in auth_service, next to the rest of the credential
|
|
# handling: a line stop, a closure and an out-of-tolerance value all ask the same
|
|
# question, and only one of the three is about production.
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _add_event(
|
|
db: AsyncSession,
|
|
run: ProductionRun,
|
|
event_type: str,
|
|
user: User,
|
|
supervisor: Optional[User] = None,
|
|
note: Optional[str] = None,
|
|
task_id: Optional[int] = None,
|
|
) -> ProductionEvent:
|
|
"""Append to the run's trace.
|
|
|
|
Added through the session rather than run.events: touching the collection would
|
|
trigger a lazy load, which raises under async SQLAlchemy.
|
|
"""
|
|
event = ProductionEvent(
|
|
run_id=run.id,
|
|
event_type=event_type,
|
|
user_id=user.id,
|
|
supervisor_id=supervisor.id if supervisor else None,
|
|
note=note,
|
|
task_id=task_id,
|
|
created_at=_now(),
|
|
)
|
|
db.add(event)
|
|
return event
|
|
|
|
|
|
async def open_run(
|
|
db: AsyncSession, data: ProductionRunCreate, operator: User,
|
|
) -> ProductionRun:
|
|
"""Start a production at a station.
|
|
|
|
Refuses if the station already has one open: two runs on the same station would
|
|
mean two timers and two histories for one physical line.
|
|
"""
|
|
station = await get_station_by_code(db, data.station_code)
|
|
|
|
existing = await get_open_run_for_station(db, station.id)
|
|
if existing is not None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=f"Station '{station.code}' already has an open production run",
|
|
)
|
|
|
|
recipe_row = await db.execute(select(Recipe).where(Recipe.id == data.recipe_id))
|
|
recipe = recipe_row.scalar_one_or_none()
|
|
if recipe is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Recipe not found",
|
|
)
|
|
if not recipe.active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Cannot start a production run with an inactive recipe",
|
|
)
|
|
await _require_recipe_assigned_to_station(db, station.id, recipe.id)
|
|
|
|
if data.version_id is not None:
|
|
version_row = await db.execute(
|
|
select(RecipeVersion).where(
|
|
RecipeVersion.id == data.version_id,
|
|
RecipeVersion.recipe_id == recipe.id,
|
|
)
|
|
)
|
|
version = version_row.scalar_one_or_none()
|
|
if version is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Recipe version not found for this recipe",
|
|
)
|
|
else:
|
|
version_row = await db.execute(
|
|
select(RecipeVersion).where(
|
|
RecipeVersion.recipe_id == recipe.id,
|
|
RecipeVersion.is_current == True, # noqa: E712
|
|
)
|
|
)
|
|
version = version_row.scalar_one_or_none()
|
|
if version is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Recipe has no current version",
|
|
)
|
|
|
|
# Traceability the recipe declares compulsory has to be there before the first
|
|
# measurement, not after it: a lot entered halfway through leaves the values
|
|
# taken up to that point unattributable.
|
|
if recipe.requires_lot and not (data.lot_number or "").strip():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
detail="This recipe requires a lot number to start a production",
|
|
)
|
|
if recipe.requires_serial and not (data.serial_number or "").strip():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
detail="This recipe requires a serial number to start a production",
|
|
)
|
|
|
|
now = _now()
|
|
# The interval is copied, not referenced: editing the recipe mid-production must
|
|
# not move the deadline of a run already under way.
|
|
interval = recipe.measurement_interval_minutes
|
|
run = ProductionRun(
|
|
station_id=station.id,
|
|
recipe_id=recipe.id,
|
|
version_id=version.id,
|
|
operator_id=operator.id,
|
|
lot_number=data.lot_number,
|
|
serial_number=data.serial_number,
|
|
status="running",
|
|
measurement_interval_minutes=interval,
|
|
next_measurement_at=(
|
|
now + timedelta(minutes=interval) if interval and interval > 0 else None
|
|
),
|
|
cycle_count=0,
|
|
started_at=now,
|
|
active_station_id=station.id,
|
|
)
|
|
db.add(run)
|
|
await db.flush()
|
|
_add_event(db, run, "start", operator)
|
|
await db.flush()
|
|
await db.refresh(run)
|
|
return run
|
|
|
|
|
|
def _require_open(run: ProductionRun) -> None:
|
|
if run.status == "closed":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Production run is already closed",
|
|
)
|
|
|
|
|
|
def _require_measurable(run: ProductionRun) -> None:
|
|
_require_open(run)
|
|
if run.status == "paused":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Production run is paused: resume it before measuring",
|
|
)
|
|
|
|
|
|
async def complete_cycle(
|
|
db: AsyncSession,
|
|
run: ProductionRun,
|
|
user: User,
|
|
task_id: Optional[int] = None,
|
|
note: Optional[str] = None,
|
|
) -> ProductionRun:
|
|
"""Record a finished measurement task, and restart the interval if it was the last.
|
|
|
|
A cycle spans every measurement task of the recipe, so only the last one closes
|
|
it. Restarting the interval at the first would make a recipe with three
|
|
measurement tasks measure three times as often as it was set up to.
|
|
|
|
An unrecognised task_id - or none at all - closes the cycle: a caller that does
|
|
not know about the sequence gets the behaviour that existed before it.
|
|
"""
|
|
_require_measurable(run)
|
|
|
|
# A cycle that contains a quote out of tolerance is not a finished cycle. This
|
|
# is the same gate as on the next measurement: closing the cycle would be
|
|
# another way of walking past it.
|
|
blocking = await measurement_service.pending_authorisation(
|
|
db, run.version_id, user.id, run.id,
|
|
)
|
|
if blocking is not None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=(
|
|
"A measurement is out of tolerance and awaiting the supervisor: "
|
|
"the cycle cannot be closed until it is authorised"
|
|
),
|
|
)
|
|
|
|
measure_ids = await measurement_task_ids(db, run.version_id)
|
|
closes_cycle = (
|
|
task_id is None or task_id not in measure_ids or task_id == measure_ids[-1]
|
|
)
|
|
|
|
if not closes_cycle:
|
|
_add_event(db, run, "task_measured", user, note=note, task_id=task_id)
|
|
await db.flush()
|
|
await db.refresh(run)
|
|
return run
|
|
|
|
now = _now()
|
|
run.cycle_count += 1
|
|
if run.measurement_interval_minutes and run.measurement_interval_minutes > 0:
|
|
# Measured from now, not from the previous deadline: a late measurement must
|
|
# not compress the following interval.
|
|
run.next_measurement_at = now + timedelta(
|
|
minutes=run.measurement_interval_minutes
|
|
)
|
|
_add_event(db, run, "cycle_completed", user, note=note, task_id=task_id)
|
|
await db.flush()
|
|
await db.refresh(run)
|
|
return run
|
|
|
|
|
|
async def remeasure(
|
|
db: AsyncSession,
|
|
run: ProductionRun,
|
|
user: User,
|
|
task_id: Optional[int] = None,
|
|
note: Optional[str] = None,
|
|
) -> ProductionRun:
|
|
"""Measure the same task again without closing the cycle.
|
|
|
|
This is the piece being turned over. The deadline and the cycle count are left
|
|
exactly as they are - a second reading of the same part is not a new cycle, and
|
|
it must not buy the operator another interval's worth of time. The values
|
|
themselves are ordinary measurements: both readings stay in the statistics,
|
|
which is the point of taking them.
|
|
"""
|
|
_require_measurable(run)
|
|
_add_event(db, run, "remeasure", user, note=note, task_id=task_id)
|
|
await db.flush()
|
|
await db.refresh(run)
|
|
return run
|
|
|
|
|
|
async def pause_run(
|
|
db: AsyncSession,
|
|
run: ProductionRun,
|
|
user: User,
|
|
supervisor: User,
|
|
note: Optional[str] = None,
|
|
) -> ProductionRun:
|
|
"""Fermo linea: suspend the run and freeze the countdown."""
|
|
_require_open(run)
|
|
if run.status == "paused":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT, detail="Production run is already paused",
|
|
)
|
|
run.status = "paused"
|
|
run.paused_at = _now()
|
|
_add_event(db, run, "line_stop", user, supervisor=supervisor, note=note)
|
|
await db.flush()
|
|
await db.refresh(run)
|
|
return run
|
|
|
|
|
|
async def resume_run(
|
|
db: AsyncSession,
|
|
run: ProductionRun,
|
|
user: User,
|
|
supervisor: User,
|
|
note: Optional[str] = None,
|
|
) -> ProductionRun:
|
|
"""Restart a paused run, pushing the deadline out by the length of the stop."""
|
|
_require_open(run)
|
|
if run.status != "paused":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT, detail="Production run is not paused",
|
|
)
|
|
now = _now()
|
|
if run.next_measurement_at is not None and run.paused_at is not None:
|
|
# Shift rather than recompute: whatever was left when the line stopped is
|
|
# what the operator gets back, so a stop neither grants nor costs time.
|
|
run.next_measurement_at = run.next_measurement_at + (now - run.paused_at)
|
|
run.status = "running"
|
|
run.paused_at = None
|
|
_add_event(db, run, "resume", user, supervisor=supervisor, note=note)
|
|
await db.flush()
|
|
await db.refresh(run)
|
|
return run
|
|
|
|
|
|
async def close_run(
|
|
db: AsyncSession,
|
|
run: ProductionRun,
|
|
user: User,
|
|
supervisor: User,
|
|
note: Optional[str] = None,
|
|
) -> ProductionRun:
|
|
"""Fine produzione: close the run for good, stop the timer, emit the statistics.
|
|
|
|
Everything here happens without the ERP. Handing the production over to GAIA
|
|
goes at the marked seam below, once D-1 and D-2 are answered; until then the
|
|
file is produced and the run is properly closed, which is what the shop floor
|
|
needs either way.
|
|
"""
|
|
_require_open(run)
|
|
run.status = "closed"
|
|
run.closed_at = _now()
|
|
run.closed_by = supervisor.id
|
|
run.next_measurement_at = None
|
|
run.paused_at = None
|
|
# Releasing the slot lets the station open a new run; the unique index on this
|
|
# column is what keeps exactly one open at a time.
|
|
run.active_station_id = None
|
|
_add_event(db, run, "close", user, supervisor=supervisor, note=note)
|
|
await db.flush()
|
|
|
|
# Statistics file for the whole production, and the measurements marked as sent.
|
|
run.statistics_path = await production_export_service.export_run_statistics(db, run)
|
|
|
|
# --- ERP hand-off (GAIA) plugs in here once the protocol is defined (D-1, D-2).
|
|
# Deliberately absent rather than stubbed: an empty call that looks wired is
|
|
# worse than none at all.
|
|
|
|
await db.flush()
|
|
await db.refresh(run)
|
|
return run
|