Samuele Locatelli
277 lines
11 KiB
Python
277 lines
11 KiB
Python
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()
|