feature(dev-branch-pacbio)

cg/services/post_processing/abstract_classes.py

@@ -0,0 +1,91 @@
+"""Post-processing service abstract classes."""
+
+from abc import abstractmethod, ABC
+from pathlib import Path
+
+from pydantic import BaseModel
+
+from cg.apps.housekeeper.hk import HousekeeperAPI
+from cg.services.post_processing.abstract_models import PostProcessingDTOs, RunMetrics, RunData
+from cg.store.store import Store
+
+
+class RunDataGenerator(ABC):
+    """Abstract class for that holds functionality to create a run data model."""
+
+    @abstractmethod
+    def _validate_run_name(self, run_name: str) -> None:
+        pass
+
+    @abstractmethod
+    def get_run_data(self, run_name: str, sequencing_dir: str) -> RunData:
+        pass
+
+
+class RunFileManager(ABC):
+    """Abstract class that manages files related to an instrument run."""
+
+    @abstractmethod
+    def get_files_to_parse(self, run_data: RunData) -> list[Path]:
+        """Get the files required for the PostProcessingMetricsService."""
+        pass
+
+    @abstractmethod
+    def get_files_to_store(self, run_data: RunData) -> list[Path]:
+        """Get the files to store for the PostProcessingHKService."""
+        pass
+
+
+class PostProcessingMetricsParser(ABC):
+
+    @abstractmethod
+    def parse_metrics(self, metrics_paths: list[Path]) -> RunMetrics:
+        pass
+
+
+class PostProcessingDataTransferService(ABC):
+    def __init__(self, metrics_service: PostProcessingMetricsParser):
+        self.metrics_service = metrics_service
+
+    def get_post_processing_dtos(self) -> PostProcessingDTOs:
+        pass
+
+
+class PostProcessingStoreService(ABC):
+    def __init__(self, store: Store, data_transfer_service: PostProcessingDataTransferService):
+        self.store: Store = store
+        self.data_transfer_service: PostProcessingDataTransferService = data_transfer_service
+
+    def _create_run_device(self, run_name):
+        pass
+
+    def _create_instrument_run(self, run_name):
+        pass
+
+    def _create_sample_run_metrics(self, run_name):
+        pass
+
+    def store_post_processing_data(self, run_name):
+        pass
+
+
+class PostProcessingHKService(ABC):
+    def __init__(self, hk_api: HousekeeperAPI):
+        self.hk_api = HousekeeperAPI
+
+    def store_files_in_housekeeper(self, file_to_store: list[Path]):
+        pass
+
+
+class PostProcessingService(ABC):
+
+    def __init(
+        self, store_service: PostProcessingStoreService, hk_service: PostProcessingHKService
+    ):
+        self.store_service = store_service
+        self.hk_service = hk_service
+
+    @abstractmethod
+    def post_process(self, run_name):
+        """Store sequencing metrics in statusdb and relevant files in housekeeper"""
+        pass

cg/services/post_processing/abstract_models.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel
+
+
+class RunData(BaseModel):
+    pass
+
+
+class RunMetrics(BaseModel):
+    pass
+
+
+class PostProcessingDTOs(BaseModel):
+    pass

cg/services/post_processing/exc.py

@@ -0,0 +1,5 @@
+from cg.exc import CgError
+
+
+class PostProcessingRunValidationError(CgError):
+    pass

cg/services/post_processing/pacbio/run_data_generator/pacbio_run_data_generator.py

@@ -0,0 +1,46 @@
+from pathlib import Path
+from cg.services.post_processing.abstract_classes import RunDataGenerator
+from cg.services.post_processing.pacbio.run_data_generator.run_data import PacBioRunData
+from cg.services.post_processing.validators import (
+    validate_name_pre_fix,
+    validate_has_expected_parts,
+)
+from cg.utils.string import get_element_from_split
+
+
+class PacBioRunDataGenerator(RunDataGenerator):
+
+    def _validate_run_name(self, run_name) -> None:
+        validate_name_pre_fix(run_name)
+        validate_has_expected_parts(run_name=run_name, expected_parts=2)
+
+    def get_run_data(self, run_name: str, sequencing_dir: str) -> PacBioRunData:
+        """
+        Get the run data for a PacBio SMRT cell run.
+        run_name should include the PacBio run including plate well, e.g. 'r84202_20240522_133539/1_A01'
+        """
+        self._validate_run_name(run_name)
+        full_path = Path(sequencing_dir, run_name)
+
+        return PacBioRunData(
+            full_path=full_path,
+            sequencing_run_name=self._get_sequencing_run_name(run_name),
+            well_name=self._get_well(run_name),
+            plate=self._get_plate(run_name),
+        )
+
+    @staticmethod
+    def _get_sequencing_run_name(run_name: str) -> str:
+        return get_element_from_split(value=run_name, element_position=0, split="/")
+
+    @staticmethod
+    def _get_plate_well(run_name: str) -> str:
+        return get_element_from_split(value=run_name, element_position=-1, split="/")
+
+    def _get_plate(self, run_name: str) -> str:
+        plate_well: str = self._get_plate_well(run_name)
+        return get_element_from_split(value=plate_well, element_position=0, split="_")
+
+    def _get_well(self, run_name: str) -> str:
+        plate_well: str = self._get_plate_well(run_name)
+        return get_element_from_split(value=plate_well, element_position=-1, split="_")

