diff --git a/assets/instructions/protocol-1-instructions.Rmd b/assets/instructions/protocol-1-instructions.Rmd index 747c79e..b2f71d3 100755 --- a/assets/instructions/protocol-1-instructions.Rmd +++ b/assets/instructions/protocol-1-instructions.Rmd @@ -13,7 +13,7 @@ library(knitr) json_data <- fromJSON(file = params$json_path) ``` \vspace{-1.5cm} -## Protocol 1: Heat-Shock Transformation in OT-2 +## Protocol 1: Heat shock transformations ### Overview DNA is transferred to competent cells. diff --git a/assets/instructions/protocol-2-instructions.Rmd b/assets/instructions/protocol-2-instructions.Rmd index 46134e9..22bb77a 100755 --- a/assets/instructions/protocol-2-instructions.Rmd +++ b/assets/instructions/protocol-2-instructions.Rmd @@ -13,15 +13,15 @@ library(knitr) json_data <- fromJSON(file = params$json_path) ``` \vspace{-1.5cm} -## Protocol 2: Agar Plate Spotting in OT-2 +## Protocol 2: Colony selection ### Overview -Transformed cells are spotted on the total of x `r json_data$agar_plate_slots` agar plates. -The pipette aspirates additional `r json_data$additional_volume` uL of transformed cells. +Transformed cells are spotted on the total of `r length(json_data$agar_plate_slot) ` agar plates. +The pipette aspirates additional `r json_data$additional_volume` $\mu$L of transformed cells. The spotting height above the agar is set at `r json_data$spotting_height` mm. The agar height is calculated automatically per plate based on: -the plate weight without agar (`r json_data$empty_agar_plate_weight`) and with agar (g), -and agar density of `r json_data$agar_density`(g/mm3). +the plate weight without agar and with agar (g), +and agar density of `r json_data$agar_density`($g mm^{-3}$). ### Preparation The plates are spotted in uL according to the layout below. @@ -50,5 +50,5 @@ for (i in seq_along(image_files)) { ### Deck Loading Check that `r json_data$pipette_name` is in the `r json_data$pipette_mount` mount. **Slot `r json_data$tiprack_slots`** load `r json_data$tiprack_name` for the `r json_data$pipette_name` pipette. -**Slot `r json_data$agar_plate_slots`** load `r json_data$agar_plate_name` containing agar. +**Slot `r json_data$agar_plate_slot`** load `r json_data$agar_plate_name` containing agar. **Slot `r json_data$source_plate_slot`** load `r json_data$source_plate_name` containing transformed cells. \ No newline at end of file diff --git a/assets/instructions/protocol-3-instructions.Rmd b/assets/instructions/protocol-3-instructions.Rmd index bd97c33..c0634a4 100755 --- a/assets/instructions/protocol-3-instructions.Rmd +++ b/assets/instructions/protocol-3-instructions.Rmd @@ -17,11 +17,11 @@ json_data <- fromJSON(file = params$json_path) ### Overview The colonies are sampled using a pipette tip and transferred to destination plate containing fresh media supplemented with appropriate antibiotic. -The robot samples colonies from total of x agar plates. -The picking height into the agar is set at x mm. +The robot samples colonies from total of `r length(json_data$agar_plate_slot) ` agar plates. +The picking height into the agar is set at `r json_data$agar_pierce_depth` mm. The agar height is calculated automatically per plate based on: -the plate weight without agar (`r json_data$empty_agar_plate_weight`) and with agar (g), -and agar density of `r json_data$agar_density`(g/mm3). +the plate weight without agar and with agar (g), +and agar density of `r json_data$agar_density`($g mm^{-3}$). ### Preparation The colonies are sampled according to the plate layout below. @@ -48,6 +48,9 @@ for (i in seq_along(image_files)) { ``` ### Deck Loading -Check that `r json_data$pipette_name` is in the `r json_data$pipette_mount` mount. -**Slot `r json_data$tiprack_slots`** load `r json_data$tiprack_name` for the `r json_data$pipette_name` pipette. -**Slot `r json_data$agar_plate_slots`** load `r json_data$agar_plate_name` containing agar. \ No newline at end of file +Check that `r json_data$right_pipette_name` is in the right mount and `r json_data$left_pipette_name` is in the left mount. +**Slot `r json_data$right_pipette_tiprack_slot`** load `r json_data$right_pipette_tiprack_name` for the `r json_data$right_pipette_name` pipette. +**Slot `r json_data$left_pipette_tiprack_slot`** load `r json_data$left_pipette_tiprack_name` for the `r json_data$left_pipette_name` pipette. +**Slot `r json_data$media_plate_slot`** load `r json_data$media_plate_name` containing fresh media with antibiotic. +**Slot `r json_data$agar_plate_slot`** load `r json_data$agar_plate_name` containing agar with developed colonies. +**Slot `r json_data$destination_plate_slot`** load `r json_data$destination_plate_name` where fresh media will be distributed and sampled cells propagated. diff --git a/assets/instructions/protocol-4-instructions.Rmd b/assets/instructions/protocol-4-instructions.Rmd index 1d827e6..b8335a3 100755 --- a/assets/instructions/protocol-4-instructions.Rmd +++ b/assets/instructions/protocol-4-instructions.Rmd @@ -49,5 +49,4 @@ for (i in seq_along(image_files)) { ### Deck Loading Check that `r json_data$right_pipette_name` is in the right mount and `r json_data$left_pipette_name` is in the left mount. **Slot `r json_data$right_pipette_tiprack_slot`** load `r json_data$right_pipette_tiprack_name` for the `r json_data$right_pipette_name` pipette. -**Slot `r json_data$left_pipette_tiprack_slot`** load `r json_data$left_pipette_tiprack_name` for the `r json_data$left_pipette_name` pipette. - +**Slot `r json_data$left_pipette_tiprack_slot`** load `r json_data$left_pipette_tiprack_name` for the `r json_data$left_pipette_name` pipette. \ No newline at end of file diff --git a/assets/protocols/protocol-1-template.py b/assets/protocols/protocol-1-template.py index b361853..2478cf3 100644 --- a/assets/protocols/protocol-1-template.py +++ b/assets/protocols/protocol-1-template.py @@ -5,10 +5,10 @@ from typing import Tuple, List, Dict, NamedTuple, Any, Optional metadata = { - "apiLevel": "2.13", - "protocolName": "Protocol 1: E. coli heat-shock transformation", - "description": "OT-2 protocol for standard E. coli heat-shock transformation using thermocycler.", - "author": "Martyna Kasprzyk" + "apiLevel": "2.15", + "protocolName": "Protocol 1: E. coli heat shock transformation", + "description": "OT-2 protocol for standard E. coli heat shock transformation using thermocycler.", + "author": "Stracquadanio Lab" } ############################################## @@ -27,123 +27,84 @@ ############################################## -def load_experiment_parameters(json_file: str) -> dict: - """ - Load and return experiment parameters from a JSON file. - """ - experiment_parameters = json.loads(json_file) - return experiment_parameters - -def load_experiment_data(csv_file: str): - """ - Parse CSV file and return data for single and multi-channel pipettes. - """ - csv_data = csv.DictReader(csv_file.splitlines()[1:]) - - single_channel_wells, multi_channel_wells = [],[] # List of wells when single or multi channel pipette is used - columns = defaultdict(list) - - for row in csv_data: - single_channel_wells.append(row) # For single channel the list of wells stays as provided in the csv file - well = row["destination_well"] - _, column_number = well[0], well[1:] # For multi - if column_number not in columns: - multi_channel_wells.append(row) - columns[column_number].append(well) - - protocol_data = namedtuple("protocol_data", ["dna_id", "dna_well", "dna_volume", "cells_id", "cells_well", "cells_volume", "media_id", "media_well", "media_volume", "destination_well"]) - single_channel_data = extract_data(single_channel_wells) - multi_channel_data = extract_data(multi_channel_wells) - return protocol_data(*single_channel_data), protocol_data(*multi_channel_data) - -def extract_data(rows: List[Dict]) -> Tuple[List[str], ...]: - """ - Extract the required info for the protocol. - """ - data = defaultdict(list) - for row in rows: - for key in row: - if key.endswith("volume"): - data[key].append(float(row[key])) - else: - data[key].append(row[key]) - return tuple(data.values()) - -def choose_pipette_volume(volumes: List[float]) -> str: - """ - Select appropriate pipette based on the volumes. - """ - volumes = [volume for volume in volumes if volume != 0] - min_volume = min(volumes) - if min_volume <= 20: - return "p20" - elif min_volume > 20: - return "p300" - else: - return None - -def choose_pipette_channel(pipette: str) -> str: - """ - Determine pipette channel based on pipette type. - """ - if "8-Channel" in pipette: - return "multi" - elif "Single-Channel" in pipette: - return "single" - else: - return "Invalid pipette type." + +def load_json_data(json_content: str) -> dict: + """Load JSON formatted string with experiment parameters.""" + return json.loads(json_content) + +def load_csv_data(csv_content: str): + """Parse CSV content with experiment data.""" + csv_reader = csv.DictReader(csv_content.splitlines()[1:]) + data = {key: [] for key in csv_reader.fieldnames if key} + for row in csv_reader: + for key, value in row.items(): + data[key].append(float(value) if 'volume' in key else value) + return namedtuple('ProtocolData', data.keys())(*[data[key] for key in data.keys()]) def load_or_reuse_labware(protocol: protocol_api.ProtocolContext, plate_info: Dict[str, str], loaded_plates: Dict[str, protocol_api.Labware]): - """ - Load a plate into the protocol or reuse an existing one if the slot is already occupied. - """ + """Load a plate into the protocol or reuse an existing one if the slot is already occupied.""" slot = plate_info["slot"] - if slot in loaded_plates: - return loaded_plates[slot] # Reuse the plate that's already loaded into this slot - else: - plate = protocol.load_labware(plate_info["name"], slot) - loaded_plates[slot] = plate # Load a new plate and store it in the dictionary - return plate - -def filter_and_transfer(sources: List[str], destinations: List[str], volumes: List[float]) -> Tuple[List[str], List[str], List[float]]: - """ - Filter out "NA" values and return filtered sources, destinations, and volumes. - """ - filtered_indices = [i for i, source in enumerate(sources) if source != "NA"] - return ([sources[i] for i in filtered_indices], - [destinations[i] for i in filtered_indices], - [volumes[i] for i in filtered_indices]) + return loaded_plates[slot] if slot in loaded_plates else loaded_plates.setdefault(slot, protocol.load_labware(plate_info["name"], slot)) -def run(protocol: protocol_api.ProtocolContext): - """ - Main function for running the protocol. - """ - json_params = load_experiment_parameters(INPUT_JSON_FILE) # Load the parameters from the json file - single, multi = load_experiment_data(INPUT_CSV_FILE) # Load data from the csv file, modified based on the channel of the pipette - - # Load pipettes - right_pipette_tipracks = [protocol.load_labware(load_name=json_params["right_pipette_tiprack_name"], location=i) for i in json_params["right_pipette_tiprack_slot"]] - right_pipette = protocol.load_instrument(instrument_name=json_params["right_pipette_name"], mount="right", tip_racks=right_pipette_tipracks) - left_pipette_tipracks = [protocol.load_labware(load_name=json_params["left_pipette_tiprack_name"], location=i) for i in json_params["left_pipette_tiprack_slot"]] - left_pipette = protocol.load_instrument(instrument_name=json_params["left_pipette_name"], mount="left", tip_racks=left_pipette_tipracks) - pipette_cells = right_pipette if choose_pipette_volume(single.cells_volume) in json_params["right_pipette_name"] else left_pipette # Choose which pipette to use based on volume - pipette_dna = right_pipette if choose_pipette_volume(single.dna_volume) in json_params["right_pipette_name"] else left_pipette - pipette_media = right_pipette if choose_pipette_volume(single.media_volume) in json_params["right_pipette_name"] else left_pipette - pipette_channel_cells = multi if choose_pipette_channel(str(pipette_cells)) == "multi" else single # Choose the csv wells based on the pipette being used - pipette_channel_dna = multi if choose_pipette_channel(str(pipette_dna)) == "multi" else single - pipette_channel_media = multi if choose_pipette_channel(str(pipette_media)) == 'multi' else single +def filter_and_transfer(pipette, sources: List[str], volumes: List[float], destinations: List[str]) -> Tuple[List[str], List[float], List[str]]: + """Filters out sources, volumes, and destinations. Adjusts wells to start with 'A' for 8-channel pipettes.""" + seen_columns, filtered_sources, filtered_volumes, filtered_destinations = set(), [], [], [] + is_multi_channel = "8-Channel" in str(pipette) - # Load hardware and labware - loaded_plates = {} # Dictionary to keep track of loaded plates + for source, volume, destination in zip(sources, volumes, destinations): + destination_column = destination[1:] + formatted_destination = 'A' + destination_column if is_multi_channel else destination + formatted_source = 'A' + source[1:] if is_multi_channel else source + + if formatted_destination in seen_columns: + continue + + seen_columns.add(formatted_destination) + filtered_sources.append(formatted_source) + filtered_volumes.append(volume) + filtered_destinations.append(formatted_destination) + + return filtered_sources, filtered_volumes, filtered_destinations + +def setup_pipettes(protocol: protocol_api.ProtocolContext, pipette_info: Dict[str, Any]) -> Dict[str, protocol_api.InstrumentContext]: + """Load specified pipettes into the protocol based on configuration details provided.""" + loaded_pipettes = {} + for side in ["right", "left"]: + if pipette_info[f"{side}_pipette_name"] != "NA": + tip_racks = [protocol.load_labware(pipette_info[f"{side}_pipette_tiprack_name"], slot) for slot in pipette_info[f"{side}_pipette_tiprack_slot"]] + pipette = protocol.load_instrument(pipette_info[f"{side}_pipette_name"], mount=side, tip_racks=tip_racks) + loaded_pipettes[pipette_info[f"{side}_pipette_name"]] = pipette + return loaded_pipettes + +def select_pipette(volume: List[float], loaded_pipettes: Dict[str, protocol_api.InstrumentContext]) -> protocol_api.InstrumentContext: + """ Determine the appropriate pipette based on the volume and available pipettes. """ + if len(loaded_pipettes) == 1: + return next(iter(loaded_pipettes.values())) + pipette_type = "p20" if min(volume, default=float('inf')) <= 20 else "p300" + for pipette_name, pipette in loaded_pipettes.items(): + if pipette_type in pipette_name: + return pipette + return None + +def run(protocol: protocol_api.ProtocolContext): + """Main function for running the protocol.""" + json_params = load_json_data(INPUT_JSON_FILE) + data = load_csv_data(INPUT_CSV_FILE) + loaded_pipettes = setup_pipettes(protocol, json_params) + + pipette_cells = select_pipette(data.cells_volume, loaded_pipettes) + pipette_dna = select_pipette(data.dna_volume, loaded_pipettes) + pipette_media = select_pipette(data.media_volume, loaded_pipettes) + + loaded_plates = {} cells_plate = load_or_reuse_labware(protocol, {"name": json_params["cells_plate_name"], "slot": json_params["cells_plate_slot"]}, loaded_plates) dna_plate = load_or_reuse_labware(protocol, {"name": json_params["dna_plate_name"], "slot": json_params["dna_plate_slot"]}, loaded_plates) media_plate = load_or_reuse_labware(protocol, {"name": json_params["media_plate_name"], "slot": json_params["media_plate_slot"]}, loaded_plates) - cells_plate = load_or_reuse_labware(protocol, {"name": json_params["cells_plate_name"], "slot": json_params["cells_plate_slot"]}, loaded_plates) protocol.set_rail_lights(True) if json_params["destination_plate_slot"] == "thermocycler": - thermocycler_mod = protocol.load_module("thermocycler") # Load the thermocycler module which takes location 7,8,10,11 + thermocycler_mod = protocol.load_module("thermocycler") destination_plate = thermocycler_mod.load_labware(json_params["destination_plate_name"]) thermocycler_mod.set_block_temperature(temperature=json_params["pre_shock_incubation_temp"]) thermocycler_mod.open_lid() @@ -152,40 +113,50 @@ def run(protocol: protocol_api.ProtocolContext): destination_plate = protocol.load_labware(json_params["destination_plate_name"], json_params["destination_plate_slot"]) ########## ADD COMPETENT CELLS ########## - cells_source, cells_destination, cells_volume = filter_and_transfer(pipette_channel_cells.cells_well, pipette_channel_cells.destination_well, pipette_channel_cells.cells_volume) + protocol.comment("Adding competent cells.") pipette_cells.pick_up_tip() - # Set to keep track of wells that have been mixed mixed_wells = set() - for well, volume in zip(cells_source, cells_volume): - if well not in mixed_wells: - # Mix the well if it hasn't been mixed yet - pipette_cells.mix(1, pipette_cells.max_volume, cells_plate.wells_by_name()[well]) - mixed_wells.add(well) - pipette_cells.transfer(volume=volume, - source=cells_plate.wells_by_name()[well], - dest=destination_plate.wells_by_name()[well], - new_tip="never") + cells_source, cells_volume, cells_destination = filter_and_transfer(pipette_cells, data.cells_well, data.cells_volume, data.destination_well) + print(data.cells_well, data.cells_volume, data.destination_well) + print(cells_source, cells_volume, cells_destination) + + cumulative_cell_volumes = defaultdict(float) + for src_well, vol_cells in zip(cells_source, cells_volume): + cumulative_cell_volumes[src_well] += vol_cells + + for src_well, vol_cells, dest_well in zip(cells_source, cells_volume, cells_destination): + if src_well not in mixed_wells: + mix_volume = cumulative_cell_volumes[src_well] + if mix_volume > pipette_cells.max_volume: + mix_volume = pipette_cells.max_volume + pipette_cells.mix(1, mix_volume, cells_plate.wells_by_name()[src_well]) + mixed_wells.add(src_well) + + pipette_cells.transfer(volume=vol_cells, + source=cells_plate.wells_by_name()[src_well], + dest=destination_plate.wells_by_name()[dest_well], + new_tip="never") pipette_cells.drop_tip() - + ########## ADD DNA ########## - pipette_dna.well_bottom_clearance.aspirate = 0.5 - dna_source, dna_destination, dna_volume = filter_and_transfer(pipette_channel_dna.dna_well, pipette_channel_dna.destination_well, pipette_channel_dna.dna_volume) - for i in zip(dna_source, dna_destination, dna_volume, cells_volume): + protocol.comment("Adding DNA.") + dna_source, dna_volume, dna_destination = filter_and_transfer(pipette_dna, data.dna_well, data.dna_volume, data.destination_well,) + for src_well, vol_dna, dest_well, vol_cells in zip(dna_source, dna_volume, dna_destination, cells_volume): pipette_dna.pick_up_tip() - pipette_dna.aspirate(volume=i[2], location=dna_plate.wells_by_name()[i[0]]) - pipette_dna.dispense(volume=i[2], location=destination_plate.wells_by_name()[i[1]]) - mixing_volume = (i[2]+i[3])/2 - pipette_dna.mix(repetitions=2, volume=mixing_volume, location=destination_plate.wells_by_name()[i[1]]) - pipette_dna.blow_out(location=destination_plate.wells_by_name()[i[1]]) - pipette_dna.move_to(destination_plate.wells_by_name()[i[1]].bottom()) # To make sure that the droplets from the blow out do not stay on the tip + pipette_dna.aspirate(volume=vol_dna, location=dna_plate.wells_by_name()[src_well]) + pipette_dna.dispense(volume=vol_dna, location=destination_plate.wells_by_name()[dest_well]) + mixing_volume = (vol_dna + vol_cells) / 2 + pipette_dna.mix(repetitions=2, volume=mixing_volume, location=destination_plate.wells_by_name()[dest_well]) + pipette_dna.blow_out(location=destination_plate.wells_by_name()[dest_well]) + pipette_dna.move_to(destination_plate.wells_by_name()[dest_well].bottom()) # To make sure that the droplets from the blow out do not stay on the tip pipette_dna.drop_tip() - ########## HEAT-SHOCK TRANSFORMATION ########## + ########## HEAT SHOCK TRANSFORMATION ########## if json_params["destination_plate_slot"] == "thermocycler": thermocycler_mod.close_lid() thermocycler_mod.set_block_temperature(temperature=json_params["pre_shock_incubation_temp"], hold_time_minutes=json_params["pre_shock_incubation_time"]) - protocol.comment("Starting heat-shock transformation.") + protocol.comment("Starting heat shock transformation.") thermocycler_mod.set_block_temperature(temperature=json_params["heat_shock_temp"], hold_time_seconds=json_params["heat_shock_time"]) thermocycler_mod.set_block_temperature(temperature=json_params["post_shock_incubation_temp"], @@ -195,22 +166,21 @@ def run(protocol: protocol_api.ProtocolContext): protocol.pause("Put plate into an external thermocycler for heat-shock transformation and return.") ######## ADD RECOVERY MEDIUM ########## - media_source, media_destination, media_volume = filter_and_transfer(pipette_channel_media.media_well, pipette_channel_media.destination_well, pipette_channel_media.media_volume) - for i in zip(media_source, media_destination, media_volume, cells_volume): + protocol.comment("Adding recovery media.") + media_source, media_volume, media_destination = filter_and_transfer(pipette_media, data.media_well, data.media_volume, data.destination_well) + for src_well, vol_media, dest_well, vol_cells in zip(media_source, media_volume, media_destination, cells_volume): pipette_media.pick_up_tip() - pipette_media.aspirate(volume=i[2], location=media_plate.wells_by_name()[i[0]]) - pipette_media.dispense(volume=i[2], location=destination_plate.wells_by_name()[i[1]]) - mixing_volume = (i[2]+i[3])/2 - for _ in range(3): - pipette_media.aspirate(volume=mixing_volume, location=destination_plate.wells_by_name()[i[1]]) - pipette_media.dispense(volume=mixing_volume, location=destination_plate.wells_by_name()[i[1]]) - pipette_media.blow_out(location=destination_plate.wells_by_name()[i[1]]) - pipette_media.move_to(destination_plate.wells_by_name()[i[1]].bottom()) + pipette_media.aspirate(volume=vol_media + vol_cells, location=media_plate.wells_by_name()[src_well]) + pipette_media.dispense(volume=vol_media, location=destination_plate.wells_by_name()[dest_well]) + for _ in range(2): + pipette_media.aspirate(volume=(vol_media + vol_cells) / 2, location=destination_plate.wells_by_name()[dest_well]) + pipette_media.dispense(volume=(vol_media + vol_cells) / 2, location=destination_plate.wells_by_name()[dest_well]) + pipette_media.blow_out(location=destination_plate.wells_by_name()[dest_well]) + pipette_media.move_to(destination_plate.wells_by_name()[dest_well].bottom()) pipette_media.drop_tip() ######## RECOVERY INCUBATION ########## if json_params["destination_plate_slot"] == "thermocycler": - thermocycler_mod.close_lid() thermocycler_mod.set_lid_temperature(temperature=json_params["recovery_temp"]) thermocycler_mod.set_block_temperature(temperature=json_params["recovery_temp"], @@ -219,4 +189,6 @@ def run(protocol: protocol_api.ProtocolContext): thermocycler_mod.deactivate() protocol.set_rail_lights(False) else: - protocol.comment("Put plate into an external thermocycler for incubation.") \ No newline at end of file + protocol.comment("Put plate into an external thermocycler for incubation.") + + protocol.set_rail_lights(False) \ No newline at end of file diff --git a/assets/protocols/protocol-2-template.py b/assets/protocols/protocol-2-template.py index 9e40a1c..46615c6 100644 --- a/assets/protocols/protocol-2-template.py +++ b/assets/protocols/protocol-2-template.py @@ -6,9 +6,9 @@ import math metadata = { - "apiLevel": "2.13", - "protocolName": "Protocol 2 - Agar Plate Spotting", - "description": "OT-2 protocol for agar plate spotting.", + "apiLevel": "2.15", + "protocolName": "Protocol 2: Colony selection", + "description": "OT-2 protocol for colony selection by selective agar plate spotting.", "author": "Martyna Kasprzyk" } @@ -28,88 +28,59 @@ ############################################## - -def load_experiment_parameters(json_file: str) -> dict: - """ - Load and return experiment parameters from a JSON file. - """ - experiment_parameters = json.loads(json_file) - return experiment_parameters - -def load_experiment_data(csv_file: str): - """ - Parse CSV file and return data for single and multi-channel pipettes. - """ - csv_data = csv.DictReader(csv_file.splitlines()[1:]) - - single_channel_wells, multi_channel_wells = [],[] # List of wells when single or multi channel pipette is used - columns = defaultdict(list) - for row in csv_data: - single_channel_wells.append(row) # For single channel the list of wells stays as provided in the csv file - well = row["destination_well"] - _, column_number = well[0], well[1:] # For multi - if column_number not in columns: - multi_channel_wells.append(row) - columns[column_number].append(well) - - protocol_data = namedtuple("protocol_data", ["id", "agar_plate_location", "source_well", "destination_well", "spotting_volume", "agar_plate_weight"]) - single_channel_data = extract_data(single_channel_wells) - multi_channel_data = extract_data(multi_channel_wells) - return protocol_data(*single_channel_data), protocol_data(*multi_channel_data) - -def extract_data(rows: List[Dict]) -> Tuple[List[str], ...]: - """ - Extract the required info for the protocol. - """ - data = defaultdict(list) - for row in rows: - for key in row: - if key.endswith("volume"): - data[key].append(float(row[key])) - else: - data[key].append(row[key]) - return tuple(data.values()) - -def choose_pipette_channel(pipette: str) -> str: - """ - Determine pipette channel based on pipette type. - """ - if "8-Channel" in pipette: - return "multi" - elif "Single-Channel" in pipette: - return "single" - else: - return "Invalid pipette type." - -def agar_height(agar_plate_weight: float, empty_agar_plate_weight: float , agar_plate_area: float, agar_density: float, spotting_height: float) -> float: - """ - Calculate the the agar height based on the base area of the plate, weight of the empty plate, the weight of plate with agar and the agar density. - """ +def load_json_data(json_content: str) -> dict: + """Load JSON formatted string with experiment parameters.""" + return json.loads(json_content) + +def load_csv_data(csv_content: str): + """Parse CSV content with experiment data.""" + csv_reader = csv.DictReader(csv_content.splitlines()[1:]) + data = {key: [] for key in csv_reader.fieldnames if key} + for row in csv_reader: + for key, value in row.items(): + data[key].append(float(value) if 'volume' in key else value) + return namedtuple('ProtocolData', data.keys())(*[data[key] for key in data.keys()]) + +def filter_and_transfer(pipette, sources: List[str], volumes: List[float], destinations: List[str], locations: List[str]) -> Tuple[List[str], List[float], List[str], List[float]]: + """Filters sources, volumes, and destinations.""" + seen_columns, filtered_sources, filtered_volumes, filtered_destinations, filtered_locations = set(), [], [], [], [] + is_multi_channel = "8-Channel" in str(pipette) + for source, volume, destination, location in zip(sources, volumes, destinations, locations): + if "|" in destination: + formatted_destination = destination + else: + formatted_destination = 'A' + destination[1:] if is_multi_channel else destination + if formatted_destination in seen_columns: + continue + formatted_source = 'A' + source[1:] if is_multi_channel and "|" not in source else source + seen_columns.add(formatted_destination) + filtered_sources.append(formatted_source) + filtered_volumes.append(volume) + filtered_destinations.append(formatted_destination) + filtered_locations.append(int(location)) + + return filtered_sources, filtered_volumes, filtered_destinations, filtered_locations + +def agar_height(agar_plate_weight: float, empty_agar_plate_weight: float, agar_plate_area: float, agar_density: float, spotting_height: float) -> float: + """Calculate the the agar height based on the base area of the plate, weight of the empty plate, the weight of plate with agar and the agar density.""" agar_weight = float(agar_plate_weight) - float(empty_agar_plate_weight) - agar_height = agar_weight/(agar_plate_area*agar_density) # Agar height calculation - height = agar_height + spotting_height # Adds spotting_height i.e. how many mm below or above the agar the robot should dispense + agar_height = agar_weight/(agar_plate_area*(agar_density/1000)) + height = agar_height + spotting_height return height def run(protocol: protocol_api.ProtocolContext): - """ - Main function for running the protocol. - """ - json_params = load_experiment_parameters(INPUT_JSON_FILE) # load the parameters from the json file - single, multi = load_experiment_data(INPUT_CSV_FILE) # load data from the csv file, modified based on the channel of the pipette + """Main function for running the protocol.""" + json_params = load_json_data(INPUT_JSON_FILE) + csv_data = load_csv_data(INPUT_CSV_FILE) - # Load pipettes pipette_tipracks = [protocol.load_labware(load_name=json_params["tiprack_name"], location=i) for i in json_params["tiprack_slots"]] pipette = protocol.load_instrument(instrument_name=json_params["pipette_name"], mount=json_params["pipette_mount"], tip_racks=pipette_tipracks) - pipette_channel = multi if choose_pipette_channel(str(pipette)) == "multi" else single # Choose the csv wells based on the pipette being used - # Load hardware and labware - agar_labware = [protocol.load_labware(load_name=json_params["agar_plate_name"], - location=slot, - label=f"Agar Plate {i+1}") - for i, slot in enumerate(json_params["agar_plate_slots"])] + protocol.set_rail_lights(True) + source_well, spotting_volume, destination_well, locations = filter_and_transfer(pipette, csv_data.source_well, csv_data.spotting_volume, csv_data.destination_well, csv_data.agar_plate_location) - mapping_dict = {int(slot): labware for slot, labware in zip(json_params["agar_plate_slots"], agar_labware)} # Create a mapping dictionary from slot to agar_labware - agar_plates = [mapping_dict[int(loc)] for loc in pipette_channel.agar_plate_location if int(loc) in mapping_dict] # Generate agar_plates list with repeating plates + agar_labware = {int(slot): protocol.load_labware(load_name=json_params["agar_plate_name"], location=slot, label=f"Agar Plate {i+1}") + for i, slot in enumerate(json_params["agar_plate_slot"])} if json_params["source_plate_slot"] == "thermocycler": thermocycler_mod = protocol.load_module("thermocycler") # Thermocycler module, takes location 7,8,10,11 @@ -117,13 +88,31 @@ def run(protocol: protocol_api.ProtocolContext): thermocycler_mod.open_lid() else: source_plate = protocol.load_labware(load_name=json_params["source_plate_name"], location=json_params["source_plate_slot"]) + + agar_info = {slot: {'empty_plate_weight': weight, 'agar_plate_weight': agar_weight} + for slot, weight, agar_weight in zip(json_params["agar_plate_slot"], json_params["empty_agar_plate_weight"], json_params["agar_plate_weight"])} + agar_plates = [agar_labware[loc] for loc in locations] + empty_plate_weight = [agar_info[loc]['empty_plate_weight'] for loc in locations if loc in agar_info] + agar_plate_weight = [agar_info[loc]['agar_plate_weight'] for loc in locations if loc in agar_info] ######## SPOTTING ########## - for plate, source, destination, weight, volume in zip(agar_plates, pipette_channel.source_well, pipette_channel.destination_well, pipette_channel.agar_plate_weight, pipette_channel.spotting_volume): - pipette.well_bottom_clearance.dispense = 1 # Reset the dispense height + for plate, source, volume, destination, empty_weight, agar_weight in zip(agar_plates, source_well, spotting_volume, destination_well, empty_plate_weight, agar_plate_weight): pipette.pick_up_tip() - pipette.mix(repetitions=3, volume=pipette.max_volume, location=source_plate[source], rate=2) # Resuspend the transformed cells - pipette.aspirate(volume=volume + json_params["additional_volume"], location=source_plate[source], rate=2) # Aspirate the spotting volume and dead volume for better accuracy - pipette.well_bottom_clearance.dispense = agar_height(weight, json_params["empty_agar_plate_weight"], json_params["agar_plate_area"], json_params["agar_density"], json_params["spotting_height"]) #Adjust height to the agar wight - pipette.dispense(volume=volume, location=plate[destination], rate=4) # Spotting - protocol.delay(seconds=5) - pipette.drop_tip() \ No newline at end of file + pipette.well_bottom_clearance.dispense = 1 + if "|" in destination: + destinations = destination.split("|") + pipette.mix(repetitions=3, volume=pipette.max_volume / 2, location=source_plate[source], rate=2) + for dest in destinations: + pipette.aspirate(volume = volume + json_params["additional_volume"], location=source_plate[source], rate=2) + pipette.well_bottom_clearance.dispense = agar_height(agar_weight, empty_weight, json_params["agar_plate_area"], json_params["agar_density"], json_params["spotting_height"]) + pipette.dispense(volume = volume, location=plate[dest], rate=4) + protocol.delay(seconds = 5) + pipette.drop_tip() + else: + pipette.mix(repetitions=3, volume=pipette.max_volume / 2, location=source_plate[source], rate=2) + pipette.aspirate(volume=volume + json_params["additional_volume"], location=source_plate[source], rate=2) + pipette.well_bottom_clearance.dispense=agar_height(agar_weight, empty_weight, json_params["agar_plate_area"], json_params["agar_density"], json_params["spotting_height"]) + pipette.dispense(volume=volume, location=plate[destination], rate=4) + protocol.delay(seconds=5) + pipette.drop_tip() + + protocol.set_rail_lights(False) \ No newline at end of file diff --git a/assets/protocols/protocol-3-template.py b/assets/protocols/protocol-3-template.py index 76b1475..44e3495 100644 --- a/assets/protocols/protocol-3-template.py +++ b/assets/protocols/protocol-3-template.py @@ -7,7 +7,7 @@ import numpy as np metadata = { - "apiLevel": "2.13", + "apiLevel": "2.15", "protocolName": "Protocol 3 - Colony Sampling", "description": "OT-2 protocol for colony sampling from agar plates.", "author": "Martyna Kasprzyk" @@ -29,157 +29,133 @@ ############################################## - -def load_experiment_parameters(json_file: str) -> dict: - """ - Load and return experiment parameters from a JSON file. - """ - experiment_parameters = json.loads(json_file) - return experiment_parameters - -def load_experiment_data(csv_file: str): - """ - Parse CSV file and return data for single and multi-channel pipettes. - """ - csv_data = csv.DictReader(csv_file.splitlines()[1:]) - - single_channel_wells, multi_channel_wells = [],[] # List of wells when single or multi channel pipette is used - columns = defaultdict(list) - for row in csv_data: - single_channel_wells.append(row) # For single channel the list of wells stays as provided in the csv file - well = row["destination_well"] - _, column_number = well[0], well[1:] # For multi - if column_number not in columns: - multi_channel_wells.append(row) - columns[column_number].append(well) - - protocol_data = namedtuple("protocol_data", ["id","agar_plate_location","agar_plate_weight","media_source_well","sampling_source_well","media_volume","destination_well"]) - single_channel_data = extract_data(single_channel_wells) - multi_channel_data = extract_data(multi_channel_wells) - return protocol_data(*single_channel_data), protocol_data(*multi_channel_data) - -def extract_data(rows: List[Dict]) -> Tuple[List[str], ...]: - """ - Extarct the required info for the protocol. - """ - data = defaultdict(list) - for row in rows: - for key in row: - if key.endswith("volume"): - data[key].append(float(row[key])) - else: - data[key].append(row[key]) - return tuple(data.values()) - -def choose_pipette_volume(volumes: List[float]) -> str: - """ - Select appropriate pipette based on the volumes. - """ - volumes = [volume for volume in volumes if volume != 0] - min_volume = min(volumes) - if min_volume <= 20: - return "p20" - elif min_volume > 20: - return "p300" +def load_json_data(json_content: str) -> dict: + """Load JSON formatted string with experiment parameters.""" + return json.loads(json_content) + +def load_csv_data(csv_content: str): + """Parse CSV content with experiment data.""" + csv_reader = csv.DictReader(csv_content.splitlines()[1:]) + data = {key: [] for key in csv_reader.fieldnames if key} + for row in csv_reader: + for key, value in row.items(): + data[key].append(float(value) if 'volume' in key else value) + return namedtuple('ProtocolData', data.keys())(*[data[key] for key in data.keys()]) + +def setup_pipettes(protocol: protocol_api.ProtocolContext, pipette_info: Dict[str, Any]) -> Dict[str, protocol_api.InstrumentContext]: + """Load specified pipettes into the protocol based on configuration details provided.""" + loaded_pipettes = {} + for side in ["right", "left"]: + if pipette_info[f"{side}_pipette_name"] != "NA": + tip_racks = [protocol.load_labware(pipette_info[f"{side}_pipette_tiprack_name"], slot) for slot in pipette_info[f"{side}_pipette_tiprack_slot"]] + pipette = protocol.load_instrument(pipette_info[f"{side}_pipette_name"], mount=side, tip_racks=tip_racks) + loaded_pipettes[pipette_info[f"{side}_pipette_name"]] = pipette + return loaded_pipettes + +def select_pipette(loaded_pipettes: Dict[str, protocol_api.InstrumentContext], volume: List[float] = None, is_sampling: bool = False) -> protocol_api.InstrumentContext: + """Determine the appropriate pipette based on volume requirements and whether it's for sampling.""" + if is_sampling: + return min(loaded_pipettes.values(), key=lambda p: p.max_volume) else: - return None - -def choose_pipette_channel(pipette: str) -> str: - """ - Determine pipette channel based on pipette type. - """ - if "8-Channel" in pipette: - return "multi" - elif "Single-Channel" in pipette: - return "single" + pipette_type = "p20" if min(volume, default=float('inf')) <= 20 else "p300" + for pipette_name, pipette in loaded_pipettes.items(): + if pipette_type in pipette_name: + return pipette + return None + +def filter_data(pipette, sources: List[str], volumes: List[float], destinations: List[str], locations: List[any] = None) -> Tuple[List[str], List[float], List[str], List[float]]: + """Filters sources, volumes, and destinations, optionally handling locations.""" + seen_columns, filtered_sources, filtered_volumes, filtered_destinations, filtered_locations = set(), [], [], [], [] + is_multi_channel = "8-Channel" in str(pipette) + + for i, (source, destination) in enumerate(zip(sources, destinations)): + formatted_destination = 'A' + destination[1:] if is_multi_channel else destination + formatted_source = 'A' + source[1:] if is_multi_channel else source + + if formatted_destination not in seen_columns: + seen_columns.add(formatted_destination) + filtered_sources.append(formatted_source) + filtered_destinations.append(formatted_destination) + + if volumes: + filtered_volumes.append(volumes[i]) + if locations: + filtered_locations.append(int(locations[i])) + if locations: + return filtered_sources, filtered_destinations, filtered_locations else: - return "Invalid pipette type." - -def agar_height(agar_plate_weight: float, empty_agar_plate_weight: float , agar_plate_area: float, agar_density: float) -> float: - """ - Calculate the the agar height based on the base area of the plate, weight of the empty plate, the weight of plate with agar and the agar density. - """ + return filtered_sources, filtered_volumes, filtered_destinations + +def agar_height(agar_plate_weight: float, empty_agar_plate_weight: float, agar_plate_area: float, agar_density: float, agar_pierce_depth: float) -> float: + """Calculate the the agar height based on the base area of the plate, weight of the empty plate, the weight of plate with agar and the agar density.""" agar_weight = float(agar_plate_weight) - float(empty_agar_plate_weight) - agar_height = agar_weight/(agar_plate_area*agar_density) # Agar height calculation - height = agar_height # Adds spotting_height i.e. how many mm below or above the agar the robot should dispense + agar_height = agar_weight/(agar_plate_area*(agar_density/1000)) + height = agar_height + agar_pierce_depth return height def calculate_spiral_coords(max_radius: float, num_points: int = 25, total_rotations: float = 3) -> tuple: """ Calculate the coordinates for a spiral sampling method. - max_radius (float): The maximum radial distance from the center to the outermost point of the spiral. num_points (int): Number of points in the spiral. Defines the granularity of the spiral curve. total_rotations (float): Total number of complete 360-degree rotations the spiral will make. - """ - b = max_radius / (total_rotations * 2 * np.pi) # Calculate the increment factor b based on the max radius and total rotations - end_theta = total_rotations * 2 * np.pi # Maximum theta based on total rotations - theta = np.linspace(0, end_theta, num_points) # Create theta values from 0 to end_theta evenly spaced based on num_points - r = b * theta # Calculate the radius for each theta - # Calculate x and y coordinates using the radius and theta values + b = max_radius / (total_rotations * 2 * np.pi) + end_theta = total_rotations * 2 * np.pi + theta = np.linspace(0, end_theta, num_points) + r = b * theta x = r * np.cos(theta) y = r * np.sin(theta) return x, y def run(protocol: protocol_api.ProtocolContext): - """ - Main function for running the protocol. - """ - json_params = load_experiment_parameters(INPUT_JSON_FILE) # load the parameters from the json file - single, multi = load_experiment_data(INPUT_CSV_FILE) # load data from the csv file, modified based on the channel of the pipette - - # Load pipettes - right_pipette_tipracks = [protocol.load_labware(load_name=json_params["right_pipette_tiprack_name"], location=i) for i in json_params["right_pipette_tiprack_slot"]] - right_pipette = protocol.load_instrument(instrument_name=json_params["right_pipette_name"], mount="right", tip_racks=right_pipette_tipracks) + """Main function for running the protocol.""" + json_params = load_json_data(INPUT_JSON_FILE) + csv_data = load_csv_data(INPUT_CSV_FILE) - left_pipette_tipracks = [protocol.load_labware(load_name=json_params["left_pipette_tiprack_name"], location=i) for i in json_params["left_pipette_tiprack_slot"]] - left_pipette = protocol.load_instrument(instrument_name=json_params["left_pipette_name"], mount="left", tip_racks=left_pipette_tipracks) - - pipette_media = right_pipette if choose_pipette_volume(single.media_volume) in json_params["right_pipette_name"] else left_pipette # Choose which pipette to use based on volume - pipette_channel_media = multi if choose_pipette_channel(str(pipette_media)) == 'multi' else single - pipette_sampling = right_pipette if choose_pipette_volume([10.0]) in json_params["right_pipette_name"] else left_pipette # Choose which pipette to use based on volume - pipette_channel_sampling = multi if choose_pipette_channel(str(pipette_sampling)) == 'multi' else single - - # Load hardware and labware - agar_labware = [protocol.load_labware(load_name=json_params["agar_plate_name"], - location=slot, - label=f"Agar Plate {i+1}") - for i, slot in enumerate(json_params["agar_plate_slot"])] - - mapping_dict = {int(slot): labware for slot, labware in zip(json_params["agar_plate_slot"], agar_labware)} # Create a mapping dictionary from slot to agar_labware - agar_plates = [mapping_dict[int(loc)] for loc in pipette_channel_sampling.agar_plate_location if int(loc) in mapping_dict] # Generate agar_plates list with repeating plates + loaded_pipettes = setup_pipettes(protocol, json_params) + pipette_media = select_pipette(loaded_pipettes, csv_data.media_volume) + pipette_sampling = select_pipette(loaded_pipettes, is_sampling=True) media_plate = protocol.load_labware(load_name = json_params["media_plate_name"], location = json_params["media_plate_slot"]) - culture_plate = protocol.load_labware(load_name = json_params["culture_plate_name"], location = json_params["culture_plate_slot"]) + culture_plate = protocol.load_labware(load_name = json_params["destination_plate_name"], location = json_params["destination_plate_slot"]) + + sampling_source_wells, sampling_destination_wells, locations = filter_data(pipette_sampling, csv_data.sampling_source_well, [], csv_data.destination_well, csv_data.agar_plate_location) + media_source_wells, media_volumes, media_destination_wells = filter_data(pipette_media, csv_data.media_source_well, csv_data.media_volume, csv_data.destination_well) + agar_labware = {int(slot): protocol.load_labware(load_name=json_params["agar_plate_name"], location=slot, label=f"Agar Plate {i+1}") + for i, slot in enumerate(json_params["agar_plate_slot"])} + agar_info = {slot: {'empty_plate_weight': weight, 'agar_plate_weight': agar_weight} + for slot, weight, agar_weight in zip(json_params["agar_plate_slot"], json_params["empty_agar_plate_weight"], json_params["agar_plate_weight"])} + agar_plates = [agar_labware[loc] for loc in locations] + empty_plate_weight = [agar_info[loc]['empty_plate_weight'] for loc in locations if loc in agar_info] + agar_plate_weight = [agar_info[loc]['agar_plate_weight'] for loc in locations if loc in agar_info] - pipette_media.transfer(volume=pipette_channel_media.media_volume, - source=[media_plate.wells_by_name()[well] for well in pipette_channel_media.media_source_well], - dest=[culture_plate.wells_by_name()[well] for well in pipette_channel_media.destination_well], + protocol.set_rail_lights(True) + + ########## DISTRIBUTE MEDIA ########## + pipette_media.transfer(volume=media_volumes, + source=[media_plate.wells_by_name()[well] for well in media_source_wells], + dest=[culture_plate.wells_by_name()[well] for well in media_destination_wells], new_tip="once") - if json_params["sampling_method"] == "spiral": - x_coords, y_coords = calculate_spiral_coords(json_params["spot_radius"]) - for plate, weight, well, culture_well in zip(agar_plates, pipette_channel_sampling.agar_plate_weight, pipette_channel_sampling.sampling_source_well, pipette_channel_sampling.destination_well): - pipette_sampling.pick_up_tip() - sampling_height = agar_height(weight, json_params["empty_agar_plate_weight"], json_params["agar_plate_area"], json_params["agar_density"]) - colony_well = plate.wells_by_name()[well] - for x,y in zip(x_coords, y_coords): - colony_sampling = colony_well.bottom(z=sampling_height-json_params["agar_stab_depth"]).move(types.Point(x=x, y=y)) + ########## SAMPLING ########## + for plate, source, destination, empty_weight, agar_weight in zip(agar_plates, sampling_source_wells, sampling_destination_wells, empty_plate_weight, agar_plate_weight): + pipette_sampling.pick_up_tip() + sampling_height = agar_height(agar_weight, empty_weight, json_params["agar_plate_area"], json_params["agar_density"], json_params["agar_pierce_depth"]) + colony_well = plate.wells_by_name()[source] + + if json_params["sampling_method"] == "spiral": + x_coords, y_coords = calculate_spiral_coords(json_params["spot_radius"]) + for x, y in zip(x_coords, y_coords): + colony_sampling = colony_well.bottom(z=sampling_height).move(types.Point(x=x, y=y)) pipette_sampling.move_to(colony_sampling) - pipette_sampling.move_to(media_plate[culture_well].bottom()) - pipette_sampling.mix(repetitions=3, rate=4) # Inoculation - pipette_sampling.drop_tip() - - elif json_params["sampling_method"] == "pierce": - for plate, weight, well, culture_well in zip(agar_plates, pipette_channel_sampling.agar_plate_weight, pipette_channel_sampling.source_well, pipette_channel_sampling.destination_well): - pipette_sampling.pick_up_tip() - sampling_height = agar_height(weight, json_params["empty_agar_plate_weight"], json_params["agar_plate_area"], json_params["agar_density"]) - print(sampling_height) - colony_well = plate.wells_by_name()[well] - colony_sampling = colony_well.bottom(z=sampling_height-json_params["agar_stab_depth"]) + elif json_params["sampling_method"] == "pierce": + colony_sampling = colony_well.bottom(z=sampling_height) pipette_sampling.move_to(colony_sampling) - pipette_sampling.move_to(media_plate[culture_well].bottom()) - pipette_sampling.mix(repetitions=3, rate=4) # Inoculation - pipette_sampling.drop_tip() \ No newline at end of file + + pipette_sampling.move_to(media_plate[destination].bottom()) + pipette_sampling.mix(repetitions=2, rate=4) + pipette_sampling.drop_tip() + + protocol.set_rail_lights(False) \ No newline at end of file diff --git a/assets/testdata/protocol-1-config.json b/assets/testdata/protocol-1-config.json index e93f23a..fb31606 100644 --- a/assets/testdata/protocol-1-config.json +++ b/assets/testdata/protocol-1-config.json @@ -12,7 +12,7 @@ "media_plate_name": "armadillo_96_wellplate_200ul_pcr_full_skirt", "media_plate_slot": 1, "destination_plate_name":"armadillo_96_wellplate_200ul_pcr_full_skirt", - "destination_plate_slot": 4, + "destination_plate_slot": "thermocycler", "pre_shock_incubation_temp":4, "pre_shock_incubation_time":30, "heat_shock_temp":42, diff --git a/assets/testdata/protocol-1-data.csv b/assets/testdata/protocol-1-data.csv index 4daadc8..5af8774 100644 --- a/assets/testdata/protocol-1-data.csv +++ b/assets/testdata/protocol-1-data.csv @@ -1,4 +1,17 @@ dna_id,dna_well,dna_volume,cells_id,cells_well,cells_volume,media_id,media_well,media_volume,destination_well -pJKR-5,A1,1,DH5a,A11,10,SOC,A12,50,A1 -pJKR-5,B1,1,DH5a,B11,10,SOC,B12,50,B1 -pJKR-5,A1,1,DH5a,A10,10,SOC,A12,50,A2 \ No newline at end of file +pEX01,A1,3,DH5a,A2,30,SOC,A3,150,A1 +pEX02,B1,3,DH5a,B2,30,SOC,B3,150,B1 +pEX03,C1,3,DH5a,C2,30,SOC,C3,150,C1 +pEX04,D1,3,DH5a,D2,30,SOC,D3,150,D1 +pEX05,E1,3,DH5a,E2,30,SOC,E3,150,E1 +pEX06,F1,3,DH5a,F2,30,SOC,F3,150,F1 +pEX07,G1,3,DH5a,G2,30,SOC,G3,150,G1 +pEX08,H1,3,DH5a,H2,30,SOC,H3,150,H1 +pEX01,A1,3,DH5a,A2,30,SOC,A4,150,A2 +pEX02,B1,3,DH5a,B2,30,SOC,B4,150,B2 +pEX03,C1,3,DH5a,C2,30,SOC,C4,150,C2 +pEX04,D1,3,DH5a,D2,30,SOC,D4,150,D2 +pEX05,E1,3,DH5a,E2,30,SOC,E4,150,E2 +pEX06,F1,3,DH5a,F2,30,SOC,F4,150,F2 +pEX07,G1,3,DH5a,G2,30,SOC,G4,150,G2 +pEX08,H1,3,DH5a,H2,30,SOC,H4,150,H2 \ No newline at end of file diff --git a/assets/testdata/protocol-2-config.json b/assets/testdata/protocol-2-config.json index 503e23f..fe80345 100644 --- a/assets/testdata/protocol-2-config.json +++ b/assets/testdata/protocol-2-config.json @@ -6,10 +6,11 @@ "source_plate_name":"armadillo_96_wellplate_200ul_pcr_full_skirt", "source_plate_slot":"1", "agar_plate_name":"armadillo_96_wellplate_200ul_pcr_full_skirt", - "agar_plate_slots":[2], + "agar_plate_slot":[2], "agar_plate_area":9469.2, - "empty_agar_plate_weight":38.92, - "agar_density":0.0008975, - "additional_volume":5, + "empty_agar_plate_weight":[38.92], + "agar_plate_weight":[68.35], + "agar_density":0.911, + "additional_volume":1, "spotting_height":0 } \ No newline at end of file diff --git a/assets/testdata/protocol-2-data.csv b/assets/testdata/protocol-2-data.csv index a8b3c67..e0ae840 100644 --- a/assets/testdata/protocol-2-data.csv +++ b/assets/testdata/protocol-2-data.csv @@ -1,7 +1,17 @@ -id,agar_plate_location,source_well,destination_well,spotting_volume,agar_plate_weight -pJKR-BL21(DE3),2,A1,A1,5,76.65 -pJKR-BL21(DE3),2,A1,A2,5,76.65 -pJKR-BL21(DE3),2,A1,A3,5,76.65 -pJKR-BL21(DE3),2,A1,A4,5,76.65 -pUC19-BL21(DE3),2,A2,A5,5,76.65 -no-DNA-BL21(DE3),2,A3,A6,5,76.65 \ No newline at end of file +id,agar_plate_location,source_well,destination_well,spotting_volume +puc19-DH5a,2,A1,A1,5 +puc19-DH5a,2,B1,B1,5 +puc19-DH5a,2,C1,C1,5 +puc19-DH5a,2,D1,D1,5 +puc19-DH5a,2,E1,E1,5 +puc19-DH5a,2,F1,F1,5 +puc19-DH5a,2,G1,G1,5 +puc19-DH5a,2,H1,H1,5 +puc19-DH5a,2,A2,A2,5 +puc19-DH5a,2,B2,B2,5 +puc19-DH5a,2,C2,C2,5 +puc19-DH5a,2,D2,D2,5 +puc19-DH5a,2,E2,E2,5 +puc19-DH5a,2,F2,F2,5 +puc19-DH5a,2,G2,G2,5 +puc19-DH5a,2,H2,H2,5 \ No newline at end of file diff --git a/assets/testdata/protocol-3-config.json b/assets/testdata/protocol-3-config.json index 2b33bb3..6514681 100644 --- a/assets/testdata/protocol-3-config.json +++ b/assets/testdata/protocol-3-config.json @@ -7,14 +7,18 @@ "left_pipette_tiprack_slot":[6], "media_plate_name": "usascientific_96_wellplate_2.4ml_deep", "media_plate_slot":2, - "culture_plate_name":"usascientific_96_wellplate_2.4ml_deep", - "culture_plate_slot":3, + "destination_plate_name":"usascientific_96_wellplate_2.4ml_deep", + "destination_plate_slot":3, "agar_plate_name":"armadillo_96_wellplate_200ul_pcr_full_skirt", "agar_plate_slot": [1], "agar_plate_area":9469.2, - "empty_agar_plate_weight":38.92, - "agar_density":0.0008975, - "agar_stab_depth":0, + "empty_agar_plate_weight":[38.92], + "agar_plate_weight":[68.92], + "agar_density":0.911, + "agar_pierce_depth":0, "sampling_method": "spiral", - "spot_radius": 2.5 + "spot_radius": 2.5, + "spiral_points": 25, + "spiral_rotations":3 + } \ No newline at end of file diff --git a/assets/testdata/protocol-3-data.csv b/assets/testdata/protocol-3-data.csv index c1c4ab7..4355b3c 100644 --- a/assets/testdata/protocol-3-data.csv +++ b/assets/testdata/protocol-3-data.csv @@ -1,2 +1,9 @@ -id,agar_plate_location,agar_plate_weight,media_source_well,sampling_source_well,media_volume,destination_well -pJKR-BL21(DE3),1,67.1377,A1,A1,500,A1 \ No newline at end of file +id,agar_plate_location,media_source_well,media_volume,sampling_source_well,destination_well +puc19-DH5a,1,A1,500,A1,A1 +puc19-DH5a,1,B1,500,B1,B1 +puc19-DH5a,1,C1,500,C1,C1 +puc19-DH5a,1,D1,500,D1,D1 +puc19-DH5a,1,E1,500,E1,E1 +puc19-DH5a,1,F1,500,F1,F1 +puc19-DH5a,1,G1,500,G1,G1 +puc19-DH5a,1,G1,500,H1,H1 \ No newline at end of file diff --git a/assets/testdata/protocol-4-config.json b/assets/testdata/protocol-4-config.json index f6f84bb..9f21082 100644 --- a/assets/testdata/protocol-4-config.json +++ b/assets/testdata/protocol-4-config.json @@ -7,12 +7,12 @@ "left_pipette_tiprack_slot":[9], "left_pipette_name":"p300_multi_gen2", "left_pipette_mount":"left", - "blank_plate_name":"usascientific_96_wellplate_2.4ml_deep", - "blank_plate_slot":3, + "media_plate_name":"usascientific_96_wellplate_2.4ml_deep", + "media_plate_slot":4, "culture_plate_name":"usascientific_96_wellplate_2.4ml_deep", - "culture_plate_slot":3, + "culture_plate_slot":1, "inducer_plate_name":"usascientific_96_wellplate_2.4ml_deep", - "inducer_plate_slot":3, + "inducer_plate_slot":2, "destination_plate_name":"armadillo_96_wellplate_200ul_pcr_full_skirt", - "destination_plate_slot":2 + "destination_plate_slot":3 } \ No newline at end of file diff --git a/assets/testdata/protocol-4-data.csv b/assets/testdata/protocol-4-data.csv index ad26767..bb82cbc 100644 --- a/assets/testdata/protocol-4-data.csv +++ b/assets/testdata/protocol-4-data.csv @@ -1,3 +1,3 @@ -source_id,source_well,source_volume,destination_id,destination_well,destination_volume -pJKR-BL21(DE3),A1,10,fresh-LB,A6,500 -pJKR-BL21(DE3),B1,10,fresh-LB,B6,500 \ No newline at end of file +culture_id,culture_well,culture_volume,media_id,media_well,media_volume,inducer_id,inducer_well,inducer_volume,destination_well +pJKR-BL21(DE3),A1,10,fresh-LB,A1,500,IPTG-0.1M,A1,2,A1 +pJKR-BL21(DE3),A2,10,fresh-LB,A2,500,IPTG-0.1M,A2,2,A2 \ No newline at end of file diff --git a/bin/create-labware.py b/bin/create-labware.py index 7436a30..a6ccd4f 100755 --- a/bin/create-labware.py +++ b/bin/create-labware.py @@ -121,35 +121,40 @@ def create_csv_protocol_4(data_csv, parameters_json, output_csv): plate_slots = json.load(f) df_all_reactants = [] - for reactant in ["blank", "culture", "inducer"]: + for reactant in ["media", "culture", "inducer"]: well_col = f"{reactant}_well" volume_col = f"{reactant}_volume" - id_col = f"ID" + id_col = f"{reactant}_id" df_id_well_volume = df.groupby(well_col).agg({volume_col: "sum", id_col: "first"}).reset_index() # sum the volumes of reactants - df_id_well_volume.rename(columns={id_col: "ID", well_col: "well_name", volume_col: "volume"}, inplace=True) + df_id_well_volume.rename(columns={id_col: "id", well_col: "well_name", volume_col: "volume"}, inplace=True) df_id_well_volume["location"] = plate_slots[f"{reactant}_plate_slot"] df_id_well_volume["labware"] = plate_slots[f"{reactant}_plate_name"] - df_all_reactants.append(df_id_well_volume) - - # Map these IDs to rows where both culture_volume and inducer_volume > 0 - def map_ids(row): - if row["culture_volume"] > 0 and row["inducer_volume"] > 0: - # Find the matching "ID" from the "culture_only_ids" series - culture_id = df.loc[(df["culture_well"] == row["culture_well"]) & (df["inducer_volume"] == 0), "ID"] - if not culture_id.empty: - return row["ID"] + "+" + culture_id.iloc[0] - return row["ID"] + df_all_reactants.append(df_id_well_volume) - df["combined_ID"] = df.apply(map_ids, axis=1) + # df['combined_id'] = df.apply(map_ids, axis=1) + # print(df['combined_id']) - # induction dataframe + # Map these IDs to rows where both culture_volume and inducer_volume > 0 + # def map_ids(row): + # if row["culture_volume"] > 0 and row["inducer_volume"] > 0: + # # Find the matching "ID" from the "culture_only_ids" series + # culture_id = df.loc[(df["culture_well"] == row["culture_well"]) & (df["inducer_volume"] == 0), "id"] + # print(culture_id) + # if not culture_id.empty: + # return row["id"] + "+" + culture_id.iloc[0] + # return row["id"] + + # df["combined_id"] = df.apply(map_ids, axis=1) + # print(df["combined_id"]) + + # # induction dataframe df_induction = pd.DataFrame({ - "ID": df["combined_ID"], + "id": df["culture_id"], "well_name": df["destination_well"], - "volume": df[["blank_volume", "culture_volume", "inducer_volume"]].sum(axis=1), + "volume": df[["media_volume", "culture_volume", "inducer_volume"]].sum(axis=1), "labware": plate_slots["destination_plate_name"], "location": plate_slots["destination_plate_slot"] }) @@ -157,9 +162,10 @@ def map_ids(row): df_all_reactants.append(df_induction) result_df = pd.concat(df_all_reactants, ignore_index=True) # Concatenate the grouped dataframes vertically - result_df = result_df.reindex(columns=["ID", "location", "labware", "well_name", "volume"]) # Reorder columns + result_df = result_df.reindex(columns=["id", "location", "labware", "well_name", "volume"]) # Reorder columns result_df.to_csv(output_csv, index=False) # write to a CSV file + def main(): args = docopt(__doc__) csv_data = args[""] @@ -174,8 +180,8 @@ def main(): elif "3" in csv_data: create_csv_protocol_3(csv_data, json_data, csv_output) - elif "4" in csv_data: + if "4" in csv_data: create_csv_protocol_4(csv_data, json_data, csv_output) if __name__ == "__main__": - main() + main() \ No newline at end of file diff --git a/main.nf b/main.nf index 7f079d9..12256c5 100644 --- a/main.nf +++ b/main.nf @@ -11,9 +11,9 @@ workflow PROTOCOL_1 { tuple(file("$params.protocol_1_config"), file("$params.protocol_1_data")), file("$params.protocol_template_dir") ) - // SIMULATE_PROTOCOL_1( - // MAKE_PROTOCOL_1.out - // ) + SIMULATE_PROTOCOL_1( + MAKE_PROTOCOL_1.out + ) CREATE_LABWARE_CSV_1( tuple(file("$params.protocol_1_data"), file("$params.protocol_1_config")) ) @@ -30,14 +30,14 @@ workflow PROTOCOL_1 { workflow PROTOCOL_2 { - // PROTOCOL 2 - SPOTTING + // PROTOCOL 2 - SELECTION MAKE_PROTOCOL_2( tuple(file("$params.protocol_2_config"), file("$params.protocol_2_data")), file("$params.protocol_template_dir") ) - // SIMULATE_PROTOCOL_2( - // MAKE_PROTOCOL_2.out - // ) + SIMULATE_PROTOCOL_2( + MAKE_PROTOCOL_2.out + ) CREATE_LABWARE_CSV_2( tuple(file("$params.protocol_2_data"), file("$params.protocol_2_config")) ) @@ -61,9 +61,9 @@ workflow PROTOCOL_3 { tuple(file("$params.protocol_3_config"), file("$params.protocol_3_data")), file("$params.protocol_template_dir") ) - // SIMULATE_PROTOCOL_3( - // MAKE_PROTOCOL_3.out - // ) + SIMULATE_PROTOCOL_3( + MAKE_PROTOCOL_3.out + ) CREATE_LABWARE_CSV_3( tuple(file("$params.protocol_3_data"), file("$params.protocol_3_config")) ) @@ -87,9 +87,9 @@ workflow PROTOCOL_4 { tuple(file("$params.protocol_4_config"), file("$params.protocol_4_data")), file("$params.protocol_template_dir") ) - // SIMULATE_PROTOCOL_4( - // MAKE_PROTOCOL_4.out - // ) + SIMULATE_PROTOCOL_4( + MAKE_PROTOCOL_4.out + ) CREATE_LABWARE_CSV_4( tuple(file("$params.protocol_4_data"), file("$params.protocol_4_config")) ) @@ -110,5 +110,5 @@ workflow { PROTOCOL_1() PROTOCOL_2() PROTOCOL_3() - // PROTOCOL_4() + PROTOCOL_4() } \ No newline at end of file