Source code for labscheduler.sila_server.feature_implementations.labconfigurationcontroller_impl
# Generated by sila2.code_generator; sila2.__version__: 0.10.3
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
import yaml
from labscheduler.sila_server.generated.labconfigurationcontroller import (
ConfigureJobShop_Responses,
FileFormatError,
LabConfigurationControllerBase,
LoadJobShopFromFile_Responses,
Machine,
)
# TODO: please chceck, if this is correct (MACHINE vs INTERNMACHINE)
from labscheduler.structures import Machine as InternMachine
if TYPE_CHECKING:
from sila2.server import MetadataDict
from labscheduler.scheduler_interface import SchedulerInterface
from labscheduler.sila_server import Server
logger = logging.getLogger(__name__)
[docs]
class LabConfigurationControllerImpl(LabConfigurationControllerBase):
def __init__(self, parent_server: Server, scheduler_interface: SchedulerInterface) -> None:
super().__init__(parent_server=parent_server)
self.scheduler_interface = scheduler_interface
[docs]
def get_CurrentJobShop(self, *, metadata: MetadataDict) -> list[Machine]:
response = []
for intern_machine in self.scheduler_interface.job_shop:
sila_machine = (
intern_machine.name,
intern_machine.type,
intern_machine.max_capacity,
intern_machine.process_capacity,
intern_machine.min_capacity,
intern_machine.allows_overlap,
)
response.append(sila_machine)
return response
[docs]
def LoadJobShopFromFile(self, ConfigurationFile: str, *, metadata: MetadataDict) -> LoadJobShopFromFile_Responses:
"""
Deprecation warning: This function is deprecated and will be removed in future versions.
Loads a job shop configuration from a YAML file.
"""
logger.warning("LoadJobShopFromFile is deprecated and will be removed in future versions.")
try:
logger.info(f"Loading job shop from {ConfigurationFile}")
job_shop = self._parse_jobshop_from_yaml_file(ConfigurationFile)
# TODO change this to marks library labconfigreader? - or rather deprecate this function !!
except Exception as e: # FIXME: which type of errors are expected here?
msg = f"Failed to parse file ({e})."
raise FileFormatError(msg) from e
self.scheduler_interface.configure_job_shop(machine_list=job_shop)
[docs]
def _parse_jobshop_from_yaml_file(self, yaml_file: str) -> list[InternMachine]:
"""
Deprecation warning: This function is deprecated and will be removed in future versions.
Parses a YAML file to create a list of Machine objects.
The YAML file should contain a dictionary with two keys:
- pythonlab_translation: a dictionary mapping device types to their corresponding classes
- sila_servers: a dictionary where each key is a device type and the value is a list of devices
with their parameters.
Each device in the sila_servers list should have a name and a dictionary of parameters,
including capacity, min_capacity, process_capacity, and allows_overlap.
The function returns a list of Machine objects created from the data in the YAML file.
"""
logger.warning("parse_jobshop_from_yaml_file is deprecated and will be removed in future versions.")
config_dict = yaml.safe_load(yaml_file)
pythonlab_translation = dict(config_dict["pythonlab_translation"])
job_shop = []
for device_type, device_list in config_dict["sila_servers"].items():
device_class = pythonlab_translation[device_type]
for device_name, param_dict in device_list.items():
max_capacity = param_dict["capacity"]
min_capacity = param_dict.get("min_capacity", 1)
process_capacity = param_dict.get("process_capacity", max_capacity)
allows_overlap = bool(param_dict["allows_overlap"]) if "allows_overlap" in param_dict else True
job_shop.append(
InternMachine(
name=device_name,
max_capacity=max_capacity,
type=device_class,
min_capacity=min_capacity,
process_capacity=process_capacity,
allows_overlap=allows_overlap,
),
)
logger.info("Available instruments:")
for m in job_shop:
logger.info(m)
return job_shop
