diff --git a/tests/auto/decisions/etm-powersync-compare-hors-motifs.py b/tests/auto/decisions/etm-powersync-compare-hors-motifs.py new file mode 100755 index 0000000..523223f --- /dev/null +++ b/tests/auto/decisions/etm-powersync-compare-hors-motifs.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +# Comparateur « IDENTIQUE HORS MOTIFS » (correctifs moteur a et b, etm70). +# +# etm-powersync-compare-hors-motifs.py [--cas ] +# +# 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())