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": 2000, "lcl": -2000, "target": 0} } # Carica i limiti dalla DB920 (DBD168 = ucl_tol, DBD172 = lcl_tol) data_920 = self.read_area("db920") if data_920[0]: try: tol_u = get_dint(data_920[0], 168) tol_l = get_dint(data_920[0], 172) # fisso x tolleranza rettilinea... if abs(tol_u) < 10000 and abs(tol_l) < 10000: results["limits"]["ucl"] = + abs(tol_u) results["limits"]["lcl"] = - abs(tol_u) else: results["limits"]["ucl"] = 2500 results["limits"]["lcl"] = -2500 except: pass # Processa le memorie configurate 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 and "misure" in results["db900"]: results["measurements"] = results["db900"]["misure"] # Salvataggio SOLO al passaggio da Acquisition a Finished if self.last_data.get("acq_status") == "Acquisition": self.save_measurements(results) else: results["acq_status"] = "waiting" # escludo 900 che รจ solo una volta ogni richiesta save.. for area_key, res in results.items(): if isinstance(res, dict) and "duration" in res and area_key != "db900": results["latencies"][area_key] = res["duration"] return results def save_measurements(self, results): try: data_920 = self.client.db_read(920, 0, 186) 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" # Formato richiesto: yyyyMMdd-HHmm date_str = datetime.now().strftime("%Y%m%d-%H%M") 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.25) 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/') 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()