Kjell Tore Guttormsen
464ad38b49
The core in modell/kjerne/ is an unmodified copy of sim.py, diagnose.py and forlop.py from den-dynamiske-rettsstaten @2badc25, pinned by sha256. kjor.py is the single entry point; kontroll.py reproduces the note's 22 data files and printed results. Skills: diagnose and forlop (Norwegian). Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>
236 lines
13 KiB
Python
236 lines
13 KiB
Python
"""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])
|