administrator/domotica
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
domotica/concentratore/concentratore.py
andrea castagnino d30984adfe primo caricamento
2026-08-27 12:28:06 +02:00

409 lines
15 KiB
Python
Raw Permalink Blame History

#!/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()
Reference in a new issue View git blame Copy permalink