cfa4ad9c6b
etm-powersync-compare-hors-motifs.py retire de chaque ligne de la référence le seul motif publié d'une charge (telemetry.loads[].decision, plan.timeSlots[].actions[].reason) et le manque chiffré d'une clôture non tenue (missingWh, closedMissingWh), puis exige l'égalité de tout le reste : consignes, powerW, états, relais, courants, budget, draw, counts, niveaux. Il liste les cas qui changent (fonction, charge, ancien → nouveau motif). Sert de preuve aux correctifs a et b (etm70) avant de régénérer la référence. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
114 lines
4.5 KiB
Python
Executable File
114 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# Comparateur « IDENTIQUE HORS MOTIFS » (correctifs moteur a et b, etm70).
|
|
#
|
|
# etm-powersync-compare-hors-motifs.py <attendu.ndjson[.xz]> <obtenu.ndjson[.xz]> [--cas <sortie.md>]
|
|
#
|
|
# Un lot qui ne change QUE des motifs doit laisser tout le reste de la référence `decisions`
|
|
# intact à l'octet près. Ce script retire de chaque ligne, et SEULEMENT d'elle :
|
|
# - le motif publié d'une charge : `telemetry.loads[].decision` (code, params, level) dans une
|
|
# ligne de cycle, `plan.timeSlots[].actions[].reason` dans une ligne de plan direct ;
|
|
# - le manque chiffré d'une clôture non tenue (correctif b) : `params.missingWh` d'un motif
|
|
# ECO_FLOOR_MISSED porté par un niveau, et `progress.closedMissingWh` du même niveau.
|
|
# Tout le reste — consignes, powerW, estimatedPowerW, états, relais, courants, budget, draw,
|
|
# counts, niveaux et leurs motifs, avancement — doit être IDENTIQUE. Code 0 si c'est le cas,
|
|
# 1 sinon (et le premier écart est montré). La liste des motifs qui changent est imprimée, ou
|
|
# écrite en Markdown avec --cas.
|
|
import json
|
|
import lzma
|
|
import sys
|
|
from collections import Counter
|
|
|
|
|
|
def lire(chemin):
|
|
ouvrir = lzma.open if chemin.endswith('.xz') else open
|
|
with ouvrir(chemin, 'rt', encoding='utf-8') as f:
|
|
return [l.rstrip('\n') for l in f]
|
|
|
|
|
|
def sans_manque(niveau, cle_motif):
|
|
motif = niveau.get(cle_motif)
|
|
if isinstance(motif, dict) and motif.get('code') == 'ECO_FLOOR_MISSED':
|
|
motif.get('params', {}).pop('missingWh', None)
|
|
prog = niveau.get('progress')
|
|
if isinstance(prog, dict):
|
|
prog.pop('closedMissingWh', None)
|
|
|
|
|
|
def normaliser(ligne):
|
|
"""Rend (ligne sans motifs, [(charge, motif retiré)])."""
|
|
d = json.loads(ligne)
|
|
retires = []
|
|
if d.get('src') == 'cycle':
|
|
for ch in (d.get('telemetry') or {}).get('loads', []) or []:
|
|
retires.append((ch.get('loadId'), ch.pop('decision', None)))
|
|
for n in ch.get('levels', []) or []:
|
|
sans_manque(n, 'decision')
|
|
elif d.get('src') == 'plan':
|
|
for s in (d.get('plan') or {}).get('timeSlots', []) or []:
|
|
for a in s.get('actions', []) or []:
|
|
retires.append((a.get('loadId'), a.pop('reason', None)))
|
|
for n in a.get('levels', []) or []:
|
|
sans_manque(n, 'reason')
|
|
return json.dumps(d, sort_keys=True, ensure_ascii=False), retires
|
|
|
|
|
|
def motif_court(m):
|
|
if not m:
|
|
return '—'
|
|
code = m.get('code', '?')
|
|
niveau = m.get('level')
|
|
manque = (m.get('params') or {}).get('missingWh')
|
|
s = code + (f' ({niveau})' if niveau else '')
|
|
return s + (f' missingWh={manque}' if manque is not None else '')
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 3:
|
|
print(__doc__ or 'usage : attendu obtenu [--cas sortie.md]', file=sys.stderr)
|
|
return 2
|
|
attendu, obtenu = lire(sys.argv[1]), lire(sys.argv[2])
|
|
sortie_cas = sys.argv[sys.argv.index('--cas') + 1] if '--cas' in sys.argv else None
|
|
|
|
if len(attendu) != len(obtenu):
|
|
print(f'NOMBRE DE LIGNES différent : {len(attendu)} attendues, {len(obtenu)} obtenues')
|
|
return 1
|
|
|
|
brutes_identiques = 0
|
|
cas = Counter()
|
|
for i, (a, b) in enumerate(zip(attendu, obtenu), 1):
|
|
if a == b:
|
|
brutes_identiques += 1
|
|
continue
|
|
na, ra = normaliser(a)
|
|
nb, rb = normaliser(b)
|
|
if na != nb:
|
|
print(f'ÉCART HORS MOTIFS à la ligne {i} :')
|
|
print('- ' + a[:3000])
|
|
print('+ ' + b[:3000])
|
|
return 1
|
|
fn = json.loads(a).get('fn')
|
|
for (charge, ma), (_, mb) in zip(ra, rb):
|
|
if ma != mb:
|
|
cas[(fn, charge, motif_court(ma), motif_court(mb))] += 1
|
|
# Le manque d'un niveau change sans que le motif de tête change : à dire aussi.
|
|
if ra == rb:
|
|
cas[(fn, '(niveau)', 'missingWh', 'missingWh corrigé')] += 1
|
|
|
|
print(f'IDENTIQUE HORS MOTIFS : {len(attendu)} lignes, dont {brutes_identiques} identiques '
|
|
f'à l\'octet et {len(attendu) - brutes_identiques} qui ne diffèrent que par les motifs')
|
|
lignes = ['| scénario (fonction) | charge | ancien motif | nouveau motif | lignes |',
|
|
'|---|---|---|---|---|']
|
|
for (fn, charge, avant, apres), n in sorted(cas.items()):
|
|
lignes.append(f'| `{fn}` | `{charge}` | {avant} | {apres} | {n} |')
|
|
texte = '\n'.join(lignes)
|
|
if sortie_cas:
|
|
with open(sortie_cas, 'w', encoding='utf-8') as f:
|
|
f.write(texte + '\n')
|
|
else:
|
|
print(texte)
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|