"""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: it prints how many Measurement/ProductionRun/ProductionEvent rows that is and refuses to touch them without --si. Refuses to run against a real deployment (APP_ENV outside dev/test, same rule as src.backend.config) unless --conferma-produzione is also given: this script deletes data, and a shop floor is not where to find that out by mistake. If --supervisor-password is not given and the capoturno account does not exist yet, a random one is generated and printed once - never committed, never a default anyone could look up. """ import argparse import asyncio import secrets import sys from sqlalchemy import delete, func, select sys.path.insert(0, "/app") from src.backend.config import settings # noqa: E402 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 _count_replace_impact(session, recipes) -> dict: """How many Measurement/ProductionRun/ProductionEvent --replace would delete. Read-only mirror of what _delete_recipes actually removes, so the count printed before the confirmation matches the count deleted after it. """ version_ids = [] run_ids = [] for recipe in recipes: version_ids += (await session.execute( select(RecipeVersion.id).where(RecipeVersion.recipe_id == recipe.id) )).scalars().all() run_ids += (await session.execute( select(ProductionRun.id).where(ProductionRun.recipe_id == recipe.id) )).scalars().all() measurements = 0 if version_ids: measurements = (await session.execute( select(func.count()).select_from(Measurement) .where(Measurement.version_id.in_(version_ids)) )).scalar_one() events = 0 if run_ids: events = (await session.execute( select(func.count()).select_from(ProductionEvent) .where(ProductionEvent.run_id.in_(run_ids)) )).scalar_one() return { "measurements": measurements, "production_runs": len(run_ids), "production_events": events, } 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. `password` may be None: it is only needed when the account does not exist yet, in which case a random one is generated here rather than falling back to a committed default. Returns (user, created, password_used) so the caller can print the generated one - the only place it appears. """ existing = (await session.execute( select(User).where(User.username == username) )).scalar_one_or_none() if existing is not None: return existing, False, password password_used = password or secrets.token_urlsafe(12) user = await auth_service.create_user( session, username=username, password=password_used, display_name="Capo Turno (collaudo)", roles=["Supervisor", "MeasurementTec"], ) await session.flush() return user, True, password_used async def seed( station_code: str, replace: bool, supervisor_password: str | None, si: bool, ) -> 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: impact = await _count_replace_impact(session, existing) print(f"--replace cancellera' ({', '.join(r.code for r in existing)}):") print(f" {impact['measurements']} misure") print(f" {impact['production_runs']} produzioni") print(f" {impact['production_events']} eventi di produzione") if not si: print("Nessuna cancellazione eseguita. Aggiungi --si per confermare.") return 1 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, password_used = 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} / {password_used}") 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("--si", action="store_true", help="conferma la cancellazione di --replace (senza, si limita a mostrarla)") parser.add_argument("--supervisor-password", default=None, help="password del capoturno, se va creato; " "se omessa ne viene generata una casuale e stampata una volta") parser.add_argument("--conferma-produzione", action="store_true", help="richiesto per lanciare lo script con APP_ENV di produzione") args = parser.parse_args() if settings.app_env not in ("dev", "test") and not args.conferma_produzione: print( f"APP_ENV={settings.app_env!r} indica una produzione: " "questo script scrive ricette e utenti di collaudo. " "Rilancia con --conferma-produzione se e' davvero quello che vuoi." ) return 2 return asyncio.run( seed(args.station, args.replace, args.supervisor_password, args.si) ) if __name__ == "__main__": raise SystemExit(main())