OT (Operational Technology)         IT (Information Technology)
────────────────────────────────     ───────────────────────────
PLCs, SCADA, DCS                     ERP, MES, Cloud, IoT platforms
Deterministic, real-time             Best-effort, flexible
Safety-critical                      Business-critical
Change-averse (uptime matters)       Agile, frequent updates
Proprietary protocols                Open standards (HTTP, MQTT)
5–25 year lifecycle                  2–5 year lifecycle
Air-gapped networks                  Internet-connected
Modbus, OPC-UA, Profibus             MQTT, REST, gRPC

IIoT bridges these two worlds ──────────────────────────────────►

# pip install pymodbus
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException
import struct

class ModbusDevice:
    """Generic Modbus TCP device abstraction."""

    def __init__(self, host, port=502, unit_id=1):
        self._client  = ModbusTcpClient(host, port=port)
        self._unit    = unit_id

    def connect(self):
        success = self._client.connect()
        print(f"Modbus {'connected' if success else 'FAILED'}: {self._client}")
        return success

    def read_holding_registers(self, address, count=1):
        """Read holding registers (function code 3)."""
        result = self._client.read_holding_registers(address, count, self._unit)
        if result.isError():
            raise ModbusException(f"Error reading reg {address}: {result}")
        return result.registers

    def read_input_registers(self, address, count=1):
        """Read input registers (function code 4) — read-only."""
        result = self._client.read_input_registers(address, count, self._unit)
        if result.isError():
            raise ModbusException(f"Error reading input reg {address}: {result}")
        return result.registers

    def write_register(self, address, value):
        """Write single holding register (function code 6)."""
        result = self._client.write_register(address, value, self._unit)
        if result.isError():
            raise ModbusException(f"Error writing reg {address}: {result}")
        return True

    def decode_float32(self, regs):
        """Decode 2 consecutive 16-bit registers into a 32-bit float."""
        raw = struct.pack('>HH', regs[0], regs[1])
        return struct.unpack('>f', raw)[0]

    def close(self):
        self._client.close()


# Example: Reading a power meter (e.g., Eastron SDM630)
class PowerMeter(ModbusDevice):
    # Eastron SDM630 register map
    REG_VOLTAGE_L1  = 0   # 2 regs, float32, Volts
    REG_CURRENT_L1  = 6   # 2 regs, float32, Amps
    REG_POWER_L1    = 12  # 2 regs, float32, Watts
    REG_ENERGY_KWH  = 342 # 2 regs, float32, kWh total import

    def read_all(self):
        return {
            "voltage_v":  round(self.decode_float32(
                              self.read_input_registers(self.REG_VOLTAGE_L1, 2)), 2),
            "current_a":  round(self.decode_float32(
                              self.read_input_registers(self.REG_CURRENT_L1, 2)), 3),
            "power_w":    round(self.decode_float32(
                              self.read_input_registers(self.REG_POWER_L1, 2)), 1),
            "energy_kwh": round(self.decode_float32(
                              self.read_input_registers(self.REG_ENERGY_KWH, 2)), 3),
        }

meter = PowerMeter("192.168.1.100", port=502, unit_id=1)
if meter.connect():
    readings = meter.read_all()
    print(readings)
    # {'voltage_v': 231.45, 'current_a': 4.832, 'power_w': 1117.2, 'energy_kwh': 1245.332}
    meter.close()

# Modbus RTU over RS-485 (USB adapter or serial port)
from pymodbus.client import ModbusSerialClient

client = ModbusSerialClient(
    method='rtu',
    port='/dev/ttyUSB0',  # or 'COM3' on Windows
    baudrate=9600,
    parity='N',
    stopbits=1,
    bytesize=8,
    timeout=1
)

if client.connect():
    # Read temperature from a Modbus RTU sensor (unit ID = 1, register 0x0001)
    result = client.read_holding_registers(0x0001, 1, unit=1)
    if not result.isError():
        raw_temp = result.registers[0]
        temperature = raw_temp / 10.0  # Typical: value in 0.1°C units
        print(f"Temperature: {temperature}°C")
    client.close()

# pip install asyncua
import asyncio
from asyncua import Client, ua

