Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def makeExternalLinks():
file_path = SRC + "/link_example.h5"
if os.path.exists(file_path):
return # don't waste time re-creating
tgt_link_path = os.path.abspath(DES) # for absolute paths in link
tgt_link_path += "/tall.h5"
f = h5py.File(file_path, 'w')
f.create_group('g1')
f.create_group('g1/g1.1')
f['soft_link'] = h5py.SoftLink('g1')
f['external_link1'] = h5py.ExternalLink('tall.h5', 'g1/g1.1')
f['external_link2'] = h5py.ExternalLink('tall', 'g1/g1.1')
f['external_link3'] = h5py.ExternalLink('tall.test.hdfgroup.org', 'g1/g1.1')
f['external_link4'] = h5py.ExternalLink(tgt_link_path, 'g1/g1.1')
f['external_link5'] = h5py.ExternalLink('tall.subdir.test.hdfgroup.org', 'g1/g1.1')
f['external_link6'] = h5py.ExternalLink('tall.subdir', 'g1/g1.1')
f['external_link7'] = h5py.ExternalLink('subdir/tall.h5', 'g1/g1.1')
f.close()
def makeExternalLinks():
file_path = SRC + "/link_example.h5"
if os.path.exists(file_path):
return # don't waste time re-creating
tgt_link_path = os.path.abspath(DES) # for absolute paths in link
tgt_link_path += "/tall.h5"
f = h5py.File(file_path, 'w')
f.create_group('g1')
f.create_group('g1/g1.1')
f['soft_link'] = h5py.SoftLink('g1')
f['external_link1'] = h5py.ExternalLink('tall.h5', 'g1/g1.1')
f['external_link2'] = h5py.ExternalLink('tall', 'g1/g1.1')
f['external_link3'] = h5py.ExternalLink('tall.test.hdfgroup.org', 'g1/g1.1')
f['external_link4'] = h5py.ExternalLink(tgt_link_path, 'g1/g1.1')
f['external_link5'] = h5py.ExternalLink('tall.subdir.test.hdfgroup.org', 'g1/g1.1')
f['external_link6'] = h5py.ExternalLink('tall.subdir', 'g1/g1.1')
f['external_link7'] = h5py.ExternalLink('subdir/tall.h5', 'g1/g1.1')
f.close()
self.assertTrue("mysoftlink" in r)
self.assertEqual(len(r), 5)
self.assertEqual(len(g1), 1)
self.assertEqual(len(g1_1), 0)
slink = r['mysoftlink']
self.assertEqual(slink.id, g1_1.id)
# create a file that we'll link to
link_target_filename = self.getFileName("link_target")
g = h5py.File(link_target_filename, 'w')
g.create_group("somepath")
g.close()
# create a external hardlink
r['myexternallink'] = h5py.ExternalLink(link_target_filename, "somepath")
# test getclass
g1_class = r.get('g1', getclass=True)
self.assertEqual(g1_class, h5py.Group)
linkee_class = r.get('mysoftlink', getclass=True)
self.assertEqual(linkee_class, h5py.Group)
link_class = r.get('mysoftlink', getclass=True, getlink=True)
self.assertEqual(link_class, h5py.SoftLink)
softlink = r.get('mysoftlink', getlink=True)
self.assertEqual(softlink.path, '/g1/g1.1')
linkee_class = r.get('myexternallink', getclass=True)
link_class = r.get('myexternallink', getclass=True, getlink=True)
self.assertEqual(link_class, h5py.ExternalLink)
external_link = r.get('myexternallink', getlink=True)
self.assertEqual(external_link.path, 'somepath')
external_link_filename = external_link.filename
def link_run(self, filepath, run_idx, continue_run=None, **kwargs):
"""Add a run from another file to this one as an HDF5 external
link. Intuitively this is like mounting a drive in a filesystem."""
# link to the external run
ext_run_link = h5py.ExternalLink(filepath, 'runs/{}'.format(run_idx))
# the run index in this file, as determined by the counter
here_run_idx = self.next_run_idx()
# set the local run as the external link to the other run
self._h5['runs/{}'.format(here_run_idx)] = ext_run_link
# run the initialization routines for adding a run
self._add_run_init(here_run_idx, continue_run=continue_run)
run_grp = self._h5['runs/{}'.format(here_run_idx)]
# add metadata if given
for key, val in kwargs.items():
if key != 'run_idx':
run_grp.attrs[key] = val
g = h5.create_group("group")
g.create_dataset("dataset", data=numpy.int64(10))
h5.create_dataset("dataset", data=numpy.int64(10))
h5["hard_link_to_group"] = h5["/group"]
h5["hard_link_to_dataset"] = h5["/dataset"]
h5["soft_link_to_group"] = h5py.SoftLink("/group")
h5["soft_link_to_dataset"] = h5py.SoftLink("/dataset")
h5["soft_link_to_nothing"] = h5py.SoftLink("/foo/bar/2000")
alltypes_filename = get_hdf5_with_all_types()
h5["external_link_to_group"] = h5py.ExternalLink(alltypes_filename, "/arrays")
h5["external_link_to_dataset"] = h5py.ExternalLink(alltypes_filename, "/arrays/cube")
h5["external_link_to_nothing"] = h5py.ExternalLink(alltypes_filename, "/foo/bar/2000")
h5["external_link_to_missing_file"] = h5py.ExternalLink("missing_file.h5", "/")
h5.close()
_file_cache[ID] = tmp
return tmp.name
--> data
--> definition (leaf)
--> instrument
--> sample
'''
output_file_name = os.path.splitext(self.params.cxi_file)[0]+'_master.h5'
f = h5py.File(output_file_name, 'w')
entry = f.create_group('entry')
entry.attrs['NX_class'] = 'NXentry'
# --> definition
self._create_scalar(entry, 'definition', 'S4', 'NXmx')
# --> data
data = entry.create_group('data')
data.attrs['NX_class'] = 'NXdata'
data_key = 'data'
data[data_key] = h5py.ExternalLink(self.params.cxi_file, "entry_1/data_1/data")
# --> sample
sample = entry.create_group('sample')
sample.attrs['NX_class'] = 'NXsample'
beam = sample.create_group('beam')
beam.attrs['NX_class'] = 'NXbeam'
if self.params.wavelength is None:
wavelengths = h5py.File(self.params.cxi_file, 'r')['instrument/photon_wavelength_A']
beam.create_dataset('incident_wavelength', (1,), data=np.mean(wavelengths),dtype='f8')
else:
beam.create_dataset('incident_wavelength', (1,), data=self.params.wavelength,dtype='f8') # 9150
beam['incident_wavelength'].attrs['units'] = 'angstrom'
# --> instrument
instrument = entry.create_group('instrument')
instrument.attrs['NX_class'] = 'NXinstrument'
agipd = instrument.create_group('AGIPD')
agipd.attrs['NX_class'] = 'NXdetector_group'
# Loop over all individual files and link in the top level groups.
for ind_file in individual_files:
# Open file.
with h5py.File(ind_file, 'r') as h5_infile:
# Links must be relative.
relative_link_target = os.path.relpath(path=ind_file, start=os.path.dirname(os.path.dirname(ind_file)))
# Link global parameters.
if not global_parameters:
global_parameters = True
h5_outfile["params"] = h5py.ExternalLink(relative_link_target, "params")
h5_outfile["info"] = h5py.ExternalLink(relative_link_target, "info")
h5_outfile["misc"] = h5py.ExternalLink(relative_link_target, "misc")
h5_outfile["version"] = h5py.ExternalLink(relative_link_target, "version")
for key in h5_infile['data']:
# Link in the data.
ds_path = "data/%s" % (key)
h5_outfile[ds_path] = h5py.ExternalLink(relative_link_target, ds_path)
# Reset output path.
self.output_path = self.output_path+".h5"
def write_link(self, **kwargs):
parent, builder = getargs('parent', 'builder', kwargs)
if builder.written:
return None
name = builder.name
target_builder = builder.builder
path = self.__get_path(target_builder)
# source will indicate target_builder's location
if parent.file.filename == target_builder.source:
link_obj = SoftLink(path)
elif target_builder.source is not None:
link_obj = ExternalLink(target_builder.source, path)
else:
msg = 'cannot create external link to %s' % path
raise ValueError(msg)
parent[name] = link_obj
builder.written = True
return link_obj
def link(self):
return h5py.ExternalLink(self.filename, self.path)
def _getDefaultIcon(self):
style = qt.QApplication.style()
if self.__isBroken:
icon = style.standardIcon(qt.QStyle.SP_MessageBoxCritical)
return icon
class_ = self.h5pyClass
if issubclass(class_, h5py.File):
return style.standardIcon(qt.QStyle.SP_FileIcon)
elif issubclass(class_, h5py.Group):
return style.standardIcon(qt.QStyle.SP_DirIcon)
elif issubclass(class_, h5py.SoftLink):
return style.standardIcon(qt.QStyle.SP_DirLinkIcon)
elif issubclass(class_, h5py.ExternalLink):
return style.standardIcon(qt.QStyle.SP_FileLinkIcon)
elif issubclass(class_, h5py.Dataset):
if len(self.obj.shape) < 4:
name = "item-%ddim" % len(self.obj.shape)
else:
name = "item-ndim"
if str(self.obj.dtype) == "object":
name = "item-object"
icon = icons.getQIcon(name)
return icon
return None