Interagire con i sistemi SCADA in Java
Java ha una presenza storica e solida nel software industriale. Molti sistemi MES, piattaforme di integrazione e prodotti SCADA commerciali sono scritti in Java, e la JVM è spesso già installata sui server delle reti di stabilimento. A questo si aggiunge un vantaggio specifico: Eclipse Milo, una delle implementazioni open source di OPC UA più complete e curate, è scritta proprio in Java.
In questo articolo costruiremo un bridge che fa da ponte tra due mondi: legge i dati di un PLC di generazione precedente via Modbus TCP, si abbona alle variabili di un PLC moderno tramite OPC UA, unifica tutto in un archivio dei tag in memoria con notifiche ai listener e offre un servizio di scrittura con controlli di sicurezza. È uno scenario molto comune negli stabilimenti reali, dove convivono macchine di epoche diverse.
Il contesto: SCADA, PLC e protocolli
Un sistema SCADA supervisiona un processo fisico attraverso una gerarchia di componenti. I PLC eseguono la logica di controllo in tempo reale; i server SCADA raccolgono i dati, gestiscono allarmi e storico; le HMI li presentano agli operatori. Il nostro bridge si inserisce come consumatore di dati accanto al server SCADA, con una regola fondamentale: non partecipa alla logica di controllo, che resta nei PLC.
I due protocolli che useremo hanno filosofie opposte. Modbus TCP espone tabelle anonime di bit e registri a 16 bit, individuati solo da un numero: tipo, scala e significato dei dati sono convenzioni documentate a parte. OPC UA espone invece uno spazio degli indirizzi di nodi tipizzati, ognuno con un identificativo, un valore, un codice di stato e dei timestamp, e offre sottoscrizioni, sicurezza a livello di canale e autenticazione degli utenti.
Dipendenze
Useremo Java 21, Eclipse Milo 1.1.8 per OPC UA e la libreria Modbus di digitalpetri, scritta dallo stesso autore principale di Milo, nella versione 2.1.6. Entrambe richiedono Java 17 o successivo.
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.example</groupId>
<artifactId>scada-bridge</artifactId>
<version>1.0.0</version>
<properties>
<maven.compiler.release>21</maven.compiler.release>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<milo.version>1.1.8</milo.version>
</properties>
<dependencies>
<dependency>
<groupId>org.eclipse.milo</groupId>
<artifactId>milo-sdk-client</artifactId>
<version>${milo.version}</version>
</dependency>
<dependency>
<groupId>com.digitalpetri.modbus</groupId>
<artifactId>modbus-tcp</artifactId>
<version>2.1.6</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>2.0.17</version>
</dependency>
</dependencies>
</project>Milo usa SLF4J per i log, e senza un'implementazione collegata ogni messaggio della libreria andrebbe perso: slf4j-simple è sufficiente per iniziare. La serie 1.2 di Milo è in fase di release candidate al momento della scrittura; il codice di questo articolo usa le API della serie 1.1, che sono state verificate sulla versione indicata.
Il modello: tag e letture
I record di Java sono ideali per descrivere dati immutabili come la definizione di un tag e il valore letto. Il modello è indipendente dai protocolli: un tag Modbus e un nodo OPC UA producono lo stesso tipo di lettura.
// src/main/java/com/example/scada/model/TagReading.java
package com.example.scada.model;
import java.time.Instant;
public record TagReading(String name, Double value, Quality quality, Instant sourceTime, String unit) {
public enum Quality { GOOD, UNCERTAIN, BAD }
public static TagReading bad(String name, String unit) {
return new TagReading(name, null, Quality.BAD, Instant.now(), unit);
}
}// src/main/java/com/example/scada/model/ModbusTag.java
package com.example.scada.model;
public record ModbusTag(
String name,
Table table,
int address,
DataType type,
WordOrder wordOrder,
double scale,
String unit,
WriteLimits writeLimits) {
public enum Table { HOLDING, INPUT }
public enum WordOrder { HIGH_FIRST, LOW_FIRST }
public enum DataType {
INT16(1), UINT16(1), INT32(2), UINT32(2), FLOAT32(2);
private final int registers;
DataType(int registers) {
this.registers = registers;
}
public int registers() {
return registers;
}
}
// Limiti di scrittura: null significa tag in sola lettura
public record WriteLimits(double min, double max) {
}
public boolean writable() {
return writeLimits != null && table == Table.HOLDING;
}
}Il campo address è l'offset a base zero trasmesso nel protocollo. La notazione storica che si trova nei manuali, come 40101, va convertita: il 4 indica gli holding register e la numerazione parte da uno, quindi l'offset è 100.
L'archivio dei tag
L'archivio conserva l'ultimo valore di ogni tag e notifica i listener solo quando valore o qualità cambiano. Usa una ConcurrentHashMap perché viene aggiornato da thread diversi: il thread di polling Modbus e i thread di Milo che consegnano le notifiche OPC UA.
// src/main/java/com/example/scada/TagStore.java
package com.example.scada;
import com.example.scada.model.TagReading;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
public final class TagStore {
private final Map<String, TagReading> current = new ConcurrentHashMap<>();
private final List<Consumer<TagReading>> listeners = new CopyOnWriteArrayList<>();
public void addListener(Consumer<TagReading> listener) {
listeners.add(listener);
}
public void update(TagReading reading) {
TagReading previous = current.put(reading.name(), reading);
boolean changed = previous == null
|| previous.quality() != reading.quality()
|| !Objects.equals(previous.value(), reading.value());
if (changed) {
for (Consumer<TagReading> listener : listeners) {
try {
listener.accept(reading);
} catch (RuntimeException e) {
// Un listener difettoso non deve interrompere la distribuzione agli altri
System.getLogger("store").log(System.Logger.Level.WARNING, "listener fallito", e);
}
}
}
}
public Map<String, TagReading> snapshot() {
return Map.copyOf(current);
}
}Decodificare i registri Modbus
La libreria digitalpetri restituisce i registri come array di byte nell'ordine di trasmissione: due byte per registro, big-endian. Per i tipi a 32 bit bisogna sapere in quale dei due registri si trova la parte alta, perché il protocollo non lo stabilisce e i costruttori seguono convenzioni diverse. ByteBuffer si occupa del resto.
// src/main/java/com/example/scada/modbus/RegisterCodec.java
package com.example.scada.modbus;
import com.example.scada.model.ModbusTag;
import java.nio.ByteBuffer;
public final class RegisterCodec {
private RegisterCodec() {
}
public static double decode(byte[] block, int offsetInRegisters, ModbusTag tag) {
int byteOffset = offsetInRegisters * 2;
int length = tag.type().registers() * 2;
byte[] bytes = new byte[length];
System.arraycopy(block, byteOffset, bytes, 0, length);
if (tag.type().registers() == 2 && tag.wordOrder() == ModbusTag.WordOrder.LOW_FIRST) {
// Scambia le due word per riportare la parte alta in testa
bytes = new byte[] {bytes[2], bytes[3], bytes[0], bytes[1]};
}
ByteBuffer buffer = ByteBuffer.wrap(bytes);
double raw = switch (tag.type()) {
case INT16 -> buffer.getShort();
case UINT16 -> Short.toUnsignedInt(buffer.getShort());
case INT32 -> buffer.getInt();
case UINT32 -> Integer.toUnsignedLong(buffer.getInt());
case FLOAT32 -> {
float value = buffer.getFloat();
if (!Float.isFinite(value)) {
throw new IllegalStateException("Float non valido per " + tag.name() + ": controllare l'ordine delle word");
}
yield value;
}
};
return raw * tag.scale();
}
public static int encodeInt16(double engineeringValue, ModbusTag tag) {
long raw = Math.round(engineeringValue / tag.scale());
if (raw < Short.MIN_VALUE || raw > Short.MAX_VALUE) {
throw new IllegalArgumentException("Valore grezzo fuori dall'intervallo int16: " + raw);
}
// Rappresentazione senza segno a 16 bit richiesta dalla richiesta di scrittura
return (int) raw & 0xFFFF;
}
}La sorgente Modbus
La sorgente Modbus esegue un ciclo di lettura periodico su un ScheduledExecutorService a thread singolo. Usare scheduleWithFixedDelay invece di scheduleAtFixedRate è importante: con il ritardo fisso, il ciclo successivo parte solo dopo la fine del precedente, quindi un PLC lento non accumula richieste arretrate. Lo stesso esecutore viene usato anche per le scritture, così tutte le operazioni sul dispositivo sono serializzate per costruzione.
Il trasporto Netty di digitalpetri, nella configurazione predefinita, mantiene la connessione persistente e si riconnette da solo. Il nostro codice deve solo gestire le eccezioni e segnare i tag come non validi quando una lettura fallisce.
// src/main/java/com/example/scada/modbus/ModbusSource.java
package com.example.scada.modbus;
import com.digitalpetri.modbus.client.ModbusTcpClient;
import com.digitalpetri.modbus.exceptions.ModbusException;
import com.digitalpetri.modbus.exceptions.ModbusResponseException;
import com.digitalpetri.modbus.pdu.ReadHoldingRegistersRequest;
import com.digitalpetri.modbus.pdu.ReadInputRegistersRequest;
import com.digitalpetri.modbus.pdu.WriteSingleRegisterRequest;
import com.digitalpetri.modbus.tcp.client.NettyTcpClientTransport;
import com.example.scada.TagStore;
import com.example.scada.model.ModbusTag;
import com.example.scada.model.TagReading;
import java.time.Duration;
import java.time.Instant;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;
public final class ModbusSource implements AutoCloseable {
private static final System.Logger LOG = System.getLogger("modbus");
private final ModbusTcpClient client;
private final int unitId;
private final List<ModbusTag> tags;
private final Map<String, ModbusTag> tagsByName;
private final TagStore store;
private final ScheduledExecutorService executor =
Executors.newSingleThreadScheduledExecutor(r -> Thread.ofPlatform().name("modbus-poller").daemon().unstarted(r));
public ModbusSource(String host, int port, int unitId, List<ModbusTag> tags, TagStore store) {
var transport = NettyTcpClientTransport.create(cfg -> {
cfg.setHostname(host);
cfg.setPort(port);
cfg.setConnectTimeout(Duration.ofSeconds(3));
});
this.client = ModbusTcpClient.create(transport, cfg -> cfg.setRequestTimeout(Duration.ofMillis(1500)));
this.unitId = unitId;
this.tags = tags.stream().sorted(Comparator.comparingInt(ModbusTag::address)).toList();
this.tagsByName = tags.stream().collect(Collectors.toUnmodifiableMap(ModbusTag::name, Function.identity()));
this.store = store;
}
public void start(Duration interval) {
executor.scheduleWithFixedDelay(this::pollSafely, 0, interval.toMillis(), TimeUnit.MILLISECONDS);
}
private void pollSafely() {
try {
if (!client.isConnected()) {
client.connect();
LOG.log(System.Logger.Level.INFO, "connesso al dispositivo Modbus");
}
poll();
} catch (Exception e) {
// Qualsiasi eccezione non gestita annullerebbe le esecuzioni successive del task pianificato
LOG.log(System.Logger.Level.WARNING, "ciclo di lettura fallito: " + e.getMessage());
tags.forEach(tag -> store.update(TagReading.bad(tag.name(), tag.unit())));
}
}
private void poll() throws ModbusException {
for (ModbusTag.Table table : ModbusTag.Table.values()) {
List<ModbusTag> tableTags = tags.stream().filter(t -> t.table() == table).toList();
if (tableTags.isEmpty()) {
continue;
}
int start = tableTags.getFirst().address();
ModbusTag last = tableTags.getLast();
int quantity = last.address() + last.type().registers() - start;
if (quantity > 125) {
throw new IllegalStateException("Blocco troppo esteso per una singola richiesta: " + quantity);
}
byte[] registers = table == ModbusTag.Table.HOLDING
? client.readHoldingRegisters(unitId, new ReadHoldingRegistersRequest(start, quantity)).registers()
: client.readInputRegisters(unitId, new ReadInputRegistersRequest(start, quantity)).registers();
Instant now = Instant.now();
for (ModbusTag tag : tableTags) {
try {
double value = RegisterCodec.decode(registers, tag.address() - start, tag);
store.update(new TagReading(tag.name(), value, TagReading.Quality.GOOD, now, tag.unit()));
} catch (IllegalStateException e) {
LOG.log(System.Logger.Level.WARNING, e.getMessage());
store.update(TagReading.bad(tag.name(), tag.unit()));
}
}
}
}
public double write(String name, double value) throws ModbusException {
ModbusTag tag = tagsByName.get(name);
if (tag == null || !tag.writable()) {
throw new IllegalArgumentException("Tag non scrivibile: " + name);
}
if (tag.type() != ModbusTag.DataType.INT16) {
throw new IllegalArgumentException("Scrittura supportata solo per tag INT16");
}
try {
// La scrittura gira sullo stesso thread del polling: nessuna sovrapposizione con le letture
return executor.submit(() -> writeAndVerify(tag, value)).get(5, TimeUnit.SECONDS);
} catch (ExecutionException e) {
if (e.getCause() instanceof ModbusException modbusException) {
throw modbusException;
}
throw new IllegalStateException(e.getCause());
} catch (Exception e) {
throw new IllegalStateException("Scrittura non completata: esito incerto", e);
}
}
private double writeAndVerify(ModbusTag tag, double value) throws ModbusException {
int raw = RegisterCodec.encodeInt16(value, tag);
client.writeSingleRegister(unitId, new WriteSingleRegisterRequest(tag.address(), raw));
byte[] check = client.readHoldingRegisters(unitId, new ReadHoldingRegistersRequest(tag.address(), 1)).registers();
double confirmed = RegisterCodec.decode(check, 0, tag);
if (Math.abs(confirmed - value) > Math.abs(tag.scale()) / 2) {
throw new ModbusResponseException(0x06, 0x03);
}
return confirmed;
}
@Override
public void close() throws Exception {
executor.shutdownNow();
executor.awaitTermination(2, TimeUnit.SECONDS);
client.disconnect();
}
}Il blocco try che avvolge tutto il ciclo in pollSafely non è facoltativo. Un dettaglio poco noto di ScheduledExecutorService è che se un task periodico solleva un'eccezione, tutte le esecuzioni successive vengono annullate in silenzio: il servizio continua a girare, ma il polling si ferma per sempre senza alcun messaggio. È uno dei bug più insidiosi in un servizio di acquisizione.
Questa versione legge un unico blocco per tabella, dal primo all'ultimo indirizzo configurato. È semplice e funziona bene quando i tag sono raccolti in un'area compatta; se gli indirizzi sono sparsi, conviene suddividerli in più blocchi contigui, e alcuni PLC lo richiedono, perché rifiutano con Illegal Data Address le letture che attraversano registri non mappati.
Quando la rilettura non conferma il valore, il metodo segnala l'errore come una risposta di eccezione Illegal Data Value (codice 3 sulla funzione 6), una scelta di comodo per riusare la gerarchia di eccezioni della libreria. In un'applicazione più grande conviene definire un'eccezione propria.
La sorgente OPC UA con Eclipse Milo
Con OPC UA non c'è bisogno di polling. Si crea una sottoscrizione, che definisce ogni quanto il server pubblica le notifiche, e vi si aggiungono dei monitored item, ciascuno associato a un nodo e con un proprio intervallo di campionamento. In Milo 1.1 le classi OpcUaSubscription e OpcUaMonitoredItem rappresentano lo stato desiderato: si configurano localmente, poi create crea la sottoscrizione sul server e synchronizeMonitoredItems allinea gli item.
// src/main/java/com/example/scada/opcua/OpcUaSource.java
package com.example.scada.opcua;
import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
import com.example.scada.TagStore;
import com.example.scada.model.TagReading;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.subscriptions.MonitoredItemSynchronizationException;
import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaMonitoredItem;
import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaSubscription;
import org.eclipse.milo.opcua.stack.core.UaException;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
public final class OpcUaSource implements AutoCloseable {
private static final System.Logger LOG = System.getLogger("opcua");
public record NodeTag(String name, String nodeId, String unit) {
}
private final OpcUaClient client;
private final List<NodeTag> nodes;
private final TagStore store;
private OpcUaSubscription subscription;
public OpcUaSource(String endpointUrl, List<NodeTag> nodes, TagStore store) throws UaException {
this.client = OpcUaClient.create(
endpointUrl,
// In laboratorio si sceglie l'endpoint senza sicurezza; in produzione Basic256Sha256 o superiore
endpoints -> endpoints.stream()
.filter(e -> SecurityPolicy.None.getUri().equals(e.getSecurityPolicyUri()))
.findFirst(),
transport -> {},
config -> config
.setApplicationName(LocalizedText.english("SCADA Bridge"))
.setApplicationUri("urn:example:scada-bridge")
.setRequestTimeout(uint(5000)));
this.nodes = nodes;
this.store = store;
}
public void start() throws UaException {
client.connect();
LOG.log(System.Logger.Level.INFO, "sessione OPC UA attiva");
subscription = new OpcUaSubscription(client, 500.0);
for (NodeTag node : nodes) {
OpcUaMonitoredItem item = OpcUaMonitoredItem.newDataItem(NodeId.parse(node.nodeId()));
item.setSamplingInterval(250.0);
item.setQueueSize(uint(10));
// Il listener va registrato prima della sincronizzazione: il primo valore arriva subito
item.setDataValueListener((monitoredItem, value) -> store.update(toReading(node, value)));
subscription.addMonitoredItem(item);
}
subscription.create();
try {
subscription.synchronizeMonitoredItems();
} catch (MonitoredItemSynchronizationException e) {
// Alcuni item possono fallire, per esempio per un NodeId inesistente: gli altri restano attivi
LOG.log(System.Logger.Level.ERROR, "creazione parziale dei monitored item: " + e.getMessage());
}
}
private static TagReading toReading(NodeTag node, DataValue value) {
StatusCode status = value.statusCode();
Object payload = value.value().value();
Instant sourceTime = value.sourceTime() != null ? value.sourceTime().getJavaInstant() : Instant.now();
if (status.isBad() || !(payload instanceof Number number)) {
return new TagReading(node.name(), null, TagReading.Quality.BAD, sourceTime, node.unit());
}
TagReading.Quality quality = status.isGood() ? TagReading.Quality.GOOD : TagReading.Quality.UNCERTAIN;
return new TagReading(node.name(), number.doubleValue(), quality, sourceTime, node.unit());
}
public void write(String nodeId, double value) throws UaException {
// Il tipo del Variant deve coincidere con il DataType del nodo: qui si assume Double
DataValue dataValue = DataValue.valueOnly(Variant.ofDouble(value));
List<StatusCode> results = client.writeValues(List.of(NodeId.parse(nodeId)), List.of(dataValue));
StatusCode status = results.getFirst();
if (!status.isGood()) {
throw new UaException(status);
}
}
public Map<String, Object> readOnce(String nodeId) throws UaException {
DataValue value = client.readValue(0.0, TimestampsToReturn.Both, NodeId.parse(nodeId));
return Map.of("status", value.statusCode().toString(), "value", String.valueOf(value.value().value()));
}
@Override
public void close() throws Exception {
try {
if (subscription != null) {
subscription.delete();
}
} finally {
client.disconnect();
}
}
}Alcuni aspetti del codice derivano direttamente dal funzionamento di OPC UA.
- Tre livelli di qualità. Un codice di stato OPC UA può essere buono, incerto o cattivo. Un valore incerto, per esempio un sensore fuori calibrazione o un dato sostituito manualmente, non va trattato come buono né scartato: va mostrato con un'indicazione esplicita. Per questo il modello ha la qualità
UNCERTAIN. - Timestamp della sorgente.
sourceTimeindica quando il valore è stato prodotto dal dispositivo, non quando è arrivato. È quello da registrare in uno storico. - Listener rapidi. Milo consegna le notifiche in modo sincrono come meccanismo di contropressione: finché il listener non ritorna, non vengono processate altre notifiche né inviate nuove richieste di pubblicazione. Il nostro listener aggiorna solo una mappa in memoria; qualsiasi operazione lenta, come una scrittura su database, va delegata a una coda separata.
- Riconnessione. Milo gestisce da solo la perdita del canale: riapre la sessione e tenta di trasferire le sottoscrizioni esistenti. Se il trasferimento fallisce, lo segnala tramite un
SubscriptionListener, e in un servizio di produzione conviene registrarne uno che ricrei la sottoscrizione.
Il metodo write mostra l'altro aspetto da non trascurare: il tipo del valore. Se il nodo è di tipo Float e si invia un Double, il server risponde con Bad_TypeMismatch e il valore non cambia. Per scritture generiche conviene leggere prima l'attributo DataType del nodo e costruire il Variant di conseguenza.
Il servizio di scrittura
Le scritture verso l'impianto passano da un unico punto, che applica i controlli e registra l'operazione. Il registro di audit qui è un logger, ma in produzione deve essere un archivio persistente e non modificabile.
// src/main/java/com/example/scada/WriteService.java
package com.example.scada;
import com.example.scada.model.ModbusTag;
import com.example.scada.model.TagReading;
import com.example.scada.modbus.ModbusSource;
import java.util.Map;
public final class WriteService {
private static final System.Logger AUDIT = System.getLogger("audit");
private final ModbusSource modbus;
private final Map<String, ModbusTag> writableTags;
private final TagStore store;
public WriteService(ModbusSource modbus, Map<String, ModbusTag> writableTags, TagStore store) {
this.modbus = modbus;
this.writableTags = writableTags;
this.store = store;
}
public double writeSetpoint(String user, String tagName, double value) throws Exception {
ModbusTag tag = writableTags.get(tagName);
if (tag == null || !tag.writable()) {
throw new SecurityException("Il tag " + tagName + " non è abilitato alla scrittura");
}
ModbusTag.WriteLimits limits = tag.writeLimits();
if (Double.isNaN(value) || value < limits.min() || value > limits.max()) {
throw new IllegalArgumentException(
"Valore %.2f fuori dai limiti [%.2f, %.2f]".formatted(value, limits.min(), limits.max()));
}
TagReading before = store.snapshot().get(tagName);
String previous = before != null && before.value() != null ? before.value().toString() : "n/d";
try {
double confirmed = modbus.write(tagName, value);
AUDIT.log(System.Logger.Level.INFO, "utente=%s tag=%s da=%s a=%s esito=OK"
.formatted(user, tagName, previous, confirmed));
return confirmed;
} catch (Exception e) {
// L'esito di una scrittura fallita è incerto: non si ritenta, si registra e si segnala
AUDIT.log(System.Logger.Level.WARNING, "utente=%s tag=%s da=%s richiesto=%s esito=ERRORE %s"
.formatted(user, tagName, previous, value, e.getMessage()));
throw e;
}
}
}Assemblare il bridge
La classe principale crea l'archivio, le due sorgenti e il servizio di scrittura, registra un listener che stampa le variazioni e un hook di arresto che chiude tutto in ordine.
// src/main/java/com/example/scada/Main.java
package com.example.scada;
import com.example.scada.model.ModbusTag;
import com.example.scada.model.ModbusTag.DataType;
import com.example.scada.model.ModbusTag.Table;
import com.example.scada.model.ModbusTag.WordOrder;
import com.example.scada.model.ModbusTag.WriteLimits;
import com.example.scada.modbus.ModbusSource;
import com.example.scada.opcua.OpcUaSource;
import com.example.scada.opcua.OpcUaSource.NodeTag;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.function.Function;
import java.util.stream.Collectors;
public final class Main {
public static void main(String[] args) throws Exception {
List<ModbusTag> modbusTags = List.of(
new ModbusTag("press.temperature", Table.HOLDING, 100, DataType.INT16, WordOrder.HIGH_FIRST, 0.1, "°C", null),
new ModbusTag("press.setpoint", Table.HOLDING, 101, DataType.INT16, WordOrder.HIGH_FIRST, 0.1, "°C",
new WriteLimits(20.0, 180.0)),
new ModbusTag("press.force", Table.HOLDING, 102, DataType.FLOAT32, WordOrder.HIGH_FIRST, 1.0, "kN", null),
new ModbusTag("press.cycles", Table.INPUT, 0, DataType.UINT32, WordOrder.LOW_FIRST, 1.0, "", null));
List<NodeTag> opcUaNodes = List.of(
new NodeTag("robot.speed", "ns=2;s=Robot1.Speed", "%"),
new NodeTag("robot.temperature", "ns=2;s=Robot1.MotorTemperature", "°C"));
TagStore store = new TagStore();
store.addListener(reading -> System.out.printf("%-20s %-10s %s %s%n",
reading.name(), reading.quality(), reading.value(), reading.unit()));
ModbusSource modbus = new ModbusSource(
System.getenv().getOrDefault("MODBUS_HOST", "127.0.0.1"), 502, 1, modbusTags, store);
OpcUaSource opcUa = new OpcUaSource(
System.getenv().getOrDefault("OPCUA_URL", "opc.tcp://127.0.0.1:4840"), opcUaNodes, store);
Map<String, ModbusTag> writable = modbusTags.stream()
.filter(ModbusTag::writable)
.collect(Collectors.toUnmodifiableMap(ModbusTag::name, Function.identity()));
WriteService writeService = new WriteService(modbus, writable, store);
CountDownLatch stopped = new CountDownLatch(1);
Runtime.getRuntime().addShutdownHook(Thread.ofPlatform().unstarted(() -> {
System.out.println("Arresto del bridge in corso");
try (opcUa; modbus) {
// Le risorse vengono chiuse in ordine inverso rispetto alla dichiarazione
} catch (Exception e) {
System.err.println("Errore durante l'arresto: " + e.getMessage());
}
stopped.countDown();
}));
modbus.start(Duration.ofSeconds(1));
opcUa.start();
// Esempio di scrittura controllata dopo qualche secondo di acquisizione
Thread.sleep(5000);
try {
double confirmed = writeService.writeSetpoint("operatore.demo", "press.setpoint", 150.0);
System.out.println("Setpoint confermato: " + confirmed);
} catch (Exception e) {
System.err.println("Scrittura rifiutata: " + e.getMessage());
}
stopped.await();
}
}Il costrutto try (opcUa; modbus) usa una caratteristica introdotta in Java 9: le variabili effettivamente finali possono essere usate direttamente come risorse, e vengono chiuse in ordine inverso, quindi prima la sorgente Modbus e poi quella OPC UA.
Esporre il bridge come server OPC UA
Uno sviluppo naturale di questo progetto è fare il percorso inverso: invece di stampare i valori, il bridge potrebbe pubblicarli a sua volta come nodi di un server OPC UA, usando il modulo milo-sdk-server. In questo modo il vecchio PLC Modbus diventerebbe visibile al sistema SCADA con nomi, tipi, unità di misura e codici di qualità, come se fosse un dispositivo moderno. È esattamente ciò che fanno molti gateway industriali commerciali, e Milo fornisce tutto il necessario per costruirne uno: lo spazio degli indirizzi personalizzato si definisce estendendo un namespace gestito e aggiornando i nodi a ogni variazione dell'archivio.
Sicurezza
- Segmentazione di rete. Modbus TCP non autentica né cifra. Il bridge deve stare in una zona dedicata, l'unica autorizzata a raggiungere i PLC, secondo il modello a zone e condotti della IEC 62443.
- OPC UA con sicurezza attiva. In produzione la selezione dell'endpoint deve scegliere una policy come
Basic256Sha256oAes256_Sha256_RsaPsscon modalitàSignAndEncrypt, e il client deve avere un certificato proprio, configurato tramite ilKeyStoreLoadero un meccanismo equivalente e approvato dall'amministratore del server. - Utenti con privilegi minimi. Se il server OPC UA lo consente, il bridge deve autenticarsi con un utente che può scrivere solo i nodi previsti.
- Scritture esplicite e tracciate. Solo i tag con limiti dichiarati sono scrivibili, ogni scrittura è verificata per rilettura e registrata, e nessuna viene ritentata automaticamente.
Conclusioni
Abbiamo visto come Java, grazie a Eclipse Milo e alla libreria Modbus di digitalpetri, permetta di costruire un bridge industriale robusto con poco codice: polling Modbus serializzato su un esecutore dedicato, sottoscrizioni OPC UA con gestione della qualità a tre livelli, un archivio dei tag thread-safe e un servizio di scrittura con controlli e audit. Le insidie principali non sono nei protocolli in sé, ma nei dettagli di esecuzione: un'eccezione che uccide in silenzio un task pianificato, un listener lento che blocca le notifiche, un Variant del tipo sbagliato. Conoscerle in anticipo fa la differenza tra un servizio che funziona in laboratorio e uno che gira per mesi in uno stabilimento.