diff --git a/cg_lims/EPPs/arnold/flow_cell.py b/cg_lims/EPPs/arnold/flow_cell.py index f0af7f42..48b8a06e 100644 --- a/cg_lims/EPPs/arnold/flow_cell.py +++ b/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"}, diff --git a/cg_lims/models/arnold/flow_cell.py b/cg_lims/models/arnold/flow_cell.py index 6081c428..7d9e13f1 100644 --- a/cg_lims/models/arnold/flow_cell.py +++ b/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 diff --git a/cg_lims/options.py b/cg_lims/options.py index 590149c5..75a0b838 100644 --- a/cg_lims/options.py +++ b/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, + ) diff --git a/cg_lims/scripts/one_time_scripts/load_arnold_flowcells.py b/cg_lims/scripts/one_time_scripts/load_arnold_flowcells.py index 3d100b27..31b0d0f5 100644 --- a/cg_lims/scripts/one_time_scripts/load_arnold_flowcells.py +++ b/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"}, diff --git a/tests/EPPs/arnold/test_arnold_load_flowcell.py b/tests/EPPs/arnold/test_arnold_load_flowcell.py index a16a5a0b..d0043ae3 100644 --- a/tests/EPPs/arnold/test_arnold_load_flowcell.py +++ b/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()