Source code for laborchestrator.start_up.example_processes.runtime_decision_process

from numpy import random

from pythonlab.resources.services.moving import MoverServiceResource
from pythonlab.resources.services.centrifugation import CentrifugeServiceResource
from pythonlab.resources.services.analysis import PlateReaderServiceResource
from pythonlab.resources.services.labware_storage import LabwareStorageResource as ContainerStorageResource
from pythonlab.resources.services.incubation import IncubatorServiceResource
from pythonlab.resources.services.liquid_handling import LiquidHandlerServiceResource
from pythonlab.resource import LabwareResource as ContainerResource
from pythonlab.process import PLProcess




class RuntimeDecisionProcess(PLProcess):
    def __init__(self, priority=10):  # 0 has highest priority
        self.num_plates = 2
        self.name = "RuntimeDecisionProcess"
        super().__init__(priority=priority)#, num_plates=2, process_name="RuntimeDecisionProcess")



    def create_resources(self):
        self.hotel = ContainerStorageResource(proc=self, name="Carousel")
        self.rotanta_transfer = ContainerStorageResource(proc=self, name="Transfer")
        self.incubator1 = IncubatorServiceResource(proc=self, name="Incubator1")
        self.incubator2 = IncubatorServiceResource(proc=self, name="Incubator2")
        self.incubator3 = IncubatorServiceResource(proc=self, name="Incubator3")
        self.incubator4 = IncubatorServiceResource(proc=self, name="Incubator4")
        self.robot_arm = MoverServiceResource(proc=self, name="Mover")
        self.reader = PlateReaderServiceResource(proc=self, name="PlateReader")
        self.reader2 = PlateReaderServiceResource(proc=self, name="Plate_Reader2")
        self.pipetter = LiquidHandlerServiceResource(proc=self, name="Liquid_Handler")
        self.centrifuge = CentrifugeServiceResource(proc=self, name="Centrifuge")
        self.reader_pool = PlateReaderServiceResource(proc=self, name=None)
        self.mover_pool = MoverServiceResource(proc=self, name=None)

        # only container 0 will be marked filled for testing purposes
        self.containers = [ContainerResource(proc=self, name=f"{self.name}_cont_{cont}", lidded=True, filled=False)
                           for cont in range(self.num_plates)]




    def init_service_resources(self):
        # setting start position of containers
        super().init_service_resources()
        for cont in self.containers:
            cont.set_start_position(
                self.hotel, self.hotel.next_free_position)




    def process(self):
        incubation_duration1 = 3
        incubation_duration2 = 9

        cont1 = self.containers[0]
        cont2 = self.containers[1]
        for cont in self.containers:
            # read the barcodes
            self.mover_pool.read_barcode(cont)
            # move to incubator
            self.mover_pool.move(cont, target_loc=self.incubator1)
        # incubate
        self.incubator1.incubate(cont1, duration=incubation_duration1, temperature=310)
        self.incubator1.incubate(cont2, duration=incubation_duration2, temperature=310)
        # move to reader
        self.mover_pool.move(cont1, target_loc=self.reader_pool, lidded=False)
        # make absorbance measurement
        absorb = self.reader_pool.single_read(cont1, method='211001_varioskan_single_well_600')
        aver_abs = self.average(absorb)
        if aver_abs < .6:
            # do some more incubation
            self.mover_pool.move(cont1, target_loc=self.incubator1, lidded=True)
            self.incubator1.incubate(cont1, duration=incubation_duration1, temperature=310)

        # move back to hotel
        self.mover_pool.move(cont1, target_loc=self.hotel, lidded=True)
        self.mover_pool.move(cont2, target_loc=self.hotel, position=27)




    def average(self, l):
        abs = random.rand()+.1
        return abs