den-dynamiske-rettsstaten/paper/diagnose.py

"""Flaskehalsdiagnose for straffesakskjeden fra egne tall.

Bruk (fra paper/):
    python3 diagnose.py eksempel_domstol.json

Inndata er en JSON-fil med ett objekt per ledd i kjeden, i rekkefølge. Alle
strømmer i saker per måned, beholdninger i saker, tider i måneder:

    {"ledd": [
      {"navn": "Domstol",
       "innstromning": 134,          # saker inn per måned nå
       "kapasitet": 105,             # saker leddet kan avgjøre per måned, slik det virker nå
       "restanse": 669,              # saker i kø nå
       "maal_tid": 3,                # mål for saksbehandlingstid (mnd)
       "prognose_innstromning": 134, # valgfritt: strømmen når balansen er gjenopprettet
       "maks_andel_utenom": 0.2,     # valgfritt: andel som kan ledes til forenklede reaksjoner
       "restanse_for_12_mnd": 400,   # valgfritt: gir observert vekst i restansen
       "restanse_maal": 300,         # valgfritt: akseptabel restanse (saker) i stedet for maal_tid;
                                     # oppgis begge, brukes restanse_maal
       "skjult_avgang": 0,           # valgfritt: saker per måned som avsluttes uten behandling og
                                     # som er regnet med i kapasitet (f.eks. henlagt pga. kapasitet);
                                     # trekkes fra kapasiteten når utnyttelse og behov regnes ut
       "innstromning_for_okning": 100}  # valgfritt, brukes bare av forlop.py: saker inn per måned
                                     # før strømmen begynte å øke
    ]}

Verktøyet følger trinn (c)-(e) i metoden (notatet, avsnitt 7.7 og 9.4-9.5):
(c) finn skranken som binder: høyest utnyttelse; er utnyttelsen lik på to desimaler,
    avgjør restanseveksten i forhold til kapasiteten;
(d) velg tiltakstype etter Theory of Constraints: utnytt, underordne, utvid;
(e) dimensjoner kapasiteten etter prognosen for strømmen, ikke etter køen, med
    den koordinerte regelen fra sim.py (simulate_endo, policy="koord"):
        Z_ny = (xi_min * F_prognose + Q_overskudd / T_avvikling) / ups_star.
Forløpet over tid og stabilitetssjekken ligger i forlop.py, som leser samme fil.
Verktøyet regner ikke med kapasitet som er bevilget, men ennå ikke virker: oppgi
kapasiteten som virker nå, og trekk det som allerede er bevilget, fra behovet.
Kolonnen «Saksbeh.tid» er restanse delt på kapasitet, ikke målt saksbehandlingstid.
Restansen over målet regnes som restanse - maal_tid * kapasitet.
sim.py trekker i tillegg fra minste behandlingstid m_D; eksempelfila er satt
slik at overskuddet blir det samme som i sim.py i måned 24.
Bruker ingenting fra sim.py og endrer ikke paper/data/.
"""
import json
import sys

UPS_STAR = 0.95   # målutnyttelse ved dimensjonering (ENDO["ups_star"] i sim.py)
UPS_HI = 1.0      # bestill når utnyttelsen er over denne (ENDO["ups_hi"])
T_CLEAR = 24.0    # restansen avvikles over denne tiden, mnd (ENDO["T_clear"])
L1, L2 = 6.0, 24.0  # bestilt kapasitet virker lineært fra L1 til L2 mnd etter vedtak

TALLFELT = ("innstromning", "kapasitet", "restanse", "maal_tid", "prognose_innstromning",
            "maks_andel_utenom", "restanse_for_12_mnd", "restanse_maal", "skjult_avgang",
            "innstromning_for_okning")


def tall(x, d=2):
    return f"{x:,.{d}f}".replace(",", " ").replace(".", ",")


def sjekk(data):
    """Stopp med en forståelig melding før regningen hvis inndata ikke kan brukes."""
    if not isinstance(data, dict) or not isinstance(data.get("ledd"), list) or not data["ledd"]:
        sys.exit('Feil i inndata: fila må ha en liste "ledd" med minst ett ledd.')
    for nr, ledd in enumerate(data["ledd"], 1):
        if not isinstance(ledd, dict):
            sys.exit(f"Feil i inndata, ledd {nr}: hvert ledd må være et objekt med felt og verdier.")
        hvor = f'ledd {nr} ({ledd.get("navn", "uten navn")})'
        for felt in ("navn", "innstromning", "kapasitet", "restanse"):
            if felt not in ledd:
                sys.exit(f'Feil i inndata, {hvor}: feltet "{felt}" mangler.')
        if not isinstance(ledd["navn"], str):
            sys.exit(f'Feil i inndata, {hvor}: "navn" må være tekst i anførselstegn.')
        if "maal_tid" not in ledd and "restanse_maal" not in ledd:
            sys.exit(f'Feil i inndata, {hvor}: oppgi "maal_tid" eller "restanse_maal".')
        for felt in TALLFELT:
            if felt not in ledd:
                continue
            verdi = ledd[felt]
            vist = json.dumps(verdi, ensure_ascii=False)
            try:
                if isinstance(verdi, bool):
                    raise TypeError
                x = float(verdi)
            except (TypeError, ValueError):
                sys.exit(f'Feil i inndata, {hvor}: "{felt}" må være et tall '
                         f"(med punktum som desimaltegn), ikke {vist}.")
            if not 0 <= x < float("inf"):   # fanger også NaN
                sys.exit(f'Feil i inndata, {hvor}: "{felt}" må være 0 eller et positivt, endelig tall, ikke {vist}.')
        if float(ledd["kapasitet"]) - float(ledd.get("skjult_avgang", 0.0)) <= 0:
            sys.exit(f"Feil i inndata, {hvor}: kapasiteten må være større enn 0, "
                     "også etter at skjult_avgang er trukket fra.")
        if float(ledd.get("maks_andel_utenom", 0.0)) >= 1:
            sys.exit(f'Feil i inndata, {hvor}: "maks_andel_utenom" er en andel og må være mindre enn 1 '
                     "(0.2 betyr 20 %).")


