10603637037

Committed 28 Aug 2024 08:06PM UTC coverage: 84.63%. First build

Build # 10603637037

Build Type

Pull #3212

github

Committed by

web-flow

Commit Message

Merge 603e15d94 into a4566a28b

Pull Request Pull Request #3212: Add scout support for tomte

Run Details

105 of 107 new or added lines in 12 files covered. (98.13%)

21903 of 25881 relevant lines covered (84.63%)

0.86 hits per line

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

84.0

/cg/meta/report/tomte.py

"""Tomte delivery report API."""

from cg.constants import (
    REQUIRED_APPLICATION_FIELDS,
    REQUIRED_CASE_FIELDS,
    REQUIRED_CUSTOMER_FIELDS,
    REQUIRED_DATA_ANALYSIS_TOMTE_FIELDS,
    REQUIRED_REPORT_FIELDS,
    REQUIRED_SAMPLE_METADATA_TOMTE_FIELDS,
    REQUIRED_SAMPLE_METHODS_FIELDS,
    REQUIRED_SAMPLE_TIMESTAMP_FIELDS,
    REQUIRED_SAMPLE_TOMTE_FIELDS,
)
from cg.constants.scout import ScoutUploadKey
from cg.meta.report.field_validators import get_million_read_pairs
from cg.meta.report.report_api import ReportAPI
from cg.meta.workflow.tomte import TomteAnalysisAPI
from cg.models.analysis import AnalysisModel, NextflowAnalysis
from cg.models.cg_config import CGConfig
from cg.models.report.metadata import TomteSampleMetadataModel
from cg.models.report.report import CaseModel, ReportRequiredFields, ScoutReportFiles
from cg.models.report.sample import SampleModel
from cg.models.tomte.tomte import TomteQCMetrics
from cg.store.models import Case, Sample


class TomteReportAPI(ReportAPI):
    """API to create Tomte delivery reports."""

    def __init__(self, config: CGConfig, analysis_api: TomteAnalysisAPI):
        super().__init__(config=config, analysis_api=analysis_api)

    def get_sample_metadata(
        self, case: Case, sample: Sample, analysis_metadata: NextflowAnalysis
    ) -> TomteSampleMetadataModel:
        """Return Tomte sample metadata to include in the delivery report."""
        sample_metrics: TomteQCMetrics = analysis_metadata.sample_metrics[sample.internal_id]
        return TomteSampleMetadataModel(
            bias_5_3=sample_metrics.median_5prime_to_3prime_bias,
            duplicates=sample_metrics.pct_duplication,
            dv200=self.lims_api.get_sample_dv200(sample.internal_id),
            gc_content=sample_metrics.after_filtering_gc_content,
            initial_qc=self.lims_api.has_sample_passed_initial_qc(sample.internal_id),
            input_amount=self.lims_api.get_latest_rna_input_amount(sample.internal_id),
            mean_length_r1=sample_metrics.after_filtering_read1_mean_length,
            million_read_pairs=get_million_read_pairs(sample_metrics.before_filtering_total_reads),
            mrna_bases=sample_metrics.pct_mrna_bases,
            pct_adapter=sample_metrics.pct_adapter,
            pct_intergenic_bases=sample_metrics.pct_intergenic_bases,
            pct_intronic_bases=sample_metrics.pct_intronic_bases,
            pct_surviving=sample_metrics.pct_surviving,
            q20_rate=sample_metrics.after_filtering_q20_rate,
            q30_rate=sample_metrics.after_filtering_q30_rate,
            ribosomal_bases=sample_metrics.pct_ribosomal_bases,
            rin=self.lims_api.get_sample_rin(sample.internal_id),
            uniquely_mapped_reads=sample_metrics.uniquely_mapped_percent,
        )

    def is_report_accredited(
        self, samples: list[SampleModel], analysis_metadata: AnalysisModel
    ) -> bool:
        """Return whether the Tomte delivery report is accredited."""
        return False

    def get_required_fields(self, case: CaseModel) -> dict:
        """Return the delivery report required fields for Tomte."""
        report_required_fields = ReportRequiredFields(
            applications=self.get_application_required_fields(
                case=case, required_fields=REQUIRED_APPLICATION_FIELDS
            ),
            case=REQUIRED_CASE_FIELDS,
            customer=REQUIRED_CUSTOMER_FIELDS,
            data_analysis=REQUIRED_DATA_ANALYSIS_TOMTE_FIELDS,
            metadata=self.get_sample_required_fields(
                case=case, required_fields=REQUIRED_SAMPLE_METADATA_TOMTE_FIELDS
            ),
            methods=self.get_sample_required_fields(
                case=case, required_fields=REQUIRED_SAMPLE_METHODS_FIELDS
            ),
            report=REQUIRED_REPORT_FIELDS,
            samples=self.get_sample_required_fields(
                case=case, required_fields=REQUIRED_SAMPLE_TOMTE_FIELDS
            ),
            timestamps=self.get_timestamp_required_fields(
                case=case, required_fields=REQUIRED_SAMPLE_TIMESTAMP_FIELDS
            ),
        )
        return report_required_fields.model_dump()

    def get_scout_uploaded_files(self, case_id: str) -> ScoutReportFiles:
        """Return files that will be uploaded to Scout."""
        return ScoutReportFiles(
            snv_vcf=self.get_scout_uploaded_file_from_hk(
                case_id=case_id, scout_key=ScoutUploadKey.SNV_VCF
            ),
            snv_research_vcf=self.get_scout_uploaded_file_from_hk(
                case_id=case_id, scout_key=ScoutUploadKey.SNV_RESEARCH_VCF
            ),
            fraser_tsv=self.get_scout_uploaded_file_from_hk(
                case_id=case_id, scout_key=ScoutUploadKey.FRASER_TSV
            ),
            outrider_tsv=self.get_scout_uploaded_file_from_hk(
                case_id=case_id, scout_key=ScoutUploadKey.OUTRIDER_TSV
            ),
        )

