Add mutant rules (#3656)

cg/services/order_validation_service/constants.py

@@ -20,6 +20,7 @@ class ElutionBuffer(StrEnum):
 WORKFLOW_PREP_CATEGORIES: dict[Workflow, list[PrepCategory]] = {
     Workflow.MICROSALT: [PrepCategory.COVID, PrepCategory.MICROBIAL],
     Workflow.TOMTE: [PrepCategory.WHOLE_TRANSCRIPTOME_SEQUENCING],
+    Workflow.MUTANT: [PrepCategory.COVID],
 }
 
 MINIMUM_VOLUME, MAXIMUM_VOLUME = 20, 130

cg/services/order_validation_service/models/aliases.py

@@ -1,8 +1,12 @@
+from cg.services.order_validation_service.workflows.microsalt.models.sample import MicrosaltSample
 from cg.services.order_validation_service.workflows.mip_dna.models.case import MipDnaCase
 from cg.services.order_validation_service.workflows.mip_dna.models.sample import MipDnaSample
+from cg.services.order_validation_service.workflows.mutant.models.sample import MutantSample
 from cg.services.order_validation_service.workflows.tomte.models.case import TomteCase
 from cg.services.order_validation_service.workflows.tomte.models.sample import TomteSample
 
 
 SampleWithRelatives = TomteSample | MipDnaSample
 CaseContainingRelatives = TomteCase | MipDnaCase
+
+NonHumanSample = MutantSample | MicrosaltSample

cg/services/order_validation_service/models/order_with_samples.py

@@ -0,0 +1,11 @@
+from cg.services.order_validation_service.models.aliases import NonHumanSample
+from cg.services.order_validation_service.models.order import Order
+from cg.services.order_validation_service.models.sample import Sample
+
+
+class OrderWithNonHumanSamples(Order):
+    samples: list[NonHumanSample]
+
+    @property
+    def enumerated_samples(self) -> enumerate[NonHumanSample]:
+        return enumerate(self.samples)

cg/services/order_validation_service/rules/sample/rules.py

@@ -0,0 +1,133 @@
+from cg.constants.constants import PrepCategory, Workflow
+from cg.services.order_validation_service.constants import WORKFLOW_PREP_CATEGORIES
+from cg.services.order_validation_service.errors.sample_errors import (
+    ApplicationArchivedError,
+    ApplicationNotCompatibleError,
+    ApplicationNotValidError,
+    ElutionBufferMissingError,
+    ExtractionMethodMissingError,
+    InvalidVolumeError,
+    OccupiedWellError,
+    OrganismDoesNotExistError,
+    SampleNameRepeatedError,
+)
+from cg.services.order_validation_service.rules.utils import (
+    is_application_not_compatible,
+    is_volume_invalid,
+)
+from cg.services.order_validation_service.workflows.microsalt.models.order import (
+    OrderWithNonHumanSamples,
+)
+from cg.services.order_validation_service.rules.sample.utils import (
+    PlateSamplesValidator,
+    get_indices_for_repeated_sample_names,
+)
+from cg.store.store import Store
+
+
+def validate_application_exists(
+    order: OrderWithNonHumanSamples, store: Store, **kwargs
+) -> list[ApplicationNotValidError]:
+    errors: list[ApplicationNotValidError] = []
+    for sample_index, sample in order.enumerated_samples:
+        if not store.get_application_by_tag(sample.application):
+            error = ApplicationNotValidError(sample_index=sample_index)
+            errors.append(error)
+    return errors
+
+
+def validate_applications_not_archived(
+    order: OrderWithNonHumanSamples, store: Store, **kwargs
+) -> list[ApplicationArchivedError]:
+    errors: list[ApplicationArchivedError] = []
+    for sample_index, sample in order.enumerated_samples:
+        if store.is_application_archived(sample.application):
+            error = ApplicationArchivedError(sample_index=sample_index)
+            errors.append(error)
+    return errors
+
+
+def validate_volume_interval(order: OrderWithNonHumanSamples, **kwargs) -> list[InvalidVolumeError]:
+    errors: list[InvalidVolumeError] = []
+    for sample_index, sample in order.enumerated_samples:
+        if is_volume_invalid(sample):
+            error = InvalidVolumeError(sample_index=sample_index)
+            errors.append(error)
+    return errors
+
+
+def validate_organism_exists(
+    order: OrderWithNonHumanSamples, store: Store, **kwargs
+) -> list[OrganismDoesNotExistError]:
+    errors: list[OrganismDoesNotExistError] = []
+    for sample_index, sample in order.enumerated_samples:
+        if not store.get_organism_by_internal_id(sample.organism):
+            error = OrganismDoesNotExistError(sample_index=sample_index)
+            errors.append(error)
+    return errors
+
+
+def validate_buffer_required(
+    order: OrderWithNonHumanSamples,
+    **kwargs,
+) -> list[ElutionBufferMissingError]:
+    errors: list[ElutionBufferMissingError] = []
+    for sample_index, sample in order.enumerated_samples:
+        if not sample.elution_buffer:
+            error = ElutionBufferMissingError(sample_index=sample_index)
+            errors.append(error)
+    return errors
+
+
+def validate_extraction_method_required(
+    order: OrderWithNonHumanSamples, **kwargs
+) -> list[ExtractionMethodMissingError]:
+    errors: list[ExtractionMethodMissingError] = []
+    for sample_index, sample in order.enumerated_samples:
+        if not sample.extraction_method:
+            error = ExtractionMethodMissingError(sample_index=sample_index)
+            errors.append(error)
+    return errors
+
+
+def validate_application_compatibility(
+    order: OrderWithNonHumanSamples,
+    store: Store,
+    **kwargs,
+) -> list[ApplicationNotCompatibleError]:
+    errors: list[ApplicationNotCompatibleError] = []
+    workflow: Workflow = order.workflow
+    allowed_prep_categories: list[PrepCategory] = WORKFLOW_PREP_CATEGORIES[workflow]
+    for sample_index, sample in order.enumerated_samples:
+        incompatible: bool = is_application_not_compatible(
+            allowed_prep_categories=allowed_prep_categories,
+            application_tag=sample.application,
+            store=store,
+        )
+        if incompatible:
+            error = ApplicationNotCompatibleError(sample_index=sample_index)
+            errors.append(error)
+    return errors
+
+
+def validate_wells_contain_at_most_one_sample(
+    order: OrderWithNonHumanSamples,
+    **kwargs,
+) -> list[OccupiedWellError]:
+    plate_samples = PlateSamplesValidator(order)
+    return plate_samples.get_occupied_well_errors()
+
+
+def validate_well_positions_required(
+    order: OrderWithNonHumanSamples,
+    **kwargs,
+) -> list[OccupiedWellError]:
+    plate_samples = PlateSamplesValidator(order)
+    return plate_samples.get_well_position_missing_errors()
+
+
+def validate_sample_names_unique(
+    order: OrderWithNonHumanSamples, **kwargs
+) -> list[SampleNameRepeatedError]:
+    sample_indices: list[int] = get_indices_for_repeated_sample_names(order)
+    return [SampleNameRepeatedError(sample_index=sample_index) for sample_index in sample_indices]

cg/services/order_validation_service/workflows/microsalt/rules/sample/utils.py → cg/services/order_validation_service/rules/sample/utils.py

@@ -4,27 +4,23 @@
     OccupiedWellError,
     WellPositionMissingError,
 )
