8780670433

Committed 22 Apr 2024 08:09AM UTC coverage: 84.096%. First build

Build # 8780670433

Build Type

Pull #3148

github

Committed by

web-flow

Commit Message

Merge 4e5f945f0 into f3a674eb2

Pull Request Pull Request #3148: add(create validation case)

Run Details

59 of 180 new or added lines in 5 files covered. (32.78%)

19548 of 23245 relevant lines covered (84.1%)

0.84 hits per line

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

96.3

/cg/meta/create_validation_cases/validation_case_data.py

"""Model for down sampling meta data."""

import logging
from pathlib import Path

from cg.apps.housekeeper.hk import HousekeeperAPI
from cg.constants import DataDelivery, Priority
from cg.constants.constants import CaseActions, CustomerId
from cg.meta.create_validation_cases.validation_data_input import ValidationDataInput
from cg.store.models import ApplicationVersion, Case, Sample
from cg.store.store import Store
from cg.utils.calculations import multiply_by_million

LOG = logging.getLogger(__name__)


class ValidationCaseData:
    def __init__(self, status_db: Store, validation_data_input: ValidationDataInput):
        """Initialize the validation sample data and perform integrity checks."""
        self.status_db: Store = status_db
        self.input_data: ValidationDataInput = validation_data_input
        self.case_id: str = validation_data_input.case_id
        self.case_name: str = validation_data_input.case_name
        self.original_case: Case = self.get_case_to_copy()
        self.original_samples: list[Sample] = self.get_samples_to_copy()
        self.validation_samples: list[Sample] = self.get_validation_samples()
        self.validation_case: Case = self._generate_validation_case()
        LOG.info(f"Validation Data checks completed for {self.case_id}")

    @staticmethod
    def _validation_sample_id(sample: Sample) -> str:
        """Return a new validation sample identifier. The identifier removes "ACC" and prepends "VAL"."""
        return "VAL" + sample.internal_id[3:]

    @property
    def validation_case_name(
        self,
    ) -> str:
        """Return a case name with _validation appended."""
        return f"{self.case_name}_validation"

    def get_samples_to_copy(self) -> list[Sample]:
        """
        Return samples associated to a case.
        """
        case: Case = self.status_db.get_case_by_internal_id(self.case_id)
        return [sample for sample in case.samples]

    def get_case_to_copy(self) -> Case:
        """
        Check if a case exists in StatusDB.
            Raises: ValueError
        """
        case: Case = self.status_db.get_case_by_internal_id(self.case_id)
        if not case:
            raise ValueError(f"Case {self.case_id} not found in StatusDB.")
        return case

    def _generate_validation_samples(
        self,
        original_sample: Sample,
    ) -> Sample:
        """
        Generate a validation sample record for StatusDB.
        The new sample contains the original sample internal id and meta data
        """
        application_version: ApplicationVersion = self._get_application_version(original_sample)
        sample_id: str = self._validation_sample_id(original_sample)
        validation_sample: Sample = self.status_db.add_sample(
            name=sample_id,
            internal_id=sample_id,
            sex=original_sample.sex,
            order=original_sample.order,
            from_sample=original_sample.internal_id,
            tumour=original_sample.is_tumour,
            priority=Priority.standard,
            customer=CustomerId.CG_INTERNAL_CUSTOMER,
            application_version=application_version,
            received=original_sample.received_at,
            prepared_at=original_sample.prepared_at,
            last_sequenced_at=original_sample.last_sequenced_at,
        )
        return validation_sample

    def get_validation_samples(self):
        return [self._generate_validation_samples(sample) for sample in self.original_samples]

    def _generate_validation_case(
        self,
    ) -> Case:
        """
        Generate a case for the validation samples.
        The new case uses existing case data.
        Customer will be set to cust000.
        """
        if self.status_db.case_with_name_exists(case_name=self.validation_case_name):
            raise ValueError(f"Case with name {self.case_name} already exists.")
        validation_case: Case = self.status_db.add_case(
            data_analysis=(self.input_data.data_analysis or self.original_case.data_analysis),
            data_delivery=(self.input_data.delivery or self.original_case.data_delivery),
            name=self.validation_case_name,
            panels=self.original_case.panels,
            priority=Priority.standard,
            ticket=self.original_case.latest_ticket,
        )
        validation_case.orders.append(self.original_case.latest_order)
        validation_case.customer = CustomerId.CG_INTERNAL_CUSTOMER
        validation_case.is_compressible = False
        validation_case.action = CaseActions.HOLD
        return validation_case

    @staticmethod
    def _get_application_version(sample: Sample) -> ApplicationVersion:
        """Return the application version for a sample."""
        return sample.application_version

