pybis.py

            if isinstance(props, str):
                props = [props]
            for prop in props:
                datasets[prop.upper()] = datasets['properties'].map(lambda x: x.get(prop.upper(), ''))
                attrs.append(prop.upper())

        return Things(
            openbis_obj = self,
            entity = 'dataset',
            df = datasets[attrs],
            identifier_name = 'permId',
            start_with=start_with,
            count=count,
            totalCount=totalCount,
        )


    def get_sample(self, sample_ident, only_data=False, withAttachments=False, props=None):
        """Retrieve metadata for the sample.
        Get metadata for the sample and any directly connected parents of the sample to allow access
        to the same information visible in the ELN UI. The metadata will be on the file system.
        :param sample_identifiers: A list of sample identifiers to retrieve.
        """

        only_one = True
        identifiers = []
        if isinstance(sample_ident, list):
            only_one = False
            for ident in sample_ident:
                identifiers.append(
                    _type_for_id(ident, 'sample')
                )
        else:
            identifiers.append(
                _type_for_id(sample_ident, 'sample')
            )

        fetchopts = {"type": {"@type": "as.dto.sample.fetchoptions.SampleTypeFetchOptions"}}

        options = ['tags', 'properties', 'attachments', 'space', 'experiment', 'registrator', 'modifier', 'dataSets']
        if self.get_server_information().project_samples_enabled:
            options.append('project')
        for option in options:
            fetchopts[option] = fetch_option[option]

        if withAttachments:
            fetchopts['attachments'] = fetch_option['attachmentsWithContent']

        for key in ['parents','children','container','components']:
            fetchopts[key] = {"@type": "as.dto.sample.fetchoptions.SampleFetchOptions"}

        request = {
            "method": "getSamples",
            "params": [
                self.token,
                identifiers,
                fetchopts
            ],
        }

        resp = self._post_request(self.as_v3, request)

        if only_one:
            if len(resp) == 0:
                raise ValueError('no such sample found: {}'.format(sample_ident))

            parse_jackson(resp)
            for sample_ident in resp:
                if only_data:
                    return resp[sample_ident]
                else:
                    return Sample(
                        openbis_obj = self,
                        type = self.get_sample_type(resp[sample_ident]["type"]["code"]),
                        data = resp[sample_ident]
                    )
        else:
            return self._sample_list_for_response(
                response=list(resp.values()),
                props=props,
            )

    def _sample_list_for_response(
        self, response, props=None,
        start_with=None, count=None, totalCount=0
    ):
        """returns a Things object, containing a DataFrame plus some additional information
        """

        parse_jackson(response)
        attrs = ['identifier', 'permId', 'experiment', 'type',
                 'registrator', 'registrationDate', 'modifier', 'modificationDate']
        if len(response) == 0:
            samples = DataFrame(columns=attrs)
        else:
            samples = DataFrame(response)
            samples['registrationDate'] = samples['registrationDate'].map(format_timestamp)
            samples['modificationDate'] = samples['modificationDate'].map(format_timestamp)
            samples['registrator'] = samples['registrator'].map(extract_person)
            samples['modifier'] = samples['modifier'].map(extract_person)
            samples['identifier'] = samples['identifier'].map(extract_identifier)
            samples['permId'] = samples['permId'].map(extract_permid)
            samples['experiment'] = samples['experiment'].map(extract_nested_identifier)
            samples['type'] = samples['type'].map(extract_nested_permid)

        if props is not None:
            if isinstance(props, str):
                props = [props]
            for prop in props:
                samples[prop.upper()] = samples['properties'].map(lambda x: x.get(prop.upper(), ''))
                attrs.append(prop.upper())

        return Things(
            openbis_obj = self,
            entity = 'sample',
            df = samples[attrs],
            identifier_name = 'identifier',
            start_with=start_with,
            count=count,
            totalCount=totalCount,
        )


    get_object = get_sample # Alias


    def get_external_data_management_system(self, permId, only_data=False):
        """Retrieve metadata for the external data management system.
        :param permId: A permId for an external DMS.
        :param only_data: Return the result data as a hash-map, not an object.
        """

        request = {
            "method": "getExternalDataManagementSystems",
            "params": [
                self.token,
                [{
                    "@type": "as.dto.externaldms.id.ExternalDmsPermId",
                    "permId": permId
                }],
                {
                    "@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):
        """ 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 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: A DataSet object
        """
        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)        

    @staticmethod
    def sample_to_sample_id(sample):
        """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
                }
            else:
                object_id = {
                    "permId": obj,
                    "@type": permIdType
                }
        else:
            object_id = {
                "identifier": obj.identifier,
                "@type": identifierType
            }
        return object_id

    @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)

    new_object = new_sample # Alias

    def new_sample_type(self, 
            code,
            **kwargs
        ):
        return SampleType(self, code=code, **kwargs)

    new_object_type = new_sample_type

    def new_dataset_type(self, 
            code,
            **kwargs
        ):
        return DataSetType(self, code=code, **kwargs)

    new_object_type = new_sample_type

    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]
        else:
            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:
            self.__dict__['data'] = data

        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
        self.attrs = [
            'api_version', 'archiving_configured', 'authentication_service',
            'enabled_technologies', 'project_samples_enabled'
        ]

    def __dir__(self):
        return self.attrs

    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


class PropertyType(
    OpenBisObject, 
    entity='propertyType',
    single_item_method_name='get_property_type'
):
    pass

class Plugin(
    OpenBisObject,
    entity='plugin',
    single_item_method_name='get_plugin'
):
    pass