10491698847

Committed 21 Aug 2024 02:18PM UTC coverage: 83.391%. First build

Build # 10491698847

Build Type

Pull #3300

github

Committed by

web-flow

Commit Message

Merge 822323b1a into 826231296

Pull Request Pull Request #3300: Add mutant qc

Run Details

290 of 379 new or added lines in 14 files covered. (76.52%)

17046 of 20441 relevant lines covered (83.39%)

1.08 hits per line

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

94.85

/cg/meta/workflow/mutant/quality_controller/quality_controller.py

from pathlib import Path
from cg.apps.lims.api import LimsAPI
from cg.constants.constants import MutantQC
from cg.constants.lims import LimsProcess
from cg.exc import CgError
from cg.meta.workflow.mutant.quality_controller.metrics_parser_utils import parse_samples_results
from cg.meta.workflow.mutant.quality_controller.models import (
    MutantPoolSamples,
    SamplePoolAndResults,
    SampleQualityResults,
    CaseQualityResult,
    MutantQualityResult,
    ParsedSampleResults,
    SamplesQualityResults,
)
from cg.meta.workflow.mutant.quality_controller.report_generator_utils import (
    get_summary,
    write_report,
)
from cg.meta.workflow.mutant.quality_controller.result_logger_utils import (
    log_case_result,
    log_results,
    log_sample_result,
)
from cg.meta.workflow.mutant.quality_controller.utils import (
    has_external_negative_control_sample_valid_total_reads,
    has_internal_negative_control_sample_valid_total_reads,
    has_sample_valid_total_reads,
)
from cg.store.models import Case, Sample
from cg.store.store import Store


