refactor

Author: rickwierenga

pylabrobot/liquid_handling/liquid_handler.py

@@ -2478,14 +2478,12 @@ async def consolidate_tip_inventory(self, tip_racks: List[TipRack]):
     """
 
     def merge_sublists(lists: List[List[int]], max_len: int) -> List[List[int]]:
-      """
-      Merge adjacent sublists if combined length <= max_len,
-        without splitting sublists."""
+      """Merge adjacent sublists if combined length <= max_len, without splitting sublists."""
       merged: List[List[int]] = []
       buffer: List[int] = []
 
       for sublist in lists:
-        if not sublist:
+        if len(sublist) == 0:
           continue  # skip empty sublists
 
         if len(buffer) + len(sublist) <= max_len:
@@ -2495,23 +2493,22 @@ def merge_sublists(lists: List[List[int]], max_len: int) -> List[List[int]]:
             merged.append(buffer)
           buffer = sublist  # start new buffer
 
-      if buffer:
+      if len(buffer) > 0:
         merged.append(buffer)
 
       return merged
 
     def divide_list_into_chunks(
       list_l: List[Any], chunk_size: int
     ) -> Generator[List[Any], None, None]:
-      """
-      Divides a list into smaller chunks of a specified size.
+      """Divides a list into smaller chunks of a specified size.
 
       Parameters:
-      - list_l (List[Any]): The list to be divided into chunks.
-      - chunk_size (int): The size of each chunk.
+        - list_l: The list to be divided into chunks.
+        - chunk_size: The size of each chunk.
 
       Returns:
-      - Generator[List[Any], None, None]: A generator that yields chunks of the list.
+        A generator that yields chunks of the list.
       """
       for i in range(0, len(list_l), chunk_size):
         yield list_l[i : i + chunk_size]
