cirro.sdk.process

  1from typing import List, Union
  2
  3from cirro_api_client.v1.models import Process, Executor, ProcessDetail, CustomPipelineSettings, PipelineCode
  4
  5from cirro.cirro_client import CirroApi
  6from cirro.models.form_specification import ParameterSpecification
  7from cirro.sdk.asset import DataPortalAssets, DataPortalAsset
  8
  9
 10class DataPortalProcess(DataPortalAsset):
 11    """Helper functions for interacting with analysis processes."""
 12
 13    def __init__(self, process: Union[Process, ProcessDetail], client: CirroApi):
 14        """
 15        Instantiate with helper method
 16
 17        ```python
 18        from cirro import DataPortal()
 19        portal = DataPortal()
 20        process = portal.get_process_by_name("Process Name")
 21        ```
 22        """
 23        self._data = process
 24        self._client = client
 25
 26    @property
 27    def id(self) -> str:
 28        """Unique identifier"""
 29        return self._data.id
 30
 31    @property
 32    def name(self) -> str:
 33        """Readable name"""
 34        return self._data.name
 35
 36    @property
 37    def description(self) -> str:
 38        """Longer description of process"""
 39        return self._data.description
 40
 41    @property
 42    def child_process_ids(self) -> List[str]:
 43        """List of processes which can be run on the output of this process"""
 44        return self._data.child_process_ids
 45
 46    @property
 47    def executor(self) -> Executor:
 48        """INGEST, CROMWELL, or NEXTFLOW"""
 49        return self._data.executor
 50
 51    @property
 52    def category(self) -> str:
 53        """Category of process"""
 54        return self._data.category
 55
 56    @property
 57    def pipeline_type(self) -> str:
 58        """Pipeline type"""
 59        return self._data.pipeline_type
 60
 61    @property
 62    def documentation_url(self) -> str:
 63        """Documentation URL"""
 64        return self._data.documentation_url
 65
 66    @property
 67    def file_requirements_message(self) -> str:
 68        """Description of files required for INGEST processes"""
 69        return self._data.file_requirements_message
 70
 71    @property
 72    def code(self) -> PipelineCode:
 73        """Pipeline code configuration"""
 74        return self._get_detail().pipeline_code
 75
 76    @property
 77    def custom_settings(self) -> CustomPipelineSettings:
 78        """Custom settings for the process"""
 79        return self._get_detail().custom_settings
 80
 81    def _get_detail(self) -> ProcessDetail:
 82        if not isinstance(self._data, ProcessDetail):
 83            self._data = self._client.processes.get(self.id)
 84        return self._data
 85
 86    def __str__(self):
 87        return '\n'.join([
 88            f"{i.title()}: {self.__getattribute__(i)}"
 89            for i in ['name', 'id', 'description']
 90        ])
 91
 92    def get_parameter_spec(self) -> ParameterSpecification:
 93        """
 94        Gets a specification used to describe the parameters used in the process.
 95        """
 96        return self._client.processes.get_parameter_spec(self.id)
 97
 98
 99class DataPortalProcesses(DataPortalAssets[DataPortalProcess]):
100    """Collection of DataPortalProcess objects."""
101    asset_name = "process"