Interagire con i sistemi SCADA in Python
Python è spesso il primo linguaggio a cui si pensa quando bisogna leggere dati da un impianto industriale per analizzarli: chi lavora con i dati di processo usa già pandas, Jupyter e le librerie di analisi, e poter interrogare direttamente un PLC dallo stesso ambiente accorcia molto il percorso. Ma Python è anche adatto a servizi di acquisizione continui, soprattutto da quando asyncio permette di gestire più connessioni industriali in un solo processo senza ricorrere ai thread.
In questo articolo costruiremo un servizio di acquisizione asincrono che legge un PLC via Modbus TCP con pymodbus, si abbona alle variabili di un server OPC UA con asyncua, applica una banda morta per ridurre il volume dei dati, valuta allarmi con isteresi e salva lo storico in SQLite. Sono gli stessi mattoni di un piccolo historian, il componente SCADA che registra l'andamento delle grandezze di processo nel tempo.
Cosa significa interagire con uno SCADA
Un sistema SCADA è stratificato. I PLC eseguono la logica di controllo leggendo sensori e comandando attuatori con cicli di scansione dell'ordine dei millisecondi. Sopra di loro un server SCADA raccoglie i dati, li presenta agli operatori tramite le HMI, gestisce allarmi e storico. Un'applicazione esterna come la nostra può collegarsi a due livelli: direttamente ai PLC, tipicamente via Modbus TCP, oppure al server SCADA o a un PLC moderno tramite OPC UA. In entrambi i casi il principio guida è lo stesso: la logica di controllo resta nel PLC, e il software esterno osserva o, al più, modifica parametri entro limiti stabiliti.
Installazione
Serve Python 3.11 o successivo. Le dipendenze si installano in un ambiente virtuale.
python3 -m venv .venv
source .venv/bin/activate
pip install "pymodbus>=3.10" asyncuaLa versione minima di pymodbus non è un dettaglio. La libreria ha cambiato la propria API più volte nel corso della serie 3: il parametro che identifica il dispositivo sul bus, storicamente chiamato slave, oggi si chiama device_id, e il vecchio BinaryPayloadDecoder è stato sostituito dai metodi convert_from_registers e convert_to_registers del client. Molti esempi che si trovano in rete si riferiscono alle versioni precedenti e non funzionano più.
Per le prove, pymodbus include un simulatore di server, mentre per OPC UA si può usare il server di esempio di asyncua stesso, che in poche righe crea un namespace con alcune variabili.
Il modello dei tag
Ogni grandezza da acquisire è un tag. Lo descriviamo con una dataclass immutabile che contiene tutto ciò che serve per leggerla e interpretarla, comprese la banda morta e le soglie di allarme.
# tags.py
from dataclasses import dataclass, field
from enum import Enum
class Table(Enum):
HOLDING = "holding"
INPUT = "input"
@dataclass(frozen=True)
class AlarmLimits:
high: float | None = None
low: float | None = None
# Isteresi: l'allarme rientra solo quando il valore torna oltre la soglia di questa quantità
hysteresis: float = 0.0
@dataclass(frozen=True)
class ModbusTag:
name: str
table: Table
address: int
data_type: str # "int16", "uint16", "int32", "uint32", "float32"
scale: float = 1.0
unit: str = ""
word_order: str = "big"
deadband: float = 0.0
alarms: AlarmLimits = field(default_factory=AlarmLimits)
@dataclass(frozen=True)
class OpcUaTag:
name: str
node_id: str
unit: str = ""
deadband: float = 0.0
alarms: AlarmLimits = field(default_factory=AlarmLimits)
@dataclass(frozen=True)
class Reading:
name: str
value: float | None
quality: str # "good" oppure "bad"
timestamp: float
unit: str = ""Il campo address è sempre l'offset a base zero trasmesso sul filo. Se la documentazione del PLC indica il registro 40101, l'indirizzo da configurare è 100: il 4 iniziale identifica la tabella degli holding register e la numerazione storica parte da uno.
Il client Modbus asincrono
Il lettore Modbus mantiene una connessione persistente, legge ciclicamente i tag e produce oggetti Reading in una coda asyncio. Per i tipi a 32 bit usiamo convert_from_registers, che accetta il parametro word_order: la specifica Modbus non stabilisce in quale dei due registri vada la parte alta, e i costruttori di PLC non sono d'accordo tra loro.
# modbus_reader.py
import asyncio
import logging
import time
from pymodbus.client import AsyncModbusTcpClient
from pymodbus.exceptions import ModbusException
from tags import ModbusTag, Reading, Table
logger = logging.getLogger("modbus")
REGISTER_COUNT = {"int16": 1, "uint16": 1, "int32": 2, "uint32": 2, "float32": 2}
class ModbusReader:
def __init__(self, host: str, port: int, device_id: int, tags: list[ModbusTag], interval: float = 1.0):
self.device_id = device_id
self.tags = tags
self.interval = interval
self.client = AsyncModbusTcpClient(host, port=port, timeout=2, retries=1)
self._lock = asyncio.Lock()
def _decode(self, registers: list[int], tag: ModbusTag) -> float:
data_type = getattr(self.client.DATATYPE, tag.data_type.upper())
raw = self.client.convert_from_registers(registers, data_type=data_type, word_order=tag.word_order)
return round(raw * tag.scale, 6)
async def _read_tag(self, tag: ModbusTag) -> float:
count = REGISTER_COUNT[tag.data_type]
if tag.table is Table.HOLDING:
response = await self.client.read_holding_registers(tag.address, count=count, device_id=self.device_id)
else:
response = await self.client.read_input_registers(tag.address, count=count, device_id=self.device_id)
# Una risposta di eccezione Modbus non solleva errori: va controllata esplicitamente
if response.isError():
raise ModbusException(f"{tag.name}: risposta di eccezione {response}")
return self._decode(response.registers, tag)
async def run(self, queue: asyncio.Queue[Reading]) -> None:
backoff = 1.0
while True:
if not self.client.connected:
if not await self.client.connect():
logger.warning("connessione fallita, nuovo tentativo tra %.0f s", backoff)
await self._publish_bad(queue)
await asyncio.sleep(backoff)
backoff = min(backoff * 2, 30.0)
continue
logger.info("connesso al dispositivo %d", self.device_id)
backoff = 1.0
cycle_start = time.monotonic()
async with self._lock:
for tag in self.tags:
try:
value = await self._read_tag(tag)
await queue.put(Reading(tag.name, value, "good", time.time(), tag.unit))
except ModbusException as error:
logger.warning("lettura di %s fallita: %s", tag.name, error)
await queue.put(Reading(tag.name, None, "bad", time.time(), tag.unit))
# Mantiene la cadenza costante sottraendo la durata del ciclo
elapsed = time.monotonic() - cycle_start
await asyncio.sleep(max(0.0, self.interval - elapsed))
async def _publish_bad(self, queue: asyncio.Queue[Reading]) -> None:
now = time.time()
for tag in self.tags:
await queue.put(Reading(tag.name, None, "bad", now, tag.unit))
async def write_setpoint(self, tag: ModbusTag, value: float, minimum: float, maximum: float) -> float:
if not (minimum <= value <= maximum):
raise ValueError(f"{value} fuori dai limiti [{minimum}, {maximum}]")
if tag.table is not Table.HOLDING:
raise ValueError("solo gli holding register sono scrivibili")
data_type = getattr(self.client.DATATYPE, tag.data_type.upper())
raw = value / tag.scale
if tag.data_type != "float32":
raw = round(raw)
registers = self.client.convert_to_registers(raw, data_type=data_type, word_order=tag.word_order)
# Il lock impedisce che la scrittura si intrecci con un ciclo di lettura in corso
async with self._lock:
response = await self.client.write_registers(tag.address, registers, device_id=self.device_id)
if response.isError():
raise ModbusException(f"scrittura rifiutata: {response}")
confirmed = await self._read_tag(tag)
if abs(confirmed - value) > abs(tag.scale) / 2:
raise ModbusException(f"il PLC ha mantenuto {confirmed} invece di {value}")
logger.info("setpoint %s impostato a %s", tag.name, confirmed)
return confirmedTre aspetti di questo codice sono tipici del dominio industriale. Primo, una risposta di eccezione Modbus, come Illegal Data Address, non solleva un'eccezione Python: pymodbus restituisce un oggetto risposta di tipo errore, e se non si chiama isError() si rischia di leggere un attributo registers inesistente. Secondo, il ciclo sottrae la propria durata all'intervallo di attesa, così la cadenza resta stabile anche se un PLC risponde lentamente. Terzo, la scrittura rilegge sempre il valore: molti PLC ignorano le scritture in modalità manuale o applicano limiti interni, e solo la rilettura dice cosa è effettivamente attivo.
La scrittura, inoltre, non viene mai ripetuta automaticamente in caso di errore. Se la risposta si perde dopo che il PLC ha eseguito il comando, ripeterlo può avere effetti indesiderati: per un setpoint il danno è nullo, per un comando che avvia un ciclo di lavorazione no. Il parametro retries=1 del client riguarda le letture, ma conviene esserne consapevoli anche per le scritture: in ambienti critici si può creare un client separato per i comandi con retries=0.
Le sottoscrizioni OPC UA
Con OPC UA non serve interrogare il server: si crea una sottoscrizione e il server notifica le variazioni. asyncua chiama un oggetto gestore per ogni notifica; il nostro gestore converte la notifica in un Reading e lo inserisce nella stessa coda usata dal lettore Modbus. In questo modo il resto della pipeline non sa, né deve sapere, da quale protocollo provenga un dato.
# opcua_reader.py
import asyncio
import logging
import time
from asyncua import Client, Node, ua
from asyncua.common.subscription import DataChangeNotif
from tags import OpcUaTag, Reading
logger = logging.getLogger("opcua")
class SubscriptionHandler:
def __init__(self, queue: asyncio.Queue[Reading], tags_by_node: dict[str, OpcUaTag]):
self.queue = queue
self.tags_by_node = tags_by_node
def datachange_notification(self, node: Node, value, data: DataChangeNotif) -> None:
tag = self.tags_by_node[node.nodeid.to_string()]
data_value = data.monitored_item.Value
good = data_value.StatusCode.is_good()
source_time = data_value.SourceTimestamp
timestamp = source_time.timestamp() if source_time else time.time()
reading = Reading(tag.name, float(value) if good else None, "good" if good else "bad", timestamp, tag.unit)
# Il gestore è sincrono: put_nowait evita di bloccare il ciclo degli eventi
try:
self.queue.put_nowait(reading)
except asyncio.QueueFull:
logger.error("coda piena, notifica di %s scartata", tag.name)
def status_change_notification(self, status: ua.StatusChangeNotification) -> None:
logger.warning("stato della sottoscrizione cambiato: %s", status.Status)
class OpcUaReader:
def __init__(self, url: str, tags: list[OpcUaTag], publishing_interval_ms: int = 500):
self.url = url
self.tags = tags
self.publishing_interval_ms = publishing_interval_ms
async def run(self, queue: asyncio.Queue[Reading]) -> None:
backoff = 1.0
while True:
try:
await self._session(queue)
except (OSError, asyncio.TimeoutError, ua.UaError) as error:
logger.warning("sessione interrotta: %s; nuovo tentativo tra %.0f s", error, backoff)
now = time.time()
for tag in self.tags:
await queue.put(Reading(tag.name, None, "bad", now, tag.unit))
await asyncio.sleep(backoff)
backoff = min(backoff * 2, 30.0)
async def _session(self, queue: asyncio.Queue[Reading]) -> None:
# In produzione: client.set_security_string("Basic256Sha256,SignAndEncrypt,cert.der,key.pem")
async with Client(url=self.url, timeout=5) as client:
logger.info("connesso a %s", self.url)
nodes = [client.get_node(tag.node_id) for tag in self.tags]
tags_by_node = {node.nodeid.to_string(): tag for node, tag in zip(nodes, self.tags)}
handler = SubscriptionHandler(queue, tags_by_node)
subscription = await client.create_subscription(self.publishing_interval_ms, handler)
await subscription.subscribe_data_change(nodes, queuesize=10)
# Verifica periodica che la sessione sia ancora viva
while True:
await asyncio.sleep(5)
await client.check_connection()Il gestore deve essere veloce, perché viene invocato nel contesto di ricezione dei messaggi del client. Per questo usa put_nowait invece di await queue.put: se la coda è piena perché il consumatore è rimasto indietro, il dato viene scartato e segnalato, invece di bloccare tutta la ricezione. Il timestamp scelto è quello della sorgente, cioè il momento in cui il valore è stato prodotto sul campo, che per uno storico è più significativo del momento di arrivo.
Il ciclo finale con check_connection serve a far emergere una connessione caduta: se il server diventa irraggiungibile, il metodo solleva un'eccezione, si esce dal blocco async with e il ciclo esterno ricrea sessione e sottoscrizione da zero.
Banda morta e allarmi con isteresi
Un trasmettitore di pressione che oscilla di qualche millesimo di bar intorno al proprio valore genererebbe un record a ogni ciclo, gonfiando lo storico senza aggiungere informazione. La banda morta risolve il problema: un nuovo valore viene registrato solo se si discosta dall'ultimo registrato di almeno una certa quantità. I cambi di qualità vengono invece sempre registrati.
Lo stesso rumore crea un problema analogo con gli allarmi. Se la soglia alta è 80 gradi e la temperatura oscilla tra 79,9 e 80,1, un allarme ingenuo scatterebbe e rientrerebbe decine di volte al minuto, il fenomeno chiamato chattering. L'isteresi stabilisce che l'allarme rientri solo quando il valore torna sotto la soglia meno una certa quantità.
# processing.py
from dataclasses import dataclass
from tags import AlarmLimits, Reading
@dataclass
class TagState:
last_value: float | None = None
last_quality: str | None = None
high_active: bool = False
low_active: bool = False
class Processor:
def __init__(self, deadbands: dict[str, float], alarms: dict[str, AlarmLimits]):
self.deadbands = deadbands
self.alarms = alarms
self.states: dict[str, TagState] = {}
def should_store(self, reading: Reading) -> bool:
state = self.states.setdefault(reading.name, TagState())
quality_changed = reading.quality != state.last_quality
if reading.quality != "good":
state.last_quality = reading.quality
return quality_changed
deadband = self.deadbands.get(reading.name, 0.0)
outside_band = state.last_value is None or abs(reading.value - state.last_value) > deadband
if quality_changed or outside_band:
state.last_value = reading.value
state.last_quality = reading.quality
return True
return False
def evaluate_alarms(self, reading: Reading) -> list[str]:
limits = self.alarms.get(reading.name)
if limits is None or reading.quality != "good":
return []
state = self.states.setdefault(reading.name, TagState())
events = []
if limits.high is not None:
if not state.high_active and reading.value > limits.high:
state.high_active = True
events.append(f"ALARM HIGH {reading.name}={reading.value}")
elif state.high_active and reading.value < limits.high - limits.hysteresis:
state.high_active = False
events.append(f"CLEAR HIGH {reading.name}={reading.value}")
if limits.low is not None:
if not state.low_active and reading.value < limits.low:
state.low_active = True
events.append(f"ALARM LOW {reading.name}={reading.value}")
elif state.low_active and reading.value > limits.low + limits.hysteresis:
state.low_active = False
events.append(f"CLEAR LOW {reading.name}={reading.value}")
return eventsUn'osservazione importante: questi allarmi sono informativi. Gli allarmi di sicurezza veri, quelli che arrestano un impianto, sono e devono restare nel PLC o in un sistema di sicurezza dedicato, perché un servizio Python che gira su un server può fermarsi, rallentare o perdere la connessione. Il nostro servizio può notificare, non proteggere.
Lo storico in SQLite
SQLite è sorprendentemente adatto a un historian di piccole dimensioni: un file, nessun servizio da gestire, prestazioni ottime con la modalità WAL. Poiché il modulo sqlite3 è sincrono, le scritture avvengono in un thread separato tramite asyncio.to_thread, raggruppate in transazioni per ridurre il costo di ogni inserimento.
# historian.py
import sqlite3
from tags import Reading
SCHEMA = """
CREATE TABLE IF NOT EXISTS readings (
tag TEXT NOT NULL,
ts REAL NOT NULL,
value REAL,
quality TEXT NOT NULL,
unit TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_readings_tag_ts ON readings (tag, ts);
CREATE TABLE IF NOT EXISTS alarm_events (
ts REAL NOT NULL,
message TEXT NOT NULL
);
"""
class Historian:
def __init__(self, path: str):
self.connection = sqlite3.connect(path, check_same_thread=False)
self.connection.execute("PRAGMA journal_mode=WAL")
self.connection.executescript(SCHEMA)
def store_batch(self, readings: list[Reading]) -> None:
with self.connection:
self.connection.executemany(
"INSERT INTO readings (tag, ts, value, quality, unit) VALUES (?, ?, ?, ?, ?)",
[(r.name, r.timestamp, r.value, r.quality, r.unit) for r in readings],
)
def store_alarms(self, timestamp: float, messages: list[str]) -> None:
with self.connection:
self.connection.executemany(
"INSERT INTO alarm_events (ts, message) VALUES (?, ?)",
[(timestamp, message) for message in messages],
)
def close(self) -> None:
self.connection.close()Mettere insieme i pezzi
Il programma principale crea la coda condivisa, avvia i due lettori come task e consuma le letture in un ciclo che applica banda morta e allarmi e accumula i record da salvare. Il salvataggio avviene ogni secondo o quando il lotto raggiunge una certa dimensione.
# main.py
import asyncio
import logging
import signal
import time
from historian import Historian
from modbus_reader import ModbusReader
from opcua_reader import OpcUaReader
from processing import Processor
from tags import AlarmLimits, ModbusTag, OpcUaTag, Reading, Table
MODBUS_TAGS = [
ModbusTag("tank1.temperature", Table.HOLDING, 100, "int16", scale=0.1, unit="°C", deadband=0.2,
alarms=AlarmLimits(high=80.0, low=5.0, hysteresis=1.5)),
ModbusTag("tank1.pressure", Table.HOLDING, 102, "float32", unit="bar", deadband=0.01,
alarms=AlarmLimits(high=6.0, hysteresis=0.2)),
ModbusTag("line1.pieces", Table.INPUT, 10, "uint32", word_order="little"),
]
OPCUA_TAGS = [
OpcUaTag("line1.speed", "ns=2;s=Line1.Speed", unit="m/min", deadband=0.5),
OpcUaTag("line1.motor_current", "ns=2;s=Line1.MotorCurrent", unit="A", deadband=0.1,
alarms=AlarmLimits(high=42.0, hysteresis=2.0)),
]
async def consume(queue: asyncio.Queue[Reading], processor: Processor, historian: Historian) -> None:
batch: list[Reading] = []
last_flush = time.monotonic()
while True:
try:
reading = await asyncio.wait_for(queue.get(), timeout=1.0)
if processor.should_store(reading):
batch.append(reading)
alarms = processor.evaluate_alarms(reading)
if alarms:
logging.getLogger("alarms").warning("; ".join(alarms))
await asyncio.to_thread(historian.store_alarms, reading.timestamp, alarms)
except asyncio.TimeoutError:
pass
if batch and (len(batch) >= 200 or time.monotonic() - last_flush >= 1.0):
await asyncio.to_thread(historian.store_batch, batch)
batch = []
last_flush = time.monotonic()
async def main() -> None:
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")
queue: asyncio.Queue[Reading] = asyncio.Queue(maxsize=10_000)
all_tags = [*MODBUS_TAGS, *OPCUA_TAGS]
processor = Processor(
deadbands={tag.name: tag.deadband for tag in all_tags},
alarms={tag.name: tag.alarms for tag in all_tags},
)
historian = Historian("historian.db")
modbus_reader = ModbusReader("127.0.0.1", 502, device_id=1, tags=MODBUS_TAGS, interval=1.0)
opcua_reader = OpcUaReader("opc.tcp://127.0.0.1:4840", OPCUA_TAGS)
stop = asyncio.Event()
loop = asyncio.get_running_loop()
for sig in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(sig, stop.set)
tasks = [
asyncio.create_task(modbus_reader.run(queue), name="modbus"),
asyncio.create_task(opcua_reader.run(queue), name="opcua"),
asyncio.create_task(consume(queue, processor, historian), name="consumer"),
]
await stop.wait()
logging.info("arresto in corso")
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
modbus_reader.client.close()
historian.close()
if __name__ == "__main__":
asyncio.run(main())La coda ha una dimensione massima di diecimila elementi. Senza un limite, un database lento o un disco pieno farebbero crescere la memoria del processo fino all'esaurimento; con il limite, il lettore Modbus rallenta naturalmente perché await queue.put attende spazio, mentre il gestore OPC UA scarta le notifiche in eccesso e lo registra nel log. È una forma elementare ma efficace di contropressione.
Interrogare lo storico
Il vantaggio di avere i dati in SQLite è che l'analisi diventa immediata. Ecco come ottenere le medie su intervalli di cinque minuti per un tag, escludendo i campioni di cattiva qualità.
# report.py
import sqlite3
query = """
SELECT CAST(ts / 300 AS INTEGER) * 300 AS bucket,
AVG(value) AS mean,
MIN(value) AS minimum,
MAX(value) AS maximum,
COUNT(*) AS samples
FROM readings
WHERE tag = ? AND quality = 'good' AND ts >= strftime('%s', 'now', '-1 day')
GROUP BY bucket
ORDER BY bucket
"""
with sqlite3.connect("historian.db") as connection:
for bucket, mean, minimum, maximum, samples in connection.execute(query, ("tank1.temperature",)):
print(f"{bucket}: media {mean:.2f}, min {minimum:.2f}, max {maximum:.2f} ({samples} campioni)")C'è una sottigliezza statistica: per effetto della banda morta, la media semplice dei record non è una media temporale esatta, perché i periodi in cui il valore è stabile sono rappresentati da pochi campioni. Per analisi rigorose si usa una media pesata sulla durata di ciascun valore, che si ottiene facilmente con le funzioni finestra di SQLite o caricando i dati in pandas.
Sicurezza
- Modbus TCP non autentica nessuno. Il servizio va eseguito in una rete segregata, con regole firewall che permettono solo le connessioni dal servizio verso i PLC e nessuna nel verso opposto.
- Le scritture vanno abilitate esplicitamente. Il metodo
write_setpointrichiede limiti espliciti e non è collegato ad alcuna interfaccia esterna. Se lo si espone tramite un'API, servono autenticazione, autorizzazione per singolo tag e un registro di audit. - OPC UA con cifratura.
asyncuasupportaset_security_stringper configurare policy, modalità e certificati. In produzione la modalità senza sicurezza non è un'opzione. - Dipendenze bloccate. In un servizio che parla con un impianto, un aggiornamento automatico di una libreria che cambia la semantica di un parametro è un rischio concreto. Le versioni vanno fissate in un file di lock e aggiornate solo dopo prove sul simulatore.
Conclusioni
Con circa trecento righe di Python abbiamo costruito un servizio che acquisisce dati da due protocolli industriali, li normalizza in un unico flusso, ne riduce il volume con la banda morta, valuta allarmi senza oscillazioni e costruisce uno storico interrogabile. La scelta di far confluire tutto in una coda comune rende l'architettura estendibile: aggiungere un nuovo protocollo significa scrivere un nuovo lettore, mentre sostituire SQLite con un database per serie temporali significa cambiare solo la classe Historian. Ed è qui che Python dà il meglio: il passo dall'acquisizione all'analisi si riduce a una query, o a un read_sql di pandas.