Newer
Older
<thead>
<tr style="text-align: right;">
<th>attribute</th>
<th>value</th>
</tr>
</thead>
<tbody>
"""
Swen Vermeul
committed
for attr in self._allowed_attrs:
if attr == 'attachments':
continue
html += "<tr> <td>{}</td> <td>{}</td> </tr>".format(
attr, nvl(getattr(self, attr, ''), '')
Swen Vermeul
committed
)
if getattr(self, '_attachments') is not None:
Swen Vermeul
committed
html += "<tr><td>attachments</td><td>"
html += "<br/>".join(att['fileName'] for att in self._attachments)
html += "</td></tr>"
html += """
</tbody>
</table>
"""
return html
def __repr__(self):
""" When using iPython, this method displays a nice table
of all attributes and their values when the object is printed.
"""
headers = ['attribute', 'value']
lines = []
for attr in self._allowed_attrs:
if attr == 'attachments':
continue
elif attr == 'users' and '_users' in self.__dict__:
lines.append([
attr,
", ".join(att['userId'] for att in self._users)
])
elif attr == 'roleAssignments' and '_roleAssignments' in self.__dict__:
roles = []
for role in self._roleAssignments:
if role.get('space') is not None:
roles.append("{} ({})".format(
role.get('role'),
role.get('space').get('code')
))
else:
roles.append(role.get('role'))
lines.append([
attr,
", ".join(roles)
])
else:
lines.append([
attr,
nvl(getattr(self, attr, ''))
])
return tabulate(lines, headers=headers)
class Sample():
""" A Sample is one of the most commonly used objects in openBIS.
"""
Swen Vermeul
committed
def __init__(self, openbis_obj, type, data=None, props=None, **kwargs):
self.__dict__['openbis'] = openbis_obj
self.__dict__['type'] = type
self.__dict__['p'] = PropertyHolder(openbis_obj, type)
self.__dict__['a'] = AttrHolder(openbis_obj, 'Sample', type)
if data is not None:
self._set_data(data)
Swen Vermeul
committed
if props is not None:
for key in props:
setattr(self.p, key, props[key])
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
Swen Vermeul
committed
def __dir__(self):
'props', 'get_parents()', 'get_children()',
'add_parents()', 'add_children()', 'del_parents()', 'del_children()',
'space', 'project', 'experiment', 'tags',
'set_tags()', 'add_tags()', 'del_tags()',
Swen Vermeul
committed
'add_attachment()', 'get_attachments()', 'download_attachments()',
'save()', 'delete()'
Swen Vermeul
committed
@property
def props(self):
return self.__dict__['p']
Swen Vermeul
committed
@property
def type(self):
@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):
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):
props = self.p._all_props()
Swen Vermeul
committed
if self.is_new:
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
else:
request["params"][1][0]["properties"] = props
self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Sample successfully updated.")
Swen Vermeul
committed
new_sample_data = self.openbis.get_sample(self.permId, only_data=True)
self._set_data(new_sample_data)
self.openbis.delete_entity(entity='Sample',id=self.permId, reason=reason)
if VERBOSE: print("Sample {} successfully 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)
Swen Vermeul
committed
def get_experiment(self):
Swen Vermeul
committed
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
Swen Vermeul
committed
4186
4187
4188
4189
4190
4191
4192
4193
4194
4195
4196
4197
4198
4199
4200
4201
4202
4203
4204
4205
4206
4207
4208
4209
4210
4211
4212
4213
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'))
Swen Vermeul
committed
def delete(self, reason='no reason specified'):
self.openbis.delete_entity(
entity='RoleAssignment', id=self.id,
reason=reason, id_name='techId'
)
4221
4222
4223
4224
4225
4226
4227
4228
4229
4230
4231
4232
4233
4234
4235
4236
4237
4238
4239
4240
4241
4242
4243
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',
Swen Vermeul
committed
'registrator', 'registrationDate','space',
'get_roles()', 'assign_role(role, space)', 'revoke_role(role)',
Swen Vermeul
committed
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):
try:
self.openbis.assign_role(role=role, person=self, **kwargs)
if VERBOSE:
print(
"Role {} successfully assigned to person {}".format(role, self.userId)
)
except ValueError as e:
if 'exists' in str(e):
if VERBOSE:
print(
"Role {} already assigned to person {}".format(role, self.userId)
)
else:
raise ValueError(str(e))
Swen Vermeul
committed
def revoke_role(self, role, space=None, project=None, reason='no reason specified'):
Swen Vermeul
committed
""" Revoke a role from this person.
"""
techId = None
Swen Vermeul
committed
if isinstance(role, int):
techId = role
else:
query = { "role": role }
if space is None:
query['space'] = ''
else:
query['space'] = space.upper()
if project is None:
query['project'] = ''
else:
query['project'] = project.upper()
# build a query string for dataframe
querystr = " & ".join(
'{} == "{}"'.format(key, value) for key, value in query.items()
)
roles = self.get_roles().df
Swen Vermeul
committed
if len(roles) == 0:
if VERBOSE:
print("Role has already been revoked from person {}".format(role, self.code))
return
techId = roles.query(querystr)['techId'].values[0]
# finally delete the role assignment
ra = self.openbis.get_role_assignment(techId)
ra.delete(reason)
if VERBOSE:
print(
"Role {} successfully revoked from person {}".format(role, self.code)
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'])
self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Person successfully updated.")
4348
4349
4350
4351
4352
4353
4354
4355
4356
4357
4358
4359
4360
4361
4362
4363
4364
4365
4366
4367
4368
4369
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()',
Swen Vermeul
committed
'get_roles()', 'assign_role()', 'revoke_role(techId)'
def get_persons(self):
Swen Vermeul
committed
""" Returns a Things object wich contains all Persons (Users)
that belong to this group.
"""
columns = ['permId', 'userId', 'firstName', 'lastName', 'email', 'space', 'registrationDate', 'active']
persons = DataFrame(self._users)
if len(persons) == 0:
persons = DataFrame(columns=columns)
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[columns],
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')
"""
Swen Vermeul
committed
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')
"""
try:
self.openbis.assign_role(role=role, group=self, **kwargs)
if VERBOSE:
print(
"Role {} successfully assigned to group {}".format(role, self.code)
)
except ValueError as e:
if 'exists' in str(e):
if VERBOSE:
print(
"Role {} already assigned to group {}".format(role, self.code)
)
else:
raise ValueError(str(e))
Swen Vermeul
committed
def revoke_role(self, role, space=None, project=None, reason='no reason specified'):
""" Revoke a role from this group.
"""
Swen Vermeul
committed
if isinstance(role, int):
Swen Vermeul
committed
else:
query = { "role": role }
if space is None:
query['space'] = ''
else:
query['space'] = space.upper()
if project is None:
query['project'] = ''
else:
query['project'] = project.upper()
# build a query string for dataframe
querystr = " & ".join(
'{} == "{}"'.format(key, value) for key, value in query.items()
)
roles = self.get_roles().df
Swen Vermeul
committed
if len(roles) == 0:
if VERBOSE:
print("Role has already been revoked from group {}".format(role, self.code))
return
techId = roles.query(querystr)['techId'].values[0]
# finally delete the role assignment
ra = self.openbis.get_role_assignment(techId)
ra.delete(reason)
if VERBOSE:
print(
"Role {} successfully revoked from group {}".format(role, self.code)
)
return
""" creates a nice table in Jupyter notebooks when the object itself displayed
"""
4478
4479
4480
4481
4482
4483
4484
4485
4486
4487
4488
4489
4490
4491
4492
4493
4494
4495
4496
4497
4498
4499
4500
4501
4502
4503
4504
4505
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>
"""
4506
4507
4508
4509
4510
4511
4512
4513
4514
4515
4516
4517
4518
4519
4520
4521
4522
4523
4524
4525
4526
4527
4528
4529
4530
4531
4532
4533
4534
4535
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 delete(self, reason='unknown'):
self.openbis.delete_entity(
entity = "AuthorizationGroup",
id = self.permId,
reason = reason
)
if VERBOSE:
print("Authorization group {} successfully deleted".format(
self.permId
))
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_group(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)
""" 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)",
'get_samples()',
'delete()'
]
def __str__(self):
return self.data.get('code', None)
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) )
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)
def new_sample(self, **kwargs):
return self.openbis.new_sample(space=self, **kwargs)
def delete(self, reason):
self.openbis.delete_entity('Space', self.permId, reason)
if VERBOSE: print("Space {} has been sucsessfully deleted.".format(self.permId))
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)
Chandrasekhar Ramakrishnan
committed
4642
4643
4644
4645
4646
4647
4648
4649
4650
4651
4652
4653
4654
4655
4656
4657
4658
4659
4660
4661
4662
4663
4664
4665
4666
4667
4668
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.df = df
self.identifier_name = identifier_name
return tabulate(self.df, headers=list(self.df))
def __len__(self):
return len(self.df)
def _repr_html_(self):
return self.df._repr_html_()
if self.entity not in ['sample', 'dataset']:
Swen Vermeul
committed
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:
parents = getattr(self.openbis, 'get_' + self.entity.lower() + 's')(withChildren=ident, **kwargs)
Swen Vermeul
committed
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)
if self.entity not in ['sample', 'dataset']:
Swen Vermeul
committed
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:
parents = getattr(self.openbis, 'get_' + self.entity.lower() + 's')(withParent=ident, **kwargs)
Swen Vermeul
committed
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
committed
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:
samples = self.openbis.get_samples(**args, **kwargs)
Swen Vermeul
committed
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
if self.entity not in ['sample', 'experiment']:
Swen Vermeul
committed
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:
datasets = self.openbis.get_datasets(**args, **kwargs)
Swen Vermeul
committed
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]
row = self.df[self.df[self.identifier_name] == key.upper()]
# invoke the openbis.get_<entity>() method
return getattr(self.openbis, 'get_' + self.entity)(row[self.identifier_name].values[0])
Swen Vermeul
committed
def __iter__(self):
for item in self.df[[self.identifier_name]][self.identifier_name].iteritems():
yield getattr(self.openbis, 'get_' + self.entity)(item[1])
Swen Vermeul
committed
# return self.df[[self.identifier_name]].to_dict()[self.identifier_name]
Swen Vermeul
committed
Swen Vermeul
committed
"""
def __init__(self, openbis_obj, type, project=None, data=None, props=None, code=None, **kwargs):
Swen Vermeul
committed
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
committed
if data is not None:
self._set_data(data)
if project is not None:
setattr(self, 'project', project)
Swen Vermeul
committed
if props is not None:
for key in props:
setattr(self.p, key, props[key])
if code is not None:
self.code = code
Swen Vermeul
committed
if kwargs is not None:
Swen Vermeul
committed
defs = _definitions('Experiment')
Swen Vermeul
committed
for key in kwargs:
Swen Vermeul
committed
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))
Swen Vermeul
committed
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
Swen Vermeul
committed
# 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
committed
def __str__(self):
return self.data['code']
Swen Vermeul
committed
def __dir__(self):
# the list of possible methods/attributes displayed
# when invoking TAB-completition
return [
'code', 'permId', 'identifier',
'type', 'project',
'props.',
'project', 'tags', 'attachments', 'data',
'get_datasets()', 'get_samples()',
'set_tags()', 'add_tags()', 'del_tags()',
Swen Vermeul
committed
'add_attachment()', 'get_attachments()', 'download_attachments()',
'save()'
Swen Vermeul
committed
]
@property
def props(self):
return self.__dict__['p']
@property
def type(self):
Swen Vermeul
committed
@type.setter
def type(self, type_name):
experiment_type = self.openbis.get_experiment_type(type_name)
self.p.__dict__['_type'] = experiment_type
self.a.__dict__['_type'] = experiment_type
Swen Vermeul
committed
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()']:
Swen Vermeul
committed
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_experiment(self.permId, properties=properties)
Swen Vermeul
committed
def save(self):
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
Swen Vermeul
committed
else:
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)
def delete(self, reason):
if self.permId is None:
return None
self.openbis.delete_entity(entity='Experiment', id=self.permId, reason=reason)
if VERBOSE: print("Experiment {} successfully deleted.".format(self.permId))
if self.permId is None:
return None
return self.openbis.get_datasets(experiment=self.permId, **kwargs)
if self.permId is None:
return None
return self.openbis.get_project(experiment=self.permId, **kwargs)
Swen Vermeul
committed
if self.permId is None:
return None
return self.openbis.get_samples(experiment=self.permId, **kwargs)
Swen Vermeul
committed
get_objects = get_samples # Alias
Swen Vermeul
committed
4925
4926
4927
4928
4929
4930
4931
4932
4933
4934
4935
4936
4937
4938
4939
4940
4941
4942
4943
4944
4945
4946
4947
4948
4949
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
Swen Vermeul
committed
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
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
committed
"fileName": self.fileName,
"title": self.title,
"description": self.description,
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
committed
def __init__(self, openbis_obj, data=None, **kwargs):
self.__dict__['openbis'] = openbis_obj
self.__dict__['a'] = AttrHolder(openbis_obj, 'Project')