SCADA Water Automation Development Stack

CREATE TABLE scada_tag (
    tag_id BIGSERIAL PRIMARY KEY,
    tag_name TEXT UNIQUE NOT NULL,
    site_id TEXT NOT NULL,
    equipment_id TEXT,
    data_type TEXT NOT NULL,
    unit TEXT,
    description TEXT,
    is_active BOOLEAN DEFAULT TRUE
);

CREATE TABLE historian_value (
    tag_id BIGINT REFERENCES scada_tag(tag_id),
    ts TIMESTAMPTZ NOT NULL,
    value_num DOUBLE PRECISION,
    value_text TEXT,
    quality_code TEXT NOT NULL DEFAULT 'GOOD',
    source TEXT,
    PRIMARY KEY (tag_id, ts)
);

SELECT
    date_trunc('hour', hv.ts) AS hour_bucket,
    AVG(hv.value_num) AS avg_pressure_psi,
    MIN(hv.value_num) AS min_pressure_psi,
    MAX(hv.value_num) AS max_pressure_psi
FROM historian_value hv
JOIN scada_tag t ON t.tag_id = hv.tag_id
WHERE t.tag_name = 'PS101_DISCHARGE_PRESSURE_PSI'
  AND hv.quality_code = 'GOOD'
  AND hv.ts > NOW() - INTERVAL '24 hours'
GROUP BY hour_bucket
ORDER BY hour_bucket;

# Bad rule
ALLOW ANY FROM IT_NETWORK TO PLC_NETWORK

# Better rule concept
ALLOW TCP 443 FROM VPN_USERS TO JUMP_SERVER
ALLOW RDP FROM JUMP_SERVER TO ENGINEERING_WORKSTATION
ALLOW OPCUA 4840 FROM SCADA_SERVER TO OPCUA_SERVER
DENY  ANY FROM INTERNET TO PLC_NETWORK
DENY  ANY FROM IT_NETWORK TO PLC_NETWORK
LOG   ALL DENIED TRAFFIC

CREATE TABLE scada_audit_log (
    audit_id BIGSERIAL PRIMARY KEY,
    event_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    username TEXT NOT NULL,
    source_ip INET,
    action TEXT NOT NULL,
    asset TEXT,
    old_value TEXT,
    new_value TEXT,
    result TEXT NOT NULL,
    notes TEXT
);

{
  "site_id": "PS101",
  "site_type": "booster_pump_station",
  "equipment": [
    {"id": "PUMP_1", "type": "pump", "motor_hp": 150, "vfd": true},
    {"id": "PUMP_2", "type": "pump", "motor_hp": 150, "vfd": true},
    {"id": "FLOW_METER_1", "type": "mag_meter", "unit": "MGD"},
    {"id": "PRESSURE_TX_1", "type": "pressure_transmitter", "unit": "PSI"}
  ]
}

point_name,io_type,signal,unit,equipment,description
PUMP1_RUN_FB,DI,24VDC,,PUMP_1,Pump 1 run feedback
PUMP1_START_CMD,DO,24VDC,,PUMP_1,Pump 1 start command
DISCH_PRESSURE,AI,4-20mA,PSI,HEADER,Discharge header pressure
PUMP1_SPEED_CMD,AO,4-20mA,%,PUMP_1,VFD speed command

(* Simplified lead/lag pressure control *)
LowPressure := Discharge_Pressure_PSI < Low_Pressure_SP;
VeryLowPressure := Discharge_Pressure_PSI < Very_Low_Pressure_SP;

IF Auto_Mode AND LowPressure THEN
    LeadPump_Start := TRUE;
END_IF;

IF Auto_Mode AND VeryLowPressure THEN
    LagPump_Start := TRUE;
END_IF;

IF Discharge_Pressure_PSI > Stop_Pressure_SP THEN
    LeadPump_Start := FALSE;
    LagPump_Start := FALSE;
END_IF;

from pymodbus.client import ModbusTcpClient

client = ModbusTcpClient("10.10.20.15", port=502, timeout=3)
client.connect()

result = client.read_holding_registers(address=0, count=2, slave=1)
if result.isError():
    raise RuntimeError("Modbus read failed")

