Samuele Locatelli
122 lines
4.6 KiB
Python
122 lines
4.6 KiB
Python
import snap7
|
|
import struct
|
|
import time
|
|
|
|
# Configurazione PLC
|
|
PLC_IP = '192.168.124.43' # Sostituisci con l'IP del tuo PLC
|
|
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 come specificato)
|
|
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 (DBD4 inizia dal byte 4)
|
|
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]
|
|
|
|
# # Saltiamo i byte intermedi se necessario
|
|
# params['durata_live'] = struct.unpack('<f', data[56:60])[0]
|
|
# params['y_live'] = struct.unpack('<f', data[60:64])[0]
|
|
# --- SEZIONI LIVE ---
|
|
raw_x = data[56:60]
|
|
raw_y = data[60:64]
|
|
params['x_live'] = struct.unpack('<f', raw_x)[0]
|
|
params['y_live'] = struct.unpack('<f', raw_y)[0]
|
|
|
|
# DEBUG: Stampa HEX per identificare lo sfasamento
|
|
# Se i valori sono sbagliati, guarda questi hex.
|
|
# Esempio: se X_Live è 1.0, dovresti vedere qualcosa tipo '3F800000'
|
|
print(f"DEBUG X_Live (56-59): {raw_x.hex().upper()}")
|
|
print(f"DEBUG Y_Live (60-63): {raw_y.hex().upper()}")
|
|
|
|
# Parsing Stringhe (DBD64, DBD96, DBD128, DBD160)
|
|
def parse_s7_string(raw_data):
|
|
if len(raw_data) < 2: return ""
|
|
length = struct.unpack('<H', raw_data[:2])[0]
|
|
return raw_data[2:2+length].decode('ascii', errors='replace').strip('\x00')
|
|
|
|
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):
|
|
# Leggiamo 4000 REAL. Ogni REAL è 4 byte -> 16000 byte totali
|
|
try:
|
|
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()
|
|
|
|
if __name__ == "__main__":
|
|
plc = SiemensClient(PLC_IP, RACK, SLOT)
|
|
if plc.connect():
|
|
try:
|
|
while True:
|
|
db100_data = plc.read_db100()
|
|
|
|
print("-" * 30)
|
|
print(f"STATO: Acq={db100_data['status']['acquisizione']} | Pronti={db100_data['status']['pronti']}")
|
|
print(f"STRINGHE: {db100_data['strings']}")
|
|
print(f"PARAMETRI: X_time={db100_data['params']['x_live']}, Y_Live={db100_data['params']['y_live']}")
|
|
#print(f"PARAMETRI: X_time={db100_data['params']['x_live']:.2f}, Y_Live={db100_data['params']['y_live']:.2f}")
|
|
|
|
# Esempio lettura DB101 (solo primi 5 valori per non intasare la console)
|
|
db101_values = plc.read_db101()
|
|
if db101_values:
|
|
print(f"DB101 (primi 5): {db101_values[:5]}")
|
|
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
print("Interruzione utente.")
|
|
finally:
|
|
plc.disconnect()
|