dg/grotti
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Grotti 2.5.1: Druckbericht, Statistik, Spulenakte und Protokoll-Fixes.

Browse Source
This commit is contained in:
Dominic Gundel committed 2026-10-17 15:32:51 +02:00
1 parent b277ca48aa
commit efd0801cd1
30 files changed
+3393 -178
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

web/app.py
+202 -10
View File
@@ -37,7 +37,7 @@ IDEA_STATUS = {"offen", "spaeter", "erledigt"}
PRINTER_FAMILIES = set(FAMILY_LABELS) | {""}
STEER_ACTIONS = {
"nozzle", "bed", "speed", "flow", "fan", "emergency_stop", "clear_error",
"exclude_object",
"exclude_object", "dryer_start", "dryer_stop", "slot_setzen",
}
@@ -580,6 +580,58 @@ def create_app(cfg):
def protokoll():
return page("protokoll.html", "Protokoll")
@app.get("/protokoll/<int:run_id>")
def protokoll_detail(run_id):
row = store.get_run_row(conn(), run_id)
if not row:
abort(404)
return page("bericht.html", "Druckbericht")
@app.get("/statistik")
def statistik_page():
return page("statistik.html", "Statistik")
@app.get("/etikett/lauf/<int:run_id>")
def etikett_lauf(run_id):
row = store.get_run_row(conn(), run_id)
if not row:
abort(404)
run = store.run_out(row)
if not run.get("chargen_nr"):
store.ensure_run_chargen(conn(), run_id, run.get("weight_used_g"))
row = store.get_run_row(conn(), run_id)
run = store.run_out(row)
if not run.get("chargen_nr"):
abort(404)
labels = store.list_run_labels(conn(), run_id)
mode = (request.args.get("mode") or "druck").strip()
qty = max(1, min(50, int(request.args.get("qty") or 1)))
if mode == "objekt":
labels = store.ensure_run_labels(conn(), run_id, run.get("objects") or [], qty=qty)
sheets = [lab for lab in labels if not lab.get("excluded")]
else:
sheets = [{"object_name": "", "chargen_nr": run["chargen_nr"], "excluded": False}]
if not sheets:
abort(404)
spools = store.etikett_spools_for_run(conn(), run)
base = grotti_config.public_base(cfg, request.url_root)
target_g = float(run.get("weight_planned_g") or run.get("weight_g") or 0)
for sheet in sheets:
sheet["qr_url"] = "%s/protokoll/%s" % (base, run_id)
sheet["target_g"] = target_g
if sheet.get("object_name"):
sheet["qr_url"] += "#objekt-%s" % quote(sheet["object_name"])
return render_template(
"etikett_lauf.html",
page="etikett",
title="Etikett %s" % run["chargen_nr"],
token="",
run=run,
sheets=sheets,
spools=spools,
target_g=target_g,
)
@app.get("/ideen")
def ideen():
return page("ideen.html", "Ideen")
@@ -672,6 +724,9 @@ def create_app(cfg):
item["queue_failed"] = queue.get("failed")
item["queue_waiting"] = queue.get("waiting", 0)
item["queue_waiting_next"] = queue.get("waiting_next", "")
live_run = store.find_live_run(conn(), host) if host else None
item["live_run_id"] = (live_run or {}).get("id")
item["live_chargen"] = (live_run or {}).get("chargen_nr") or ""
by_id = {row["host"]: row for row in rows}
if request.args.get("live", "1") == "0":
@@ -873,7 +928,7 @@ def create_app(cfg):
row = store.get_file_row(conn(), file_id)
if not row:
abort(404)
png = library.preview_png(row)
png = library.preview_png(row, conn=conn(), data_dir=cfg["data_dir"])
if not png:
abort(404)
return png, 200, {"Content-Type": "image/png", "Cache-Control": "private, max-age=300"}
@@ -883,8 +938,8 @@ def create_app(cfg):
row = store.get_file_row(conn(), file_id)
if not row:
raise GrottiError("Die Datei fehlt.", "datei", 404)
path = Path(row["stored_path"])
if not path.is_file():
path = library.resolve_stored_path(conn(), row, cfg["data_dir"])
if path is None or not path.is_file():
raise GrottiError("Die Datei liegt nicht mehr auf der Platte.", "datei", 404)
cache_dir = Path(cfg["data_dir"]) / "toolpath"
cache_dir.mkdir(parents=True, exist_ok=True)
@@ -1506,11 +1561,28 @@ def create_app(cfg):
@app.get("/api/v1/spulen/by-code/<path:code>")
def spool_by_code(code):
row = store.get_spool_by_code(conn(), code)
text = (code or "").strip()
if text.upper().startswith("D-"):
run = store.get_run_by_chargen(conn(), text)
if not run:
raise GrottiError("Kein Druck mit dieser Charge.", "charge", 404)
return jsonify({"kind": "lauf", "run": store.run_out(run), "redirect": "/protokoll/%s" % run["id"]})
row = store.get_spool_by_code(conn(), text)
if not row:
raise GrottiError("Keine Spule mit diesem Code.", "spule", 404)
return jsonify(row)
@app.get("/api/v1/spulen/<int:spool_id>/akte")
def spool_akte(spool_id):
spool = store.get_spool_row(conn(), spool_id)
if not spool:
raise GrottiError("Die Spule fehlt.", "spule", 404)
return jsonify({
"spool": store.spool_out(spool),
"runs": store.list_spool_runs(conn(), spool_id),
"stats": store.spool_usage_stats(conn(), spool_id),
})
@app.post("/api/v1/spulen")
def spool_create():
return jsonify(store.create_spool(conn(), spool_fields(body()))), 201
@@ -1531,6 +1603,18 @@ def create_app(cfg):
spool = store.get_spool_row(conn(), spool_id)
if not spool:
raise GrottiError("Die Spule fehlt.", "spule", 404)
live_run = store.find_live_run(conn(), host)
mm_now = float((live_run or {}).get("filament_used_mm") or 0)
printing = bool(
live_run and (live_run.get("state") or "") in ("in_progress", "printing", "paused")
)
if printing:
try:
snap = moonraker.query(host)
if isinstance(snap.get("filament_used"), (int, float)):
mm_now = float(snap["filament_used"])
except Exception:
pass
# Alte Rolle am Platz / externe Rolle am Host freimachen
if slot:
old = conn().execute(
@@ -1542,22 +1626,32 @@ def create_app(cfg):
"SELECT * FROM spools WHERE printer_host = %s AND ace_slot IS NULL AND id <> %s",
(host, spool_id),
).fetchall()
printer = store.printer_by_host(conn(), host) or {}
if printing:
closed = store.close_open_segments(conn(), live_run["id"], slot, mm_now)
if not closed and old:
# Erster Wechsel: alter Abschnitt von 0 bis jetzt
for row in old:
store.open_segment(conn(), live_run["id"], row["id"], slot, 0)
store.close_open_segments(conn(), live_run["id"], slot, mm_now)
for row in old:
store.update_spool(conn(), row["id"], {"printer_host": "", "ace_slot": None})
store.add_event(
conn(),
host,
(store.printer_by_host(conn(), host) or {}).get("name") or host,
printer.get("name") or host,
"spule",
"%s: „%s“ ins Regal%s" % (
(store.printer_by_host(conn(), host) or {}).get("name") or host,
printer.get("name") or host,
row["name"],
(" (Platz %s)" % slot) if slot else "",
),
"",
(live_run or {}).get("filename") or "",
run_id=(live_run or {}).get("id"),
)
updated = store.update_spool(conn(), spool_id, {"printer_host": host, "ace_slot": slot})
printer = store.printer_by_host(conn(), host) or {}
if printing:
store.open_segment(conn(), live_run["id"], spool_id, slot, mm_now)
store.add_event(
conn(),
host,
@@ -1568,7 +1662,8 @@ def create_app(cfg):
updated["name"],
(" auf Platz %s" % slot) if slot else "",
),
"",
(live_run or {}).get("filename") or "",
run_id=(live_run or {}).get("id"),
)
return jsonify(updated)
@@ -1704,6 +1799,36 @@ def create_app(cfg):
def run_list():
return jsonify(store.list_runs(conn()))
@app.get("/api/v1/laeufe/<int:run_id>")
def run_get(run_id):
row = store.get_run_row(conn(), run_id)
if not row:
raise GrottiError("Der Lauf fehlt.", "lauf", 404)
run = store.run_out(row)
if not run.get("chargen_nr"):
store.ensure_run_chargen(conn(), run_id)
row = store.get_run_row(conn(), run_id) or row
run = store.run_out(row)
if run.get("chargen_nr") and run.get("objects"):
store.ensure_run_labels(conn(), run_id, run.get("objects") or [], qty=1)
file_row = store.get_file_row(conn(), run["file_id"]) if run.get("file_id") else None
project = store.get_project(conn(), run["project_id"]) if run.get("project_id") else None
samples = store.list_run_samples(conn(), run_id)
# Anzeige auf ca. 1 Punkt / Minute verdichten
if len(samples) > 180:
step = max(1, len(samples) // 120)
samples = samples[::step]
return jsonify({
"run": run,
"spools": store.list_run_spools(conn(), run_id),
"segments": store.list_run_segments(conn(), run_id),
"samples": samples,
"labels": store.list_run_labels(conn(), run_id),
"events": store.list_run_events(conn(), run_id),
"file": store.file_out(file_row) if file_row else None,
"project": project,
})
@app.post("/api/v1/laeufe/abgleich")
def run_sync():
library.sync_watch(conn(), cfg, force=False)
@@ -1726,4 +1851,71 @@ def create_app(cfg):
row = store.run_out(current)
return jsonify(row)
@app.get("/api/v1/statistik")
def statistik_api():
since = request.args.get("since") or None
until = request.args.get("until") or None
host = (request.args.get("host") or "").strip()
material = (request.args.get("material") or "").strip()
brand = (request.args.get("brand") or "").strip()
project_id = optional_id(request.args.get("project_id"))
return jsonify(store.statistics(
conn(),
since=since,
until=until,
host=host,
material=material or "",
brand=brand or "",
project_id=project_id,
))
@app.get("/api/v1/statistik.csv")
def statistik_csv():
import csv
import io
from flask import Response
since = request.args.get("since") or None
until = request.args.get("until") or None
host = (request.args.get("host") or "").strip()
material = (request.args.get("material") or "").strip()
brand = (request.args.get("brand") or "").strip()
runs = store.list_runs(
conn(),
limit=2000,
since=since,
until=until,
host=host,
material=material,
brand=brand,
)
buf = io.StringIO()
writer = csv.writer(buf, delimiter=";")
writer.writerow([
"id", "drucker", "datei", "zustand", "start", "ende", "dauer_s",
"soll_g", "ist_g", "quelle", "filament_eur", "strom_eur", "urteil", "charge",
])
for run in runs:
writer.writerow([
run["id"],
run.get("printer_name") or "",
run.get("filename") or "",
run.get("state") or "",
run.get("started_at") or "",
run.get("ended_at") or "",
run.get("duration_s") or 0,
run.get("weight_planned_g") or 0,
run.get("weight_used_g") or 0,
run.get("weight_source") or "",
run.get("filament_cost") or 0,
run.get("energy_cost") or 0,
run.get("verdict") or "",
run.get("chargen_nr") or "",
])
return Response(
buf.getvalue(),
mimetype="text/csv; charset=utf-8",
headers={"Content-Disposition": "attachment; filename=grotti-laeufe.csv"},
)
return app