Adriano Dal Pastro
919ee7ac79
Uno script che prepara sul sistema quello che i test dimostrano ma nessuno ha ancora visto su uno schermo: due ricette con impostazioni opposte, perche' i punti da collaudare le vogliono cosi'. COLLAUDO-A e' quella da guidare — due task di misura con task documentali prima, in mezzo e dopo, lotto obbligatorio, digitazione ammessa, intervallo di due minuti. Serve per il ciclo di misura, il conto alla rovescia che va oltre lo zero, il ritorno automatico alla misura e il gate del fuori tolleranza. COLLAUDO-B e' quella da guardare — solo calibro, lotto e seriale obbligatori: il tastierino non viene disegnato e Avvia non parte finche' mancano i dati. Le quote hanno limiti scelti perche' ogni esito sia a un tasto di distanza: 10.00 conforme, 10.30 attenzione, 12.00 fuori tolleranza. Crea anche un capoturno con il ruolo Supervisor: un amministratore passerebbe il controllo, ma il collaudo deve esercitare il ruolo che usera' l'officina. Rilanciarlo trova le ricette e si ferma; --replace le rifa' da zero, misure comprese, che e' quello che serve fra una sessione e l'altra e mai per sbaglio. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
322 lines
12 KiB
Python
322 lines
12 KiB
Python
"""Seed the recipes used to accept points 2, 3, 5, 8, 9 and 11 on a real screen.
|
|
|
|
The tests prove the rules hold; this puts them in front of an operator. Two
|
|
recipes, because they need opposite settings:
|
|
|
|
COLLAUDO-A typing allowed, lot compulsory, two measurement tasks with
|
|
documental tasks before, between and after them, interval of two
|
|
minutes. This is the one to drive: the measurement loop, the
|
|
bidirectional countdown, the out-of-tolerance gate, the cycle that
|
|
closes only on the last measurement task.
|
|
|
|
COLLAUDO-B caliper only, lot and serial compulsory. This one is to look at:
|
|
the keypad is not drawn, and Avvia does not start until both
|
|
traceability fields are filled in.
|
|
|
|
Run it inside the server container, where src.backend and the database settings
|
|
are already in place:
|
|
|
|
docker compose cp scripts/seed_collaudo.py server:/tmp/seed_collaudo.py
|
|
docker compose exec server uv run python /tmp/seed_collaudo.py --station ST-DEFAULT
|
|
|
|
Re-running finds the recipes already there and stops. --replace deletes them
|
|
first, along with anything measured against them - which is what you want between
|
|
one acceptance session and the next, and never what you want by accident.
|
|
"""
|
|
import argparse
|
|
import asyncio
|
|
import sys
|
|
|
|
from sqlalchemy import delete, select
|
|
|
|
sys.path.insert(0, "/app")
|
|
|
|
from src.backend.database import async_session_factory # noqa: E402
|
|
from src.backend.models.orm.measurement import Measurement # noqa: E402
|
|
from src.backend.models.orm.production import ( # noqa: E402
|
|
ProductionEvent, ProductionRun,
|
|
)
|
|
from src.backend.models.orm.recipe import Recipe, RecipeVersion # noqa: E402
|
|
from src.backend.models.orm.station import ( # noqa: E402
|
|
Station, StationRecipeAssignment,
|
|
)
|
|
from src.backend.models.orm.task import RecipeSubtask, RecipeTask # noqa: E402
|
|
from src.backend.models.orm.user import User # noqa: E402
|
|
from src.backend.services import auth_service # noqa: E402
|
|
|
|
CODES = ("COLLAUDO-A", "COLLAUDO-B")
|
|
|
|
# Limits chosen so every outcome is one deliberate keystroke away:
|
|
# 10.00 -> conforme 10.30 -> attenzione (dentro tolleranza)
|
|
# 12.00 -> fuori tolleranza, serve il capoturno
|
|
QUOTES_A1 = [
|
|
dict(marker_number=1, description="Diametro esterno",
|
|
nominal=10.0, ltl=9.5, lwl=9.8, uwl=10.2, utl=10.5, unit="mm"),
|
|
dict(marker_number=2, description="Spessore parete",
|
|
nominal=25.0, ltl=24.5, lwl=24.8, uwl=25.2, utl=25.5, unit="mm"),
|
|
]
|
|
QUOTES_A2 = [
|
|
dict(marker_number=3, description="Lunghezza totale",
|
|
nominal=100.0, ltl=99.0, lwl=99.5, uwl=100.5, utl=101.0, unit="mm"),
|
|
]
|
|
QUOTES_B = [
|
|
dict(marker_number=1, description="Diametro con calibro",
|
|
nominal=10.0, ltl=9.5, lwl=9.8, uwl=10.2, utl=10.5, unit="mm"),
|
|
]
|
|
|
|
# A description with a blank line and a bold phrase: point 11 is visible or it is
|
|
# not, and this is where you look.
|
|
NOTA_INIZIALE = (
|
|
"Prendere il pezzo dalla cassetta **a sinistra** della postazione.\n"
|
|
"Pulire la superficie prima di misurare.\n\n"
|
|
"**Attenzione**: non misurare pezzi ancora caldi."
|
|
)
|
|
NOTA_INTERMEDIA = (
|
|
"Girare il pezzo di **90 gradi** prima della seconda misura.\n"
|
|
"Se il pezzo non appoggia in piano, ripetere il posizionamento."
|
|
)
|
|
NOTA_FINALE = (
|
|
"Riporre il pezzo nella cassetta **a destra**.\n"
|
|
"Segnalare al capoturno qualsiasi quota fuori tolleranza."
|
|
)
|
|
|
|
|
|
async def _find_recipes(session, codes):
|
|
result = await session.execute(select(Recipe).where(Recipe.code.in_(codes)))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def _delete_recipes(session, recipes):
|
|
"""Remove the collaudo recipes and everything measured against them."""
|
|
for recipe in recipes:
|
|
versions = (await session.execute(
|
|
select(RecipeVersion.id).where(RecipeVersion.recipe_id == recipe.id)
|
|
)).scalars().all()
|
|
runs = (await session.execute(
|
|
select(ProductionRun.id).where(ProductionRun.recipe_id == recipe.id)
|
|
)).scalars().all()
|
|
|
|
if versions:
|
|
await session.execute(
|
|
delete(Measurement).where(Measurement.version_id.in_(versions))
|
|
)
|
|
if runs:
|
|
await session.execute(
|
|
delete(ProductionEvent).where(ProductionEvent.run_id.in_(runs))
|
|
)
|
|
await session.execute(
|
|
delete(ProductionRun).where(ProductionRun.id.in_(runs))
|
|
)
|
|
await session.execute(
|
|
delete(StationRecipeAssignment).where(
|
|
StationRecipeAssignment.recipe_id == recipe.id
|
|
)
|
|
)
|
|
# Versions, tasks and subtasks go with the recipe through the ORM cascade.
|
|
await session.delete(recipe)
|
|
await session.flush()
|
|
|
|
|
|
async def _add_task(session, version_id, order_index, title, task_type,
|
|
directive=None, description=None, quotes=()):
|
|
task = RecipeTask(
|
|
version_id=version_id,
|
|
order_index=order_index,
|
|
title=title,
|
|
task_type=task_type,
|
|
directive=directive,
|
|
description=description,
|
|
)
|
|
session.add(task)
|
|
await session.flush()
|
|
for quote in quotes:
|
|
session.add(RecipeSubtask(task_id=task.id, **quote))
|
|
await session.flush()
|
|
return task
|
|
|
|
|
|
async def _create_recipe(session, user_id, *, code, name, description,
|
|
interval, requires_lot, requires_serial,
|
|
allow_manual_input):
|
|
recipe = Recipe(
|
|
code=code,
|
|
name=name,
|
|
description=description,
|
|
created_by=user_id,
|
|
active=True,
|
|
measurement_interval_minutes=interval,
|
|
requires_lot=requires_lot,
|
|
requires_serial=requires_serial,
|
|
allow_manual_input=allow_manual_input,
|
|
)
|
|
session.add(recipe)
|
|
await session.flush()
|
|
|
|
version = RecipeVersion(
|
|
recipe_id=recipe.id,
|
|
version_number=1,
|
|
is_current=True,
|
|
created_by=user_id,
|
|
change_notes="Ricetta di collaudo",
|
|
)
|
|
session.add(version)
|
|
await session.flush()
|
|
return recipe, version
|
|
|
|
|
|
async def _assign(session, station, recipe, user_id):
|
|
session.add(StationRecipeAssignment(
|
|
station_id=station.id, recipe_id=recipe.id, assigned_by=user_id,
|
|
))
|
|
await session.flush()
|
|
|
|
|
|
async def _ensure_supervisor(session, username, password):
|
|
"""A capoturno with the role, not an admin standing in for one.
|
|
|
|
An admin is accepted by the authorisation check, but the acceptance session
|
|
should exercise the role the shop floor will actually use.
|
|
"""
|
|
existing = (await session.execute(
|
|
select(User).where(User.username == username)
|
|
)).scalar_one_or_none()
|
|
if existing is not None:
|
|
return existing, False
|
|
|
|
user = await auth_service.create_user(
|
|
session,
|
|
username=username,
|
|
password=password,
|
|
display_name="Capo Turno (collaudo)",
|
|
roles=["Supervisor", "MeasurementTec"],
|
|
)
|
|
await session.flush()
|
|
return user, True
|
|
|
|
|
|
async def seed(station_code: str, replace: bool, supervisor_password: str) -> int:
|
|
async with async_session_factory() as session:
|
|
owner = (await session.execute(
|
|
select(User).where(User.is_admin == True) # noqa: E712
|
|
)).scalars().first()
|
|
if owner is None:
|
|
print("Nessun utente amministratore: impossibile intestare le ricette.")
|
|
return 2
|
|
|
|
station = (await session.execute(
|
|
select(Station).where(Station.code == station_code)
|
|
)).scalar_one_or_none()
|
|
if station is None:
|
|
print(f"Stazione '{station_code}' inesistente.")
|
|
return 2
|
|
|
|
existing = await _find_recipes(session, CODES)
|
|
if existing and not replace:
|
|
found = ", ".join(r.code for r in existing)
|
|
print(f"Ricette di collaudo gia' presenti ({found}).")
|
|
print("Usa --replace per rifarle da zero (cancella anche le misure).")
|
|
return 1
|
|
if existing:
|
|
await _delete_recipes(session, existing)
|
|
print(f"Rimosse: {', '.join(r.code for r in existing)}")
|
|
|
|
# ---- COLLAUDO-A: the one to drive ----
|
|
recipe_a, version_a = await _create_recipe(
|
|
session, owner.id,
|
|
code="COLLAUDO-A",
|
|
name="Collaudo — ciclo di misura",
|
|
description=(
|
|
"Due task di misura con task documentali intorno. "
|
|
"Lotto obbligatorio, inserimento manuale consentito, "
|
|
"intervallo di 2 minuti."
|
|
),
|
|
interval=2,
|
|
requires_lot=True,
|
|
requires_serial=False,
|
|
allow_manual_input=True,
|
|
)
|
|
await _add_task(
|
|
session, version_a.id, 0, "Preparazione del pezzo", "note",
|
|
directive="Leggere prima di iniziare",
|
|
description=NOTA_INIZIALE,
|
|
)
|
|
await _add_task(
|
|
session, version_a.id, 1, "Misura 1 — diametro e spessore", "measure",
|
|
directive="Misurare le due quote in sequenza",
|
|
description="Quota **1** e quota **2**.\nUsare il calibro o il tastierino.",
|
|
quotes=QUOTES_A1,
|
|
)
|
|
await _add_task(
|
|
session, version_a.id, 2, "Riposizionamento", "note",
|
|
directive="Fra la prima e la seconda misura",
|
|
description=NOTA_INTERMEDIA,
|
|
)
|
|
await _add_task(
|
|
session, version_a.id, 3, "Misura 2 — lunghezza", "measure",
|
|
directive="Ultimo task di misura: chiude il ciclo",
|
|
description="Alla conferma di questa quota **riparte l'intervallo**.",
|
|
quotes=QUOTES_A2,
|
|
)
|
|
await _add_task(
|
|
session, version_a.id, 4, "Chiusura", "note",
|
|
directive="Dopo le misure",
|
|
description=NOTA_FINALE,
|
|
)
|
|
await _assign(session, station, recipe_a, owner.id)
|
|
|
|
# ---- COLLAUDO-B: the one to look at ----
|
|
recipe_b, version_b = await _create_recipe(
|
|
session, owner.id,
|
|
code="COLLAUDO-B",
|
|
name="Collaudo — solo calibro",
|
|
description=(
|
|
"Lotto e seriale obbligatori, inserimento manuale vietato: "
|
|
"il tastierino non compare e Avvia non parte finche' mancano i dati."
|
|
),
|
|
interval=5,
|
|
requires_lot=True,
|
|
requires_serial=True,
|
|
allow_manual_input=False,
|
|
)
|
|
await _add_task(
|
|
session, version_b.id, 0, "Misura con calibro", "measure",
|
|
directive="Il valore deve arrivare dal calibro",
|
|
description="Nessun tastierino: la ricetta **non ammette valori digitati**.",
|
|
quotes=QUOTES_B,
|
|
)
|
|
await _assign(session, station, recipe_b, owner.id)
|
|
|
|
supervisor, created = await _ensure_supervisor(
|
|
session, "capoturno", supervisor_password,
|
|
)
|
|
|
|
await session.commit()
|
|
|
|
print(f"Ricette create e assegnate alla stazione {station_code}:")
|
|
print(" COLLAUDO-A ciclo di misura, lotto obbligatorio, intervallo 2 min")
|
|
print(" COLLAUDO-B solo calibro, lotto e seriale obbligatori")
|
|
if created:
|
|
print(f"Capoturno creato: {supervisor.username} / {supervisor_password}")
|
|
print(" cambiare la password prima di qualunque uso reale.")
|
|
else:
|
|
print(f"Capoturno gia' presente: {supervisor.username} (password invariata)")
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--station", default="ST-DEFAULT",
|
|
help="codice della stazione a cui assegnare le ricette")
|
|
parser.add_argument("--replace", action="store_true",
|
|
help="cancella le ricette di collaudo esistenti e le misure fatte su di esse")
|
|
parser.add_argument("--supervisor-password", default="Collaudo2026!",
|
|
help="password del capoturno, se va creato")
|
|
args = parser.parse_args()
|
|
return asyncio.run(
|
|
seed(args.station, args.replace, args.supervisor_password)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|