1	"""Model for down sampling meta data."""
2
3	import logging	1✔
4	from pathlib import Path	1✔
5
6	from cg.apps.housekeeper.hk import HousekeeperAPI	1✔
7	from cg.constants import DataDelivery, Priority	1✔
8	from cg.constants.constants import CaseActions, CustomerId	1✔
9	from cg.meta.create_validation_cases.validation_data_input import ValidationDataInput	1✔
10	from cg.store.models import ApplicationVersion, Case, Sample	1✔
11	from cg.store.store import Store	1✔
12	from cg.utils.calculations import multiply_by_million	1✔
13
14	LOG = logging.getLogger(__name__)	1✔
15
16
17	class ValidationCaseData:	1✔
18	def __init__(self, status_db: Store, validation_data_input: ValidationDataInput):	1✔
19	"""Initialize the validation sample data and perform integrity checks."""
20	self.status_db: Store = status_db	1✔
21	self.input_data: ValidationDataInput = validation_data_input	1✔
22	self.case_id: str = validation_data_input.case_id	1✔
23	self.case_name: str = validation_data_input.case_name	1✔
24	self.original_case: Case = self.get_case_to_copy()	1✔
25	self.original_samples: list[Sample] = self.get_samples_to_copy()	1✔
26	self.validation_samples: list[Sample] = self.get_validation_samples()	1✔
27	self.validation_case: Case = self._generate_validation_case()	1✔
28	LOG.info(f"Validation Data checks completed for {self.case_id}")	1✔
29
30	@staticmethod	1✔
31	def _validation_sample_id(sample: Sample) -> str:	1✔
32	"""Return a new validation sample identifier. The identifier removes "ACC" and prepends "VAL"."""
33	return "VAL" + sample.internal_id[3:]	1✔
34
35	@property	1✔
36	def validation_case_name(	1✔
37	self,
38	) -> str:
39	"""Return a case name with _validation appended."""
40	return f"{self.case_name}_validation"	1✔
41
42	def get_samples_to_copy(self) -> list[Sample]:	1✔
43	"""
44	Return samples associated to a case.
45	"""
46	case: Case = self.status_db.get_case_by_internal_id(self.case_id)	1✔
47	return [sample for sample in case.samples]	1✔
48
49	def get_case_to_copy(self) -> Case:	1✔
50	"""
51	Check if a case exists in StatusDB.
52	Raises: ValueError
53	"""
54	case: Case = self.status_db.get_case_by_internal_id(self.case_id)	1✔
55	if not case:	1✔
NEW 56	raise ValueError(f"Case {self.case_id} not found in StatusDB.")	×
57	return case	1✔
58
59	def _generate_validation_samples(	1✔
60	self,
61	original_sample: Sample,
62	) -> Sample:
63	"""
64	Generate a validation sample record for StatusDB.
65	The new sample contains the original sample internal id and meta data
66	"""
67	application_version: ApplicationVersion = self._get_application_version(original_sample)	1✔
68	sample_id: str = self._validation_sample_id(original_sample)	1✔
69	validation_sample: Sample = self.status_db.add_sample(	1✔
70	name=sample_id,
71	internal_id=sample_id,
72	sex=original_sample.sex,
73	order=original_sample.order,
74	from_sample=original_sample.internal_id,
75	tumour=original_sample.is_tumour,
76	priority=Priority.standard,
77	customer=CustomerId.CG_INTERNAL_CUSTOMER,
78	application_version=application_version,
79	received=original_sample.received_at,
80	prepared_at=original_sample.prepared_at,
81	last_sequenced_at=original_sample.last_sequenced_at,
82	)
83	return validation_sample	1✔
84
85	def get_validation_samples(self):	1✔
86	return [self._generate_validation_samples(sample) for sample in self.original_samples]	1✔
87
88	def _generate_validation_case(	1✔
89	self,
90	) -> Case:
91	"""
92	Generate a case for the validation samples.
93	The new case uses existing case data.
94	Customer will be set to cust000.
95	"""
96	if self.status_db.case_with_name_exists(case_name=self.validation_case_name):	1✔
NEW 97	raise ValueError(f"Case with name {self.case_name} already exists.")	×
98	validation_case: Case = self.status_db.add_case(	1✔
99	data_analysis=(self.input_data.data_analysis or self.original_case.data_analysis),
100	data_delivery=(self.input_data.delivery or self.original_case.data_delivery),
101	name=self.validation_case_name,
102	panels=self.original_case.panels,
103	priority=Priority.standard,
104	ticket=self.original_case.latest_ticket,
105	)
106	validation_case.orders.append(self.original_case.latest_order)	1✔
107	validation_case.customer = CustomerId.CG_INTERNAL_CUSTOMER	1✔
108	validation_case.is_compressible = False	1✔
109	validation_case.action = CaseActions.HOLD	1✔
110	return validation_case	1✔
111
112	@staticmethod	1✔
113	def _get_application_version(sample: Sample) -> ApplicationVersion:	1✔
114	"""Return the application version for a sample."""
115	return sample.application_version	1✔

Clinical-Genomics / cg / 8780670433

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