etm-powersync-energy-plugin-etm/tests/auto/decisions/etm-powersync-compare-hors-motifs.py

#!/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>]
#                                         [--achat] [--strategie] [--lignes <sortie.txt>]
#
# 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.
# Avec --achat (correctif c, etm71), il retire EN PLUS `progress.purchasedWh` de chaque niveau :
# l'acheté se compte désormais sur l'énergie mesurée, et il est absent sans mesure. Un lot qui
# change cette VALEUR PUBLIÉE sans toucher aux consignes doit passer ce mode-là.
# Avec --strategie (brief 4 T6, décision du 2026-10-17), il ramène EN PLUS `plan.strategy`
# `rule-based` à `rules` d'une ligne de plan direct : le renommage du plan rule-based, et lui seul.
# 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 ; --lignes écrit le numéro et le contenu de chaque ligne qui
# change (attendu puis obtenu).
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)


ABSENT = '(absent)'


def sans_achat(niveau, charge, achats):
    prog = niveau.get('progress')
    if isinstance(prog, dict):
        achats.append((charge, prog.pop('purchasedWh', ABSENT)))


def normaliser(ligne, achat=False, strategie=False):
    """Rend (ligne sans motifs, [(charge, motif retiré)], [(charge, purchasedWh retiré)])."""
    d = json.loads(ligne)
    retires = []
    achats = []
    if strategie and d.get('src') == 'plan' and (d.get('plan') or {}).get('strategy') == 'rule-based':
        d['plan']['strategy'] = 'rules'
    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')
                if achat:
                    sans_achat(n, ch.get('loadId'), achats)
    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')
                    if achat:
                        sans_achat(n, a.get('loadId'), achats)
    return json.dumps(d, sort_keys=True, ensure_ascii=False), retires, achats


def achat_court(v):
    if isinstance(v, (int, float)):
        return f'{round(v, 2):g}'
    return str(v)


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
    sortie_lignes = sys.argv[sys.argv.index('--lignes') + 1] if '--lignes' in sys.argv else None
    achat = '--achat' in sys.argv
    strategie = '--strategie' in sys.argv
    lignes_changees = []

    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, aa = normaliser(a, achat, strategie)
        nb, rb, ab = normaliser(b, achat, strategie)
        lignes_changees.append((i, a, b))
        if na != nb:
            print(f'ÉCART HORS MOTIFS{" ET HORS ACHAT" if achat else ""}'
                  f'{" ET HORS STRATÉGIE" if strategie else ""} à 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
        # L'acheté d'un niveau éco (--achat) : ancienne valeur → nouvelle, ou devenu absent.
        for (charge, va), (_, vb) in zip(aa, ab):
            if va != vb:
                cas[(fn, charge, f'purchasedWh {achat_court(va)}', f'purchasedWh {achat_court(vb)}')] += 1
        sa = (json.loads(a).get('plan') or {}).get('strategy')
        sb = (json.loads(b).get('plan') or {}).get('strategy')
        if sa != sb:
            cas[(fn, '(plan)', f'strategy {sa}', f'strategy {sb}')] += 1
        # Le manque d'un niveau change sans que le motif de tête change : à dire aussi.
        elif ra == rb and aa == ab:
            cas[(fn, '(niveau)', 'missingWh', 'missingWh corrigé')] += 1

    hors = 'HORS MOTIFS' + (' ET HORS ACHAT' if achat else '') + (' ET HORS STRATÉGIE' if strategie else '')
    print(f'IDENTIQUE {hors} : {len(attendu)} lignes, dont {brutes_identiques} identiques '
          f'à l\'octet et {len(attendu) - brutes_identiques} qui ne diffèrent que par '
          f'{"les motifs" + (", purchasedWh" if achat else "") + (" ou plan.strategy" if strategie else "")}')
    if sortie_lignes:
        with open(sortie_lignes, 'w', encoding='utf-8') as f:
            for i, a, b in lignes_changees:
                f.write(f'## ligne {i}\n- {a}\n+ {b}\n')
    lignes = ['| scénario (fonction) | charge | avant | après | 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())