#!/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()