etm-powersync-plugins-modbus/sunspec/tools/read-float-order.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Quel ordre de mots un onduleur SunSpec émet-il pour ses float32 ?

La spécification SunSpec dit big endian, mot de poids fort en premier. Mais
libnymea-sunspec décode par défaut `ByteOrderLittleEndian`, c'est à dire
`registers.at(1)` PUIS `registers.at(0)` — mot de poids faible d'abord
(sunspecdatapoint.cpp, convertToUInt32). Le vendor Fronius du plugin garde ce
défaut. Si l'appareil émet en big endian, le plugin lit des dénormaux (~1e-41)
et aucune correction de SIGNE n'y changera quoi que ce soit.

Ce script tranche la question par la mesure, sur deux points dont la valeur
attendue est connue d'avance et non ambiguë :
  - Hz      : ~50 Hz
  - PhVphA  : ~230..240 V
Un ordre donne ces valeurs, l'autre donne des dénormaux. Il n'y a pas de doute
possible entre les deux.

LECTURE SEULE : ce script n'émet QUE des fonctions Modbus 3 (read holding
registers). Aucune écriture, aucun code fonction d'écriture n'est implémenté.

    python3 read-float-order.py 192.168.3.230
    python3 read-float-order.py 192.168.3.230 --inverter-unit 1 --meter-unit 200

