Skip to content
Snippets Groups Projects
Select Git revision
  • 8a410e5c8b1d4642c427354acd95f97c865e9406
  • main default protected
  • 65_async_query
  • 169_codemeta
  • issue/152_cleaning_up
  • adex-main protected
  • sdc380-aladin-cone-search
  • sdc-222_multi_archive_query_review
  • 69_add_diracIAM
  • dev-nico
  • dev-dirac
  • acceptance
  • dev-zooniverse
13 results

SampGrid.js

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    alerta_lofar.py 1.92 KiB
    import os
    import json
    import logging
    
    from alerta.plugins import PluginBase
    import alerta.models.alarms.isa_18_2 as isa_18_2
    
    LOG = logging.getLogger()
    
    
    class EnhanceLOFAR(PluginBase):
        """
        Plugin for enhancing alerts with LOFAR-specific information
        """
    
        @staticmethod
        def _fix_severity(alert):
            """
              Force conversion of severity to ISA 18.2 model, to allow Alerta to parse the alert.
    
              For example, the 'prometheus' webhook by default uses the 'warning' severity,
              but also users might specify a non-existing severity level.
            """
                
            if alert.severity not in isa_18_2.SEVERITY_MAP:
                # Save original severity
                alert.attributes['unparsableSeverity'] = alert.severity
    
                translation = {
                    "normal":   isa_18_2.OK,
                    "ok":       isa_18_2.OK,
                    "cleared":  isa_18_2.OK,
                    "warning":  isa_18_2.LOW,
                    "minor":    isa_18_2.MEDIUM,
                    "major":    isa_18_2.HIGH,
                    "critical": isa_18_2.CRITICAL,
                }
    
                alert.severity = translation.get(alert.severity.lower(), isa_18_2.MEDIUM)
    
        def pre_receive(self, alert, **kwargs):
            self._fix_severity(alert)
    
            # Parse LOFAR-specific fields
            for tag in alert.tags:
                try:
                    key, value = tag.split("=", 1)
                except ValueError:
                    continue
    
                if key == "device":
                    alert.attributes['lofarDevice'] = value
    
                if key == "name":
                    alert.attributes['lofarAttribute'] = value
    
                if key == "station":
                    alert.resource = value
    
            return alert
    
        def post_receive(self, alert, **kwargs):
            return
    
        def status_change(self, alert, status, text, **kwargs):
            return
    
        def take_action(self, alert, action, text, **kwargs):
            raise NotImplementedError