From e4de724046f83f9224514cd40086952e2b2d2170 Mon Sep 17 00:00:00 2001 From: Dan Porter Date: Wed, 26 Aug 2026 11:55:45 +0100 Subject: [PATCH] add links to datasets and groups ## hdfmap_class.py - add external_file to Groups and Datasets to store link locations. - add *HdfMap.find_links()* to return paths and files of links. All tests pass --- src/hdfmap/hdfmap_class.py | 33 ++++++++++++++++++++++++++++----- src/hdfmap/nexus.py | 16 +++++++++++----- tests/test_hdfmap_class.py | 12 ++++++++++++ 3 files changed, 51 insertions(+), 10 deletions(-) diff --git a/src/hdfmap/hdfmap_class.py b/src/hdfmap/hdfmap_class.py index 6625fea..eb58c44 100644 --- a/src/hdfmap/hdfmap_class.py +++ b/src/hdfmap/hdfmap_class.py @@ -31,6 +31,7 @@ class Group(typing.NamedTuple): datasets: list[str] parent: "Group | None" default: bool + external_file: str | None class Dataset(typing.NamedTuple): @@ -40,6 +41,7 @@ class Dataset(typing.NamedTuple): shape: tuple[int] attrs: dict parent: Group + external_file: str | None def generate_alt_name(hdf_dataset: h5py.Dataset) -> str | None: @@ -274,7 +276,7 @@ def _store_class(self, name, path): if path not in self.classes[name]: self.classes[name].append(path) - def _store_group(self, hdf_group: h5py.Group, path: str, name: str): + def _store_group(self, hdf_group: h5py.Group, path: str, name: str, external: str | None): parent = self.groups.get(hdf_group.parent.name, None) attrs = attrs2dict(hdf_group) @@ -287,14 +289,15 @@ def _store_group(self, hdf_group: h5py.Group, path: str, name: str): attrs=attrs, datasets=[key for key, item in hdf_group.items() if isinstance(item, h5py.Dataset)], parent=parent, - default=nx_default + default=nx_default, + external_file=external ) self._store_class(name, path) self._store_class(nx_class, path) logger.debug(f"{path} HDFGroup: {nx_class}") return nx_class - def _store_dataset(self, hdf_dataset: h5py.Dataset, hdf_path: str, name: str): + def _store_dataset(self, hdf_dataset: h5py.Dataset, hdf_path: str, name: str, external: str | None): # New: add group_name to namespace as standard, helps with names like s5/x + s4/x # this significantly increases the number of names in namespaces group = self.groups[SEP.join(hdf_path.split(SEP)[:-1])] # group is already stored @@ -311,6 +314,7 @@ def _store_dataset(self, hdf_dataset: h5py.Dataset, hdf_path: str, name: str): shape=hdf_dataset.shape, attrs=attrs2dict(hdf_dataset), parent=group, + external_file=external ) if hdf_dataset.ndim > 0: if is_image(hdf_dataset.shape): @@ -353,17 +357,18 @@ def _populate(self, hdf_group: h5py.Group, root: str = '', # New: store all paths in file, useful for checking if anything was missed, but might be slow self.all_paths.append(hdf_path) name = generate_identifier(hdf_path) + external_file = link.filename if isinstance(link, h5py.ExternalLink) else None logger.debug(f"{hdf_path}: {name}, link={repr(link)}") # Group if isinstance(obj, h5py.Group): - nx_class = self._store_group(obj, hdf_path, name) + nx_class = self._store_group(obj, hdf_path, name, external_file) if recursive and (key in groups or nx_class in groups if groups else True): self._populate(obj, hdf_path, recursive) # Dataset elif isinstance(obj, h5py.Dataset): #18 remove link omission - self._store_dataset(obj, hdf_path, name) + self._store_dataset(obj, hdf_path, name, external_file) def add_local(self, **kwargs): """Add value to the local namespace, used in eval""" @@ -743,6 +748,24 @@ def find_names(self, string: str, match_case=False) -> list[str]: return [name for name in self.combined if string in name] return [name for name in self.combined if string.lower() in name.lower()] + def find_links(self, *names_or_classes: str) -> dict[str, str]: + """ + Find datasets and groups within the hdfmap that are links to external files + :param names_or_classes: if names is given, only return links to these names + :return: dict[hdf_path, 'external_filename'] + """ + if names_or_classes: + group_paths = self.find_groups(*names_or_classes) + dataset_paths = self.find_datasets(*names_or_classes) + groups = {path: self.groups[path] for path in group_paths} + datasets = {path: self.datasets[path] for path in dataset_paths} + else: + groups = self.groups + datasets = self.datasets + group_links = {path: group.external_file for path, group in groups.items() if group.external_file} + dataset_links = {path: ds.external_file for path, ds in datasets.items() if ds.external_file} + return {**group_links, **dataset_links} + def find_attr(self, attr_name: str) -> list[str]: """ Find any dataset or group path with an attribute that contains attr_name. diff --git a/src/hdfmap/nexus.py b/src/hdfmap/nexus.py index c3e48e3..af35b5a 100644 --- a/src/hdfmap/nexus.py +++ b/src/hdfmap/nexus.py @@ -238,8 +238,8 @@ def info_nexus(self, scannables=True, image_data=True, metadata=False) -> str: out += f"" return out - def _store_group(self, hdf_group: h5py.Group, path: str, name: str): - super()._store_group(hdf_group, path, name) + def _store_group(self, hdf_group: h5py.Group, path: str, name: str, external: str | None): + super()._store_group(hdf_group, path, name, external) if NX_DEFINITION in hdf_group: definition = hdf_group[NX_DEFINITION].asstr()[()] # e.g. NXmx or NXxas self._store_class(definition, path) @@ -251,12 +251,16 @@ def _store_default_nexus_paths(self, hdf_file): nx_entry_name = default_nxentry(hdf_file) nx_entry = hdf_file[nx_entry_name] nx_entry_path = build_hdf_path(nx_entry_name) - self._store_group(nx_entry, nx_entry_path, NX_ENTRY) + nx_entry_link = hdf_file.get(nx_entry_name, getlink=True) + external_file = nx_entry_link.filename if isinstance(nx_entry_link, h5py.ExternalLink) else None + self._store_group(nx_entry, nx_entry_path, NX_ENTRY, external_file) # find the default NXdata group nx_data_name = default_nxdata(nx_entry) nx_data = nx_entry[nx_data_name] nx_data_path = build_hdf_path(nx_entry_name, nx_data_name) - self._store_group(nx_data, nx_data_path, NX_DATA) + nx_data_link = nx_entry.get(nx_data_name, getlink=True) + external_file = nx_data_link.filename if isinstance(nx_data_link, h5py.ExternalLink) else None + self._store_group(nx_data, nx_data_path, NX_DATA, external_file) axes_paths, signal_paths = find_nexus_defaults(hdf_file, nx_data_path) if axes_paths and isinstance(hdf_file.get(axes_paths[0]), h5py.Dataset): @@ -471,10 +475,12 @@ def populate(self, hdf_file: h5py.File, groups=None, default_entry_only=False): nx_entry = hdf_file.get(entry) if nx_entry is None: continue # group may be missing due to a broken link + nx_entry_link = hdf_file.get(entry, getlink=True) + external_file = nx_entry_link.filename if isinstance(nx_entry_link, h5py.ExternalLink) else None hdf_path = build_hdf_path(entry) logger.debug(f"NX Entry: {hdf_path}") self.all_paths.append(hdf_path) - self._store_group(nx_entry, hdf_path, entry) + self._store_group(nx_entry, hdf_path, entry, external_file) self._populate(nx_entry, root=hdf_path, groups=groups) # nx_entry.name can be wrong! if not self.datasets: diff --git a/tests/test_hdfmap_class.py b/tests/test_hdfmap_class.py index 5e40ef6..ad5237d 100644 --- a/tests/test_hdfmap_class.py +++ b/tests/test_hdfmap_class.py @@ -119,6 +119,18 @@ def test_save_load(hdf_map): assert save == new_save +def test_find_links(hdf_map): + links = hdf_map.find_links() + assert links == { + '/entry1/instrument/pil3_100k/data': '1049598-pilatus3_100k-files/1049598.hdf', + '/entry1/pil3_100k/data': '1049598-pilatus3_100k-files/1049598.hdf' + } + links = hdf_map.find_links('NXdetector') + assert links == { + '/entry1/instrument/pil3_100k/data': '1049598-pilatus3_100k-files/1049598.hdf', + } + + "--------------------------------------------------------" "---------------------- FILE READERS --------------------" "--------------------------------------------------------"