"""Read-only, no-auth JSON API for the Temper-Linie overview page. The page is served same-origin under ``/module/temper_overview`` and calls ``GET /module/temper_overview/api/overview?range=1d`` for the whole-line payload. No ``verify_token`` dependency: this is the deliberate no-auth read surface for the Grafana iframe (non-sensitive shop-floor data). Keep every route narrow and read-only — NEVER proxy arbitrary SQL. ── STATUS: REAL DATA ───────────────────────────────────────────────────────────── This binds the whole-line board to the live database, **un-scoped across presses** but scoped to one plant (``PLANT_NAME`` env, default ``Cadolzburg`` — same convention as ``modules/fastpress/press/_identity.py``). It reuses the temper status-derivation the fastpress module already relies on: a cart's live position is derived from its planned timestamps vs ``NOW()`` (``on_way`` while ``NOW() < ttIntoOvenAt``; ``in_oven`` until ``ttOutOfOvenAt``; ``done`` after), computed in SQL so the DB clock is the single source of truth and the timezone is never guessed. Data sources (all through the kernel SELECT-only proxy ``.._db.run_select_query``): * ``press_settings.presses`` — the canonical press list for the plant. * ``production.active_orders`` — per-press active orders + produced good parts. * ``production.rejects`` — reject totals per order (no plant column; keyed by ``rejOrdernumber`` like the fastpress overview). * ``production.cycles_new`` — last cycle time + run/stop recency per press. * ``production.downtimes`` — the latest downtime reason per press (the note). * ``settings.downtime_reasons`` — reason texts (DE) for the note. * ``production.temper_tracking`` — carts: on-the-way, in-oven, finished, and the parts sent to temper (for "nicht verladen"). Decisions (confirmed with the user, 2026-07-02): * run/stop/idle = **cycle recency**. idle = no active order; run = a cycle ended within ``RUN_RECENCY_MINUTES`` (env, default 10); else stop. The note text comes from the latest ``downtimes.Reason_int`` (99 "Im Betrieb" is treated as "no note"). * The oven **temperature is omitted** (no sensor feed exists) — the frontend hides the °C chip. ``ovenAvgMin`` is a real planned-oven-minutes average. * Reprint rows (``ttReprintOf IS NOT NULL``) are EXCLUDED from every parts/cart sum so a re-printed label never double-counts its cart. Never echoes customer/price fields — only order number, article description, counts. """ from __future__ import annotations import logging import os from datetime import datetime from typing import Any, Optional from fastapi import APIRouter, Query from .._db import run_select_query logger = logging.getLogger(__name__) router = APIRouter(prefix="/module/temper_overview/api", tags=["temper_overview"]) # ── Config (module-only; the generic kernel never reads these) ────────────────── # The plant this board scopes to — same convention as fastpress _identity.PLANT_NAME. PLANT_NAME = os.getenv("PLANT_NAME", "Cadolzburg") # A press counts as "running" if a cycle ended within this many minutes. RUN_RECENCY_MINUTES = int(os.getenv("TEMPER_OVERVIEW_RUN_RECENCY_MIN", "10") or 10) # Fallback average oven duration when no cart carries a planned time yet. DEFAULT_OVEN_MINUTES = int(os.getenv("TEMPER_DEFAULT_MINUTES", "120") or 120) _DEFAULT_RANGE = "1d" # finished-figures window → minutes back from NOW() (mirrors the design's picker). _RANGE_MINUTES: dict[str, int] = { "12h": 12 * 60, "1d": 24 * 60, "7d": 7 * 24 * 60, "14d": 14 * 24 * 60, "30d": 30 * 24 * 60, } # Reason_int values that mean "running / no stoppage" → not shown as a stop note. _RUNNING_REASONS = {99} # How many recent finished carts to list in the "Fertig/Lager" band. _FINISHED_LIST_LIMIT = 8 # ── Small pure helpers (unit-testable, no DB) ─────────────────────────────────── def _range_code(raw: str) -> str: return raw if raw in _RANGE_MINUTES else _DEFAULT_RANGE def _color_map(*ordernumbers: str) -> dict[str, int]: """Assign each distinct order number a colour INDEX 0..3 (the frontend maps the index → hex via ``src/data/orderColors.ts``). Sorted so the mapping is stable for a given SET of orders and distinct for up to four concurrent orders; it wraps beyond four. Falsy names are ignored.""" names = sorted({o for o in ordernumbers if o}) return {name: i % 4 for i, name in enumerate(names)} def _fmt_ago(minutes: Optional[int]) -> str: """German 'time ago' label from whole minutes (e.g. 8 → 'vor 8 min', 66 → 'vor 1 h 6 min'). ``None``/negative collapses to 'gerade eben'.""" if minutes is None or minutes < 1: return "gerade eben" if minutes < 60: return f"vor {minutes} min" h, m = divmod(minutes, 60) return f"vor {h} h {m} min" if m else f"vor {h} h" def _portion(orderno: str, qty: Any, colors: dict[str, int]) -> dict[str, Any]: return {"orderNo": orderno, "qty": int(qty or 0), "colorIndex": colors.get(orderno, 0)} def _group_carts(rows: list[dict[str, Any]], colors: dict[str, int]) -> "dict[str, dict[str, Any]]": """Group temper rows into physical carts by ``cartId`` (falling back to ``#ttId`` when a row predates the shared cart id). Preserves first-seen order. Returns ``{displayCartId: {"cartId", "orders": [portion...], "_rows": [row...]}}`` — the caller reads whatever cart-level fields it needs (eta / remaining / ago) off the grouped rows (all rows of one cart share the same planned timestamps).""" carts: dict[str, dict[str, Any]] = {} for r in rows: cid = r.get("cartId") or f"#{r['ttId']}" cart = carts.get(cid) if cart is None: cart = {"cartId": cid, "orders": [], "_rows": []} carts[cid] = cart cart["orders"].append(_portion(r["orderno"], r["qty"], colors)) cart["_rows"].append(r) return carts # ── SQL fetches (each degrades to empty on failure so the board never 500s) ───── def _fetch_presses(plant: str) -> list[str]: try: rows = run_select_query( "SELECT psPress AS press FROM press_settings.presses WHERE psPlant = :plant " "ORDER BY CAST(psPress AS UNSIGNED), psPress", {"plant": plant}, ) except Exception as e: # noqa: BLE001 logger.warning("temper_overview: press list unavailable: %s", e) return [] return [str(r["press"]) for r in rows if r.get("press") is not None] def _fetch_active_orders(plant: str) -> list[dict[str, Any]]: try: rows = run_select_query( "SELECT aoPress AS press, aoOrdernumber AS orderno, aoArticleDesc AS descr, " "COALESCE(aoActualCycles,0) AS actual, COALESCE(aoCavities,0) AS cavities " "FROM production.active_orders WHERE aoPlant = :plant " "ORDER BY aoPress, aoOrdernumber", {"plant": plant}, ) except Exception as e: # noqa: BLE001 logger.warning("temper_overview: active orders unavailable: %s", e) return [] out = [] for r in rows: cavities = int(r["cavities"] or 0) out.append( { "press": str(r["press"]), "orderno": r["orderno"], "produced": int(r["actual"] or 0) * cavities, } ) return out def _fetch_rejects(ordernumbers: list[str]) -> dict[str, int]: """Reject totals per order. ``production.rejects`` has no plant column, so it is keyed by ``rejOrdernumber`` (as the fastpress overview does) over just this plant's active orders.""" if not ordernumbers: return {} placeholders = ", ".join(f":o{i}" for i in range(len(ordernumbers))) binds = {f"o{i}": o for i, o in enumerate(ordernumbers)} try: rows = run_select_query( "SELECT rejOrdernumber AS orderno, COALESCE(SUM(rejQuantity),0) AS qty " f"FROM production.rejects WHERE rejOrdernumber IN ({placeholders}) " "GROUP BY rejOrdernumber", binds, ) except Exception as e: # noqa: BLE001 logger.warning("temper_overview: rejects unavailable: %s", e) return {} return {r["orderno"]: int(r["qty"] or 0) for r in rows} def _fetch_last_cycle(plant: str) -> dict[str, dict[str, Any]]: """Per press: the last cycle time (T_Cycle_Act of the newest cycle by id) and whether it ended within ``RUN_RECENCY_MINUTES`` (recency resolved in SQL against the DB clock). ``{press: {"ct": float|None, "recent": bool}}``.""" try: rows = run_select_query( "SELECT c.Press AS press, c.T_Cycle_Act AS ct, " "(c.DT_CycEnd >= NOW() - INTERVAL :recency MINUTE) AS recent " "FROM production.cycles_new c " "JOIN (SELECT Press, MAX(id) AS mid FROM production.cycles_new " " WHERE Plant = :plant GROUP BY Press) m " " ON c.id = m.mid AND c.Press = m.Press " "WHERE c.Plant = :plant", {"plant": plant, "recency": RUN_RECENCY_MINUTES}, ) except Exception as e: # noqa: BLE001 logger.warning("temper_overview: cycle recency unavailable: %s", e) return {} out: dict[str, dict[str, Any]] = {} for r in rows: try: ct = round(float(r["ct"]), 1) if r["ct"] is not None else None except (TypeError, ValueError): ct = None out[str(r["press"])] = {"ct": ct, "recent": bool(r["recent"])} return out def _fetch_downtime_reasons(plant: str) -> dict[str, int]: """The latest downtime ``Reason_int`` per press (by ``DT_start``). Used only for the press note text.""" try: rows = run_select_query( "SELECT d.Press AS press, d.Reason_int AS reason " "FROM production.downtimes d " "JOIN (SELECT Press, MAX(DT_start) AS mx FROM production.downtimes " " WHERE Plant = :plant GROUP BY Press) m " " ON d.Press = m.Press AND d.DT_start = m.mx " "WHERE d.Plant = :plant", {"plant": plant}, ) except Exception as e: # noqa: BLE001 logger.warning("temper_overview: downtime reasons unavailable: %s", e) return {} out: dict[str, int] = {} for r in rows: p = str(r["press"]) if p not in out and r["reason"] is not None: # first wins on a DT_start tie out[p] = int(r["reason"]) return out def _fetch_reason_texts() -> dict[int, str]: try: rows = run_select_query( "SELECT id, reason_DE AS de FROM settings.downtime_reasons", {} ) except Exception as e: # noqa: BLE001 logger.warning("temper_overview: reason texts unavailable: %s", e) return {} return {int(r["id"]): r["de"] for r in rows if r.get("de")} def _fetch_temper_sent(plant: str) -> dict[tuple[str, str], int]: """Good parts sent to temper per (press, order) — the basis for 'nicht verladen'. Excludes reprint rows so a re-printed label doesn't double-count.""" try: rows = run_select_query( "SELECT ttPress AS press, ttOrdernumber AS orderno, " "COALESCE(SUM(ttPartsGood),0) AS sent " "FROM production.temper_tracking " "WHERE ttPlant = :plant AND ttReprintOf IS NULL " "GROUP BY ttPress, ttOrdernumber", {"plant": plant}, ) except Exception as e: # noqa: BLE001 logger.warning("temper_overview: temper-sent aggregate unavailable: %s", e) return {} return {(str(r["press"]), r["orderno"]): int(r["sent"] or 0) for r in rows} def _fetch_active_carts(plant: str) -> list[dict[str, Any]]: """Carts not yet out of the oven — on-the-way + in-oven — one row per (cart,order). Live position + eta/remaining minutes are derived in SQL from the planned times.""" try: rows = run_select_query( "SELECT ttCartId AS cartId, ttId, ttPress AS press, ttOrdernumber AS orderno, " "COALESCE(ttPartsGood,0) AS qty, COALESCE(ttTemperMinutes,0) AS totalMin, " "CASE WHEN NOW() < ttIntoOvenAt THEN 'on_way' ELSE 'in_oven' END AS st, " "CEIL(GREATEST(0, TIMESTAMPDIFF(SECOND, NOW(), ttIntoOvenAt)) / 60) AS etaMin, " "CEIL(GREATEST(0, TIMESTAMPDIFF(SECOND, NOW(), ttOutOfOvenAt)) / 60) AS remainingMin " "FROM production.temper_tracking " "WHERE ttPlant = :plant AND ttReprintOf IS NULL " " AND ttIntoOvenAt IS NOT NULL AND ttOutOfOvenAt IS NOT NULL " " AND NOW() < ttOutOfOvenAt " "ORDER BY ttIntoOvenAt, ttId", {"plant": plant}, ) except Exception as e: # noqa: BLE001 logger.warning("temper_overview: active carts unavailable: %s", e) return [] return [ { "cartId": r.get("cartId"), "ttId": r["ttId"], "press": str(r["press"]), "orderno": r["orderno"], "qty": int(r["qty"] or 0), "totalMin": int(r["totalMin"] or 0), "st": r["st"], "etaMin": int(r["etaMin"] or 0), "remainingMin": int(r["remainingMin"] or 0), } for r in rows ] def _fetch_finished_carts(plant: str, window_min: int) -> list[dict[str, Any]]: """Carts out of the oven within the selected window — one row per (cart,order), newest first. Excludes reprints.""" try: rows = run_select_query( "SELECT ttCartId AS cartId, ttId, ttPress AS press, ttOrdernumber AS orderno, " "COALESCE(ttPartsGood,0) AS qty, COALESCE(ttScrap,0) AS scrap, " "COALESCE(ttTemperMinutes,0) AS temperMin, " "TIMESTAMPDIFF(MINUTE, ttOutOfOvenAt, NOW()) AS agoMin " "FROM production.temper_tracking " "WHERE ttPlant = :plant AND ttReprintOf IS NULL " " AND ttOutOfOvenAt IS NOT NULL AND NOW() >= ttOutOfOvenAt " " AND ttOutOfOvenAt >= NOW() - INTERVAL :window_min MINUTE " "ORDER BY ttOutOfOvenAt DESC, ttId DESC", {"plant": plant, "window_min": window_min}, ) except Exception as e: # noqa: BLE001 logger.warning("temper_overview: finished carts unavailable: %s", e) return [] return [ { "cartId": r.get("cartId"), "ttId": r["ttId"], "press": str(r["press"]), "orderno": r["orderno"], "qty": int(r["qty"] or 0), "scrap": int(r["scrap"] or 0), "temperMin": int(r["temperMin"] or 0), "agoMin": int(r["agoMin"]) if r["agoMin"] is not None else None, } for r in rows ] # ── Pure payload builder (no DB — testable with fabricated rows) ──────────────── def _build_payload( *, plant: str, range_code: str, presses: list[str], active_orders: list[dict[str, Any]], rejects: dict[str, int], last_cycle: dict[str, dict[str, Any]], downtime_reason: dict[str, int], reason_texts: dict[int, str], temper_sent: dict[tuple[str, str], int], active_carts: list[dict[str, Any]], finished_carts: list[dict[str, Any]], generated_at: str, ) -> dict[str, Any]: """Assemble the whole-line payload from already-fetched rows. Pure: given the same rows it always produces the same dict (the only I/O-derived input is ``generated_at``).""" # One stable colour index per order number, over every order that appears anywhere. colors = _color_map( *[o["orderno"] for o in active_orders], *[c["orderno"] for c in active_carts], *[c["orderno"] for c in finished_carts], ) # Per-press good parts + order chips, folding in rejects (produced good = produced − rejects). good_by_press: dict[str, int] = {} chips_by_press: dict[str, list[dict[str, Any]]] = {} not_verladen_by_press: dict[str, int] = {} seen_chip: dict[str, set] = {} for o in active_orders: press, orderno = o["press"], o["orderno"] good = max(0, o["produced"] - rejects.get(orderno, 0)) good_by_press[press] = good_by_press.get(press, 0) + good sent = temper_sent.get((press, orderno), 0) not_verladen_by_press[press] = not_verladen_by_press.get(press, 0) + max(0, good - sent) seen = seen_chip.setdefault(press, set()) if orderno not in seen: seen.add(orderno) chips_by_press.setdefault(press, []).append( {"orderNo": orderno, "colorIndex": colors.get(orderno, 0)} ) presses_with_order = {o["press"] for o in active_orders} presses_out: list[dict[str, Any]] = [] for press in presses: has_order = press in presses_with_order cyc = last_cycle.get(press, {}) recent = bool(cyc.get("recent")) if not has_order: # No active order → the defining fact is "no order". Match the design mock # (P7: idle · Kein Auftrag) and ignore any stale downtime reason. status = "idle" note = "Kein Auftrag" elif recent: status = "run" note = None else: status = "stop" reason = downtime_reason.get(press) note = reason_texts.get(reason) if (reason is not None and reason not in _RUNNING_REASONS) else None presses_out.append( { "name": press, "status": status, "orders": chips_by_press.get(press, []), "good": good_by_press.get(press, 0), "cycleTime": (f"{cyc['ct']:.1f}" if (status == "run" and cyc.get("ct") is not None) else None), "notVerladen": not_verladen_by_press.get(press, 0), "note": note, "placeholder": False, } ) # On-the-way carts, keyed by origin press. Each cart's eta = its planned into-oven. on_way_rows = [c for c in active_carts if c["st"] == "on_way"] on_way: dict[str, list[dict[str, Any]]] = {} for press in {c["press"] for c in on_way_rows}: carts = _group_carts([c for c in on_way_rows if c["press"] == press], colors) on_way[press] = [ { "cartId": cart["cartId"], "orders": cart["orders"], "etaMin": min(r["etaMin"] for r in cart["_rows"]), } for cart in carts.values() ] # In-oven batches (flat; the frontend sorts soonest-out first). oven_rows = [c for c in active_carts if c["st"] == "in_oven"] oven_carts = _group_carts(oven_rows, colors) oven: list[dict[str, Any]] = [] for cart in oven_carts.values(): rows = cart["_rows"] total_min = max((r["totalMin"] for r in rows), default=0) or DEFAULT_OVEN_MINUTES oven.append( { "cartId": cart["cartId"], "orders": cart["orders"], "totalMin": total_min, "remainingMin": min(r["remainingMin"] for r in rows), } ) # Finished figures for the window. fin_carts = _group_carts(finished_carts, colors) finished_parts = sum(c["qty"] for c in finished_carts) finished_scrap = sum(c["scrap"] for c in finished_carts) temper_mins = [c["temperMin"] for c in finished_carts if c["temperMin"] > 0] finished_avg = round(sum(temper_mins) / len(temper_mins)) if temper_mins else DEFAULT_OVEN_MINUTES by_order_parts: dict[str, int] = {} for c in finished_carts: by_order_parts[c["orderno"]] = by_order_parts.get(c["orderno"], 0) + c["qty"] by_order = [ {"orderNo": o, "colorIndex": colors.get(o, 0), "parts": p} for o, p in sorted(by_order_parts.items(), key=lambda kv: kv[1], reverse=True) ] finished = { "carts": len(fin_carts), "parts": finished_parts, "scrap": finished_scrap, "avgOvenMin": finished_avg, "byOrder": by_order, } finished_list = [ { "cartId": cart["cartId"], "orders": cart["orders"], "ago": _fmt_ago(min((r["agoMin"] for r in cart["_rows"] if r["agoMin"] is not None), default=None)), } for cart in list(fin_carts.values())[:_FINISHED_LIST_LIMIT] ] # Header oven average = mean planned minutes of the carts currently in the oven, # falling back to the finished-window average, then the global default. in_oven_mins = [b["totalMin"] for b in oven if b["totalMin"] > 0] oven_avg_min = round(sum(in_oven_mins) / len(in_oven_mins)) if in_oven_mins else finished_avg return { "plant": plant, "generatedAt": generated_at, "range": range_code, # ovenTempC intentionally omitted — no sensor feed; the frontend hides the chip. "ovenAvgMin": oven_avg_min, "sample": False, "presses": presses_out, "onWay": on_way, "oven": oven, "finished": finished, "finishedList": finished_list, } @router.get("/overview", summary="Whole temper-line overview (live data)") def overview(range: str = Query(_DEFAULT_RANGE, description="finished-figures window: 12h|1d|7d|14d|30d")) -> dict[str, Any]: code = _range_code(range) window_min = _RANGE_MINUTES[code] plant = PLANT_NAME presses = _fetch_presses(plant) active_orders = _fetch_active_orders(plant) rejects = _fetch_rejects([o["orderno"] for o in active_orders]) last_cycle = _fetch_last_cycle(plant) downtime_reason = _fetch_downtime_reasons(plant) reason_texts = _fetch_reason_texts() temper_sent = _fetch_temper_sent(plant) active_carts = _fetch_active_carts(plant) finished_carts = _fetch_finished_carts(plant, window_min) return _build_payload( plant=plant, range_code=code, presses=presses, active_orders=active_orders, rejects=rejects, last_cycle=last_cycle, downtime_reason=downtime_reason, reason_texts=reason_texts, temper_sent=temper_sent, active_carts=active_carts, finished_carts=finished_carts, generated_at=datetime.now().astimezone().isoformat(), )