SPA_TemperProductionOverview/api/routes.py
Erik aa115a0b8d feat: scaffold Temper-Linie live-overview page-module (temper_overview)
Vue 3 + Vite 8 + TS + PrimeVue 4 (Nora/Industrial) + Pinia + vue-i18n SPA, served
by the SPRO gateway as a static page-module under /module/temper_overview and
embedded into Grafana via iframe.

- Ports the LOCKED design #4 "Shop-floor Schematic": one connected line
  Pressen -> (cyan) -> Ofen -> (green) -> Fertig; carts under their origin press;
  oven trays as dwell-progress bars; finished band; range picker scoping only the
  Fertig figures. Flow/order colours match PressV (fleet recognisability).
- Python page-module wrapper: register(app) mounts committed dist/ + a no-auth
  read API (GET /api/overview). Kernel DB seam in _db.py.
- Backend returns SAMPLE data; frontend falls back to bundled sample so `npm run
  dev` renders with zero backend (TODO markers point at the real temper aggregates).
- Rich README bakes in the design, contract, wiring + Grafana embedding for
  standalone development.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-01 20:41:04 +02:00

142 lines
7.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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: BASE SCAFFOLD ────────────────────────────────────────────────────────
This returns realistic SAMPLE data shaped EXACTLY like the frontend contract
(``src/types/overview.ts``) so the page renders end-to-end with no database. The
``range`` query param scales only the "Fertig" (finished) figures, mirroring the
locked design (presses / on-the-way / oven are always real-time).
TODO — wire real data (see the gateway repo, this repo's README, and the memory
``.claude/memory/temper-line-overview.md``):
* Import the kernel SELECT proxy: ``from .._db import run_select_query``
* Whole-line, un-scoped across presses. Reuse the temper status-derivation from
``modules/fastpress/press/temper.py`` (``_temper_aggregates_all`` derives
on_way / in_oven / finished from planned timestamps vs NOW()); read
``production.temper_tracking`` (carts, grouped by ``ttCartId``, keyed by origin
press ``ttPress``) + ``production.active_orders``/``finished_orders``.
* "nichtVerladen" = produced good parts (cycles) parts logged to temper
(open question: per-press or plant-wide — see the memory).
* Scope the finished aggregates by ``range`` (12h..30d) on ``ttOutOfOvenAt``.
* Never echo customer/price fields.
Alternatively, add a public ``GET /press/temper/overview`` in fastpress and have
the page call that instead — decide before wiring; keep the payload shape below.
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from fastapi import APIRouter, Query
router = APIRouter(prefix="/module/temper_overview/api", tags=["temper_overview"])
# ── Sample data (ported verbatim from the locked prototype) ─────────────────────
# Order identity: key → (Auftragsnummer, colour index 0..3). The 4 colours live in
# the frontend (src/data/orderColors.ts) — the API only sends the index, so the
# page and legend always line up.
_ORDERS: dict[str, tuple[str, int]] = {
"o1": ("O-4471", 0),
"o2": ("O-4460", 1),
"o3": ("O-4455", 2),
"o4": ("O-4489", 3),
}
# range code → factor applied to the finished figures only (from the prototype).
_RANGE_FACTORS: dict[str, float] = {"12h": 0.5, "1d": 1.0, "7d": 6.4, "14d": 12.7, "30d": 27.2}
_DEFAULT_RANGE = "1d"
def _orders(*specs: tuple[str, int]) -> list[dict[str, Any]]:
"""[(order_key, qty), ...] → [{orderNo, qty, colorIndex}, ...]."""
out: list[dict[str, Any]] = []
for key, qty in specs:
name, color = _ORDERS[key]
out.append({"orderNo": name, "qty": qty, "colorIndex": color})
return out
def _order_chips(*keys: str) -> list[dict[str, Any]]:
"""order keys → [{orderNo, colorIndex}, ...] (no qty — for the press header)."""
return [{"orderNo": _ORDERS[k][0], "colorIndex": _ORDERS[k][1]} for k in keys]
_PRESSES: list[dict[str, Any]] = [
{"name": "P1", "status": "run", "orders": _order_chips("o1"), "good": 1240, "cycleTime": "4.2", "notVerladen": 34, "note": None, "placeholder": False},
{"name": "P2", "status": "run", "orders": _order_chips("o2"), "good": 880, "cycleTime": "6.1", "notVerladen": 18, "note": None, "placeholder": False},
{"name": "P3", "status": "run", "orders": _order_chips("o1", "o4"), "good": 612, "cycleTime": "4.3", "notVerladen": 52, "note": None, "placeholder": False},
{"name": "P4", "status": "stop", "orders": _order_chips("o2"), "good": 430, "cycleTime": None, "notVerladen": 0, "note": "Werkzeug rüsten", "placeholder": False},
{"name": "P5", "status": "run", "orders": _order_chips("o3"), "good": 1975, "cycleTime": "2.8", "notVerladen": 41, "note": None, "placeholder": False},
{"name": "P6", "status": "run", "orders": _order_chips("o2"), "good": 744, "cycleTime": "6.0", "notVerladen": 23, "note": None, "placeholder": False},
{"name": "P7", "status": "idle", "orders": [], "good": 0, "cycleTime": None, "notVerladen": 0, "note": "Kein Auftrag", "placeholder": False},
{"name": "P8", "status": "run", "orders": _order_chips("o4"), "good": 1102, "cycleTime": "3.9", "notVerladen": 29, "note": None, "placeholder": False},
{"name": "P9", "status": "run", "orders": _order_chips("o1"), "good": 968, "cycleTime": "4.1", "notVerladen": 12, "note": None, "placeholder": False},
{"name": "P10", "status": "idle", "orders": [], "good": 0, "cycleTime": None, "notVerladen": 0, "note": None, "placeholder": True},
]
# carts on the way, keyed by origin press (a press may have several → its column scrolls).
_ON_WAY: dict[str, list[dict[str, Any]]] = {
"P3": [{"cartId": "C-014a", "orders": _orders(("o1", 54), ("o2", 42)), "etaMin": 3}],
"P5": [{"cartId": "C-014b", "orders": _orders(("o3", 60)), "etaMin": 7}],
"P8": [
{"cartId": "C-015", "orders": _orders(("o1", 30), ("o4", 30), ("o2", 30), ("o3", 30)), "etaMin": 9},
{"cartId": "C-019", "orders": _orders(("o4", 36)), "etaMin": 14},
{"cartId": "C-020", "orders": _orders(("o1", 28)), "etaMin": 19},
],
}
_OVEN: list[dict[str, Any]] = [
{"cartId": "C-011", "orders": _orders(("o1", 88)), "totalMin": 120, "remainingMin": 42},
{"cartId": "C-012", "orders": _orders(("o2", 60), ("o4", 44)), "totalMin": 120, "remainingMin": 71},
{"cartId": "C-013", "orders": _orders(("o3", 72)), "totalMin": 120, "remainingMin": 12},
{"cartId": "C-016", "orders": _orders(("o1", 40), ("o2", 36)), "totalMin": 120, "remainingMin": 98},
{"cartId": "C-017", "orders": _orders(("o4", 80)), "totalMin": 120, "remainingMin": 104},
]
_FINISHED_BASE = {"carts": 47, "parts": 5120, "scrap": 63, "avgOvenMin": 118}
_FINISHED_BY_ORDER: dict[str, int] = {"o1": 1680, "o2": 1290, "o3": 1150, "o4": 1000}
_FINISHED_LIST: list[dict[str, Any]] = [
{"cartId": "C-010", "orders": _orders(("o2", 50), ("o3", 40)), "ago": "vor 8 min"},
{"cartId": "C-009", "orders": _orders(("o1", 64)), "ago": "vor 21 min"},
{"cartId": "C-008", "orders": _orders(("o4", 58)), "ago": "vor 35 min"},
{"cartId": "C-007", "orders": _orders(("o2", 44), ("o1", 20)), "ago": "vor 52 min"},
{"cartId": "C-006", "orders": _orders(("o3", 72)), "ago": "vor 1 h 6 min"},
]
@router.get("/overview", summary="Whole temper-line overview (SAMPLE data)")
def overview(range: str = Query(_DEFAULT_RANGE, description="finished-figures window: 12h|1d|7d|14d|30d")) -> dict[str, Any]:
code = range if range in _RANGE_FACTORS else _DEFAULT_RANGE
factor = _RANGE_FACTORS[code]
finished = {
"carts": round(_FINISHED_BASE["carts"] * factor),
"parts": round(_FINISHED_BASE["parts"] * factor),
"scrap": round(_FINISHED_BASE["scrap"] * factor),
"avgOvenMin": _FINISHED_BASE["avgOvenMin"],
"byOrder": [
{"orderNo": _ORDERS[k][0], "colorIndex": _ORDERS[k][1], "parts": round(v * factor)}
for k, v in _FINISHED_BY_ORDER.items()
],
}
return {
"plant": "Werk 1",
"generatedAt": datetime.now().astimezone().isoformat(),
"range": code,
"ovenTempC": 185,
"ovenAvgMin": 120,
"sample": True, # frontend shows a "Demo-Daten" hint while this is set
"presses": _PRESSES,
"onWay": _ON_WAY,
"oven": _OVEN,
"finished": finished,
"finishedList": _FINISHED_LIST,
}