Source code for daq_queuing_service.plugins.converter

import importlib

from blueapi.service.model import TaskRequest
from blueapi.worker.event import TaskError, TaskResult

from daq_queuing_service.external_interaction.blueapi.blueapi_call import (
    BlueapiCall,
    BlueapiCallResponse,
)
from daq_queuing_service.task_queue.task import (
    Experiment,
    Task,
    TaskWithPosition,
)


class ConverterError(Exception):
    def __init__(self, original: Exception):
        super().__init__(f"{type(original).__name__}: {original}")
        self.original = original


class ValidateError(Exception): ...




class Converter:
    def __init__(self): ...



    def validate(self, experiments: list[TaskRequest | Experiment]) -> None:
        """This gets run on new experiments added to the queue through the api endpoint.
        Any error raised will be handled and cause the experiments to not be added to
        the queue.

        Args:
            experiments (list[TaskRequest  |  Experiment]): List of experiments or plans
            added to the queue.

        Raises:
            Exception: Any exception raised due to experiments failing validation
        """
        ...




    def pre_process(
        self,
        current_task: TaskWithPosition | None,
        queue: list[Task],
        history: list[TaskWithPosition],
        call_history: list[BlueapiCallResponse],
    ) -> list[Task]:
        """This gets run whenever something changes in the queue, before
        'construct_blueapi_calls'. It is an opportunity to automatically modify the
        queue based on it's current state and history.

        For example, it could be used to add calibration scans before any experiments
        that require them. It could also be used to query another service such as ulims
        and add experiments to the queue.

        If not overridden, this method returns the queue as is with no modification.

        Args:
            current_task (TaskWithPosition | None): The current running task, if one
                exists.
            queue (list[Task]): List of tasks currently in the queue
            history (list[TaskWithPosition]): List of completed tasks
            call_history (list[BlueapiCallResponse]): List of completed blueapi calls

        Returns:
            list[Task]: The new list of tasks in the queue
        """
        return queue




    def construct_blueapi_calls(
        self,
        queue: list[TaskWithPosition],
        history: list[TaskWithPosition],
        call_history: list[BlueapiCallResponse],
    ) -> list[BlueapiCall]:
        """Converts the list of queued tasks into a list of blueapi calls. This is
        needed if the queue contains ulims experiments, as these must be mapped onto
        bluesky plans. If the queue contains TaskRequests, they can just be
        wrapped into BlueapiCall objects with no conversion, as this default
        implementation does.

        Args:
            queue (list[TaskWithPosition]): List of tasks in the queue
            history (list[TaskWithPosition]): List of completed tasks
            call_history (list[BlueapiCallResponse]): List of completed blueapi calls

        Returns:
            list[BlueapiCall]: List of blueapi calls to execute.
        """

        call_list = [
            BlueapiCall(
                parent_task_id=task.id,
                task_request=self._construct_blueapi_task_request(task.experiment),
            )
            for task in queue
        ]
        return call_list




    def pause_on_complete_blueapi_call(
        self, call: BlueapiCall, result: TaskResult
    ) -> bool:
        """Decide whether to pause the queue after a blueapi call completes successfully

        Args:
            call (BlueapiCall): The completed call
            result (TaskResult): The result of the call from blueapi

        Returns:
            bool: True to pause the queue, False to leave it running.
        """
        return False




    def pause_on_fail_blueapi_call(
        self, call: BlueapiCall, errors: list[str | TaskError] | None
    ) -> bool:
        """Decide whether to pause the queue after a blueapi call fails

        Args:
            call (BlueapiCall): The failed call
            errors (list[str  |  TaskError] | None): The errors that occurred

        Returns:
            bool: True to pause the queue, False to leave it running.
        """
        return True


    def _construct_blueapi_task_request(
        self,
        experiment: Experiment | TaskRequest,
    ) -> TaskRequest:
        match experiment:
            case TaskRequest():
                return experiment
            case Experiment() as experiment:
                raise NotImplementedError(
                    f"No conversion implemented for {type(experiment)}. "
                    + "Try using a different converter"
                )



def get_converter(path: str, name: str) -> Converter:
    """Instantiates a converter based on a path and class name

    Args:
        path (str): Path to converter class. For example:
            "daq_queuing_service.plugins.converter"
        converter_name (str): Name of the converter class. For example:
            "Converter"

    Returns:
        Converter: Converter instance
    """
    module = importlib.import_module(path)
    converter_cls = getattr(module, name)
    converter = converter_cls()
    if not isinstance(converter, Converter):
        raise TypeError(f"Converter is not of type Converter, it is {type(converter)}")
    return converter