Source code for synbiopython.lab_automation.picklist.PickList

# pylint: disable=C0330,C0103,E0102,R1705,R0913
"""Classes to represent picklists and liquid transfers in general."""
from copy import deepcopy
from synbiopython.lab_automation.picklist.Transfer import Transfer

[docs]class PickList: """Representation of a list of well-to-well transfers. :param transfers_list: A list of Transfer objects that will be part of the same dispensing operation, in the order in which they are meant to be simulated. :param data: A dict with information on the picklist. """ def __init__(self, transfers_list=(), data=None): self.transfers_list = list(transfers_list) = {} if data is None else data
[docs] def add_transfer( self, source_well=None, destination_well=None, volume=None, data=None, transfer=None, ): """Add a transfer to the picklist's tranfers list. You can either provide a ``Transfer`` object with the ``transfer`` parameter, or the parameters. """ if transfer is None: transfer = Transfer( source_well=source_well, destination_well=destination_well, volume=volume, data=data, ) self.transfers_list.append(transfer)
[docs] def to_plain_string(self): """Return the list of transfers in human-readable format.""" return "\n".join(transfer.to_plain_string() for transfer in self.transfers_list)
[docs] def to_plain_textfile(self, filename): """Write the picklist in a file in a human reable format.""" with open(filename, "w+") as f: f.write(self.to_plain_string())
[docs] def simulate(self, content_field="content", inplace=True): """Simulate the execution of the picklist.""" if not inplace: all_plates = set( plate for transfer in self.transfers_list for plate in [ transfer.source_well.plate, transfer.destination_well.plate, ] ) new_plates = {plate: deepcopy(plate) for plate in all_plates} new_transfer_list = [] for transfer in self.transfers_list: new_source_plate = new_plates[transfer.source_well.plate] new_dest_plate = new_plates[transfer.destination_well.plate] new_source_well = new_source_plate.wells[] new_dest_well = new_dest_plate.wells[] new_transfer_list.append( Transfer( volume=transfer.volume, source_well=new_source_well, destination_well=new_dest_well, ) ) new_picklist = PickList(transfers_list=new_transfer_list) new_picklist.simulate( content_field=content_field, inplace=True, ) return new_plates else: for transfer in self.transfers_list: transfer.apply() return None
[docs] def restricted_to( self, transfer_filter=None, source_well=None, destination_well=None ): """Return a version of the picklist restricted to transfers with the right source/destination well. You can provide ``source_well`` and ``destination_well`` or alternatively just a function ``transfer_filter`` with signature (transfer)=>True/False that will be used to filter out transfers (for which it returns false). """ if transfer_filter is None: def transfer_filter(tr): source_well_is_ok = (source_well is None) or ( source_well == tr.source_well ) dest_well_is_ok = (destination_well is None) or ( destination_well == tr.destination_well ) return source_well_is_ok and dest_well_is_ok transfers = [tr for tr in self.transfers_list if transfer_filter(tr)] return PickList(transfers, data={"parent": self})
[docs] def sorted_by(self, sorting_method="source_well"): """Return a new version of the picklist sorted by some parameter. The ``sorting_method`` is either the name of an attribute of the transfers, such as "source_well", or a function f(transfer) -> value. """ if not hasattr(sorting_method, "__call__"): def sorting_method(transfer): return transfer.__dict__[sorting_method] return PickList( sorted(self.transfers_list, key=sorting_method), data={"parent": self}, )
[docs] def total_transferred_volume(self): """Return the sum of all volumes from all transfers.""" return sum([transfer.volume for transfer in self.transfers_list])
[docs] def enforce_maximum_dispense_volume(self, max_dispense_volume): """Return a new picklist were every too-large dispense is broken down into smaller dispenses.""" transfers = [] for trf in self.transfers_list: n_additional_dispense = int(trf.volume / max_dispense_volume) rest = trf.volume - n_additional_dispense * max_dispense_volume for _ in range(n_additional_dispense): transfers.append(trf.with_new_volume(max_dispense_volume)) if rest > 0: transfers.append(trf.with_new_volume(rest)) return PickList(transfers_list=transfers)
def __add__(self, other): return PickList(self.transfers_list + other.transfers_list)
[docs] @staticmethod def merge_picklists(picklists_list): """Merge the list of picklists into a single picklist. The transfers in the final picklist are the concatenation of the transfers in the different picklists, in the order in which they appear in the list. """ return sum(picklists_list, PickList([]))