Has anyone found a reasonably performant script or expression that, for a given tagPath, returns the highest alarm level for any alarm on any tag underneath that path?
I've made a modified version of the alarm summary script I found on here at one time to do just that. I literally made the change yesterday to it, although mine has a lot of other stuff I'm doing too, so let me see if I can clean it up, but essentially it's a Gateway script that runs like once a second and keeps track of all the alarms and dumps that in a global variable, then in each folder I have an "area" tag that parses that global tag and finds all the alarm stats for tags starting with the same starting path.
Here's my Gateway.AlarmSummary script:
logger = system.util.getLogger(system.util.getProjectName() + '.' + __name__)
alarmSummaryGlobal = {}
def getAllAlarms():
'''
Description:
Query all alarms using system function 'system.alarm.queryStatus' and store in
a global vaiable dictionary, 'alarmSummaryGlobal'. This function calls 'system.alarm.queryStatus'
only one time vs calling it several times per instance, which can cause higher than normal CPU usage
when several instances are used.
This function is typically called from a gateway event, like a timer event.
The frequency of this event can vary depending on the specific project needs, but has been
tested using a one second delay with very good results.
Args:
None
Returns:
None
History:
Ver. Date Author Comment
---------------------------------------------------------------------------------------
1.0 2022-10-20 James Landwerlen Initial
1.1 2022-10-29 James Landwerlen Added shelving
1.2 2023-02-13 Michael Flagler Split out alarm styles
'''
global alarmSummaryGlobal
global testInt
alarmSummary = {}
state = ["ActiveUnacked", "ActiveAcked", "ClearUnacked"]
# Get an Alarm Query Result for all alarms, including alarms that are shelved
alarms = system.alarm.queryStatus(state=state, includeShelved=True)
for alarm in alarms:
# Iterate through all alarms and place into the global dictionary 'alarmSummaryGlobal' to be called later
# on by each UDT instance
source = alarm.source
# Build a typical tag path [MyTagProvider]MyTag
tagPath = '[' + alarm.source.getPathComponent('prov') +']' + alarm.source.getPathComponent('tag')
name = alarm.name
state = alarm.state.name()
priority = alarm.priority.name()
# If the alarm is shelved, flag as shelved and set state/priority to Shelved so the counts
# will be ignored later on
if alarm.isShelved():
state = priority = 'Shelved'
alarmSummary[source] = {'tagPath':tagPath, 'name':name, 'state':state, 'priority':priority, 'shelved':alarm.isShelved()}
alarmSummaryGlobal = alarmSummary
#testInt += 1
#logger.warn(str(testInt))
def getUDTInstances(tp, alarmPath):
'''
Desc:
This function is called from each UDT instance using `runScript` in the element `summary` and
returns a json with alarm information specific to the path.
The argument `tp` is populated automatically in each UDT instance using a value change script
in the element `TagProvider`.
Args:
tp (str): Tag Provider
alarmPath (str): Path to the UDT instance parent folder. Root will be blank.
Returns:
str: JSON encoded string of UDT instances
History:
Ver. Date Author Comment
---------------------------------------------------------------------------------------
1.0 2022-10-20 James Landwerlen Initial
1.1 2022-10-29 James Landwerlen Added shelving
1.2 2023-02-13 Michael Flagler Split out alarm styles
1.3 2024-07-24 Michael Flagler Added priorityUnack
'''
global alarmSummaryGlobal
#global testInt
#logger.warn(str(testInt))
alarmPath = '[' + tp + ']' + alarmPath
alarmPriorities = {'': -2, 'Shelved':-1, 'Diagnostic':0, 'Low':1, 'Medium':2, 'High':3, 'Critical':4}
statePriorities = {'': -2, 'Shelved':-1, 'ClearAcked':0, 'ClearUnacked':1, 'ActiveAcked':2, 'ActiveUnacked':3}
priorityCounts = {'Shelved':0, 'Diagnostic': 0, 'Low': 0, 'Medium': 0, 'High': 0, 'Critical': 0}
stateCounts = {'Shelved':0, 'ClearAcked':0, 'ClearUnacked':0, 'ActiveAcked':0, 'ActiveUnacked':0}
priority = ''
priorityUnack = ''
state = ''
styleClass = ''
hasUnackAlarm = False
resultDict = {'counts': priorityCounts, 'priority': '', 'state': '', 'class': '', 'hasUnackAlarm': ''}
alarmPrioritiesList = []
highestPriorityUnack = -2
#logger.warn(str(len(alarmSummaryGlobal)))
for alarm in alarmSummaryGlobal:
tagPath = alarmSummaryGlobal[alarm]['tagPath']
# If alarm found within the summary then get counts and priorities
# The tagPath must match exactly starting from the left, this prevents partial matches
if tagPath.startswith(alarmPath):
# Create a list of alarm priorities, so we can later find the highest priority as well as state for that priority
alarmPriority = alarmSummaryGlobal[alarm]['priority']
alarmState = alarmSummaryGlobal[alarm]['state']
alarmPrioritiesList.append([alarmPriorities[alarmPriority], statePriorities[alarmState]])
shelved = alarmSummaryGlobal[alarm]['shelved']
if not shelved and (alarmState in ['ClearedUnacked', 'ActiveUnacked']):
highestPriorityUnack = max(highestPriorityUnack, alarmPriorities[alarmPriority])
hasUnackAlarm = True
priorityCounts[alarmPriority] += 1
stateCounts[alarmState] += 1
if alarmPrioritiesList:
# Get max priority/state combination from list of alarms
# Here, we're comparing the priority first, and if the same, comparing the state to determine highest/max
alarmListMax = max(alarmPrioritiesList, key=lambda item: (item[0], item[1]))
priority = alarmPriorities.keys()[alarmPriorities.values().index(alarmListMax[0])]
state = statePriorities.keys()[statePriorities.values().index(alarmListMax[1])]
priorityUnack = alarmPriorities.keys()[alarmPriorities.values().index(highestPriorityUnack)]
styleClass = getAlarmStyle(str(priority), str(state))
#Take the results and place in a dictionary
resultDict = {
'priorityCounts': priorityCounts, # A dicionary holding counts for each priority
'stateCounts': stateCounts, # A dicionary holding counts for each state
'priority': priority, # The highest priority
'state': state, # The most important state
'class': styleClass, # The style class for indication purposes
'hasUnackAlarm': hasUnackAlarm, # At least one unacknowledged alarm exists
'priorityUnack': priorityUnack # Highest priority of all unacked alarms
}
json = system.util.jsonEncode(resultDict, 0)
return json
def getAlarmStyle(priority, state):
'''
Description:
This function is called from each UDT instance using `runScript` in the element `summary` and
returns a json with alarm information specific to the path.
Args:
priority (str): Priority name (Shelved, Diagnostic, Low, Medium, High, Critical)
state (str): State name (Shelved, ClearAcked, ClearUnacked, ActiveAcked, ActiveUnacked)
History:
Ver. Date Author Comment
---------------------------------------------------------------------------------------
1.0 2022-10-20 James Landwerlen Initial
1.1 2022-10-29 James Landwerlen Added shelving
1.2 2023-02-13 Michael Flagler Split out alarm styles
'''
#These styles were not part of the resource, but any style can be added and will be selected based on the state and priority
styleStates = { 'DiagnosticActiveUnacked': 'common/navigation/diagnostic-unacked',
'DiagnosticActiveAcked': 'common/navigation/diagnostic-acked',
'LowActiveUnacked': 'common/navigation/low-unacked',
'LowActiveAcked': 'common/navigation/low-acked',
'MediumActiveUnacked': 'common/navigation/medium-unacked',
'MediumActiveAcked': 'common/navigation/medium-acked',
'HighActiveUnacked': 'common/navigation/high-unacked',
'HighActiveAcked': 'common/navigation/high-acked',
'CriticalActiveUnacked': 'common/navigation/critical-unacked',
'CriticalActiveAcked': 'common/navigation/critical-acked'
}
return styleStates.get(priority + state)
Which I have a gateway timer script running at 1000ms:
Gateway.AlarmSummary.getAllAlarms()
And I've attached the export of my UDT for the AlarmSummary tag which will give both the highest priority alarm, and the highest unacked alarm priority, plus a bunch of other nice stats at that folder level the tag is in or any subfolder.
AlarmSummaryUDT.json (6.0 KB)
2 Likes
I put something on the exchange that may work for you, Ignition Exchange | Inductive Automation, which I think Michael edited to suit his needs. @nminchin does something similar. I haven't updated the exchange the resource, but just as Nick does now, I use writeBlocking to update the instances now vs expressions in each UDT instance, that seems to be a lot more performant.
2 Likes
Just be aware that this will add all of these writes currently into the audit log, which is fun 
This is where 10's of GBs come from in my audit log at one site.. I've been meaning to setup a script to delete the useless entries periodically, but at this stage there are literally 10's of millions of these records that it would likley crash the database or lock stuff for a long time
2 Likes
I was thinking about modifying this to write to a globalVarMap() but have been advised that's probably not even performant enough to run on our hardware. Hoping this is built-in when 8.3 drops.
And then reading this directly from the View and filtering to the tagpath needed?
This is essentially what I used to do, except instead of from the View, I did it in a runscript within expression tags added into the various folder levels I needed summaries for. I moved away from these expression tags though in favour of a single script to update all alarm summary tags added to the various folder levels.
Then the Views simply bind to the area alarm summary tags for their display
Yep, you're right, I kept thinking @nminchin provided it, but I kept your name in the script, so when I went to pull it up, I knew that wasn't right, but I remembered @nminchin working on something similar.
I don't know that I'd want to do writes to all those tags vs the bindings as they are now. Is it really that much better performance, and why not writeAsync instead of writeBlocking? I would think the tag writes, audit logs, etc would be less performant than the way it is now.
Depends how many summary tags you've got. I think I have 10 or so per alarm summary UDT instance, so alarm summary UDT instance tags add up quickly for a large 400k tag project. I don't recall how many i've got, but it's something like 50-100, so 500-1000 tags. If all of those tags were calling runScript, that's a lot of processes to run (which I assume is wildly inefficient @pturmel @PGriffith ?), as opposed to just one periodic script that updates them all via a single tag.write* call 
2 Likes
I've only got about a dozen or so instances of my summary tag. but runscript is only ran once per UDT instance tag then using jsonGet expression tags to pull the data out of the document tag.