180 lines
6.9 KiB
Python
180 lines
6.9 KiB
Python
# alarm_simulator.py — Build-a-Thon alarm activity generator (Jython, Ignition 8.3)
|
|
#
|
|
# Drives the Boolean memory tags in [default]BuildathonSim/ (import
|
|
# simulation_tags.json first). Designed to be called every 5 seconds:
|
|
#
|
|
# RECOMMENDED — Gateway timer script (survives Designer close):
|
|
# 1. Designer > Project Library: create a script named "alarmsim", paste this file.
|
|
# 2. Project > Gateway Events > Timer: new script, Delay 5000 ms,
|
|
# Fixed Delay, Dedicated thread. Body: alarmsim.tick()
|
|
# 3. Save the project. Alarm activity starts immediately and runs forever.
|
|
#
|
|
# ALTERNATIVE — Designer Script Console (blocks the console while running):
|
|
# Paste this whole file, then add at the bottom:
|
|
# run_console(minutes=30)
|
|
#
|
|
# Patterns produced (all timestamps land in the alarm journal):
|
|
# - Baseline: random alarms across 5 areas / 3 priorities, active 30 s - 5 min,
|
|
# averaging a few activations per minute
|
|
# - Chattering: *_Chatter tags re-trigger every 30-60 s (short 5-15 s actives)
|
|
# - Standing: *_Standing tags go active on the first tick and never clear
|
|
# - Fleeting: *_Fleeting tags are active < 10 s
|
|
# - Flood: roughly every 30-60 min (or on demand via force_flood()), one area
|
|
# bursts 20-40 activations over 10 minutes
|
|
#
|
|
# Manual helpers (Script Console, gateway scope via system.util.sendRequest not
|
|
# needed — just call from a console if running console mode, or temporarily from
|
|
# the timer script):
|
|
# alarmsim.force_flood() -> start a flood burst now
|
|
# alarmsim.reset() -> clear all sim tags and internal state
|
|
|
|
import random
|
|
import time
|
|
|
|
BASE = "[default]BuildathonSim"
|
|
|
|
BASELINE = {
|
|
"Intake": ["Pump1_Fault", "Pump2_Fault", "ScreenDiffPressHigh", "InletFlowLow",
|
|
"ValveV101_FailToOpen", "WetWellLevelHigh", "SampleTempHigh", "PowerMonitorAlarm"],
|
|
"BoilerHouse": ["FeedPumpA_Fault", "FeedPumpB_Fault", "SteamPressHigh", "DrumLevelLow",
|
|
"StackTempHigh", "FuelGasPressLow", "EconomizerDPHigh", "BlowdownConductivityHigh"],
|
|
"Packaging": ["LabelerFault", "CapperTorqueLow", "FillerLevelDeviation", "ConveyorOverload",
|
|
"CaseSealerJam", "PrinterInkLow", "RejectRateHigh", "GuardDoorOpen"],
|
|
"Utilities": ["AirCompA_Fault", "AirHeaderPressLow", "ChillerTripped", "CoolingTowerVibration",
|
|
"GlycolTempHigh", "N2PressLow", "UPSOnBattery", "WaterSoftenerFault"],
|
|
"TankFarm": ["T101_LevelHigh", "T102_LevelHigh", "T103_TempHigh", "TransferPumpFault",
|
|
"ContainmentSumpHigh", "VaporRecoveryFault", "T104_PressHigh", "ManifoldLeakDetect"],
|
|
}
|
|
|
|
CHATTER = [
|
|
"Intake/ConveyorJam_Chatter",
|
|
"Packaging/ShrinkTunnelTemp_Chatter",
|
|
"Utilities/AirDryerDewpoint_Chatter",
|
|
]
|
|
STANDING = [
|
|
"BoilerHouse/DrumLevelXmtrFail_Standing",
|
|
"TankFarm/T105_LevelXmtrFail_Standing",
|
|
]
|
|
FLEETING = [
|
|
"Intake/CommsGlitch_Fleeting",
|
|
"BoilerHouse/FlameScannerFlicker_Fleeting",
|
|
"Packaging/EStopPulse_Fleeting",
|
|
"TankFarm/PumpSealFlushLow_Fleeting",
|
|
]
|
|
|
|
# Tuning (per 5 s tick)
|
|
BASELINE_PROB = 0.30 # ~3-4 baseline activations/min across the plant
|
|
FLEETING_PROB = 0.08 # ~1 fleeting alarm/min
|
|
FLOOD_START_PROB = 0.0025 # expected flood every ~30-60 min
|
|
FLOOD_DURATION = 600 # 10 minutes
|
|
FLOOD_EVENT_PROB = 0.35 # per tick during flood -> ~25 activations / 10 min
|
|
|
|
|
|
def _path(rel):
|
|
return "%s/%s" % (BASE, rel)
|
|
|
|
|
|
def _state():
|
|
g = system.util.getGlobals()
|
|
if "buildathon_alarmsim" not in g:
|
|
g["buildathon_alarmsim"] = {
|
|
"clears": {}, # tag path -> epoch seconds to write False
|
|
"chatter_next": {}, # chatter path -> epoch seconds of next re-trigger
|
|
"flood": None, # {"area": name, "until": epoch} while flooding
|
|
"standing_set": False,
|
|
}
|
|
return g["buildathon_alarmsim"]
|
|
|
|
|
|
def _write(pairs):
|
|
if pairs:
|
|
system.tag.writeBlocking([p for p, v in pairs], [v for p, v in pairs])
|
|
|
|
|
|
def tick():
|
|
st = _state()
|
|
now = time.time()
|
|
writes = []
|
|
|
|
# 1. Standing alarms: activate once, never clear
|
|
if not st["standing_set"]:
|
|
writes += [(_path(p), True) for p in STANDING]
|
|
st["standing_set"] = True
|
|
|
|
# 2. Process scheduled clears
|
|
for path in list(st["clears"].keys()):
|
|
if now >= st["clears"][path]:
|
|
writes.append((path, False))
|
|
del st["clears"][path]
|
|
|
|
# 3. Baseline activity: random tag, active 30 s - 5 min
|
|
if random.random() < BASELINE_PROB:
|
|
area = random.choice(list(BASELINE.keys()))
|
|
path = _path("%s/%s" % (area, random.choice(BASELINE[area])))
|
|
if path not in st["clears"]:
|
|
writes.append((path, True))
|
|
st["clears"][path] = now + random.uniform(30, 300)
|
|
|
|
# 4. Chattering: re-trigger every 30-60 s, active 5-15 s each time
|
|
for rel in CHATTER:
|
|
path = _path(rel)
|
|
nxt = st["chatter_next"].get(path, 0)
|
|
if now >= nxt and path not in st["clears"]:
|
|
writes.append((path, True))
|
|
st["clears"][path] = now + random.uniform(5, 15)
|
|
st["chatter_next"][path] = now + random.uniform(30, 60)
|
|
|
|
# 5. Fleeting: active 2-8 s
|
|
if random.random() < FLEETING_PROB:
|
|
path = _path(random.choice(FLEETING))
|
|
if path not in st["clears"]:
|
|
writes.append((path, True))
|
|
st["clears"][path] = now + random.uniform(2, 8)
|
|
|
|
# 6. Flood burst: one area, 15+ activations inside 10 minutes
|
|
if st["flood"] is None:
|
|
if random.random() < FLOOD_START_PROB:
|
|
_start_flood(st, now)
|
|
elif now >= st["flood"]["until"]:
|
|
st["flood"] = None
|
|
else:
|
|
if random.random() < FLOOD_EVENT_PROB:
|
|
area = st["flood"]["area"]
|
|
path = _path("%s/%s" % (area, random.choice(BASELINE[area])))
|
|
# During a flood, alarms clear fast and may re-trigger, inflating the
|
|
# event rate the way a real upset does. Overwrite any pending clear.
|
|
writes.append((path, True))
|
|
st["clears"][path] = now + random.uniform(10, 45)
|
|
|
|
_write(writes)
|
|
|
|
|
|
def _start_flood(st, now):
|
|
area = random.choice(list(BASELINE.keys()))
|
|
st["flood"] = {"area": area, "until": now + FLOOD_DURATION}
|
|
system.util.getLogger("alarmsim").info("Flood burst started in area: %s" % area)
|
|
|
|
|
|
def force_flood():
|
|
"""Start a flood burst immediately (handy for demo/testing)."""
|
|
_start_flood(_state(), time.time())
|
|
|
|
|
|
def reset():
|
|
"""Clear all simulation tags and internal state."""
|
|
g = system.util.getGlobals()
|
|
g.pop("buildathon_alarmsim", None)
|
|
paths = []
|
|
for area, names in BASELINE.items():
|
|
paths += [_path("%s/%s" % (area, n)) for n in names]
|
|
paths += [_path(p) for p in CHATTER + STANDING + FLEETING]
|
|
system.tag.writeBlocking(paths, [False] * len(paths))
|
|
|
|
|
|
def run_console(minutes=30):
|
|
"""Script Console runner: calls tick() every 5 s for `minutes`. Blocks the console."""
|
|
end = time.time() + minutes * 60
|
|
while time.time() < end:
|
|
tick()
|
|
time.sleep(5)
|