Complete Bloom lab-action live proof support

Author: iamh2o

bloom_lims/domain/lab_actions.py

@@ -46,6 +46,9 @@
 POOL_CONTENT_TEMPLATE_CODE = "content/pool/sequencing-library/1.0/"
 GDNA_QUANT_TEMPLATE_CODE = "data/quantification/gdna/1.0/"
 EXTRACTION_QC_TEMPLATE_CODE = "data/operation/extraction-qc/1.0/"
+SEQUENCED_LIBRARY_ASSIGNMENT_TEMPLATE_CODE = (
+    "data/library-index-assignment/sequencing-library/1.0/"
+)
 
 
 def re_split_multi(value: str) -> list[str]:
@@ -795,15 +798,77 @@ def create_seq_run_set(self, request: SeqRunSetRequest) -> dict[str, Any]:
             self._create_lineage(run_set.euid, request.instrument_euid, "run_uses_instrument")
         for reagent_euid in request.reagent_euids:
             self._create_lineage(run_set.euid, reagent_euid, "run_uses_reagent")
+        assignments = self._create_sequenced_library_assignments(
+            run_set=run_set,
+            pool_content=pool_content,
+            flowcell_barcode=request.flowcell_barcode,
+        )
         self.bdb.session.commit()
         return {
             "set_euid": run_set.euid,
             "status": run_set.bstatus,
             "platform": request.platform,
             "pool_tube_euid": pool_tube.euid,
             "pool_content_euid": pool_content.euid,
+            "assignments": assignments,
         }
 
+    def _create_sequenced_library_assignments(
+        self,
+        *,
+        run_set,
+        pool_content,
+        flowcell_barcode: str,
+    ) -> list[dict[str, str]]:
+        assignments: list[dict[str, str]] = []
+        source_libraries = [
+            lineage.parent_instance
+            for lineage in self._lineages_to_child(pool_content, "DERIVED_FROM")
+        ]
+        for index, library in enumerate(source_libraries, start=1):
+            props = self._props(library)
+            barcode = str(props.get("index_barcode") or "").strip()
+            lane = str(props.get("lane") or props.get("lane_index") or "1").strip()
+            assignment = self._create_by_code(
+                SEQUENCED_LIBRARY_ASSIGNMENT_TEMPLATE_CODE,
+                name=f"{run_set.euid}:{lane}:{barcode or library.euid}",
+                properties={
+                    "flowcell_id": flowcell_barcode,
+                    "lane": lane,
+                    "library_barcode": barcode,
+                    "library_prep_output_euid": library.euid,
+                    "library_material_euid": library.euid,
+                    "barcode_reagent_euid": str(props.get("index_euid") or "").strip(),
+                    "metadata": {
+                        "created_by": "bloom_lab_actions",
+                        "pool_content_euid": pool_content.euid,
+                    },
+                },
+            )
+            self._create_lineage(
+                run_set.euid,
+                assignment.euid,
+                "beta_sequenced_library_assignment",
+            )
+            self._create_lineage(
+                library.euid,
+                assignment.euid,
+                "beta_assignment_source",
+                edge_type="RUN_CONSUMED",
+                source_role="sequencing_run_set",
+                target_role="library_material",
+            )
+            assignments.append(
+                {
+                    "assignment_euid": assignment.euid,
+                    "library_content_euid": library.euid,
+                    "flowcell_id": flowcell_barcode,
+                    "lane": lane,
+                    "library_barcode": barcode,
+                }
+            )
+        return assignments
+
     def plate_mapping_rows(self, plate_euid: str) -> list[dict[str, str]]:
         plate = self._require(plate_euid)
         wells = self._plate_wells(plate)
@@ -1302,10 +1367,22 @@ def import_spreadsheet(
         return {"filename": filename, "dry_run": dry_run, "actions": actions}
 
     def print_euids(self, request: PrintEuidRequest) -> dict[str, Any]:
-        service = ZebraDayService()
         results = []
         for euid in request.euids:
             self._require(euid)
+            if request.dry_run:
+                results.append(
+                    {
+                        "dry_run": True,
+                        "lab": request.lab,
+                        "printer_id": request.printer_id,
+                        "label_zpl_style": request.label_zpl_style,
+                        "euid": euid,
+                        "copies": request.copies,
+                    }
+                )
+                continue
+            service = ZebraDayService()
             results.append(
                 service.submit_print_job(
                     lab=request.lab,
@@ -1315,4 +1392,4 @@ def print_euids(self, request: PrintEuidRequest) -> dict[str, Any]:
                     print_n=request.copies,
                 )
             )
-        return {"printed": len(results), "results": results}
+        return {"printed": 0 if request.dry_run else len(results), "dry_run": request.dry_run, "results": results}

bloom_lims/schemas/lab_actions.py

@@ -260,6 +260,7 @@ class PrintEuidRequest(BaseModel):
     printer_id: str
     label_zpl_style: str
     copies: int = Field(default=1, ge=1, le=20)
+    dry_run: bool = False
 
     @field_validator("euids")
     @classmethod

templates/modern/lab_actions.html

@@ -240,6 +240,10 @@ <h2>Print EUID Barcodes</h2>
                 <input id="print-style" class="form-control" data-testid="bloom-lab-print-style" value="euid">
             </div>
         </div>
+        <label class="checkbox-row">
+            <input type="checkbox" id="print-dry-run" data-testid="bloom-lab-print-dry-run" checked>
+            Preview only; do not send to printer
+        </label>
         <button class="btn btn-outline" type="button" data-action="print-euids" data-testid="bloom-lab-print-submit">
             <i class="fas fa-barcode"></i> Print Selected EUIDs
         </button>
@@ -468,6 +472,7 @@ <h2>Action Results</h2>
       printer_id: document.getElementById('print-printer').value.trim(),
       label_zpl_style: document.getElementById('print-style').value.trim(),
       copies: 1,
+      dry_run: document.getElementById('print-dry-run').checked,
     }),
   };
 

tests/test_lab_actions_api.py

@@ -201,6 +201,9 @@ def test_lab_actions_full_ilmn_flow_and_csv_export() -> None:
     assert seq_run.status_code == 200, seq_run.text
     run_payload = seq_run.json()
     assert run_payload["set_euid"]
+    assert run_payload["assignments"]
+    assert run_payload["assignments"][0]["flowcell_id"] == "PYTEST-FLOWCELL"
+    assert run_payload["assignments"][0]["lane"] == "1"
 
     sample_sheet = client.get(
         f"/api/v1/lab-actions/seq-runs/{run_payload['set_euid']}/samplesheet"
@@ -282,11 +285,14 @@ def test_object_creation_count_and_print_euid_mock() -> None:
                 "printer_id": "pytest-printer",
                 "label_zpl_style": "euid",
                 "copies": 1,
+                "dry_run": True,
             },
         )
     assert response.status_code == 200, response.text
-    assert response.json()["printed"] == 3
-    assert service.submit_print_job.call_count == 3
+    assert response.json()["printed"] == 0
+    assert response.json()["dry_run"] is True
+    assert service.submit_print_job.call_count == 0
+    assert len(response.json()["results"]) == 3
 
     recursive_batch = client.post(
         "/api/v1/object-creation/create",