async def read_plc_data(endpoint="opc.tcp://localhost:4840"):
    async with Client(url=endpoint) as client:
        # OPC-UA uses node IDs to address data points
        # Browse the server namespace to find available nodes
        node = client.get_node("ns=2;i=1001")  # Example node ID

        # Read a single value
        value = await node.get_value()
        print(f"Machine speed: {value} RPM")

        # Read multiple nodes at once (more efficient)
        nodes = [
            client.get_node("ns=2;i=1001"),  # Speed
            client.get_node("ns=2;i=1002"),  # Temperature
            client.get_node("ns=2;i=1003"),  # Vibration
        ]
        values = await client.read_values(nodes)
        print(dict(zip(["speed", "temperature", "vibration"], values)))

        # Subscribe to data changes (push instead of poll)
        handler = SubHandler()
        sub = await client.create_subscription(500, handler)  # 500ms interval
        await sub.subscribe_data_change(nodes)

        await asyncio.sleep(30)  # Receive 30s of updates
        await sub.delete()

class SubHandler:
    def datachange_notification(self, node, val, data):
        print(f"Change → {node}: {val}")

asyncio.run(read_plc_data())

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

class PredictiveMaintenanceEngine:
    """
    Detect abnormal machine behavior using Isolation Forest.
    Train on historical 'normal' operation data.
    """
    def __init__(self, contamination=0.05):
        self.scaler = StandardScaler()
        self.model  = IsolationForest(
            contamination=contamination,  # Expected anomaly fraction
            random_state=42,
            n_estimators=100
        )
        self._trained = False

    def train(self, df_normal):
        """Train on normal operating data."""
        features = df_normal[['vibration', 'temperature', 'current', 'speed']].dropna()
        X_scaled = self.scaler.fit_transform(features)
        self.model.fit(X_scaled)
        self._trained = True
        print(f"Trained on {len(features)} samples of normal operation")

    def predict(self, df_new):
        """
        Predict anomalies on new data.
        Returns DataFrame with 'anomaly' (True/False) and 'score' columns.
        """
        if not self._trained:
            raise RuntimeError("Model not trained yet")

        features = df_new[['vibration', 'temperature', 'current', 'speed']].fillna(method='ffill')
        X_scaled = self.scaler.transform(features)

        # Isolation Forest: -1 = anomaly, 1 = normal
        predictions = self.model.predict(X_scaled)
        scores      = self.model.score_samples(X_scaled)

        df_result = df_new.copy()
        df_result['anomaly'] = predictions == -1
        df_result['anomaly_score'] = -scores  # Higher = more anomalous

        anomalies = df_result[df_result['anomaly']]
        print(f"Anomalies detected: {len(anomalies)}/{len(df_result)} "
              f"({len(anomalies)/len(df_result)*100:.1f}%)")

        return df_result

# Example usage
normal_data = pd.DataFrame({
    'vibration':   np.random.normal(2.0, 0.3, 5000),
    'temperature': np.random.normal(65.0, 5.0, 5000),
    'current':     np.random.normal(15.0, 1.5, 5000),
    'speed':       np.random.normal(1450, 20, 5000),
})

# Simulate new data with some anomalies injected
new_data = normal_data.copy().iloc[:100]
new_data.loc[40:45, 'vibration']   = 8.0   # Imbalance spike
new_data.loc[70:72, 'temperature'] = 120.0  # Overheating
new_data.loc[85:88, 'current']     = 35.0   # Motor overload

engine = PredictiveMaintenanceEngine(contamination=0.05)
engine.train(normal_data)
results = engine.predict(new_data)

critical = results[results['anomaly_score'] > 0.7]
print(f"\nCritical anomalies (score > 0.7):")
print(critical[['vibration', 'temperature', 'current', 'anomaly_score']])

Field Level (OT)
  PLCs, sensors, actuators, VFDs
  Protocols: Modbus RTU, Profibus, HART, 4-20mA
        │
        │ (Serial RS-485 / Ethernet)
        ▼
Control Level
  SCADA / DCS / HMI
  OPC-UA server (exposes all data as addressable nodes)
        │
        │ OPC-UA / MQTT bridge
        ▼
Edge Level
  Industrial PC / Edge gateway
  Data filtering, ML inference, local alerting
  Protocol translation: Modbus → MQTT
        │
        │ MQTT / HTTPS (encrypted)
        ▼
Cloud Level
  InfluxDB (time-series), Data Lake (S3)
  ML training (SageMaker, Azure ML)
  Grafana dashboards
  ERP integration (SAP, Oracle)