Add support for upload of NovaSeq X flow cell documents to Arnold

cg_lims/EPPs/arnold/flow_cell.py

@@ -6,17 +6,29 @@
 import requests
 from requests import Response
 
+from cg_lims import options
 from cg_lims.exceptions import LimsError
 from cg_lims.get.artifacts import OutputGenerationType, OutputType, get_output_artifacts
 
-from cg_lims.models.arnold.flow_cell import FlowCell, Lane
+from cg_lims.models.arnold.flow_cell import NovaSeq6000FlowCell, NovaSeqXFlowCell, Lane
 
 
 LOG = logging.getLogger(__name__)
 
 
-def build_flow_cell_document(process: Process, lanes: List[Artifact]) -> FlowCell:
-    flowcell = FlowCell(**dict(process.udf.items()), date=process.date_run)
+def build_novaseq_6000_document(process: Process, lanes: List[Artifact]) -> NovaSeq6000FlowCell:
+    flowcell = NovaSeq6000FlowCell(**dict(process.udf.items()), date=process.date_run)
+
+    for lane in lanes:
+        if not lane.location:
+            continue
+        flowcell.lanes.append(Lane(name=lane.name, **dict(lane.udf.items())))
+
+    return flowcell
+
+
+def build_novaseq_x_document(process: Process, lanes: List[Artifact]) -> NovaSeqXFlowCell:
+    flowcell = NovaSeqXFlowCell(**dict(process.udf.items()), date=process.date_run)
 
     for lane in lanes:
         if not lane.location:
@@ -27,9 +39,10 @@ def build_flow_cell_document(process: Process, lanes: List[Artifact]) -> FlowCel
 
 
 @click.command()
+@options.novaseq_x_flow_cell()
 @click.pass_context
-def flow_cell(ctx):
-    """Creating FlowCell documents from a run in the arnold flow_cell collection."""
+def flow_cell(ctx, novaseq_x: bool):
+    """Creating flow cell documents from a run in the arnold flow_cell collection."""
 
     LOG.info(f"Running {ctx.command_path} with params: {ctx.params}")
     process: Process = ctx.obj["process"]
@@ -41,7 +54,14 @@ def flow_cell(ctx):
         lims=lims,
         output_type=OutputType.RESULT_FILE,
     )
