mht-siemens/solution/client/plc_client.py

import snap7
import struct
import time
import os

# Configurazione PLC
PLC_IP = '192.168.124.43'  # L'IP che hai confermato
RACK = 0
SLOT = 2

# Codici Colore ANSI
GREEN = "\033[92m"
RED = "\033[91m"
RESET = "\033[0m"

class SiemensClient:
    def __init__(self, ip, rack, slot):
        self.client = snap7.client.Client()
        self.rack = rack
        self.slot = slot
        self.ip = ip

    def connect(self):
        try:
            self.client.connect(self.ip, self.rack, self.slot)
            print(f"Connesso a {self.ip}")
            return True
        except Exception as e:
            print(f"Errore di connessione: {e}")
            return False

    def read_db100(self):
        # Leggiamo l'intera DB100 (192 byte)
        data = self.client.db_read(100, 0, 192)
        
        # Parsing Status Bits (DBD0)
        status_bytes = data[0:4]
        # Bit 0: Acquisizione (1=On, 0=Off)
        acquisizione = (status_bytes[0] & 0x01) != 0
        # Bit 8 (primo bit del secondo byte)
        # Nota: Se il PLC mappa il bit 9 come il primo bit del byte 1, 
        # lo shift è 0x80 (10000000 in binario)
        pronti = (status_bytes[1] & 0x80) != 0

        # Parsing REAL (32-bit Float) in BIG-ENDIAN (>)
        params = {}
        params['x_iniziale'] = struct.unpack('>f', data[4:8])[0]
        params['x_finale'] = struct.unpack('>f', data[8:12])[0]
        params['limite_sup'] = struct.unpack('>f', data[12:16])[0]
        params['limite_inf'] = struct.unpack('>f', data[16:20])[0]
        params['tipo_lettura'] = struct.unpack('>f', data[20:24])[0]
        params['conteggio'] = struct.unpack('>f', data[24:28])[0]
        params['param_arco'] = struct.unpack('>f', data[28:32])[0]
        params['param_extra1'] = struct.unpack('>f', data[32:36])[0]
        params['param_extra2'] = struct.unpack('>f', data[36:40])[0]
        params['param_extra3'] = struct.unpack('>f', data[40:44])[0]
        
        # X_Live: DBD56 (Byte 56-59), Y_Live: DBD60 (Byte 60-63)
        params['x_live'] = struct.unpack('>f', data[56:60])[0]
        params['y_live'] = struct.unpack('>f', data[60:64])[0]

        # Parsing Stringhe con pulizia profonda
        def parse_s7_string(raw_data):
            if len(raw_data) < 2: return ""
            length = struct.unpack('<H', raw_data[:2])[0]
            content = raw_data[2:2+length]
            # Decodifica, sostituisce caratteri nulli con spazi e pulisce gli spazi bianchi
            return content.decode('ascii', errors='replace').replace('\x00', ' ').strip()

        strings = {
            "Comm": parse_s7_string(data[64:96]),
            "Art": parse_s7_string(data[96:128]),
            "IdCont": parse_s7_string(data[128:160]),
            "Altro": parse_s7_string(data[160:192])
        }

        return {
            "status": {"acquisizione": acquisizione, "pronti": pronti},
            "params": params,
            "strings": strings
        }

    def read_db101(self):
        try:
            # Leggiamo 40 REAL
            data = self.client.db_read(101, 0, 160)
            values = struct.unpack('>40f', data)
            return values
        except Exception as e:
            print(f"Errore lettura DB101: {e}")
            return None

    def disconnect(self):
        self.client.disconnect()

def clear_screen():
    os.system('cls' if os.name == 'nt' else 'clear')

if __name__ == "__main__":
    plc = SiemensClient(PLC_IP, RACK, SLOT)
    if plc.connect():
        try:
            while True:
                clear_screen()
                db100_data = plc.read_db100()
                
                print("=" * 60)
                print(f"  PLC MONITOR - {PLC_IP}  ")
                print("=" * 60)
                print(f" STATO: Acquisizione={db100_data['status']['acquisizione']} | Pronti={db100_data['status']['pronti']}")
                print("-" * 60)
                print(f" VALORI LIVE:")
                print(f"  X_Live (DBD56): {db100_data['params']['x_live']:>10.4f}")
                print(f"  Y_Live (DBD60): {db100_data['params']['y_live']:>10.4f}")
                print("-" * 60)
                print(f" DATI STRINGHE:")
                print(f"  Commessa:   {db100_data['strings']['Comm']}")
                print(f"  Articolo:   {db100_data['strings']['Art']}")
                print(f"  IdCont:     {db100_data['strings']['IdCont']}")
                print(f"  Altro:      {db100_data['strings']['Altro']}")
                print("-" * 60)
                
                # Gestione DB101
                db101_raw = plc.read_db101()
                if db101_raw:
                    # Prendi i primi 20 valori
                    data_subset = db101_raw[:20]
                    
                    print(f" VISUALIZZAZIONE DB101 (Primi 20):")
                    if data_subset:
                        # Trova il valore massimo assoluto per scalare correttamente le barre
                        max_abs = max(abs(v) for v in data_subset)
                        if max_abs == 0: max_abs = 1.0
                        
                        for i, val in enumerate(data_subset):
                            # Calcola larghezza barra (max 30 caratteri)
                            bar_len = int((abs(val) / max_abs) * 30)
                            bar = "█" * bar_len + "-" * (30 - bar_len)
                            
                            # Determina colore
                            color = RESET
                            if val > 0: color = GREEN
                            elif val < 0: color = RED
                            
                            # Stampa riga con colore
                            # [01] indica l'indice, il valore è colorato
                            print(f"  [{i+1:02d}] {color}{bar} {val:>8.2f}{RESET}")
                
                print("-" * 60)
                time.sleep(0.5)
        except KeyboardInterrupt:
            print("\nInterruzione utente.")
        finally:
            plc.disconnect()