Skip to content
Snippets Groups Projects
pybis.py 174 KiB
Newer Older
  • Learn to ignore specific revisions
  •         # put the properties in the self.p namespace (without checking them)
            for key, value in data['properties'].items():
                self.p.__dict__[key.lower()] = value
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            return [
    
                'props', 'get_parents()', 'get_children()', 
                'add_parents()', 'add_children()', 'del_parents()', 'del_children()',
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                'get_datasets()', 'get_experiment()',
    
                'space', 'project', 'experiment', 'tags',
    
                'set_tags()', 'add_tags()', 'del_tags()',
    
                'add_attachment()', 'get_attachments()', 'download_attachments()',
                'save()', 'delete()'
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            ]
    
            return self.__dict__['type']
    
            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)
    
            return self.a._repr_html_()
    
        def __repr__(self):
            return self.a.__repr__()
    
    
        def set_properties(self, properties):
            self.openbis.update_sample(self.permId, properties=properties)
    
        def save(self):
    
                request = self._new_attrs()
    
                request["params"][1][0]["properties"] = props
    
                resp = self.openbis._post_request(self.openbis.as_v3, request)
    
                if VERBOSE: print("Sample successfully created.")
    
                new_sample_data = self.openbis.get_sample(resp[0]['permId'], only_data=True)
                self._set_data(new_sample_data)
                return self
    
                request = self._up_attrs()
    
                request["params"][1][0]["properties"] = props
    
                self.openbis._post_request(self.openbis.as_v3, request)
    
                if VERBOSE: print("Sample successfully updated.")
    
                new_sample_data = self.openbis.get_sample(self.permId, only_data=True)
                self._set_data(new_sample_data)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def delete(self, reason):
    
            self.openbis.delete_entity(entity='Sample',id=self.permId, reason=reason)
            if VERBOSE: print("Sample {} has been sucessfully deleted.".format(self.permId))
    
        def get_datasets(self, **kwargs):
            return self.openbis.get_datasets(sample=self.permId, **kwargs)
    
        def get_projects(self, **kwargs):
            return self.openbis.get_project(withSamples=[self.permId], **kwargs)
    
                return self.openbis.get_experiment(self._experiment['identifier'])
            except Exception:
                pass
    
    
        @property
        def experiment(self):
    
                return self.openbis.get_experiment(self._experiment['identifier'])
            except Exception:
                pass
    
    class RoleAssignment(OpenBisObject):
        """ managing openBIS role assignments
        """
    
        def __init__(self, openbis_obj, data=None, **kwargs):
            self.__dict__['openbis'] = openbis_obj
            self.__dict__['a'] = AttrHolder(openbis_obj, 'RoleAssignment' )
    
            if data is not None:
                self.a(data)
                self.__dict__['data'] = data
    
            if kwargs is not None:
                for key in kwargs:
                    setattr(self, key, kwargs[key])
    
    
        def __dir__(self):
            """all the available methods and attributes that should be displayed
            when using the autocompletion feature (TAB) in Jupyter
            """
            return [
                'id', 'role', 'roleLevel', 'space', 'project','group'
            ]
    
        def __str__(self):
            return "{}".format(self.get('role'))
    
        def delete(self, reason):
            self.openbis.delete_entity(
                entity='RoleAssignment', id=self.id, 
                reason=reason, id_name='techId'
            ) 
            if VERBOSE: print("RoleAssignment has been sucessfully deleted.".format(self.permId))
    
    
    
    class Person(OpenBisObject):
        """ managing openBIS persons
        """
    
        def __init__(self, openbis_obj, data=None, **kwargs):
            self.__dict__['openbis'] = openbis_obj
            self.__dict__['a'] = AttrHolder(openbis_obj, 'Person' )
    
            if data is not None:
                self.a(data)
                self.__dict__['data'] = data
    
            if kwargs is not None:
                for key in kwargs:
                    setattr(self, key, kwargs[key])
    
    
        def __dir__(self):
            """all the available methods and attributes that should be displayed
            when using the autocompletion feature (TAB) in Jupyter
            """
            return [
                'permId', 'userId', 'firstName', 'lastName', 'email',
    
                'registrator', 'registrationDate','space',
                'get_roles()', 'assign_role()', 'revoke_role(techId)',
    
    
        def get_roles(self, **search_args):
            """ Get all roles that are assigned to this person.
            Provide additional search arguments to refine your search.
    
            Usage::
                person.get_roles()
                person.get_roles(space='TEST_SPACE')
            """
            return self.openbis.get_role_assignments(person=self, **search_args)
    
    
        def assign_role(self, role, **kwargs):
            self.openbis.assign_role(role=role, person=self, **kwargs)
    
            if VERBOSE:
                print(
                    "Role {} successfully assigned to person {}".format(role, self.userId)
                ) 
    
    
        def revoke_role(self, role, reason='no specific reason'):
            """ Revoke a role from this person. 
            """
            if isinstance(role, int):
                ra = self.openbis.get_role_assignment(role)
                ra.delete(reason)
                if VERBOSE:
                    print(
                        "Role {} successfully revoked from person {}".format(role, self.code)
                    ) 
                return
            else:
                raise ValueError("Please provide the techId of the role assignment you want to revoke")
    
        def __str__(self):
            return "{} {}".format(self.get('firstName'), self.get('lastName'))
    
        def delete(self, reason):
            raise ValueError("Persons cannot be deleted")
    
        def save(self):
            if self.is_new:
                request = self._new_attrs()
                # for new and updated objects, the parameter is
                # unfortunately called homeSpaceId, spaceId throws no error
                # but makes no change either
                if "spaceId" in request['params'][1][0]:
                    request['params'][1][0]['homeSpaceId'] =  request['params'][1][0]['spaceId']
                    del(request['params'][1][0]['spaceId'])
                resp = self.openbis._post_request(self.openbis.as_v3, request)
    
                if VERBOSE: print("Person successfully created.")
    
                new_person_data = self.openbis.get_person(resp[0]['permId'], only_data=True)
                self._set_data(new_person_data)
                return self
    
            else:
                request = self._up_attrs()
                # for new and updated objects, the parameter is
                # unfortunately called homeSpaceId, spaceId throws no error
                # but makes no change either
                if "spaceId" in request['params'][1][0]:
                    request['params'][1][0]['homeSpaceId'] =  request['params'][1][0]['spaceId']
                    del(request['params'][1][0]['spaceId'])
    
    
                return json.dumps(request)
    
                self.openbis._post_request(self.openbis.as_v3, request)
    
                if VERBOSE: print("Person successfully updated.")
    
                new_person_data = self.openbis.get_person(self.permId, only_data=True)
                self._set_data(new_person_data)
    
    
    class Group(OpenBisObject):
        """ Managing openBIS authorization groups
        """
        
        def __init__(self, openbis_obj, data=None, **kwargs):
            self.__dict__['openbis'] = openbis_obj
            self.__dict__['a'] = AttrHolder(openbis_obj, 'AuthorizationGroup')
    
            if data is not None:
                self.a(data)
                self.__dict__['data'] = data
    
            if kwargs is not None:
                for key in kwargs:
                    setattr(self, key, kwargs[key])
    
        def __dir__(self):
            return [
    
                'code','description','users','roleAssignments',
                'get_users()', 'set_users()', 'add_users()', 'del_users()',
    
                'get_roles()', 'assign_role()', 'revoke_role(techId)'
    
            """ Returns a Things object wich contains all Persons (Users)
            that belong to this group.
            """
    
            persons = DataFrame(self._users)
            persons['permId'] = persons['permId'].map(extract_permid)
            persons['registrationDate'] = persons['registrationDate'].map(format_timestamp)
            persons['space'] = persons['space'].map(extract_nested_permid)
            p = Things(
                self.openbis, entity='person', 
                df=persons[['permId', 'userId', 'firstName', 'lastName', 'email', 'space', 'registrationDate', 'active']],
                identifier_name='permId'
            )
            return p
    
        get_users = get_persons  # Alias
    
    
        def get_roles(self, **search_args):
            """ Get all roles that are assigned to this group.
            Provide additional search arguments to refine your search.
    
            Usage::
                group.get_roles()
                group.get_roles(space='TEST_SPACE')
            """
    
            return self.openbis.get_role_assignments(group=self, **search_args)
    
    
        def assign_role(self, role, **kwargs):
            """ Assign a role to this group. If no additional attribute is provided,
            roleLevel will default to INSTANCE. If a space attribute is provided,
            the roleLevel will be SPACE. If a project attribute is provided,
            roleLevel will be PROJECT.
    
            Usage::
                group.assign_role(role='ADMIN')
                group.assign_role(role='ADMIN', space='TEST_SPACE')
    
            """
    
            self.openbis.assign_role(role=role, group=self, **kwargs)
    
            if VERBOSE:
                print(
                    "Role {} successfully assigned to group {}".format(role, self.code)
                ) 
    
    
        def revoke_role(self, role, reason='no specific reason'):
    
            if isinstance(role, int):
                ra = self.openbis.get_role_assignment(role)
                ra.delete(reason)
                if VERBOSE:
                    print(
                        "Role {} successfully revoked from group {}".format(role, self.code)
                    ) 
                return
            else:
                raise ValueError("Please provide the techId of the role assignment you want to revoke")
            
    
        def _repr_html_(self):
    
            """ creates a nice table in Jupyter notebooks when the object itself displayed
            """
    
            def nvl(val, string=''):
                if val is None:
                    return string
                return val
    
            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._allowed_attrs:
                if attr in ['users','roleAssignments']:
                    continue
                html += "<tr> <td>{}</td> <td>{}</td> </tr>".format(
                    attr, nvl(getattr(self, attr, ''), '')
                )
    
            html += """
                </tbody>
                </table>
            """
    
    
            if getattr(self, '_users') is not None:
                html += """
                    <br/>
                    <b>Users</b>
                    <table border="1" class="dataframe">
                    <thead>
                        <tr style="text-align: right;">
                        <th>userId</th>
                        <th>FirstName</th>
                        <th>LastName</th>
                        <th>Email</th>
                        <th>Space</th>
                        <th>active</th>
                        </tr>
                    </thead>
                    <tbody>
                """
                for user in self._users:
                    html += "<tr><td>{}</td><td>{}</td><td>{}</td><td>{}</td><td>{}</td><td>{}</td></tr>".format(
                        user.get('userId'),
                        user.get('firstName'),
                        user.get('lastName'),
                        user.get('email'),
                        user.get('space').get('code') if user.get('space') is not None else '',
                        user.get('active'),
                    )
                html += """
                    </tbody>
                    </table>
                """
    
        def save(self):
            if self.is_new:
                request = self._new_attrs()
                resp = self.openbis._post_request(self.openbis.as_v3, request)
    
                if VERBOSE: print("Group successfully created.")
    
                # re-fetch group from openBIS
                new_data = self.openbis.get_person(resp[0]['permId'], only_data=True)
                self._set_data(new_data)
                return self
    
            else:
                request = self._up_attrs()
                self.openbis._post_request(self.openbis.as_v3, request)
    
                if VERBOSE: print("Group successfully updated.")
    
                # re-fetch group from openBIS
                new_data = self.openbis.get_group(self.permId, only_data=True)
                self._set_data(new_data)
    
    
    
    class Space(OpenBisObject):
    
        """ managing openBIS spaces
        """
    
    
        def __init__(self, openbis_obj, data=None, **kwargs):
    
            self.__dict__['openbis'] = openbis_obj
    
            self.__dict__['a'] = AttrHolder(openbis_obj, 'Space' )
    
            if data is not None:
                self.a(data)
                self.__dict__['data'] = data
    
            if kwargs is not None:
                for key in kwargs:
                    setattr(self, key, kwargs[key])
    
        def __dir__(self):
            """all the available methods and attributes that should be displayed
            when using the autocompletion feature (TAB) in Jupyter
            """
    
            return [
                'code', 'permId', 'description', 'registrator', 'registrationDate', 'modificationDate', 
                'get_projects()', 
    
                "new_project(code='', description='', attachments)", 
    
        def __str__(self):
            return self.data.get('code', None)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_samples(self, **kwargs):
            return self.openbis.get_samples(space=self.code, **kwargs)
    
        get_objects = get_samples #Alias
    
    
        def get_sample(self, sample_code):
            if is_identifier(sample_code) or is_permid(sample_code):
                return self.openbis.get_sample(sample_code)
            else:
                # we assume we just got the code
                return self.openbis.get_sample('/{}/{}'.format(self.code,sample_code) )
    
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_projects(self, **kwargs):
            return self.openbis.get_projects(space=self.code, **kwargs)
    
        def new_project(self, code, description=None, **kwargs):
            return self.openbis.new_project(self.code, code, description, **kwargs)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    
        def new_sample(self, **kwargs):
            return self.openbis.new_sample(space=self, **kwargs)
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def delete(self, reason):
            self.openbis.delete_entity('Space', self.permId, reason)
    
            if VERBOSE: print("Space {} has been sucessfully deleted.".format(self.permId))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    
        def save(self):
            if self.is_new:
                request = self._new_attrs()
                resp = self.openbis._post_request(self.openbis.as_v3, request)
    
                if VERBOSE: print("Space successfully created.")
    
                new_space_data = self.openbis.get_space(resp[0]['permId'], only_data=True)
                self._set_data(new_space_data)
                return self
    
            else:
                request = self._up_attrs()
                self.openbis._post_request(self.openbis.as_v3, request)
    
                if VERBOSE: print("Space successfully updated.")
    
                new_space_data = self.openbis.get_space(self.permId, only_data=True)
                self._set_data(new_space_data)
    
    
    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 Things():
        """An object that contains a DataFrame object about an entity  available in openBIS.
           
        """
    
    
        def __init__(self, openbis_obj, entity, df, identifier_name='code'):
    
            self.openbis = openbis_obj
    
            self.entity = entity
    
            self.df = df
            self.identifier_name = identifier_name
    
    
        def __repr__(self):
    
            return tabulate(self.df, headers=list(self.df))
    
        def __len__(self):
            return len(self.df)
    
    
        def _repr_html_(self):
            return self.df._repr_html_()
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_parents(self, **kwargs):
    
            if self.entity not in ['sample', 'dataset']:
    
                raise ValueError("{}s do not have parents".format(self.entity))
    
            if self.df is not None and len(self.df) > 0:
                dfs = []
                for ident in self.df[self.identifier_name]:
                    # get all objects that have this object as a child == parent
                    try:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        parents = getattr(self.openbis, 'get_' + self.entity.lower() + 's')(withChildren=ident, **kwargs)
    
                        dfs.append(parents.df)
                    except ValueError:
                        pass
    
                if len(dfs) > 0:
                    return Things(self.openbis, self.entity, pd.concat(dfs), self.identifier_name)
                else:
                    return Things(self.openbis, self.entity, DataFrame(), self.identifier_name)
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_children(self, **kwargs):
    
            if self.entity not in ['sample', 'dataset']:
    
                raise ValueError("{}s do not have children".format(self.entity))
    
            if self.df is not None and len(self.df) > 0:
                dfs = []
                for ident in self.df[self.identifier_name]:
                    # get all objects that have this object as a child == parent
                    try:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        parents = getattr(self.openbis, 'get_' + self.entity.lower() + 's')(withParent=ident, **kwargs)
    
                        dfs.append(parents.df)
                    except ValueError:
                        pass
    
                if len(dfs) > 0:
                    return Things(self.openbis, self.entity, pd.concat(dfs), self.identifier_name)
                else:
                    return Things(self.openbis, self.entity, DataFrame(), self.identifier_name)
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_samples(self, **kwargs):
    
            if self.entity not in ['space', 'project', 'experiment']:
                raise ValueError("{}s do not have samples".format(self.entity))
    
            if self.df is not None and len(self.df) > 0:
                dfs = []
                for ident in self.df[self.identifier_name]:
                    args = {}
                    args[self.entity.lower()] = ident
                    try:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        samples = self.openbis.get_samples(**args, **kwargs)
    
                        dfs.append(samples.df)
                    except ValueError:
                        pass
    
                if len(dfs) > 0:
                    return Things(self.openbis, 'sample', pd.concat(dfs), 'identifier')
                else:
                    return Things(self.openbis, 'sample', DataFrame(), 'identifier')
    
    
        get_objects = get_samples # Alias
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_datasets(self, **kwargs):
    
            if self.entity not in ['sample', 'experiment']:
    
                raise ValueError("{}s do not have datasets".format(self.entity))
    
            if self.df is not None and len(self.df) > 0:
                dfs = []
                for ident in self.df[self.identifier_name]:
                    args = {}
                    args[self.entity.lower()] = ident
                    try:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        datasets = self.openbis.get_datasets(**args, **kwargs)
    
                        dfs.append(datasets.df)
                    except ValueError:
                        pass
    
                if len(dfs) > 0:
                    return Things(self.openbis, 'dataset', pd.concat(dfs), 'permId')
                else:
                    return Things(self.openbis, 'dataset', DataFrame(), 'permId')
    
    
        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_<entity>() method
    
                    return getattr(self.openbis, 'get_' + self.entity)(row[self.identifier_name].values[0])
    
        def __iter__(self):
            for item in self.df[[self.identifier_name]][self.identifier_name].iteritems():
    
                yield getattr(self.openbis, 'get_' + self.entity)(item[1])
    
                # return self.df[[self.identifier_name]].to_dict()[self.identifier_name]
    
    class Experiment(OpenBisObject):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def __init__(self, openbis_obj, type, project=None, data=None, props=None, code=None, **kwargs):
    
            self.__dict__['openbis'] = openbis_obj
            self.__dict__['type'] = type
            self.__dict__['p'] = PropertyHolder(openbis_obj, type)
    
            self.__dict__['a'] = AttrHolder(openbis_obj, 'Experiment', type)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if project is not None:
                setattr(self, 'project', project)
    
    
            if props is not None:
                for key in props:
                    setattr(self.p, key, props[key])
    
    
            if code is not None:
                self.code = code
    
    
                    if key in defs['attrs']:
                        setattr(self, key, kwargs[key])
                    else:
                        raise ValueError("{attr} is not a known attribute for an Experiment".format(attr=key))
    
    
            # 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
    
        def __str__(self):
            return self.data['code']
    
    
        def __dir__(self):
            # the list of possible methods/attributes displayed
            # when invoking TAB-completition
            return [
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                'code', 'permId', 'identifier',
                'type', 'project',
                'props.', 
    
                'project', 'tags', 'attachments', 'data',
                'get_datasets()', 'get_samples()',
    
                'set_tags()', 'add_tags()', 'del_tags()',
    
                'add_attachment()', 'get_attachments()', 'download_attachments()',
                'save()'
    
            ]
    
        @property
        def props(self):
            return self.__dict__['p']
    
        @property
        def type(self):
    
            return self.__dict__['type']
    
            experiment_type = self.openbis.get_experiment_type(type_name)
            self.p.__dict__['_type'] = experiment_type
            self.a.__dict__['_type'] = experiment_type
    
    
        def __getattr__(self, name):
            return getattr(self.__dict__['a'], name)
    
        def __setattr__(self, name, value):
    
            if name in ['set_properties', 'add_tags()', 'del_tags()', 'set_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
    
            self.openbis.update_experiment(self.permId, properties=properties)
    
            if self.is_new:
                request = self._new_attrs()
                props = self.p._all_props()
    
                request["params"][1][0]["properties"] = props
    
                resp = self.openbis._post_request(self.openbis.as_v3, request)
    
    
                if VERBOSE: print("Experiment successfully created.")
    
                new_exp_data = self.openbis.get_experiment(resp[0]['permId'], only_data=True)
                self._set_data(new_exp_data)
                return self
    
                request = self._up_attrs()
                props = self.p._all_props()
    
                request["params"][1][0]["properties"] = props
    
                self.openbis._post_request(self.openbis.as_v3, request)
    
                if VERBOSE: print("Experiment successfully updated.")
    
                new_exp_data = self.openbis.get_experiment(resp[0]['permId'], only_data=True)
                self._set_data(new_exp_data)
    
            if self.permId is None:
                return None
    
            self.openbis.delete_entity(entity='Experiment', id=self.permId, reason=reason)
            if VERBOSE: print("Experiment {} has been sucessfully deleted.".format(self.permId))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_datasets(self, **kwargs):
    
            if self.permId is None:
                return None
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            return self.openbis.get_datasets(experiment=self.permId, **kwargs)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_projects(self, **kwargs):
    
            if self.permId is None:
                return None
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            return self.openbis.get_project(experiment=self.permId, **kwargs)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_samples(self, **kwargs):
    
            if self.permId is None:
                return None
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            return self.openbis.get_samples(experiment=self.permId, **kwargs)
    
        get_objects = get_samples # Alias
    
    
        def add_samples(self, *samples):
    
            for sample in samples:
                if isinstance(sample, str):
                    obj = self.openbis.get_sample(sample)
                else:
                    # we assume we got a sample object
                    obj = sample
    
                # a sample can only belong to exactly one experiment
                if obj.experiment is not None:
                    raise ValueError(
                        "sample {} already belongs to experiment {}".format(
                            obj.code, obj.experiment
                        )
                    )
                else:
                    if self.is_new:
                        raise ValueError("You need to save this experiment first before you can assign any samples to it")
                    else:
                        # update the sample directly
                        obj.experiment = self.identifier
                        obj.save()
                        self.a.__dict__['_samples'].append(obj._identifier)
    
    
        add_objects = add_samples # Alias
    
    
        def del_samples(self, samples):
            if not isinstance(samples, list):
                samples = [samples]
    
            
            for sample in samples:
                if isinstance(sample, str):
                    obj = self.openbis.get_sample(sample)
                    objects.append(obj)
                else:
                    # we assume we got an object
                    objects.append(obj)
            
            self.samples = objects
    
        del_objects = del_samples # Alias
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    class Attachment():
        def __init__(self, filename, title=None, description=None):
            if not os.path.exists(filename):
                raise ValueError("File not found: {}".format(filename))
            self.fileName = filename
            self.title = title
            self.description = description
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_data_short(self):
            return {
    
                "fileName": self.fileName,
                "title": self.title,
                "description": self.description,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            }
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_data(self):
            with open(self.fileName, 'rb') as att:
                content = att.read()
                contentb64 = base64.b64encode(content).decode()
            return {
    
                "fileName": self.fileName,
                "title": self.title,
                "description": self.description,
                "content": contentb64,
                "@type": "as.dto.attachment.create.AttachmentCreation",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            }
    
    class Project(OpenBisObject):
    
        def __init__(self, openbis_obj, data=None, **kwargs):
            self.__dict__['openbis'] = openbis_obj
            self.__dict__['a'] = AttrHolder(openbis_obj, 'Project')
    
            if data is not None:
                self.a(data)
                self.__dict__['data'] = data
    
            if kwargs is not None:
                for key in kwargs:
                    setattr(self, key, kwargs[key])
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def _modifiable_attrs(self):
            return
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def __dir__(self):
            """all the available methods and attributes that should be displayed
            when using the autocompletion feature (TAB) in Jupyter
            """
    
            return ['code', 'permId', 'identifier', 'description', 'space', 'registrator',
                    'registrationDate', 'modifier', 'modificationDate', 'add_attachment()',
                    'get_attachments()', 'download_attachments()',
                    'get_experiments()', 'get_samples()', 'get_datasets()',
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_samples(self, **kwargs):
            return self.openbis.get_samples(project=self.permId, **kwargs)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    
        def get_sample(self, sample_code):
            if is_identifier(sample_code) or is_permid(sample_code):
                return self.openbis.get_sample(sample_code)
            else:
                # we assume we just got the code
                return self.openbis.get_sample(project=self, code=sample_code)
    
    
        get_objects = get_samples # Alias
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_experiments(self):
            return self.openbis.get_experiments(project=self.permId)
    
        def get_datasets(self):
            return self.openbis.get_datasets(project=self.permId)
    
    
        def delete(self, reason):
    
            self.openbis.delete_entity(entity='Project', id=self.permId, reason=reason)
            if VERBOSE: print("Project {} has been sucessfully deleted.".format(self.permId))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def save(self):
    
            if self.is_new:
                request = self._new_attrs()
    
                resp = self.openbis._post_request(self.openbis.as_v3, request)
    
                self.a.__dict__['_is_new'] = False
    
                if VERBOSE: print("Project successfully created.")
    
                new_project_data = self.openbis.get_project(resp[0]['permId'], only_data=True)
                self._set_data(new_project_data)
                return self
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            else:
    
                request = self._up_attrs()
                self.openbis._post_request(self.openbis.as_v3, request)
    
                if VERBOSE: print("Project successfully updated.")
    
    
    
    class SemanticAnnotation():
        def __init__(self, openbis_obj, isNew=True, **kwargs):
            self._openbis = openbis_obj
            self._isNew = isNew;
            
            self.permId = kwargs.get('permId')
            self.entityType = kwargs.get('entityType')
            self.propertyType = kwargs.get('propertyType')
            self.predicateOntologyId = kwargs.get('predicateOntologyId')
            self.predicateOntologyVersion = kwargs.get('predicateOntologyVersion')
            self.predicateAccessionId = kwargs.get('predicateAccessionId')
            self.descriptorOntologyId = kwargs.get('descriptorOntologyId')
            self.descriptorOntologyVersion = kwargs.get('descriptorOntologyVersion')
            self.descriptorAccessionId = kwargs.get('descriptorAccessionId')
            self.creationDate = kwargs.get('creationDate')
    
        def __dir__(self):
    
            return [
                'permId', 'entityType', 'propertyType', 
                'predicateOntologyId', 'predicateOntologyVersion', 
                'predicateAccessionId', 'descriptorOntologyId',
                'descriptorOntologyVersion', 'descriptorAccessionId', 
                'creationDate', 
                'save()', 'delete()' 
            ]
    
    
        def save(self):
            if self._isNew:
                self._create()
            else:
                self._update()
                
        def _create(self):
            
            creation = {
                "@type": "as.dto.semanticannotation.create.SemanticAnnotationCreation"
            }
    
            if self.entityType is not None and self.propertyType is not None:
                creation["propertyAssignmentId"] = {
                    "@type": "as.dto.property.id.PropertyAssignmentPermId",
                    "entityTypeId" : {
                        "@type": "as.dto.entitytype.id.EntityTypePermId",
                        "permId" : self.entityType,
                        "entityKind" : "SAMPLE"
                    },
                    "propertyTypeId" : {
                        "@type" : "as.dto.property.id.PropertyTypePermId",
                        "permId" : self.propertyType
                    }
                }
            elif self.entityType is not None:
                creation["entityTypeId"] = {
                    "@type": "as.dto.entitytype.id.EntityTypePermId",
                    "permId" : self.entityType,
                    "entityKind" : "SAMPLE"
                }
            elif self.propertyType is not None:
                creation["propertyTypeId"] = {
                    "@type" : "as.dto.property.id.PropertyTypePermId",
                    "permId" : self.propertyType
                }
                
            for attr in ['predicateOntologyId', 'predicateOntologyVersion', 'predicateAccessionId', 'descriptorOntologyId', 'descriptorOntologyVersion', 'descriptorAccessionId']:
                creation[attr] = getattr(self, attr)
    
            request = {
                "method": "createSemanticAnnotations",
                "params": [
                    self._openbis.token,
                    [creation]
                ]
            }
            
            self._openbis._post_request(self._openbis.as_v3, request)
            self._isNew = False
            
    
            if VERBOSE: print("Semantic annotation successfully created.")
    
        
        def _update(self):
            
            update = {
                "@type": "as.dto.semanticannotation.update.SemanticAnnotationUpdate",
                "semanticAnnotationId" : {
                    "@type" : "as.dto.semanticannotation.id.SemanticAnnotationPermId",
                    "permId" : self.permId
                }
            }
            
            for attr in ['predicateOntologyId', 'predicateOntologyVersion', 'predicateAccessionId', 'descriptorOntologyId', 'descriptorOntologyVersion', 'descriptorAccessionId']:
                update[attr] = {
                    "@type" : "as.dto.common.update.FieldUpdateValue",
                    "isModified" : True,
                    "value" : getattr(self, attr)
                }
                
            request = {
                "method": "updateSemanticAnnotations",
                "params": [
                    self._openbis.token,
                    [update]
                ]
            }
            
            self._openbis._post_request(self._openbis.as_v3, request)
    
            if VERBOSE: print("Semantic annotation successfully updated.")
    
            self._openbis.delete_entity(entity='SemanticAnnotation', id=self.permId, reason=reason)
    
            if VERBOSE: print("Semantic annotation successfully deleted.")