import time import random import snap7 import ctypes # Modificato l'import per supportare le ultime versioni di python-snap7 from snap7.type import SrvArea from snap7.util import set_bool, set_real, set_string, set_int, set_dint # --- CONFIGURAZIONE SERVER S7 --- server = snap7.server.Server() # Inizializzazione delle aree di memoria corrette per Snap7 (vettori ctypes) # DB100: 192 byte totali db100_buffer = (ctypes.c_ubyte * 192)() # DB101: 800 byte totali db101_buffer = (ctypes.c_ubyte * 800)() # DB900: 16384 byte totali (16KB) db900_buffer = (ctypes.c_ubyte * 16384)() # DB901: 38 byte totali db901_buffer = (ctypes.c_ubyte * 38)() # DB920: 186 byte totali db920_buffer = (ctypes.c_ubyte * 186)() # Registrazione delle aree di memoria nel server (Corretto usando SrvArea.DB) server.register_area(SrvArea.DB, 100, db100_buffer) server.register_area(SrvArea.DB, 101, db101_buffer) server.register_area(SrvArea.DB, 900, db900_buffer) server.register_area(SrvArea.DB, 901, db901_buffer) server.register_area(SrvArea.DB, 920, db920_buffer) # Avvio del server sulla porta standard S7 (102) # NOTA: Su Linux/macOS sono necessari i permessi di root (sudo) per la porta 102 server.start() print("Server S7 simulato avviato sulla porta 102.") print("In attesa di connessioni da client (PLC, SCADA, HMI)...") # --- POPOLAMENTO CONFIGURAZIONE STATICA (DB100) --- # Popolamento stringhe (Formato S7: max 30 caratteri, occupa 32 byte ciascuna) set_string(db100_buffer, 64, "COMMESSA_2026", max_size=30) set_string(db100_buffer, 96, "ARTICOLO_XYZ", max_size=30) set_string(db100_buffer, 128, "OK_CONFRONTO", max_size=30) set_string(db100_buffer, 160, "SIMULAZIONE_ATTIVA", max_size=30) # Popolamento parametri REAL fissi (Offset corretti a passi di 4 byte) set_real(db100_buffer, 4, 0.0) # Parametro x iniziale set_real(db100_buffer, 8, 100.0) # Parametro x finale set_real(db100_buffer, 12, 1.8) # Limite superiore set_real(db100_buffer, 16, -1.8) # Limite inferiore # Parametri liberi fissati a 0 (da DBD20 a DBD40 inclusi) for offset in range(20, 44, 4): set_real(db100_buffer, offset, 0.0) # --- POPOLAMENTO CONFIGURAZIONE STATICA (DB920) --- set_string(db920_buffer, 0, "CLIENTE_TEST", max_size=30) set_string(db920_buffer, 32, "ARTICOLO_TEST", max_size=30) set_string(db920_buffer, 64, "MATRICOLA_TEST", max_size=30) set_string(db920_buffer, 96, "OPERATORE_TEST", max_size=30) set_string(db920_buffer, 128, "SUPERVISORE_TEST", max_size=30) set_dint(db920_buffer, 160, 160) # Codice Produzione PEP set_dint(db920_buffer, 164, 1234) # Numero di Certificato set_dint(db920_buffer, 168, 2000) # Massima Tolleranza Rettilinietà (micron) set_dint(db920_buffer, 172, 2000) # Massima Tolleranza Rotondità (micron) set_dint(db920_buffer, 176, 0) # Errore Riscontrato di Rettilinietà (micron, max) set_dint(db920_buffer, 180, 0) # Errore Riscontrato di Rotondità (micron, max) set_bool(db920_buffer, 184, 0, False) # Bit 0: Attivazione Pagina Certificato (1=ON) set_bool(db920_buffer, 184, 1, False) # Bit 1: Richiesta salvataggio certificato (1=ON) set_bool(db920_buffer, 185, 0, False) # Bit 0: Conferma Salvataggio Certificato Richiesta (1=ON) # --- PARAMETRI DI SIMULAZIONE --- intervallo = 0.10 # Generazione dati ogni 100ms durata_simulazione = 10.0 # Durata del ciclo di 10 secondi campioni_totali = int(durata_simulazione / intervallo) # 100 campioni # --- LOOP PRINCIPALE DEL SERVER --- try: while True: storico_campioni = [] print(f"\n--- Inizio Nuova Simulazione ({durata_simulazione} sec) ---") for step in range(campioni_totali): tempo_corrente = step * intervallo # 1. Gestione Bit di Stato (DB100.DBD0) # Reset dei primi 4 byte per pulire stati precedenti db100_buffer[0:4] = b'\x00\x00\x00\x00' set_bool(db100_buffer, 0, 0, True) # 1° bit del 1° byte: Acquisizione attiva (1) set_bool(db100_buffer, 1, 0, False) # 9° bit totale (1° bit del 2° byte): Fine simulazione (0) # DB901 bits (DBD0) - Reset e impostazione stato for b in range(4): db901_buffer[b] = 0 set_bool(db901_buffer, 0, 0, True) # Bit 0: Stato Ready set_bool(db901_buffer, 0, 1, True) # Bit 1: Acquisizione (1=ON) # 2. Aggiornamento durata processo (DB100.DBD56) set_real(db100_buffer, 56, float(tempo_corrente)) # 3. Generazione valore REAL casuale tra -2.0 e 2.0 (DB100.DBD60) valore_simulato = random.uniform(-2.0, 2.0) set_real(db100_buffer, 60, valore_simulato) # 4. Aggiornamento dati live in DB901 # DBD22: Valore Attuale Misura (real time) in Micron (DINT) set_dint(db901_buffer, 22, int(valore_simulato * 1000)) # DBD26: Posizione longitudinale della misura Asse Z (real time) (REAL) # Se deve essere in micron, moltiplichiamo per 1000.0 set_real(db901_buffer, 26, tempo_corrente * 1000.0) # Memorizza il campione per il blocco finale di DB101 storico_campioni.append(valore_simulato) # log ogni 10 valori... if(step%10 == 0): print(f"[{tempo_corrente:3.1f}s] Sim | Valore DB101: {valore_simulato:+.4f} | DB901: {int(valore_simulato*1000)}") time.sleep(intervallo) # --- FINE CICLO 10 SECONDI --- print("Fine ciclo. Aggiornamento stati e popolamento DB101/DB900...") # 1. Aggiornamento Bit di Stato Finale su DB100 - Reset sicuro db100_buffer[0:4] = b'\x00\x00\x00\x00' set_bool(db100_buffer, 0, 0, False) # Stato acquisizione -> 0 (Attesa) set_bool(db100_buffer, 1, 0, True) # 9° bit totale -> 1 (Simulazione terminata) set_real(db100_buffer, 56, durata_simulazione) # Durata finale fissata a 10.0 # 2. Aggiornamento stati finali su DB901 for b in range(4): db901_buffer[b] = 0 set_bool(db901_buffer, 0, 0, False) # Ready -> 0 set_bool(db901_buffer, 0, 1, False) # Acquisizione -> 0 set_bool(db901_buffer, 0, 2, True) # Fine Acquisizione -> 1 set_bool(db901_buffer, 0, 3, True) # Dati Pronti su DB900 -> 1 # 3. Popolamento DB101 con i 20 campioni ctypes.memset(db101_buffer, 0, ctypes.sizeof(db101_buffer)) # Scrive i 20 valori reali simulati (Offset da DBD0 a DBD76) for i, val in enumerate(storico_campioni): offset_db101 = i * 4 set_real(db101_buffer, offset_db101, val) # 4. Popolamento DB900 con i dati di DB101 in formato DINT (x1000) set_dint(db900_buffer, 0, len(storico_campioni)) # DBD4-16387: Array di misure DINT for i, val in enumerate(storico_campioni): # Ogni DINT occupa 4 byte offset_db900 = 4 + (i * 4) if offset_db900 < 16384: set_dint(db900_buffer, offset_db900, int(val * 1000)) print(f"DB101 aggiornata ({len(storico_campioni)} campioni). DB900 aggiornata con {len(storico_campioni)} misure.") print("Pausa prima del prossimo ciclo...") # Pausa statica di fine ciclo prima di ricominciare time.sleep(5.0) except KeyboardInterrupt: print("\nSpegnimento del Server S7...") finally: server.stop() print("Server S7 arrestato.")