class MutantQualityController:
    def __init__(self, status_db: Store, lims: LimsAPI) -> None:
        self.status_db: Store = status_db
        self.lims: LimsAPI = lims

    def get_quality_control_result(
        self, case: Case, case_results_file_path: Path, case_qc_report_path: Path
    ) -> MutantQualityResult:
        """Perform QC check on a case and generate the QC_report."""
        sample_pool_and_results: SamplePoolAndResults = self._get_sample_pool_and_results(
            case_results_file_path=case_results_file_path,
            case=case,
        )

        samples_quality_results: SamplesQualityResults = self._get_samples_quality_results(
            sample_pool_and_results=sample_pool_and_results
        )
        case_quality_result: CaseQualityResult = self._get_case_quality_result(
            samples_quality_results
        )

        write_report(
            case_qc_report_path=case_qc_report_path,
            samples_quality_results=samples_quality_results,
            case_quality_result=case_quality_result,
        )

        log_results(
            case_quality_result=case_quality_result,
            samples_quality_results=samples_quality_results,
            report_file_path=case_qc_report_path,
        )

        summary: str = get_summary(
            case_quality_result=case_quality_result,
            samples_quality_results=samples_quality_results,
        )

        return MutantQualityResult(
            case_quality_result=case_quality_result,
            samples_quality_results=samples_quality_results,
            summary=summary,
        )

    def _get_samples_quality_results(
        self, sample_pool_and_results: SamplePoolAndResults
    ) -> SamplesQualityResults:
        samples_quality_results: list[SampleQualityResults] = []
        for sample in sample_pool_and_results.pool.samples:
            sample_results: ParsedSampleResults = sample_pool_and_results.results[
                sample.internal_id
            ]
            sample_quality_results: (
                SampleQualityResults
            ) = self._get_sample_quality_result_for_sample(
                sample=sample, sample_results=sample_results
            )
            samples_quality_results.append(sample_quality_results)

        internal_negative_control_sample: (
            Sample
        ) = sample_pool_and_results.pool.internal_negative_control
        internal_negative_control_quality_metrics: (
            SampleQualityResults
        ) = self._get_sample_quality_result_for_internal_negative_control_sample(
            sample=internal_negative_control_sample
        )

        external_negative_control_sample: (
            Sample
        ) = sample_pool_and_results.pool.external_negative_control
        external_negative_control_sample_results: (
            ParsedSampleResults
        ) = sample_pool_and_results.results[external_negative_control_sample.internal_id]
        external_negative_control_quality_metrics: (
            SampleQualityResults
        ) = self._get_sample_quality_result_for_external_negative_control_sample(
            sample=external_negative_control_sample,
            sample_results=external_negative_control_sample_results,
        )

        return SamplesQualityResults(
            samples=samples_quality_results,
            internal_negative_control=internal_negative_control_quality_metrics,
            external_negative_control=external_negative_control_quality_metrics,
        )

    @staticmethod
    def _get_sample_quality_result_for_sample(
        sample: Sample, sample_results: ParsedSampleResults
    ) -> SampleQualityResults:
        does_sample_pass_reads_threshold: bool = has_sample_valid_total_reads(sample=sample)
        does_sample_pass_qc: bool = does_sample_pass_reads_threshold and sample_results.passes_qc
        sample_quality_result = SampleQualityResults(
            sample_id=sample.internal_id,
            passes_qc=does_sample_pass_qc,
            passes_reads_threshold=does_sample_pass_reads_threshold,
            passes_mutant_qc=sample_results.passes_qc,
        )

        log_sample_result(
            result=sample_quality_result,
        )
        return sample_quality_result

    @staticmethod
    def _get_sample_quality_result_for_internal_negative_control_sample(
        sample: Sample,
    ) -> SampleQualityResults:
        does_sample_pass_reads_threshold: (
            bool
        ) = has_internal_negative_control_sample_valid_total_reads(sample=sample)
        sample_quality_result = SampleQualityResults(
            sample_id=sample.internal_id,
            passes_qc=does_sample_pass_reads_threshold,
            passes_reads_threshold=does_sample_pass_reads_threshold,
        )

        log_sample_result(result=sample_quality_result, is_external_negative_control=True)
        return sample_quality_result

    @staticmethod
    def _get_sample_quality_result_for_external_negative_control_sample(
        sample: Sample, sample_results: ParsedSampleResults
    ) -> SampleQualityResults:
        does_sample_pass_reads_threshold: (
            bool
        ) = has_external_negative_control_sample_valid_total_reads(sample=sample)
        sample_passes_qc: bool = does_sample_pass_reads_threshold and not sample_results.passes_qc
        sample_quality_result = SampleQualityResults(
            sample_id=sample.internal_id,
            passes_qc=sample_passes_qc,
            passes_reads_threshold=does_sample_pass_reads_threshold,
            passes_mutant_qc=sample_results.passes_qc,
        )

        log_sample_result(result=sample_quality_result, is_external_negative_control=True)
        return sample_quality_result

    def _get_case_quality_result(
        self, samples_quality_results: SamplesQualityResults
    ) -> CaseQualityResult:
        external_negative_control_pass_qc: (
            bool
        ) = samples_quality_results.external_negative_control.passes_qc
        internal_negative_control_pass_qc: (
            bool
        ) = samples_quality_results.internal_negative_control.passes_qc

        samples_pass_qc: bool = self._samples_pass_qc(
            samples_quality_results=samples_quality_results
        )

        case_passes_qc: bool = (
            samples_pass_qc
            and internal_negative_control_pass_qc
            and external_negative_control_pass_qc
        )

        result = CaseQualityResult(
            passes_qc=case_passes_qc,
            internal_negative_control_passes_qc=internal_negative_control_pass_qc,
            external_negative_control_passes_qc=external_negative_control_pass_qc,
            fraction_samples_passes_qc=samples_pass_qc,
        )

        log_case_result(result)
        return result

    @staticmethod
    def _samples_pass_qc(samples_quality_results: SamplesQualityResults) -> bool:
        fraction_failed_samples: float = (
            samples_quality_results.failed_samples_count
            / samples_quality_results.total_samples_count
        )
        return fraction_failed_samples < MutantQC.FRACTION_OF_SAMPLES_WITH_FAILED_QC_TRESHOLD

    def _get_internal_negative_control_id_for_case(self, case: Case) -> str:
        """Query lims to retrive internal_negative_control_id for a mutant case sequenced in one pool."""

        sample_internal_id = case.sample_ids[0]
        internal_negative_control_id: (
            str
        ) = self.lims.get_internal_negative_control_id_from_sample_in_pool(
            sample_internal_id=sample_internal_id, pooling_step=LimsProcess.COVID_POOLING_STEP
        )
        return internal_negative_control_id

    def _get_internal_negative_control_sample_for_case(
        self,
        case: Case,
    ) -> Sample:
        internal_negative_control_id: str = self._get_internal_negative_control_id_for_case(
            case=case
        )
        return self.status_db.get_sample_by_internal_id(internal_id=internal_negative_control_id)

    def _get_mutant_pool_samples(self, case: Case) -> MutantPoolSamples:
        samples = []
        external_negative_control = None

        for sample in case.samples:
            if sample.is_negative_control:
                external_negative_control = sample
                continue
            samples.append(sample)

        if not external_negative_control:
            raise CgError(f"No external negative control sample found for case {case.internal_id}.")

        internal_negative_control: Sample = self._get_internal_negative_control_sample_for_case(
            case=case
        )

        return MutantPoolSamples(
            samples=samples,
            external_negative_control=external_negative_control,
            internal_negative_control=internal_negative_control,
        )

    def _get_sample_pool_and_results(
        self, case_results_file_path: Path, case: Case
    ) -> SamplePoolAndResults:
        try:
            samples: MutantPoolSamples = self._get_mutant_pool_samples(case=case)
        except Exception as exception_object:
            raise CgError(
                f"Not possible to retrieve samples for case {case.internal_id}: {exception_object}"
            ) from exception_object

        try:
            samples_results: dict[str, ParsedSampleResults] = parse_samples_results(
                case=case, results_file_path=case_results_file_path
            )
        except Exception as exception_object:
            raise CgError(
                f"Not possible to retrieve results for case {case.internal_id}: {exception_object}"
            )

        return SamplePoolAndResults(pool=samples, results=samples_results)

