Samuele Locatelli
335 lines
12 KiB
Python
335 lines
12 KiB
Python
import snap7
|
|
import struct
|
|
import time
|
|
import os
|
|
import json
|
|
import logging
|
|
import threading
|
|
from datetime import datetime
|
|
from flask import Flask, render_template, jsonify
|
|
from snap7.util import get_int, get_dint, get_real, get_string
|
|
|
|
# Configurazione Logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
|
|
logger = logging.getLogger("PLC_Client_Web")
|
|
|
|
class PLCManager:
|
|
def __init__(self, config_path="config.json"):
|
|
self.config_path = config_path
|
|
self.config = {}
|
|
self.load_config()
|
|
self.client = snap7.client.Client()
|
|
self.connected = False
|
|
self.last_data = {}
|
|
self.lock = threading.Lock()
|
|
|
|
self.ip = self.config["PLC"]["IP"]
|
|
self.rack = self.config["PLC"]["RACK"]
|
|
self.slot = self.config["PLC"]["SLOT"]
|
|
|
|
def load_config(self):
|
|
try:
|
|
with open(self.config_path, 'r') as f:
|
|
self.config = json.load(f)
|
|
logger.info(f"Configurazione caricata da {self.config_path}")
|
|
except Exception as e:
|
|
logger.error(f"Errore nel caricamento del config.json: {e}")
|
|
raise
|
|
|
|
def connect(self):
|
|
try:
|
|
self.client.connect(self.ip, self.rack, self.slot)
|
|
self.connected = True
|
|
logger.info(f"Connesso a {self.ip} (Rack: {self.rack}, Slot: {self.slot})")
|
|
return True
|
|
except Exception as e:
|
|
self.connected = False
|
|
logger.error(f"Errore di connessione a {self.ip}: {e}")
|
|
return False
|
|
|
|
def disconnect(self):
|
|
try:
|
|
self.client.disconnect()
|
|
except:
|
|
pass
|
|
self.connected = False
|
|
logger.info("Client disconnesso.")
|
|
|
|
def read_area(self, area_key):
|
|
if not self.connected:
|
|
return None, 0
|
|
|
|
area_cfg = self.config["MEMORIES"].get(area_key)
|
|
if not area_cfg:
|
|
return None, 0
|
|
|
|
db_id = area_cfg["id"]
|
|
size = area_cfg["size"]
|
|
|
|
try:
|
|
start_time = time.perf_counter()
|
|
data = self.client.db_read(db_id, 0, size)
|
|
duration = (time.perf_counter() - start_time) * 1000
|
|
return data, duration
|
|
except Exception as e:
|
|
logger.warning(f"Errore lettura DB{db_id}: {e}")
|
|
self.connected = False
|
|
return None, 0
|
|
|
|
def fetch_all_data(self):
|
|
results = {
|
|
"connected": self.connected,
|
|
"plc_ip": self.ip,
|
|
"plc_model": "Siemens S7-300",
|
|
"polling_period": "5Hz",
|
|
"acq_status": "waiting",
|
|
"latencies": {},
|
|
"measurements": [],
|
|
"limits": {"ucl": 212, "lcl": 208, "target": 0}
|
|
}
|
|
|
|
# Carica i limiti dalla DB920 (DBD168 = ucl_tol, DBD172 = lcl_tol)
|
|
data_920 = self.read_area("db920")
|
|
if data_920[0]:
|
|
try:
|
|
# Leggiamo i valori DINT. Assumiamo che siano tolleranze assolute.
|
|
tol_u = get_dint(data_920[0], 168)
|
|
tol_l = get_dint(data_920[0], 172)
|
|
# Se i valori sono troppo grandi, potrebbe esserci un errore di offset o tipo
|
|
if abs(tol_u) < 10000 and abs(tol_l) < 10000:
|
|
results["limits"]["ucl"] = 0 + abs(tol_u) # Target 0 + tol
|
|
results["limits"]["lcl"] = 0 - abs(tol_l)
|
|
# Se il target è 210 di base, aggiustiamo:
|
|
# results["limits"]["ucl"] = 210 + abs(tol_u)
|
|
# results["limits"]["lcl"] = 210 - abs(tol_l)
|
|
else:
|
|
# Fallback se i dati sono sballati
|
|
results["limits"]["ucl"] = 212
|
|
results["limits"]["lcl"] = 208
|
|
except:
|
|
pass
|
|
|
|
for area_key, area_cfg in self.config["MEMORIES"].items():
|
|
data, dur = self.read_area(area_key)
|
|
if data is None:
|
|
continue
|
|
|
|
parsing_type = area_cfg.get("type", "complex")
|
|
|
|
if parsing_type == "complex":
|
|
params = {}
|
|
for p in area_cfg.get("parameters", []):
|
|
p_type = p.get("type")
|
|
if p_type == "real":
|
|
params[p["name"]] = get_real(data, p["offset"])
|
|
elif p_type == "dint":
|
|
params[p["name"]] = get_dint(data, p["offset"])
|
|
elif p_type == "int":
|
|
params[p["name"]] = get_int(data, p["offset"])
|
|
elif p_type == "bool":
|
|
params[p["name"]] = (data[p["offset"]] & (1 << p.get("bit", 0))) != 0
|
|
|
|
strings = {}
|
|
for s in area_cfg.get("strings", []):
|
|
strings[s["name"]] = get_string(data, s["offset"])
|
|
|
|
results[area_key] = {
|
|
"params": params,
|
|
"strings": strings,
|
|
"duration": dur
|
|
}
|
|
|
|
elif parsing_type == "dint_array":
|
|
if area_key == "db900":
|
|
num_misure = get_dint(data, 0)
|
|
arr = []
|
|
for i in range(num_misure):
|
|
curr_off = 4 + (i * 4)
|
|
if curr_off + 4 <= area_cfg["size"]:
|
|
arr.append(get_dint(data, curr_off))
|
|
else:
|
|
arr.append(0)
|
|
results[area_key] = {
|
|
"misure": arr,
|
|
"duration": dur
|
|
}
|
|
else:
|
|
count = area_cfg.get("count", 21)
|
|
offset = area_cfg.get("offset", 4)
|
|
arr = []
|
|
for i in range(count):
|
|
curr_off = offset + (i * 4)
|
|
if curr_off + 4 <= area_cfg["size"]:
|
|
arr.append(get_dint(data, curr_off))
|
|
else:
|
|
arr.append(0)
|
|
results[area_key] = {
|
|
"misure": arr,
|
|
"duration": dur
|
|
}
|
|
|
|
elif parsing_type == "array_real":
|
|
count = area_cfg.get("count", 40)
|
|
try:
|
|
values = struct.unpack(f">{count}f", data)
|
|
results[area_key] = {
|
|
"values": values,
|
|
"duration": dur
|
|
}
|
|
except:
|
|
results[area_key] = {"error": "Parsing array_real failed"}
|
|
results[area_key]["duration"] = dur
|
|
|
|
else:
|
|
params = {}
|
|
for p in area_cfg.get("parameters", []):
|
|
if p["type"] == "real":
|
|
params[p["name"]] = get_real(data, p["offset"])
|
|
elif p["type"] == "dint":
|
|
params[p["name"]] = get_dint(data, p["offset"])
|
|
results[area_key] = {
|
|
"params": params,
|
|
"duration": dur
|
|
}
|
|
|
|
if "db901" in results:
|
|
db901_res = results["db901"]
|
|
if "params" in db901_res:
|
|
acq = db901_res["params"].get("acquisizione", False)
|
|
fine = db901_res["params"].get("fine_acq", False)
|
|
pronti = db901_res["params"].get("pronti", False)
|
|
|
|
if acq and not fine:
|
|
results["acq_status"] = "Acquisition"
|
|
elif fine and pronti:
|
|
results["acq_status"] = "Finished"
|
|
if "db900" in results:
|
|
results["measurements"] = results["db900"].get("misure", [])
|
|
self.save_measurements(results)
|
|
else:
|
|
results["acq_status"] = "waiting"
|
|
|
|
# Correzione x_val e y_val
|
|
# x_val: DB901.DBD26 (REAL) -> dividere per 1000 come indicato dall'utente
|
|
# y_val: DB901.DBD22 (DINT)
|
|
if "x_val" in db901_res["params"]:
|
|
val_x = db901_res["params"]["x_val"]
|
|
db901_res["params"]["x"] = val_x / 1000.0
|
|
if "y_val" in db901_res["params"]:
|
|
db901_res["params"]["y"] = db901_res["params"]["y_val"]
|
|
|
|
for area_key, res in results.items():
|
|
if isinstance(res, dict) and "duration" in res and area_key != "db901":
|
|
results["latencies"][area_key] = res["duration"]
|
|
|
|
return results
|
|
|
|
def save_measurements(self, results):
|
|
try:
|
|
data_920 = self.client.db_read(920, 0, 186)
|
|
# get_string(data, offset, size)
|
|
articolo = get_string(data_920, 32, 30).strip()
|
|
cod_produzione = get_string(data_920, 160, 30).strip()
|
|
num_certificato = get_dint(data_920, 164)
|
|
|
|
if not articolo: articolo = "Sconosciuto"
|
|
if not cod_produzione: cod_produzione = "Sconosciuto"
|
|
|
|
date_str = datetime.now().strftime("%Y-%m-%d")
|
|
path_dir = os.path.join("measurements", articolo, cod_produzione)
|
|
if not os.path.exists(path_dir):
|
|
os.makedirs(path_dir, exist_ok=True)
|
|
|
|
filename = f"{num_certificato}_{date_str}.json"
|
|
path_file = os.path.join(path_dir, filename)
|
|
|
|
save_data = {
|
|
"date": datetime.now().isoformat(),
|
|
"articolo": articolo,
|
|
"cod_produzione": cod_produzione,
|
|
"num_certificato": num_certificato,
|
|
"misure": results["measurements"]
|
|
}
|
|
|
|
with open(path_file, 'w') as f:
|
|
json.dump(save_data, f, indent=4)
|
|
logger.info(f"Misure salvate in: {path_file}")
|
|
except Exception as e:
|
|
logger.error(f"Errore nel salvataggio misure: {e}")
|
|
|
|
def get_data(self):
|
|
with self.lock:
|
|
return self.last_data
|
|
|
|
def get_history(self):
|
|
history = []
|
|
history_path = "measurements"
|
|
if not os.path.exists(history_path):
|
|
return history
|
|
for root, dirs, files in os.walk(history_path):
|
|
for file in files:
|
|
if file.endswith(".json"):
|
|
try:
|
|
parts = root.split(os.sep)
|
|
articolo = parts[-3]
|
|
cod_produzione = parts[-2]
|
|
history.append({
|
|
"file_path": os.path.join(root, file),
|
|
"articolo": articolo,
|
|
"cod_produzione": cod_produzione,
|
|
"filename": file
|
|
})
|
|
except:
|
|
continue
|
|
return history
|
|
|
|
def load_history_file(self, file_path):
|
|
with open(file_path, 'r') as f:
|
|
return json.load(f)
|
|
|
|
def run_loop(self):
|
|
while True:
|
|
if not self.connected:
|
|
if not self.connect():
|
|
time.sleep(self.config["PLC"].get("RECONNECT_DELAY", 5.0))
|
|
continue
|
|
data = self.fetch_all_data()
|
|
if data:
|
|
with self.lock:
|
|
self.last_data = data
|
|
time.sleep(0.5)
|
|
|
|
app = Flask(__name__, static_folder='static', template_folder='templates')
|
|
plc_manager = PLCManager("../client/config.json")
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
@app.route('/data')
|
|
def data():
|
|
return jsonify(plc_manager.get_data())
|
|
|
|
@app.route('/history')
|
|
def history():
|
|
return jsonify(plc_manager.get_history())
|
|
|
|
@app.route('/history/<path:file_path>')
|
|
def load_history(file_path):
|
|
safe_path = os.path.normpath(file_path).lstrip('/')
|
|
if not safe_path.startswith('measurements'):
|
|
return jsonify({"error": "Forbidden"}), 403
|
|
try:
|
|
content = plc_manager.load_history_file(safe_path)
|
|
return jsonify(content)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
def run_app():
|
|
thread = threading.Thread(target=plc_manager.run_loop, daemon=True)
|
|
thread.start()
|
|
app.run(host='0.0.0.0', port=5000, debug=False)
|
|
|
|
if __name__ == "__main__":
|
|
run_app()
|