add new epp for revio seq setings

cg_lims/EPPs/files/smrt_link/models.py

@@ -94,7 +94,7 @@ def __init__(self, artifact: Artifact):
         self.system_name = process.udf.get("Sequencing Instrument")
         self.binding_kit = process.udf.get("Binding Kit")
         self.number_of_samples = len(artifact.samples)
-        self.application = process.udf.get("Prep Type")
+        self.application = process.udf.get("Revio Application")
         self.available_volume = artifact.udf.get("Volume (ul)")
         self.starting_concentration = artifact.udf.get("Input Concentration (ng/ul)")
         self.size = artifact.udf.get("Size (bp)")
@@ -305,7 +305,7 @@ def _create_smrt_cell_settings(self) -> str:
                 pool.udf.get("Revio Application"),
                 pool.udf.get("Polymerase Kit"),
                 self.movie_acquisition_time,
-                pool.udf.get("Average Size (bp)"),
+                pool.udf.get("Mean Size (bp)"),
                 pool.udf.get("Library Concentration (pM)"),
                 self.adaptive_loading,
                 self.base_kinetics,

cg_lims/EPPs/udf/set/base.py

@@ -3,6 +3,7 @@
 from cg_lims.EPPs.udf.set.set_barcode import assign_barcode
 from cg_lims.EPPs.udf.set.set_method import method_document
 from cg_lims.EPPs.udf.set.set_ont_sequencing_settings import set_ont_sequencing_settings
+from cg_lims.EPPs.udf.set.set_revio_sequencing_settings import set_revio_sequencing_settings
 from cg_lims.EPPs.udf.set.set_sample_date import set_sample_date
 from cg_lims.EPPs.udf.set.set_samples_reads_missing import set_reads_missing_on_new_samples
 from cg_lims.EPPs.udf.set.set_sequencing_settings import set_sequencing_settings
@@ -24,3 +25,4 @@ def set(context: click.Context):
 set.add_command(set_sequencing_settings)
 set.add_command(replace_flow_cell_output_path)
 set.add_command(updated_sample_volume)
+set.add_command(set_revio_sequencing_settings)

cg_lims/EPPs/udf/set/set_revio_sequencing_settings.py

@@ -0,0 +1,113 @@
+import logging
+import sys
+from datetime import date
+from typing import List
+
+import click
+import numpy as np
+from cg_lims.exceptions import InvalidValueError, LimsError
+from cg_lims.get.artifacts import get_artifacts, get_latest_artifact, get_non_pooled_artifacts
+from genologics.entities import Artifact, Container, Process
+from genologics.lims import Lims
+
+LOG = logging.getLogger(__name__)
+
+
+def get_run_name(process: Process) -> str:
+    """Create and return a run name from today's date and the step ID."""
+    return f"{date.today().strftime('%y%m%d')}_{process.id}"
+
+
+def set_run_name(process: Process) -> None:
+    """Set the run name to both process and artifact UDFs."""
+    run_name: str = get_run_name(process=process)
+    process.udf["Run Name"] = run_name
+    process.put()
+
+
+def set_plates(process: Process) -> None:
+    """Sets default values for the plates in the step."""
+    containers: List[Container] = process.output_containers()
+    containers.sort(key=lambda container_object: container_object.id)
+    if len(containers) > 2:
+        raise InvalidValueError(
+            f"Error: The maximum number of plates you can have for a sequencing run is 2!"
+        )
+    i = 1
+    for container in containers:
+        process.udf[f"Plate {i}"] = container.name
+        i += 1
+    process.put()
+
+
+def calculate_weighted_average_size(artifact: Artifact) -> float:
+    """"""
+    if len(artifact.samples) <= 1:
+        input_artifact = artifact.input_artifact_list()[0]
+        return input_artifact.udf.get("Size (bp)")
+    input_artifacts: List[Artifact] = get_non_pooled_artifacts(artifact=artifact)
+    sizes: List[float] = []
+    volumes: List[float] = []
+    for input_artifact in input_artifacts:
+        sizes.append(input_artifact.udf.get("Size (bp)"))
+        volumes.append(input_artifact.udf.get("Volume of sample (ul)"))
+    return np.average(sizes, weights=volumes)
+
+
+def get_library_concentration(artifact: Artifact) -> float:
+    """"""
+    lims: Lims = artifact.lims
+    sample_id: str = artifact.samples[0].id
+    if len(artifact.samples) <= 1:
+        udf_name: str = "Loading Concentration (pM)"
+        step_name: str = "Preparing ABC Complex (Revio)"
+    else:
+        udf_name: str = "Target Pool Concentration (pM)"
+        step_name: str = "Pooling Samples for Sequencing (Revio)"
+    all_artifacts = lims.get_artifacts(process_type=step_name, samplelimsid=sample_id)
+    artifact: Artifact = get_latest_artifact(lims_artifacts=all_artifacts)
+    parent_process: Process = artifact.parent_process
+    return parent_process.udf[udf_name]
+
+
+def get_polymerase_kit(artifact: Artifact) -> float:
+    """"""
+    lims: Lims = artifact.lims
+    sample_id: str = artifact.samples[0].id
+    all_artifacts = lims.get_artifacts(
+        process_type="Preparing ABC Complex (Revio)", samplelimsid=sample_id
+    )
+    artifact: Artifact = get_latest_artifact(lims_artifacts=all_artifacts)
+    parent_process: Process = artifact.parent_process
+    return parent_process.udf["Binding Kit"]
+
+
+def set_artifact_values(artifacts: List[Artifact]) -> None:
+    """"""
+    for artifact in artifacts:
+        artifact.udf["Mean Size (bp)"] = calculate_weighted_average_size(artifact=artifact)
+        artifact.udf["Library Concentration (pM)"] = get_library_concentration(artifact=artifact)
+        artifact.udf["Polymerase Kit"] = get_polymerase_kit(artifact=artifact)
+        artifact.put()
+
+
+@click.command()
+@click.pass_context
+def set_revio_sequencing_settings(ctx):
+    """Sets the values required for sequencing of Revio SMRT cells."""
+
+    LOG.info(f"Running {ctx.command_path} with params: {ctx.params}")
+
+    process: Process = ctx.obj["process"]
+
+    try:
+        artifacts: List[Artifact] = get_artifacts(process=process)
+        set_run_name(process=process)
+        set_plates(process=process)
+        set_artifact_values(artifacts=artifacts)
+        message: str = "Revio sequencing settings have been successfully set."
+        LOG.info(message)
+        click.echo(message)
+    except LimsError as e:
+        LOG.error(e.message)
+        sys.exit(e.message)