add(pacbio run validator) (#3738) (patch)

# Description
add run validator for pacbio

Author: ChristianOertlin

cg/constants/pacbio.py

@@ -12,6 +12,8 @@ class PacBioDirsAndFiles:
     SMRTLINK_DATASETS_REPORT: str = "smrtlink-datasets.json"
     STATISTICS_DIR: str = "statistics"
     UNZIPPED_REPORTS_DIR: str = "unzipped_reports"
+    METADATA_DIR: str = "metadata"
+    RUN_IS_VALID: str = "is_valid"
 
 
 class CCSAttributeIDs:
@@ -80,3 +82,6 @@ class PacBioBundleTypes:
     PacBioDirsAndFiles.SMRTLINK_DATASETS_REPORT: PacBioBundleTypes.SMRT_CELL,
     f"{PacBioDirsAndFiles.HIFI_READS}{FileExtensions.BAM}$": PacBioBundleTypes.SAMPLE,
 }
+
+ZIPPED_REPORTS_PATTERN: str = "*reports.zip"
+MANIFEST_FILE_PATTERN: str = "*.transferdone"

cg/models/cg_config.py

@@ -26,6 +26,7 @@
 from cg.constants.priority import SlurmQos
 from cg.meta.delivery.delivery import DeliveryAPI
 from cg.services.analysis_service.analysis_service import AnalysisService
+from cg.services.decompression_service.decompressor import Decompressor
 from cg.services.fastq_concatenation_service.fastq_concatenation_service import (
     FastqConcatenationService,
 )
@@ -54,11 +55,15 @@
 from cg.services.run_devices.pacbio.run_file_manager.run_file_manager import (
     PacBioRunFileManager,
 )
+from cg.services.run_devices.pacbio.run_validator.pacbio_run_validator import PacBioRunValidator
 from cg.services.sequencing_qc_service.sequencing_qc_service import SequencingQCService
 from cg.services.slurm_service.slurm_cli_service import SlurmCLIService
 from cg.services.slurm_service.slurm_service import SlurmService
 from cg.services.slurm_upload_service.slurm_upload_config import SlurmUploadConfig
 from cg.services.slurm_upload_service.slurm_upload_service import SlurmUploadService
+from cg.services.validate_file_transfer_service.validate_file_transfer_service import (
+    ValidateFileTransferService,
+)
 from cg.store.database import initialize_database
 from cg.store.store import Store
 
@@ -612,6 +617,11 @@ def get_pacbio_post_processing_service(self) -> PacBioPostProcessingService:
         LOG.debug("Instantiating PacBio post-processing service")
         run_data_generator = PacBioRunDataGenerator()
         file_manager = PacBioRunFileManager()
+        run_validator = PacBioRunValidator(
+            file_manager=file_manager,
+            decompressor=Decompressor(),
+            file_transfer_validator=ValidateFileTransferService(),
+        )
         metrics_parser = PacBioMetricsParser(file_manager=file_manager)
         transfer_service = PacBioDataTransferService(metrics_service=metrics_parser)
         store_service = PacBioStoreService(
@@ -623,6 +633,7 @@ def get_pacbio_post_processing_service(self) -> PacBioPostProcessingService:
             metrics_parser=metrics_parser,
         )
         return PacBioPostProcessingService(
+            run_validator=run_validator,
             run_data_generator=run_data_generator,
             hk_service=hk_service,
             store_service=store_service,

cg/services/decompression_service/decompressor.py

@@ -6,6 +6,8 @@ class Decompressor:
 
     def decompress(self, source_path: Path, destination_path: Path) -> Path:
         """Decompress a file to a directory."""
+        if not source_path.exists():
+            raise FileNotFoundError(f"Source path {source_path} does not exist.")
         with zipfile.ZipFile(source_path, "r") as zip_file:
             zip_file.extractall(destination_path)
         if not self._decompressed_path_exists(destination_path):

cg/services/run_devices/abstract_classes.py

@@ -16,6 +16,15 @@ def get_run_data(self, run_name: str, sequencing_dir: str) -> RunData:
         pass
 
 
+class RunValidator(ABC):
+    """Abstract class that holds functionality to validate a run and ensure it can start post processing."""
+
+    @abstractmethod
+    def ensure_post_processing_can_start(self, run_data: RunData):
+        """Ensure a post processing run can start."""
+        pass
+
+
 class RunFileManager(ABC):
     """Abstract class that manages files related to an instrument run."""
 
@@ -79,12 +88,3 @@ def _touch_post_processing_complete(run_data: RunData) -> None:
         """Touch the post-processing complete file."""
         processing_complete_file = Path(run_data.full_path, POST_PROCESSING_COMPLETED)
         processing_complete_file.touch()
-
-
-class FileTransferValidationService(ABC):
-    """Abstract class that validates file transfers for instrument runs from NAS to Hasta."""
-
-    @abstractmethod
-    def validate_file_transfer(self, run_data: RunData):
-        """Validate an instrument run transfer."""
-        pass

cg/services/run_devices/pacbio/post_processing_service.py

@@ -20,6 +20,7 @@
     PacBioRunDataGenerator,
 )
 from cg.services.run_devices.pacbio.run_data_generator.run_data import PacBioRunData
