Skip to content
Snippets Groups Projects
pybis.py 86.6 KiB
Newer Older
  • Learn to ignore specific revisions
  •     @property
        def terms_kv(self):
             return [ 
                {voc["code"]:voc["label"] }
                for voc 
                in sorted(self.data['objects'], key=lambda v: v["ordinal"]) 
            ]
    
        @property
        def terms(self):
             return [ 
                voc["code"]
                for voc 
                in sorted(self.data['objects'], key=lambda v: v["ordinal"]) 
            ]
    
    
        def _repr_html_(self):
            html = """
                <table border="1" class="dataframe">
                <thead>
                    <tr style="text-align: right;">
                    <th>vocabulary term</th>
                    <th>label</th>
                    <th>description</th>
                    <th>vocabulary</th>
                    </tr>
                </thead>
                <tbody>
            """
    
            for voc in sorted(
                self.data['objects'], 
                key=lambda v: (v["permId"]["vocabularyCode"], v["ordinal"])
            ):
    
                html += "<tr> <td>{}</td> <td>{}</td> <td>{}</td> <td>{}</td> </tr>".format(
                    voc['code'],
                    voc['label'],
                    voc['description'],
                    voc['permId']['vocabularyCode']
                )
    
            html += """
                </tbody>
                </table>
            """
            return html
    
    
    
    class PropertyHolder():
    
        def __init__(self, openbis_obj, type):
            self.__dict__['_openbis'] = openbis_obj
            self.__dict__['_type'] = type
            self.__dict__['_property_names'] = []
            for prop in type.data['propertyAssignments']:
                self._property_names.append(prop['propertyType']['code'].lower())
    
    
        def _get_terms(self, vocabulary):
            return self._openbis.get_terms(vocabulary)
    
        def _all_props(self):
            props = {}
            for code in self._type.codes():
                props[code] = getattr(self, code)
            return props
    
        def __getattr__(self, name):
            if name.endswith('_'):
                name = name.rstrip('_')
                property_type = self._type.prop[name]['propertyType']
                if property_type['dataTypeCode'] == 'CONTROLLEDVOCABULARY':
                    return self._openbis.get_terms(name)
                else:
                    return { property_type["label"] : property_type["dataTypeCode"]}
            else: return None
    
    
        def __setattr__(self, name, value):
            if name not in self._property_names:
                raise KeyError("No such property: {}".format(name)) 
    
            property_type = self._type.prop[name]['propertyType']
    
            data_type = property_type['dataTypeCode']
            if data_type == 'CONTROLLEDVOCABULARY':
    
                voc = self._openbis.get_terms(name)
                if value not in voc.terms:
                    raise ValueError("Value must be one of these terms: " + ", ".join(voc.terms))
    
            elif data_type in ('INTEGER', 'BOOLEAN', 'VARCHAR'):
                if not check_datatype(data_type, value):
                    raise ValueError("Value must be of type {}".format(data_type))
            self.__dict__[name] = value
    
        def __dir__(self):
            return self._property_names
    
        def _repr_html_(self):
            html = """
                <table border="1" class="dataframe">
                <thead>
                    <tr style="text-align: right;">
                    <th>property</th>
                    <th>value</th>
                    </tr>
                </thead>
                <tbody>
            """
    
            for prop in self._property_names:
                value = ''
                try:
                    value = getattr(self, prop)
                except Exception:
                    pass
    
                html += "<tr> <td>{}</td> <td>{}</td> </tr>".format(
                    prop,
                    value
                )
    
            html += """
                </tbody>
                </table>
            """
            return html
    
    
    class AttrHolder():
    
        """ General class for both samples and experiments that hold all common attributes, such as:
        - space
        - experiment
        - parents
        - children
        - tags
        """
    
        def __init__(self, openbis_obj, type):
            self.__dict__['_openbis'] = openbis_obj
            self.__dict__['_type_obj'] = type
    
            entity_type = type.data['@type'].split('.')[2].capitalize()
            self.__dict__['_entity_type'] = entity_type
            self.__dict__['_allowed_attrs'] = _definitions(entity_type)['attrs']
            self.__dict__['_identifier'] = None
            self.__dict__['_is_new'] = True
    
        def __call__(self, data):
            self.__dict__['_is_new'] = False
            for key in  "code permId identifier type container components attachments".split():
                self.__dict__['_'+key] = data.get(key, None)
    
            for key in "space project experiment".split():
                d =  data.get(key, None)
                if d is not None:
                    d = d['permId']
                self.__dict__['_'+key] = d
    
            for key in "parents children".split():
                self.__dict__['_'+key] = []
                for item in data[key]:
                    self.__dict__['_'+key].append(item['identifier'])
    
            for key in "tags".split():
                self.__dict__['_'+key] = []
                for item in data[key]:
                    self.__dict__['_'+key].append({
                        "code": item['code'],
                        "@type": "as.dto.tag.id.TagCode"
                    })
    
    
        def _all_attrs(self):
            defs = _definitions(self._entity_type)
            attr2ids = _definitions('attr2ids')
            ids2type = _definitions('ids2type')
    
            request = {}
    
            for attr in defs['attrs']:
                # only continue if attribute is actually defined
                if '_'+attr in self.__dict__:
                    # handle multivalue attributes (parents, children, tags etc.)
                    if attr in defs['multi']:
                        items = self.__dict__.get('_'+attr, [])
                        if items == None:
                            items = []
                        request[attr2ids[attr]] = {
                            "actions": [
                                {
                                    "items": items,
                                    "@type": "as.dto.common.update.ListUpdateActionSet",
                                }
                            ],
                            "@type": "as.dto.common.update.IdListUpdateValue"
                        }
                    else:
                        ids = attr2ids[attr]
                        request[ids] = self.__dict__.get('_'+attr, None)
    
            if self.__dict__.get('_code', None) is None:
                request['autoGeneratedCode'] = True
            else:
                pass
                
            return request
    
        def __getattr__(self, name):
            """ handles all attribute requests dynamically. Values are returned in a sensible way,
                for example the identifiers of parents, children and components are returned
                as an array of values.
            """
                    
            int_name = '_'+name
            if int_name in self.__dict__: 
                if int_name in ["_experiment"]:
                    exp = ''
                    try:
                        exp = self.__dict__[int_name]['identifier']['identifier']
                    except TypeError:
                        pass
                    return exp
                elif isinstance(self.__dict__[int_name], list):
                    values = []
                    for item in self.__dict__[int_name]:
                        values.append(
                            item.get("code",
                                item.get("identifier",
                                    item.get("permId", None)))
                        )
                    return values
                elif isinstance(self.__dict__[int_name], dict):
                    return self.__dict__[int_name].get(name,
                        self.__dict__[int_name].get("code",
                            self.__dict__[int_name].get("identifier",
                                self.__dict__[int_name].get("permId", None))))
                else:
                    return self.__dict__[int_name]
            else:
                return None
    
        def __setattr__(self, name, value):
            if name in ["parents", "children", "components"]:
                if not isinstance(value, list):
                    value = [value]
                objs = []
                for val in value:
                    # fetch objects in openBIS, make sure they actually exists
                    obj = getattr(self._openbis, 'get_'+self._entity_type.lower())(val)
                    objs.append(obj)
                self.__dict__['_'+name] = {
                    "@type": "as.dto.common.update.IdListUpdateValue",
                    "actions": [{
                        "@type": "as.dto.common.update.ListUpdateActionSet",
                        "items": [item._permId for item in objs]
                    }]
                }
            elif name in ["tags"]:
                tags = []
                for val in value:
                    tags.append({
                        "@type": "as.dto.tag.id.TagCode",
                        "code": val
                    })
    
                self.__dict__['_tags'] = tags
    
            elif name in ["space", "project", "experiment"]:
                # fetch object in openBIS, make sure it actually exists
                obj = getattr(self._openbis, "get_"+name)(value)
                self.__dict__['_'+name] = obj['permId']
                # mark attribute as modified, if it's an existing entity
                if self.__dict__['_is_new']:
                    pass
                else:
                    self.__dict__['_'+name]['is_modified'] = True
    
            elif name in ["identifier", "project"]:
                raise KeyError("you can not modify the {}".format(name))
            elif name == "code":
                if self.__dict__['_type_obj'].data['autoGeneratedCode']:
                    raise KeyError("for this {}Type you can not set a code".format(self.__dict__['_entity_type']))
                else:
                    self.__dict__['_code'] = value
            else:
                raise KeyError("no such attribute: {}".format(name))
    
        def get_type(self):
            return self._type_obj
    
        def get_parents(self):
            return getattr(self._openbis, 'get_'+self._entity_type.lower()+'s')(withChildren=self.identifier)
    
        def get_children(self):
            return getattr(self._openbis, 'get_'+self._entity_type.lower()+'s')(withParents=self.identifier)
    
        @property 
    
        def set_tags(self, tags):
            tagIds = _tagIds_for_tags(tags, 'Set')
            self.openbis.update_sample(self.permId, tagIds=tagIds)
    
        def add_tags(self, tags):
            tagIds = _tagIds_for_tags(tags, 'Add')
            self.openbis.update_sample(self.permId, tagIds=tagIds)
    
        def del_tags(self, tags):
            tagIds = _tagIds_for_tags(tags, 'Remove')
            self.openbis.update_sample(self.permId, tagIds=tagIds)
    
        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 key in ["identifier","permId", "code", "type", "space","project", "experiment", "parents", "children", "tags"]:
                html += "<tr> <td>{}</td> <td>{}</td> </tr>".format(key, getattr(self, key, None))
    
        """ A Sample is one of the most commonly used objects in openBIS.
    
        def __init__(self, openbis_obj, type, data=None, **kwargs):
    
            self.__dict__['openbis'] = openbis_obj
            self.__dict__['type'] = type
            self.__dict__['p'] = PropertyHolder(openbis_obj, type)
            self.__dict__['a'] = AttrHolder(openbis_obj, type)
    
            if kwargs is not None:
                for key in kwargs:
                    setattr(self, key, kwargs[key])
    
        def _set_data(self, data):
                # assign the attribute data to self.a by calling it 
                # (invoking the AttrHolder.__call__ function)
                self.a(data)
                self.__dict__['data'] = data
    
    
                # put the properties in the self.p namespace (without checking them)
                for key, value in data['properties'].items():
                    self.p.__dict__[key.lower()] = value
    
    
    
        @property
        def type(self):
            return self.__dict__['type'].data['code']
    
        @type.setter
        def type(self, type_name):
                sample_type = self.openbis.get_sample_type(type_name)
                self.p.__dict__['_type'] = sample_type
                self.a.__dict__['_type'] = sample_type
    
    
        def __getattr__(self, name):
            return getattr(self.__dict__['a'], name)
    
        def __setattr__(self, name, value):
            if name in ['set_properties', 'set_tags', 'add_tags']:
                raise ValueError("These are methods which should not be overwritten")
    
    
            setattr(self.__dict__['a'], name, value)
    
    
        def _repr_html_(self):
            html = self.a._repr_html_()
            return html
    
        def set_properties(self, properties):
            self.openbis.update_sample(self.permId, properties=properties)
    
        def save(self):
    
            props = self.p._all_props()
            attrs = self.a._all_attrs()
            attrs["properties"] = props
    
            if self.identifier is None:
                # create a new sample
                attrs["@type"] = "as.dto.sample.create.SampleCreation"
                attrs["typeId"] = self.__dict__['type'].data['permId']
                request = {
                    "method": "createSamples",
                    "params": [ self.openbis.token,
                        [ attrs ]
                    ]
                }
                resp = self.openbis._post_request(self.openbis.as_v3, request)
                new_sample_data = self.openbis.get_sample(resp[0]['permId'], only_data=True)
                self._set_data(new_sample_data)
                return self
                
    
                attrs["@type"] = "as.dto.sample.update.SampleUpdate"
                attrs["sampleId"] = {
                    "permId": self.permId,
                    "@type": "as.dto.sample.id.SamplePermId"
                }
                request = {
                    "method": "updateSamples",
                    "params": [ self.openbis.token,
                        [ attrs ]
                    ]
                }
                resp = self.openbis._post_request(self.openbis.as_v3, request)
                print('Sample successfully updated')
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def delete(self, reason):
    
            self.openbis.delete_entity('sample', self.permId, reason)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            objects = self.dataSets
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            datasets = DataFrame(objects)
            datasets['registrationDate'] = datasets['registrationDate'].map(format_timestamp)
            datasets['properties'] = datasets['properties'].map(extract_properties)
            datasets['type'] = datasets['type'].map(extract_code)
    
    class Space(dict):
        """ managing openBIS spaces
        """
    
        def __init__(self, openbis_obj, *args, **kwargs):
            super(Space, self).__init__(*args, **kwargs)
            self.__dict__ = self
            self.openbis = openbis_obj
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_samples(self, *args, **kwargs):
    
            """ Lists all samples in a given space. A pandas DataFrame object is returned.
            """
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
            return self.openbis.get_samples(space=self.code, *args, **kwargs)
    
    
        @property
        def projects(self):
            return self.openbis.get_projects(space=self.code)
        
        def new_project(self, **kwargs):
            return self.openbis.new_project(space=self.code, **kwargs)
    
        @property
        def experiments(self):
            return self.openbis.get_experiments(space=self.code)
    
    
    
    class Things():
        """An object that contains a DataFrame object about an entity  available in openBIS.
           
        """
    
        def __init__(self, openbis_obj, what, df, identifier_name='code'):
            self.openbis = openbis_obj
            self.what = what
            self.df = df
            self.identifier_name = identifier_name
    
        def _repr_html_(self):
            return self.df._repr_html_()
    
        def __getitem__(self, key):
            if self.df is not None and len(self.df) > 0:
                row = None
                if isinstance(key, int):
                    # get thing by rowid
                    row = self.df.loc[[key]]
    
                elif isinstance(key, list):
                    # treat it as a normal dataframe
                    return self.df[key]
    
                else:
                    # get thing by code
                    row = self.df[self.df[self.identifier_name]==key.upper()]
    
                if row is not None:
                    # invoke the openbis.get_what() method
                    return getattr(self.openbis, 'get_'+self.what)(row[self.identifier_name].values[0])
    
    
    
        """ managing openBIS experiments
        """
    
    
        def __init__(self, openbis_obj, data):
    
            self.openbis = openbis_obj
    
            self.permId = data['permId']['permId']
            self.identifier  = data['identifier']['identifier']
            self.properties = data['properties']
            self.tags = extract_tags(data['tags'])
            self.attachments = extract_attachments(data['attachments'])
            self.project = data['project']['code']
            self.data = data
    
        def set_properties(self, properties):
    
            self.openbis.update_experiment(self.permId, properties=properties)
    
    
        def set_tags(self, tags):
            tagIds = _tagIds_for_tags(tags, 'Set')
    
            self.openbis.update_experiment(self.permId, tagIds=tagIds)
    
    
        def add_tags(self, tags):
            tagIds = _tagIds_for_tags(tags, 'Add')
    
            self.openbis.update_experiment(self.permId, tagIds=tagIds)
    
    
        def del_tags(self, tags):
            tagIds = _tagIds_for_tags(tags, 'Remove')
    
            self.openbis.update_experiment(self.permId, tagIds=tagIds)
    
            self.openbis.delete_entity('experiment', self.permId, reason)
    
    
        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>
        <tr> <th>permId</th> <td>{}</td> </tr>
        <tr> <th>identifier</th> <td>{}</td> </tr>
        <tr> <th>project</th> <td>{}</td> </tr>
        <tr> <th>properties</th> <td>{}</td> </tr>
        <tr> <th>tags</th> <td>{}</td> </tr>
        <tr> <th>attachments</th> <td>{}</td> </tr>
      </tbody>
    </table>
            """
    
            return html.format(self.permId, self.identifier, self.project, self.properties, self.tags, self.attachments)
    
    
        def __repr__(self):
            data = {}
    
            data["identifier"] = self.identifier
    
            data["properties"] = self.properties
            data["tags"] = self.tags
    
            return repr(data)
    
    
        """ holds are properties, that are assigned to an entity, eg. sample or experiment
        """
    
    
        def __init__(self, openbis_obj, data):
            self.openbis = openbis_obj
            self.data = data
            self.prop = {}
            for pa in self.data['propertyAssignments']:
                self.prop[pa['propertyType']['code'].lower()] = pa
    
    
        def codes(self):
            codes = []
            for pa in self.data['propertyAssignments']:
                codes.append(pa['propertyType']['code'].lower())
            return codes
    
    <p>{}: <b>{}</b>
    <p>description: {}</p>
    <p>Code autogenerated: {}</p>
    
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th>property</th>
          <th>label</th>
          <th>description</th>
          <th>dataTypeCode</th>
          <th>mandatory</th>
        </tr>
      </thead>
      <tbody>
    
            """.format(
                self.data['@type'].split('.')[-1],
                self.data['code'], 
                self.data['description'], 
                self.data['autoGeneratedCode']
            )
    
    
            for pa in self.data['propertyAssignments']:
                html += "<tr> <th>{}</th> <td>{}</td> <td>{}</td> <td>{}</td> <td>{}</td> </tr>".format(
                    pa['propertyType']['code'].lower(),    
                    pa['propertyType']['label'],    
                    pa['propertyType']['description'],    
                    pa['propertyType']['dataTypeCode'],    
                    pa['mandatory']
                )
    
            html += """
                </tbody>
                </table>
            """
            return html