Bibliothèque standard uniquement : rien à installer.
"""

import argparse
import socket
import struct
import sys

SUNS = (0x5375, 0x6E53)

# Offsets comptés depuis le registre ID du modèle (les données commencent à +2).
# Chaque point float32 occupe DEUX registres, d'où le pas de 2.
INV_F = {"W": 22, "Hz": 24, "PhVphA": 16, "PhVphB": 18, "PhVphC": 20, "DCW": 38}
MET_F = {"W": 28, "WphA": 30, "WphB": 32, "WphC": 34,
         "Hz": 26, "PhV": 10, "PhVphA": 12, "PhVphB": 14, "PhVphC": 16}

INVERTER_FLOAT = (111, 112, 113)
METER_FLOAT = (211, 212, 213, 214)
INVERTER_INT = (101, 102, 103)
METER_INT = (201, 202, 203, 204)


def read_holding(sock, unit, addr, count):
    """Fonction 3, et rien d'autre."""
    sock.sendall(struct.pack(">HHHBBHH", 1, 0, 6, unit, 3, addr, count))
    hdr = b""
    while len(hdr) < 8:
        chunk = sock.recv(8 - len(hdr))
        if not chunk:
            raise IOError("connexion fermée (en-tête)")
        hdr += chunk
    _, _, length, _, fc = struct.unpack(">HHHBB", hdr)
    body = b""
    while len(body) < length - 2:
        chunk = sock.recv(length - 2 - len(body))
        if not chunk:
            raise IOError("connexion fermée (corps)")
        body += chunk
    if fc & 0x80:
        raise IOError("exception modbus %d en lisant %d registres à %d"
                      % (body[0], count, addr))
    return list(struct.unpack(">%dH" % (body[0] // 2), body[1:1 + body[0]]))


def decode(r0, r1, low_word_first):
    hi, lo = (r1, r0) if low_word_first else (r0, r1)
    return struct.unpack(">f", struct.pack(">HH", hi, lo))[0]


def find_base(sock, unit):
    for base in (40000, 50000, 0):
        try:
            if tuple(read_holding(sock, unit, base, 2)) == SUNS:
                return base
        except IOError:
            continue
    return None


def walk(sock, unit, base):
    """(model_id, adresse du registre ID, longueur) pour toute la chaîne."""
    out, addr = [], base + 2
    while addr < base + 5000:
        mid, mlen = read_holding(sock, unit, addr, 2)
        if mid == 0xFFFF:
            return out
        out.append((mid, addr, mlen))
        addr += 2 + mlen
    return out


def verdict(hz_low, hz_big):
    """Quel ordre donne une fréquence réseau plausible ?"""
    def plausible(v):
        return 45.0 <= v <= 65.0
    if plausible(hz_low) and not plausible(hz_big):
        return "MOT DE POIDS FAIBLE D'ABORD (ce que le plugin lit déjà)"
    if plausible(hz_big) and not plausible(hz_low):
        return ("BIG ENDIAN, conforme à la spec SunSpec — un plugin qui garde le "
                "défaut de libnymea-sunspec lira des dénormaux")
    if hz_low == 0.0 and hz_big == 0.0:
        return "INDÉCIDABLE : registres à zéro, aucun ordre ne se distingue"
    return "INDÉCIDABLE : aucune des deux lectures ne donne 45..65 Hz"


def probe(host, port, unit, wanted_float, wanted_int, table, points, role):
    print("=" * 74)
    print("%s — unité %d" % (role, unit))
    print("=" * 74)
    sock = socket.create_connection((host, port), timeout=10)
    try:
        base = find_base(sock, unit)
        if base is None:
            print("  aucun marqueur « SunS » : rien à cette unité.\n")
            return
        chain = walk(sock, unit, base)
        print("  base %d · chaîne : %s" % (
            base, ", ".join("%d@%d(L=%d)" % (m, a, l) for m, a, l in chain)))

        found = [c for c in chain if c[0] in wanted_float]
        if not found:
            ints = [c for c in chain if c[0] in wanted_int]
            if ints:
                print("  modèle %d : ENTIER+SF, pas de float32 — la question de "
                      "l'ordre des mots ne se pose pas ici.\n" % ints[0][0])
            else:
                print("  aucun modèle attendu dans cette chaîne.\n")
            return

        mid, id_addr, mlen = found[0]
        print("  modèle float %d, registre ID %d, longueur %d\n" % (mid, id_addr, mlen))
        print("  %-8s %-9s %-14s %14s %16s  %s" % (
            "point", "registre", "brut (hex)", "mot faible 1er", "big endian", "lecture"))
        print("  " + "-" * 86)
        hz = None
        for name in points:
            off = table[name]
            addr = id_addr + off
            r0, r1 = read_holding(sock, unit, addr, 2)
            low, big = decode(r0, r1, True), decode(r0, r1, False)
            if name == "Hz":
                hz = (low, big)
            note = "nul — n'apprend rien" if (r0 == 0 and r1 == 0) else ""
            print("  %-8s %-9d %-14s %14.4f %16.4e  %s" % (
                name, addr, "%04X %04X" % (r0, r1), low, big, note))
        print()
        if hz:
            print("  >>> %s" % verdict(*hz))
        print()
    finally:
        sock.close()


def main():
    ap = argparse.ArgumentParser(
        description="Ordre des mots des float32 SunSpec — LECTURE SEULE.")
    ap.add_argument("host")
    ap.add_argument("--port", type=int, default=502)
    ap.add_argument("--inverter-unit", type=int, default=1)
    ap.add_argument("--meter-unit", type=int, default=200)
    args = ap.parse_args()

    # Le COMPTEUR d'abord : c'est lui qui tranche. Il mesure le réseau en
    # permanence, donc ses registres portent des valeurs même onduleur en veille.
    # Un onduleur au repos sert un bloc à zéro, et zéro se décode en 0.0 dans les
    # DEUX ordres : ça n'apprend rien.
    probe(args.host, args.port, args.meter_unit, METER_FLOAT, METER_INT,
          MET_F, ("Hz", "PhVphA", "PhVphB", "PhVphC", "W", "WphA"),
          "COMPTEUR — mesure décisive")
    probe(args.host, args.port, args.inverter_unit, INVERTER_FLOAT, INVERTER_INT,
          INV_F, ("Hz", "PhVphA", "PhVphB", "PhVphC", "W"),
          "ONDULEUR — confirmation, nul si en veille")

    print("Lecture : la colonne qui donne ~50 Hz et ~230..240 V est l'ordre que")
    print("l'appareil émet. L'autre affiche des dénormaux (~1e-41) : c'est la")
    print("signature d'une mauvaise interprétation, pas une vraie mesure.")
    print()
    print("Si « big endian » est la bonne colonne, tout lecteur qui garde le défaut")
    print("de libnymea-sunspec (mot faible d'abord) lira ces modèles de travers.")
    print("Le vendor Fronius du plugin force big endian depuis 1.15.2+etm9.")
    return 0


if __name__ == "__main__":
    sys.exit(main())