Module enrgdaq.daq.jobs.handle_alerts

Classes

class DAQJobHandleAlerts (config: DAQJobConfig,
supervisor_info: SupervisorInfo | None = None,
instance_id: int | None = None,
message_in: Any = None,
message_out: Any = None,
raw_config: str | None = None,
zmq_xpub_url: str | None = None,
zmq_xsub_url: str | None = None)
Expand source code 
class DAQJobHandleAlerts(DAQJob):
    """
    DAQJobHandleAlerts is a job that stores alert messages (DAQJobMessageAlert).
    """

    allowed_message_in_types = [DAQJobMessageAlert]
    config_type = DAQJobHandleAlertsConfig
    config: DAQJobHandleAlertsConfig

    @override
    def start(self):
        while not self._has_been_freed:
            time.sleep(1)

    def handle_message(self, message: DAQJobMessage) -> bool:
        if not super().handle_message(message):
            return False
        if not isinstance(message, DAQJobMessageAlert):
            return True

        keys = [
            "timestamp",
            "daq_job",
            "severity",
            "message",
        ]

        assert message.daq_job_info is not None
        data_to_send = [
            [
                get_unix_timestamp_ms(message.date),
                message.daq_job_info.daq_job_type,
                message.alert_info.severity,
                message.alert_info.message,
            ]
        ]

        self._put_message_out(
            DAQJobMessageStoreTabular(
                store_config=self.config.store_config,
                keys=keys,
                data=data_to_send,
            )
        )

        return True

DAQJobHandleAlerts is a job that stores alert messages (DAQJobMessageAlert).

Ancestors

Class variables

var allowed_message_in_types
var config : DAQJobHandleAlertsConfig
var config_type : type[DAQJobConfig]

Configuration class for DAQJobHandleAlerts. Inherits from StorableDAQJobConfig.

Methods

def start(self)
Expand source code 
@override
def start(self):
    while not self._has_been_freed:
        time.sleep(1)

Inherited members

class DAQJobHandleAlertsConfig (store_config: DAQJobStoreConfig,
*,
daq_job_type: str,
verbosity: LogVerbosity = LogVerbosity.INFO,
daq_job_unique_id: str | None = None,
use_shm_when_possible: bool = True,
topics_to_subscribe: list[str] = <factory>)
Expand source code 
class DAQJobHandleAlertsConfig(StorableDAQJobConfig):
    """
    Configuration class for DAQJobHandleAlerts.
    Inherits from StorableDAQJobConfig.
    """

    pass

Configuration class for DAQJobHandleAlerts. Inherits from StorableDAQJobConfig.

Ancestors