diff --git a/scripts/seed_collaudo.py b/scripts/seed_collaudo.py new file mode 100644 index 0000000..4ade59f --- /dev/null +++ b/scripts/seed_collaudo.py @@ -0,0 +1,321 @@ +"""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())