rippy/docker/api/eta.py

"""Restzeit-Schätzung für laufende Jobs — aus dem gemessenen Fortschritt.

Commander-Anforderung 26.07.2026: *„Der Server Status muss dringend überarbeitet
werden, das Dashboard soll ja quasi alles auf einen Blick zeigen"* — dazu eine
**ETA** im Dashboard und beim externen Worker.

## Warum hier und nicht im Browser

Der naheliegende Weg wäre, den Fortschritt im UI mitzuschreiben. Drei Gründe
dagegen: Ein Seitenwechsel setzt die Messreihe zurück; zwei offene Browser
zeigten verschiedene Zahlen; und die Kompression läuft womöglich auf einer
ANDEREN Maschine (genau dort wollte der Commander die ETA sehen). Die Reihe
gehört also dorthin, wo der Fortschritt ankommt.

## Warum nicht einfach „vergangene Zeit / Prozent"

Weil ein Job zwei völlig verschiedene Phasen hat: Der Rip dauert rund eine
Stunde, die Kompression Stunden bis Tage (auf der Rippy-VM gemessene 28-55 h je
4K-Film). Aus dem Gesamt-Mittel entstünde bei jedem Phasenwechsel eine
haarsträubende Zahl. Deshalb wird die Messreihe bei JEDEM Statuswechsel
verworfen und die Rate nur innerhalb der laufenden Phase bestimmt.

## Ehrlichkeit vor Zahl

Lieber „wird noch geschätzt" als eine erfundene Minute:

- Unter MINDEST_PUNKTE Messwerten gibt es keine Schätzung.
- Die Reihe muss MINDEST_SPANNE_SEKUNDEN abdecken UND MINDEST_FORTSCHRITT
  Prozentpunkte gestiegen sein — bei 4K bewegt sich eine halbe Stunde lang
  nichts, daraus ließe sich sonst „fertig in 3 Minuten" ableiten.
- Nur die letzten FENSTER Punkte zählen: HandBrake wird bei komplexen Szenen
  langsamer, die frühen Werte lügen dann.
- Über OBERGRENZE_SEKUNDEN wird nicht mehr aufs Detail gerechnet, sondern
  „mehr als 2 Tage" gesagt. Eine Zahl wie „51:23 h" wirkt genau, ist es aber
  nicht.
"""

MINDEST_PUNKTE = 2
MINDEST_SPANNE_SEKUNDEN = 60
MINDEST_FORTSCHRITT = 1
FENSTER = 10
OBERGRENZE_SEKUNDEN = 48 * 3600

# Status, für die eine Restzeit überhaupt Sinn hat.
LAUFENDE_STATUS = ("running", "processing", "ripping", "transcoding")


def beobachtung_hinzufuegen(reihe, status: str, progress: int, jetzt: float) -> dict:
    """Neuen Messpunkt anfügen (pure Funktion) → die aktualisierte Reihe.

    `reihe` ist {"status": str, "punkte": [[zeit, prozent], ...]} oder None.
    Bei Statuswechsel beginnt die Reihe neu — siehe Modul-Doku (Rip und
    Kompression haben nichts miteinander zu tun).

    Ein unveränderter Fortschritt wird NICHT als neuer Punkt angefügt, aber der
    letzte Punkt behält seine ursprüngliche Zeit. Das ist wichtig: Ein Stillstand
    verlängert damit automatisch die gemessene Spanne und macht die Schätzung
    langsamer — genau richtig, denn er heißt ja, dass es langsam vorangeht.
    """
    alt = reihe or {}
    punkte = list(alt.get("punkte") or []) if alt.get("status") == status else []
    if not punkte or punkte[-1][1] != progress:
        punkte.append([jetzt, progress])
    return {"status": status, "punkte": punkte[-FENSTER:]}


def restzeit_sekunden(reihe, jetzt: float) -> int:
    """Geschätzte Restzeit in Sekunden — oder -1 für „noch keine Aussage".

    -1 statt None, damit der Wert unverändert durch JSON und das
    Antwort-Modell passt (dasselbe Muster wie get_progress_from_line im
    Worker, wo -1 „keine Angabe" heißt).
    """
    punkte = (reihe or {}).get("punkte") or []
    if len(punkte) < MINDEST_PUNKTE:
        return -1
    erste_zeit, erster_prozent = punkte[0]
    letzte_zeit, letzter_prozent = punkte[-1]
    # Die Spanne bis JETZT, nicht bis zum letzten Punkt: sonst zeigt ein
    # stehender Job dauerhaft die Rate von vor drei Stunden.
    spanne = max(jetzt, letzte_zeit) - erste_zeit
    gewachsen = letzter_prozent - erster_prozent
    if spanne < MINDEST_SPANNE_SEKUNDEN or gewachsen < MINDEST_FORTSCHRITT:
        return -1
    rest_prozent = 100 - letzter_prozent
    if rest_prozent <= 0:
        return 0
    pro_prozent = spanne / gewachsen
    return int(rest_prozent * pro_prozent)


