Merge c00ace3 into 767b883

.travis.yml

@@ -7,7 +7,7 @@ install:
   - pip install -r requirements-dev.txt -r requirements.txt -e .
 
 script:
-  - py.test --cov=cg_lims -rsxv tests/* 
+  - py.test --cov=cg_lims -rsxv tests/*
 
 after_success: coveralls
 

cg_lims/EPPs/udf/calculate/base.py

@@ -9,6 +9,7 @@
     calculate_water_volume_rna,
 )
 from cg_lims.EPPs.udf.calculate.get_missing_reads import get_missing_reads
+from cg_lims.EPPs.udf.calculate.maf_calculate_volume import maf_calculate_volume
 from cg_lims.EPPs.udf.calculate.molar_concentration import molar_concentration
 from cg_lims.EPPs.udf.calculate.sum_missing_reads_in_pool import missing_reads_in_pool
 from cg_lims.EPPs.udf.calculate.twist_aliquot_amount import twist_aliquot_amount
@@ -40,4 +41,5 @@ def calculate(ctx):
 calculate.add_command(molar_concentration)
 calculate.add_command(calculate_beads)
 calculate.add_command(missing_reads_in_pool)
+calculate.add_command(maf_calculate_volume)
 calculate.add_command(calculate_water_volume_rna)

cg_lims/EPPs/udf/calculate/maf_calculate_volume.py

@@ -0,0 +1,130 @@
+"""
+CLI module to calculate dilution volumes from Concentration udf
+
+For an explanation of the sample volume calculation and setting of the QG flag, see
+AM-document 1243 Method - Preparing Plate for Genotyping, section 3.3.2 Preparing the plate
+"""
+
+import logging
+import sys
+from typing import List, Optional
+
+import click
+from genologics.entities import Artifact
+from pydantic import BaseModel, validator
+
+from cg_lims.exceptions import LimsError
+from cg_lims.get.artifacts import get_artifacts
+
+LOG = logging.getLogger(__name__)
+
+FINAL_CONCENTRATION = 4
+MINIMUM_TOTAL_VOLUME = 15
+QC_FAILED = "FAILED"
+QC_PASSED = "PASSED"
+STANDARD_SAMPLE_VOLUME = 3
+
+
+class MafVolumes(BaseModel):
+    """Pydantic model for calculating MAF volumes based on the sample concentration"""
+
+    sample_concentration: float
+    sample_volume: Optional[float]
+    final_volume: Optional[float]
+    water_volume: Optional[float]
+    qc_flag: Optional[str]
+
+    @validator("sample_concentration", always=True)
+    def set_sample_concentration(cls, v, values):
+        """Set sample concentration and handle low or missing concentration"""
+        if v < FINAL_CONCENTRATION:
+            message = "Too low or missing concentration"
+            LOG.error(message)
+            raise ValueError(message)
+        return v
+
+    @validator("sample_volume", always=True)
+    def set_sample_volume(cls, v, values):
+        """Calculate the sample volume. For an explanation about how sample volumes are
+        determined for various sample concentration ranges, refer to AM document 1243 Method -
+        Preparing Plate for Genotyping, section 3.3.2"""
+        sample_concentration = values.get("sample_concentration")
+        if sample_concentration < 20:
+            return MINIMUM_TOTAL_VOLUME * FINAL_CONCENTRATION / sample_concentration
+        elif 20 <= sample_concentration < 244:
+            return 3
+        elif 244 <= sample_concentration < 364:
+            return 2
+        elif 364 <= sample_concentration < 724:
+            return 1
+        elif 724 <= sample_concentration < 1444:
+            return 0.5
+        else:
+            message = "Sample concentration not in valid range"
+            LOG.error(message)
+            raise ValueError(message)
+
+    @validator("water_volume", always=True)
+    def set_water_volume(cls, v, values):
+        """Calculate the water volume for a sample"""
+        return values.get("final_volume") - values.get("sample_volume")
+
+    @validator("final_volume", always=True)
+    def set_final_volume(cls, v, values):
+        """Calculates the final volume for a sample"""
+        return (
+            values.get("sample_volume")
+            * values.get("sample_concentration")
+            / FINAL_CONCENTRATION
+        )
+
+    @validator("qc_flag", always=True)
+    def set_qc_flag(cls, v, values):
+        """Set the QC flag on a sample"""
+        if values.get("sample_volume") == STANDARD_SAMPLE_VOLUME:
+            return QC_PASSED
+        else:
+            return QC_FAILED
+
+
+def calculate_volume(artifacts: List[Artifact]) -> None:
+    """Determines the final volume, water volume, and sample volume"""
+    failed_artifacts = []
+
+    for artifact in artifacts:
+        try:
+            volumes = MafVolumes(sample_concentration=artifact.udf.get("Concentration"))
+            artifact.qc_flag = volumes.qc_flag
+            artifact.udf["Final Volume (uL)"] = volumes.final_volume
+            artifact.udf["Volume H2O (ul)"] = volumes.water_volume
+            artifact.udf["Volume of sample (ul)"] = volumes.sample_volume
+            artifact.put()
+        except Exception:
+            LOG.warning(
+                f"Could not calculate sample volume for sample {artifact.samples[0].name}."
+            )
+            failed_artifacts.append(artifact)
+            continue
+
+    if failed_artifacts:
+        raise LimsError(
+            f"MAF volume calculations failed for {len(failed_artifacts)} samples, "
+            f"{len(artifacts) - len(failed_artifacts)} updates successful. "
+        )
+
+
+@click.command()
+@click.pass_context
+def maf_calculate_volume(context: click.Context):
+    """Calculate dilution volumes based on the Concentration udf"""
+    LOG.info(f"Running {context.command_path} with params: {context.params}")
+    process = context.obj["process"]
+    try:
+        artifacts: List[Artifact] = get_artifacts(process=process, input=False)
+        calculate_volume(artifacts=artifacts)
+        message = "MAF volumes have been calculated."
+        LOG.info(message)
+        click.echo(message)
+    except LimsError as e:
+        LOG.error(e.message)
+        sys.exit(e.message)

tests/EPPs/udf/calculate/test_maf_calculate_volume.py

@@ -0,0 +1,160 @@
+"""Unit tests for cg_lims.EPPs.udf.calculate.maf_calculate_volume"""
+import logging
+
+import pytest
+from genologics.entities import Artifact
+from pydantic import ValidationError
+
+from cg_lims.EPPs.udf.calculate.maf_calculate_volume import (
+    QC_FAILED,
+    QC_PASSED,
+    MafVolumes,
+    calculate_volume,
+)
+
+LOG = logging.getLogger(__name__)
+
+
+@pytest.mark.parametrize(
+    "sample_concentration, final_volume, water_volume, sample_volume, qc_flag",
+    [
+        (19, 15.0, 11.842105263157894, 3.1578947368421053, QC_FAILED),
+        (20, 15.0, 12.0, 3, QC_PASSED),
+        (21, 15.75, 12.75, 3, QC_PASSED),
+        (239, 179.25, 176.25, 3, QC_PASSED),
+        (243, 182.25, 179.25, 3, QC_PASSED),
+        (244, 122.0, 120.0, 2, QC_FAILED),
+        (363, 181.5, 179.5, 2, QC_FAILED),
+        (364, 91.0, 90.0, 1, QC_FAILED),
+        (723, 180.75, 179.75, 1, QC_FAILED),
+        (724, 90.5, 90.0, 0.5, QC_FAILED),
+        (1443, 180.375, 179.875, 0.5, QC_FAILED),
+    ],
+)
+def test_maf_volume_model(
+    sample_concentration, sample_volume, final_volume, water_volume, qc_flag
+):
+    # GIVEN a pydantic model for MAF volumes
+
+    # WHEN calculating the volumes in the validators with valid values for 'sample_concentration'
+    result = MafVolumes(sample_concentration=sample_concentration)
+
+    # THEN the correct results should be set
+    assert result.sample_concentration == sample_concentration
+    assert result.final_volume == final_volume
+    assert result.water_volume == water_volume
+    assert result.sample_volume == sample_volume
+    assert result.qc_flag == qc_flag
+
+
+@pytest.mark.parametrize(
+    "sample_concentration",
+    [
+        3.99,
+    ],
+)
+def test_maf_volume_model_sample_concentration_value_error(sample_concentration):
+    # GIVEN a pydantic model for MAF volumes
+
+    # WHEN calculating the volumes in the validators with an invalid value for
+    # 'sample_concentration'
+    with pytest.raises(ValueError) as error_message:
+        MafVolumes(sample_concentration=sample_concentration)
+
+    # THEN the validator for sample_concentration should raise a ValueError exception
+    assert error_message.value.errors()[0]["loc"][0] == "sample_concentration"
+    assert error_message.value.errors()[0]["type"] == "value_error"
+    assert error_message.value.errors()[0]["msg"] == "Too low or missing concentration"
+
+
+@pytest.mark.parametrize(
+    "sample_concentration",
+    [
+        None,
+    ],
+)
+def test_maf_volume_model_sample_concentration_validation_error(sample_concentration):
+    # GIVEN a pydantic model for MAF volumes
+
+    # WHEN calculating the volumes in the validator when there is no 'sample_concentration'
+    with pytest.raises(ValidationError) as error_message:
+        MafVolumes(sample_concentration=sample_concentration)
+
+    # THEN pydantic should raise a ValidationError exception
+    assert error_message.value.errors()[0]["loc"][0] == "sample_concentration"
+    assert error_message.value.errors()[0]["type"] == "type_error.none.not_allowed"
+    assert error_message.value.errors()[0]["msg"] == "none is not an allowed value"
+
+
+@pytest.mark.parametrize(
+    "sample_concentration",
+    [
+        9001,
+    ],
+)
+def test_maf_volume_model_sample_volume_value_error(sample_concentration):
+    # GIVEN a pydantic model for MAF volumes
+
+    # WHEN calculating the volumes in the validator when there is no 'sample_concentration'
+    with pytest.raises(ValueError) as error_message:
+        MafVolumes(sample_concentration=sample_concentration)
+
+    # THEN the validator for sample_volume should raise a ValueError exception
+    assert error_message.value.errors()[0]["loc"][0] == "sample_volume"
+    assert error_message.value.errors()[0]["type"] == "value_error"
+    assert (
+        error_message.value.errors()[0]["msg"]
+        == "Sample concentration not in valid range"
+    )
+
+
+@pytest.mark.parametrize(
+    "sample_concentration, final_volume, water_volume, sample_volume, qc_flag",
+    [
+        (243, 182.25, 179.25, 3, QC_PASSED),
+    ],
+)
+def test_calculate_volume(
+    sample_concentration,
+    sample_volume,
+    final_volume,
+    water_volume,
+    qc_flag,
+    artifact_1: Artifact,
+):
+    # GIVEN a sample with a valid sample concentration:
+    artifact_1.udf["Concentration"] = sample_concentration
+    artifacts = [artifact_1]
+
+    # WHEN calculating the volumes
+    calculate_volume(artifacts)
+
+    # THEN the sample udfs should be set to the correct values
+    artifact_1.qc_flag = qc_flag
+    artifact_1.udf["Final Volume (uL)"] = final_volume
+    artifact_1.udf["Volume H2O (ul)"] = water_volume
+    artifact_1.udf["Volume of sample (ul)"] = sample_volume
+
+
+@pytest.mark.parametrize(
+    "sample_concentration",
+    [
+        3.99,
+    ],
+)
+def test_calculate_volume_exception(
+    sample_concentration,
+    artifact_1: Artifact,
+    caplog,
+):
+    # GIVEN a sample with a sample concentration that is below the required final concentration
+    artifact_1.udf["Concentration"] = sample_concentration
+    artifacts = [artifact_1]
+
+    # WHEN calculating the volumes
+    with pytest.raises(Exception) as error_message:
+        calculate_volume(artifacts)
+
+    # THEN an Exception should be raised
+    assert "Too low or missing concentration" in caplog.text
+    assert "MAF volume calculations failed for" in error_message.value.message