def analyser(ledd):
    F = float(ledd["innstromning"])
    skjult = float(ledd.get("skjult_avgang", 0.0))
    Z_tot = float(ledd["kapasitet"])          # all avgang, også den skjulte
    Z = Z_tot - skjult                        # avgang med behandling
    Q = float(ledd["restanse"])
    har_restansemaal = "restanse_maal" in ledd
    maal_tid = float(ledd["maal_tid"]) if "maal_tid" in ledd and not har_restansemaal else None
    Q_ref = float(ledd["restanse_maal"]) if har_restansemaal else maal_tid * Z_tot
    ups = F / Z
    # beholdning / avgangsrate; bare meningsfull når restansen er hele køen
    tid = None if har_restansemaal else Q / Z_tot
    if "restanse_for_12_mnd" in ledd:
        vekst = (Q - float(ledd["restanse_for_12_mnd"])) / 12.0
        vekst_kilde = "observert siste 12 mnd"
    else:
        vekst = max(F - Z_tot, 0.0)           # det som ikke avsluttes, legger seg i kø
        vekst_kilde = "innstrømning minus kapasitet"
    return dict(navn=ledd["navn"], F=F, Z=Z, Q=Q, Q_ref=Q_ref, maal_tid=maal_tid, ups=ups, tid=tid,
                vekst=vekst, vekst_kilde=vekst_kilde, skjult=skjult,
                Fp=float(ledd.get("prognose_innstromning", F)),
                har_prognose="prognose_innstromning" in ledd,
                F_for=float(ledd["innstromning_for_okning"]) if "innstromning_for_okning" in ledd else None,
                andel=float(ledd.get("maks_andel_utenom", 0.0)))


def skranke(rader):
    """Trinn (c): høyest utnyttelse, deretter raskest voksende restanse relativt til kapasiteten."""
    return max(rader, key=lambda a: (round(a["ups"], 2), a["vekst"] / a["Z"]))


def binder(b):
    return b["ups"] > UPS_HI or b["vekst"] > 0


def tiltak(a):
    """Trinn (d) og (e) for det bindende leddet."""
    ut = []
    xi_min = 1.0 - a["andel"]
    Q_exc = max(a["Q"] - a["Q_ref"], 0.0)          # restanse over målet
    ut.append("1. Utnytt leddet: fjern tapt tid i skranken (ikke regnet her).")
    if a["andel"] > 0:
        ups_u = xi_min * a["Fp"] / a["Z"]
        ut.append(f"2. Led saker utenom: med {tall(a['andel'] * 100, 0)} % til forenklede reaksjoner "
                  f"blir utnyttelsen {tall(ups_u)} (fra {tall(a['Fp'] / a['Z'])} "
                  + ("med prognosen)." if a["har_prognose"] else "med innstrømningen nå)."))
        if ups_u <= UPS_HI and Q_exc == 0:
            ut.append("   Det rekker: kapasiteten trenger ikke utvides.")
            return ut
    else:
        ut.append("2. Led saker utenom: ikke oppgitt at noen andel kan ledes utenom dette leddet.")
    behov = (xi_min * a["Fp"] + Q_exc / T_CLEAR) / UPS_STAR
    if behov * UPS_STAR > UPS_HI * a["Z"]:         # som sim.py: bestill når behovet før marginen er over kapasiteten
        okning = behov / a["Z"] - 1.0
        grunnlag = (f"prognosen ({tall(a['Fp'], 1)} saker/mnd)" if a["har_prognose"]
                    else f"innstrømningen nå ({tall(a['Fp'], 1)} saker/mnd; ingen prognose oppgitt)")
        ut.append(f"3. Utvid: dimensjoner etter {grunnlag}, ikke etter køen. "
                  f"Restanse over målet: {tall(Q_exc, 0)} saker, avviklet over {T_CLEAR:.0f} mnd.")
        ut.append(f"   Kapasitet: ({tall(xi_min)} x {tall(a['Fp'], 1)} + {tall(Q_exc, 0)}/{T_CLEAR:.0f}) / "
                  f"{tall(UPS_STAR)} = {tall(behov, 1)} saker/mnd, det vil si {tall(okning * 100, 0)} % mer"
                  + (f" enn kapasiteten etter skjult avgang ({tall(a['Z'], 1)} saker/mnd)."
                     if a["skjult"] > 0 else "."))
        ut.append(f"   Bestilt kapasitet virker gradvis fra {L1:.0f} til {L2:.0f} mnd etter vedtak. "
                  "Trekk fra kapasitet som er bevilget, men ikke virker ennå."
                  + (" Når restansen er avviklet, går andelen utenom tilbake, og behovet vurderes på nytt."
                     if a["andel"] > 0 else ""))
    else:
        ut.append("3. Utvid: ikke nødvendig med disse tallene.")
    return ut


