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"
"""