1	from pathlib import Path	1✔
2	from cg.apps.lims.api import LimsAPI	1✔
3	from cg.constants.constants import MutantQC	1✔
4	from cg.constants.lims import LimsProcess	1✔
5	from cg.exc import CgError	1✔
6	from cg.meta.workflow.mutant.quality_controller.metrics_parser_utils import parse_samples_results	1✔
7	from cg.meta.workflow.mutant.quality_controller.models import (	1✔
8	MutantPoolSamples,
9	SamplePoolAndResults,
10	SampleQualityResults,
11	CaseQualityResult,
12	MutantQualityResult,
13	ParsedSampleResults,
14	SamplesQualityResults,
15	)
16	from cg.meta.workflow.mutant.quality_controller.report_generator_utils import (	1✔
17	get_summary,
18	write_report,
19	)
20	from cg.meta.workflow.mutant.quality_controller.result_logger_utils import (	1✔
21	log_case_result,
22	log_results,
23	log_sample_result,
24	)
25	from cg.meta.workflow.mutant.quality_controller.utils import (	1✔
26	has_external_negative_control_sample_valid_total_reads,
27	has_internal_negative_control_sample_valid_total_reads,
28	has_sample_valid_total_reads,
29	)
30	from cg.store.models import Case, Sample	1✔
31	from cg.store.store import Store	1✔
32
33
34	class MutantQualityController:	1✔
35	def __init__(self, status_db: Store, lims: LimsAPI) -> None:	1✔
36	self.status_db: Store = status_db	1✔
37	self.lims: LimsAPI = lims	1✔
38
39	def get_quality_control_result(	1✔
40	self, case: Case, case_results_file_path: Path, case_qc_report_path: Path
41	) -> MutantQualityResult:
42	"""Perform QC check on a case and generate the QC_report."""
43	sample_pool_and_results: SamplePoolAndResults = self._get_sample_pool_and_results(	1✔
44	case_results_file_path=case_results_file_path,
45	case=case,
46	)
47
48	samples_quality_results: SamplesQualityResults = self._get_samples_quality_results(	1✔
49	sample_pool_and_results=sample_pool_and_results
50	)
51	case_quality_result: CaseQualityResult = self._get_case_quality_result(	1✔
52	samples_quality_results
53	)
54
55	write_report(	1✔
56	case_qc_report_path=case_qc_report_path,
57	samples_quality_results=samples_quality_results,
58	case_quality_result=case_quality_result,
59	)
60
61	log_results(	1✔
62	case_quality_result=case_quality_result,
63	samples_quality_results=samples_quality_results,
64	report_file_path=case_qc_report_path,
65	)
66
67	summary: str = get_summary(	1✔
68	case_quality_result=case_quality_result,
69	samples_quality_results=samples_quality_results,
70	)
71
72	return MutantQualityResult(	1✔
73	case_quality_result=case_quality_result,
74	samples_quality_results=samples_quality_results,
75	summary=summary,
76	)
77
78	def _get_samples_quality_results(	1✔
79	self, sample_pool_and_results: SamplePoolAndResults
80	) -> SamplesQualityResults:
81	samples_quality_results: list[SampleQualityResults] = []	1✔
82	for sample in sample_pool_and_results.pool.samples:	1✔
83	sample_results: ParsedSampleResults = sample_pool_and_results.results[	1✔
84	sample.internal_id
85	]
86	sample_quality_results: (	1✔
87	SampleQualityResults
88	) = self._get_sample_quality_result_for_sample(
89	sample=sample, sample_results=sample_results
90	)
91	samples_quality_results.append(sample_quality_results)	1✔
92
93	internal_negative_control_sample: (	1✔
94	Sample
95	) = sample_pool_and_results.pool.internal_negative_control
96	internal_negative_control_quality_metrics: (	1✔
97	SampleQualityResults
98	) = self._get_sample_quality_result_for_internal_negative_control_sample(
99	sample=internal_negative_control_sample
100	)
101
102	external_negative_control_sample: (	1✔
103	Sample
104	) = sample_pool_and_results.pool.external_negative_control
105	external_negative_control_sample_results: (	1✔
106	ParsedSampleResults
107	) = sample_pool_and_results.results[external_negative_control_sample.internal_id]
108	external_negative_control_quality_metrics: (	1✔
109	SampleQualityResults
110	) = self._get_sample_quality_result_for_external_negative_control_sample(
111	sample=external_negative_control_sample,
112	sample_results=external_negative_control_sample_results,
113	)
114
115	return SamplesQualityResults(	1✔
116	samples=samples_quality_results,
117	internal_negative_control=internal_negative_control_quality_metrics,
118	external_negative_control=external_negative_control_quality_metrics,
119	)
120
121	@staticmethod	1✔
122	def _get_sample_quality_result_for_sample(	1✔
123	sample: Sample, sample_results: ParsedSampleResults
124	) -> SampleQualityResults:
125	does_sample_pass_reads_threshold: bool = has_sample_valid_total_reads(sample=sample)	1✔
126	does_sample_pass_qc: bool = does_sample_pass_reads_threshold and sample_results.passes_qc	1✔
127	sample_quality_result = SampleQualityResults(	1✔
128	sample_id=sample.internal_id,
129	passes_qc=does_sample_pass_qc,
130	passes_reads_threshold=does_sample_pass_reads_threshold,
131	passes_mutant_qc=sample_results.passes_qc,
132	)
133
134	log_sample_result(	1✔
135	result=sample_quality_result,
136	)
137	return sample_quality_result	1✔
138
139	@staticmethod	1✔
140	def _get_sample_quality_result_for_internal_negative_control_sample(	1✔
141	sample: Sample,
142	) -> SampleQualityResults:
143	does_sample_pass_reads_threshold: (	1✔
144	bool
145	) = has_internal_negative_control_sample_valid_total_reads(sample=sample)
146	sample_quality_result = SampleQualityResults(	1✔
147	sample_id=sample.internal_id,
148	passes_qc=does_sample_pass_reads_threshold,
149	passes_reads_threshold=does_sample_pass_reads_threshold,
150	)
151
152	log_sample_result(result=sample_quality_result, is_external_negative_control=True)	1✔
153	return sample_quality_result	1✔
154
155	@staticmethod	1✔
156	def _get_sample_quality_result_for_external_negative_control_sample(	1✔
157	sample: Sample, sample_results: ParsedSampleResults
158	) -> SampleQualityResults:
159	does_sample_pass_reads_threshold: (	1✔
160	bool
161	) = has_external_negative_control_sample_valid_total_reads(sample=sample)
162	sample_passes_qc: bool = does_sample_pass_reads_threshold and not sample_results.passes_qc	1✔
163	sample_quality_result = SampleQualityResults(	1✔
164	sample_id=sample.internal_id,
165	passes_qc=sample_passes_qc,
166	passes_reads_threshold=does_sample_pass_reads_threshold,
167	passes_mutant_qc=sample_results.passes_qc,
168	)
169
170	log_sample_result(result=sample_quality_result, is_external_negative_control=True)	1✔
171	return sample_quality_result	1✔
172
173	def _get_case_quality_result(	1✔
174	self, samples_quality_results: SamplesQualityResults
175	) -> CaseQualityResult:
176	external_negative_control_pass_qc: (	1✔
177	bool
178	) = samples_quality_results.external_negative_control.passes_qc
179	internal_negative_control_pass_qc: (	1✔
180	bool
181	) = samples_quality_results.internal_negative_control.passes_qc
182
183	samples_pass_qc: bool = self._samples_pass_qc(	1✔
184	samples_quality_results=samples_quality_results
185	)
186
187	case_passes_qc: bool = (	1✔
188	samples_pass_qc
189	and internal_negative_control_pass_qc
190	and external_negative_control_pass_qc
191	)
192
193	result = CaseQualityResult(	1✔
194	passes_qc=case_passes_qc,
195	internal_negative_control_passes_qc=internal_negative_control_pass_qc,
196	external_negative_control_passes_qc=external_negative_control_pass_qc,
197	fraction_samples_passes_qc=samples_pass_qc,
198	)
199
200	log_case_result(result)	1✔
201	return result	1✔
202
203	@staticmethod	1✔
204	def _samples_pass_qc(samples_quality_results: SamplesQualityResults) -> bool:	1✔
205	fraction_failed_samples: float = (	1✔
206	samples_quality_results.failed_samples_count
207	/ samples_quality_results.total_samples_count
208	)
209	return fraction_failed_samples < MutantQC.FRACTION_OF_SAMPLES_WITH_FAILED_QC_TRESHOLD	1✔
210
211	def _get_internal_negative_control_id_for_case(self, case: Case) -> str:	1✔
212	"""Query lims to retrive internal_negative_control_id for a mutant case sequenced in one pool."""
213
214	sample_internal_id = case.sample_ids[0]	1✔
215	internal_negative_control_id: (	1✔
216	str
217	) = self.lims.get_internal_negative_control_id_from_sample_in_pool(
218	sample_internal_id=sample_internal_id, pooling_step=LimsProcess.COVID_POOLING_STEP
219	)
220	return internal_negative_control_id	1✔
221
222	def _get_internal_negative_control_sample_for_case(	1✔
223	self,
224	case: Case,
225	) -> Sample:
226	internal_negative_control_id: str = self._get_internal_negative_control_id_for_case(	1✔
227	case=case
228	)
229	return self.status_db.get_sample_by_internal_id(internal_id=internal_negative_control_id)	1✔
230
231	def _get_mutant_pool_samples(self, case: Case) -> MutantPoolSamples:	1✔
232	samples = []	1✔
233	external_negative_control = None	1✔
234
235	for sample in case.samples:	1✔
236	if sample.is_negative_control:	1✔
237	external_negative_control = sample	1✔
238	continue	1✔
239	samples.append(sample)	1✔
240
241	if not external_negative_control:	1✔
NEW 242	raise CgError(f"No external negative control sample found for case {case.internal_id}.")	×
243
244	internal_negative_control: Sample = self._get_internal_negative_control_sample_for_case(	1✔
245	case=case
246	)
247
248	return MutantPoolSamples(	1✔
249	samples=samples,
250	external_negative_control=external_negative_control,
251	internal_negative_control=internal_negative_control,
252	)
253
254	def _get_sample_pool_and_results(	1✔
255	self, case_results_file_path: Path, case: Case
256	) -> SamplePoolAndResults:
257	try:	1✔
258	samples: MutantPoolSamples = self._get_mutant_pool_samples(case=case)	1✔
NEW 259	except Exception as exception_object:	×
NEW 260	raise CgError(	×
261	f"Not possible to retrieve samples for case {case.internal_id}: {exception_object}"
262	) from exception_object
263
264	try:	1✔
265	samples_results: dict[str, ParsedSampleResults] = parse_samples_results(	1✔
266	case=case, results_file_path=case_results_file_path
267	)
NEW 268	except Exception as exception_object:	×
NEW 269	raise CgError(	×
270	f"Not possible to retrieve results for case {case.internal_id}: {exception_object}"
271	)
272
273	return SamplePoolAndResults(pool=samples, results=samples_results)	1✔

Clinical-Genomics / cg / 10491698847

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