<reponame>timoklimmer/neanno<filename>neanno/utils/signals.py
"""Defines helper functions for working with signals emitted during a ParallelWorker job."""

import time


def emit_top_header(text, signals):
    """ Emits a top header."""
    signals.message.emit(text, True)
    signals.message.emit(("=") * len(text), True)
    signals.message.emit("", True)


def emit_sub_header(text, signals):
    """ Emits a sub header."""
    signals.message.emit(text, True)
    signals.message.emit("-" * len(text), True)


def emit_message(message, signals):
    """Emits a simple message."""
    signals.message.emit(message, True)


def emit_image(image_bytes, image_format, signals):
    """Emits the passed image."""
    signals.image.emit(image_bytes, image_format)

def emit_partial_message(message, signals):
    """Emits the given message but does not advance to the next line."""
    signals.message.emit(message, False)


def emit_new_line(signals):
    """Emits a new line to improve readability."""
    signals.message.emit("", True)


def emit_start_time(signals):
    """Emits the start time and returns it for later when the duration needs to be calculated."""
    start_time = time.time()
    signals.message.emit(
        "Start time: {}".format(time.strftime("%X", time.localtime(start_time))), True
    )
    return start_time


def emit_end_time_duration(start_time, activity_name, signals):
    """Emits the end time and duration of something that has started before."""
    end_time = time.time()
    signals.message.emit(
        "End time: {}".format(time.strftime("%X", time.localtime(end_time))), True
    )
    signals.message.emit(
        "{} took (hh:mm:ss): {}.".format(
            activity_name, time.strftime("%H:%M:%S", time.gmtime(end_time - start_time))
        ),
        True,
    )

"""Defines helper functions for working with signals emitted during a ParallelWorker job."""

import time


def emit_top_header(text, signals):
    """ Emits a top header."""
    signals.message.emit(text, True)
    signals.message.emit(("=") * len(text), True)
    signals.message.emit("", True)


def emit_sub_header(text, signals):
    """ Emits a sub header."""
    signals.message.emit(text, True)
    signals.message.emit("-" * len(text), True)


def emit_message(message, signals):
    """Emits a simple message."""
    signals.message.emit(message, True)


def emit_image(image_bytes, image_format, signals):
    """Emits the passed image."""
    signals.image.emit(image_bytes, image_format)

def emit_partial_message(message, signals):
    """Emits the given message but does not advance to the next line."""
    signals.message.emit(message, False)


def emit_new_line(signals):
    """Emits a new line to improve readability."""
    signals.message.emit("", True)


def emit_start_time(signals):
    """Emits the start time and returns it for later when the duration needs to be calculated."""
    start_time = time.time()
    signals.message.emit(
        "Start time: {}".format(time.strftime("%X", time.localtime(start_time))), True
    )
    return start_time


def emit_end_time_duration(start_time, activity_name, signals):
    """Emits the end time and duration of something that has started before."""
    end_time = time.time()
    signals.message.emit(
        "End time: {}".format(time.strftime("%X", time.localtime(end_time))), True
    )
    signals.message.emit(
        "{} took (hh:mm:ss): {}.".format(
            activity_name, time.strftime("%H:%M:%S", time.gmtime(end_time - start_time))
        ),
        True,
    )
Defines helper functions for working with signals emitted during a ParallelWorker job. Emits a top header. Emits a sub header. Emits a simple message. Emits the passed image. Emits the given message but does not advance to the next line. Emits a new line to improve readability. Emits the start time and returns it for later when the duration needs to be calculated. Emits the end time and duration of something that has started before.