import snap7 import struct import time import os from snap7.util import get_int, get_dint, get_real, get_string # Configurazione PLC PLC_IP = '127.0.0.1' # PLC_IP = '192.168.124.43' 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_db900_db901(self): # Lettura DB900 (16KB) start_time = time.perf_counter() #~nel test leggo solo 40 valori + counter iniziale db900_data = self.client.db_read(900, 0, 404) end_time = time.perf_counter() db900_duration = (end_time - start_time) * 1000 # Lettura DB901 (38 byte) start_time_901 = time.perf_counter() db901_data = self.client.db_read(901, 0, 38) end_time_901 = time.perf_counter() db901_duration = (end_time_901 - start_time_901) * 1000 # Parsing DB901 (DINT e Bit) status_901 = db901_data[0] acquisizione_901 = (status_901 & 0x02) != 0 fine_acq_901 = (status_901 & 0x04) != 0 pronti_901 = (status_901 & 0x08) != 0 x_live = get_real(db901_data, 26) y_live = get_dint(db901_data, 22) # Parsing DB900 (DINT) - Primi 21 valori (DBD0-DBD80) # DBD0: Numero misure (DINT) num_misure = get_dint(db900_data,0) # Array misure (da DBD4 in poi) valori_db900 = [] for i in range(21): offset = (i * 4) if offset + 4 <= 164: valori_db900.append(get_dint(db900_data,offset)) else: valori_db900.append(0) return { "db900_data": valori_db900, "x_live": x_live, "y_live": y_live, "db900_num_misure": num_misure, "db901_status": { "acquisizione": acquisizione_901, "pronti": pronti_901, "fine_acq": fine_acq_901 }, "db901_duration": db901_duration, "db900_duration": db900_duration } def read_db100(self): start_time = time.perf_counter() data = self.client.db_read(100, 0, 192) end_time = time.perf_counter() duration = (end_time - start_time) * 1000 # in ms # 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'] = get_real(data,4) params['x_finale'] = get_real(data,8) params['limite_sup'] = get_real(data,12) params['limite_inf'] = get_real(data,16) params['tipo_lettura'] = get_real(data,20) params['conteggio'] = get_real(data,24) params['param_arco'] = get_real(data,28) params['param_extra1'] = get_real(data,32) params['param_extra2'] = get_real(data,36) params['param_extra3'] = get_real(data,40) # # X_Live: DBD56 (Byte 56-59), Y_Live: DBD60 (Byte 60-63) # params['x_live'] = get_real(data,56) # params['y_live'] = get_real(data,60) # Tempo di lettura della DB100 params['db100_time'] = duration strings = { "Comm": get_string(data,64), "Art": get_string(data,96), "IdCont": get_string(data,128), "Altro": get_string(data,160) } return { "status": {"acquisizione": acquisizione, "pronti": pronti}, "params": params, "strings": strings } def read_db101(self): start_time = time.perf_counter() data = self.client.db_read(101, 0, 160) end_time = time.perf_counter() duration = (end_time - start_time) * 1000 # in ms try: values = struct.unpack('>40f', data) return values, duration except Exception as e: print(f"Errore lettura DB101: {e}") return None, duration def read_db920(self): start_time = time.perf_counter() data = self.client.db_read(920, 0, 186) end_time = time.perf_counter() duration = (end_time - start_time) * 1000 # in ms # Parsing Status Bits (DBD0) status_bytes = data[184:185] # Bit 0: generazione (1=On, 0=Off) acquisizione = (status_bytes[0] & 0x01) != 0 # Bit 1: salva salva = (status_bytes[0] & 0x02) != 0 # Parsing REAL (32-bit Float) in BIG-ENDIAN (>) params = {} params['cod_prod'] = get_dint(data,160) params['nr_cert'] = get_dint(data,164) params['max_toll_rett'] = get_dint(data,168) params['max_toll_rot'] = get_dint(data,172) params['errore_rett'] = get_dint(data,176) params['errore_rot'] = get_dint(data,180) # Tempo di lettura della DB100 params['db920_time'] = duration strings = { "Cliente": get_string(data,0), "Art": get_string(data,32), "MatrPezzo": get_string(data,64), "Operatore": get_string(data,96), "Supervisore": get_string(data,128) } return { "params": params, "strings": strings } 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() db101_data, db101_time = plc.read_db101() db900_data = plc.read_db900_db901() db920_data = plc.read_db920() print("=" * 60) print(f" PLC MONITOR - {PLC_IP} ") print("=" * 60) print(f" STATO: Acquisizione={db100_data['status']['acquisizione']} | Pronti={db100_data['status']['pronti']}") print(f" LATENZA: DB100={db100_data['params']['db100_time']:.2f}ms | DB901={db900_data['db901_duration']:.2f}ms") print("-" * 60) print(f" VALORI LIVE:") print(f" Misure (DB900.DBD0) : {db900_data['db900_num_misure']:0.0f}") print(f" X_Live (DB901.DBD26): {db900_data['x_live']:0.2f}") print(f" Y_Live (DB901.DBD22): {db900_data['y_live']:0.0f}") print(f" Toll 1 (DB920.DBD168) : {db920_data['params']['max_toll_rett']:0.0f}") print(f" Toll 2 (DB920.DBD172) : {db920_data['params']['max_toll_rot']:0.0f}") print("-" * 60) print(f" DATI STRINGHE:") print(f" Cliente : {db920_data['strings']['Cliente']}") print(f" Articolo: {db920_data['strings']['Art']}") print(f" MatrPezzo: {db920_data['strings']['MatrPezzo']}") print(f" Operatore: {db920_data['strings']['Operatore']}") print(f" Supervisore: {db920_data['strings']['Supervisore']}") print("-" * 60) # Gestione DB101 con grafico a barre centrato if db101_data: # Prendi i primi 20 valori data_subset = db101_data[:20] print(f" VISUALIZZAZIONE DB101 (Primi {len(data_subset)}):") if data_subset: # Scala fissa basata sui tuoi dati (-2..+2) scale_max = 2.0 for i, val in enumerate(data_subset): # Calcola larghezza barra (max 15 caratteri per lato del centro) bar_len = int((abs(val) / scale_max) * 15) bar_len = min(bar_len, 15) bar = "█" * bar_len if val > 0: # Cresce verso destra: [ |███████████████] left_side = "-" * 15 right_side = bar + "-" * (15 - bar_len) color = GREEN line = f" [{i+1:02d}] {left_side}|{color}{right_side}{RESET} {val:>8.4f}" elif val < 0: # Cresce verso sinistra: [-------█████████████| ] left_side = "-" * (15 - bar_len) + bar right_side = "-" * 15 color = RED line = f" [{i+1:02d}] {color}{left_side}{RESET}|{right_side} {val:>8.4f}" else: # Valore zero: [ | ] line = f" [{i+1:02d}] {'-' * 15}|{'-' * 15} {val:>8.4f}" print(line) print("-" * 60) print(f" DATI DB900 (Primi {len(db900_data['db900_data'])} DINT):") # Mostra i primi 21 valori della DB900 for i, val in enumerate(db900_data['db900_data']): # Il primo valore (index 0) è il numero di misure if i == 0: print(f" [000] Num. Misure: {val}") else: print(f" [{i:03d}] Valore DINT: {val}") print("-" * 60) # Campionamento dinamico if db100_data['status']['acquisizione']: time.sleep(0.5) else: time.sleep(2.0) except KeyboardInterrupt: print("\nInterruzione utente.") finally: plc.disconnect()