"""MeasurementTec blueprint - recipe selection and measurement execution.""" from flask import ( Blueprint, flash, jsonify, redirect, render_template, request, session, url_for, ) from flask_babel import gettext as _ from blueprints.auth import login_required, role_required from config import Config from services.api_client import api_client from services.file_proxy import proxy_file measure_bp = Blueprint("measure", __name__) # Session key holding a station chosen at runtime, overriding Config.STATION_CODE. STATION_OVERRIDE_KEY = "station_override" def _current_station() -> tuple[str | None, bool]: """Return the station this client acts as, and whether it is an override.""" override = session.get(STATION_OVERRIDE_KEY) if override: return override, True return Config.STATION_CODE, False def _apply_station_switch() -> None: """Honour ?station=CODE, used to exercise several stations from one PC. Only active when STATION_SWITCH_ENABLED is set: in production the station identity comes from the local install, and silently measuring against another station's recipes would break traceability. An empty value clears the override and falls back to the configured station. The requested code is validated against the server before being stored, so a typo leaves the operator on the current station with an explanation rather than stranded on a station that does not exist. """ if "station" not in request.args: return requested = (request.args.get("station") or "").strip().upper() if not Config.STATION_SWITCH_ENABLED: flash(_("Il cambio stazione da URL non è abilitato su questa postazione."), "error") return if not requested: if session.pop(STATION_OVERRIDE_KEY, None): flash(_("Stazione riportata a quella configurata."), "info") return try: probe = api_client.get_station_recipes(requested) except Exception: flash(_("Impossibile contattare il server per cambiare stazione."), "error") return if isinstance(probe, dict) and probe.get("error"): flash( _("Stazione '%(code)s' inesistente o disattivata: stazione invariata.", code=requested), "error", ) return session[STATION_OVERRIDE_KEY] = requested flash(_("Stazione corrente: %(code)s", code=requested), "info") # --------------------------------------------------------------------------- # Route: Recipe selection # --------------------------------------------------------------------------- @measure_bp.route("/select") @login_required @role_required("MeasurementTec") def select_recipe(): """Recipe selection page with search and barcode support.""" _apply_station_switch() station_code, station_overridden = _current_station() # Fail-fast if no station is configured and none was chosen if not station_code: return render_template("errors/station_not_configured.html"), 503 # Load recipes filtered by station try: resp = api_client.get_station_recipes(station_code) except Exception as e: return render_template( "errors/station_not_configured.html", error=str(e), ), 502 if isinstance(resp, dict) and resp.get("error"): flash( _("Errore nel caricamento delle ricette: %(detail)s", detail=resp.get("detail", "")), "error", ) recipes = [] else: # API may return paginated envelope or plain list recipes = resp.get("items", resp) if isinstance(resp, dict) else resp # Auto-fill from query params auto_recipe_code = request.args.get("recipe", "") auto_lot = request.args.get("lot", session.get("lot_number", "")) auto_serial = request.args.get("serial", session.get("serial_number", "")) return render_template( "measure/select_recipe.html", recipes=recipes, station_code=station_code, station_overridden=station_overridden, station_configured=Config.STATION_CODE, auto_recipe_code=auto_recipe_code, auto_lot=auto_lot, auto_serial=auto_serial, ) # --------------------------------------------------------------------------- # Route: Task list for a recipe # --------------------------------------------------------------------------- @measure_bp.route("/tasks/") @login_required @role_required("MeasurementTec") def task_list(recipe_id: int): """Task list for selected recipe.""" # Persist lot/serial from query params into session lot_number = request.args.get( "lot_number", session.get("lot_number", ""), ) serial_number = request.args.get( "serial_number", session.get("serial_number", ""), ) if lot_number: session["lot_number"] = lot_number if serial_number: session["serial_number"] = serial_number # Load recipe details recipe_resp = api_client.get(f"/api/recipes/{recipe_id}") if recipe_resp.get("error"): flash( _("Ricetta non trovata: %(detail)s", detail=recipe_resp.get("detail", "")), "error", ) return redirect(url_for("measure.select_recipe")) # Load tasks for this recipe tasks_resp = api_client.get(f"/api/recipes/{recipe_id}/tasks") if isinstance(tasks_resp, dict) and tasks_resp.get("error"): flash( _("Errore nel caricamento dei task: %(detail)s", detail=tasks_resp.get("detail", "")), "error", ) tasks = [] else: tasks = tasks_resp if isinstance(tasks_resp, list) else tasks_resp.get("items", []) return render_template( "measure/task_list.html", recipe=recipe_resp, tasks=tasks, lot_number=lot_number, serial_number=serial_number, ) # --------------------------------------------------------------------------- # Route: Task execution (measurement input) # --------------------------------------------------------------------------- @measure_bp.route("/execute/") @login_required @role_required("MeasurementTec") def task_execute(task_id: int): """Execute measurements for a task.""" # Load task + subtasks task_resp = api_client.get(f"/api/tasks/{task_id}") if task_resp.get("error"): flash( _("Task non trovato: %(detail)s", detail=task_resp.get("detail", "")), "error", ) return redirect(url_for("measure.select_recipe")) lot_number = session.get("lot_number", "") serial_number = session.get("serial_number", "") # Load all task IDs for this recipe (ordered) for auto-advance recipe_id = task_resp.get("recipe_id") all_task_ids = [] measurement_interval_minutes = None if recipe_id: tasks_resp = api_client.get(f"/api/recipes/{recipe_id}/tasks") if isinstance(tasks_resp, list): sorted_tasks = sorted(tasks_resp, key=lambda t: t.get("order_index", 0)) all_task_ids = [t["id"] for t in sorted_tasks] recipe_resp = api_client.get(f"/api/recipes/{recipe_id}") if not recipe_resp.get("error"): measurement_interval_minutes = recipe_resp.get("measurement_interval_minutes") return render_template( "measure/task_execute.html", task=task_resp, lot_number=lot_number, serial_number=serial_number, all_task_ids=all_task_ids, measurement_interval_minutes=measurement_interval_minutes, ) # --------------------------------------------------------------------------- # Route: Task completion summary # --------------------------------------------------------------------------- @measure_bp.route("/complete/") @login_required @role_required("MeasurementTec") def task_complete(recipe_id: int): """Task completion summary with all measurements.""" # Retrieve version_id from query params version_id = request.args.get("version_id") # Load recipe for context recipe_resp = api_client.get(f"/api/recipes/{recipe_id}") if recipe_resp.get("error"): flash( _("Ricetta non trovata: %(detail)s", detail=recipe_resp.get("detail", "")), "error", ) return redirect(url_for("measure.select_recipe")) # Load tasks+subtasks for this recipe to build subtask and task lookup tasks_resp = api_client.get(f"/api/recipes/{recipe_id}/tasks") subtask_map = {} subtask_task_map = {} # subtask_id → task info if isinstance(tasks_resp, list): for task in tasks_resp: for st in task.get("subtasks", []): subtask_map[st["id"]] = st subtask_task_map[st["id"]] = { "id": task["id"], "title": task.get("title", ""), "order_index": task.get("order_index", 0), } # Load measurements if version_id provided measurements = [] if version_id: meas_resp = api_client.get( "/api/measurements", params={"version_id": version_id, "per_page": 500}, ) if not (isinstance(meas_resp, dict) and meas_resp.get("error")): raw = ( meas_resp if isinstance(meas_resp, list) else meas_resp.get("items", []) ) # Enrich each measurement with nested subtask data for m in raw: st = subtask_map.get(m.get("subtask_id"), {}) m["subtask"] = st m["task_info"] = subtask_task_map.get(m.get("subtask_id"), {}) # Compute deviation if not present if m.get("deviation") is None and st.get("nominal") is not None: try: m["deviation"] = m["value"] - st["nominal"] except (TypeError, KeyError): m["deviation"] = 0.0 measurements = raw lot_number = session.get("lot_number", "") serial_number = session.get("serial_number", "") return render_template( "measure/task_complete.html", recipe=recipe_resp, measurements=measurements, lot_number=lot_number, serial_number=serial_number, ) # --------------------------------------------------------------------------- # Route: Barcode lookup (AJAX) # --------------------------------------------------------------------------- @measure_bp.route("/lookup-barcode", methods=["POST"]) @login_required @role_required("MeasurementTec") def lookup_barcode(): """Look up a recipe by barcode/code. Returns JSON for AJAX calls.""" data = request.get_json(silent=True) or {} code = data.get("code", "").strip() if not code: return jsonify({"error": True, "detail": _("Codice non fornito")}), 400 resp = api_client.get(f"/api/recipes/code/{code}") if resp.get("error"): return jsonify({ "error": True, "detail": resp.get("detail", _("Ricetta non trovata")), }), 404 return jsonify(resp) # --------------------------------------------------------------------------- # Route: Save lot/serial to session (AJAX) # --------------------------------------------------------------------------- @measure_bp.route("/save-traceability", methods=["POST"]) @login_required @role_required("MeasurementTec") def save_traceability(): """Save lot_number and serial_number to session.""" data = request.get_json(silent=True) or {} lot = data.get("lot_number", "").strip() serial = data.get("serial_number", "").strip() if lot: session["lot_number"] = lot if serial: session["serial_number"] = serial return jsonify({"ok": True}) # --------------------------------------------------------------------------- # Route: Save measurement (AJAX proxy to FastAPI) # --------------------------------------------------------------------------- @measure_bp.route("/save-measurement", methods=["POST"]) @login_required @role_required("MeasurementTec") def save_measurement(): """Save a single measurement value via API proxy. Expects JSON body: subtask_id: int task_id: int value: float pass_fail: str ('pass' | 'warning' | 'fail') deviation: float lot_number: str (optional) serial_number: str (optional) Returns JSON with the created measurement or error. """ data = request.get_json(silent=True) or {} # Validate required fields subtask_id = data.get("subtask_id") version_id = data.get("version_id") value = data.get("value") if subtask_id is None or version_id is None or value is None: return jsonify({ "error": True, "detail": _("Dati mancanti: subtask_id, version_id e value sono obbligatori"), }), 400 # Build payload for the FastAPI backend payload = { "subtask_id": subtask_id, "version_id": version_id, "value": value, "lot_number": data.get("lot_number", session.get("lot_number", "")), "serial_number": data.get("serial_number", session.get("serial_number", "")), "input_method": data.get("input_method", "manual"), "input_duration_ms": data.get("input_duration_ms"), # Attaches the measurement to the production under way, so fine produzione # can hand the whole run to the statistics file. "production_run_id": data.get("production_run_id"), } # Trailing slash matters: the route is declared as "/", so omitting it costs # a 307 redirect on every single measurement saved. resp = api_client.post("/api/measurements/", data=payload) if resp.get("error"): status_code = resp.get("status_code", 500) return jsonify({ "error": True, "detail": resp.get("detail", _("Errore nel salvataggio")), }), status_code if status_code >= 400 else 500 return jsonify(resp), 201 # --------------------------------------------------------------------------- # Route: Validate supervisor credentials (AJAX) # --------------------------------------------------------------------------- @measure_bp.route("/validate-supervisor", methods=["POST"]) @login_required @role_required("MeasurementTec") def validate_supervisor(): """Validate supervisor (capoturno) credentials for out-of-tolerance authorization.""" data = request.get_json(silent=True) or {} username = data.get("username", "").strip() password = data.get("password", "") if not username or not password: return jsonify({"error": True, "detail": _("Username e password richiesti")}), 400 resp = api_client.post("/api/auth/login", data={"username": username, "password": password}) if resp.get("error"): return jsonify({"error": True, "detail": _("Credenziali non valide")}), 401 user = resp.get("user", {}) is_supervisor = "Supervisor" in (user.get("roles") or []) if not (is_supervisor or user.get("is_admin")): return jsonify({"error": True, "detail": _("Utente non autorizzato (richiesto capoturno)")}), 403 return jsonify({"authorized": True, "supervisor": user.get("display_name", username)}), 200 # --------------------------------------------------------------------------- # Routes: Production run (state that must outlive the page) # --------------------------------------------------------------------------- @measure_bp.route("/api/production/current", methods=["GET"]) @login_required @role_required("MeasurementTec") def api_current_production(): """Proxy: the production open at this station, or null. Every page asks this on load. Before, the timer and the cycle count lived in the Alpine component and a change of task - a full page load - wiped them. """ station_code, _overridden = _current_station() if not station_code: return jsonify({"error": True, "detail": _("Stazione non configurata")}), 503 resp = api_client.get( "/api/production-runs/current", params={"station_code": station_code}, ) if isinstance(resp, dict) and resp.get("error"): return jsonify(resp), resp.get("status_code", 500) return jsonify(resp), 200 @measure_bp.route("/api/production/start", methods=["POST"]) @login_required @role_required("MeasurementTec") def api_start_production(): """Proxy: open a production at this station.""" station_code, _overridden = _current_station() if not station_code: return jsonify({"error": True, "detail": _("Stazione non configurata")}), 503 data = request.get_json(silent=True) or {} payload = { "station_code": station_code, "recipe_id": data.get("recipe_id"), "version_id": data.get("version_id"), "lot_number": data.get("lot_number") or session.get("lot_number") or None, "serial_number": data.get("serial_number") or session.get("serial_number") or None, } resp = api_client.post("/api/production-runs", data=payload) if isinstance(resp, dict) and resp.get("error"): return jsonify(resp), resp.get("status_code", 500) return jsonify(resp), 201 @measure_bp.route("/api/production//cycle", methods=["POST"]) @login_required @role_required("MeasurementTec") def api_complete_cycle(run_id: int): """Proxy: record a finished measurement cycle and restart the interval.""" data = request.get_json(silent=True) or {} resp = api_client.post( f"/api/production-runs/{run_id}/cycle", data={"note": data.get("note")}, ) if isinstance(resp, dict) and resp.get("error"): return jsonify(resp), resp.get("status_code", 500) return jsonify(resp), 200 def _supervised_action(run_id: int, action: str): """Forward an action that needs the capoturno's authorisation. The credentials go straight to the API, which checks them and records who authorised what on the run's trace. Validating them separately first would leave no such record, and is an extra round trip besides. """ data = request.get_json(silent=True) or {} username = (data.get("supervisor_username") or "").strip() password = data.get("supervisor_password") or "" if not username or not password: return jsonify({ "error": True, "detail": _("Username e password richiesti"), }), 400 resp = api_client.post( f"/api/production-runs/{run_id}/{action}", data={ "supervisor_username": username, "supervisor_password": password, "note": data.get("note"), }, ) if isinstance(resp, dict) and resp.get("error"): return jsonify(resp), resp.get("status_code", 500) return jsonify(resp), 200 @measure_bp.route("/api/production//pause", methods=["POST"]) @login_required @role_required("MeasurementTec") def api_pause_production(run_id: int): """Proxy: fermo linea - suspends the run and freezes the countdown.""" return _supervised_action(run_id, "pause") @measure_bp.route("/api/production//resume", methods=["POST"]) @login_required @role_required("MeasurementTec") def api_resume_production(run_id: int): """Proxy: restart a stopped line, giving back the time the stop took.""" return _supervised_action(run_id, "resume") @measure_bp.route("/api/production//close", methods=["POST"]) @login_required @role_required("MeasurementTec") def api_close_production(run_id: int): """Proxy: fine produzione - closes the run and emits the statistics file.""" return _supervised_action(run_id, "close") # --------------------------------------------------------------------------- # Route: File proxy (browser can't send X-API-Key directly) # --------------------------------------------------------------------------- @measure_bp.route("/api/files/", methods=["GET"]) @login_required def api_get_file(file_path: str): """Proxy: Serve file from API server (browser can't send X-API-Key).""" return proxy_file(file_path)