@@ -2520,33 +2517,35 @@ def divide_list_into_chunks(
 
     for idx, tip_rack in enumerate(tip_racks):
       # Only consider partially-filled tip_racks
-      tip_status = [tip_spot.tracker.has_tip for tip_spot in tip_rack.children]
-      partially_filled = any(tip_status) and not all(tip_status)
-
-      if partially_filled:
-        tipspots_w_tips = [i for b, i in zip(tip_status, tip_rack.children) if b]
+      tip_status = [tip_spot.tracker.has_tip for tip_spot in tip_rack.get_all_items()]
 
-        # Identify model by hashed unique physical characteristics
-        current_model = hash(tipspots_w_tips[0].tracker.get_tip())
+      if not (any(tip_status) and not all(tip_status)):
+        continue  # ignore non-partially-filled tip_racks
 
-        num_empty_tipspots = len(tip_status) - len(tipspots_w_tips)
+      tipspots_w_tips = [tip_spot for has_tip, tip_spot in zip(tip_status, tip_rack.children) if has_tip]
 
-        sanity_check = all(
-          hash(tip_spot.tracker.get_tip()) == current_model for tip_spot in tipspots_w_tips[1:]
+      # Identify model by hashed unique physical characteristics
+      current_model = hash(tipspots_w_tips[0].tracker.get_tip())
+      if not all(
+        hash(tip_spot.tracker.get_tip()) == current_model for tip_spot in tipspots_w_tips[1:]
+      ):
+        raise ValueError(
+          f"Tip rack {tip_rack.name} has mixed tip models, cannot consolidate: "
+          f"{[tip_spot.tracker.get_tip() for tip_spot in tipspots_w_tips]}"
         )
 
-        if sanity_check:
-          clusters_by_model.setdefault(current_model, []).append((tip_rack, num_empty_tipspots))
+      num_empty_tipspots = len(tip_status) - len(tipspots_w_tips)
+      clusters_by_model.setdefault(current_model, []).append((tip_rack, num_empty_tipspots))
 
-    # Sort partially-filled tipracks by minimal fill_len
+    # Sort partially-filled tipracks from most to least empty
     for model, rack_list in clusters_by_model.items():
       rack_list.sort(key=lambda x: x[1])
 
     # Consolidate one tip model at a time across all tip_racks of that model
     for model, rack_list in clusters_by_model.items():
-      print(f"Consolidating:\n - {', '.join([rack.name for rack, num in rack_list])}")
+      print(f"Consolidating: - {', '.join([rack.name for rack, _ in rack_list])}")
 
-      all_tip_spots_list = [tip for tip_rack, _ in rack_list for tip in tip_rack.children]
+      all_tip_spots_list = [tip_spot for tip_rack, _ in rack_list for tip_spot in tip_rack.children]
 
       # 1: Record current tip state
       current_tip_presence_list = [tip_spot.has_tip() for tip_spot in all_tip_spots_list]
@@ -2555,11 +2554,7 @@ def divide_list_into_chunks(
       total_length = len(all_tip_spots_list)
       num_tips_per_model = sum(current_tip_presence_list)
 
-      target_tip_presence_list = [
-        # True if i < num_tips_per_model else False for i in range(total_length)
-        i < num_tips_per_model
-        for i in range(total_length)
-      ]
+      target_tip_presence_list = [i < num_tips_per_model for i in range(total_length)]
 
       # 3: Calculate tip_spots involved in tip movement
       tip_movement_list = [
@@ -2572,23 +2567,24 @@ def divide_list_into_chunks(
       tip_target_indices = [i for i, v in enumerate(tip_movement_list) if v == -1]
       all_target_tip_spots = [all_tip_spots_list[idx] for idx in tip_target_indices]
 
+      # Only continue if tip_racks are not already consolidated
+      if len(all_target_tip_spots) == 0:
+        print("Tips already optimally consolidated!")
+        continue
+
       # 4: Cluster target tip_spots by BOTH parent tip_rack & x-coordinate
       sorted_tip_spots = sorted(
-        all_target_tip_spots, key=lambda tip: (str(tip.parent), round(tip.location.x, 3))
+        all_target_tip_spots, key=lambda tip: (tip.parent.name, round(tip.location.x, 3))
       )
 
       target_tip_clusters_by_parent_x: Dict[Tuple[str, float], List[TipSpot]] = {}
 
       for tip_spot in sorted_tip_spots:
-        key = (str(tip_spot.parent), round(tip_spot.location.x, 3))
+        key = (tip_spot.parent.name, round(tip_spot.location.x, 3))
         if key not in target_tip_clusters_by_parent_x:
           target_tip_clusters_by_parent_x[key] = []
         target_tip_clusters_by_parent_x[key].append(tip_spot)
 
-      # Only continue if tip_racks are not already consolidated
-      if len(target_tip_clusters_by_parent_x) > 0:
-        raise ValueError(f"No channel capable of handling tips on deck: {current_tip_model}")
-
       current_tip_model = all_origin_tip_spots[0].tracker.get_tip()
 
       # Ensure there are channels that can pick up the tip model
@@ -2600,31 +2596,29 @@ def divide_list_into_chunks(
         ]
       )
 
-      # 5: Optimise speed
-      if num_channels_available > 0:
-        # by aggregating drop columns i.e. same drop column should not be visited twice!
-        if num_channels_available >= 8:  # physical constraint of tip_rack's having 8 rows
-          merged_target_tip_clusters = merge_sublists(
-            target_tip_clusters_by_parent_x.values(), max_len=8
-          )
-
-        else:  # by chunking drop tip_spots list into size of available channels
-          merged_target_tip_clusters = list(
-            divide_list_into_chunks(all_target_tip_spots, chunk_size=num_channels_available)
-          )
+      # 5: Optimize speed
+      if num_channels_available == 0:
+        raise ValueError(f"No channel capable of handling tips on deck: {current_tip_model}")
 
-        len_transfers = len(merged_target_tip_clusters)
+      # by aggregating drop columns i.e. same drop column should not be visited twice!
+      if num_channels_available >= 8:  # physical constraint of tip_rack's having 8 rows
+        merged_target_tip_clusters = merge_sublists(
+          target_tip_clusters_by_parent_x.values(), max_len=8
+        )
+      else:  # by chunking drop tip_spots list into size of available channels
+        merged_target_tip_clusters = list(
+          divide_list_into_chunks(all_target_tip_spots, chunk_size=num_channels_available)
+        )
 
-        # 6: Execute tip movement/consolidation
-        for idx, target_tip_spots in enumerate(merged_target_tip_clusters):
-          print(f"     - tip transfer cycle: {idx} / {len_transfers - 1}")
-          num_channels = len(target_tip_spots)
-          use_channels = list(range(num_channels))
+      len_transfers = len(merged_target_tip_clusters)
 
-          origin_tip_spots = [all_origin_tip_spots.pop(0) for idx in range(num_channels)]
+      # 6: Execute tip movement/consolidation
+      for idx, target_tip_spots in enumerate(merged_target_tip_clusters):
+        print(f"   - tip transfer cycle: {idx+1} / {len_transfers}")
+        num_channels = len(target_tip_spots)
+        use_channels = list(range(num_channels))
 
-          await self.pick_up_tips(origin_tip_spots, use_channels=use_channels)
+        origin_tip_spots = [all_origin_tip_spots.pop(0) for _ in range(num_channels)]
 
-          await self.drop_tips(target_tip_spots, use_channels=use_channels)
-      else:
-        print("Tips already optimally consolidated!")
+        await self.pick_up_tips(origin_tip_spots, use_channels=use_channels)
+        await self.drop_tips(target_tip_spots, use_channels=use_channels)