cg/services/post_processing/pacbio/run_data_generator/run_data.py

@@ -0,0 +1,12 @@
+from pathlib import Path
+
+from cg.services.post_processing.abstract_models import RunData
+
+
+class PacBioRunData(RunData):
+    """Holds information on a single SMRTcell of a PacBio run."""
+
+    full_path: Path
+    sequencing_run_name: str
+    well_name: str
+    plate: int

cg/services/post_processing/validators.py

@@ -0,0 +1,11 @@
+from cg.services.post_processing.exc import PostProcessingRunValidationError
+
+
+def validate_name_pre_fix(run_name: str) -> None:
+    if not run_name.startswith("r"):
+        raise PostProcessingRunValidationError
+
+
+def validate_has_expected_parts(run_name: str, expected_parts: int) -> None:
+    if len(run_name.split("/")) != expected_parts:
+        raise PostProcessingRunValidationError

cg/utils/string.py

@@ -0,0 +1,10 @@
+"""Utils related to string manipulation."""
+
+from cg.exc import CgError
+
+
+def get_element_from_split(value: str, element_position: int, split: str) -> str:
+    elements: list[str] = value.split(split)
+    if len(elements) < element_position:
+        raise CgError(message="Provided element position out of bounds.")
+    return elements[element_position]

tests/fixture_plugins/pacbio_fixtures/name_fixtures.py

@@ -5,6 +5,11 @@
 from cg.constants.pacbio import PacBioDirsAndFiles
 
 
+@pytest.fixture
+def pac_bio_smrt_cell_name() -> str:
+    return "1_A01"
+
+
 @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

@@ -31,9 +31,9 @@ def pac_bio_test_run_dir(pac_bio_runs_dir: Path, pac_bio_test_run_name: str) ->
 
 
 @pytest.fixture
-def pac_bio_smrt_cell_dir_1_a01(pac_bio_test_run_dir: Path) -> Path:
+def pac_bio_smrt_cell_dir_1_a01(pac_bio_test_run_dir: Path, pac_bio_smrt_cell_name: str) -> Path:
     """Return the path to a PacBio SMRT cell directory."""
-    return Path(pac_bio_test_run_dir, "1_A01")
+    return Path(pac_bio_test_run_dir, pac_bio_smrt_cell_name)
 
 
 @pytest.fixture

tests/services/post_processing/pacbio/conftest.py

@@ -0,0 +1,19 @@
+"""Fixtures for the PacBio post processing services."""
+
+from pathlib import Path
+
+import pytest
+
+from cg.services.post_processing.pacbio.run_data_generator.run_data import PacBioRunData
+
+
+@pytest.fixture
+def expected_pac_bio_run_data(
+    pac_bio_test_run_name: str, pac_bio_fixtures_dir: Path, pac_bio_smrt_cell_name: str
+) -> PacBioRunData:
+    return PacBioRunData(
+        full_path=Path(pac_bio_fixtures_dir, pac_bio_test_run_name, pac_bio_smrt_cell_name),
+        sequencing_run_name=pac_bio_test_run_name,
+        well_name="A01",
+        plate="1",
+    )

tests/services/post_processing/pacbio/run_data_generator/test_pacbio_run_data_generator.py

@@ -0,0 +1,27 @@
+"""Tests for the PacBioRunDataGenerator"""
+
+from pathlib import Path
+
+from cg.services.post_processing.pacbio.run_data_generator.pacbio_run_data_generator import (
+    PacBioRunDataGenerator,
+)
+from cg.services.post_processing.pacbio.run_data_generator.run_data import PacBioRunData
+
+
+def test_get_run_data(
+    pac_bio_fixtures_dir: Path,
+    pac_bio_test_run_name: str,
+    pac_bio_smrt_cell_name: str,
+    expected_pac_bio_run_data: PacBioRunData,
+):
+    # GIVEN a run directory, a run name and a SMRT cell name
+    run_name: str = "/".join([pac_bio_test_run_name, pac_bio_smrt_cell_name])
+
+    # WHEN Generating run data
+    run_data_generator = PacBioRunDataGenerator()
+    run_data: PacBioRunData = run_data_generator.get_run_data(
+        run_name=run_name, sequencing_dir=pac_bio_fixtures_dir.as_posix()
+    )
+
+    # THEN the correct run data are returned
+    assert run_data == expected_pac_bio_run_data

tests/utils/test_string_utils.py

@@ -0,0 +1,30 @@
+"""Test for the string utilities,"""
+
+import pytest
+
+from cg.exc import CgError
+from cg.utils.string import get_element_from_split
+
+
+def test_get_element_from_split():
+
+    # GIVEN a string with a seperator
+    separated_string: str = "zero_one_two_three"
+
+    # WHEN getting an element divided by a separator based on the position
+    element: str = get_element_from_split(value=separated_string, element_position=2, split="_")
+
+    # THEN the expected element is returned
+    assert element == "two"
+
+
+def test_get_element_from_split_error():
+
+    # GIVEN a string with a seperator
+    separated_string: str = "zero_one_two_three"
+
+    # WHEN getting an element divided by a separator based on the position that is out of bounds
+    with pytest.raises(CgError):
+        get_element_from_split(value=separated_string, element_position=12, split="_")
+
+        # THEN an error is raised