10110106824

Committed 26 Jul 2024 10:54AM UTC coverage: 84.375%. First build

Build # 10110106824

Build Type

Pull #3467

github

Committed by

web-flow

Commit Message

Merge f73d6b83c into 53a3cd219

Pull Request Pull Request #3467: refactor pacbio metrics parser

Run Details

30 of 38 new or added lines in 5 files covered. (78.95%)

20742 of 24583 relevant lines covered (84.38%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.15

/cg/services/post_processing/abstract_classes.py

"""Post-processing service abstract classes."""

from abc import ABC, abstractmethod
from pathlib import Path

from cg.apps.housekeeper.hk import HousekeeperAPI
from cg.services.post_processing.abstract_models import PostProcessingDTOs, RunData, RunMetrics
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):

    def __init__(self, file_manager: RunFileManager):
        self.file_manager: RunFileManager = file_manager

    def parse_metrics(self, run_data: RunData) -> 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

1	"""Post-processing service abstract classes."""
2
3	from abc import ABC, abstractmethod	1✔
4	from pathlib import Path	1✔
5
6	from cg.apps.housekeeper.hk import HousekeeperAPI	1✔
7	from cg.services.post_processing.abstract_models import PostProcessingDTOs, RunData, RunMetrics	1✔
8	from cg.store.store import Store	1✔
9
10
11	class RunDataGenerator(ABC):	1✔
12	"""Abstract class for that holds functionality to create a run data model."""
13
14	@abstractmethod	1✔
15	def _validate_run_name(self, run_name: str) -> None:	1✔
16	pass	×
17
18	@abstractmethod	1✔
19	def get_run_data(self, run_name: str, sequencing_dir: str) -> RunData:	1✔
20	pass	×
21
22
23	class RunFileManager(ABC):	1✔
24	"""Abstract class that manages files related to an instrument run."""
25
26	@abstractmethod	1✔
27	def get_files_to_parse(self, run_data: RunData) -> list[Path]:	1✔
28	"""Get the files required for the PostProcessingMetricsService."""
29	pass	×
30
31	@abstractmethod	1✔
32	def get_files_to_store(self, run_data: RunData) -> list[Path]:	1✔
33	"""Get the files to store for the PostProcessingHKService."""
34	pass	×
35
36
37	class PostProcessingMetricsParser(ABC):	1✔
38
39	def __init__(self, file_manager: RunFileManager):	1✔
NEW 40	self.file_manager: RunFileManager = file_manager	×
41
42	def parse_metrics(self, run_data: RunData) -> RunMetrics:	1✔
43	pass	×
44
45
46	class PostProcessingDataTransferService(ABC):	1✔
47	def __init__(self, metrics_service: PostProcessingMetricsParser):	1✔
48	self.metrics_service = metrics_service	×
49
50	def get_post_processing_dtos(self) -> PostProcessingDTOs:	1✔
51	pass	×
52
53
54	class PostProcessingStoreService(ABC):	1✔
55	def __init__(self, store: Store, data_transfer_service: PostProcessingDataTransferService):	1✔
56	self.store: Store = store	×
57	self.data_transfer_service: PostProcessingDataTransferService = data_transfer_service	×
58
59	def _create_run_device(self, run_name):	1✔
60	pass	×
61
62	def _create_instrument_run(self, run_name):	1✔
63	pass	×
64
65	def _create_sample_run_metrics(self, run_name):	1✔
66	pass	×
67
68	def store_post_processing_data(self, run_name):	1✔
69	pass	×
70
71
72	class PostProcessingHKService(ABC):	1✔
73	def __init__(self, hk_api: HousekeeperAPI):	1✔
74	self.hk_api = HousekeeperAPI	×
75
76	def store_files_in_housekeeper(self, file_to_store: list[Path]):	1✔
77	pass	×
78
79
80	class PostProcessingService(ABC):	1✔
81
82	def __init(	1✔
83	self, store_service: PostProcessingStoreService, hk_service: PostProcessingHKService
84	):
85	self.store_service = store_service	×
86	self.hk_service = hk_service	×
87
88	@abstractmethod	1✔
89	def post_process(self, run_name):	1✔
90	"""Store sequencing metrics in statusdb and relevant files in housekeeper"""
91	pass	×

Clinical-Genomics / cg / 10110106824

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous