feat: model core, launcher, reproduction check and the first two skills
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>
This commit is contained in:
commit
464ad38b49
SSH key fingerprint:
SHA256:JakMjO6FTBBzN0Bhfj9saOoEjaFxlSdYuZQQpM/lF9Q
51 changed files with 11207 additions and 0 deletions
18
modell/kjerne/KILDE.json
Normal file
18
modell/kjerne/KILDE.json
Normal file
|
|
@ -0,0 +1,18 @@
|
|||
{
|
||||
"repo": "https://git.fromaitochitta.com/open/den-dynamiske-rettsstaten",
|
||||
"commit": "2badc25e5134ee26309bd8c3bd008ac64afe55a4",
|
||||
"filer": {
|
||||
"sim.py": {
|
||||
"kilde": "paper/sim.py",
|
||||
"sha256": "96857a92492f5768bc5d1d98bec30f4292b603c1273913e69901b6a1b220a01e"
|
||||
},
|
||||
"diagnose.py": {
|
||||
"kilde": "paper/diagnose.py",
|
||||
"sha256": "e5a20c68615836dd854c1414346c7932037eefaf1adb29a18458dbd590a3367a"
|
||||
},
|
||||
"forlop.py": {
|
||||
"kilde": "paper/forlop.py",
|
||||
"sha256": "5b369e30f98783b712f76e5dc90bcea4f711603944690c4f19da01efd29c6e21"
|
||||
}
|
||||
}
|
||||
}
|
||||
236
modell/kjerne/diagnose.py
Normal file
236
modell/kjerne/diagnose.py
Normal file
|
|
@ -0,0 +1,236 @@
|
|||
"""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])
|
||||
527
modell/kjerne/forlop.py
Normal file
527
modell/kjerne/forlop.py
Normal file
|
|
@ -0,0 +1,527 @@
|
|||
"""Forløp over tid og stabilitetssjekk for leddet som binder, fra egne tall.
|
||||
|
||||
Bruk (fra paper/, med samme Python-miljø som sim.py):
|
||||
../.venv/bin/python forlop.py eksempel_domstol.json
|
||||
../.venv/bin/python forlop.py norge_2025.json "Tingrett, meddomsrett" # velg ledd selv
|
||||
|
||||
Leser samme fil som diagnose.py. To ting kan legges til:
|
||||
|
||||
{"styring": {"forsinkelse": 3, # måneder fra kriminalitetsbildet endrer seg til innsatsen
|
||||
# er endret (statistikk + beslutning + iverksetting)
|
||||
"reaktivitet": 3}, # prosent mer innsats per prosent avvik fra målet
|
||||
"ledd": [{..., "innstromning_for_okning": 100}]} # saker inn per måned før strømmen økte
|
||||
|
||||
Uten "styring" brukes regime A i notatet (3 og 3). Uten "innstromning_for_okning"
|
||||
antas leddet 95 % utnyttet før økningen, som i notatet.
|
||||
|
||||
Hva skriptet gjør:
|
||||
1. Stabilitetssjekk: ligger forsinkelsen under den kritiske forsinkelsen for denne
|
||||
reaktiviteten (notatet, proposisjon 1 og avsnitt 9.3; critical_delay i sim.py)?
|
||||
2. Forløp: leddet som binder, legges inn som modellens ledd med kø (domstolen i
|
||||
avsnitt 9.5), og den koordinerte regelen kjøres måned for måned. Størrelsen på
|
||||
økningen settes slik at regelens første budsjettvedtak ser prognosen i fila.
|
||||
Restansen legges inn direkte: forløpet starter i restansen i fila, på
|
||||
tidspunktet tallene gjelder. «Nå» er første budsjettvedtak, tre måneder etter.
|
||||
3. Resten av kjeden: politiets innsats, andelen ledet utenom, og hva flere avgjorte
|
||||
saker betyr for leddene som står etter i fila.
|
||||
|
||||
Kapasitet, restanse og saksbehandlingstid følger tallene i fila. Kriminalitet,
|
||||
tillit og politiets innsats følger modellens stiliserte parametere (tabell 3 i
|
||||
notatet), som ikke er kalibrert på norske data.
|
||||
|
||||
Bruker sim.py uendret og skriver ingenting til paper/data/. For å kunne starte i
|
||||
restansen i fila har skriptet sin egen utgave av regnestegene i simulate_endo
|
||||
(simuler() under). Hver kjøring kontrollerer at den gir nøyaktig samme svar som
|
||||
simulate_endo når restansen ikke legges inn.
|
||||
"""
|
||||
import json
|
||||
import math
|
||||
import re
|
||||
import sys
|
||||
|
||||
import numpy as np
|
||||
from scipy.optimize import brentq
|
||||
|
||||
import diagnose
|
||||
from diagnose import tall
|
||||
from sim import (CH0, DT, ENDO, P, UPS0, VARSIGMA, Omega, critical_delay, deriv_chain_xz,
|
||||
init_chain, predict_F, simulate_endo)
|
||||
|
||||
J0 = P["gamma"] * P["Lam0"] * P["T0"] # modellens strøm før økningen (0,10 per måned)
|
||||
NAA = 24.0 # modellmåneden for første budsjettvedtak etter økningen
|
||||
OBS = NAA - ENDO["lagD"] # vedtaket bygger på tall som er lagD måneder gamle
|
||||
SLUTT = 180.0 # som tabell 7 og 10 i notatet
|
||||
LAM_MYK = P["Lam0"] - 0.05 # myk grense for tillit (0,75), som i mpc_step
|
||||
K_GULV, T_GULV = 0.80, 0.50 # under dette (1 = før økningen) regnes forløpet ikke
|
||||
UPS_FOR_MIN = ENDO["ups_lo"] + 0.01 # under dette bygger modellens regel ned kapasitet før økningen
|
||||
UPS_FOR_MAKS = 0.98 # over dette bestiller den kapasitet før økningen
|
||||
RADER = (-3, 0, 6, 12, 18, 24, 36, 48, 60, 72, 96, 120, 156)
|
||||
|
||||
|
||||
def idx(mnd):
|
||||
return int(round(mnd / DT))
|
||||
|
||||
|
||||
def les_styring(data):
|
||||
s = data.get("styring", {})
|
||||
if not isinstance(s, dict):
|
||||
sys.exit('Feil i inndata: "styring" må være et objekt med "forsinkelse" og "reaktivitet".')
|
||||
ut = {}
|
||||
for felt, standard, maks in (("forsinkelse", 3.0, 60.0), ("reaktivitet", 3.0, 50.0)):
|
||||
verdi = s.get(felt, standard)
|
||||
try:
|
||||
x = float("nan") if isinstance(verdi, bool) else float(verdi)
|
||||
except (TypeError, ValueError):
|
||||
x = float("nan")
|
||||
if not 0 < x <= maks:
|
||||
sys.exit(f'Feil i inndata, styring: "{felt}" må være et tall større enn 0 og høyst {maks:.0f}, '
|
||||
f"ikke {json.dumps(verdi, ensure_ascii=False)}.")
|
||||
ut[felt] = x
|
||||
ut["oppgitt"] = "styring" in data
|
||||
return ut
|
||||
|
||||
|
||||
def stabilitet(tau, kp):
|
||||
"""Lineær stabilitet i referanselikevekten: x' = -a x - b x(t - tau)."""
|
||||
c = P["gamma"] * P["Lam0"] # fjerningsrate per enhet innsats (0,10 per måned)
|
||||
a = c * P["T0"] + P["delta"]
|
||||
b = c * kp
|
||||
tc = critical_delay(a, b)
|
||||
print("Stabilitetssjekk: tåler styringen sin egen forsinkelse?\n")
|
||||
print(f"Forsinkelse {tall(tau, 1)} mnd, reaktivitet {tall(kp, 1)} "
|
||||
f"(10 % over målet gir {tall(kp * 10, 0)} % mer innsats).")
|
||||
if math.isinf(tc):
|
||||
print("Stabil uansett forsinkelse: reaksjonen er svakere enn systemets egen demping "
|
||||
f"(reaktivitet høyst {tall(a / c, 1)}).")
|
||||
return True
|
||||
stabil = tau < tc
|
||||
print(f"Kritisk forsinkelse: {tall(tc, 1)} mnd. " +
|
||||
(f"Forsinkelsen er kortere, så styringen er stabil (margin {tall(tc - tau, 1)} mnd)." if stabil else
|
||||
"Forsinkelsen er lengre, så styringen skaper svingninger selv: satsinger og kutt avløser "
|
||||
f"hverandre. Ved stabilitetsgrensen er perioden om lag "
|
||||
f"{tall(2 * math.pi / math.sqrt(b * b - a * a), 0)} mnd."))
|
||||
if critical_delay(a, c * 50.0) < tau:
|
||||
kp_maks = brentq(lambda k: critical_delay(a, c * k) - tau, a / c + 1e-6, 50.0)
|
||||
print(f"Med denne forsinkelsen er største stabile reaktivitet {tall(kp_maks, 1)}; "
|
||||
f"tommelfingerregelen gir under {tall(math.pi / (2 * c * tau), 1)}.")
|
||||
if not stabil:
|
||||
print(f"To veier ut: demp reaksjonen, eller kort forsinkelsen ned til under {tall(tc, 1)} mnd.")
|
||||
print("(Regnet med modellens stiliserte tall: innsatsen fjerner 10 % av kriminaliteten per måned, "
|
||||
"og 2 % faller fra av seg selv.)")
|
||||
return stabil
|
||||
|
||||
|
||||
def modell(b):
|
||||
"""Skaler leddet til modellens enheter. Returnerer skala, kjede, regelparametere, målene
|
||||
regelens første vedtak skal se (prognose og restanse over målet), og køen som svarer til
|
||||
restansen i fila, alt i modellens enheter."""
|
||||
Z_u = b["Z"]
|
||||
antatt = b["F_for"] is None
|
||||
F_for = UPS0 * Z_u if antatt else b["F_for"]
|
||||
hva = (f"innstrømningen før økningen ({tall(F_for, 1)} saker/mnd"
|
||||
+ (f", antatt {tall(UPS0 * 100, 0)} % av kapasiteten)" if antatt else ")"))
|
||||
if not UPS_FOR_MIN * Z_u <= F_for <= UPS_FOR_MAKS * Z_u:
|
||||
sys.exit(f"Forløpet forutsetter et ledd som var i balanse før økningen: {hva} må ligge mellom "
|
||||
f"{tall(UPS_FOR_MIN * 100, 0)} og {tall(UPS_FOR_MAKS * 100, 0)} % av kapasiteten "
|
||||
f"({tall(Z_u, 1)}). Utenfor det ville modellens regel ha endret kapasiteten allerede før "
|
||||
"økningen.")
|
||||
if b["Fp"] <= F_for:
|
||||
sys.exit(f"Forløpet har ingenting å vise for {b['navn']}: prognosen ({tall(b['Fp'], 1)} saker/mnd) "
|
||||
f"er ikke høyere enn {hva}.")
|
||||
skala = F_for / J0 # saker per måned per modellenhet
|
||||
Z = Z_u / skala
|
||||
tau_ref = b["Q_ref"] / Z_u # målet som måneder av kapasiteten modellen bruker
|
||||
if tau_ref <= 0:
|
||||
sys.exit("Forløpet trenger et mål for saksbehandlingstid eller restanse som er større enn 0.")
|
||||
m = tau_ref - J0 / (Z * (1.0 / tau_ref + VARSIGMA)) # som MD i sim.py: gir tau_ref før økningen
|
||||
ch = dict(CH0, Z=Z, m=m)
|
||||
e = dict(xi_min=1.0 - b["andel"])
|
||||
# Modellen viser restansen som (m + Q/Z) * Z. Køen Q som gir restansen i fila, er derfor
|
||||
# restansen minus m * Z; modellen kan ikke ha mindre restanse enn m * Z.
|
||||
return dict(skala=skala, ch=ch, e=e, F_for=F_for, antatt=antatt, tau_ref=tau_ref,
|
||||
F_maal=b["Fp"] / skala, Q_maal=max(b["Q"] - b["Q_ref"], 0.0) / skala,
|
||||
Q_inn=max(b["Q"] / skala - m * Z, 0.0), Q_min=m * Z)
|
||||
|
||||
|
||||
def simuler(policy, tau, p, ch, e, kp, tend, inn=None):
|
||||
"""Samme regnesteg som simulate_endo i sim.py for policy "koord" og "sched" (uten
|
||||
kvalitetssløyfe og uten tidsplaner for kapasitet og andel utenom), men køen kan settes
|
||||
direkte: inn = (steg, dQ) legger dQ til køen i det tidssteget (køen blir aldri negativ). Uten
|
||||
inn skal svaret være nøyaktig det samme som fra simulate_endo; kontroller_integrator() prøver
|
||||
det i hver kjøring. Endres simulate_endo, må denne endres likt."""
|
||||
e = dict(ENDO, **(e or {}))
|
||||
n = int(tend / DT) + 1
|
||||
t = np.arange(n) * DT
|
||||
K = np.ones(n); U = np.full(n, p["U0"]); L = np.full(n, p["Lam0"]); T = np.full(n, p["T0"])
|
||||
Q = np.zeros(n); tauD = np.zeros(n); JD = np.zeros(n); RD = np.zeros(n); Fa = np.zeros(n)
|
||||
Z = np.full(n, ch["Z"]); xi = np.ones(n)
|
||||
Q0, tau_ref = init_chain(dict(ch, Zfun=None, xifun=None), p)
|
||||
Q[0] = Q0
|
||||
lag = int(round(tau / DT)); lagD = int(round(e["lagD"] / DT))
|
||||
xi_tgt = 1.0
|
||||
binding = False
|
||||
orders = []
|
||||
Z_down = None
|
||||
log = []
|
||||
for k in range(n - 1):
|
||||
if inn is not None and k == inn[0]:
|
||||
Q[k] = max(Q[k] + inn[1], 0.0) # opp eller ned til restansen i fila
|
||||
tk = t[k]
|
||||
Kobs = K[k - lag] if k - lag >= 0 else 1.0
|
||||
T[k] = np.clip(p["T0"] + kp * (Kobs - p["Kstar"]), 0.0, p["Tmax"])
|
||||
i = k - lagD
|
||||
if i >= 0:
|
||||
Q_hat, F_hat = Q[i], Fa[i]
|
||||
else:
|
||||
Q_hat, F_hat = Q0, 0.1
|
||||
at_month = abs(tk - round(tk)) < 1e-9
|
||||
if policy == "koord":
|
||||
Q_ref = (tau_ref - ch["m"]) * Z[k]
|
||||
Q_exc = max(Q_hat - Q_ref, 0.0)
|
||||
if at_month and k > 0:
|
||||
binding = binding or F_hat > e["ups_hi"] * Z[k]
|
||||
xi_tgt = float(np.clip((e["ups_star"] * Z[k] - Q_exc / e["T_clear"]) / max(F_hat, 1e-9),
|
||||
e["xi_min"], 1.0)) if binding else 1.0
|
||||
if at_month and k > 0 and abs(tk % e["period"]) < 1e-9:
|
||||
pending = sum(dZ * (1 - np.clip((tk - a) / (b - a), 0, 1)) for a, b, dZ in orders)
|
||||
Z_comm = Z[k] + pending
|
||||
i_ = max(i, 0)
|
||||
F1 = predict_F(K[i_], U[i_], L[i_], Q[i_], Omega(t[i_], p), p, ch, tau_ref, e["H_pred"], kp, lag)
|
||||
need, surplus_ref = e["xi_min"] * F1 + Q_exc / e["T_clear"], F1
|
||||
sig = f"F_pred={F1:.3f} Q_exc={Q_exc:.2f}"
|
||||
if need > e["ups_hi"] * Z_comm:
|
||||
dZ = e["kZ"] * (need / e["ups_star"] - Z_comm)
|
||||
orders.append((tk + e["L1"], tk + e["L2"], dZ)); Z_down = None
|
||||
log.append(f"mnd {tk:.0f}: bestill +{dZ / ch['Z'] * 100:.0f} % ({sig})")
|
||||
elif surplus_ref < e["ups_lo"] * Z[k] and pending < 1e-9:
|
||||
Z_down = surplus_ref / e["ups_star"]
|
||||
log.append(f"mnd {tk:.0f}: bygg ned mot {Z_down / ch['Z']:.2f} ({sig})")
|
||||
rate = sum(dZ / (b - a) for a, b, dZ in orders if a <= tk < b)
|
||||
if Z_down is not None and rate == 0.0 and Z[k] > Z_down:
|
||||
rate -= (Z[k] - Z_down) / e["tau_down"]
|
||||
Z[k + 1] = Z[k] + DT * rate
|
||||
xi[k + 1] = xi[k] + DT * (xi_tgt - xi[k]) / e["tau_xi"]
|
||||
elif policy != "sched":
|
||||
raise ValueError(policy)
|
||||
dK, dU, dL, dQ, tauD[k], JD[k], RD[k], Fa[k] = deriv_chain_xz(
|
||||
K[k], U[k], L[k], T[k], Q[k], Z[k], xi[k], 0.0, tk, p, ch, tau_ref, mu3=0.0)
|
||||
K[k + 1] = K[k] + DT * dK
|
||||
U[k + 1] = U[k] + DT * dU
|
||||
L[k + 1] = np.clip(L[k] + DT * dL, 1e-4, 1 - 1e-4)
|
||||
Q[k + 1] = max(Q[k] + DT * dQ, 0.0)
|
||||
for arr in (T, tauD, JD, RD, Fa):
|
||||
arr[-1] = arr[-2]
|
||||
return dict(t=t, K=K, U=U, L=L, T=T, Q=Q, tauD=tauD, JD=JD, RD=RD, Z=Z, xi=xi, tau_ref=tau_ref, log=log)
|
||||
|
||||
|
||||
def kontroller_integrator(tau, p, ch, e, kp):
|
||||
"""Stopp hvis simuler() uten innlagt restanse ikke gir nøyaktig samme svar som simulate_endo."""
|
||||
for policy in ("koord", "sched"):
|
||||
egen = simuler(policy, tau, p, ch, e, kp, SLUTT)
|
||||
fasit = simulate_endo(policy, tau=tau, p=p, ch=ch, e=e, kp=kp, tend=SLUTT)
|
||||
like = (egen["log"] == fasit["log"] and egen["tau_ref"] == fasit["tau_ref"]
|
||||
and all(np.array_equal(egen[n], fasit[n])
|
||||
for n in ("t", "K", "U", "L", "T", "Q", "tauD", "JD", "RD", "Z", "xi")))
|
||||
if not like:
|
||||
sys.exit("Intern kontroll feilet: regnestegene i forlop.py (simuler) gir ikke lenger samme svar "
|
||||
f'som simulate_endo i sim.py (policy "{policy}"). Er sim.py endret, må simuler() '
|
||||
"oppdateres likt. Forløpet er ikke regnet.")
|
||||
|
||||
|
||||
def sett_ved_vedtak(okning, steg, mo, tau, kp, H=ENDO["H_pred"]):
|
||||
"""Det den koordinerte regelen ser ved første vedtak (sim.py, simulate_endo, budsjettsløyfen):
|
||||
prognosen F1 og restansen over målet, begge fra tilstanden lagD måneder før. Returnerer også
|
||||
køen modellen selv har bygd opp på det tidspunktet."""
|
||||
p = dict(P, shock=okning, t_shock=(steg - 0.5) * DT)
|
||||
r = simulate_endo("koord", tau=tau, p=p, ch=mo["ch"], e=mo["e"], kp=kp, tend=OBS)
|
||||
i = idx(OBS)
|
||||
F1 = predict_F(r["K"][i], r["U"][i], r["L"][i], r["Q"][i], Omega(r["t"][i], p), p, mo["ch"],
|
||||
r["tau_ref"], H, kp, idx(tau))
|
||||
Q_exc = max(r["Q"][i] - (r["tau_ref"] - mo["ch"]["m"]) * mo["ch"]["Z"], 0.0)
|
||||
return F1, Q_exc, r["Q"][i]
|
||||
|
||||
|
||||
def kalibrer(mo, tau, kp):
|
||||
"""Finn økningen i pådraget, slik at første vedtak ser F_maal, og tidssteget den begynner, slik at
|
||||
restansen modellen selv bygger opp, kommer nærmest Q_maal. Økningen må begynne etter at budsjettet
|
||||
i måned 12 har sett sine tall, og før tallene til vedtaket i måned 24 er tatt ut. Returnerer
|
||||
(økning, steg, kant), der kant er "tidlig" eller "sent" når restansen ikke kan treffes innenfor
|
||||
det vinduet. Avviket som står igjen, dekkes ved at restansen i fila legges inn direkte (main)."""
|
||||
def okning_for(steg):
|
||||
f = lambda x: sett_ved_vedtak(x, steg, mo, tau, kp)[0] - mo["F_maal"]
|
||||
lav, hoy = 1.0001, 3.0
|
||||
if f(lav) > 0:
|
||||
sys.exit("Forløpet har ingenting å vise: prognosen ligger for nær innstrømningen før økningen.")
|
||||
if f(hoy) < 0:
|
||||
sys.exit("Forløpet dekker ikke disse tallene: prognosen ligger for høyt over innstrømningen før "
|
||||
f"økningen ({tall(mo['F_maal'] / J0, 1)} ganger; modellen rekker om lag 3).")
|
||||
return brentq(f, lav, hoy, xtol=1e-6)
|
||||
|
||||
def avvik(steg):
|
||||
x = okning_for(steg)
|
||||
return x, sett_ved_vedtak(x, steg, mo, tau, kp)[1] - mo["Q_maal"]
|
||||
|
||||
tidlig, sent = idx(NAA - 12.0 - ENDO["lagD"]) + 1, idx(OBS) - 1
|
||||
x_t, g_t = avvik(tidlig)
|
||||
if g_t <= 0: # selv den lengste forhistorien gir for liten restanse
|
||||
return x_t, tidlig, "tidlig"
|
||||
x_s, g_s = avvik(sent)
|
||||
if g_s >= 0: # selv den korteste gir for stor
|
||||
return x_s, sent, "sent"
|
||||
while sent - tidlig > 1:
|
||||
midt = (tidlig + sent) // 2
|
||||
x_m, g_m = avvik(midt)
|
||||
if g_m > 0:
|
||||
tidlig, x_t, g_t = midt, x_m, g_m
|
||||
else:
|
||||
sent, x_s, g_s = midt, x_m, g_m
|
||||
return (x_t, tidlig, None) if abs(g_t) <= abs(g_s) else (x_s, sent, None)
|
||||
|
||||
|
||||
def naar(mnd):
|
||||
if mnd == 0:
|
||||
return "nå"
|
||||
return f"om {mnd:.0f} mnd" if mnd > 0 else f"for {-mnd:.0f} mnd siden"
|
||||
|
||||
|
||||
def vedtak(r, mo):
|
||||
"""Budsjettvedtakene i forløpet, fra loggen til simulate_endo. Samme nedbygging år etter år
|
||||
vises én gang."""
|
||||
ut = []
|
||||
for linje in r["log"]:
|
||||
t = re.match(r"mnd (\d+): bestill \+(\d+) %", linje)
|
||||
if t:
|
||||
ut.append((float(t.group(1)) - NAA, f"bestill {t.group(2)} % mer kapasitet", int(t.group(2))))
|
||||
continue
|
||||
t = re.match(r"mnd (\d+): bygg ned mot ([\d.]+)", linje)
|
||||
if not t:
|
||||
sys.exit(f"Kjenner ikke igjen vedtaket «{linje}» fra sim.py; forlop.py må oppdateres.")
|
||||
tekst = ("bygg ned ved naturlig avgang mot om lag "
|
||||
f"{tall(float(t.group(2)) * mo['ch']['Z'] * mo['skala'], 0)} saker/mnd")
|
||||
if not (ut and ut[-1][1] == tekst):
|
||||
ut.append((float(t.group(1)) - NAA, tekst, None))
|
||||
return ut
|
||||
|
||||
|
||||
def nokkeltall(r, Z0, fra):
|
||||
t = r["t"]
|
||||
return dict(tid=r["tauD"].max(), tillit=r["L"].min(), krim=r["K"][t >= fra].mean(),
|
||||
kap=r["Z"].max() / Z0 - 1.0,
|
||||
kapaar=np.trapezoid(np.maximum(r["Z"] - Z0, 0) / Z0, t) / 12)
|
||||
|
||||
|
||||
def ikke_regnet(grunn):
|
||||
print(f"Forløp: ikke regnet. {grunn}")
|
||||
|
||||
|
||||
def main(fil, valgt=None):
|
||||
data = diagnose.les(fil)
|
||||
st = les_styring(data)
|
||||
tau, kp = st["forsinkelse"], st["reaktivitet"]
|
||||
rader = [diagnose.analyser(ledd) for ledd in data["ledd"]]
|
||||
if valgt is None:
|
||||
b = diagnose.skranke(rader)
|
||||
else:
|
||||
treff = [a for a in rader if a["navn"] == valgt]
|
||||
if not treff:
|
||||
sys.exit(f'Fant ikke leddet "{valgt}". Leddene i fila: ' + ", ".join(a["navn"] for a in rader) + ".")
|
||||
b = treff[0]
|
||||
|
||||
stabil = stabilitet(tau, kp)
|
||||
if not st["oppgitt"]:
|
||||
print('(Fila har ikke "styring"; forsinkelse 3 mnd og reaktivitet 3 er regime A i notatet.)')
|
||||
print()
|
||||
if valgt is None and not diagnose.binder(b):
|
||||
print(f"Forløp: ingen ledd binder (høyeste utnyttelse er {tall(b['ups'])}, {b['navn']}), "
|
||||
"så det er ikke noe tiltak å følge over tid.")
|
||||
return
|
||||
if not stabil:
|
||||
ikke_regnet("Når styringen selv skaper svingninger, finnes det ingen ny balanse å dimensjonere etter, "
|
||||
"og prognosen regelen bruker, blir tilfeldig. Rett styringen først (trinn (e) i metoden "
|
||||
"krever det), og kjør på nytt.")
|
||||
return
|
||||
|
||||
mo = modell(b)
|
||||
okning, steg, kant = kalibrer(mo, tau, kp)
|
||||
s, ch = mo["skala"], mo["ch"]
|
||||
Z0 = ch["Z"]
|
||||
F1, _, Q_egen = sett_ved_vedtak(okning, steg, mo, tau, kp)
|
||||
F2 = sett_ved_vedtak(okning, steg, mo, tau, kp, H=2 * ENDO["H_pred"])[0]
|
||||
if abs(F2 - F1) > 0.01 * F1:
|
||||
ikke_regnet("Stabilitetssjekken er bestått, men prognosen for strømmen har ikke falt til ro innen "
|
||||
f"horisonten på {tall(ENDO['H_pred'], 0)} mnd ({tall(F1 * s, 1)} saker/mnd, mot "
|
||||
f"{tall(F2 * s, 1)} med dobbel horisont): styringen er for nær grensen eller for treg. Da "
|
||||
"finnes det ikke ett tall å dimensjonere etter. Prøv med kortere forsinkelse.")
|
||||
return
|
||||
Q_u = max(b["Q"] - b["Q_ref"], 0.0)
|
||||
behov = ((1.0 - b["andel"]) * b["Fp"] + Q_u / diagnose.T_CLEAR) / diagnose.UPS_STAR
|
||||
|
||||
def bestilling(x): # som regelen: ingen bestilling når behovet før marginen er dekket
|
||||
return x / b["Z"] - 1.0 if x * ENDO["ups_star"] > ENDO["ups_hi"] * b["Z"] else 0.0
|
||||
|
||||
# Restansen i fila legges inn i tidssteget tallene til første vedtak tas ut (OBS): køen flyttes
|
||||
# fra det modellens egen forhistorie ga, til restansen i fila. Resten av tilstanden (kriminalitet,
|
||||
# tillit, politiinnsats) kommer fra forhistorien. Kjøringen uten tiltak får samme flytting, så
|
||||
# den starter høyere når regelen har ledet saker utenom i forhistorien (som i tabell 10).
|
||||
p = dict(P, shock=okning, t_shock=(steg - 0.5) * DT)
|
||||
kontroller_integrator(tau, p, ch, mo["e"], kp)
|
||||
inn = (idx(OBS), mo["Q_inn"] - Q_egen)
|
||||
r = simuler("koord", tau, p, ch, mo["e"], kp, SLUTT, inn=inn)
|
||||
r0 = simuler("sched", tau, p, ch, mo["e"], kp, SLUTT, inn=inn)
|
||||
K_min, T_min = float(r["K"][idx(OBS):].min()), float(r["T"][idx(OBS):].min())
|
||||
if K_min < K_GULV or T_min < T_GULV:
|
||||
ikke_regnet(f"Mens restansen i fila ({tall(b['Q'], 0)} saker, {tall(b['Q'] / b['Z'], 0)} måneders "
|
||||
f"kapasitet) avvikles, faller kriminaliteten i modellen til {tall(max(K_min, 0.0))} og "
|
||||
f"politiets innsats til {tall(T_min)} (1,00 = nivået før økningen). Skriptet regner ikke "
|
||||
f"forløpet når kriminaliteten kommer under {tall(K_GULV)} eller innsatsen under "
|
||||
f"{tall(T_GULV)}: modellen legger halvparten av virkningen av en sak til tidspunktet saken "
|
||||
"avgjøres, og det holder ikke når restansen er mange ganger større enn strømmen. Bruk "
|
||||
"diagnosen.")
|
||||
return
|
||||
alle = vedtak(r, mo)
|
||||
if any(mnd < 0 for mnd, _, _ in alle):
|
||||
sys.exit("Forløpet kan ikke legges med disse tallene: modellens regel gjør vedtak før økningen er "
|
||||
"observert. Kontroller kapasiteten og innstrømningen før økningen.")
|
||||
|
||||
print(f"Forløp over tid for {'leddet som binder' if valgt is None else 'leddet du valgte'}: {b['navn']}\n")
|
||||
print("Slik er tallene lagt inn i modellen:")
|
||||
print("- Leddet regnes som modellens ledd med kø: det tar imot sakene politiets innsats gir "
|
||||
"(domstolen i notatet).")
|
||||
print(f"- Fra fila: kapasitet {tall(b['Z'], 1)} saker/mnd"
|
||||
+ (" (etter skjult avgang)" if b["skjult"] > 0 else "")
|
||||
+ f", prognose {tall(b['Fp'], 1)} saker/mnd"
|
||||
+ ("" if b["har_prognose"] else " (innstrømningen nå; ingen prognose oppgitt)")
|
||||
+ f", restanse {tall(b['Q'], 0)} ({tall(Q_u, 0)} over målet på {tall(b['Q_ref'], 0)}), "
|
||||
f"høyst {tall(b['andel'] * 100, 0)} % ledet utenom.")
|
||||
print(f"- Før økningen: {tall(mo['F_for'], 1)} saker/mnd inn, utnyttelse {tall(mo['F_for'] / b['Z'])}"
|
||||
+ (' (antatt; oppgi "innstromning_for_okning" for å endre).' if mo["antatt"] else " (oppgitt)."))
|
||||
print(f"- Modellen gjenskaper prognosen med en varig økning i det strukturelle pådraget på "
|
||||
f"{tall((okning - 1) * 100, 0)} % som begynte for {tall(NAA - steg * DT, 0)} mnd siden"
|
||||
+ ({"sent": " (det seneste modellen tillater)", "tidlig": " (det tidligste modellen tillater)"}[kant]
|
||||
if kant else "") + ".")
|
||||
print(f"- «Nå» er første budsjettvedtak etter økningen. Tallene det bygger på, er "
|
||||
f"{tall(ENDO['lagD'], 0)} mnd gamle, som tallene i fila.")
|
||||
start_u = (mo["Q_inn"] + mo["Q_min"]) * s # restansen forløpet starter i, i saker
|
||||
egen_u = (Q_egen + mo["Q_min"]) * s # restansen modellens egen forhistorie ga
|
||||
print(f"- Restansen er lagt inn direkte: forløpet starter i {tall(start_u, 0)} saker på tidspunktet "
|
||||
f"tallene gjelder (for {tall(ENDO['lagD'], 0)} mnd siden)."
|
||||
+ (f" Modellens egen forhistorie ville gitt {tall(egen_u, 0)}."
|
||||
if abs(egen_u - start_u) > max(0.02 * b["Q_ref"], 1.0) else ""))
|
||||
if start_u - b["Q"] > 0.5:
|
||||
print(f"- Avvik: restansen i fila ({tall(b['Q'], 0)}) er lavere enn modellen kan ha. Med denne "
|
||||
f"kapasiteten regner modellen minst {tall(mo['Q_min'] / Z0, 1)} mnd behandlingstid, altså "
|
||||
f"{tall(start_u, 0)} saker.")
|
||||
if start_u - egen_u > max(0.25 * b["Q_ref"], 1.0):
|
||||
print(f"- Forbehold: en så stor restanse har vokst lenger enn modellens forhistorie på "
|
||||
f"{tall(NAA - steg * DT, 0)} mnd. Kriminalitet, tillit og politiinnsats starter derfor som om "
|
||||
"restansen var ny, og virkningen av lang saksbehandlingstid kommer i modellen først etter "
|
||||
"start. I virkeligheten har den alt virket en stund.")
|
||||
if b["andel"] > 0 and r["xi"][idx(NAA)] < 0.995:
|
||||
print(f"- I modellens forhistorie har den koordinerte regelen alt begynt å lede saker utenom "
|
||||
f"({tall((1 - r['xi'][idx(NAA)]) * 100, 0)} % nå).")
|
||||
if K_min < 0.995 or T_min < 0.995:
|
||||
print(f"- Forbehold: mens restansen avvikles, faller kriminaliteten i modellen til {tall(K_min)} og "
|
||||
f"politiets innsats til {tall(T_min)} (1,00 = nivået før økningen). Modellen legger halvparten "
|
||||
"av virkningen av en sak til tidspunktet saken avgjøres. Om det holder for gamle saker, er "
|
||||
"usikkert.")
|
||||
print("\nVedtak i forløpet (budsjettet vedtar én gang i året; bestilt kapasitet virker gradvis "
|
||||
f"fra {tall(ENDO['L1'], 0)} til {tall(ENDO['L2'], 0)} mnd etter vedtak):")
|
||||
for mnd, tekst, _ in alle:
|
||||
print(f"- {naar(mnd)}: {tekst}")
|
||||
if not alle:
|
||||
print("- ingen: regelen bestiller ikke kapasitet med disse tallene.")
|
||||
forste = next((pst for mnd, _, pst in alle if mnd == 0 and pst is not None), None)
|
||||
diag = round(bestilling(behov) * 100)
|
||||
if forste is not None and forste == diag:
|
||||
print(f" Første vedtak er det samme som diagnose.py regner ut ({diag} %).")
|
||||
elif forste is not None:
|
||||
print(f" diagnose.py regner {diag} % for første vedtak; forskjellen er avrunding i modellens "
|
||||
"prognose.")
|
||||
if len([1 for _, _, pst in alle if pst is not None]) > 1:
|
||||
print(" Senere bestillinger kommer fordi restansen har vokst, eller prognosen er endret, mens den "
|
||||
"første kapasiteten bygges opp.")
|
||||
|
||||
print(f"\n{'Mnd fra nå':>10}{'Kapasitet':>11}{'Utenom':>8}{'Restanse':>10}{'Saksbeh.tid':>13}"
|
||||
f"{'Politiinnsats':>15}{'Kriminalitet':>14}{'Tillit':>8}")
|
||||
for mnd in RADER:
|
||||
k = idx(mnd + NAA)
|
||||
print(f"{mnd:>10}{tall(r['Z'][k] * s, 1):>11}{tall((1 - r['xi'][k]) * 100, 0) + ' %':>8}"
|
||||
f"{tall(r['tauD'][k] * r['Z'][k] * s, 0):>10}{tall(r['tauD'][k], 1):>13}"
|
||||
f"{tall(r['T'][k]):>15}{tall(r['K'][k]):>14}{tall(r['L'][k]):>8}")
|
||||
print("(Kapasitet i saker/mnd, restanse i saker, saksbehandlingstid i mnd = restanse delt på kapasitet. "
|
||||
"Politiinnsats og kriminalitet: 1 = nivået før økningen. Tillit: 0,80 før økningen.)")
|
||||
|
||||
fra = steg * DT
|
||||
med, uten = nokkeltall(r, Z0, fra), nokkeltall(r0, Z0, fra)
|
||||
print(f"\nMed og uten tiltak i kjeden, fra økningen begynte til {tall(SLUTT - NAA, 0)} mnd fra nå:\n")
|
||||
print(f"{'':<34}{'Med tiltakene':>14}{'Uten':>10}")
|
||||
print(f"{'Lengste saksbehandlingstid (mnd)':<34}{tall(med['tid'], 1):>14}{tall(uten['tid'], 1):>10}")
|
||||
print(f"{'Laveste tillit':<34}{tall(med['tillit']):>14}{tall(uten['tillit']):>10}")
|
||||
print(f"{'Kriminalitet, gjennomsnitt':<34}{tall(med['krim']):>14}{tall(uten['krim']):>10}")
|
||||
print(f"{'Høyeste kapasitet':<34}{'+' + tall(med['kap'] * 100, 0) + ' %':>14}{'+0 %':>10}")
|
||||
print(f"{'Kapasitetsår over utgangsnivået':<34}{tall(med['kapaar'], 1):>14}{tall(0, 1):>10}")
|
||||
uten_u = (r0["Q"][idx(OBS)] + mo["Q_min"]) * s
|
||||
if abs(uten_u - start_u) >= 1.0:
|
||||
print(f"(Uten tiltak starter restansen i {tall(uten_u, 0)} saker, ikke {tall(start_u, 0)}: i den "
|
||||
"kjøringen er ingen saker ledet utenom før nå.)")
|
||||
if med["tillit"] < LAM_MYK:
|
||||
print(f"Tilliten kommer under modellens myke grense ({tall(LAM_MYK)}). Trinn (f) peker da på "
|
||||
"prosessuell kvalitet i tillegg.")
|
||||
sent = r["t"] > SLUTT - 60.0
|
||||
spenn = {n: (float(r[n][sent].min()), float(r[n][sent].max())) for n in ("K", "tauD", "Z")}
|
||||
i_ro = (spenn["K"][1] - spenn["K"][0] <= 0.01
|
||||
and spenn["tauD"][1] - spenn["tauD"][0] <= 0.05 * mo["tau_ref"]
|
||||
and spenn["Z"][1] - spenn["Z"][0] <= 0.01 * Z0)
|
||||
if i_ro:
|
||||
print(f"Forløpet faller til ro: kriminaliteten ender på {tall(r['K'][-1])} og saksbehandlingstiden på "
|
||||
f"{tall(r['tauD'][-1], 1)} mnd.")
|
||||
ups_slutt = r["JD"][-1] / r["Z"][-1]
|
||||
if ups_slutt < ENDO["ups_star"] - 0.03:
|
||||
print(f"Kapasiteten ender høyere enn strømmen krever (utnyttelse {tall(ups_slutt)} mot målet "
|
||||
f"{tall(ENDO['ups_star'])}): det som ble bestilt for å avvikle restansen, blir stående, fordi "
|
||||
f"regelen først bygger ned under {tall(ENDO['ups_lo'])}.")
|
||||
else:
|
||||
print("Forløpet har ikke falt til ro ved slutten. De siste fem årene ligger kriminaliteten mellom "
|
||||
f"{tall(spenn['K'][0])} og {tall(spenn['K'][1])}, saksbehandlingstiden mellom "
|
||||
f"{tall(spenn['tauD'][0], 1)} og {tall(spenn['tauD'][1], 1)} mnd og kapasiteten mellom "
|
||||
f"{tall(spenn['Z'][0] * s, 1)} og {tall(spenn['Z'][1] * s, 1)} saker/mnd. Sluttallene over er da "
|
||||
"ikke en ny balanse.")
|
||||
|
||||
print("\nResten av kjeden:")
|
||||
print(f"- Politiet: innsatsen går fra 1,00 til høyst {tall(r['T'].max())} og ender på {tall(r['T'][-1])}. "
|
||||
"Å bremse politiet til leddets takt er ikke et tiltak (diagnose.py).")
|
||||
if b["andel"] > 0:
|
||||
utenom = 1.0 - r["xi"]
|
||||
print(f"- Saker ledet utenom {b['navn']}: høyst {tall(utenom.max() * 100, 0)} % "
|
||||
f"({naar(round(r['t'][int(np.argmax(utenom))] - NAA))}) og {tall(utenom[-1] * 100, 0)} % til "
|
||||
"slutt. Hvor de sakene havner, er ikke regnet.")
|
||||
R_data = r["RD"][idx(OBS) - 1] # steget før restansen legges inn: avgangen følger køen i modellen
|
||||
topp, slutt = r["RD"].max() / R_data, r["RD"][-1] / R_data
|
||||
print(f"- Ut av {b['navn']}: i modellen avgjør leddet på det meste {tall((topp - 1) * 100, 0)} % flere saker "
|
||||
f"enn på tidspunktet tallene i fila gjelder ({naar(round(r['t'][int(np.argmax(r['RD']))] - NAA))}), "
|
||||
f"og {tall((slutt - 1) * 100, 0)} % flere til slutt.")
|
||||
plass = next(i for i, a in enumerate(rader) if a is b)
|
||||
for a in rader[plass + 1:]:
|
||||
if max(topp, slutt) <= 1.005:
|
||||
print(f"- {a['navn']}: uendret innstrømning herfra.")
|
||||
continue
|
||||
ny_topp, ny_slutt = a["F"] * topp, a["F"] * slutt
|
||||
dom = ("Leddet binder allerede, og presset øker." if a["ups"] > diagnose.UPS_HI else
|
||||
"Skranken flytter seg dit." if ny_slutt > diagnose.UPS_HI * a["Z"] else
|
||||
"Leddet er over full utnyttelse mens restansen avvikles, ikke varig."
|
||||
if ny_topp > diagnose.UPS_HI * a["Z"] else "Leddet tar det unna.")
|
||||
print(f"- {a['navn']}: får leddet sakene sine gjennom {b['navn']}, går innstrømningen fra "
|
||||
f"{tall(a['F'], 1)} til om lag {tall(ny_topp, 1)} per mnd på det meste og {tall(ny_slutt, 1)} til "
|
||||
f"slutt, mot kapasitet {tall(a['Z'], 1)}: utnyttelse {tall(ny_topp / a['Z'])} og "
|
||||
f"{tall(ny_slutt / a['Z'])}. {dom}")
|
||||
if not rader[plass + 1:]:
|
||||
print(f"- Fila har ingen ledd etter {b['navn']}.")
|
||||
print("\nLes forløpet slik: kapasitet, restanse og saksbehandlingstid følger tallene i fila. Politiinnsats, "
|
||||
"kriminalitet og tillit følger modellens stiliserte parametere, som ikke er kalibrert på norske data. "
|
||||
"Rekkefølgen og retningen er poenget, ikke tallene. Bare leddet som binder, har kø i modellen.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) not in (2, 3):
|
||||
sys.exit('Bruk: python forlop.py <inndata.json> ["navn på ledd"]')
|
||||
main(*sys.argv[1:])
|
||||
1077
modell/kjerne/sim.py
Normal file
1077
modell/kjerne/sim.py
Normal file
File diff suppressed because it is too large
Load diff
589
modell/kjor.py
Normal file
589
modell/kjor.py
Normal file
|
|
@ -0,0 +1,589 @@
|
|||
# Copyright (c) 2026 Kjell Tore Guttormsen. MIT License, see LICENSE.
|
||||
"""Én inngang til modellen bak notatet «Den dynamiske rettsstaten».
|
||||
|
||||
Bruk:
|
||||
python3 kjor.py [--data <mappe>] [--modell <mappe>] <kommando> [...]
|
||||
|
||||
Bare standardbiblioteket, Python 3.9 eller nyere. Kommandoene står i HJELP under.
|
||||
|
||||
Exit-koder: 0 ok, 1 forskjeller funnet av kontroll, 2 feil i inndata eller bruk,
|
||||
3 miljøet mangler (numpy eller scipy).
|
||||
|
||||
Valg av Python for kommandoene som trenger numpy og scipy: (1) denne tolken hvis den har
|
||||
numpy 2.0 eller nyere (numpy.trapezoid) og scipy; (2) ellers Python i den egne mappa
|
||||
<data>/venv, som kjører samme kommando på nytt; (3) ellers exit-kode 3 med en melding om
|
||||
hva som mangler. Ingenting installeres uten «miljo --opprett».
|
||||
|
||||
Grensesnitt for modulene i modell-mappa (scenario.py, kontroll.py, arbeidskopi.py):
|
||||
hver har ``main(argv: list[str]) -> int``, der argv er argumentene etter kommandonavnet.
|
||||
Før main kalles, ligger modell-mappa og modell-mappa/kjerne først i sys.path, slik at
|
||||
``import sim`` gir kjernen i samme modell-mappe. ``sys.exit("melding")`` gir exit-kode 2
|
||||
og meldingen uendret; ``sys.exit(n)`` gir exit-kode n.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
import traceback
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, NamedTuple, TypedDict
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
from types import ModuleType
|
||||
|
||||
HER = Path(__file__).resolve().parent
|
||||
VENV_MERKE = "DYNAMISK_RETTSSTAT_VENV" # satt når kjor.py er startet på nytt med Python fra den egne mappa
|
||||
KJERNEMODULER = ("sim", "diagnose", "forlop")
|
||||
MODELLMODULER = ("scenario", "kontroll", "arbeidskopi")
|
||||
|
||||
OK, FORSKJELLER, FEIL_BRUK, MILJO_MANGLER = 0, 1, 2, 3
|
||||
|
||||
HJELP = """\
|
||||
Bruk: python3 kjor.py [--data <mappe>] [--modell <mappe>] <kommando> [...]
|
||||
|
||||
Kommandoer:
|
||||
diagnose <fil> finn leddet som binder, fra egne tall (trenger bare Python)
|
||||
forlop <fil> ["ledd"] stabilitetssjekk og forløp måned for måned (numpy og scipy)
|
||||
scenario <fil> [--ut <mappe>] kjør en scenariofil (numpy og scipy)
|
||||
notatet [--ut <mappe>] kjør notatets simuleringer uendret, om lag 1-2 min (numpy og scipy)
|
||||
kontroll [--rask] kontroller at modellen regner som notatet
|
||||
miljo [--opprett] vis Python, numpy og scipy; --opprett installerer dem i en egen mappe
|
||||
arbeidskopi <mappe> lag en kopi av modellen som du kan endre
|
||||
|
||||
Valg før kommandoen:
|
||||
--data <mappe> mappa der den egne Python-mappa (venv) med numpy og scipy ligger
|
||||
(standard ~/.dynamisk-rettsstat)
|
||||
--modell <mappe> bruk modellen i en arbeidskopi i stedet for pluginens
|
||||
|
||||
Exit-koder: 0 ok, 1 kontroll fant forskjeller, 2 feil i inndata eller bruk,
|
||||
3 numpy eller scipy mangler.
|
||||
"""
|
||||
|
||||
TREGER_REGNEMILJO = ("forlop", "scenario", "notatet")
|
||||
KOMMANDOER = ("diagnose", "forlop", "scenario", "notatet", "kontroll", "miljo", "arbeidskopi")
|
||||
|
||||
|
||||
class Bruksfeil(Exception):
|
||||
"""Feil i kommandolinjen. Meldingen er til brukeren og gir exit-kode 2."""
|
||||
|
||||
|
||||
class Valg(NamedTuple):
|
||||
"""Kommandolinjen, tolket."""
|
||||
|
||||
argv: list[str] # hele kommandolinjen, slik den kom inn
|
||||
data: Path # mappa der den egne venv ligger
|
||||
data_oppgitt: bool # om --data ble oppgitt med en brukbar verdi
|
||||
modell: Path # modell-mappa som brukes (pluginens eller en arbeidskopi)
|
||||
kommando: str
|
||||
args: list[str] # argumentene etter kommandoen
|
||||
|
||||
|
||||
class Beskrivelse(TypedDict):
|
||||
"""Hva en Python-tolk har av numpy og scipy."""
|
||||
|
||||
python: str
|
||||
sti: str
|
||||
numpy: str | None
|
||||
trapezoid: bool
|
||||
scipy: str | None
|
||||
|
||||
|
||||
class Tolk(NamedTuple):
|
||||
"""Hvilken Python som skal regne: «denne», «venv» eller «mangler»."""
|
||||
|
||||
hvor: str
|
||||
python: Path | None
|
||||
mangler: list[str]
|
||||
|
||||
|
||||
# ---------------- kommandolinjen ----------------
|
||||
|
||||
def standard_datamappe() -> Path:
|
||||
"""Mappa som brukes når --data ikke er oppgitt."""
|
||||
return Path.home() / ".dynamisk-rettsstat"
|
||||
|
||||
|
||||
def datamappe(verdi: str | None) -> tuple[Path, bool]:
|
||||
"""Tom verdi, eller en verdi med «${» (ikke satt inn av Claude), betyr «ikke oppgitt»."""
|
||||
if verdi is None or not verdi.strip() or "${" in verdi:
|
||||
return standard_datamappe(), False
|
||||
return Path(verdi).expanduser().resolve(), True
|
||||
|
||||
|
||||
def modellmappe(verdi: str | None) -> Path:
|
||||
"""Modell-mappa: pluginens egen, eller en arbeidskopi (mappa selv eller mappa over modell/)."""
|
||||
if verdi is None:
|
||||
return HER
|
||||
if not verdi.strip():
|
||||
melding = "--modell må ha en mappe etter seg."
|
||||
raise Bruksfeil(melding)
|
||||
sti = Path(verdi).expanduser().resolve()
|
||||
for kandidat in (sti, sti / "modell"):
|
||||
if (kandidat / "kjerne").is_dir():
|
||||
return kandidat
|
||||
melding = f"Fant ingen modell i {sti}: mappa må ha undermappa kjerne/ eller modell/kjerne/."
|
||||
raise Bruksfeil(melding)
|
||||
|
||||
|
||||
def tolk_argumenter(argv: list[str]) -> Valg:
|
||||
"""Tolk de globale valgene og finn kommandoen."""
|
||||
verdier: dict[str, str | None] = {"--data": None, "--modell": None}
|
||||
i = 0
|
||||
while i < len(argv) and argv[i].startswith("-"):
|
||||
arg = argv[i]
|
||||
if arg in {"-h", "--hjelp", "--help"}:
|
||||
return Valg(argv=argv, data=standard_datamappe(), data_oppgitt=False, modell=HER, kommando="hjelp",
|
||||
args=[])
|
||||
navn, likhetstegn, verdi = arg.partition("=")
|
||||
if navn not in verdier:
|
||||
melding = f"Ukjent valg {arg}. Gyldige valg før kommandoen: --data <mappe>, --modell <mappe>."
|
||||
raise Bruksfeil(melding)
|
||||
if likhetstegn:
|
||||
i += 1
|
||||
elif i + 1 < len(argv):
|
||||
verdi = argv[i + 1]
|
||||
i += 2
|
||||
else:
|
||||
melding = f"{navn} må ha en mappe etter seg."
|
||||
raise Bruksfeil(melding)
|
||||
verdier[navn] = verdi
|
||||
if i >= len(argv):
|
||||
melding = "Mangler kommando."
|
||||
raise Bruksfeil(melding)
|
||||
kommando, args = argv[i], argv[i + 1:]
|
||||
if kommando not in KOMMANDOER and kommando != "_beskriv":
|
||||
melding = f"Ukjent kommando «{kommando}». Gyldige kommandoer: {', '.join(KOMMANDOER)}."
|
||||
raise Bruksfeil(melding)
|
||||
data, oppgitt = datamappe(verdier["--data"])
|
||||
return Valg(argv, data, oppgitt, modellmappe(verdier["--modell"]), kommando, args)
|
||||
|
||||
|
||||
# ---------------- Python-miljøet ----------------
|
||||
|
||||
def beskriv_denne() -> Beskrivelse:
|
||||
"""Python, numpy og scipy i tolken som kjører nå."""
|
||||
ut = Beskrivelse(python=platform.python_version(), sti=sys.executable, numpy=None, trapezoid=False, scipy=None)
|
||||
try:
|
||||
numpy = importlib.import_module("numpy")
|
||||
except ImportError:
|
||||
pass
|
||||
else:
|
||||
ut["numpy"] = str(getattr(numpy, "__version__", "ukjent versjon"))
|
||||
ut["trapezoid"] = hasattr(numpy, "trapezoid")
|
||||
try:
|
||||
importlib.import_module("scipy.optimize")
|
||||
scipy = importlib.import_module("scipy")
|
||||
except ImportError:
|
||||
pass
|
||||
else:
|
||||
ut["scipy"] = str(getattr(scipy, "__version__", "ukjent versjon"))
|
||||
return ut
|
||||
|
||||
|
||||
def mangler(beskrivelse: Beskrivelse) -> list[str]:
|
||||
"""Det som mangler for å regne med numpy og scipy, i ord."""
|
||||
ut = []
|
||||
if beskrivelse["numpy"] is None:
|
||||
ut.append("numpy")
|
||||
elif not beskrivelse["trapezoid"]:
|
||||
ut.append(f"numpy 2.0 eller nyere (har {beskrivelse['numpy']})")
|
||||
if beskrivelse["scipy"] is None:
|
||||
ut.append("scipy")
|
||||
return ut
|
||||
|
||||
|
||||
def beskriv_tolk(python: Path) -> Beskrivelse | None:
|
||||
"""Beskriv en annen Python-tolk ved å la den kjøre denne fila. None hvis det ikke går."""
|
||||
try:
|
||||
resultat = subprocess.run(
|
||||
[str(python), str(Path(__file__).resolve()), "_beskriv"],
|
||||
capture_output=True, text=True, encoding="utf-8", check=False,
|
||||
)
|
||||
except OSError:
|
||||
return None
|
||||
if resultat.returncode != 0:
|
||||
return None
|
||||
try:
|
||||
data = json.loads(resultat.stdout.strip().splitlines()[-1])
|
||||
except (ValueError, IndexError):
|
||||
return None
|
||||
return Beskrivelse(
|
||||
python=str(data["python"]), sti=str(data["sti"]), numpy=data["numpy"],
|
||||
trapezoid=bool(data["trapezoid"]), scipy=data["scipy"],
|
||||
)
|
||||
|
||||
|
||||
def venv_mappe(data: Path) -> Path:
|
||||
"""Den egne Python-mappa med numpy og scipy."""
|
||||
return data / "venv"
|
||||
|
||||
|
||||
def venv_python(data: Path) -> Path:
|
||||
"""Python-tolken i den egne mappa."""
|
||||
if os.name == "nt":
|
||||
return venv_mappe(data) / "Scripts" / "python.exe"
|
||||
return venv_mappe(data) / "bin" / "python"
|
||||
|
||||
|
||||
def velg_tolk(data: Path) -> Tolk:
|
||||
"""Velg Python for kommandoene som trenger numpy og scipy."""
|
||||
mangel = mangler(beskriv_denne())
|
||||
if not mangel:
|
||||
return Tolk("denne", Path(sys.executable), [])
|
||||
kandidat = venv_python(data)
|
||||
if kandidat.exists() and os.environ.get(VENV_MERKE) != "1":
|
||||
return Tolk("venv", kandidat, mangel)
|
||||
return Tolk("mangler", None, mangel)
|
||||
|
||||
|
||||
def kjor_paa_nytt(python: Path, argv: list[str]) -> int:
|
||||
"""Kjør samme kommando med en annen Python og gi videre exit-koden."""
|
||||
sys.stdout.flush()
|
||||
miljo = dict(os.environ)
|
||||
miljo[VENV_MERKE] = "1"
|
||||
return subprocess.run([str(python), str(Path(__file__).resolve()), *argv], env=miljo, check=False).returncode
|
||||
|
||||
|
||||
def kommando_tekst(valg: Valg, *args: str) -> str:
|
||||
"""Kommandolinjen brukeren kan skrive for å kjøre kjor.py med samme --data."""
|
||||
deler = ["python3", str(Path(__file__).resolve())]
|
||||
if valg.data_oppgitt:
|
||||
deler += ["--data", str(valg.data)]
|
||||
return " ".join(shlex.quote(d) for d in [*deler, *args])
|
||||
|
||||
|
||||
def melding_mangler(valg: Valg, tolk: Tolk) -> str:
|
||||
"""Norsk melding for exit-kode 3."""
|
||||
linjer = [
|
||||
f"«{valg.kommando}» trenger numpy (2.0 eller nyere) og scipy. Mangler: {', '.join(tolk.mangler)}.",
|
||||
f"Python som ble brukt: {sys.executable} ({platform.python_version()}).",
|
||||
]
|
||||
if os.environ.get(VENV_MERKE) == "1":
|
||||
linjer.append(f"Den egne mappa {venv_mappe(valg.data)} mangler dem også. Kommandoen under installerer "
|
||||
"dem der på nytt fra PyPI (krever internett):")
|
||||
else:
|
||||
linjer.append(f"Fant ingen egen mappe med numpy og scipy i {venv_mappe(valg.data)}. Kommandoen under "
|
||||
"lager den og installerer numpy og scipy fra PyPI der (krever internett):")
|
||||
linjer.append(f" {kommando_tekst(valg, 'miljo', '--opprett')}")
|
||||
linjer.append("Ingenting installeres uten at du ber om det. diagnose virker uten numpy og scipy.")
|
||||
return "\n".join(linjer)
|
||||
|
||||
|
||||
# ---------------- å kjøre kjernen og modulene ----------------
|
||||
|
||||
def forbered_stier(modell: Path) -> None:
|
||||
"""Legg modell-mappa og kjernen først i sys.path, og glem moduler lastet fra en annen mappe."""
|
||||
sys.dont_write_bytecode = True # ingen __pycache__ i plugin-mappa
|
||||
for navn in (*KJERNEMODULER, *MODELLMODULER):
|
||||
sys.modules.pop(navn, None)
|
||||
fjern = {str(modell), str(modell / "kjerne")}
|
||||
if modell != HER:
|
||||
fjern.add(str(HER))
|
||||
sys.path[:] = [sti for sti in sys.path if str(Path(sti or ".").resolve()) not in fjern]
|
||||
sys.path[:0] = [str(modell / "kjerne"), str(modell)]
|
||||
|
||||
|
||||
def exit_kode(kode: object) -> int:
|
||||
"""Exit-koden for et sys.exit(kode) fra kjernen eller en modul."""
|
||||
if kode is None:
|
||||
return OK
|
||||
if isinstance(kode, int):
|
||||
return kode
|
||||
sys.stdout.flush()
|
||||
print(kode, file=sys.stderr)
|
||||
return FEIL_BRUK
|
||||
|
||||
|
||||
def fang(funksjon: Callable[[], object], hva: str) -> int:
|
||||
"""Kjør funksjonen og gjør sys.exit og uventede feil om til exit-koder."""
|
||||
try:
|
||||
resultat = funksjon()
|
||||
except SystemExit as stopp:
|
||||
return exit_kode(stopp.code)
|
||||
except Exception: # noqa: BLE001 - alt som ikke er SystemExit, er en feil i koden, ikke i inndata
|
||||
sys.stdout.flush()
|
||||
traceback.print_exc()
|
||||
print(f"Uventet feil i {hva} (se over).", file=sys.stderr)
|
||||
return FEIL_BRUK
|
||||
if isinstance(resultat, int) and not isinstance(resultat, bool):
|
||||
return resultat
|
||||
return OK
|
||||
|
||||
|
||||
def last_kjerne(modell: Path, navn: str) -> ModuleType:
|
||||
"""Last en kjernemodul fra modell-mappa/kjerne."""
|
||||
forbered_stier(modell)
|
||||
return importlib.import_module(navn)
|
||||
|
||||
|
||||
def last_modul(modell: Path, navn: str) -> ModuleType | None:
|
||||
"""Last modell-mappa/<navn>.py. None hvis fila ikke finnes."""
|
||||
fil = modell / f"{navn}.py"
|
||||
if not fil.is_file():
|
||||
return None
|
||||
forbered_stier(modell)
|
||||
spec = importlib.util.spec_from_file_location(navn, fil)
|
||||
if spec is None or spec.loader is None:
|
||||
return None
|
||||
modul = importlib.util.module_from_spec(spec)
|
||||
sys.modules[navn] = modul
|
||||
spec.loader.exec_module(modul)
|
||||
return modul
|
||||
|
||||
|
||||
def kjor_modul(valg: Valg, navn: str) -> int:
|
||||
"""Kall main(argv) i modell-mappa/<navn>.py med argumentene etter kommandoen."""
|
||||
fil = valg.modell / f"{navn}.py"
|
||||
if not fil.is_file():
|
||||
print(f"Kommandoen «{valg.kommando}» er ikke på plass i denne modellen ennå: fant ikke {fil}.",
|
||||
file=sys.stderr)
|
||||
return FEIL_BRUK
|
||||
|
||||
def start() -> object:
|
||||
main = getattr(last_modul(valg.modell, navn), "main", None)
|
||||
if not callable(main):
|
||||
sys.exit(f"{fil} mangler funksjonen main(argv).")
|
||||
return main(list(valg.args))
|
||||
|
||||
return fang(start, f"{navn}.py")
|
||||
|
||||
|
||||
# ---------------- kommandoene ----------------
|
||||
|
||||
def kommando_diagnose(valg: Valg) -> int:
|
||||
"""Kjernens diagnose.main."""
|
||||
if len(valg.args) != 1:
|
||||
melding = "Bruk: kjor.py diagnose <inndata.json>"
|
||||
raise Bruksfeil(melding)
|
||||
diagnose = last_kjerne(valg.modell, "diagnose")
|
||||
return fang(lambda: diagnose.main(valg.args[0]), "diagnose.py")
|
||||
|
||||
|
||||
def kommando_forlop(valg: Valg) -> int:
|
||||
"""Kjernens forlop.main."""
|
||||
if len(valg.args) not in {1, 2}:
|
||||
melding = 'Bruk: kjor.py forlop <inndata.json> ["navn på ledd"]'
|
||||
raise Bruksfeil(melding)
|
||||
forlop = last_kjerne(valg.modell, "forlop")
|
||||
return fang(lambda: forlop.main(*valg.args), "forlop.py")
|
||||
|
||||
|
||||
def kommando_notatet(valg: Valg) -> int:
|
||||
"""Kjør kjernens sim.py uendret, med ut-mappa som arbeidsmappe (sim.py skriver data/ der)."""
|
||||
ut = Path.cwd() / "resultater" / "notatet"
|
||||
args = list(valg.args)
|
||||
if args:
|
||||
if len(args) != 2 or args[0] != "--ut": # noqa: PLR2004 - «--ut <mappe>»
|
||||
melding = "Bruk: kjor.py notatet [--ut <mappe>]"
|
||||
raise Bruksfeil(melding)
|
||||
ut = Path(args[1]).expanduser()
|
||||
ut = ut.resolve()
|
||||
if ut == valg.modell or valg.modell in ut.parents:
|
||||
melding = f"Ut-mappa {ut} ligger inne i modell-mappa {valg.modell}. Velg en annen med --ut <mappe>."
|
||||
raise Bruksfeil(melding)
|
||||
if ut.exists() and not ut.is_dir():
|
||||
melding = f"{ut} finnes og er ikke en mappe. Velg en annen med --ut <mappe>."
|
||||
raise Bruksfeil(melding)
|
||||
ut.mkdir(parents=True, exist_ok=True)
|
||||
print(f"Kjører notatets simuleringer (kjerne/sim.py, uendret) i {ut}. Det tar om lag 1-2 minutter.\n",
|
||||
flush=True)
|
||||
miljo = dict(os.environ, PYTHONIOENCODING="utf-8", PYTHONDONTWRITEBYTECODE="1", PYTHONUNBUFFERED="1")
|
||||
utskrift = ut / "sim_utskrift.txt"
|
||||
with utskrift.open("w", encoding="utf-8") as logg, subprocess.Popen(
|
||||
[sys.executable, str(valg.modell / "kjerne" / "sim.py")],
|
||||
cwd=str(ut), stdout=subprocess.PIPE, text=True, encoding="utf-8", env=miljo,
|
||||
) as prosess:
|
||||
if prosess.stdout is not None:
|
||||
for linje in prosess.stdout:
|
||||
sys.stdout.write(linje)
|
||||
logg.write(linje)
|
||||
if prosess.returncode != 0:
|
||||
print(f"sim.py stoppet med exit-kode {prosess.returncode}.", file=sys.stderr)
|
||||
return FEIL_BRUK
|
||||
antall = len(list((ut / "data").glob("*.dat")))
|
||||
print(f"\nFerdig: {antall} datafiler i {ut / 'data'}, utskriften i {utskrift}.")
|
||||
print("«kontroll» sammenligner pluginens egen kjøring med notatet.")
|
||||
return OK
|
||||
|
||||
|
||||
def laaste_versjoner(modell: Path) -> dict[str, str]:
|
||||
"""Notatets låste versjoner fra modell/requirements.txt."""
|
||||
ut = {}
|
||||
fil = modell / "requirements.txt"
|
||||
if fil.is_file():
|
||||
for linje in fil.read_text(encoding="utf-8").splitlines():
|
||||
navn, skille, versjon = linje.split("#")[0].strip().partition("==")
|
||||
if skille:
|
||||
ut[navn.strip().lower()] = versjon.strip()
|
||||
return ut
|
||||
|
||||
|
||||
def vis_tolk(beskrivelse: Beskrivelse, laast: dict[str, str]) -> list[str]:
|
||||
"""Linjer som beskriver Python, numpy og scipy i en tolk."""
|
||||
linjer = [f" Python {beskrivelse['python']} ({beskrivelse['sti']})"]
|
||||
for navn in ("numpy", "scipy"):
|
||||
versjon = beskrivelse["numpy"] if navn == "numpy" else beskrivelse["scipy"]
|
||||
notatet = f" (notatet: {laast[navn]})" if navn in laast else ""
|
||||
linjer.append(f" {navn}: {versjon or 'mangler'}{notatet}")
|
||||
return linjer
|
||||
|
||||
|
||||
def samme_som_notatet(beskrivelse: Beskrivelse, laast: dict[str, str]) -> bool:
|
||||
"""Sjekk om tolken har notatets låste versjoner av numpy og scipy."""
|
||||
return beskrivelse["numpy"] == laast.get("numpy") and beskrivelse["scipy"] == laast.get("scipy")
|
||||
|
||||
|
||||
def rapport_miljo(valg: Valg) -> None:
|
||||
"""Skriv ut hva denne Python og den egne mappa har."""
|
||||
laast = laaste_versjoner(valg.modell)
|
||||
denne = beskriv_denne()
|
||||
print("Python som kjører nå:")
|
||||
print("\n".join(vis_tolk(denne, laast)))
|
||||
vpy = venv_python(valg.data)
|
||||
print(f"\nEgen mappe for numpy og scipy: {venv_mappe(valg.data)}")
|
||||
venv: Beskrivelse | None = None
|
||||
if not vpy.exists():
|
||||
print(" finnes ikke. «miljo --opprett» lager den og installerer numpy og scipy fra PyPI "
|
||||
"(krever internett).")
|
||||
else:
|
||||
venv = beskriv_tolk(vpy)
|
||||
if venv is None:
|
||||
print(f" finnes, men {vpy} lot seg ikke starte. «miljo --opprett» prøver å lage den på nytt.")
|
||||
else:
|
||||
print("\n".join(vis_tolk(venv, laast)))
|
||||
print()
|
||||
regner = None
|
||||
if not mangler(denne):
|
||||
regner = denne
|
||||
print("forlop, scenario, notatet og kontroll regner med Python som kjører nå.")
|
||||
elif venv is not None and not mangler(venv):
|
||||
regner = venv
|
||||
print("forlop, scenario, notatet og kontroll regner med Python i den egne mappa.")
|
||||
else:
|
||||
hva = ", ".join(mangler(venv if venv is not None else denne))
|
||||
print(f"forlop, scenario, notatet og kontroll kan ikke regne nå: {hva} mangler.")
|
||||
if regner is not None:
|
||||
if samme_som_notatet(regner, laast):
|
||||
print("Versjonene av numpy og scipy er de samme som notatets, så kontroll kan gi bit for bit.")
|
||||
else:
|
||||
print("Versjonene av numpy og scipy er ikke de samme som notatets: kontroll sammenligner da "
|
||||
"med toleranse, ikke bit for bit.")
|
||||
|
||||
|
||||
def kjor_prosess(kommando: list[str]) -> int:
|
||||
"""Kjør et program (pip, venv) med utskriften rett til skjermen. Exit-koden tilbake."""
|
||||
sys.stdout.flush()
|
||||
try:
|
||||
return subprocess.run(kommando, check=False).returncode
|
||||
except OSError as feil:
|
||||
print(f"Kunne ikke starte {kommando[0]}: {feil}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
def opprett_miljo(valg: Valg) -> int:
|
||||
"""Lag den egne mappa og installer numpy og scipy: låste versjoner først, ellers nyeste."""
|
||||
mappe = venv_mappe(valg.data)
|
||||
vpy = venv_python(valg.data)
|
||||
if not vpy.exists():
|
||||
print(f"Lager egen Python-mappa {mappe} med {sys.executable} ({platform.python_version()}).")
|
||||
valg.data.mkdir(parents=True, exist_ok=True)
|
||||
if kjor_prosess([sys.executable, "-m", "venv", str(mappe)]) != 0 or not vpy.exists():
|
||||
print(f"Kunne ikke lage {mappe}. På noen Linux-systemer må pakken python3-venv installeres først.",
|
||||
file=sys.stderr)
|
||||
return MILJO_MANGLER
|
||||
pip = [str(vpy), "-m", "pip", "install", "--disable-pip-version-check"]
|
||||
krav = valg.modell / "requirements.txt"
|
||||
laast = laaste_versjoner(valg.modell)
|
||||
print(f"\nInstallerer notatets versjoner ({', '.join(f'{n} {v}' for n, v in laast.items())}) fra PyPI.")
|
||||
toleranse = False
|
||||
if kjor_prosess([*pip, "-r", str(krav)]) != 0:
|
||||
print("\nNotatets låste versjoner lot seg ikke installere for denne Python-versjonen. "
|
||||
"Prøver numpy 2.0 eller nyere og scipy.")
|
||||
toleranse = True
|
||||
if kjor_prosess([*pip, "numpy>=2.0", "scipy"]) != 0:
|
||||
print(f"\nInstallasjonen feilet. Se meldingene over. Mappa {mappe} kan slettes.", file=sys.stderr)
|
||||
return MILJO_MANGLER
|
||||
print()
|
||||
rapport_miljo(valg)
|
||||
if toleranse:
|
||||
print("\nNumpy og scipy er ikke notatets låste versjoner. Resultatene kontrolleres da med toleranse, "
|
||||
"ikke bit for bit.")
|
||||
return OK
|
||||
|
||||
|
||||
def kommando_miljo(valg: Valg) -> int:
|
||||
"""Vis miljøet, eller lag den egne mappa med --opprett."""
|
||||
if valg.args == ["--opprett"]:
|
||||
return opprett_miljo(valg)
|
||||
if valg.args:
|
||||
melding = "Bruk: kjor.py miljo [--opprett]"
|
||||
raise Bruksfeil(melding)
|
||||
rapport_miljo(valg)
|
||||
return OK
|
||||
|
||||
|
||||
def kommando_beskriv(_valg: Valg) -> int:
|
||||
"""Intern kommando for beskriv_tolk(): skriv beskrivelsen av denne tolken som JSON."""
|
||||
print(json.dumps(beskriv_denne()))
|
||||
return OK
|
||||
|
||||
|
||||
def kommando_hjelp(_valg: Valg) -> int:
|
||||
"""Skriv bruken."""
|
||||
print(HJELP, end="")
|
||||
return OK
|
||||
|
||||
|
||||
def kommando_regner(valg: Valg) -> int:
|
||||
"""Kommandoene som trenger numpy og scipy: velg Python først (se toppen av fila)."""
|
||||
if valg.kommando == "scenario" and not (valg.modell / "scenario.py").is_file():
|
||||
return kjor_modul(valg, "scenario") # sier at kommandoen ikke er på plass ennå
|
||||
if valg.kommando == "forlop" and valg.args and not Path(valg.args[0]).exists():
|
||||
print(f"Fant ikke fila {valg.args[0]}.", file=sys.stderr)
|
||||
return FEIL_BRUK
|
||||
tolk = velg_tolk(valg.data)
|
||||
if tolk.hvor == "venv" and tolk.python is not None:
|
||||
return kjor_paa_nytt(tolk.python, valg.argv)
|
||||
if tolk.hvor == "mangler" and valg.kommando != "kontroll": # kontroll gjør det den kan uten
|
||||
print(melding_mangler(valg, tolk), file=sys.stderr)
|
||||
return MILJO_MANGLER
|
||||
handlinger: dict[str, Callable[[Valg], int]] = {
|
||||
"forlop": kommando_forlop, "notatet": kommando_notatet,
|
||||
"scenario": lambda v: kjor_modul(v, "scenario"), "kontroll": lambda v: kjor_modul(v, "kontroll"),
|
||||
}
|
||||
return handlinger[valg.kommando](valg)
|
||||
|
||||
|
||||
def utfor(valg: Valg) -> int:
|
||||
"""Utfør kommandoen."""
|
||||
handlinger: dict[str, Callable[[Valg], int]] = {
|
||||
"hjelp": kommando_hjelp, "_beskriv": kommando_beskriv,
|
||||
"diagnose": kommando_diagnose, "miljo": kommando_miljo,
|
||||
"arbeidskopi": lambda v: kjor_modul(v, "arbeidskopi"),
|
||||
}
|
||||
return handlinger.get(valg.kommando, kommando_regner)(valg)
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
"""Inngangen. Returnerer exit-koden."""
|
||||
try:
|
||||
if not argv:
|
||||
print(HJELP, end="", file=sys.stderr)
|
||||
return FEIL_BRUK
|
||||
return utfor(tolk_argumenter(argv))
|
||||
except Bruksfeil as feil:
|
||||
sys.stdout.flush()
|
||||
print(f"{feil}\nSkriv «python3 kjor.py --hjelp» for bruk.", file=sys.stderr)
|
||||
return FEIL_BRUK
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(sys.argv[1:]))
|
||||
432
modell/kontroll.py
Normal file
432
modell/kontroll.py
Normal file
|
|
@ -0,0 +1,432 @@
|
|||
# Copyright (c) 2026 Kjell Tore Guttormsen. MIT License, see LICENSE.
|
||||
"""Gjenskapingskontroll: sjekk at modellen i denne mappa regner som notatet.
|
||||
|
||||
Bruk:
|
||||
python3 kjor.py kontroll [--rask]
|
||||
python3 kontroll.py [--rask] (med en Python som har numpy og scipy)
|
||||
|
||||
Fire kontroller, hver med én linje dom og detaljer bare ved avvik:
|
||||
1. Kjernen: sha256 for filene i kjerne/ mot kjerne/KILDE.json.
|
||||
2. diagnose.py på begge eksempelfilene mot fasit/ (tekst).
|
||||
3. forlop.py på begge eksempelfilene mot fasit/ (tekst; trenger numpy og scipy).
|
||||
4. Full kjøring av sim.py i en midlertidig mappe: datafilene mot fasit/data/ byte for byte,
|
||||
med største absolutte avvik per kolonne når en fil er ulik, og utskriften mot
|
||||
fasit/sim_utskrift.txt. Om lag 1-2 min; --rask hopper over den.
|
||||
I tillegg kjøres grensetilfelle() i hver utvidelse i utvidelser/; hvert avvik skal være eksakt 0.
|
||||
|
||||
Nivået til slutt:
|
||||
- «bit for bit som notatet»: alt er likt byte for byte.
|
||||
- «innenfor toleranse»: teksten er lik, og hvert tall avviker med høyst én enhet i siste
|
||||
viste siffer (det som kan skje med andre versjoner av numpy og scipy, eller en annen maskin).
|
||||
- «endret»: alt annet. I en arbeidskopi (mappa over modell/ har LES-MEG.md, som
|
||||
arbeidskopi.py lager) er det en opplysning, ikke en feil: detaljene sier hvilke av
|
||||
notatets resultater som har flyttet seg, og hvor mye.
|
||||
|
||||
Exit-kode: 0 for bit for bit og innenfor toleranse, 1 for endret, 3 når numpy eller scipy
|
||||
mangler for noe av kontrollen (det som kunne kjøres, er kjørt), 2 for feil bruk.
|
||||
|
||||
Grensesnitt: main(argv: list[str]) -> int, som kjor.py kaller.
|
||||
Utvidelser: utvidelser/<navn>.py (ikke filer som begynner med _) med
|
||||
grensetilfelle() -> list av (beskrivelse, største absolutte avvik mot kjernen). Hver
|
||||
kjøres i en egen prosess med modell/kjerne og modell/ først i sys.path.
|
||||
Bare standardbiblioteket her; kjernen kjøres som egne prosesser med samme Python.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import difflib
|
||||
import hashlib
|
||||
import importlib
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
MODELL = Path(__file__).resolve().parent
|
||||
KJERNE = MODELL / "kjerne"
|
||||
ROT = MODELL.parent
|
||||
FASIT = ROT / "fasit"
|
||||
EKSEMPLER = ROT / "eksempler"
|
||||
EKSEMPELFILER = ("eksempel_domstol", "norge_2025")
|
||||
ARBEIDSKOPI_MERKE = "LES-MEG.md"
|
||||
|
||||
LIK, TOLERANSE, ENDRET, IKKE_KJORT = "lik", "toleranse", "endret", "ikke kjørt"
|
||||
RANG = {LIK: 0, TOLERANSE: 1, IKKE_KJORT: 2, ENDRET: 3}
|
||||
MAKS_LINJER = 40 # linjer med forskjeller som vises per sammenligning
|
||||
SLINGRING = 1e-9 # relativ slingring når avviket sammenlignes med én enhet i siste siffer
|
||||
|
||||
TALL = re.compile(r"[-+]?\d+(?:[.,]\d+)?(?:[eE][-+]?\d+)?")
|
||||
|
||||
MILJO = dict(os.environ, PYTHONIOENCODING="utf-8", PYTHONDONTWRITEBYTECODE="1")
|
||||
|
||||
UTVIDELSE_KJORER = """\
|
||||
import importlib.util, json, sys
|
||||
modell, fil = sys.argv[1], sys.argv[2]
|
||||
sys.path[:0] = [modell + "/kjerne", modell]
|
||||
spec = importlib.util.spec_from_file_location("utvidelse", fil)
|
||||
modul = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(modul)
|
||||
print()
|
||||
print(json.dumps([[str(b), float(a)] for b, a in modul.grensetilfelle()]))
|
||||
"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class Dom:
|
||||
"""Resultatet av én kontroll."""
|
||||
|
||||
status: str
|
||||
linje: str
|
||||
detaljer: list[str] = field(default_factory=list)
|
||||
avvik: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class Sammenligning:
|
||||
"""Resultatet av å sammenligne en fil eller en utskrift med fasit."""
|
||||
|
||||
status: str
|
||||
avvik: float = 0.0
|
||||
detaljer: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
# ---------------- sammenligning ----------------
|
||||
|
||||
def enhet(tall: str) -> float:
|
||||
"""Én enhet i siste viste siffer: «1,28» -> 0,01, «7.21» -> 0,01, «1.11e-15» -> 1e-17."""
|
||||
mantisse, _, eksponent = tall.lower().partition("e")
|
||||
desimaler = len(mantisse.replace(",", ".").partition(".")[2])
|
||||
return float(10.0 ** (int(eksponent or "0") - desimaler))
|
||||
|
||||
|
||||
def verdi(tall: str) -> float:
|
||||
"""Tallet i en tekst, med punktum eller komma som desimaltegn."""
|
||||
return float(tall.replace(",", "."))
|
||||
|
||||
|
||||
def innenfor(fasit: str, naa: str) -> tuple[bool, float]:
|
||||
"""Er to tall like innenfor én enhet i siste viste siffer? Returnerer også avviket."""
|
||||
try:
|
||||
a, b = verdi(fasit), verdi(naa)
|
||||
except ValueError:
|
||||
return fasit == naa, 0.0 if fasit == naa else math.inf
|
||||
avvik = abs(a - b) if math.isfinite(a) and math.isfinite(b) else (0.0 if a == b else math.inf)
|
||||
return avvik <= max(enhet(fasit), enhet(naa)) * (1 + SLINGRING), avvik
|
||||
|
||||
|
||||
def vis_tall(x: float) -> str:
|
||||
"""Et avvik med norsk desimaltegn."""
|
||||
if x == 0:
|
||||
return "0"
|
||||
if math.isinf(x):
|
||||
return "uendelig"
|
||||
return f"{x:.1e}".replace(".", ",")
|
||||
|
||||
|
||||
def forskjeller(fasit: list[str], naa: list[str]) -> list[str]:
|
||||
"""Linjene som er ulike, som en kort diff (notatet / nå)."""
|
||||
linjer = list(difflib.unified_diff(fasit, naa, "notatet", "nå", lineterm="", n=0))
|
||||
if len(linjer) > MAKS_LINJER:
|
||||
linjer = [*linjer[:MAKS_LINJER], f"... ({len(linjer) - MAKS_LINJER} linjer til)"]
|
||||
return linjer
|
||||
|
||||
|
||||
def sammenlign_tekst(fasit: str, naa: str) -> Sammenligning:
|
||||
"""Sammenlign en utskrift med fasit: lik, innenfor toleranse eller endret."""
|
||||
if fasit == naa:
|
||||
return Sammenligning(LIK)
|
||||
a, b = fasit.splitlines(), naa.splitlines()
|
||||
detaljer = forskjeller(a, b)
|
||||
if len(a) != len(b):
|
||||
return Sammenligning(ENDRET, math.inf, detaljer)
|
||||
storst = 0.0
|
||||
for linje_a, linje_b in zip(a, b):
|
||||
if linje_a == linje_b:
|
||||
continue
|
||||
if TALL.sub("#", linje_a) != TALL.sub("#", linje_b):
|
||||
return Sammenligning(ENDRET, math.inf, detaljer)
|
||||
for x, y in zip(TALL.findall(linje_a), TALL.findall(linje_b)):
|
||||
ok, avvik = innenfor(x, y)
|
||||
if not ok:
|
||||
return Sammenligning(ENDRET, avvik, detaljer)
|
||||
storst = max(storst, avvik)
|
||||
return Sammenligning(TOLERANSE, storst, detaljer)
|
||||
|
||||
|
||||
def les_dat(tekst: str) -> tuple[list[str], list[list[str]]]:
|
||||
"""Kolonnenavn og rader i en datafil fra sim.py (første linje er kolonnenavnene)."""
|
||||
linjer = [linje.split() for linje in tekst.splitlines() if linje.strip()]
|
||||
if not linjer:
|
||||
return [], []
|
||||
return linjer[0], linjer[1:]
|
||||
|
||||
|
||||
def sammenlign_dat(fasit: str, naa: str) -> Sammenligning:
|
||||
"""Sammenlign en datafil med fasit; ved avvik største absolutte avvik per kolonne."""
|
||||
if fasit == naa:
|
||||
return Sammenligning(LIK)
|
||||
kol_a, rader_a = les_dat(fasit)
|
||||
kol_b, rader_b = les_dat(naa)
|
||||
if kol_a != kol_b:
|
||||
return Sammenligning(ENDRET, math.inf,
|
||||
[f"kolonnene er ulike: notatet {' '.join(kol_a)}, nå {' '.join(kol_b)}"])
|
||||
if len(rader_a) != len(rader_b) or any(len(x) != len(y) for x, y in zip(rader_a, rader_b)):
|
||||
return Sammenligning(ENDRET, math.inf,
|
||||
[f"antall rader er ulikt: notatet {len(rader_a)}, nå {len(rader_b)}"])
|
||||
per_kolonne = [0.0] * len(kol_a)
|
||||
alle_innenfor = True
|
||||
for rad_a, rad_b in zip(rader_a, rader_b):
|
||||
for j, (x, y) in enumerate(zip(rad_a, rad_b)):
|
||||
if x == y:
|
||||
continue
|
||||
ok, avvik = innenfor(x, y)
|
||||
alle_innenfor = alle_innenfor and ok
|
||||
per_kolonne[j] = max(per_kolonne[j], avvik)
|
||||
deler = [f"{navn} {vis_tall(avvik)}" for navn, avvik in zip(kol_a, per_kolonne) if avvik > 0]
|
||||
detaljer = ["største absolutte avvik per kolonne: " + (", ".join(deler) or "ingen (bare formatering)")]
|
||||
return Sammenligning(TOLERANSE if alle_innenfor else ENDRET, max(per_kolonne), detaljer)
|
||||
|
||||
|
||||
def verst(statuser: list[str]) -> str:
|
||||
"""Den dårligste statusen i en liste."""
|
||||
return max(statuser, key=RANG.__getitem__, default=LIK)
|
||||
|
||||
|
||||
# ---------------- miljø og prosesser ----------------
|
||||
|
||||
def har_numpy_scipy() -> bool:
|
||||
"""Sjekk om denne Python har numpy 2.0 eller nyere (numpy.trapezoid) og scipy."""
|
||||
try:
|
||||
numpy = importlib.import_module("numpy")
|
||||
importlib.import_module("scipy.optimize")
|
||||
except ImportError:
|
||||
return False
|
||||
return hasattr(numpy, "trapezoid")
|
||||
|
||||
|
||||
def kjor_python(args: list[str], cwd: Path) -> subprocess.CompletedProcess[str]:
|
||||
"""Kjør denne Python med args i mappa cwd og fang utskriften."""
|
||||
return subprocess.run([sys.executable, *args], cwd=str(cwd), env=MILJO, capture_output=True, text=True,
|
||||
encoding="utf-8", check=False)
|
||||
|
||||
|
||||
def feilmelding(prosess: subprocess.CompletedProcess[str]) -> list[str]:
|
||||
"""De siste linjene på stderr, til detaljene."""
|
||||
linjer = prosess.stderr.strip().splitlines()
|
||||
return [f"stoppet med exit-kode {prosess.returncode}:", *linjer[-MAKS_LINJER // 2:]]
|
||||
|
||||
|
||||
def sha256(fil: Path) -> str:
|
||||
"""sha256 av en fil."""
|
||||
return hashlib.sha256(fil.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
# ---------------- kontrollene ----------------
|
||||
|
||||
def kontroll_kjerne() -> Dom:
|
||||
"""1. Kjernen mot sha256 i KILDE.json."""
|
||||
kildefil = KJERNE / "KILDE.json"
|
||||
if not kildefil.is_file():
|
||||
return Dom(ENDRET, f"Kjernen: fant ikke {kildefil}, så kjernen kan ikke sammenlignes med notatet.")
|
||||
kilde = json.loads(kildefil.read_text(encoding="utf-8"))
|
||||
endret = []
|
||||
for navn, info in kilde["filer"].items():
|
||||
fil = KJERNE / navn
|
||||
if not fil.is_file():
|
||||
endret.append(f"{navn} mangler")
|
||||
elif sha256(fil) != info["sha256"]:
|
||||
endret.append(f"{navn} er endret")
|
||||
versjon = f"{kilde['repo']} @ {kilde['commit'][:7]}"
|
||||
if not endret:
|
||||
return Dom(LIK, f"Kjernen: uendret fra notatet ({len(kilde['filer'])} filer, {versjon}).")
|
||||
return Dom(ENDRET, f"Kjernen: endret fra notatet ({versjon}): {', '.join(endret)}.")
|
||||
|
||||
|
||||
def kontroll_utskrift(skript: str, mappe: Path) -> Dom:
|
||||
"""2 og 3. Utskriften fra diagnose.py eller forlop.py på begge eksempelfilene mot fasit."""
|
||||
statuser, detaljer, storst = [], [], 0.0
|
||||
for eksempel in EKSEMPELFILER:
|
||||
fasitfil = FASIT / f"{skript}_{eksempel}.txt"
|
||||
inndata = EKSEMPLER / f"{eksempel}.json"
|
||||
mangler = [str(f) for f in (fasitfil, inndata) if not f.is_file()]
|
||||
if mangler:
|
||||
statuser.append(ENDRET)
|
||||
detaljer.append(f"{eksempel}: fant ikke {', '.join(mangler)}")
|
||||
continue
|
||||
prosess = kjor_python([str(KJERNE / f"{skript}.py"), str(inndata)], mappe)
|
||||
if prosess.returncode != 0:
|
||||
statuser.append(ENDRET)
|
||||
detaljer += [f"{eksempel}:", *(" " + linje for linje in feilmelding(prosess))]
|
||||
continue
|
||||
resultat = sammenlign_tekst(fasitfil.read_text(encoding="utf-8"), prosess.stdout)
|
||||
statuser.append(resultat.status)
|
||||
storst = max(storst, resultat.avvik)
|
||||
if resultat.status != LIK:
|
||||
detaljer += [f"{eksempel}:", *(" " + linje for linje in resultat.detaljer)]
|
||||
status = verst(statuser)
|
||||
begge = " og ".join(EKSEMPELFILER)
|
||||
linje = {
|
||||
LIK: f"{skript}: lik notatet for {begge}.",
|
||||
TOLERANSE: f"{skript}: innenfor toleranse for {begge} (største avvik {vis_tall(storst)}).",
|
||||
ENDRET: f"{skript}: endret (se under).",
|
||||
}[status]
|
||||
return Dom(status, linje, detaljer, storst)
|
||||
|
||||
|
||||
def kontroll_full(mappe: Path) -> Dom:
|
||||
"""4. Full kjøring av sim.py i en midlertidig mappe mot fasit/data og fasit/sim_utskrift.txt."""
|
||||
print(" (kjører sim.py, om lag 1-2 minutter ...)", flush=True)
|
||||
prosess = kjor_python([str(KJERNE / "sim.py")], mappe)
|
||||
if prosess.returncode != 0:
|
||||
return Dom(ENDRET, "Full kjøring: sim.py stoppet med en feil.", feilmelding(prosess), math.inf)
|
||||
fasit_data = FASIT / "data"
|
||||
navn_fasit = sorted(f.name for f in fasit_data.glob("*.dat"))
|
||||
navn_naa = sorted(f.name for f in (mappe / "data").glob("*.dat"))
|
||||
statuser, detaljer, storst, like = [], [], 0.0, 0
|
||||
for navn in sorted(set(navn_fasit) | set(navn_naa)):
|
||||
if navn not in navn_naa or navn not in navn_fasit:
|
||||
statuser.append(ENDRET)
|
||||
detaljer.append(f"{navn}: " + ("mangler i kjøringen" if navn not in navn_naa else "finnes ikke i fasit"))
|
||||
continue
|
||||
resultat = sammenlign_dat((fasit_data / navn).read_text(encoding="utf-8"),
|
||||
(mappe / "data" / navn).read_text(encoding="utf-8"))
|
||||
statuser.append(resultat.status)
|
||||
storst = max(storst, resultat.avvik)
|
||||
if resultat.status == LIK:
|
||||
like += 1
|
||||
else:
|
||||
dom = "innenfor toleranse" if resultat.status == TOLERANSE else "endret"
|
||||
detaljer.append(f"{navn} ({dom}): {'; '.join(resultat.detaljer)}")
|
||||
utskriftfil = FASIT / "sim_utskrift.txt"
|
||||
if utskriftfil.is_file():
|
||||
utskrift = sammenlign_tekst(utskriftfil.read_text(encoding="utf-8"), prosess.stdout)
|
||||
else:
|
||||
utskrift = Sammenligning(ENDRET, math.inf, [f"fant ikke {utskriftfil}"])
|
||||
statuser.append(utskrift.status)
|
||||
storst = max(storst, utskrift.avvik)
|
||||
if utskrift.status != LIK:
|
||||
detaljer += ["utskriften fra sim.py:", *(" " + linje for linje in utskrift.detaljer)]
|
||||
status = verst(statuser)
|
||||
tekst_utskrift = {LIK: "lik", TOLERANSE: "innenfor toleranse", ENDRET: "endret"}[utskrift.status]
|
||||
linje = (f"Full kjøring: {like} av {len(navn_fasit)} datafiler like byte for byte med notatet; "
|
||||
f"utskriften {tekst_utskrift}.")
|
||||
return Dom(status, linje, detaljer, storst)
|
||||
|
||||
|
||||
def kontroll_utvidelser(mappe: Path, *, regnemiljo: bool) -> Dom:
|
||||
"""Grensetilfellet i hver utvidelse skal gjenskape kjernen eksakt (avvik 0)."""
|
||||
utvidelser = sorted(f for f in (MODELL / "utvidelser").glob("*.py") if not f.name.startswith("_"))
|
||||
if not utvidelser:
|
||||
return Dom(LIK, "Utvidelser: ingen å kontrollere.")
|
||||
if not regnemiljo:
|
||||
return Dom(IKKE_KJORT, f"Utvidelser: ikke kjørt ({len(utvidelser)} stk.), numpy og scipy mangler.")
|
||||
statuser, detaljer, storst = [], [], 0.0
|
||||
for fil in utvidelser:
|
||||
prosess = kjor_python(["-c", UTVIDELSE_KJORER, str(MODELL), str(fil)], mappe)
|
||||
try:
|
||||
tilfeller = json.loads(prosess.stdout.strip().splitlines()[-1]) if prosess.returncode == 0 else None
|
||||
except (ValueError, IndexError):
|
||||
tilfeller = None
|
||||
if tilfeller is None:
|
||||
statuser.append(ENDRET)
|
||||
detaljer += [f"{fil.stem}: grensetilfelle() kunne ikke kjøres", *(" " + linje for linje in
|
||||
feilmelding(prosess))]
|
||||
continue
|
||||
if not tilfeller:
|
||||
statuser.append(ENDRET)
|
||||
detaljer.append(f"{fil.stem}: grensetilfelle() ga ingen tilfeller; en utvidelse skal ha minst ett.")
|
||||
continue
|
||||
for beskrivelse, avvik in tilfeller:
|
||||
storst = max(storst, abs(avvik))
|
||||
if avvik != 0.0:
|
||||
statuser.append(ENDRET)
|
||||
detaljer.append(f"{fil.stem}: {beskrivelse}: avvik {vis_tall(abs(avvik))}, skal være eksakt 0")
|
||||
else:
|
||||
statuser.append(LIK)
|
||||
status = verst(statuser)
|
||||
navn = ", ".join(f.stem for f in utvidelser)
|
||||
linje = (f"Utvidelser: grensetilfellene gjenskaper kjernen eksakt ({navn})." if status == LIK
|
||||
else f"Utvidelser: grensetilfellet gjenskaper ikke kjernen eksakt ({navn}; se under).")
|
||||
return Dom(status, linje, detaljer, storst)
|
||||
|
||||
|
||||
# ---------------- rapporten ----------------
|
||||
|
||||
def niva(dommer: list[Dom], *, rask: bool, arbeidskopi: bool) -> list[str]:
|
||||
"""Linjene om nivået som er oppnådd."""
|
||||
kjort = [d for d in dommer if d.status != IKKE_KJORT]
|
||||
status = verst([d.status for d in kjort])
|
||||
deler = (["uten full kjøring, --rask"] if rask else []) + (
|
||||
["bare det som er kjørt"] if any(d.status == IKKE_KJORT for d in dommer) else [])
|
||||
hva = f" ({'; '.join(deler)})" if deler else ""
|
||||
if status == LIK:
|
||||
linjer = [f"Nivå{hva}: bit for bit som notatet."]
|
||||
elif status == TOLERANSE:
|
||||
storst = max(d.avvik for d in kjort)
|
||||
linjer = [(f"Nivå{hva}: innenfor toleranse (største avvik {vis_tall(storst)}): hvert tall avviker med "
|
||||
"høyst én enhet i siste viste siffer.")]
|
||||
else:
|
||||
endret = [d.linje.split(":")[0] for d in kjort if d.status == ENDRET]
|
||||
linjer = [f"Nivå{hva}: endret: {', '.join(endret)}."]
|
||||
if arbeidskopi:
|
||||
linjer.append("Dette er en arbeidskopi, så det er en opplysning, ikke en feil: detaljene over viser "
|
||||
"hvilke av notatets resultater som har flyttet seg, og hvor mye.")
|
||||
else:
|
||||
linjer.append("Pluginens modell skal regne som notatet, så dette er en feil.")
|
||||
if any(d.status == IKKE_KJORT for d in dommer):
|
||||
linjer.append(f"Ikke kjørt fordi numpy eller scipy mangler i {sys.executable}. "
|
||||
"«kjor.py miljo --opprett» installerer dem i en egen mappe.")
|
||||
return linjer
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
"""Kjør kontrollene og skriv dommen. Returnerer exit-koden."""
|
||||
if argv not in ([], ["--rask"]):
|
||||
print("Bruk: kjor.py kontroll [--rask]", file=sys.stderr)
|
||||
return 2
|
||||
rask = argv == ["--rask"]
|
||||
arbeidskopi = (ROT / ARBEIDSKOPI_MERKE).is_file()
|
||||
regnemiljo = har_numpy_scipy()
|
||||
print(f"Kontroll av modellen i {MODELL}" + (" (arbeidskopi)" if arbeidskopi else "") + "\n", flush=True)
|
||||
dommer: list[Dom] = []
|
||||
|
||||
def vis(nummer: str, dom: Dom) -> None:
|
||||
dommer.append(dom)
|
||||
print(f"{nummer} {dom.linje}")
|
||||
for linje in dom.detaljer:
|
||||
print(f" {linje}")
|
||||
sys.stdout.flush()
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix="dynamisk-rettsstat-") as tmp:
|
||||
mappe = Path(tmp)
|
||||
vis("1.", kontroll_kjerne())
|
||||
vis("2.", kontroll_utskrift("diagnose", mappe))
|
||||
if regnemiljo:
|
||||
vis("3.", kontroll_utskrift("forlop", mappe))
|
||||
else:
|
||||
vis("3.", Dom(IKKE_KJORT, "forlop: ikke kjørt, numpy og scipy mangler."))
|
||||
if rask:
|
||||
print("4. Full kjøring: hoppet over (--rask).")
|
||||
elif regnemiljo:
|
||||
vis("4.", kontroll_full(mappe))
|
||||
else:
|
||||
vis("4.", Dom(IKKE_KJORT, "Full kjøring: ikke kjørt, numpy og scipy mangler."))
|
||||
vis("5.", kontroll_utvidelser(mappe, regnemiljo=regnemiljo))
|
||||
print()
|
||||
print("\n".join(niva(dommer, rask=rask, arbeidskopi=arbeidskopi)))
|
||||
statuser = [d.status for d in dommer]
|
||||
if ENDRET in statuser:
|
||||
return 1
|
||||
if IKKE_KJORT in statuser:
|
||||
return 3
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.dont_write_bytecode = True
|
||||
sys.exit(main(sys.argv[1:]))
|
||||
2
modell/requirements.txt
Normal file
2
modell/requirements.txt
Normal file
|
|
@ -0,0 +1,2 @@
|
|||
numpy==2.5.4
|
||||
scipy==1.18.1
|
||||
Loading…
Add table
Add a link
Reference in a new issue