-    flow_cell_document: FlowCell = build_flow_cell_document(process=process, lanes=lanes)
+    if novaseq_x:
+        flow_cell_document: NovaSeqXFlowCell = build_novaseq_x_document(
+            process=process, lanes=lanes
+        )
+    else:
+        flow_cell_document: NovaSeq6000FlowCell = build_novaseq_6000_document(
+            process=process, lanes=lanes
+        )
     response: Response = requests.post(
         url=f"{arnold_host}/flow_cell",
         headers={"Content-Type": "application/json"},

cg_lims/models/arnold/flow_cell.py

@@ -28,7 +28,7 @@ class Lane(BaseModel):
     percent_phasing_r2: Optional[float] = Field(None, alias="% Phasing R2")
 
 
-class FlowCell(BaseModel):
+class NovaSeq6000FlowCell(BaseModel):
     instrument: Optional[str] = Field(None, alias="instrument")
     date: Optional[date]
     done: Optional[bool] = Field(None, alias="Done")
@@ -69,3 +69,49 @@ def vali_date(cls, v, values) -> Optional[str]:
 
     class Config:
         allow_population_by_field_name = True
+
+
+class NovaSeqXFlowCell(BaseModel):
+    flow_cell_id: Optional[str] = Field(None, alias="Flow Cell ID")
+    instrument: Optional[str] = Field(None, alias="instrument")
+    date: Optional[date]
+    done: Optional[bool] = Field(None, alias="Done")
+    buffer_expiration_date: Optional[date] = Field(None, alias="Buffer Expiration Date")
+    buffer_lot_number: Optional[str] = Field(None, alias="Buffer Lot Number")
+    buffer_part_number: Optional[str] = Field(None, alias="Buffer Part Number")
+    buffer_serial_barcode: Optional[str] = Field(None, alias="Buffer Serial Number")
+    flow_cell_expiration_date: Optional[date] = Field(None, alias="Flow Cell Expiration Date")
+    flow_cell_lot_number: Optional[str] = Field(None, alias="Flow Cell Lot Number")
+    flow_cell_mode: Optional[str] = Field(None, alias="Flow Cell Mode")
+    flow_cell_part_number: Optional[str] = Field(None, alias="Flow Cell Part Number")
+    reagent_expiration_date: Optional[date] = Field(None, alias="Reagent Expiration Date")
+    reagent_lot_number: Optional[str] = Field(None, alias="Reagent Lot Number")
+    reagent_part_number: Optional[str] = Field(None, alias="Reagent Part Number")
+    reagent_serial_barcode: Optional[str] = Field(None, alias="Reagent Serial Number")
+    run_id: Optional[str] = Field(None, alias="Run ID")
+    sample_tube_expiration_date: Optional[date] = Field(None, alias="Sample Tube Expiration Date")
+    sample_tube_lot_number: Optional[str] = Field(None, alias="Sample Tube Lot Number")
+    sample_tube_part_number: Optional[str] = Field(None, alias="Sample Tube Part Number")
+    sample_tube_serial_barcode: Optional[str] = Field(None, alias="Sample Tube Serial Number")
+    lyo_expiration_date: Optional[date] = Field(None, alias="Lyo Expiration Date")
+    lyo_lot_number: Optional[str] = Field(None, alias="Lyo Lot Number")
+    lyo_part_number: Optional[str] = Field(None, alias="Lyo Part Number")
+    lyo_serial_barcode: Optional[str] = Field(None, alias="Lyo Serial Number")
+    lanes: Optional[List[Lane]] = []
+
+    @validator(
+        "reagent_expiration_date",
+        "sample_tube_expiration_date",
+        "lyo_expiration_date",
+        "flow_cell_expiration_date",
+        "buffer_expiration_date",
+        "date",
+        always=True,
+    )
+    def vali_date(cls, v, values) -> Optional[str]:
+        if isinstance(v, date):
+            return datetime(v.year, v.month, v.day).__str__()
+        return None
+
+    class Config:
+        allow_population_by_field_name = True

cg_lims/options.py

@@ -365,3 +365,15 @@ def udf_values(
         multiple=True,
         help=help,
     )
+
+
+def novaseq_x_flow_cell(
+    help: str = "Flag to set flow cell type to NovaSeq X",
+) -> click.option:
+    return click.option(
+        "-nsx",
+        "--novaseq-x",
+        required=False,
+        is_flag=True,
+        help=help,
+    )

cg_lims/scripts/one_time_scripts/load_arnold_flowcells.py

@@ -7,9 +7,9 @@
 from requests import Response
 
 from cg_lims import options
-from cg_lims.EPPs.arnold.flow_cell import build_flow_cell_document
+from cg_lims.EPPs.arnold.flow_cell import build_novaseq_6000_document
 from cg_lims.get.artifacts import OutputGenerationType, OutputType, get_output_artifacts
-from cg_lims.models.arnold.flow_cell import FlowCell
+from cg_lims.models.arnold.flow_cell import NovaSeq6000FlowCell
 
 LOG = logging.getLogger(__name__)
 
@@ -35,7 +35,9 @@ def update_arnold_flow_cells(ctx, process_types: List[str]):
                 lims=lims,
                 output_type=OutputType.RESULT_FILE,
             )
-            flow_cell_document: FlowCell = build_flow_cell_document(process=process, lanes=lanes)
+            flow_cell_document: NovaSeq6000FlowCell = build_novaseq_6000_document(
+                process=process, lanes=lanes
+            )
             response: Response = requests.post(
                 url=f"{arnold_host}/flow_cell",
                 headers={"Content-Type": "application/json"},

tests/EPPs/arnold/test_arnold_load_flowcell.py

@@ -1,8 +1,8 @@
 from genologics.entities import Process
 
-from cg_lims.EPPs.arnold.flow_cell import build_flow_cell_document
+from cg_lims.EPPs.arnold.flow_cell import build_novaseq_6000_document
 from cg_lims.get.artifacts import OutputGenerationType, OutputType, get_output_artifacts
-from cg_lims.models.arnold.flow_cell import FlowCell
+from cg_lims.models.arnold.flow_cell import NovaSeq6000FlowCell
 
 from tests.conftest import server
 
@@ -21,7 +21,7 @@ def test_load_flowcell(lims, flow_cell_fixture):
     )
 
     # WHEN running build_flow_cell_document
-    flow_cell: FlowCell = build_flow_cell_document(process=process, lanes=lanes)
+    flow_cell: NovaSeq6000FlowCell = build_novaseq_6000_document(process=process, lanes=lanes)
 
     # THEN assert the flow_cell document with its lanes was created.
     flow_cell_dict = flow_cell.dict()