1	"""Tomte delivery report API."""
2
3	from cg.constants import (	1✔
4	REQUIRED_APPLICATION_FIELDS,
5	REQUIRED_CASE_FIELDS,
6	REQUIRED_CUSTOMER_FIELDS,
7	REQUIRED_DATA_ANALYSIS_TOMTE_FIELDS,
8	REQUIRED_REPORT_FIELDS,
9	REQUIRED_SAMPLE_METADATA_TOMTE_FIELDS,
10	REQUIRED_SAMPLE_METHODS_FIELDS,
11	REQUIRED_SAMPLE_TIMESTAMP_FIELDS,
12	REQUIRED_SAMPLE_TOMTE_FIELDS,
13	)
14	from cg.constants.scout import ScoutUploadKey	1✔
15	from cg.meta.report.field_validators import get_million_read_pairs	1✔
16	from cg.meta.report.report_api import ReportAPI	1✔
17	from cg.meta.workflow.tomte import TomteAnalysisAPI	1✔
18	from cg.models.analysis import AnalysisModel, NextflowAnalysis	1✔
19	from cg.models.cg_config import CGConfig	1✔
20	from cg.models.report.metadata import TomteSampleMetadataModel	1✔
21	from cg.models.report.report import CaseModel, ReportRequiredFields, ScoutReportFiles	1✔
22	from cg.models.report.sample import SampleModel	1✔
23	from cg.models.tomte.tomte import TomteQCMetrics	1✔
24	from cg.store.models import Case, Sample	1✔
25
26
27	class TomteReportAPI(ReportAPI):	1✔
28	"""API to create Tomte delivery reports."""
29
30	def __init__(self, config: CGConfig, analysis_api: TomteAnalysisAPI):	1✔
31	super().__init__(config=config, analysis_api=analysis_api)	1✔
32
33	def get_sample_metadata(	1✔
34	self, case: Case, sample: Sample, analysis_metadata: NextflowAnalysis
35	) -> TomteSampleMetadataModel:
36	"""Return Tomte sample metadata to include in the delivery report."""
37	sample_metrics: TomteQCMetrics = analysis_metadata.sample_metrics[sample.internal_id]	1✔
38	return TomteSampleMetadataModel(	1✔
39	bias_5_3=sample_metrics.median_5prime_to_3prime_bias,
40	duplicates=sample_metrics.pct_duplication,
41	dv200=self.lims_api.get_sample_dv200(sample.internal_id),
42	gc_content=sample_metrics.after_filtering_gc_content,
43	initial_qc=self.lims_api.has_sample_passed_initial_qc(sample.internal_id),
44	input_amount=self.lims_api.get_latest_rna_input_amount(sample.internal_id),
45	mean_length_r1=sample_metrics.after_filtering_read1_mean_length,
46	million_read_pairs=get_million_read_pairs(sample_metrics.before_filtering_total_reads),
47	mrna_bases=sample_metrics.pct_mrna_bases,
48	pct_adapter=sample_metrics.pct_adapter,
49	pct_intergenic_bases=sample_metrics.pct_intergenic_bases,
50	pct_intronic_bases=sample_metrics.pct_intronic_bases,
51	pct_surviving=sample_metrics.pct_surviving,
52	q20_rate=sample_metrics.after_filtering_q20_rate,
53	q30_rate=sample_metrics.after_filtering_q30_rate,
54	ribosomal_bases=sample_metrics.pct_ribosomal_bases,
55	rin=self.lims_api.get_sample_rin(sample.internal_id),
56	uniquely_mapped_reads=sample_metrics.uniquely_mapped_percent,
57	)
58
59	def is_report_accredited(	1✔
60	self, samples: list[SampleModel], analysis_metadata: AnalysisModel
61	) -> bool:
62	"""Return whether the Tomte delivery report is accredited."""
63	return False	×
64
65	def get_required_fields(self, case: CaseModel) -> dict:	1✔
66	"""Return the delivery report required fields for Tomte."""
67	report_required_fields = ReportRequiredFields(	×
68	applications=self.get_application_required_fields(
69	case=case, required_fields=REQUIRED_APPLICATION_FIELDS
70	),
71	case=REQUIRED_CASE_FIELDS,
72	customer=REQUIRED_CUSTOMER_FIELDS,
73	data_analysis=REQUIRED_DATA_ANALYSIS_TOMTE_FIELDS,
74	metadata=self.get_sample_required_fields(
75	case=case, required_fields=REQUIRED_SAMPLE_METADATA_TOMTE_FIELDS
76	),
77	methods=self.get_sample_required_fields(
78	case=case, required_fields=REQUIRED_SAMPLE_METHODS_FIELDS
79	),
80	report=REQUIRED_REPORT_FIELDS,
81	samples=self.get_sample_required_fields(
82	case=case, required_fields=REQUIRED_SAMPLE_TOMTE_FIELDS
83	),
84	timestamps=self.get_timestamp_required_fields(
85	case=case, required_fields=REQUIRED_SAMPLE_TIMESTAMP_FIELDS
86	),
87	)
88	return report_required_fields.model_dump()	×
89
90	def get_scout_uploaded_files(self, case_id: str) -> ScoutReportFiles:	1✔
91	"""Return files that will be uploaded to Scout."""
NEW 92	return ScoutReportFiles(	×
93	snv_vcf=self.get_scout_uploaded_file_from_hk(
94	case_id=case_id, scout_key=ScoutUploadKey.SNV_VCF
95	),
96	snv_research_vcf=self.get_scout_uploaded_file_from_hk(
97	case_id=case_id, scout_key=ScoutUploadKey.SNV_RESEARCH_VCF
98	),
99	fraser_tsv=self.get_scout_uploaded_file_from_hk(
100	case_id=case_id, scout_key=ScoutUploadKey.FRASER_TSV
101	),
102	outrider_tsv=self.get_scout_uploaded_file_from_hk(
103	case_id=case_id, scout_key=ScoutUploadKey.OUTRIDER_TSV
104	),
105	)

Clinical-Genomics / cg / 10603637037

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