Merge pull request #43 from AD-SDL/data_improvements

Data improvements

Makefile

@@ -22,10 +22,10 @@ checks: # Runs all the pre-commit checks
 	@pre-commit install
 	@pre-commit run --all-files || { echo "Checking fixes\n" ; pre-commit run --all-files; }
 
-# test: init .env paths # Runs all the tests
-# 	@docker compose -f wei.compose.yaml --env-file .env up --build -d
-# 	@docker compose -f wei.compose.yaml --env-file .env exec ot2_module pytest -p no:cacheprovider -m "not hardware" ot2_module
-# 	@docker compose -f wei.compose.yaml --env-file .env down
+test: init .env paths #test
+	@docker compose -f wei.compose.yaml --env-file .env up --build -d
+	@docker compose -f wei.compose.yaml --env-file .env exec ot2_module pytest -p no:cacheprovider -m "not hardware" ot2_module
+	@docker compose -f wei.compose.yaml --env-file .env down
 
 # hardware_test: init .env paths # Runs all the tests
 # 	@docker compose -f wei.compose.yaml --env-file .env up --build -d

ot2_driver/ot2_driver_http.py

@@ -14,6 +14,8 @@
 
 
 class RobotStatus(Enum):
+    """status of ot2"""
+
     IDLE = "idle"
     RUNNING = "running"
     FINISHING = "finishing"
@@ -23,6 +25,8 @@ class RobotStatus(Enum):
 
 
 class RunStatus(Enum):
+    """status of run on ot2"""
+
     IDLE = "idle"
     RUNNING = "running"
     FINISHING = "finishing"
@@ -327,6 +331,7 @@ def reset_robot_data(self):
                 requests.delete(url=delete_url + run["runID"], headers=self.headers)
 
     def change_lights_status(self, status: bool = False):
+        """switch the lights"""
         change_lights_url = f"{self.base_url}/robot/lights"
         payload = {"on": status}
 

ot2_driver/protopiler/protopiler.py

@@ -41,9 +41,7 @@ class ProtoPiler:
     def __init__(
         self,
         config_path: Optional[PathLike] = None,
-        template_dir: PathLike = (
-            Path(__file__).parent.resolve() / "protocol_templates"
-        ),
+        template_dir: PathLike = None,
         resource_file: Optional[PathLike] = None,
     ) -> None:
         """Can initialize with the resources we need, or it can be done after initialization
@@ -57,7 +55,10 @@ def __init__(
         resource_file : Optional[PathLike], optional
             path to the resource file, if using a config and it does not exist, it will be created, by default None
         """
-        self.template_dir = template_dir
+        if template_dir is None:
+            self.template_dir = Path(__file__).parent.resolve() / "protocol_templates"
+        else:
+            self.template_dir = template_dir
         self.resource_file = resource_file
 
         if self.resource_file:
