def getActivePriorityCount(tagPath, Priority = ['Diagnostic', 'Low','Medium','High','Critical']): ### # This function returns the number of active alarms of a specificed priority other wise all active alarms return len(getActiveAlarms(tagPath, Priority, state = ['ActiveUnacked', 'ActiveAcked'], sortByTime = False)) def getActiveCriticalCount(tagPath, Priority = ['Critical']): ### # This function returns the number of Active Critical Alarms return len(getActiveAlarms(tagPath, Priority, state = ['ActiveUnacked', 'ActiveAcked'], sortByTime = False)) def getActiveHighCount(tagPath, Priority = ['High']): ### # This function returns the number of Active High Alarms return len(getActiveAlarms(tagPath, Priority, state = ['ActiveUnacked', 'ActiveAcked'], sortByTime = False)) def getActiveMediumCount(tagPath, Priority = ['Medium']): ### # This function returns the number of Active Medium Alarms return len(getActiveAlarms(tagPath, Priority, state = ['ActiveUnacked', 'ActiveAcked'], sortByTime = False)) def getActiveLowCount(tagPath, Priority = ['Low']): ### # This function returns the number of Active Low Alarms return len(getActiveAlarms(tagPath, Priority, state = ['ActiveUnacked', 'ActiveAcked'], sortByTime = False)) def getActiveDiagCount(tagPath, Priority = ['Diagnostic']): ### # This function returns the number of Active Diagnostic Alarms return len(getActiveAlarms(tagPath, Priority, state = ['ActiveUnacked', 'ActiveAcked'], sortByTime = False)) def getActiveAlarms(tagPath, priority = ['Diagnostic', 'Low','Medium','High','Critical'], state = ['ActiveUnacked', 'ActiveAcked'], sortByTime = False): ### # This function grabs a list of active alarms to use in a table # tagpath - path of device alarms to collect of type string # priority - list of priority types to collect # state - a list of states to collect # sort - option to sort by timestamp ### alarmList = [] if tagPath: i = tagPath.find(']') path = '*{}/*'.format(tagPath[i + 1:]) results = system.alarm.queryStatus(source = [path], priority = priority, state = state) for each in results: alarm = { 'Time': each.get('eventTime'), 'Alarm': each.getDisplayPath(), 'Notes': each.getNotes(), 'Priority': str(each.getPriority()), 'Alarm Priority': str(Global.AlmTables.getCustomPriority(str(each.getPriority()))), 'Name': each.getName(), 'id': each.getId(), 'Source': each.getSource(), 'State': each.get('State') } alarmList.append(alarm) if sortByTime: return sorted(alarmList, key=lambda d: (d['Time'], priorityWeight[d['Priority']]), reverse=True) else: priorityWeight = {'Diagnostic': 0, 'Low': 1, 'Medium': 2, 'High': 3, 'Critical': 4} return sorted(alarmList, key=lambda d: (priorityWeight[d['Priority']], d['Time']), reverse=True) return alarmList def getCustomPriority(priority, customList = ['Diagnostic', 'Info', 'Undefined', 'Warning', 'Critical']): ### # This function converts standard alarm priorities to custom defined priorites # priority - priority or type string # customList - list of custom priorites if priority == 'Diagnostic': return customList[0] if priority == 'Low': return customList[1] if priority == 'Medium': return customList[2] if priority == 'High': return customList[3] if priority == 'Critical': return customList[4] return 'UNKNOWN' def printTagAlarmCnts(tagPath): #Diag## # This function is used for quickly testing alarm counts for a tagpath # results printed to console # this function is purley for testing count1 = Global.AlmTables.getActiveCriticalCount(tagPath) count2 = Global.AlmTables.getActiveHighCount(tagPath) count3 = Global.AlmTables.getActiveMediumCount(tagPath) count4 = Global.AlmTables.getActiveLowCount(tagPath) count5 = Global.AlmTables.getActiveDiagCount(tagPath) print "Alarm total counts for:" + tagPath + " are: \n" + "Critical: " + str(count1) + "\nHigh: " + str(count2) + "\nMedium: " + str(count3) + "\nLow: " + str(count4) + "\nDiag: " + str(count5) def updateAlarmCount(tagPath): ### # This function writes the current alarm count back to the tag device tagList = ['Diagnostic', 'Low','Medium','High','Critical'] for i in tagList: count = getActivePriorityCount(tagPath, Priority = [i]) tag = tagPath + "/Alarm Data/"+ i + " Count" #tag = tagPath + "/" + i + " Count" #system.tag.writeAsync(tag, count, callback) system.tag.writeAsync(tag, count) def printThis(text = "Testing"): system.perspective.print(text) return text