mirror of
https://github.com/p2p-ld/nwb-linkml.git
synced 2024-11-10 00:34:29 +00:00
remove unused code
This commit is contained in:
parent
e331b8756e
commit
23a5412854
1 changed files with 0 additions and 74 deletions
|
@ -91,7 +91,6 @@ class HDF5IO():
|
||||||
else:
|
else:
|
||||||
return queue.completed[path].result
|
return queue.completed[path].result
|
||||||
|
|
||||||
|
|
||||||
def make_provider(self) -> SchemaProvider:
|
def make_provider(self) -> SchemaProvider:
|
||||||
"""
|
"""
|
||||||
Create a :class:`~.providers.schema.SchemaProvider` by
|
Create a :class:`~.providers.schema.SchemaProvider` by
|
||||||
|
@ -120,56 +119,6 @@ class HDF5IO():
|
||||||
return provider
|
return provider
|
||||||
|
|
||||||
|
|
||||||
def process_group(self, group:h5py.Group|h5py.File) -> dict | list:
|
|
||||||
attrs = dict(group.attrs)
|
|
||||||
|
|
||||||
# how to process the group?
|
|
||||||
# --------------------------------------------------
|
|
||||||
# list-like
|
|
||||||
# --------------------------------------------------
|
|
||||||
# a list of data classes
|
|
||||||
if 'neurodata_type' not in attrs and \
|
|
||||||
all([isinstance(v, h5py.Group) for v in group.values()]) and \
|
|
||||||
all(['neurodata_type' in v.attrs for v in group.values()]):
|
|
||||||
|
|
||||||
return [self.process_group(v) for v in group.values()]
|
|
||||||
|
|
||||||
# --------------------------------------------------
|
|
||||||
# dict-like
|
|
||||||
# --------------------------------------------------
|
|
||||||
|
|
||||||
res = {}
|
|
||||||
|
|
||||||
|
|
||||||
for key, val in group.items():
|
|
||||||
if isinstance(val, h5py.Group):
|
|
||||||
res[key] = self.process_group(val)
|
|
||||||
elif isinstance(val, h5py.Dataset):
|
|
||||||
res[key] = self.process_dataset(val)
|
|
||||||
return res
|
|
||||||
|
|
||||||
def process_dataset(self, data: h5py.Dataset) -> dict | list:
|
|
||||||
if len(data.shape) == 1:
|
|
||||||
return list(data[:])
|
|
||||||
|
|
||||||
|
|
||||||
def finish_root_hackily(queue: ReadQueue) -> dict:
|
|
||||||
root = {'name': 'root'}
|
|
||||||
for k, v in queue.queue.items():
|
|
||||||
if isinstance(v.result, dict):
|
|
||||||
res_dict = {}
|
|
||||||
for inner_k, inner_v in v.result.items():
|
|
||||||
if isinstance(inner_v, HDF5_Path):
|
|
||||||
inner_res = queue.completed.get(inner_v)
|
|
||||||
if inner_res is not None:
|
|
||||||
res_dict[inner_k] = inner_res.result
|
|
||||||
else:
|
|
||||||
res_dict[inner_k] = inner_v
|
|
||||||
root[res_dict['name']] = res_dict
|
|
||||||
else:
|
|
||||||
root[v.path.split('/')[-1]] = v.result
|
|
||||||
return root
|
|
||||||
|
|
||||||
def read_specs_as_dicts(group: h5py.Group) -> dict:
|
def read_specs_as_dicts(group: h5py.Group) -> dict:
|
||||||
"""
|
"""
|
||||||
Utility function to iterate through the `/specifications` group and
|
Utility function to iterate through the `/specifications` group and
|
||||||
|
@ -197,18 +146,6 @@ def read_specs_as_dicts(group: h5py.Group) -> dict:
|
||||||
return spec_dict
|
return spec_dict
|
||||||
|
|
||||||
|
|
||||||
def get_model(cls: h5py.Group | h5py.Dataset) -> Type[BaseModel]:
|
|
||||||
attrs = cls.attrs
|
|
||||||
ns = attrs.get('namespace')
|
|
||||||
model_name = attrs.get('neurodata_type')
|
|
||||||
|
|
||||||
try:
|
|
||||||
return SchemaProvider().get_class(ns, model_name)
|
|
||||||
except:
|
|
||||||
# try to get parent class
|
|
||||||
mod = get_model(cls.parent)
|
|
||||||
return mod.model_fields[cls.name.split('/')[-1]].annotation
|
|
||||||
|
|
||||||
def find_references(h5f: h5py.File, path: str) -> List[str]:
|
def find_references(h5f: h5py.File, path: str) -> List[str]:
|
||||||
"""
|
"""
|
||||||
Find all objects that make a reference to a given object in
|
Find all objects that make a reference to a given object in
|
||||||
|
@ -264,8 +201,6 @@ def find_references(h5f: h5py.File, path: str) -> List[str]:
|
||||||
return references
|
return references
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def truncate_file(source: Path, target: Optional[Path] = None, n:int=10) -> Path:
|
def truncate_file(source: Path, target: Optional[Path] = None, n:int=10) -> Path:
|
||||||
"""
|
"""
|
||||||
Create a truncated HDF5 file where only the first few samples are kept.
|
Create a truncated HDF5 file where only the first few samples are kept.
|
||||||
|
@ -350,12 +285,3 @@ def truncate_file(source: Path, target: Optional[Path] = None, n:int=10) -> Path
|
||||||
return target
|
return target
|
||||||
|
|
||||||
|
|
||||||
def submodel_by_path(model: BaseModel, path:str) -> Type[BaseModel | dict | list]:
|
|
||||||
"""
|
|
||||||
Given a pydantic model and an absolute HDF5 path, get the type annotation
|
|
||||||
"""
|
|
||||||
parts = path.split('/')
|
|
||||||
for part in parts:
|
|
||||||
ann = model.model_fields[part].annotation
|
|
||||||
|
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue