Skip to content
Snippets Groups Projects
pybis.py 113 KiB
Newer Older
  • Learn to ignore specific revisions
  • Swen Vermeul's avatar
    Swen Vermeul committed
                    {
                        "@type": "as.dto.externaldms.fetchoptions.ExternalDmsFetchOptions",
                    },
    
                ],
            }
    
            resp = self._post_request(self.as_v3, request)
            parse_jackson(resp)
    
            if resp is None or len(resp) == 0:
    
                raise ValueError('no such external DMS found: ' + permId)
    
            else:
                for ident in resp:
                    if only_data:
                        return resp[ident]
                    else:
                        return ExternalDMS(self, resp[ident])
    
    
        def new_space(self, **kwargs):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """ Creates a new space in the openBIS instance.
    
            return Space(self, None, **kwargs)
    
    
        def new_git_data_set(self, data_set_type, path, commit_id, repository_id, dms, sample=None, experiment=None, properties={},
    
                             dss_code=None, parents=None, data_set_code=None, contents=[]):
    
            """ Create a link data set.
            :param data_set_type: The type of the data set
    
            :param path: The path to the git repository
            :param commit_id: The git commit id
    
            :param repository_id: The git repository id - same for copies
    
            :param dms: An external data managment system object or external_dms_id
            :param sample: A sample object or sample id.
    
            :param dss_code: Code for the DSS -- defaults to the first dss if none is supplied.
    
            :param properties: Properties for the data set.
            :param parents: Parents for the data set.
    
            :param data_set_code: A data set code -- used if provided, otherwise generated on the server
    
            :param contents: A list of dicts that describe the contents:
                {'file_length': [file length],
                 'crc32': [crc32 checksum],
                 'directory': [is path a directory?]
                 'path': [the relative path string]}
    
            return pbds.GitDataSetCreation(self, data_set_type, path, commit_id, repository_id, dms, sample, experiment,
    
                                           properties, dss_code, parents, data_set_code, contents).new_git_data_set()
    
        def new_content_copy(self, path, commit_id, repository_id, edms_id, data_set_id):
            """
            Create a content copy in an existing link data set.
            :param path: path of the new content copy
            "param commit_id: commit id of the new content copy
            "param repository_id: repository id of the content copy
            "param edms_id: Id of the external data managment system of the content copy
            "param data_set_id: Id of the data set to which the new content copy belongs
            """
    
            return pbds.GitDataSetUpdate(self, data_set_id).new_content_copy(path, commit_id, repository_id, edms_id)
    
    
        def search_files(self, data_set_id, dss_code=None):
            return pbds.GitDataSetFileSearch(self, data_set_id).search_files()        
    
    
        def delete_content_copy(self, data_set_id, content_copy):
            """
            Deletes a content copy from a data set.
            :param data_set_id: Id of the data set containing the content copy
            :param content_copy: The content copy to be deleted
            """
            return pbds.GitDataSetUpdate(self, data_set_id).delete_content_copy(content_copy)        
    
            """Take sample which may be a string or object and return an identifier for it."""
    
            return Openbis._object_to_object_id(sample, "as.dto.sample.id.SampleIdentifier", "as.dto.sample.id.SamplePermId");
    
        @staticmethod
        def experiment_to_experiment_id(experiment):
            """Take experiment which may be a string or object and return an identifier for it."""
            return Openbis._object_to_object_id(experiment, "as.dto.experiment.id.ExperimentIdentifier", "as.dto.experiment.id.SamplePermId");
    
        @staticmethod
        def _object_to_object_id(obj, identifierType, permIdType):
            object_id = None
            if isinstance(obj, str):
                if (is_identifier(obj)):
                    object_id = {
                        "identifier": obj,
                        "@type": identifierType
    
                    object_id = {
                        "permId": obj,
                        "@type": permIdType
    
                object_id = {
                    "identifier": obj.identifier,
                    "@type": identifierType
    
        @staticmethod
        def data_set_to_data_set_id(data_set):
            if isinstance(data_set, str):
                code = data_set
            else:
                code = data_set.permId
            return {
                "permId": code,
                "@type": "as.dto.dataset.id.DataSetPermId"
            }
    
    
        def external_data_managment_system_to_dms_id(self, dms):
            if isinstance(dms, str):
                dms_id = {
                    "permId": dms,
                    "@type": "as.dto.externaldms.id.ExternalDmsPermId"
                }
            else:
                dms_id = {
                    "identifier": dms.code,
                    "@type": "as.dto.sample.id.SampleIdentifier"
                }
            return dms_id
    
    
        def new_sample(self, type, project=None, props=None, **kwargs):
    
            """ Creates a new sample of a given sample type.
    
            if 'collection' in kwargs:
                kwargs['experiment'] = kwargs['collection']
                kwargs.pop('collection', None)
    
            return Sample(self, type=self.get_sample_type(type), project=project, data=None, props=props, **kwargs)
    
        def new_sample_type(self, 
                code,
                **kwargs
            ):
            """ Creates a new sample type
            """
            return SampleType(self, code=code, **kwargs)
    
    
    
        def new_dataset(self, type=None, kind='PHYSICAL_DATA', files=None, props=None, folder=None, **kwargs):
    
            """ Creates a new dataset of a given sample type.
            """
    
            type_obj = self.get_dataset_type(type.upper())
    
            if 'object' in kwargs:
                kwargs['sample'] = kwargs['object']
                kwargs.pop('object', None)
            if 'collection' in kwargs:
                kwargs['experiment'] = kwargs['collection']
                kwargs.pop('collection', None)
    
            return DataSet(self, type=type_obj, kind=kind, files=files, folder=folder, props=props, **kwargs)
    
        
        def new_semantic_annotation(self, entityType=None, propertyType=None, **kwargs):
    
            """ Note: not functional yet. """
    
            return SemanticAnnotation(
    
                openbis_obj=self, isNew=True,
    
                entityType=entityType, propertyType=propertyType, **kwargs
            )    
    
        def new_vocabulary(self, code, terms, managedInternally=False, internalNameSpace=False, chosenFromList=True, **kwargs):
            """ Creates a new vocabulary
            Usage::
                new_vocabulary(
                    code = 'vocabulary_code',
                    description = '',
                    terms = [
                        { "code": "term1", "label": "label1", "description": "description1" },
                        { "code": "term2", "label": "label2", "description": "description2" },
                    ]
                )
            """
            kwargs['code'] = code
            kwargs['managedInternally'] = managedInternally
            kwargs['internalNameSpace'] = internalNameSpace
            kwargs['chosenFromList'] = chosenFromList
            return Vocabulary(self, data=None, terms=terms, **kwargs)
    
            
    
    
        def _get_dss_url(self, dss_code=None):
            """ internal method to get the downloadURL of a datastore.
            """
    
            dss = self.get_datastores()
            if dss_code is None:
                return dss['downloadUrl'][0]
    
                return dss[dss['code'] == dss_code]['downloadUrl'][0]
    
    class ExternalDMS():
        """ managing openBIS external data management systems
    
        def __init__(self, openbis_obj, data=None, **kwargs):
            self.__dict__['openbis'] = openbis_obj
    
            if data is not None:
    
            if kwargs is not None:
                for key in kwargs:
                    setattr(self, key, kwargs[key])
    
        def __getattr__(self, name):
            return self.__dict__['data'].get(name)
    
    
        def __dir__(self):
    
            """all the available methods and attributes that should be displayed
            when using the autocompletion feature (TAB) in Jupyter
            """
            return ['code', 'label', 'urlTemplate', 'address', 'addressType', 'openbis']
    
        def __str__(self):
            return self.data.get('code', None)
    
    class ServerInformation():
    
        def __init__(self, info):
            self._info = info
    
                'api_version', 'archiving_configured', 'authentication_service',
    
                'enabled_technologies', 'project_samples_enabled'
            ]
    
    
        def __getattr__(self, name):
    
            return self._info.get(name.replace('_', '-'))
    
        
        def get_major_version(self):
            return int(self._info["api-version"].split(".")[0]);
        
        def get_minor_version(self):
            return int(self._info["api-version"].split(".")[1]);
        
        def is_openbis_1605(self):
            return (self.get_major_version() == 3) and (self.get_minor_version() <= 2);
        
        def is_openbis_1806(self):
            return (self.get_major_version() == 3) and (self.get_minor_version() >= 5);
    
    
        def _repr_html_(self):
            html = """
                <table border="1" class="dataframe">
                <thead>
                    <tr style="text-align: right;">
                    <th>attribute</th>
                    <th>value</th>
                    </tr>
                </thead>
                <tbody>
            """
    
            for attr in self.attrs:
                html += "<tr> <td>{}</td> <td>{}</td> </tr>".format(
                    attr, getattr(self, attr, '')
                )
    
            html += """
                </tbody>
                </table>
            """
            return html