Merge branch 'main' into stable

icfree/__init__.py

@@ -0,0 +1,5 @@
+from icfree.instructor import main as instructor
+from icfree.sampler import main as sampler
+from icfree.plate_designer import main as plate_designer
+
+__all__ = ['instructor', 'sampler', 'plate_designer']

icfree/__main__.py

@@ -1,6 +1,11 @@
 import argparse
-import subprocess
 import os
+import sys
+
+# Import main functions from the modules
+from icfree.sampler import main as sampler_main
+from icfree.plate_designer import main as plate_designer_main
+from icfree.instructor import main as instructor_main
 
 def generate_snakefile(args):
     output_folder = args.plate_designer_output_folder
@@ -69,6 +74,38 @@ def generate_snakefile(args):
     with open('Snakefile', 'w') as file:
         file.write(snakefile_content)
 
+def run_snakemake(args):
+    # Simulating Snakemake workflow by directly calling the main functions
+
+    # SAMPLER
+    sampler_main(args.sampler_input_filename, args.plate_designer_sampling_file, args.sampler_nb_samples, args.sampler_step, args.sampler_seed)
+
+    # PLATE_DESIGNER
+    plate_designer_main(
+        args.plate_designer_sampling_file,
+        args.plate_designer_sample_volume,
+        args.plate_designer_start_well_src_plt,
+        args.plate_designer_start_well_dst_plt,
+        "16x24",  # Assuming default plate dimensions
+        args.plate_designer_well_capacity,
+        args.plate_designer_default_well_capacity,
+        args.plate_designer_dead_volumes,
+        args.plate_designer_default_dead_volume,
+        args.plate_designer_num_replicates,
+        args.plate_designer_output_folder
+    )
+
+    # INSTRUCTOR
+    instructor_main(
+        f"{args.plate_designer_output_folder}/source_plate.csv",
+        f"{args.plate_designer_output_folder}/destination_plate.csv",
+        args.instructor_output_filename,
+        args.instructor_source_plate_type,
+        args.instructor_max_transfer_volume,
+        args.instructor_split_threshold,
+        args.instructor_split_components
+    )
+
 def main():
     parser = argparse.ArgumentParser(description="Generate and run a Snakemake workflow based on user parameters.")
     parser.add_argument('--sampler_input_filename', required=True, help="Input filename for the SAMPLER step.")
@@ -95,8 +132,8 @@ def main():
 
     generate_snakefile(args)
 
-    # Run Snakemake
-    subprocess.run(["snakemake", "--cores", "1"])
+    # Run Snakemake simulation by calling main functions directly
+    run_snakemake(args)
 
 if __name__ == "__main__":
     main()

icfree/instructor.py

@@ -92,51 +92,63 @@ def split_volumes(source_well, dest_well, volume, component, plate_type):
     instructions_df = pd.DataFrame(instructions)
     return instructions_df.groupby('Sample ID', as_index=False).apply(lambda x: x)
 
-def main():
+def main(source_plate_file, destination_plate_file, output_file, source_plate_type="default:384PP_AQ_GP3",
+         max_transfer_volume=None, split_threshold=None, split_components=None):
     """
-    Main function to parse arguments, read input files, generate ECHO instructions, and write the output to files.
-    """
-    parser = argparse.ArgumentParser(description="Generate ECHO liquid handler instructions.")
-    parser.add_argument("source_plate_file", type=str, help="Path to the source plate file.")
-    parser.add_argument("destination_plate_file", type=str, help="Path to the destination plate file.")
-    parser.add_argument("output_file", type=str, help="Path to the output instructions file.")
-    parser.add_argument("--source_plate_type", type=str, default="default:384PP_AQ_GP3",
-                        help="Comma-separated list of component and plate type pairs, e.g., 'Component_1:384PP_AQ_CP,Component_2:384PP_AQ_GP3'. Default for all is 384PP_AQ_GP3.")
-    parser.add_argument("--max_transfer_volume", type=int, help="Maximum volume for a single transfer. If not specified, no splitting will be performed.")
-    parser.add_argument("--split_threshold", type=int, help="Volume threshold above which transfers need to be split. If not specified, no splitting will be performed.")
-    parser.add_argument("--split_components", type=str, help="Comma-separated list of component names to create separate files for.")
-
-    args = parser.parse_args()
+    Main function to read input files, generate ECHO instructions, and write the output to files.
     