def les(fil):
    try:
        with open(fil, encoding="utf-8") as f:
            data = json.load(f)
    except FileNotFoundError:
        sys.exit(f"Fant ikke fila {fil}.")
    except UnicodeDecodeError:
        sys.exit(f"Fila {fil} er ikke lagret som UTF-8.")
    except json.JSONDecodeError as e:
        sys.exit(f"Fila {fil} er ikke gyldig JSON: feil på linje {e.lineno}, kolonne {e.colno}.")
    except OSError:
        sys.exit(f"Kunne ikke lese {fil}. Er det en mappe, eller mangler du lesetilgang?")
    sjekk(data)
    return data


def main(fil):
    rader = [analyser(ledd) for ledd in les(fil)["ledd"]]
    w = max(22, max(len(a["navn"]) for a in rader) + 2)
    print("Trinn (c): utnyttelse og restanse i hvert ledd\n")
    print(f"{'Ledd':<{w}}{'Utnyttelse':>11}{'Saksbeh.tid':>13}{'Mål':>6}{'Restansevekst/mnd':>19}")
    for a in rader:
        maal = tall(a['maal_tid'], 1) if a['maal_tid'] is not None else "-"
        tid = tall(a['tid'], 1) if a['tid'] is not None else "-"
        print(f"{a['navn']:<{w}}{tall(a['ups']):>11}{tid:>13}{maal:>6}"
              f"{tall(a['vekst'], 0):>19}")
    if any(a["tid"] is not None for a in rader):
        print("(Saksbeh.tid er restanse delt på kapasitet, ikke målt saksbehandlingstid.)")
    b = skranke(rader)
    print()
    if not binder(b):
        vokser = [a["navn"] for a in rader if a["vekst"] > 0]
        if vokser:
            print(f"Ingen ledd er over full utnyttelse, men restansen vokser i: {', '.join(vokser)}. "
                  "Kontroller tallene for innstrømning og kapasitet der.")
        else:
            print(f"Ingen ledd binder: høyeste utnyttelse er {tall(b['ups'])} ({b['navn']}), "
                  "og ingen restanse vokser.")
        return
    print(f"Skranken som binder: {b['navn']} (utnyttelse {tall(b['ups'])}, restanse {tall(b['Q'], 0)} "
          f"mot mål {tall(b['Q_ref'], 0)}; vekst: {b['vekst_kilde']}).")
    like = [a["navn"] for a in rader if a is not b and round(a["ups"], 2) == round(b["ups"], 2)]
    if like:
        print(f"Samme utnyttelse på to desimaler: {', '.join(like)}. Verktøyet velger da leddet der "
              "restansen vokser raskest i forhold til kapasiteten. Forskjellen er for liten til å skille leddene.")
    andre = [f"{a['navn']} ({tall(a['ups'])})" for a in rader
             if a is not b and (a["ups"] > UPS_HI or a["vekst"] > 0)]
    if andre:
        print(f"Også over full utnyttelse eller med voksende restanse: {', '.join(andre)}. "
              "Skranken kan flytte seg dit når den første er løst.")
    plass = next(i for i, a in enumerate(rader) if a is b)
    nedstroms = [a["navn"] for a in rader[plass + 1:]]
    if nedstroms:
        print(f"Merk: i den grad {', '.join(nedstroms)} får sakene sine gjennom skranken, viser tallene der "
              "bare det som slipper gjennom. Når den løses, kan innstrømningen dit øke.")
    print("\nTrinn (d)-(e): tiltak etter Theory of Constraints, i denne rekkefølgen\n")
    for linje in tiltak(b):
        print(linje)
    print()
    if plass > 0:
        print("Å bremse leddene foran skranken til dens takt er ikke et tiltak: "
              "i modellens domstolseksempel gir det det dårligste utfallet for kriminaliteten.")
    print("Kontroller så de rettsstatlige skrankene (trinn f): det er sakstyper, ikke prosentandeler, som flyttes.")


if __name__ == "__main__":
    if len(sys.argv) != 2:
        sys.exit("Bruk: python3 diagnose.py <inndata.json>")
    main(sys.argv[1])