Source code for pythonlab.resources.services.liquid_handling

from typing import List, Union

from pythonlab.resource import ServiceResource, Position, LabwareResource, DynamicLabwareResource
import logging




class LiquidHandlerServiceResource(ServiceResource):
    """


    :param Resource: [description]
    :type Resource: [type]
    """

    def __init__(self, proc, name: str):
        super().__init__(proc=proc, name=name)
        self._transfer_positon = Position(self, position=0)

    @property
    def transfer_position(self):
        return self._transfer_positon



    def executeProtocol(self, labware: Union[List[LabwareResource], LabwareResource, None], protocol: str,
                        **kwargs):
        # wrap a single labware into a list (usability and backwards compatibility)
        if isinstance(labware, LabwareResource):
            labware = [labware]
        elif labware is None:
            labware = []
        logging.debug(f"\n execute {protocol} on {[c.name for c in labware]}.\n")
        # bring all needed reagents to the liquid handler
        last_job = {}
        if "reagents" in kwargs:
            for i, reagent in enumerate(kwargs['reagents']):
                if isinstance(reagent, DynamicLabwareResource):
                    # copy their current position, so we can later reset them
                    last_job[reagent.name] = self.proc.last_job[reagent.name].copy()
                    action_kwargs = dict(
                        resource=self.proc.mover_pool, labware=[reagent], fct='move', target=type(self),
                        target_name=self.name)
                    if "reagent_pos" in kwargs:
                        preferred_pos = kwargs["reagent_pos"]
                        if preferred_pos[i] is not None:
                            action_kwargs['position'] = preferred_pos[i]
                    # we assume reagent_pos to be a dictionary[reagent.name -> position]
                    if 'reagent_pos' in kwargs:
                        if reagent.name in kwargs['reagent_pos']:
                            action_kwargs['position'] = kwargs['reagent_pos'][reagent.name]
                    self.proc.add_process_step(is_movement=True, **action_kwargs)
                    reagent.wait_cost(reagent.outside_cost)
        if "duration" not in kwargs:
            kwargs['duration'] = 60
        kwargs.update(dict(fct='executeProtocol', method=protocol))
        self.proc.add_process_step(self, labware, **kwargs)

        # bring all needed reagents back to their storage
        if "reagents" in kwargs:
            for reagent in kwargs['reagents']:
                if isinstance(reagent, DynamicLabwareResource):
                    reagent.wait_cost(reagent.outside_cost)
                    self.proc.add_process_step(is_movement=True, resource=self.proc.mover_pool, labware=[reagent],
                                               fct='move',
                                               target=type(reagent.start_position.resource),
                                               target_name=reagent.start_position.resource.name,
                                               position=reagent.start_position.pos)
                    # reset the reagents las job
                    self.proc.last_job[reagent.name] = last_job[reagent.name]