def formatiere_restzeit(sekunden: int) -> str:
    """Restzeit als Text fürs UI. Leer, wenn es keine Aussage gibt."""
    if sekunden is None or sekunden < 0:
        return ""
    if sekunden > OBERGRENZE_SEKUNDEN:
        return "mehr als 2 Tage"
    if sekunden < 60:
        return "unter einer Minute"
    minuten = sekunden // 60
    if minuten < 60:
        return f"noch ca. {minuten} min"
    stunden, rest_minuten = divmod(minuten, 60)
    if stunden < 24:
        return f"noch ca. {stunden} h {rest_minuten:02d} min"
    tage, rest_stunden = divmod(stunden, 24)
    return f"noch ca. {tage} Tag{'e' if tage > 1 else ''} {rest_stunden} h"


def schluessel(job_id: str) -> str:
    return f"eta:{job_id}"


#: Messreihen im eigenen Prozess — der Rückfall, wenn es kein Redis gibt.
#:
#: ## Der Befund des Commanders (29.08.2026)
#:
#: > „Über den kompletten vorgang steht dort ‚Restzeit wird gemessen' aber
#: > messung wird nicht abgeschlossen. Das heißt man hat kein ETA"
#:
#: Die Messreihe lag ausschließlich im Cache, und der Modul-Kopf sagte dazu:
#: „Der Cache (Redis) ist schon da". Im Container stimmt das. Auf einem
#: Windows-PC gibt es kein Redis — `cache_get` gab bei JEDEM Aufruf `None`
#: zurück, `beobachtung_hinzufuegen` legte also jedes Mal eine frische Reihe
#: mit EINEM Punkt an, und `restzeit_sekunden` braucht `MINDEST_PUNKTE = 2`.
#: Ergebnis: über den ganzen Rip hinweg „noch keine Aussage".
#:
#: Ein Wörterbuch im Prozess reicht hier vollkommen: Die Reihe ist ein paar
#: Zahlen, sie gilt nur für die Dauer eines Jobs, und im eigenständigen
#: Betrieb gibt es ohnehin nur diesen einen Prozess. Redis bleibt der bessere
#: Ort, wo es eins gibt — es überlebt einen API-Neustart.
_REIHEN: dict = {}

#: Wie lange eine Reihe im Prozess aufgehoben wird (wie `expire` im Cache).
REIHE_HALTBAR_SEKUNDEN = 86400


def _aufraeumen(jetzt: float) -> None:
    """Alte Reihen wegwerfen — sonst waechst das Woerterbuch unbegrenzt."""
    for k, (stand, _) in list(_REIHEN.items()):
        if jetzt - stand > REIHE_HALTBAR_SEKUNDEN:
            _REIHEN.pop(k, None)


def aktualisiere_und_schaetze(job_id: str, status: str, progress: int,
                             jetzt: float, cache_get, cache_set) -> dict:
    """Messreihe fortschreiben und die Restzeit zurückgeben.

    Gespeichert wird in BEIDEN Ablagen: im Cache, wo es einen gibt (er
    überlebt einen API-Neustart), und im Prozess, damit die Schätzung auch
    ohne Redis zustande kommt. Siehe `_REIHEN`.
    """
    if status not in LAUFENDE_STATUS or progress <= 0:
        return {"sekunden": -1, "text": ""}
    k = schluessel(job_id)
    try:
        reihe = cache_get(k)
    except Exception:
        reihe = None
    if not reihe:
        # Kein Cache (oder leer) — dann der eigene Vorrat.
        eintrag = _REIHEN.get(k)
        reihe = eintrag[1] if eintrag else None
    reihe = beobachtung_hinzufuegen(reihe, status, progress, jetzt)
    try:
        # Eine Reihe ohne Fortschritt ist nach einem Tag wertlos.
        cache_set(k, reihe, expire=REIHE_HALTBAR_SEKUNDEN)
    except Exception:
        pass
    _REIHEN[k] = (jetzt, reihe)
    _aufraeumen(jetzt)
    sekunden = restzeit_sekunden(reihe, jetzt)
    return {"sekunden": sekunden, "text": formatiere_restzeit(sekunden)}