EVCS/tools/analyse-historique-pylontech.py

#!/usr/bin/env python3
"""Analyse les exports d'historique BatteryView d'un module Pylontech.

BatteryView enregistre la sortie brute de la console du module. Chaque ligne
de `datalist history dtls <n>` porte l'etat du module ET les quinze cellules,
toutes les trente minutes. Le module en garde 1798, soit trente-sept jours.

    python3 tools/analyse-historique-pylontech.py docs/mesures/batteries/*_history_*.txt

Ce que l'outil cherche : quelles cellules divergent, de combien, a quel niveau
de charge, et si cela s'aggrave. Les tensions sont en millivolts.
"""
import collections
import glob
import re
import sys

# Echelle d'alarme reelle, lue le 05/09/2026 dans `sysinfo` du LVHUB.
# Ce ne sont pas des seuils choisis ici : ce sont ceux du materiel.
#   haut  : niveau 1 3480 mV · niveau 2 3520 · niveau 3 3550 · niveau 4 3600
#   bas   : niveau 1 3100 mV · niveau 2 2900
SEUIL_HAUT_MV = 3550   # niveau 3 : le BMS reduit le courant de charge
SEUIL_BAS_MV = 3100    # niveau 1 bas


def lire(chemin):
    """Retourne les enregistrements, dedoublonnes par identifiant.

    Les pages de `datalist history dtls` se chevauchent : le meme
    enregistrement apparait dans plusieurs exports. L'identifiant tranche.
    """
    recs = {}
    brut = open(chemin, "rb").read()
    txt = brut.decode("utf-16" if brut[:2] in (b"\xff\xfe", b"\xfe\xff") else "utf-8",
                      errors="replace").replace("\r", "")
    for ligne in txt.split("\n"):
        ch = ligne.split()
        if len(ch) < 18 + 15 * 6:
            continue
        if not re.fullmatch(r"\d+", ch[0]) or not re.fullmatch(r"\d\d-\d\d-\d\d", ch[1]):
            continue
        recs[int(ch[0])] = dict(
            id=int(ch[0]), jour=ch[1], heure=ch[2],
            volt=int(ch[3]), curr=int(ch[4]),
            etat=ch[10], soc=int(ch[16].rstrip("%")),
            cellules=[int(ch[18 + k * 6 + 1]) for k in range(15)],
        )
    return recs


def main(motifs):
    recs = {}
    for m in motifs:
        for f in sorted(glob.glob(m)):
            recs.update(lire(f))
    r = [recs[k] for k in sorted(recs)]
    if not r:
        sys.exit("Aucun enregistrement lisible.")

    print(f"{len(r)} enregistrements, du {r[0]['jour']} {r[0]['heure']} "
          f"au {r[-1]['jour']} {r[-1]['heure']} (heure INTERNE du module)")
    print()

    print("TENSION EXTREME ATTEINTE PAR CHAQUE CELLULE")
    hauts = []
    for i in range(15):
        vs = [x["cellules"][i] for x in r]
        marque = ""
        if max(vs) > SEUIL_HAUT_MV:
            marque = "  <-- monte trop haut ET descend trop bas"
            hauts.append(i + 1)
        print(f"  cellule {i+1:>2} : max {max(vs)} mV   min {min(vs)} mV{marque}")
    print()
    if hauts:
        print(f"  Les cellules {', '.join(map(str, hauts))} se distinguent des "
              f"{15 - len(hauts)} autres.")
        print("  Monter plus haut en charge ET descendre plus bas en decharge est la")
        print("  signature d'une capacite moindre : elles se remplissent et se vident")
        print("  avant les autres. Ce n'est pas un defaut de mesure.")
    print()

    print("ECART ENTRE CELLULES, EN CHARGE, SELON LE NIVEAU DE CHARGE")
    print(f"  {'SOC':>9} {'releves':>8} {'ecart median':>14} {'ecart max':>11} {'cellule la plus haute':>22}")
    tr = collections.defaultdict(list)
    for x in r:
        if x["etat"] == "Charge":
            tr[min(90, x["soc"] // 10 * 10)].append(x)
    for k in sorted(tr):
        e = sorted(max(x["cellules"]) - min(x["cellules"]) for x in tr[k])
        vm = max(max(x["cellules"]) for x in tr[k])
        print(f"  {k:>4}-{k+9:<4} {len(e):>8} {e[len(e)//2]:>11} mV {max(e):>8} mV {vm:>19} mV")
    print()

    print(f"DEPASSEMENTS DE {SEUIL_HAUT_MV} mV, PAR JOUR")
    j = collections.Counter(x["jour"] for x in r if max(x["cellules"]) > SEUIL_HAUT_MV)
    if not j:
        print("  aucun")
    else:
        for k in sorted(j):
            print(f"  {k} : {j[k]:>2} releve(s) sur 48")
        n = sorted(j.values())
        print(f"  total {sum(j.values())} releves sur {len(r)}, repartis sur {len(j)} jours ; "
              f"median {n[len(n)//2]} par jour")
        print("  Un nombre stable jour apres jour decrit un desequilibre chronique,")
        print("  pas une degradation en cours.")
    print()

    print("LES DIX PIRES MOMENTS")
    for x in sorted(r, key=lambda z: max(z["cellules"]) - min(z["cellules"]), reverse=True)[:10]:
        hi, lo = max(x["cellules"]), min(x["cellules"])
        print(f"  {x['jour']} {x['heure']}  SOC {x['soc']:>3}%  {x['etat']:<7} "
              f"parc {x['volt']/1000:5.2f} V  {x['curr']/1000:>6.2f} A  "
              f"ecart {hi-lo:>3} mV  haute = cellule {x['cellules'].index(hi)+1} ({hi} mV)")


if __name__ == "__main__":
    main(sys.argv[1:] or ["docs/mesures/batteries/*_history_*.txt"])