8920038569

Committed 02 May 2024 07:13AM UTC coverage: 84.527%. First build

Build # 8920038569

Build Type

Pull #3148

github

Committed by

web-flow

Commit Message

Merge a6d83dc9b into 73c972545

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

Run Details

174 of 193 new or added lines in 7 files covered. (90.16%)

19797 of 23421 relevant lines covered (84.53%)

0.85 hits per line

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

96.43

/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, Customer
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)
        customer_id: int = self.status_db.get_customer_by_internal_id(
            CustomerId.CG_INTERNAL_CUSTOMER.value
        ).id
        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_id=customer_id,
            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)
        customer: Customer = self.status_db.get_customer_by_internal_id(
            CustomerId.CG_INTERNAL_CUSTOMER
        )
        validation_case.customer = 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, Customer	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	customer_id: int = self.status_db.get_customer_by_internal_id(	1✔
70	CustomerId.CG_INTERNAL_CUSTOMER.value
71	).id
72	validation_sample: Sample = self.status_db.add_sample(	1✔
73	name=sample_id,
74	internal_id=sample_id,
75	sex=original_sample.sex,
76	order=original_sample.order,
77	from_sample=original_sample.internal_id,
78	tumour=original_sample.is_tumour,
79	priority=Priority.standard,
80	customer_id=customer_id,
81	application_version=application_version,
82	received=original_sample.received_at,
83	prepared_at=original_sample.prepared_at,
84	last_sequenced_at=original_sample.last_sequenced_at,
85	)
86	return validation_sample	1✔
87
88	def get_validation_samples(self):	1✔
89	return [self._generate_validation_samples(sample) for sample in self.original_samples]	1✔
90
91	def _generate_validation_case(	1✔
92	self,
93	) -> Case:
94	"""
95	Generate a case for the validation samples.
96	The new case uses existing case data.
97	Customer will be set to cust000.
98	"""
99	if self.status_db.case_with_name_exists(case_name=self.validation_case_name):	1✔
NEW 100	raise ValueError(f"Case with name {self.case_name} already exists.")	×
101	validation_case: Case = self.status_db.add_case(	1✔
102	data_analysis=(self.input_data.data_analysis or self.original_case.data_analysis),
103	data_delivery=(self.input_data.delivery or self.original_case.data_delivery),
104	name=self.validation_case_name,
105	panels=self.original_case.panels,
106	priority=Priority.standard,
107	ticket=self.original_case.latest_ticket,
108	)
109	validation_case.orders.append(self.original_case.latest_order)	1✔
110	customer: Customer = self.status_db.get_customer_by_internal_id(	1✔
111	CustomerId.CG_INTERNAL_CUSTOMER
112	)
113	validation_case.customer = customer	1✔
114	validation_case.is_compressible = False	1✔
115	validation_case.action = CaseActions.HOLD	1✔
116	return validation_case	1✔
117
118	@staticmethod	1✔
119	def _get_application_version(sample: Sample) -> ApplicationVersion:	1✔
120	"""Return the application version for a sample."""
121	return sample.application_version	1✔

Clinical-Genomics / cg / 8920038569

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