+    Parameters:
+    - source_plate_file: Path to the source plate file.
+    - destination_plate_file: Path to the destination plate file.
+    - output_file: Path to the output instructions file.
+    - source_plate_type: Comma-separated list of component and plate type pairs.
+    - max_transfer_volume: Maximum volume for a single transfer. If not specified, no splitting will be performed.
+    - split_threshold: Volume threshold above which transfers need to be split. If not specified, no splitting will be performed.
+    - split_components: Comma-separated list of component names to create separate files for.
+    """
     # Parse the source plate types from the string argument
-    source_plate_types = parse_plate_types(args.source_plate_type)
+    source_plate_types = parse_plate_types(source_plate_type)
 
     # Read the source and destination plate data from CSV files
-    source_plate_df = pd.read_csv(args.source_plate_file)
-    destination_plate_df = pd.read_csv(args.destination_plate_file)
+    source_plate_df = pd.read_csv(source_plate_file)
+    destination_plate_df = pd.read_csv(destination_plate_file)
 
     # Generate the ECHO instructions
     instructions_df = generate_echo_instructions(source_plate_df, destination_plate_df, source_plate_types,
-                                                 args.max_transfer_volume, args.split_threshold)
+                                                 max_transfer_volume, split_threshold)
 
     # Handle splitting of components into separate files if specified
-    if args.split_components:
-        split_components = args.split_components.split(',')
-        for component in split_components:
+    if split_components:
+        split_components_list = split_components.split(',')
+        for component in split_components_list:
             component_df = instructions_df[instructions_df['Sample ID'] == component]
-            output_file = f"{os.path.splitext(args.output_file)[0]}_{component}.csv"
-            component_df.to_csv(output_file, index=False)
-            print(f"Instructions for {component} have been generated and saved to {output_file}")
+            component_output_file = f"{os.path.splitext(output_file)[0]}_{component}.csv"
+            component_df.to_csv(component_output_file, index=False)
+            print(f"Instructions for {component} have been generated and saved to {component_output_file}")
 
         # Write remaining components to the original output file
-        remaining_df = instructions_df[~instructions_df['Sample ID'].isin(split_components)]
+        remaining_df = instructions_df[~instructions_df['Sample ID'].isin(split_components_list)]
         if not remaining_df.empty:
-            remaining_df.to_csv(args.output_file, index=False)
-            print(f"Instructions for remaining components have been generated and saved to {args.output_file}")
+            remaining_df.to_csv(output_file, index=False)
+            print(f"Instructions for remaining components have been generated and saved to {output_file}")
     else:
         # Write all instructions to the original output file
-        instructions_df.to_csv(args.output_file, index=False)
-        print(f"Instructions have been generated and saved to {args.output_file}")
+        instructions_df.to_csv(output_file, index=False)
+        print(f"Instructions have been generated and saved to {output_file}")
 
 if __name__ == "__main__":
-    main()
+    import sys
+    parser = argparse.ArgumentParser(description="Generate ECHO liquid handler instructions.")
+    parser.add_argument("source_plate_file", type=str, help="Path to the source plate file.")
+    parser.add_argument("destination_plate_file", type=str, help="Path to the destination plate file.")
+    parser.add_argument("output_file", type=str, help="Path to the output instructions file.")
+    parser.add_argument("--source_plate_type", type=str, default="default:384PP_AQ_GP3",
+                        help="Comma-separated list of component and plate type pairs, e.g., 'Component_1:384PP_AQ_CP,Component_2:384PP_AQ_GP3'. Default for all is 384PP_AQ_GP3.")
+    parser.add_argument("--max_transfer_volume", type=int, help="Maximum volume for a single transfer. If not specified, no splitting will be performed.")
+    parser.add_argument("--split_threshold", type=int, help="Volume threshold above which transfers need to be split. If not specified, no splitting will be performed.")
+    parser.add_argument("--split_components", type=str, help="Comma-separated list of component names to create separate files for.")
+
+    args = parser.parse_args()
+
+    main(args.source_plate_file, args.destination_plate_file, args.output_file, args.source_plate_type,
+         args.max_transfer_volume, args.split_threshold, args.split_components)

icfree/plate_designer.py

@@ -174,23 +174,51 @@ def write_output_files(source_data, destination_data, output_folder):
     print(f"Destination plate data written to {destination_path}")
     print(f"Source plate data written to {source_path}")
 
-def main():
+def main(
+    sampling_file, sample_volume, start_well_src_plt='A1', start_well_dst_plt='A1', 
+    plate_dims='16x24', well_capacity='', default_well_capacity=60000, 
+    dead_volumes='', default_dead_volume=15000, num_replicates=1, output_folder='.'):
     """
     Main function to prepare source and destination well-plate mappings and write the results to files.