+from cg.services.run_devices.pacbio.run_validator.pacbio_run_validator import PacBioRunValidator
 
 LOG = logging.getLogger(__name__)
 
@@ -29,11 +30,13 @@ class PacBioPostProcessingService(PostProcessingService):
 
     def __init__(
         self,
+        run_validator: PacBioRunValidator,
         run_data_generator: PacBioRunDataGenerator,
         hk_service: PacBioHousekeeperService,
         store_service: PacBioStoreService,
         sequencing_dir: str,
     ):
+        self.run_validator: PacBioRunValidator = run_validator
         self.run_data_generator: PacBioRunDataGenerator = run_data_generator
         self.hk_service: PacBioHousekeeperService = hk_service
         self.store_service: PacBioStoreService = store_service
@@ -52,6 +55,7 @@ def post_process(self, run_name: str, dry_run: bool = False) -> None:
         run_data: PacBioRunData = self.run_data_generator.get_run_data(
             run_name=run_name, sequencing_dir=self.sequencing_dir
         )
+        self.run_validator.ensure_post_processing_can_start(run_data)
         self.store_service.store_post_processing_data(run_data=run_data, dry_run=dry_run)
         self.hk_service.store_files_in_housekeeper(run_data=run_data, dry_run=dry_run)
-        self._touch_post_processing_complete(run_data=run_data)
+        self._touch_post_processing_complete(run_data)

cg/services/run_devices/pacbio/run_file_manager/models.py

@@ -0,0 +1,9 @@
+from pathlib import Path
+
+from pydantic import BaseModel
+
+
+class PacBioRunValidatorFiles(BaseModel):
+    manifest_file: Path
+    decompression_target: Path
+    decompression_destination: Path

cg/services/run_devices/pacbio/run_file_manager/run_file_manager.py

@@ -1,11 +1,12 @@
 from pathlib import Path
 
 from cg.constants import FileExtensions
-from cg.constants.pacbio import PacBioDirsAndFiles
+from cg.constants.pacbio import PacBioDirsAndFiles, MANIFEST_FILE_PATTERN, ZIPPED_REPORTS_PATTERN
 from cg.services.run_devices.abstract_classes import RunFileManager
 from cg.services.run_devices.error_handler import handle_post_processing_errors
 from cg.services.run_devices.exc import PostProcessingRunFileManagerError
 from cg.services.run_devices.pacbio.run_data_generator.run_data import PacBioRunData
+from cg.services.run_devices.pacbio.run_file_manager.models import PacBioRunValidatorFiles
 from cg.services.run_devices.validators import validate_files_or_directories_exist
 from cg.utils.files import get_files_matching_pattern
 
