IH5 skeletons and stubs (low-level structures used by IH5MFRecord).
A skeleton is documenting the tree structure of a HDF5-like container,
ignoring the actual data content (attribute values and datasets).
This can be used to implement manifest file and support "patching in thin air",
i.e. without having the actual container.
Bases: BaseModel
Source code in src/metador_core/ih5/skeleton.py
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76 | class IH5Skeleton(BaseModel):
__root__: Dict[str, SkeletonNodeInfo]
def with_patch_index(self, idx: int):
"""Copy of skeletonwith patch index of all nodes and attributes modified."""
ret = self.copy()
ret.__root__ = dict(self.__root__)
for k, v in ret.__root__.items():
ret.__root__[k] = v.with_patch_index(idx)
return ret
@classmethod
def for_record(cls, rec: IH5Record):
"""Return mapping from all paths in a IH5 record to their type.
The attributes are represented as special paths with the shape `a/b/.../n@attr`,
pointing to the attribute named `attr` at the path `a/b/.../n`.
First component is a H5Type enum value,
Second component is more detailed type for attribute values and `IH5Dataset`s.
"""
skel = {"/": SkeletonNodeInfo.for_node(rec["/"])}
def add_paths(_, node):
skel[node.name] = SkeletonNodeInfo.for_node(node)
rec.visititems(add_paths)
return cls(__root__=skel)
|
with_patch_index(idx: int)
Copy of skeletonwith patch index of all nodes and attributes modified.
Source code in src/metador_core/ih5/skeleton.py
| def with_patch_index(self, idx: int):
"""Copy of skeletonwith patch index of all nodes and attributes modified."""
ret = self.copy()
ret.__root__ = dict(self.__root__)
for k, v in ret.__root__.items():
ret.__root__[k] = v.with_patch_index(idx)
return ret
|
for_record(rec: IH5Record)
Return mapping from all paths in a IH5 record to their type.
The attributes are represented as special paths with the shape a/b/.../n@attr
,
pointing to the attribute named attr
at the path a/b/.../n
.
First component is a H5Type enum value,
Second component is more detailed type for attribute values and IH5Dataset
s.
Source code in src/metador_core/ih5/skeleton.py
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76 | @classmethod
def for_record(cls, rec: IH5Record):
"""Return mapping from all paths in a IH5 record to their type.
The attributes are represented as special paths with the shape `a/b/.../n@attr`,
pointing to the attribute named `attr` at the path `a/b/.../n`.
First component is a H5Type enum value,
Second component is more detailed type for attribute values and `IH5Dataset`s.
"""
skel = {"/": SkeletonNodeInfo.for_node(rec["/"])}
def add_paths(_, node):
skel[node.name] = SkeletonNodeInfo.for_node(node)
rec.visititems(add_paths)
return cls(__root__=skel)
|
init_stub_skeleton(ds: IH5Record, skel: IH5Skeleton)
Fill a passed fresh container with stub structure based on a skeleton.
Source code in src/metador_core/ih5/skeleton.py
83
84
85
86
87
88
89
90
91
92
93
94
95
96 | def init_stub_skeleton(ds: IH5Record, skel: IH5Skeleton):
"""Fill a passed fresh container with stub structure based on a skeleton."""
if len(ds) or len(ds.attrs):
raise ValueError("Container not empty, cannot initialize stub structure here!")
for k, v in skel.__root__.items():
if v.node_type == H5Type.group:
if k not in ds:
ds.create_group(k)
elif v.node_type == H5Type.dataset:
ds[k] = h5py.Empty(None)
for a in v.attrs.keys():
ds[k].attrs[a] = h5py.Empty(None)
|
init_stub_base(
target: IH5Record,
src_ub: IH5UserBlock,
src_skel: IH5Skeleton,
)
Prepare a stub base container, given empty target, source user block and skeleton.
Patches on top of this container will work with the original container.
Source code in src/metador_core/ih5/skeleton.py
99
100
101
102
103
104
105
106 | def init_stub_base(target: IH5Record, src_ub: IH5UserBlock, src_skel: IH5Skeleton):
"""Prepare a stub base container, given empty target, source user block and skeleton.
Patches on top of this container will work with the original container.
"""
init_stub_skeleton(target, src_skel)
# mark as base container
target._set_ublock(-1, src_ub.copy(update={"prev_patch": None}))
|