+    
+    Args:
+        sampling_file (str): Path to the sampling file.
+        sample_volume (int): Wanted sample volume in the destination plate.
+        start_well_src_plt (str): Starting well for the source plate.
+        start_well_dst_plt (str): Starting well for the destination plate.
+        plate_dims (str): Plate dimensions (Format: NxM).
+        well_capacity (str): Well capacities for specific components in format component1=capacity1,component2=capacity2,...
+        default_well_capacity (int): Default well capacity in nL for components not specified in well_capacity.
+        dead_volumes (str): Dead volumes for specific components in format component1=volume1,component2=volume2,...
+        default_dead_volume (int): Default dead volume in nL for the source plate.
+        num_replicates (int): Number of wanted replicates.
+        output_folder (str): Output folder for the result files.
     """
-    args = parse_args()
-
     # Read the sampling data from the specified file
-    sampling_data = pd.read_csv(args.sampling_file)
+    sampling_data = pd.read_csv(sampling_file)
 
     # Prepare the destination plate data
-    destination_data = prepare_destination_plate(sampling_data, args.start_well_dst_plt, args.plate_dims, args.sample_volume, args.num_replicates)
+    destination_data = prepare_destination_plate(sampling_data, start_well_dst_plt, plate_dims, sample_volume, num_replicates)
 
     # Prepare the source plate data
-    source_data = prepare_source_plate(destination_data, args.dead_volumes, args.default_dead_volume, args.well_capacity, args.default_well_capacity, args.start_well_src_plt)
+    source_data = prepare_source_plate(destination_data, dead_volumes, default_dead_volume, well_capacity, default_well_capacity, start_well_src_plt)
 
     # Write the output files to the specified output folder
-    write_output_files(source_data, destination_data, Path(args.output_folder))
+    write_output_files(source_data, destination_data, Path(output_folder))
 
 if __name__ == "__main__":
-    main()
+    # If the script is executed directly, parse command-line arguments
+    args = parse_args()
+    main(
+        sampling_file=args.sampling_file,
+        sample_volume=args.sample_volume,
+        start_well_src_plt=args.start_well_src_plt,
+        start_well_dst_plt=args.start_well_dst_plt,
+        plate_dims=args.plate_dims,
+        well_capacity=args.well_capacity,
+        default_well_capacity=args.default_well_capacity,
+        dead_volumes=args.dead_volumes,
+        default_dead_volume=args.default_dead_volume,
+        num_replicates=args.num_replicates,
+        output_folder=args.output_folder
+    )

icfree/sampler.py

@@ -48,6 +48,24 @@ def generate_lhs_samples(input_file, num_samples, step, seed=None):
     samples_df = pd.DataFrame(samples, columns=components_df['Component'])
     return samples_df
 
+def main(input_file, output_file, num_samples, step=2.5, seed=None):
+    """
+    Main function to generate LHS samples and save them to a CSV file.
+    
+    Parameters:
+    - input_file: Path to the input file containing components and their max values.
+    - output_file: Path to the output CSV file where samples will be written.
+    - num_samples: Number of samples to generate.
+    - step: Step size for creating discrete ranges (default: 2.5).
+    - seed: Random seed for reproducibility (optional).
+    """
+    # Generate LHS samples
+    samples_df = generate_lhs_samples(input_file, num_samples, step, seed)
+
+    # Write the samples to a CSV file
+    samples_df.to_csv(output_file, index=False)
+    print(f"Generated {num_samples} samples and saved to {output_file}")
+
 if __name__ == "__main__":
     # Setup command line argument parsing
     parser = argparse.ArgumentParser(description="Generate Latin Hypercube Samples for given components.")
@@ -60,9 +78,5 @@ def generate_lhs_samples(input_file, num_samples, step, seed=None):
     # Parse arguments
     args = parser.parse_args()
 
-    # Generate LHS samples
-    samples_df = generate_lhs_samples(args.input_file, args.num_samples, args.step, args.seed)
-
-    # Write the samples to a CSV file
-    samples_df.to_csv(args.output_file, index=False)
-    print(f"Generated {args.num_samples} samples and saved to {args.output_file}")
+    # Run the main function with the parsed arguments
+    main(args.input_file, args.output_file, args.num_samples, args.step, args.seed)

tests/test_plate_designer.py

@@ -142,7 +142,20 @@ def test_main(self):
 
         with patch.object(sys, 'argv', test_args):
             from icfree.plate_designer import main
-            main()
+            args = parse_args()
+            main(
+                sampling_file=args.sampling_file,
+                sample_volume=args.sample_volume,
+                start_well_src_plt=args.start_well_src_plt,
+                start_well_dst_plt=args.start_well_dst_plt,
+                plate_dims=args.plate_dims,
+                well_capacity=args.well_capacity,
+                default_well_capacity=args.default_well_capacity,
+                dead_volumes=args.dead_volumes,
+                default_dead_volume=args.default_dead_volume,
+                num_replicates=args.num_replicates,
+                output_folder=args.output_folder
+            )
             self.assertTrue((Path(self.temp_dir.name) / 'destination_plate.csv').exists())
             self.assertTrue((Path(self.temp_dir.name) / 'source_plate.csv').exists())