primo caricamento

concentratore/concentratore.py

@@ -0,0 +1,409 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+CONCENTRATORE Modbus - si presenta al CCI come se fosse il gateway degli inverter.
+
+Il CCI (o mbpoll, o Modula) si collega qui con gli STESSI slave ID e gli STESSI
+registri che userebbe sul Waveshare. Il concentratore:
+
+  - risponde SUBITO alle scritture (poche decine di ms invece di ~310 ms)
+  - propaga il comando agli inverter veri e lo tiene RINFRESCATO, perche' i
+    50KTL-V3 lasciano decadere il limite se nessuno lo riscrive
+  - tiene in memoria le letture di potenza e le serve istantaneamente
+  - dichiara in due registri di stato se il comando e' stato davvero recepito
+
+Nessuna libreria da installare: solo la libreria standard di Python 3.
+
+Avvio:
+    python3 concentratore.py --gateway 192.168.4.240 --slaves 1,2,3,4,5
+
+    (porta 5020 di default: sotto la 1024 servirebbero i permessi di root)
+
+Poi si prova puntandoci contro mbpoll, cambiando SOLO l'indirizzo:
+    mbpoll -a 1 -t 4 -p 5020 -r 4359 -c 1 -1 127.0.0.1
+    mbpoll -a 1 -t 4 -p 5020 -r 4358 127.0.0.1 1 200
+
+REGISTRI ESPOSTI (identici a quelli degli inverter ZCS serie V3):
+    4357  0x1105  abilitazione controllo remoto   lettura/scrittura
+    4358  0x1106  setpoint in decimi di percento  lettura/scrittura
+    1157  0x0485  potenza attiva (decine di W)    sola lettura
+
+REGISTRI DI STATO (aggiunti da noi - vanno dichiarati ad AiLux):
+    8192  0x2000  1 = comando confermato su questo inverter, 0 = non confermato
+    8193  0x2001  decimi di secondo dall'ultima conferma (65535 = mai)
+    8194  0x2002  quanti inverter su N hanno il comando confermato
+    8195  0x2003  potenza totale d'impianto in decine di W
+
+NOTE DI VERSIONE
+    1.0.2  - corretta la corsa fra CCI e motore: la conferma ora riguarda i
+             valori davvero inviati, non quelli presenti al ritorno della
+             scrittura. Prima un comando che arrivava durante la transazione
+             upstream veniva dichiarato confermato senza essere mai spedito.
+           - il thread motore non muore piu' in silenzio su un'eccezione
+             imprevista: logga e riprende. Il watchdog dell'add-on controlla
+             solo che la porta sia aperta, quindi non avrebbe visto nulla.
+           - la potenza di un inverter che smette di rispondere viene azzerata
+             dopo un periodo di grazia, invece di restare congelata per sempre
+             e continuare a sommarsi nel registro 8195.
+           - i comandi in attesa vengono propagati prima delle letture, cosi'
+             un setpoint nuovo non aspetta un giro di scansione completo.
+"""
+
+import argparse
+import socket
+import socketserver
+import struct
+import threading
+import time
+
+REG_ENABLE = 4357        # 0x1105
+REG_SETPOINT = 4358      # 0x1106
+REG_POWER = 1157         # 0x0485
+REG_ST_OK = 8192         # 0x2000
+REG_ST_ETA = 8193        # 0x2001
+REG_ST_CONTA = 8194      # 0x2002
+REG_ST_TOT = 8195        # 0x2003
+
+cfg = {}
+stato = {}
+_stato_lock = threading.Lock()
+
+
+# ------------------------------------------------------------------ upstream
+
+class Upstream:
+    """Connessione persistente verso il Waveshare. Una transazione alla volta."""
+
+    def __init__(self, ip, porta, timeout):
+        self.ip, self.porta, self.timeout = ip, porta, timeout
+        self.sock = None
+        self.tid = 0
+        self.lock = threading.Lock()
+
+    def _apri(self):
+        if self.sock:
+            return
+        self.sock = socket.create_connection((self.ip, self.porta), timeout=self.timeout)
+        self.sock.settimeout(self.timeout)
+
+    def _chiudi(self):
+        if self.sock:
+            try:
+                self.sock.close()
+            except OSError:
+                pass
+        self.sock = None
+
+    def txn(self, unit, pdu):
+        """Restituisce il PDU di risposta, oppure None. Verifica id e slave."""
+        with self.lock:
+            for tentativo in range(2):
+                try:
+                    self._apri()
+                    self.tid = (self.tid + 1) % 65000
+                    mio = self.tid
+                    self.sock.sendall(struct.pack('>HHHB', mio, 0, len(pdu) + 1, unit) + pdu)
+                    for _ in range(4):
+                        testa = self._leggi(6)
+                        t, proto, ln = struct.unpack('>HHH', testa)
+                        corpo = self._leggi(ln)
+                        if t == mio and corpo[0] == unit:
+                            return corpo[1:]
+                    # quattro risposte spaiate: il flusso e' disallineato,
+                    # meglio ripartire da una connessione pulita
+                    self._chiudi()
+                    return None
+                except (OSError, struct.error):
+                    self._chiudi()
+                    if tentativo == 0:
+                        time.sleep(0.1)
+            return None
+
+    def _leggi(self, n):
+        buf = b''
+        while len(buf) < n:
+            c = self.sock.recv(n - len(buf))
+            if not c:
+                raise OSError('connessione chiusa')
+            buf += c
+        return buf
+
+    def leggi_reg(self, unit, addr, quanti=1):
+        r = self.txn(unit, struct.pack('>BHH', 3, addr, quanti))
+        if not r or r[0] != 3:
+            return None
+        n = r[1]
+        if len(r) < 2 + n:
+            return None
+        return list(struct.unpack('>%dH' % quanti, r[2:2 + quanti * 2]))
+
+    def scrivi_reg(self, unit, addr, valori):
+        dati = b''.join(struct.pack('>H', v & 0xFFFF) for v in valori)
+        r = self.txn(unit, struct.pack('>BHHB', 16, addr, len(valori), len(dati)) + dati)
+        return bool(r) and r[0] == 16
+
+
+# ------------------------------------------------------------------ motore
+
+def da_propagare(u):
+    """C'e' un comando da mandare a questo inverter, adesso?"""
+    with _stato_lock:
+        s = stato[u]
+        if s['sporco']:
+            return True
+        # il rinfresco serve solo mentre il limite e' attivo: se il controllo
+        # e' disabilitato basta scrivere una volta al cambiamento
+        return bool(s['enable']) and (time.time() - s['ultimo_invio']) >= cfg['refresh']
+
+
+def propaga(up, u):
+    """Manda enable+setpoint a un inverter e registra cosa e' stato inviato."""
+    with _stato_lock:
+        s = stato[u]
+        inviato = (s['enable'], s['setpoint'])
+
+    ok = up.scrivi_reg(u, REG_ENABLE, list(inviato))
+
+    with _stato_lock:
+        s = stato[u]
+        s['ultimo_invio'] = time.time()
+        if ok:
+            s['confermato'] = inviato
+            s['ultima_conferma'] = time.time()
+            # se nel frattempo il CCI ha scritto qualcosa di diverso, il lavoro
+            # non e' finito: si resta sporchi e si riparte al giro successivo
+            if (s['enable'], s['setpoint']) == inviato:
+                s['sporco'] = False
+            s['errori'] = 0
+        else:
+            s['errori'] += 1
+        errori = s['errori']
+
+    if cfg['verboso']:
+        print('  -> inv%d  enable=%d sp=%d  %s'
+              % (u, inviato[0], inviato[1], 'ok' if ok else 'PERSA (x%d)' % errori))
+
+
+def aggiorna_potenza(up, u):
+    """Legge la potenza e la invecchia se l'inverter smette di rispondere."""
+    regs = up.leggi_reg(u, REG_POWER, 1)
+    with _stato_lock:
+        s = stato[u]
+        if regs:
+            s['potenza'] = regs[0]
+            s['letta'] = time.time()
+        elif s['letta'] and (time.time() - s['letta']) > cfg['scadenza']:
+            # meglio dichiarare zero che lasciare il CCI a regolare su una
+            # potenza fantasma di un inverter che non c'e' piu'
+            s['potenza'] = 0
+            s['letta'] = None
+            if cfg['verboso']:
+                print('  !! inv%d  muto da oltre %.0f s: potenza azzerata'
+                      % (u, cfg['scadenza']))
+
+
+def ciclo(up):
+    inizio = time.time()
+
+    # prima i comandi in attesa, tutti: un setpoint nuovo non deve aspettare
+    # che finisca il giro delle letture
+    for u in cfg['slaves']:
+        if da_propagare(u):
+            propaga(up, u)
+
+    # poi le letture, ricontrollando fra una e l'altra se e' arrivato altro
+    for u in cfg['slaves']:
+        if da_propagare(u):
+            propaga(up, u)
+        aggiorna_potenza(up, u)
+
+    durata = time.time() - inizio
+    if durata < 0.2:
+        time.sleep(0.2 - durata)
+
+
+def motore(up):
+    """Propaga i comandi agli inverter e aggiorna le letture. Gira da solo.
+
+    Qualunque eccezione imprevista viene loggata e il ciclo riprende: se questo
+    thread morisse, il concentratore continuerebbe a rispondere al CCI con dati
+    congelati e nessuno se ne accorgerebbe (il watchdog guarda solo la porta).
+    """
+    while True:
+        try:
+            ciclo(up)
+        except Exception as e:
+            print('MOTORE: eccezione imprevista, riprendo -> %r' % e)
+            time.sleep(1.0)
+
+
+# ------------------------------------------------------------------ registri
+
+def leggi_virtuale(unit, addr, quanti):
+    """Serve le letture dalla memoria, senza toccare il bus."""
+    fuori = []
+    with _stato_lock:
+        s = stato.get(unit)
+        if not s:
+            return None
+        totale = sum(x['potenza'] for x in stato.values())
+        confermati = sum(1 for x in stato.values()
+                         if x['confermato'] == (x['enable'], x['setpoint']))
+        for i in range(quanti):
+            a = addr + i
+            if a == REG_ENABLE:
+                fuori.append(s['enable'])
+            elif a == REG_SETPOINT:
+                fuori.append(s['setpoint'])
+            elif a == REG_POWER:
+                fuori.append(s['potenza'])
+            elif a == REG_ST_OK:
+                fuori.append(1 if s['confermato'] == (s['enable'], s['setpoint']) else 0)
+            elif a == REG_ST_ETA:
+                if not s['ultima_conferma']:
+                    fuori.append(65535)
+                else:
+                    fuori.append(min(65534, int((time.time() - s['ultima_conferma']) * 10)))
+            elif a == REG_ST_CONTA:
+                fuori.append(confermati)
+            elif a == REG_ST_TOT:
+                fuori.append(min(65535, totale))
+            else:
+                return None
+    return fuori
+
+
+def scrivi_virtuale(unit, addr, valori):
+    """Accetta il comando, risponde subito, lascia al motore il lavoro sporco."""
+    with _stato_lock:
+        s = stato.get(unit)
+        if not s:
+            return False
+        for i, v in enumerate(valori):
+            a = addr + i
+            if a == REG_ENABLE:
+                s['enable'] = v & 0xFFFF
+            elif a == REG_SETPOINT:
+                s['setpoint'] = max(0, min(1000, v & 0xFFFF))
+            else:
+                return False
+        s['sporco'] = True
+    return True
+
+
+# ------------------------------------------------------------------ server
+
+class Handler(socketserver.BaseRequestHandler):
+
+    def handle(self):
+        c = self.request
+        c.settimeout(120)
+        try:
+            while True:
+                testa = self._leggi(c, 6)
+                if not testa:
+                    return
+                tid, proto, ln = struct.unpack('>HHH', testa)
+                corpo = self._leggi(c, ln)
+                if not corpo or proto != 0:
+                    return
+                unit, fc = corpo[0], corpo[1]
+                t0 = time.time()
+                pdu = self._servi(unit, fc, corpo[2:])
+                c.sendall(struct.pack('>HHHB', tid, 0, len(pdu) + 1, unit) + pdu)
+                if cfg['verboso']:
+                    print('CCI  inv%d  fc%-3d  risposto in %5.1f ms'
+                          % (unit, fc, 1000 * (time.time() - t0)))
+        except (OSError, struct.error):
+            pass
+
+    def _leggi(self, c, n):
+        buf = b''
+        while len(buf) < n:
+            try:
+                d = c.recv(n - len(buf))
+            except socket.timeout:
+                return None
+            if not d:
+                return None
+            buf += d
+        return buf
+
+    def _servi(self, unit, fc, dati):
+        try:
+            if fc in (3, 4):
+                addr, quanti = struct.unpack('>HH', dati[:4])
+                v = leggi_virtuale(unit, addr, quanti)
+                if v is None:
+                    return struct.pack('>BB', fc | 0x80, 2)
+                return struct.pack('>BB', fc, quanti * 2) + b''.join(
+                    struct.pack('>H', x & 0xFFFF) for x in v)
+
+            if fc == 6:
+                addr, val = struct.unpack('>HH', dati[:4])
+                if not scrivi_virtuale(unit, addr, [val]):
+                    return struct.pack('>BB', 0x86, 2)
+                return struct.pack('>BHH', 6, addr, val)
+
+            if fc == 16:
+                addr, quanti = struct.unpack('>HH', dati[:4])
+                nbyte = dati[4]
+                valori = list(struct.unpack('>%dH' % quanti, dati[5:5 + nbyte]))
+                if not scrivi_virtuale(unit, addr, valori):
+                    return struct.pack('>BB', 0x90, 2)
+                return struct.pack('>BHH', 16, addr, quanti)
+        except (struct.error, IndexError):
+            return struct.pack('>BB', fc | 0x80, 3)
+        return struct.pack('>BB', fc | 0x80, 1)
+
+
+class Server(socketserver.ThreadingTCPServer):
+    allow_reuse_address = True
+    daemon_threads = True
+
+
+# ------------------------------------------------------------------ avvio
+
+def main():
+    ap = argparse.ArgumentParser(description='Concentratore Modbus per inverter ZCS')
+    ap.add_argument('--gateway', default='192.168.4.240', help='IP del Waveshare')
+    ap.add_argument('--gateway-porta', type=int, default=502)
+    ap.add_argument('--slaves', default='1,2,3,4,5')
+    ap.add_argument('--host', default='0.0.0.0', help='su quale interfaccia ascoltare')
+    ap.add_argument('--porta', type=int, default=5020, help='porta di ascolto per il CCI')
+    ap.add_argument('--refresh', type=float, default=2.0,
+                    help='ogni quanti secondi riscrivere il setpoint agli inverter')
+    ap.add_argument('--timeout', type=float, default=1.5)
+    ap.add_argument('--zitto', action='store_true', help='niente log a schermo')
+    a = ap.parse_args()
+
+    slaves = [int(x) for x in a.slaves.split(',') if x.strip()]
+    cfg.update(slaves=slaves,
+               refresh=a.refresh,
+               verboso=not a.zitto,
+               # dopo quanto una lettura di potenza va considerata morta:
+               # abbondante rispetto al giro di scansione, ma non infinita
+               scadenza=max(10.0, 5 * a.refresh))
+
+    for u in cfg['slaves']:
+        stato[u] = dict(enable=0, setpoint=1000, sporco=False,
+                        confermato=None, ultimo_invio=0.0, ultima_conferma=None,
+                        potenza=0, letta=None, errori=0)
+
+    up = Upstream(a.gateway, a.gateway_porta, a.timeout)
+    threading.Thread(target=motore, args=(up,), daemon=True).start()
+
+    srv = Server((a.host, a.porta), Handler)
+    print('CONCENTRATORE in ascolto su %s:%d' % (a.host, a.porta))
+    print('inverter a valle: %s  via %s:%d' % (cfg['slaves'], a.gateway, a.gateway_porta))
+    print('rinfresco del setpoint ogni %.1f s (solo mentre il limite e\' attivo)' % a.refresh)
+    print('potenza azzerata dopo %.0f s di silenzio di un inverter' % cfg['scadenza'])
+    print('Ctrl+C per chiudere.\n')
+    try:
+        srv.serve_forever()
+    except KeyboardInterrupt:
+        print('\nchiuso. Attenzione: gli inverter restano con l\'ultimo comando dato.')
+
+
+if __name__ == '__main__':
+    main()