@@ -29,10 +30,22 @@ def get_files_to_store(self, run_data: PacBioRunData) -> list[Path]:
         run_path: Path = run_data.full_path
         return self.get_files_to_parse(run_data) + self._get_hifi_read_files(run_path)
 
-    @staticmethod
-    def _get_ccs_report_file(run_path: Path) -> Path:
+    @handle_post_processing_errors(
+        to_except=(FileNotFoundError,), to_raise=PostProcessingRunFileManagerError
+    )
+    def get_run_validation_files(self, run_data: PacBioRunData) -> PacBioRunValidatorFiles:
+        manifest_file: Path = self._get_manifest_file(run_data.full_path)
+        decompression_target: Path = self._get_zipped_reports_file(run_data.full_path)
+        decompression_destination: Path = self._get_unzipped_reports_dir(run_data.full_path)
+        return PacBioRunValidatorFiles(
+            manifest_file=manifest_file,
+            decompression_target=decompression_target,
+            decompression_destination=decompression_destination,
+        )
+
+    def _get_ccs_report_file(self, run_path: Path) -> Path:
         """Return the path to the CCS report file."""
-        statistics_dir: Path = Path(run_path, PacBioDirsAndFiles.STATISTICS_DIR)
+        statistics_dir: Path = self._get_statistics_dir(run_path)
         files: list[Path] = get_files_matching_pattern(
             directory=statistics_dir, pattern=f"*{PacBioDirsAndFiles.CCS_REPORT_SUFFIX}"
         )
@@ -42,9 +55,7 @@ def _get_ccs_report_file(run_path: Path) -> Path:
 
     def _get_report_files(self, run_path: Path) -> list[Path]:
         """Return the paths to the unzipped report files."""
-        unzipped_dir: Path = Path(
-            run_path, PacBioDirsAndFiles.STATISTICS_DIR, PacBioDirsAndFiles.UNZIPPED_REPORTS_DIR
-        )
+        unzipped_dir: Path = self._get_unzipped_reports_dir(run_path)
         report_files: list[Path] = [
             Path(unzipped_dir, PacBioDirsAndFiles.CONTROL_REPORT),
             Path(unzipped_dir, PacBioDirsAndFiles.LOADING_REPORT),
@@ -64,3 +75,28 @@ def _get_hifi_read_files(run_path: Path) -> list[Path]:
         )
         validate_files_or_directories_exist(bam_files)
         return bam_files
+
+    @staticmethod
+    def _get_unzipped_reports_dir(run_path) -> Path:
+        return Path(
+            run_path, PacBioDirsAndFiles.STATISTICS_DIR, PacBioDirsAndFiles.UNZIPPED_REPORTS_DIR
+        )
+
+    @staticmethod
+    def _get_statistics_dir(run_path) -> Path:
+        return Path(run_path, PacBioDirsAndFiles.STATISTICS_DIR)
+
+    @staticmethod
+    def _get_manifest_file(run_path) -> Path:
+        file_list: list[Path] = get_files_matching_pattern(
+            directory=Path(run_path, PacBioDirsAndFiles.METADATA_DIR), pattern=MANIFEST_FILE_PATTERN
+        )
+        if not file_list:
+            raise FileNotFoundError(f"No Manifest file found in {run_path}")
+        return file_list[0]
+
+    def _get_zipped_reports_file(self, run_path) -> Path:
+        return get_files_matching_pattern(
+            directory=self._get_statistics_dir(run_path),
+            pattern=ZIPPED_REPORTS_PATTERN,
+        )[0]

cg/services/run_devices/pacbio/run_validator/pacbio_run_validator.py

