Source code for labscheduler.sila_server.generated.schedulingservice.schedulingservice_base

# Generated by sila2.code_generator; sila2.__version__: 0.12.2
from __future__ import annotations

from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, List, Optional

from sila2.server import FeatureImplementationBase, MetadataDict, ObservableCommandInstance

if TYPE_CHECKING:
    from datetime import timedelta

    from ...server import Server
    from .schedulingservice_types import (
        AlgorithmMetaData,
        ComputeSchedule_Responses,
        SelectAlgorithm_Responses,
        WorkflowGraph,
    )


[docs] class SchedulingServiceBase(FeatureImplementationBase, ABC): parent_server: Server ComputeSchedule_default_lifetime_of_execution: timedelta | None def __init__(self, parent_server: Server): """ Provides an interface for the PythonLabOrchestrator or any other lab environment to interact with the scheduling module. It provides an observable command to schedule a workflow(exact syntax explained in command). You can choose between different algorithms as get their metadata. """ super().__init__(parent_server=parent_server) self.ComputeSchedule_default_lifetime_of_execution = None
[docs] @abstractmethod def get_AvailableAlgorithms(self, *, metadata: MetadataDict) -> list[AlgorithmMetaData]: """ List of algorithm info for all available algorithms :param metadata: The SiLA Client Metadata attached to the call :return: List of algorithm info for all available algorithms """
[docs] @abstractmethod def get_CurrentAlgorithm(self, *, metadata: MetadataDict) -> AlgorithmMetaData: """ The currently selected algorithm :param metadata: The SiLA Client Metadata attached to the call :return: The currently selected algorithm """
[docs] @abstractmethod def SelectAlgorithm(self, AlgorithmName: str, *, metadata: MetadataDict) -> SelectAlgorithm_Responses: """ Selects the algorithm to be used in the ComputeSchedule command :param AlgorithmName: Name of the Algorithm :param metadata: The SiLA Client Metadata attached to the call """
[docs] @abstractmethod def ComputeSchedule( self, WorkflowGraph: WorkflowGraph, MaxComputationTime: float, *, metadata: MetadataDict, instance: ObservableCommandInstance, ) -> ComputeSchedule_Responses: """ Takes a workflow graph (see parameter description) and computes a schedule for the currently configured lab environment (LabConfigurationController) with the currently selected algorithm. You have to give the maximum computation time. :param WorkflowGraph: The workflow graph of an experiment in the required SiLA structure. :param MaxComputationTime: Maximum computation time in seconds :param metadata: The SiLA Client Metadata attached to the call :param instance: The command instance, enabling sending status updates to subscribed clients :return: A tuple with: - The computed schedule - A string indicating whether the schedule is "optimal", "feasible", or "infeasible" """