cirro.sdk.process

  1from typing import List, Union
  2
  3from cirro_api_client.v1.models import Process, Executor, ProcessDetail, CustomPipelineSettings, PipelineCode, \
  4    RunAnalysisRequest, RunAnalysisRequestParams
  5
  6from cirro.cirro_client import CirroApi
  7from cirro.models.form_specification import ParameterSpecification
  8from cirro.sdk.asset import DataPortalAssets, DataPortalAsset
  9from cirro.sdk.exceptions import DataPortalInputError
 10
 11
 12class DataPortalProcess(DataPortalAsset):
 13    """Helper functions for interacting with analysis processes."""
 14
 15    def __init__(self, process: Union[Process, ProcessDetail], client: CirroApi):
 16        """
 17        Instantiate with helper method
 18
 19        ```python
 20        from cirro import DataPortal()
 21        portal = DataPortal()
 22        process = portal.get_process_by_name("Process Name")
 23        ```
 24        """
 25        self._data = process
 26        self._client = client
 27
 28    @property
 29    def id(self) -> str:
 30        """Unique identifier"""
 31        return self._data.id
 32
 33    @property
 34    def name(self) -> str:
 35        """Readable name"""
 36        return self._data.name
 37
 38    @property
 39    def description(self) -> str:
 40        """Longer description of process"""
 41        return self._data.description
 42
 43    @property
 44    def child_process_ids(self) -> List[str]:
 45        """List of processes which can be run on the output of this process"""
 46        return self._data.child_process_ids
 47
 48    @property
 49    def executor(self) -> Executor:
 50        """INGEST, CROMWELL, or NEXTFLOW"""
 51        return self._data.executor
 52
 53    @property
 54    def category(self) -> str:
 55        """Category of process"""
 56        return self._data.category
 57
 58    @property
 59    def pipeline_type(self) -> str:
 60        """Pipeline type"""
 61        return self._data.pipeline_type
 62
 63    @property
 64    def documentation_url(self) -> str:
 65        """Documentation URL"""
 66        return self._data.documentation_url
 67
 68    @property
 69    def file_requirements_message(self) -> str:
 70        """Description of files required for INGEST processes"""
 71        return self._data.file_requirements_message
 72
 73    @property
 74    def code(self) -> PipelineCode:
 75        """Pipeline code configuration"""
 76        return self._get_detail().pipeline_code
 77
 78    @property
 79    def custom_settings(self) -> CustomPipelineSettings:
 80        """Custom settings for the process"""
 81        return self._get_detail().custom_settings
 82
 83    def _get_detail(self) -> ProcessDetail:
 84        if not isinstance(self._data, ProcessDetail):
 85            self._data = self._client.processes.get(self.id)
 86        return self._data
 87
 88    def __str__(self):
 89        return '\n'.join([
 90            f"{i.title()}: {self.__getattribute__(i)}"
 91            for i in ['name', 'id', 'description']
 92        ])
 93
 94    def get_parameter_spec(self) -> ParameterSpecification:
 95        """
 96        Gets a specification used to describe the parameters used in the process.
 97        """
 98        return self._client.processes.get_parameter_spec(self.id)
 99
100    def run_analysis(
101            self,
102            name: str = None,
103            project_id: str = None,
104            datasets: list = None,
105            description: str = "",
106            params=None,
107            notifications_emails: List[str] = None,
108            compute_environment: str = None,
109            resume_dataset_id: str = None,
110            source_sample_ids: List[str] = None
111    ) -> str:
112        """
113        Runs this process on one or more input datasets, returns the ID of the newly created dataset.
114
115        Args:
116            name (str): Name of newly created dataset
117            project_id (str): ID of the project to run the analysis in
118            datasets (List[DataPortalDataset or str]): One or more input datasets
119             (as DataPortalDataset objects or dataset ID strings)
120            description (str): Description of newly created dataset
121            params (dict): Analysis parameters
122            notifications_emails (List[str]): Notification email address(es)
123            compute_environment (str): Name or ID of compute environment to use,
124             if blank it will run in AWS
125            resume_dataset_id (str): ID of dataset to resume from, used for caching task execution.
126             It will attempt to re-use the previous output to minimize duplicate work.
127            Note that Nextflow does not require this parameter, as it will automatically resume
128             from any previous attempts using a global cache.
129            source_sample_ids (List[str]): List of sample IDs to use as input for the analysis.
130
131        Returns:
132            dataset_id (str): ID of newly created dataset
133        """
134        if name is None:
135            raise DataPortalInputError("Must specify 'name' for run_analysis")
136        if project_id is None:
137            raise DataPortalInputError("Must specify 'project_id' for run_analysis")
138        if not datasets:
139            raise DataPortalInputError("Must specify 'datasets' for run_analysis")
140        if notifications_emails is None:
141            notifications_emails = []
142        if params is None:
143            params = {}
144
145        # Accept DataPortalDataset objects or raw ID strings
146        source_dataset_ids = [
147            ds if isinstance(ds, str) else ds.id
148            for ds in datasets
149        ]
150
151        if compute_environment:
152            compute_environment_name = compute_environment
153            compute_environments = self._client.compute_environments.list_environments_for_project(
154                project_id=project_id
155            )
156            compute_environment = next(
157                (env for env in compute_environments
158                 if env.name == compute_environment or env.id == compute_environment),
159                None
160            )
161            if compute_environment is None:
162                raise DataPortalInputError(f"Compute environment '{compute_environment_name}' not found")
163
164        resp = self._client.execution.run_analysis(
165            project_id=project_id,
166            request=RunAnalysisRequest(
167                name=name,
168                description=description,
169                process_id=self.id,
170                source_dataset_ids=source_dataset_ids,
171                params=RunAnalysisRequestParams.from_dict(params),
172                notification_emails=notifications_emails,
173                resume_dataset_id=resume_dataset_id,
174                source_sample_ids=source_sample_ids,
175                compute_environment_id=compute_environment.id if compute_environment else None
176            )
177        )
178        return resp.id
179
180
181class DataPortalProcesses(DataPortalAssets[DataPortalProcess]):
182    """Collection of DataPortalProcess objects."""
183    asset_name = "process"