@@ -0,0 +1,66 @@
+from pathlib import Path
+import logging
+from cg.constants.constants import FileFormat
+from cg.constants.pacbio import PacBioDirsAndFiles
+from cg.services.decompression_service.decompressor import Decompressor
+from cg.services.run_devices.abstract_classes import RunValidator
+from cg.services.run_devices.pacbio.run_data_generator.run_data import PacBioRunData
+from cg.services.run_devices.pacbio.run_file_manager.models import PacBioRunValidatorFiles
+from cg.services.run_devices.pacbio.run_file_manager.run_file_manager import PacBioRunFileManager
+from cg.services.validate_file_transfer_service.validate_file_transfer_service import (
+    ValidateFileTransferService,
+)
+
+LOG = logging.getLogger(__name__)
+
+
+class PacBioRunValidator(RunValidator):
+    """
+    PacBio run validator.
+    Ensure that the post-processing of a pacbio run can start.
+    """
+
+    def __init__(
+        self,
+        decompressor: Decompressor,
+        file_transfer_validator: ValidateFileTransferService,
+        file_manager: PacBioRunFileManager,
+    ):
+        self.decompressor = decompressor
+        self.file_transfer_validator = file_transfer_validator
+        self.file_manager = file_manager
+
+    def ensure_post_processing_can_start(self, run_data: PacBioRunData) -> None:
+        """
+        Ensure that a post-processing run can start.
+        1. Check if all files are present listed in a manifest file.
+        2. Decompresses the zipped reports.
+        3. Touches a file to indicate that the run is validated
+        4. Skips validation if the run is already validated
+        """
+        if self._is_validated(run_data.full_path):
+            LOG.debug(f"Run for {run_data.full_path} is validated.")
+            return
+        paths_information: PacBioRunValidatorFiles = self.file_manager.get_run_validation_files(
+            run_data
+        )
+        self.file_transfer_validator.validate_file_transfer(
+            manifest_file=paths_information.manifest_file,
+            source_dir=run_data.full_path,
+            manifest_file_format=FileFormat.TXT,
+        )
+
+        self.decompressor.decompress(
+            source_path=paths_information.decompression_target,
+            destination_path=paths_information.decompression_destination,
+        )
+        self._touch_is_validated(run_data.full_path)
+        LOG.debug(f"Run for {run_data.full_path} is validated.")
+
+    @staticmethod
+    def _touch_is_validated(run_path: Path):
+        Path(run_path, PacBioDirsAndFiles.RUN_IS_VALID).touch()
+
+    @staticmethod
+    def _is_validated(run_path: Path) -> bool:
+        return Path(run_path, PacBioDirsAndFiles.RUN_IS_VALID).exists()

tests/fixture_plugins/pacbio_fixtures/name_fixtures.py

@@ -8,6 +8,11 @@ def pac_bio_smrt_cell_name() -> str:
     return "1_A01"
 
 
+@pytest.fixture
+def pac_bio_another_smrt_cell_name() -> str:
+    return "1_B01"
+
+
 @pytest.fixture
 def pac_bio_test_run_name() -> str:
     """Return the name of a PacBio SMRT cell."""

tests/fixture_plugins/pacbio_fixtures/path_fixtures.py

@@ -5,6 +5,7 @@
 import pytest
 
 from cg.constants.pacbio import PacBioDirsAndFiles
+from cg.services.run_devices.pacbio.run_file_manager.models import PacBioRunValidatorFiles
 
 
 # Directory fixtures