flow_mgd = result.registers[0] / 100.0
pressure_psi = result.registers[1] / 10.0
print({"flow_mgd": flow_mgd, "pressure_psi": pressure_psi})

client.close()

import json, time
import paho.mqtt.client as mqtt

client = mqtt.Client(client_id="ps101_rtu")
client.connect("10.10.50.20", 1883, 60)

payload = {
    "timestamp": int(time.time() * 1000),
    "metrics": [
        {"name": "Pump1_Run", "value": True, "type": "Boolean"},
        {"name": "Discharge_Pressure_PSI", "value": 64.2, "type": "Float"},
        {"name": "Flow_MGD", "value": 3.18, "type": "Float"}
    ]
}

client.publish("spBv1.0/water/DDATA/PS101/RTU01", json.dumps(payload), qos=1)
client.disconnect()

from opcua import Client

client = Client("opc.tcp://10.10.30.12:4840")
client.connect()

try:
    pressure_node = client.get_node("ns=2;s=PS101.DischargePressurePSI")
    pressure_psi = pressure_node.get_value()
    print("Pressure:", pressure_psi)
finally:
    client.disconnect()

{
  "tag_name": "PS101_DISCHARGE_PRESSURE_PSI",
  "site_id": "PS101",
  "equipment_id": "DISCHARGE_HEADER",
  "data_type": "float",
  "unit": "psi",
  "source_protocol": "OPC_UA",
  "quality_code": "GOOD",
  "alarm_limits": {
    "low_low": 35,
    "low": 45,
    "high": 95,
    "high_high": 110
  }
}

CREATE TABLE scada_tag (
    tag_id BIGSERIAL PRIMARY KEY,
    tag_name TEXT UNIQUE NOT NULL,
    site_id TEXT NOT NULL,
    equipment_id TEXT,
    unit TEXT,
    description TEXT
);

CREATE TABLE historian_value (
    tag_id BIGINT REFERENCES scada_tag(tag_id),
    ts TIMESTAMPTZ NOT NULL,
    value_num DOUBLE PRECISION,
    value_text TEXT,
    quality_code TEXT NOT NULL,
    source TEXT,
    PRIMARY KEY (tag_id, ts)
);

<div class="status-card normal">
  <div class="label">Pump 1</div>
  <div class="value">RUNNING</div>
  <div class="meta">64.2 PSI · 3.18 MGD · Quality: GOOD</div>
</div>

<style>
.status-card { padding: 14px; border-radius: 14px; border: 1px solid #ddd; }
.status-card.normal { background: #f0fdf4; color: #166534; }
.status-card.alarm { background: #fef2f2; color: #991b1b; }
.label { font-size: 12px; opacity: .75; }
.value { font-size: 22px; font-weight: 800; }
.meta { font-size: 12px; margin-top: 4px; }
</style>

def evaluate_pressure_alarm(pressure_psi: float, quality: str) -> dict:
    if quality != "GOOD":
        return {"active": True, "priority": "HIGH", "alarm": "PRESSURE_DATA_BAD"}

    if pressure_psi < 35:
        return {"active": True, "priority": "CRITICAL", "alarm": "LOW_PRESSURE_CRITICAL"}

    if pressure_psi < 45:
        return {"active": True, "priority": "HIGH", "alarm": "LOW_PRESSURE_WARNING"}

    return {"active": False}

INSERT INTO scada_audit_log
(username, source_ip, action, asset, old_value, new_value, result)
VALUES
('operator01', '10.20.5.44', 'SETPOINT_CHANGE',
 'GST202_LOW_LEVEL_SP', '14.0', '13.5', 'SUCCESS');

{
  "event_type": "LOW_PRESSURE",
  "first_questions": [
    "Which pressure zone is affected?",
    "Are pumps running and available?",
    "Are tanks dropping?",
    "Is there abnormal high flow?",
    "Is the data quality good or stale?"
  ],
  "operator_actions": [
    "Verify SCADA alarm and historian trend",
    "Check pump station and tank status",
    "Contact field crew for confirmation",
    "Escalate if public health or service risk exists"
  ]
}