Source code for laborchestrator.database_integration.status_db_dummy

"""
A mockup implementation of the status database interface.
It does basically nothing and will be used if no database support is set up for the orchestrator.
"""
from typing import Optional, List, Tuple

from laborchestrator.database_integration import StatusDBInterface
from laborchestrator.structures import ProcessStep, ContainerInfo



[docs]
class StatusDBDummyImplementation(StatusDBInterface):

[docs]
    def position_empty(self, device: str, pos: int) -> bool:
        return True



[docs]
    def get_available_processes(self) -> List[Tuple[str, str]]:
        return []



[docs]
    def add_process_to_db(self, name: str, src: str) -> str:
        return ""



[docs]
    def create_experiment(self, process_id: str) -> str:
        return "Hello World"



[docs]
    def get_all_positions(self, device: str) -> List[int]:
        return [0]



[docs]
    def get_container_at_position(self, device: str, pos: int) -> Optional[ContainerInfo]:
        return None



[docs]
    def moved_container(self, source_device: str, source_pos: int, target_device: str, target_pos: int,
                        barcode: Optional[str] = None):
        pass



[docs]
    def unlidded_container(self, cont_info: ContainerInfo, lid_device: str, lid_pos: int):
        pass



[docs]
    def lidded_container(self, cont_info: ContainerInfo, lid_device: Optional[str] = None, lid_pos: Optional[int] = None):
        pass



[docs]
    def get_cont_info_by_barcode(self, barcode: str) -> Optional[ContainerInfo]:
        return None



[docs]
    def add_container(self, cont: ContainerInfo):
        pass



[docs]
    def remove_container(self, cont: ContainerInfo):
        pass



[docs]
    def set_barcode_at_position(self, barcode: str, device: str, pos: int):
        pass



[docs]
    def set_barcode(self, cont: ContainerInfo):
        pass



[docs]
    def update_lid_position(self, cont: ContainerInfo):
        pass



[docs]
    def get_estimated_duration(self, step: ProcessStep, confidence=.95) -> Optional[float]:
        return None



[docs]
    def get_estimated_durations(self, steps: List[ProcessStep], confidence=.95) -> List[Optional[float]]:
        return [None] * len(steps)



[docs]
    def safe_step_to_db(self, step: ProcessStep, container_info: ContainerInfo, experiment_uuid: str):
        pass



[docs]
    def write_server_certificate(self, device_name, cert: str) -> None:
        pass



[docs]
    def get_server_certificate(self, device_name) -> str:
        return ""