Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
can't understand the resulting array dtype.
"""
if name.find('/') != -1:
parent_path = op.dirname(name)
basename = op.basename(name)
if not basename:
raise KeyError("Group path can not end with '/'")
parent_uuid, link_json = self._get_link_json(parent_path)
if parent_uuid is None:
raise KeyError("group path: {} not found".format(parent_path))
if link_json["class"] != 'H5L_TYPE_HARD':
raise IOError("cannot create subgroup of softlink")
parent_uuid = link_json["id"]
req = "/groups/" + parent_uuid
group_json = self.GET(req)
tgt = Group(GroupID(self, group_json))
tgt[basename] = obj
elif isinstance(obj, HLObject):
body = {'id': obj.id.uuid }
req = "/groups/" + self.id.uuid + "/links/" + name
self.PUT(req, body=body)
elif isinstance(obj, SoftLink):
body = {'h5path': obj.path }
req = "/groups/" + self.id.uuid + "/links/" + name
self.PUT(req, body=body)
#self.id.links.create_soft(name, self._e(obj.path),
# lcpl=lcpl, lapl=self._lapl)
elif isinstance(obj, ExternalLink):
body = {'h5path': obj.path,
elif uuid.startswith("d-"):
collection_type = "datasets"
else:
raise IOError("Unexpected uuid: {}".format(uuid))
objdb = self.id.http_conn.getObjDb()
if objdb and uuid in objdb:
# we should be able to construct an object from objdb json
obj_json = objdb[uuid]
else:
# will need to get JSON from server
req = "/" + collection_type + "/" + uuid
# make server request
obj_json = self.GET(req)
if collection_type == 'groups':
tgt = Group(GroupID(self, obj_json))
elif collection_type == 'datatypes':
tgt = Datatype(TypeID(self, obj_json))
elif collection_type == 'datasets':
# create a Table if the daset is one dimensional and compound
shape_json = obj_json["shape"]
dtype_json = obj_json["type"]
if "dims" in shape_json and len(shape_json["dims"]) == 1 and dtype_json["class"] == 'H5T_COMPOUND':
tgt = Table(DatasetID(self, obj_json))
else:
tgt = Dataset(DatasetID(self, obj_json))
else:
raise IOError("Unexpecrted collection_type: {}".format(collection_type))
return tgt
def require_group(self, name):
""" Return a group, creating it if it doesn't exist.
TypeError is raised if something with that name already exists that
isn't a group.
"""
if not name in self:
return self.create_group(name)
grp = self[name]
if not isinstance(grp, Group):
raise TypeError("Incompatible object (%s) already exists" % grp.__class__.__name__)
return grp
try:
rsp_json = self.GET(req)
except IOError as ioe:
self.log.debug("Got ioe: {}".format(ioe))
create_group = True
if create_group:
link_json = {'id': parent_uuid, 'name': link}
body = {'link': link_json }
self.log.debug("create group with body: {}".format(body))
rsp = self.POST('/groups', body=body)
group_json = rsp
groupId = GroupID(self, group_json)
sub_group = Group(groupId)
if parent_name:
if parent_name[-1] == '/':
parent_name = parent_name + link
else:
parent_name = parent_name + '/' + link
self.log.debug("create group - parent name: {}".format(parent_name))
sub_group._name = parent_name
parent_uuid = sub_group.id.id
else:
# sub-group already exsits
self.log.debug("create group - found subgroup: {}".format(link))
if "link" not in rsp_json:
raise IOError("Unexpected Error")
link_json = rsp_json["link"]
if link_json["class"] != 'H5L_TYPE_HARD':
# TBD: get the referenced object for softlink?
from __future__ import absolute_import
import os
import time
import json
from .objectid import GroupID
from .group import Group
from .httpconn import HttpConn
from .config import Config
VERBOSE_REFRESH_TIME=1.0 # 1 second
class File(Group):
"""
Represents an HDF5 file.
"""
@property
def attrs(self):
""" Attributes attached to this object """
# hdf5 complains that a file identifier is an invalid location for an
# attribute. Instead of self, pass the root group to AttributeManager:
from . import attrs
#parent_obj = {"id": self.id.uuid}
#return attrs.AttributeManager(self['/'])
return attrs.AttributeManager(self)
@property
if not (getclass or getlink):
try:
return self[name]
except KeyError:
return default
if not name in self:
return default
elif getclass and not getlink:
obj = self.__getitem__(name)
if obj is None:
return None
if obj.id.__class__ is GroupID:
return Group
elif obj.id.__class__ is DatasetID:
return Dataset
elif obj.id.__class__ is TypeID:
return Datatype
else:
raise TypeError("Unknown object type")
elif getlink:
parent_uuid, link_json = self._get_link_json(name)
typecode = link_json['class']
if typecode == 'H5L_TYPE_SOFT':
if getclass:
return SoftLink
return SoftLink(link_json['h5path'])
rsp = http_conn.GET(req)
if rsp.status_code != 200:
http_conn.close()
raise IOError(rsp.status_code, "Unexpected Error")
group_json = json.loads(rsp.text)
groupid = GroupID(None, group_json, http_conn=http_conn)
# end else
self._name = '/'
self._id = groupid
self._verboseInfo = None # aditional state we'll get when requested
self._verboseUpdated = None # when the verbose data was fetched
Group.__init__(self, self._id)