Source code for

# Copyright 2011-2014 Biomedical Imaging Group Rotterdam, Departments of
# Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

from enum import Enum
from typing import Callable
from logging import Handler, LogRecord

[docs]class EventType(Enum): job_updated = 'job_updated' run_started = 'run_started' run_finished = 'run_finished' log_record_emitted = 'log_record_emitted'
EVENT_LISTENERS = {x: [] for x in EventType}
[docs]def register_listener(event_type: EventType, function: Callable[[object], None]): """ Register a listeners to a specific event type :param event_type: The EventType to listen on :param function: The callable that will be called on each event """ if event_type not in EventType: raise ValueError('Trying to register listener to invalid event type: {}'.format(event_type)) if function not in EVENT_LISTENERS[event_type]: EVENT_LISTENERS[event_type].append(function)
[docs]def remove_listener(event_type: EventType, function: Callable[[object], None]): """ Remove a listeren from a type of event :param event_type: The event type to remove the listeners from :param function: The function to remove """ while function in EVENT_LISTENERS[event_type]: EVENT_LISTENERS[event_type].remove(function)
[docs]def emit_event(event_type: EventType, data): """ Emit an event to all listeners :param event_type: The type of event to emit :param data: The data object to send along """ for listener in EVENT_LISTENERS[event_type]: listener(data)
[docs]class FastrLogEventHandler(Handler): """ Logging handler that sends the log records into the event system """
[docs] def emit(self, record: LogRecord): emit_event(EventType.log_record_emitted, record)