#!/usr/bin/env python3 """ Intercepteur Growatt autonome - Remplace Grott Compatible avec les onduleurs SPF (T06NNNNSPF) """ import socket import threading import time import struct import json import codecs from datetime import datetime from flask import Flask, jsonify, render_template_string from itertools import cycle import logging # Configuration du logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) app = Flask(__name__) # Configuration CONFIG = { 'proxy_port': 5279, # Port d'écoute pour les ESP 'web_port': 8080, # Port pour l'interface web 'growatt_server': '47.91.67.66', # Serveur Growatt réel (vu dans vos logs) 'growatt_port': 5279, 'verbose': True, 'forward_to_growatt': True, # Transférer les données vers Growatt } # Stockage global des données devices_data = {} raw_messages = [] statistics = { 'total_messages': 0, 'last_update': None, 'devices_seen': set() } class GrowattDecryptor: """Déchiffrement des données Growatt""" @staticmethod def decrypt(data): """Déchiffre les données selon l'algorithme Growatt""" if len(data) < 8: return None # Masque de déchiffrement mask = "Growatt" hex_mask = [ord(x) for x in mask] nmask = len(hex_mask) # Header non chiffré (8 premiers octets) unscrambled = list(data[0:8]) # Déchiffrement XOR du reste for i, j in zip(range(len(data) - 8), cycle(range(nmask))): unscrambled.append(data[i + 8] ^ hex_mask[j]) return bytes(unscrambled) class SPFParser: """Parser pour les onduleurs SPF (layout T06NNNNSPF)""" @staticmethod def parse(decrypted_data): """Parse les données déchiffrées selon le layout SPF""" try: hex_string = decrypted_data.hex() data = {} # Helper functions def get_string(start, length): """Extrait une chaîne ASCII""" hex_part = hex_string[start:start + length * 2] if hex_part: return codecs.decode(hex_part, "hex").decode('ascii', errors='ignore').rstrip('\x00') return "" def get_uint16(start): """Extrait un uint16 (2 octets)""" if start + 4 <= len(hex_string): return int(hex_string[start:start + 4], 16) return 0 def get_uint32(start): """Extrait un uint32 (4 octets)""" if start + 8 <= len(hex_string): return int(hex_string[start:start + 8], 16) return 0 # Extraction des données selon le layout T06NNNNSPF # Positions basées sur l'analyse des logs Grott # Serials data['datalogserial'] = get_string(16, 10) # Position 16, 10 bytes data['pvserial'] = get_string(76, 10) # Position 76, 10 bytes # Date/Time (position 136) try: year = int(hex_string[136:138], 16) month = int(hex_string[138:140], 16) day = int(hex_string[140:142], 16) hour = int(hex_string[142:144], 16) minute = int(hex_string[144:146], 16) second = int(hex_string[146:148], 16) data['datetime'] = f"20{year:02d}-{month:02d}-{day:02d}T{hour:02d}:{minute:02d}:{second:02d}" except: data['datetime'] = datetime.now().isoformat() # Status et données principales data['pvstatus'] = get_uint16(148) # Tensions PV (position 152+, divisé par 10) data['vpv1'] = get_uint16(152) / 10.0 data['vpv2'] = get_uint16(156) / 10.0 # Puissances PV (position 160+, divisé par 10) data['ppv1'] = get_uint16(160) / 10.0 data['ppv2'] = get_uint16(164) / 10.0 # Courants Buck (position 168+, divisé par 10) data['buck1curr'] = get_uint16(168) / 10.0 data['buck2curr'] = get_uint16(172) / 10.0 # Puissance de sortie (position 176+, divisé par 10) data['op_watt'] = get_uint16(176) / 10.0 data['pvpowerout'] = data['op_watt'] # VA de sortie (position 180+) data['op_va'] = get_uint32(180) / 10.0 # Puissances de charge AC (position 188+) data['acchr_watt'] = get_uint16(188) / 10.0 data['acchr_VA'] = get_uint16(192) / 10.0 # Batterie (position 196+) data['bat_Volt'] = get_uint16(196) / 100.0 data['batterySoc'] = get_uint16(200) # Bus voltage (position 204+, divisé par 10) data['bus_volt'] = get_uint16(204) / 10.0 # Grid (position 208+, divisé par 10) data['grid_volt'] = get_uint16(208) / 10.0 data['line_freq'] = get_uint16(212) / 100.0 # Output (position 216+, divisé par 10) data['outputvolt'] = get_uint16(216) / 10.0 data['outputfreq'] = get_uint16(220) / 100.0 # Températures (position 224+, divisé par 10) data['invtemp'] = get_uint16(224) / 10.0 data['dcdctemp'] = get_uint16(228) / 10.0 # Load (position 232+, divisé par 10) data['loadpercent'] = get_uint16(232) / 10.0 # Températures NTC (position 236+, divisé par 10) data['buck1_ntc'] = get_uint16(236) / 10.0 data['buck2_ntc'] = get_uint16(240) / 10.0 # Courants (position 244+, divisé par 10) data['OP_Curr'] = get_uint16(244) / 10.0 data['Inv_Curr'] = get_uint16(248) / 10.0 # Puissances AC (position 252+) data['AC_InWatt'] = get_uint16(252) / 10.0 data['AC_InVA'] = get_uint16(256) / 10.0 # Bits de défaut (position 260+) data['faultBit'] = get_uint16(260) data['warningBit'] = get_uint16(264) # Énergies (position 280+, divisé par 10) data['epvtoday'] = get_uint32(280) / 10.0 data['epvtotal'] = get_uint32(288) / 10.0 # Énergies AC charge (position 296+, divisé par 10) data['eacCharToday'] = get_uint32(296) / 10.0 data['eacCharTotal'] = get_uint32(304) / 10.0 # Énergies batterie décharge (position 312+, divisé par 10) data['ebatDischarToday'] = get_uint32(312) / 10.0 data['ebatDischarTotal'] = get_uint32(320) / 10.0 # Énergies AC décharge (position 328+, divisé par 10) data['eacDischarToday'] = get_uint32(328) / 10.0 data['eacDischarTotal'] = get_uint32(336) / 10.0 # Courant AC charge (position 344+, divisé par 10) data['ACCharCurr'] = get_uint16(344) / 10.0 # Puissances décharge (position 348+, divisé par 10) data['ACDischarWatt'] = get_uint16(348) / 10.0 data['ACDischarVA'] = get_uint16(352) / 10.0 # Puissances batterie (position 356+, divisé par 10) data['BatDischarWatt'] = get_uint16(356) / 10.0 data['BatDischarVA'] = get_uint16(360) / 10.0 data['BatWatt'] = get_uint16(364) / 10.0 return data except Exception as e: logger.error(f"Erreur parsing SPF: {e}") return None class GrowattProxy: """Proxy pour intercepter les communications ESP -> Growatt""" def __init__(self): self.running = False self.server_socket = None def start(self): """Démarre le serveur proxy""" try: self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server_socket.bind(('0.0.0.0', CONFIG['proxy_port'])) self.server_socket.listen(10) self.running = True logger.info(f"✅ Proxy démarré sur le port {CONFIG['proxy_port']}") while self.running: try: client_socket, address = self.server_socket.accept() logger.info(f"📡 Nouvelle connexion ESP: {address}") # Traiter dans un thread séparé thread = threading.Thread( target=self.handle_client, args=(client_socket, address) ) thread.daemon = True thread.start() except Exception as e: if self.running: logger.error(f"Erreur accept: {e}") except Exception as e: logger.error(f"Erreur démarrage proxy: {e}") def handle_client(self, client_socket, address): """Gère une connexion client (ESP)""" growatt_socket = None try: # Se connecter au serveur Growatt si nécessaire if CONFIG['forward_to_growatt']: try: growatt_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) growatt_socket.settimeout(10) growatt_socket.connect((CONFIG['growatt_server'], CONFIG['growatt_port'])) logger.info(f"✅ Connecté au serveur Growatt") except Exception as e: logger.warning(f"⚠️ Impossible de se connecter à Growatt: {e}") growatt_socket = None # Recevoir et traiter les données while True: data = client_socket.recv(4096) if not data: break # Traiter les données self.process_data(data, address) # Transférer vers Growatt si connecté if growatt_socket: try: growatt_socket.send(data) # Recevoir et transférer la réponse response = growatt_socket.recv(4096) if response: client_socket.send(response) except: pass else: # Envoyer un ACK simple si pas de connexion Growatt if len(data) > 8: # Construire un ACK basique ack = data[:8] + bytes([0x00, 0x00]) client_socket.send(ack) except Exception as e: logger.error(f"Erreur handling client {address}: {e}") finally: client_socket.close() if growatt_socket: growatt_socket.close() logger.info(f"🔌 Connexion fermée: {address}") def process_data(self, data, address): """Traite les données reçues""" try: # Ignorer les paquets trop courts if len(data) < 100: logger.debug(f"Paquet trop court ignoré: {len(data)} octets") return # Déchiffrer decrypted = GrowattDecryptor.decrypt(data) if not decrypted: return # Identifier le type de message header = decrypted[:8].hex() msg_type = header[14:16] logger.info(f"📦 Message type: {msg_type}, taille: {len(data)} octets") # Parser selon le type if msg_type == "04": # Données SPF parsed = SPFParser.parse(decrypted) if parsed: device_id = parsed.get('pvserial', 'unknown') # Mettre à jour les données devices_data[device_id] = { 'last_update': datetime.now().isoformat(), 'data': parsed, 'address': address[0] } # Statistiques statistics['total_messages'] += 1 statistics['last_update'] = datetime.now().isoformat() statistics['devices_seen'].add(device_id) # Garder les derniers messages raw_messages.append({ 'timestamp': datetime.now().isoformat(), 'device': device_id, 'type': msg_type, 'size': len(data), 'data': parsed }) if len(raw_messages) > 100: raw_messages.pop(0) logger.info(f"✅ Données mises à jour pour {device_id}") logger.info(f" PV: {parsed.get('ppv1', 0):.1f}W + {parsed.get('ppv2', 0):.1f}W") logger.info(f" Sortie: {parsed.get('pvpowerout', 0):.1f}W") logger.info(f" Batterie: {parsed.get('batterySoc', 0)}%") except Exception as e: logger.error(f"Erreur traitement données: {e}") # Instance du proxy proxy = GrowattProxy() # Routes Flask @app.route('/') def index(): """Page d'accueil avec TOUTES les données""" return render_template_string(""" 🌱 Growatt Interceptor - Vue Complète

