"""
The formal interface a database must implement to enable
"""
from typing import Optional, List, Tuple
from laborchestrator.structures import ContainerInfo, ProcessStep
from abc import ABC, abstractmethod
[docs]
class StatusDBInterface(ABC):
#def __init__(self):
# pass
@classmethod
def __subclasshook__(cls, subclass):
# todo: can't i do this with an iteration?
return (hasattr(subclass, 'get_all_positions')
and callable(subclass.get_all_positions)
and hasattr(subclass, 'add_process_to_db')
and callable(subclass.add_process_to_db)
and hasattr(subclass, 'get_available_processes')
and callable(subclass.get_available_processes)
and hasattr(subclass, 'get_process')
and callable(subclass.get_process)
and hasattr(subclass, 'create_experiment')
and callable(subclass.create_experiment)
and hasattr(subclass, 'moved_container')
and callable(subclass.moved_container)
and hasattr(subclass, 'unlidded_container')
and callable(subclass.unlidded_container)
and hasattr(subclass, 'lidded_container')
and callable(subclass.lidded_container)
and hasattr(subclass, 'get_cont_info_by_barcode')
and callable(subclass.get_cont_info_by_barcode)
and hasattr(subclass, 'add_container')
and callable(subclass.add_container)
and hasattr(subclass, 'remove_container')
and callable(subclass.remove_container)
and hasattr(subclass, 'set_barcode_at_position')
and callable(subclass.set_barcode_at_position)
and hasattr(subclass, 'set_barcode')
and callable(subclass.set_barcode)
and hasattr(subclass, 'update_lid_position')
and callable(subclass.update_lid_position)
and hasattr(subclass, 'get_estimated_duration')
and callable(subclass.get_estimated_duration)
and hasattr(subclass, 'set_barcode_at_position')
and callable(subclass.safe_step_to_db)
or NotImplemented)
[docs]
@abstractmethod
def get_all_positions(self, device: str) -> List[int]:
"""
Provides the indices of all positions in the specified device (independent of emptiness).
:param device:
:return:
"""
[docs]
def add_process_to_db(self, name: str, src: str) -> str:
"""
:param name:
:param src:
:return: the uuid of the created process
"""
[docs]
def get_available_processes(self) -> List[Tuple[str, str]]:
"""
:return: A List of tuples [process name, process uuid] for all processes in the db
"""
[docs]
def get_process(self, process_uuid: str) -> str:
"""
:param process_uuid: The unique identifier of the process in the db
:return: the source code defining the process
"""
[docs]
def create_experiment(self, process_id: str) -> str:
"""
:param process_id: the unique id of the process running in this experiment
:return: the uuid of the created experiment
"""
[docs]
@abstractmethod
def get_container_at_position(self, device: str, pos: int) -> Optional[ContainerInfo]:
"""
Checks if there is a container at a specified position.
:param device:
:param pos:
:return: None if there is no container at the specified position. All available information otherwise.
"""
[docs]
@abstractmethod
def position_empty(self, device: str, pos: int) -> bool:
"""
Checks whether the specified position is empty
:param device: Name of the device
:param pos: Index of the slot
:return:
"""
[docs]
@abstractmethod
def moved_container(self, source_device: str, source_pos: int, target_device: str, target_pos: int,
barcode: Optional[str] = None):
"""
Saves a movement of a container (specified by its prior position) to the database
:param source_device:
:param source_pos:
:param target_device:
:param target_pos:
:param barcode:
:return:
"""
[docs]
@abstractmethod
def unlidded_container(self, cont_info: ContainerInfo, lid_device: str, lid_pos: int):
"""
Saves the removal of the lid and its new position to the database
:param cont:
:param lid_device:
:param lid_pos:
:return:
"""
pass
[docs]
@abstractmethod
def lidded_container(self, cont_info: ContainerInfo, lid_device: Optional[str], lid_pos: Optional[int]):
"""
Saved to the database, that a containers lid was put back on. The position of the used lid can be specified,
so the database will/can check that it was the correct one.
:param cont:
:param lid_device:
:param lid_pos:
:return:
"""
pass
[docs]
@abstractmethod
def get_cont_info_by_barcode(self, barcode: str) -> Optional[ContainerInfo]:
"""
Retrieves all available information for the container with the given barcode
:param barcode:
:return:
"""
[docs]
@abstractmethod
def add_container(self, cont: ContainerInfo):
"""
Adds a container to the database with its starting position as the current position.
:param cont: The container to add
:return: nothin
"""
[docs]
@abstractmethod
def remove_container(self, cont: ContainerInfo):
"""
Marks the given container as removed from the platform. The information is still kept in the database
:param cont: The container to mark removed
:return: nothing
"""
[docs]
@abstractmethod
def set_barcode(self, cont: ContainerInfo):
"""
Sets the barcode of an existing container. Assumes the barcode is already saved in the ContainerInfo.
:param cont:
:return:
"""
[docs]
@abstractmethod
def update_lid_position(self, cont: ContainerInfo):
"""
Sets the barcode of an existing container. Assumes the barcode is already saved in the ContainerInfo.
:param cont:
:return:
"""
[docs]
@abstractmethod
def get_estimated_duration(self, step: ProcessStep, confidence=.95) -> Optional[float]:
"""
Checks the database for similar jobs to estimate the duration of a job.
:param step: the job of which the duration shall be estimated
:param confidence: chance, that the actual duration is less or equal the estimated duration.
:return: duration in seconds or None if no information was found in the database
"""
[docs]
@abstractmethod
def get_estimated_durations(self, steps: List[ProcessStep], confidence=.95) -> List[Optional[float]]:
"""
Same as get_estimated_duration(), but with many steps at once to save time
:return: The list will match the length and order of the given steps
"""
[docs]
@abstractmethod
def safe_step_to_db(self, step: ProcessStep, container_info: ContainerInfo, experiment_uuid: str):
"""
Saves a finished job to the database. It automatically recognizes move jobs.
:param experiment_uuid: The uuid of the experiment in the database, this step belongs to
:param container_info: information about the processed container
:param step: the structures.ProcessStep to save
:return: nothing
"""
[docs]
@abstractmethod
def write_server_certificate(self, device_name, cert: str) -> None:
"""
:param device_name:
:param cert:
:return:
"""
[docs]
@abstractmethod
def get_server_certificate(self, device_name) -> str:
"""
:param device_name:
:return:
"""
