feat(objekt): Zugfahrzeug-Flag am Objekttyp + Leitungs-Dashboard-Kacheln
CI / backend-tests (push) Failing after 1m26s
CI / frontend-build (push) Successful in 24s

Objekttyp.ist_zugfahrzeug (Migration 0011) grenzt die Fahrzeug-Zuordnung
serverseitig auf tatsächliche Zugfahrzeuge ein (422 bei Verstoß), Frontend
filtert die Fahrzeug-Dropdowns entsprechend.

Neue Dashboard-Endpunkte /dashboard/einsatzbereitschaft und
/dashboard/prueftermine plus DashboardPage mit drei Kacheln
(Einsatzbereitschaft, Prüftermine, Ablaufdaten). Aktive
Mindermengen-Genehmigung zählt als "eingeschränkt einsatzbereit", nicht als
voll einsatzbereit (U2/Karte 08).

Neue FehlbestandListePage: Fehlbestände eigenständig nachfüllen/Mindermenge
genehmigen, ohne komplette Kontrolle des Objekts zu starten (Backend
existierte bereits, nur das Frontend fehlte).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KC8HYvv6UkCVYheYiTw9DD
This commit is contained in:
2026-09-05 10:33:00 +02:00
co-authored by Claude Sonnet 5
parent 38b8ce1795
commit 54fc29690a
20 changed files with 894 additions and 19 deletions
+138
View File
@@ -4,6 +4,8 @@ from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
from app.models.geraet_instanz import GeraetInstanz
from app.models.kontrolle import Kontrolle, KontrollStatus
from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
from app.models.objekt import Objekt
from app.models.objektposition import Objektposition
@@ -89,3 +91,139 @@ async def bevorstehende_ablaufdaten(db: AsyncSession) -> list[dict]:
)
eintraege.sort(key=lambda e: e["verbleibende_tage"])
return eintraege
async def einsatzbereitschaft(db: AsyncSession) -> dict:
"""Karte-Nachtrag (Leitungs-Dashboard Kachel 1): ein Objekt ist einsatzbereit,
wenn kein offener Fehlbestand (deckt auch Geräte-Instanzen ab, da
Statuswechsel defekt/in_reparatur automatisch einen Fehlbestand erzeugt,
siehe app/services/geraet_instanz.py), kein abgelaufenes ablaufdatum, keine
überfällige Geräteprüfung und keine laufende Kontrolle (Sperre) vorliegt.
Aktive Mindermengen-Genehmigung erledigt den Fehlbestand fachlich NICHT
(Leitplanke U2/Karte 08) - das Objekt bleibt daher "eingeschränkt
einsatzbereit", nicht "einsatzbereit"."""
heute = datetime.now(timezone.utc).date()
objekte = (await db.execute(select(Objekt))).scalars().all()
offene_fehlbestaende = (
await db.execute(
select(Fehlbestand.objekt_id, Fehlbestand.id).where(Fehlbestand.status.in_(OFFEN_STATI))
)
).all()
objekt_ids_mit_fehlbestand: dict[int, set] = {}
for objekt_id, fehlbestand_id in offene_fehlbestaende:
objekt_ids_mit_fehlbestand.setdefault(objekt_id, set()).add(fehlbestand_id)
fehlbestand_ids_genehmigt = set(
(
await db.execute(
select(func.distinct(MindermengenGenehmigung.fehlbestand_id)).where(
MindermengenGenehmigung.status == MindermengeStatus.aktiv
)
)
)
.scalars()
.all()
)
objekt_ids_abgelaufen = set(
(
await db.execute(
select(func.distinct(Objektposition.objekt_id)).where(
Objektposition.ablaufdatum.is_not(None), Objektposition.ablaufdatum < heute
)
)
)
.scalars()
.all()
)
objekt_ids_pruefung_ueberfaellig = set(
(
await db.execute(
select(func.distinct(Objektposition.objekt_id))
.join(GeraetInstanz, GeraetInstanz.objektposition_id == Objektposition.id)
.where(GeraetInstanz.naechste_pruefung.is_not(None), GeraetInstanz.naechste_pruefung < heute)
)
)
.scalars()
.all()
)
objekt_ids_gesperrt = set(
(
await db.execute(
select(func.distinct(Kontrolle.objekt_id)).where(Kontrolle.status == KontrollStatus.in_bearbeitung)
)
)
.scalars()
.all()
)
bereit = eingeschraenkt = 0
details: list[dict] = []
for objekt in objekte:
gruende: list[str] = []
fehlbestand_ids = objekt_ids_mit_fehlbestand.get(objekt.id, set())
ungenehmigt = fehlbestand_ids - fehlbestand_ids_genehmigt
if ungenehmigt:
gruende.append("fehlbestand_offen")
if objekt.id in objekt_ids_abgelaufen:
gruende.append("ablaufdatum_abgelaufen")
if objekt.id in objekt_ids_pruefung_ueberfaellig:
gruende.append("pruefung_ueberfaellig")
if objekt.id in objekt_ids_gesperrt:
gruende.append("kontrolle_laeuft")
if not gruende and fehlbestand_ids and fehlbestand_ids <= fehlbestand_ids_genehmigt:
eingeschraenkt += 1
details.append(
{"objekt_id": objekt.id, "code": objekt.code, "name": objekt.name, "gruende": ["mindermenge_genehmigt"]}
)
elif not gruende:
bereit += 1
else:
details.append({"objekt_id": objekt.id, "code": objekt.code, "name": objekt.name, "gruende": gruende})
return {
"einsatzbereit": bereit,
"eingeschraenkt_einsatzbereit": eingeschraenkt,
"nicht_einsatzbereit": len(objekte) - bereit - eingeschraenkt,
"gesamt": len(objekte),
"details": details,
}
async def bevorstehende_prueftermine(db: AsyncSession, warnzeitraum_tage: int = 30) -> list[dict]:
"""Karte-Nachtrag (Leitungs-Dashboard Kachel 2): Geräteprüfungen (Karte 14),
nicht Ablaufdaten von Verbrauchsmaterial (bleibt eigene, bestehende Kachel/
Endpoint /dashboard/ablaufdaten) - Prüfung ist Wartungsintervall, Ablauf ist
Verbrauchsgut-Verfall, fachlich unterschiedlich (Nutzer-Entscheidung)."""
heute = datetime.now(timezone.utc).date()
result = await db.execute(
select(GeraetInstanz, Objektposition, Material)
.join(Objektposition, GeraetInstanz.objektposition_id == Objektposition.id)
.join(Material, Objektposition.material_id == Material.id)
.where(GeraetInstanz.naechste_pruefung.is_not(None))
)
eintraege = []
for geraet, objektposition, material in result.all():
status_wert, verbleibende_tage = _ablauf_status(geraet.naechste_pruefung, warnzeitraum_tage, heute)
if status_wert == "gueltig":
continue
eintraege.append(
{
"geraet_instanz_id": geraet.id,
"objekt_id": objektposition.objekt_id,
"objektposition_id": objektposition.id,
"material_id": material.id,
"material_name": material.name,
"seriennummer": geraet.seriennummer,
"geraet_status": geraet.status,
"naechste_pruefung": geraet.naechste_pruefung,
"verbleibende_tage": verbleibende_tage,
"status": status_wert,
}
)
eintraege.sort(key=lambda e: e["verbleibende_tage"])
return eintraege