@@ -377,9 +378,7 @@ def yaml_to_protocol(
         self,
         config_path: Optional[PathLike] = None,
         payload: Optional[Dict] = None,
-        protocol_out_path: PathLike = Path(
-            f"./protocol_{datetime.now().strftime('%Y%m%d-%H%M%S')}.py"
-        ),
+        protocol_out_path: PathLike = None,
         resource_file: Optional[PathLike] = None,
         resource_file_out: Optional[PathLike] = None,
         write_resources: bool = True,
@@ -410,7 +409,10 @@ def yaml_to_protocol(
         Tuple[Path]
             returns the path to the protocol.py file as well as the resource file (if it does not exist, None)
         """
-
+        if protocol_out_path is None:
+            protocol_out_path = Path(
+                f"./protocol_{datetime.now().strftime('%Y%m%d-%H%M%S')}.py"
+            )
         if not self.config:
             self.load_config(config_path)
 
@@ -924,6 +926,42 @@ def _create_commands(self, payload: Optional[Dict]) -> List[str]:
                     and isinstance(command_block.reps, int)
                     and isinstance(command_block.mix_volume, float)
                 ):
+                    pipette_mount = self.resource_manager.determine_pipette(
+                        command_block.mix_volume, False
+                    )
+                    if pipette_mount is None:
+                        raise Exception(
+                            f"No pipette available for {block_name} with volume: {command_block.mix_volume}"
+                        )
+                    # TODO: make more robust
+                    # # check for tip
+                    if not tip_loaded[pipette_mount]:
+                        load_command = pick_tip_template.replace(
+                            "#pipette#", f'pipettes["{pipette_mount}"]'
+                        )
+                        pipette_name = self.resource_manager.mount_to_pipette[
+                            pipette_mount
+                        ]
+
+                        # TODO: define flag to grab from specific well or just use the ones defined by the OT2
+                        if True:
+                            (
+                                rack_location,
+                                well_location,
+                            ) = self.resource_manager.get_next_tip(pipette_name, 1)
+                            location_string = (
+                                f'deck["{rack_location}"].wells()[{well_location}]'
+                            )
+                            load_command = load_command.replace(
+                                "#location#", location_string
+                            )
+                        else:
+                            load_command = load_command.replace("#location#", "")
+                            self.resource_manager.update_tip_usage(pipette_name)
+
+                        commands.append(load_command)
+                        tip_loaded[pipette_mount] = True
+
                     mix_command = mix_template.replace(
                         "#reps#", str(command_block.reps)
                     )
@@ -985,6 +1023,42 @@ def _create_commands(self, payload: Optional[Dict]) -> List[str]:
                     else:
                         mix_reps = command_block.reps
 
+                    pipette_mount = self.resource_manager.determine_pipette(
+                        command_block.mix_volume, False
+                    )
+
+                    if pipette_mount is None:
+                        raise Exception(
+                            f"No pipette available for {block_name} with volume: {command_block.mix_volume}"
+                        )
+                    # # check for tip
+                    if not tip_loaded[pipette_mount]:
+                        load_command = pick_tip_template.replace(
+                            "#pipette#", f'pipettes["{pipette_mount}"]'
+                        )
+                        pipette_name = self.resource_manager.mount_to_pipette[
+                            pipette_mount
+                        ]
+
+                        # TODO: define flag to grab from specific well or just use the ones defined by the OT2
+                        if True:
+                            (
+                                rack_location,
+                                well_location,
+                            ) = self.resource_manager.get_next_tip(pipette_name, 1)
+                            location_string = (
+                                f'deck["{rack_location}"].wells()[{well_location}]'
+                            )
+                            load_command = load_command.replace(
+                                "#location#", location_string
+                            )
+                        else:
+                            load_command = load_command.replace("#location#", "")
+                            self.resource_manager.update_tip_usage(pipette_name)
+
+                        commands.append(load_command)
+                        tip_loaded[pipette_mount] = True
+
                     for loc, mix_vols, rep in zip(locations, mix_volumes, mix_reps):
                         mix_command = mix_template.replace("#reps#", str(rep))
                         mix_command = mix_command.replace("#volume#", str(mix_vols))
@@ -1007,11 +1081,15 @@ def _create_commands(self, payload: Optional[Dict]) -> List[str]:
             elif isinstance(command_block, Replace_Tip):
                 if not isinstance(command_block.replace_tip, bool):
                     raise Exception("replace_tip must be bool")
-                replace_tip_command = return_tip_template.replace(
-                    "#pipette#", f'pipettes["{pipette_mount}"]'
-                )
-                commands.append(replace_tip_command)
-                tip_loaded[pipette_mount] = False
+                # check if tip on pipette
+                if tip_loaded[pipette_mount] is False:
+                    print("NO TIP TO REPLACE")
+                else:
+                    replace_tip_command = return_tip_template.replace(
+                        "#pipette#", f'pipettes["{pipette_mount}"]'
+                    )
+                    commands.append(replace_tip_command)
+                    tip_loaded[pipette_mount] = False
 
             elif isinstance(command_block, Clear_Pipette):
                 if not isinstance(command_block.clear, bool):

ot2_driver/protopiler/resource_manager.py

@@ -69,7 +69,7 @@ def load_equipment(
         # setup the resource tracker, if exists leave as is, else, create it
         resources = None
         try:
-            self.resources
+            print(self.resources)
         except AttributeError:
             if resource_file:
                 self.resource_file = Path(resource_file)
@@ -397,9 +397,8 @@ def update_tip_count(self, loc, well, tip_num) -> None:
             raise Exception("ERROR no more available tips")
         # update usage
         if tip_num == 1:
-            for i in range(tip_num):
-                self.resources[loc]["wells_used"].add(str(int(well)))
-                self.resources[loc]["used"] += 1
+            self.resources[loc]["wells_used"].add(str(int(well)))
+            self.resources[loc]["used"] += 1
         else:
             for i in range(tip_num):
                 self.resources[loc]["wells_used"].add(str(int(well[i])))

ot2_driver/protopiler/test_configs/basic_config.py

@@ -1,3 +1,4 @@
+"""basic config module"""
 from opentrons import protocol_api
 
 metadata = {
@@ -9,6 +10,7 @@
 
 
 def run(protocol: protocol_api.ProtocolContext):
+    """run basic config"""
     deck = {}
     pipettes = {}
 

scripts/delete_all_runs.py

@@ -1,9 +1,11 @@
+"""delete all runs"""
 from argparse import ArgumentParser
 
 import requests
 
 
 def main(args):
+    """main deletion"""
     base_url = "http://{ip_address}:31950/{extension}"
     headers = {"Opentrons-Version": "2"}