Samuele Locatelli
143 lines
5.5 KiB
Python
143 lines
5.5 KiB
Python
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
|
|
|
|
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 (40 * 4 byte = 160 byte)
|
|
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():
|
|
# Pulisce la console in base al sistema operativo
|
|
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:
|
|
# Filtra i primi 20 valori > 0
|
|
filtered_data = [v for v in db101_raw][:20]
|
|
# filtered_data = [v for v in db101_raw if v > 0][:20]
|
|
|
|
print(f" VISUALIZZAZIONE DB101 (Primi 20 > 0):")
|
|
if not filtered_data:
|
|
print(" Nessun dato > 0 rilevato.")
|
|
else:
|
|
max_val = max(filtered_data)
|
|
for i, val in enumerate(filtered_data):
|
|
# Calcola lunghezza barra (max 30 caratteri)
|
|
bar_len = int((val / max_val) * 30) if max_val > 0 else 0
|
|
bar = "█" * bar_len + "-" * (30 - bar_len)
|
|
print(f" [{i+1:02d}] {bar} {val:>8.2f}")
|
|
|
|
print("-" * 60)
|
|
time.sleep(0.5)
|
|
except KeyboardInterrupt:
|
|
print("\nInterruzione utente.")
|
|
finally:
|
|
plc.disconnect()
|