🌱 Growatt Interceptor - Vue Complète

Affichage de TOUTES les données pour debug

📊 Statistiques Globales

Messages reçus: -
Appareils détectés: -
Dernière mise à jour: -
🔍 API JSON 📝 Messages bruts 🐛 Mode Debug

🔌 Onduleurs - Données Complètes

Chargement...

""") @app.route('/api/devices') def api_devices(): """API JSON pour les données des appareils""" return jsonify({ 'devices': devices_data, 'statistics': { 'total_messages': statistics['total_messages'], 'devices_count': len(statistics['devices_seen']), 'last_update': statistics['last_update'] } }) @app.route('/api/raw') def api_raw(): """API pour les messages bruts""" return jsonify({ 'messages': raw_messages[-50:], # Derniers 50 messages 'count': len(raw_messages) }) def main(): """Point d'entrée principal""" print("=" * 60) print("🌱 GROWATT INTERCEPTOR - Remplaçant de Grott") print("=" * 60) print(f"📡 Port proxy: {CONFIG['proxy_port']}") print(f"🌐 Port web: {CONFIG['web_port']}") print(f"🔄 Transfert vers Growatt: {CONFIG['forward_to_growatt']}") print("=" * 60) print("\n⚠️ IMPORTANT: Pour utiliser cet intercepteur:") print("1. Arrêtez Grott (docker stop grott)") print("2. Lancez ce script") print("3. Les ESP se connecteront automatiquement") print("\n📊 Interface web: http://localhost:8080") print("=" * 60) # Démarrer le proxy dans un thread proxy_thread = threading.Thread(target=proxy.start) proxy_thread.daemon = True proxy_thread.start() # Démarrer Flask app.run(host='0.0.0.0', port=CONFIG['web_port'], debug=False) if __name__ == '__main__': main()