cirro.sdk.developer

  1from io import StringIO
  2
  3from cirro_api_client.v1.api.datasets import get_sample_sheets, ingest_samples
  4from cirro_api_client.v1.api.processes import validate_file_name_patterns
  5from cirro_api_client.v1.models import SampleSheets, ValidateFileNamePatternsRequest, FileNameMatch
  6
  7from cirro.cirro_client import CirroApi
  8from cirro.helpers import PreprocessDataset
  9
 10
 11class Matches(list[FileNameMatch]):
 12    def print(self):
 13        """
 14        Prints the file name validation matches in a readable format.
 15        """
 16        print(f'Matches: {len(self)}')
 17        print()
 18        for match in self:
 19            print(f'{match.file_name}')
 20            print(f'Sample name: {match.sample_name}')
 21            print(f'Matched regex: {match.regex_pattern_match}')
 22            print()
 23
 24
 25class DeveloperHelper:
 26    """
 27    Helper class for developer-related tasks,
 28    such as adding samplesheet preprocessing for a pipeline
 29    or testing file name validation and sample autopopulation.
 30    """
 31
 32    def __init__(self, client: CirroApi):
 33        self.client = client
 34
 35    def generate_preprocess_for_input_datasets(self,
 36                                               project_id: str,
 37                                               input_dataset_ids: list[str],
 38                                               params=None) -> PreprocessDataset:
 39        """
 40        Generates a PreprocessDataset object for the given datasets
 41
 42        With optional parameters to pass into the preprocess script.
 43        `metadata` is not available in this context, so it is mocked.
 44        """
 45        samplesheets = self._generate_samplesheets_for_datasets(project_id, input_dataset_ids)
 46        return PreprocessDataset(
 47            samplesheet=samplesheets.samples,
 48            files=samplesheets.files,
 49            params=params or {},
 50            # Mock metadata
 51            metadata={
 52                'dataset': {},
 53                'project': {},
 54                'inputs': [],
 55                'process': {}
 56            }
 57        )
 58
 59    def test_file_name_validation_for_dataset(self,
 60                                              project_id: str,
 61                                              dataset_id: str,
 62                                              file_name_patterns: list[str]) -> Matches:
 63        """
 64        Tests the file name validation for a given dataset against specified regex patterns.
 65
 66        Used when configuring Cirro's sample autopopulation feature.
 67        More info: https://docs.cirro.bio/features/samples/#using-auto-population
 68        """
 69        dataset_files = self.client.datasets.get_assets_listing(project_id=project_id, dataset_id=dataset_id).files
 70        file_names = [file.relative_path for file in dataset_files]
 71        return self.test_file_name_validation(file_names, file_name_patterns)
 72
 73    def test_file_name_validation(self,
 74                                  file_names: list[str],
 75                                  file_name_patterns: list[str]) -> Matches:
 76        """
 77        Tests the file name validation for a list of file names against specified regex patterns.
 78        """
 79        request_body = ValidateFileNamePatternsRequest(
 80            file_names=file_names,
 81            file_name_patterns=file_name_patterns
 82        )
 83
 84        matches = validate_file_name_patterns.sync(
 85            process_id="test",
 86            body=request_body,
 87            client=self.client.api_client
 88        )
 89        return Matches(matches)
 90
 91    def generate_samplesheets_for_dataset(self, project_id: str, dataset_id: str) -> SampleSheets:
 92        """
 93        Generates Cirro samplesheets for a given dataset
 94        """
 95        return get_sample_sheets.sync(
 96            project_id=project_id,
 97            dataset_id=dataset_id,
 98            client=self.client.api_client
 99        )
100
101    def rerun_sample_ingest_for_dataset(self, project_id: str, dataset_id: str):
102        """
103        Reruns the sample ingest process for a given dataset.
104        You'll want to do this if you have updated the file name patterns in your pipeline (or data type)
105        """
106        ingest_samples.sync_detailed(
107            project_id=project_id,
108            dataset_id=dataset_id,
109            client=self.client.api_client
110        )
111
112    def _generate_samplesheets_for_datasets(self, project_id: str, dataset_ids: list[str]) -> SampleSheets:
113        """
114        Generates Cirro samplesheets for multiple datasets in a project.
115        """
116        # Concatenate samplesheets using pandas
117        import pandas
118        samplesheets_dfs = []
119        files_dfs = []
120        for dataset_id in dataset_ids:
121            samplesheet = self.generate_samplesheets_for_dataset(project_id, dataset_id)
122            samplesheets_dfs.append(pandas.read_csv(StringIO(samplesheet.samples)))
123            files_dfs.append(pandas.read_csv(StringIO(samplesheet.files)))
124
125        samplesheets_df = pandas.concat(samplesheets_dfs, ignore_index=True)
126        files_df = pandas.concat(files_dfs, ignore_index=True)
127        return SampleSheets(
128            samples=samplesheets_df.to_csv(index=False),
129            files=files_df.to_csv(index=False)
130        )