-from cg.services.order_validation_service.workflows.microsalt.models.order import (
-    MicrosaltOrder,
-)
-from cg.services.order_validation_service.workflows.microsalt.models.sample import (
-    MicrosaltSample,
-)
+from cg.services.order_validation_service.models.order_with_samples import OrderWithNonHumanSamples
+from cg.services.order_validation_service.models.sample import Sample
 
 
 class PlateSamplesValidator:
 
-    def __init__(self, order: MicrosaltOrder):
+    def __init__(self, order: OrderWithNonHumanSamples):
         self.wells: dict[tuple[str, str], list[int]] = {}
-        self.plate_samples: list[tuple[int, MicrosaltSample]] = []
+        self.plate_samples: list[tuple[int, Sample]] = []
         self._initialize_wells(order)
 
-    def _initialize_wells(self, order: MicrosaltOrder):
+    def _initialize_wells(self, order: OrderWithNonHumanSamples):
         """
         Construct a dict with keys being a (container_name, well_position) pair.
         The value will be a list of sample indices for samples located in the well.
         """
-        for sample_index, sample in order.enumerated_new_samples:
+        for sample_index, sample in order.enumerated_samples:
             if sample.is_on_plate:
                 self.plate_samples.append((sample_index, sample))
                 key: tuple[str, str] = (sample.container_name, sample.well_position)
