Newer
Older
tagIds = _create_tagIds(tags)
for tagId in tagIds:
if tagId in self.__dict__['_tags']:
self.__dict__['_tags'].remove(tagId)
def get_attachments(self):
if getattr(self, '_attachments') is None:
return None
else:
return DataFrame(self._attachments)[['fileName', 'title', 'description', 'version']]
def add_attachment(self, fileName, title=None, description=None):
att = Attachment(filename=fileName, title=title, description=description)
if getattr(self, '_attachments') is None:
self.__dict__['_attachments'] = []
self._attachments.append(att.get_data_short())
if getattr(self, '_new_attachments') is None:
self.__dict__['_new_attachments'] = []
self._new_attachments.append(att)
def download_attachments(self):
method = 'get' + self.entity + 's'
entity = self.entity.lower()
request = {
"method": method,
"params": [
self._openbis.token,
[self._permId],
dict(
attachments=fetch_option['attachmentsWithContent'],
**fetch_option[entity]
)
]
}
resp = self._openbis._post_request(self._openbis.as_v3, request)
attachments = resp[self.permId]['attachments']
file_list = []
for attachment in attachments:
filename = os.path.join(
self._openbis.hostname,
self.permId,
attachment['fileName']
)
os.makedirs(os.path.dirname(filename), exist_ok=True)
with open(filename, 'wb') as att:
content = base64.b64decode(attachment['content'])
att.write(content)
def _repr_html_(self):
Swen Vermeul
committed
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>
"""
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
4245
4246
4247
4248
4249
4250
4251
4252
4253
4254
4255
4256
4257
4258
4259
4260
4261
4262
4263
4264
4265
4266
4267
4268
4269
4270
4271
4272
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'
)
4280
4281
4282
4283
4284
4285
4286
4287
4288
4289
4290
4291
4292
4293
4294
4295
4296
4297
4298
4299
4300
4301
4302
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.")
4407
4408
4409
4410
4411
4412
4413
4414
4415
4416
4417
4418
4419
4420
4421
4422
4423
4424
4425
4426
4427
4428
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
"""
4537
4538
4539
4540
4541
4542
4543
4544
4545
4546
4547
4548
4549
4550
4551
4552
4553
4554
4555
4556
4557
4558
4559
4560
4561
4562
4563
4564
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>
"""
4565
4566
4567
4568
4569
4570
4571
4572
4573
4574
4575
4576
4577
4578
4579
4580
4581
4582
4583
4584
4585
4586
4587
4588
4589
4590
4591
4592
4593
4594
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
4701
4702
4703
4704
4705
4706
4707
4708
4709
4710
4711
4712
4713
4714
4715
4716
4717
4718
4719
4720
4721
4722
4723
4724
4725
4726
4727
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
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: