Industrial IoT — SCADA, PLC Communication, OPC-UA, Modbus, and Digital Twins Explained

# modbus_simulator.py — simulated Modbus TCP communication
import struct, socket

class ModbusTCPMaster:
    def __init__(self, host: str, port: int = 502):
        self.host = host
        self.port = port

    def read_holding_registers(self, start_address: int, count: int) -> list:
        """Read holding registers from Modbus slave"""
        # Build Modbus TCP frame
        transaction_id = 1
        protocol_id = 0
        length = 6 + count * 2  # Bytes following unit ID
        unit_id = 1
        function_code = 3  # Read Holding Registers

        request = struct.pack('>HHHBBHH',
            transaction_id, protocol_id, length,
            unit_id, function_code,
            start_address, count
        )

        print(f"Modbus request: Read {count} registers starting at {start_address}")

        # Simulated response (in real: send via socket)
        # Return mock temperature, pressure, vibration values
        return [25 + i * 5 for i in range(count)]

# Simulate reading from a PLC
master = ModbusTCPMaster("192.168.1.100")
registers = master.read_holding_registers(0, 3)
print(f"Register 0 (Temperature): {registers[0]}°C")
print(f"Register 1 (Pressure): {registers[1]} psi")
print(f"Register 2 (Vibration): {registers[2]} mm/s")

# opcua_simulator.py — simulated OPC-UA client (real library: opcua-asyncio)
class OPCUAClient:
    def __init__(self, endpoint: str):
        self.endpoint = endpoint

    def read_variable(self, node_id: str):
        """Read an OPC-UA variable"""
        print(f"OPC-UA connect to {self.endpoint}")
        print(f"Reading node: {node_id}")

        # Simulated values based on node ID
        values = {
            "ns=2;s=Temperature": 75.3,
            "ns=2;s=Pressure": 145.2,
            "ns=2;s=FlowRate": 12.7,
            "ns=2;s=ValvePosition": 45.0,
        }
        return values.get(node_id, 0.0)

client = OPCUAClient("opc.tcp://192.168.1.100:4840")
temp = client.read_variable("ns=2;s=Temperature")
pressure = client.read_variable("ns=2;s=Pressure")
print(f"Temperature: {temp}°C")
print(f"Pressure: {pressure} psi")

# predictive_maintenance.py
import random, json, time
from datetime import datetime, timedelta

class PredictiveMaintenanceEngine:
    def __init__(self):
        self.model = None  # In production: trained ML model

    def analyze(self, vibration: float, temperature: float, pressure: float) -> dict:
        # Simplified rule-based model
        risk_score = 0.0
        warnings = []

        if vibration > 7.0:
            risk_score += 0.4
            warnings.append(f"Critical vibration: {vibration:.1f} mm/s")
        elif vibration > 4.0:
            risk_score += 0.2
            warnings.append(f"Elevated vibration: {vibration:.1f} mm/s")

        if temperature > 90.0:
            risk_score += 0.3
            warnings.append(f"Overheating: {temperature:.1f}°C")
        elif temperature > 80.0:
            risk_score += 0.15

        if pressure > 200.0:
            risk_score += 0.3
            warnings.append(f"Overpressure: {pressure:.1f} psi")

        risk_score = min(1.0, risk_score)
        remaining_days = max(0, int((1.0 - risk_score) * 30))

        return {
            "risk_score": round(risk_score, 2),
            "estimated_lifetime_days": remaining_days,
            "warnings": warnings,
            "action": "IMMEDIATE SHUTDOWN" if risk_score > 0.7
                      else "Schedule maintenance" if risk_score > 0.4
                      else "Monitor normally",
        }

# Simulate a motor degrading over time
engine = PredictiveMaintenanceEngine()
motor = {"vibration": 2.0, "temperature": 60.0, "pressure": 100.0}

print("Motor degradation simulation (10 rounds):")
for i in range(10):
    motor["vibration"] += random.uniform(0.5, 1.5)
    motor["temperature"] += random.uniform(1.0, 3.0)
    motor["pressure"] += random.uniform(5.0, 15.0)

    result = engine.analyze(**motor)
    print(f"Round {i+1}: Risk={result['risk_score']}, "
          f"Days remaining={result['estimated_lifetime_days']}, "
          f"Action={result['action']}")
    if result["warnings"]:
        for w in result["warnings"]:
            print(f"  ⚠ {w}")
    time.sleep(0.2)

# digital_twin.py
class DigitalTwin:
    def __init__(self, asset_name: str):
        self.name = asset_name
        self.state = {
            "temperature": 25.0,
            "pressure": 100.0,
            "flow_rate": 50.0,
            "valve_open": True,
        }
        self.history = []

    def sync_from_physical(self, sensor_data: dict):
        """Update twin with real sensor readings"""
        self.state.update(sensor_data)
        self.history.append({**self.state, "timestamp": datetime.utcnow().isoformat()})
        print(f"Twin synced: {self.state}")

    def simulate(self, what_if: dict) -> dict:
        """Run what-if simulation"""
        temp_state = {**self.state, **what_if}
        if not temp_state["valve_open"]:
            temp_state["flow_rate"] = 0.0
            temp_state["pressure"] += 20.0
        print(f"Simulation result: {temp_state}")
        return temp_state

twin = DigitalTwin("Pump-Unit-42")
twin.sync_from_physical({"temperature": 78.5, "pressure": 145.0})
twin.simulate({"valve_open": False})

import random, time, json

class IndustrialSensor:
    def __init__(self, name: str, normal_min: float, normal_max: float, critical_min: float, critical_max: float):
        self.name = name
        self.normal_min = normal_min
        self.normal_max = normal_max
        self.critical_min = critical_min
        self.critical_max = critical_max
        self.value = (normal_min + normal_max) / 2

    def read(self) -> dict:
        # Drift toward normal range
        self.value += random.uniform(-1, 1)
        self.value = max(self.critical_min, min(self.critical_max, self.value))
        status = "NORMAL"
        if self.value < self.critical_min or self.value > self.critical_max:
            status = "CRITICAL"
        elif self.value < self.normal_min or self.value > self.normal_max:
            status = "WARNING"
        return {"name": self.name, "value": round(self.value, 1), "status": status}

sensors = [
    IndustrialSensor("Vibration", 2.0, 5.0, 0.5, 8.0),
    IndustrialSensor("Temperature", 60.0, 85.0, 40.0, 100.0),
    IndustrialSensor("Pressure", 80.0, 150.0, 50.0, 200.0),
]

print("IIoT Dashboard (press Ctrl+C to stop):")
for _ in range(10):
    timestamp = time.strftime("%H:%M:%S")
    readings = [s.read() for s in sensors]
    status_line = " | ".join(f"{r['name']}: {r['value']} [{r['status']}]" for r in readings)
    print(f"[{timestamp}] {status_line}")
    time.sleep(0.5)