@@ -34,7 +30,7 @@ def _initialize_wells(self, order: MicrosaltOrder):
 
     def get_occupied_well_errors(self) -> list[OccupiedWellError]:
         """Get errors for samples assigned to wells that are already occupied."""
-        conflicting_samples: list[MicrosaltSample] = []
+        conflicting_samples: list[Sample] = []
         for samples_indices in self.wells.values():
             if len(samples_indices) > 1:
                 conflicting_samples.extend(samples_indices[1:])
@@ -57,10 +53,10 @@ def get_missing_well_errors(sample_indices: list[int]) -> list[WellPositionMissi
     return [WellPositionMissingError(sample_index) for sample_index in sample_indices]
 
 
-def get_indices_for_repeated_sample_names(order: MicrosaltOrder) -> list[int]:
+def get_indices_for_repeated_sample_names(order: OrderWithNonHumanSamples) -> list[int]:
     counter = Counter([sample.name for sample in order.samples])
     indices: list[int] = []
-    for index, sample in order.enumerated_new_samples:
+    for index, sample in order.enumerated_samples:
         if counter.get(sample.name) > 1:
             indices.append(index)
     return indices

cg/services/order_validation_service/workflows/microsalt/models/order.py

@@ -1,4 +1,4 @@
-from cg.services.order_validation_service.models.order import Order
+from cg.services.order_validation_service.models.order_with_samples import OrderWithNonHumanSamples
 from cg.services.order_validation_service.workflows.microsalt.constants import (
     MicrosaltDeliveryType,
 )
@@ -7,26 +7,10 @@
 )
 
 
-class MicrosaltOrder(Order):
+class MicrosaltOrder(OrderWithNonHumanSamples):
     delivery_type: MicrosaltDeliveryType
     samples: list[MicrosaltSample]
 
     @property
     def enumerated_samples(self) -> enumerate[MicrosaltSample]:
         return enumerate(self.samples)
-
-    @property
-    def enumerated_new_samples(self) -> list[tuple[int, MicrosaltSample]]:
-        samples: list[tuple[int, MicrosaltSample]] = []
-        for sample_index, sample in self.enumerated_samples:
-            if sample.is_new:
-                samples.append((sample_index, sample))
-        return samples
-
-    @property
-    def enumerated_existing_samples(self) -> list[tuple[int, MicrosaltSample]]:
-        samples: list[tuple[int, MicrosaltSample]] = []
-        for sample_index, sample in self.enumerated_samples:
-            if not sample.is_new:
-                samples.append((sample_index, sample))
-        return samples

cg/services/order_validation_service/workflows/microsalt/validation_rules.py

@@ -1,4 +1,4 @@
-from cg.services.order_validation_service.workflows.microsalt.rules.sample.rules import (
+from cg.services.order_validation_service.rules.sample.rules import (
     validate_application_compatibility,
     validate_application_exists,
     validate_applications_not_archived,