den-dynamiske-rettsstaten/paper/diagnose.py

"""Flaskehalsdiagnose for straffesakskjeden fra egne tall (første versjon).

Bruk (fra paper/):
    ../.venv/bin/python 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 ledd kan avgjøre per måned
       "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
       "skjult_avgang": 0}           # valgfritt: saker per måned som avsluttes uten behandling
                                     # (f.eks. henlagt pga. kapasitet); trekkes fra kapasiteten
    ]}

Verktøyet følger trinn (c)-(e) i metoden (artikkelen, avsnitt 9.4-9.5):
(c) finn skranken som binder: høyest utnyttelse, deretter raskest voksende restanse;
(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 (simulering med egne tall) er ikke med i denne versjonen.
Restansen over målet regnes som restanse - maal_tid * kapasitet (Littles lov).
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


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


def analyser(ledd):
    F = float(ledd["innstromning"])
    Z = float(ledd["kapasitet"]) - float(ledd.get("skjult_avgang", 0.0))
    Q = float(ledd["restanse"])
    tid_maal = float(ledd["maal_tid"]) if "maal_tid" in ledd else None
    Q_ref = float(ledd["restanse_maal"]) if "restanse_maal" in ledd else tid_maal * Z
    ups = F / Z
    # Littles lov: beholdning / avgangsrate; bare meningsfull når restansen er hele køen
    tid = Q / Z if "restanse_maal" not in ledd else None
    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 = F - min(F, Z)                     # det som ikke avgjøres, 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, tid_maal=tid_maal, ups=ups, tid=tid,
                vekst=vekst, vekst_kilde=vekst_kilde,
                Fp=float(ledd.get("prognose_innstromning", F)),
                andel=float(ledd.get("maks_andel_utenom", 0.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 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_HI * a["Z"]:
        okning = behov / a["Z"] - 1.0
        ut.append(f"3. Utvid: dimensjoner etter prognosen ({tall(a['Fp'], 1)} saker/mnd), 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.")
        ut.append(f"   Bestilt kapasitet virker gradvis fra {L1:.0f} til {L2:.0f} mnd etter vedtak."
                  + (" 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 main(fil):
    with open(fil, encoding="utf-8") as f:
        data = json.load(f)
    rader = [analyser(l) for l in data["ledd"]]
    print("Trinn (c): utnyttelse og restanse i hvert ledd\n")
    print(f"{'Ledd':<22}{'Utnyttelse':>11}{'Saksbeh.tid':>13}{'Mål':>6}{'Restansevekst/mnd':>19}")
    for a in rader:
        maal = tall(a['tid_maal'], 1) if a['tid_maal'] is not None else "-"
        tid = tall(a['tid'], 1) if a['tid'] is not None else "-"
        print(f"{a['navn']:<22}{tall(a['ups']):>11}{tid:>13}{maal:>6}"
              f"{tall(a['vekst'], 0):>19}")
    # skranken: høyest utnyttelse, deretter raskest voksende restanse relativt til kapasiteten
    b = max(rader, key=lambda a: (round(a["ups"], 2), a["vekst"] / a["Z"]))
    print()
    if b["ups"] <= UPS_HI and b["vekst"] <= 0:
        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']}).")
    andre = [a["navn"] 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 flytter seg dit når den første er løst.")
    nedstroms = [a["navn"] for a in rader[rader.index(b) + 1:]]
    if nedstroms:
        print(f"Merk: tallene for {', '.join(nedstroms)} viser bare det som slipper gjennom skranken. "
              "Når den løses, øker innstrømningen dit.")
    print("\nTrinn (d)-(e): tiltak etter Theory of Constraints, i denne rekkefølgen\n")
    for linje in tiltak(b):
        print(linje)
    print("\nÅ bremse leddene foran skranken til dens takt er ikke et tiltak: i modellen gir det det dårligste utfallet.")
    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: python diagnose.py <inndata.json>")
    main(sys.argv[1])