@@ -38,6 +39,14 @@ def pac_bio_smrt_cell_dir_1_a01(pac_bio_test_run_dir: Path, pac_bio_smrt_cell_na
     return Path(pac_bio_test_run_dir, pac_bio_smrt_cell_name)
 
 
+@pytest.fixture
+def pac_bio_smrt_cell_dir_1_b01(
+    pac_bio_test_run_dir: Path, pac_bio_another_smrt_cell_name: str
+) -> Path:
+    """Return the path to a PacBio SMRT cell directory."""
+    return Path(pac_bio_test_run_dir, pac_bio_another_smrt_cell_name)
+
+
 @pytest.fixture
 def pac_bio_hifi_reads_dir(pac_bio_smrt_cell_dir_1_a01: Path) -> Path:
     """Return the path to a PacBio HiFi reads directory."""
@@ -50,12 +59,30 @@ def pac_bio_run_statistics_dir(pac_bio_smrt_cell_dir_1_a01: Path) -> Path:
     return Path(pac_bio_smrt_cell_dir_1_a01, PacBioDirsAndFiles.STATISTICS_DIR)
 
 
+@pytest.fixture
+def pac_bio_run_statistics_dir_1_b01(pac_bio_smrt_cell_dir_1_b01: Path) -> Path:
+    """Return the path to the PacBio SMRT cell statistics directory."""
+    return Path(pac_bio_smrt_cell_dir_1_b01, PacBioDirsAndFiles.STATISTICS_DIR)
+
+
 @pytest.fixture
 def pac_bio_run_reports_dir(pac_bio_run_statistics_dir: Path) -> Path:
     """Return the path to the PacBio SMRT cell unzipped_reports directory"""
     return Path(pac_bio_run_statistics_dir, PacBioDirsAndFiles.UNZIPPED_REPORTS_DIR)
 
 
+@pytest.fixture
+def pac_bio_run_reports_dir_1_b01(pac_bio_run_statistics_dir_1_b01: Path) -> Path:
+    """Return the path to the PacBio SMRT cell unzipped_reports directory"""
+    return Path(pac_bio_run_statistics_dir_1_b01, PacBioDirsAndFiles.UNZIPPED_REPORTS_DIR)
+
+
+@pytest.fixture
+def pac_bio_run_metadata_dir_1_b01(pac_bio_smrt_cell_dir_1_b01: Path) -> Path:
+    """Return the path to the PacBio SMRT cell metadata directory"""
+    return Path(pac_bio_smrt_cell_dir_1_b01, PacBioDirsAndFiles.METADATA_DIR)
+
+
 @pytest.fixture
 def pac_bio_wrong_metrics_file(pac_bio_wrong_metrics_dir: Path) -> Path:
     """Return the path to a temporary PacBio statistics directory."""
@@ -100,6 +127,18 @@ def pac_bio_smrtlink_datasets_report_file(pac_bio_run_reports_dir: Path) -> Path
     return Path(pac_bio_run_reports_dir, PacBioDirsAndFiles.SMRTLINK_DATASETS_REPORT)
 
 
+@pytest.fixture
+def pac_bio_transferdone_file_1_b01(pac_bio_run_metadata_dir_1_b01: Path) -> Path:
+    """Return the path to the PacBio SMRTLink datasets report file."""
+    return Path(pac_bio_run_metadata_dir_1_b01, "m84202_240522_155607_s2.transferdone")
+
+
+@pytest.fixture
+def pac_bio_zipped_reports_file_1_b01(pac_bio_run_statistics_dir_1_b01: Path) -> Path:
+    """Return the path to the PacBio SMRTLink datasets report file."""
+    return Path(pac_bio_run_statistics_dir_1_b01, "m84202_240522_155607_s2.reports.zip")
+
+
 @pytest.fixture
 def pac_bio_report_files_to_parse(
     pac_bio_ccs_report_file: Path,
@@ -122,3 +161,17 @@ def pac_bio_report_files_to_parse(
 def pac_bio_hifi_read_file(pac_bio_hifi_reads_dir: Path, pac_bio_1_a01_cell_full_name: str) -> Path:
     """Return the PacBio HiFi read file."""
     return Path(pac_bio_hifi_reads_dir, f"{pac_bio_1_a01_cell_full_name}.hifi_reads.bam")
+
+
+@pytest.fixture
+def expected_1_b01_run_validation_files(
+    pac_bio_transferdone_file_1_b01: Path,
+    pac_bio_zipped_reports_file_1_b01: Path,
+    pac_bio_run_reports_dir_1_b01: Path,
+) -> PacBioRunValidatorFiles:
+    """Return the expected run validation files for the 1_B01 SMRT cell."""
+    return PacBioRunValidatorFiles(
+        manifest_file=pac_bio_transferdone_file_1_b01,
+        decompression_target=pac_bio_zipped_reports_file_1_b01,
+        decompression_destination=pac_bio_run_reports_dir_1_b01,
+    )