feature(streamline mutant delivery and upload)

cg/apps/lims/api.py

@@ -555,3 +555,9 @@ def _get_negative_controls_from_list(samples: list[Sample]) -> list[Sample]:
             ):
                 negative_controls.append(sample)
         return negative_controls
+
+    def get_sample_region_and_lab_code(self, sample_id: str) -> str:
+        """Return the reqgion code and lab code for a sample formatted as a suffix string."""
+        region_code = self.get_sample_attribute(lims_id=sample_id, key="region_code").split(" ")[0]
+        lab_code = self.get_sample_attribute(lims_id=sample_id, key="lab_code").split(" ")[0]
+        return f"{region_code}_{lab_code}_"

cg/models/cg_config.py

@@ -755,6 +755,7 @@ def delivery_service_factory(self) -> DeliveryServiceFactory:
             LOG.debug("Instantiating delivery service factory")
             factory = DeliveryServiceFactory(
                 store=self.status_db,
+                lims_api=self.lims_api,
                 hk_api=self.housekeeper_api,
                 tb_service=self.trailblazer_api,
                 rsync_service=self.delivery_rsync_service,

cg/services/deliver_files/deliver_files_service/deliver_files_service_factory.py

@@ -3,6 +3,7 @@
 from typing import Type
 
 from cg.apps.housekeeper.hk import HousekeeperAPI
+from cg.apps.lims import LimsAPI
 from cg.apps.tb import TrailblazerAPI
 from cg.constants import DataDelivery, Workflow
 from cg.constants.constants import PrepCategory
@@ -49,12 +50,14 @@ class DeliveryServiceFactory:
     def __init__(
         self,
         store: Store,
+        lims_api: LimsAPI,
         hk_api: HousekeeperAPI,
         rsync_service: DeliveryRsyncService,
         tb_service: TrailblazerAPI,
         analysis_service: AnalysisService,
     ):
         self.store = store
+        self.lims_api = lims_api
         self.hk_api = hk_api
         self.rsync_service = rsync_service
         self.tb_service = tb_service

cg/services/deliver_files/file_formatter/utils/mutant_sample_service.py

@@ -0,0 +1,87 @@
+from pathlib import Path
+
+from cg.apps.lims import LimsAPI
+from cg.services.deliver_files.file_fetcher.models import SampleFile
+from cg.services.deliver_files.file_formatter.models import FormattedFile
+from cg.services.deliver_files.file_formatter.utils.sample_concatenation_service import (
+    SampleFileConcatenationFormatter,
+)
+from cg.services.deliver_files.file_formatter.utils.sample_service import FileManagingService
+
+
+class MutantFileFormatter:
+    def __init__(
+        self,
+        lims_api: LimsAPI,
+        file_formatter: SampleFileConcatenationFormatter,
+        file_manager: FileManagingService,
+    ):
+        self.lims_api: LimsAPI = lims_api
+        self.file_formatter: SampleFileConcatenationFormatter = file_formatter
+        self.file_manager = file_manager
+
+    def format_files(
+        self, moved_files: list[SampleFile], ticket_dir_path: Path
+    ) -> list[FormattedFile]:
+        formatted_files: list[FormattedFile] = self.file_formatter.format_files(
+            moved_files=moved_files, ticket_dir_path=ticket_dir_path
+        )
+        appended_formatted_files: list[FormattedFile] = self._add_lims_metadata_to_file_name(
+            formatted_files=formatted_files, sample_files=moved_files
+        )
+        unique_formatted_files: list[FormattedFile] = self._filter_unique_path_combinations(
+            appended_formatted_files
+        )
+        for unique_files in unique_formatted_files:
+            self.file_manager.rename_file(
+                src=unique_files.original_path, dst=unique_files.formatted_path
+            )
+        return unique_formatted_files
+
+    def _add_lims_metadata_to_file_name(
+        self, formatted_files: list[FormattedFile], sample_files: list[SampleFile]
+    ) -> list[FormattedFile]:
+        """This functions adds the region and lab code to the file name of the formatted files."""
+        appended_formatted_files: list[FormattedFile] = []
+        for formatted_file in formatted_files:
+            sample_id: str = self._get_sample_id_by_original_path(
+                original_path=formatted_file.original_path, sample_files=sample_files
+            )
+            lims_meta_data = self.lims_api.get_sample_region_and_lab_code(sample_id)
+
+            new_original_path: Path = formatted_file.formatted_path
+            new_formatted_path = Path(
+                formatted_file.formatted_path.parent,
+                f"{lims_meta_data}{formatted_file.formatted_path.name}",
+            )
+            appended_formatted_files.append(
+                FormattedFile(original_path=new_original_path, formatted_path=new_formatted_path)
+            )
+        return appended_formatted_files
+
+    @staticmethod
+    def _get_sample_id_by_original_path(original_path: Path, sample_files: list[SampleFile]) -> str:
+        for sample_file in sample_files:
+            if sample_file.file_path == original_path:
+                return sample_file.sample_id
+        raise ValueError(f"Could not find sample file with path {original_path}")
+
+    @staticmethod
+    def _filter_unique_path_combinations(
+        formatted_files: list[FormattedFile],
+    ) -> list[FormattedFile]:
+        """
+        During fastq concatenation Sample_R1 and Sample_R2 files are concatenated and moved to the same file Concat_Sample.
+        This mean that there can be multiple entries for the same concatenated file in the formatted_files list coming
+        from the SampleFileConcatenationService.
+        This function filters out the duplicates to avoid moving the same file multiple times
+        which would result in an error the second time since the files is no longer in the original path.
+        """
+        unique_combinations = set()
+        unique_files: list[FormattedFile] = []
+        for formatted_file in formatted_files:
+            combination = (formatted_file.original_path, formatted_file.formatted_path)
+            if combination not in unique_combinations:
+                unique_combinations.add(combination)
+                unique_files.append(formatted_file)
+        return unique_files

cg/services/deliver_files/file_formatter/utils/sample_concatenation_service.py

@@ -1,5 +1,6 @@
 from pathlib import Path
 
+from cg.apps.lims import LimsAPI
 from cg.constants.constants import ReadDirection, FileFormat, FileExtensions
 
 from cg.services.fastq_concatenation_service.fastq_concatenation_service import (

cg/services/deliver_files/file_formatter/utils/sample_service.py

@@ -1,5 +1,6 @@
 import os
 from pathlib import Path
+
 from cg.services.deliver_files.file_fetcher.models import SampleFile
 from cg.services.deliver_files.file_formatter.models import FormattedFile
 

tests/fixture_plugins/delivery_fixtures/delivery_files_models_fixtures.py

@@ -1,4 +1,4 @@
-from pathlib import Path
+from pathlib import Path, PosixPath
 
 import pytest
 
@@ -15,6 +15,7 @@
     DeliveryMetaData,
     SampleFile,
 )
+from cg.services.deliver_files.file_formatter.models import FormattedFile
 from cg.store.models import Case
 from cg.store.store import Store
 
@@ -243,3 +244,29 @@ def swap_file_paths_with_inbox_paths(
         new_file_model.file_path = Path(inbox_dir_path, file_model.file_path.name)
         new_file_models.append(new_file_model)
     return new_file_models
+
+
+@pytest.fixture
+def lims_naming_matadata() -> str:
+    return "01_SE100_"
+
+
+@pytest.fixture
+def expected_mutant_formatted_files(
+    expected_concatenated_fastq_formatted_files, lims_naming_matadata
+) -> list[FormattedFile]:
+    unique_combinations = []
+    for formatted_file in expected_concatenated_fastq_formatted_files:
+        formatted_file.original_path = formatted_file.formatted_path
+        formatted_file.formatted_path = Path(
+            formatted_file.formatted_path.parent,
+            f"{lims_naming_matadata}{formatted_file.formatted_path.name}",
+        )
+        if formatted_file not in unique_combinations:
+            unique_combinations.append(formatted_file)
+    return unique_combinations
+
+
+@pytest.fixture
+def mutant_moved_files(fastq_concatenation_sample_files) -> list[SampleFile]:
+    return fastq_concatenation_sample_files

tests/services/file_delivery/delivery_file_service/test_service_builder.py

@@ -80,6 +80,7 @@ class DeliveryServiceScenario(BaseModel):
 def test_build_delivery_service(scenario: DeliveryServiceScenario, request: FixtureRequest):
     # GIVEN a delivery service builder with mocked store and hk_api
     builder = DeliveryServiceFactory(
+        lims_api=MagicMock(),
         store=request.getfixturevalue(scenario.store_name),
         hk_api=MagicMock(),
         rsync_service=MagicMock(),

tests/services/file_delivery/file_formatter/utils/test_formatter_utils.py

@@ -1,7 +1,11 @@
 import os
+from unittest import mock
+from unittest.mock import Mock
 import pytest
 from pathlib import Path
 
+from cg.apps.lims import LimsAPI
+from cg.services.deliver_files.file_formatter.utils.mutant_sample_service import MutantFileFormatter
 from cg.services.fastq_concatenation_service.fastq_concatenation_service import (
     FastqConcatenationService,
 )
@@ -78,3 +82,43 @@ def test_file_formatter_utils(
     for file in formatted_files:
         assert file.formatted_path.exists()
         assert not file.original_path.exists()
+
+
+def test_mutant_file_formatter(
+    mutant_moved_files: list[SampleFile],
+    expected_mutant_formatted_files: list[FormattedFile],
+    lims_naming_matadata: str,
+):
+    # GIVEN existing ticket directory path and a customer inbox
+    ticket_dir_path: Path = mutant_moved_files[0].file_path.parent
+
+    os.makedirs(ticket_dir_path, exist_ok=True)
+
+    for moved_file in mutant_moved_files:
+        moved_file.file_path.touch()
+
+    lims_mock = Mock()
+    lims_mock.get_sample_region_and_lab_code.return_value = lims_naming_matadata
+
+    # Initialize file_formatter
+    file_formatter = MutantFileFormatter(
+        file_manager=FileManagingService(),
+        file_formatter=SampleFileConcatenationFormatter(
+            file_manager=FileManagingService(),
+            file_formatter=SampleFileNameFormatter(),
+            concatenation_service=FastqConcatenationService(),
+        ),
+        lims_api=lims_mock,
+    )
+
+    # WHEN formatting the files
+    formatted_files: list[FormattedFile] = file_formatter.format_files(
+        moved_files=mutant_moved_files,
+        ticket_dir_path=ticket_dir_path,
+    )
+
+    # THEN the files should be formatted
+    assert formatted_files == expected_mutant_formatted_files
+    for file in formatted_files:
+        assert file.formatted_path.exists()
+        assert not file.original_path.exists()