added HDF5 read capability. Closes #27
This commit is contained in:
@@ -7,6 +7,7 @@ import collections
|
||||
import threading
|
||||
import traceback
|
||||
import io
|
||||
import numpy
|
||||
|
||||
class DataPool(collections.abc.MutableMapping):
|
||||
"""
|
||||
@@ -107,7 +108,7 @@ class DataPool(collections.abc.MutableMapping):
|
||||
self.__registered_listeners=None
|
||||
|
||||
def write_hdf5(self,hdffile,where="/",name="data_pool", complib=None, complevel=None):
|
||||
if type(hdffile) is bytes:
|
||||
if isinstance(hdffile, (bytes, str)):
|
||||
dump_file=tables.open_file(hdffile, mode="a")
|
||||
elif isinstance(hdffile,tables.File):
|
||||
dump_file=hdffile
|
||||
@@ -156,7 +157,7 @@ class DataPool(collections.abc.MutableMapping):
|
||||
value=self.__mydict[key]
|
||||
self.__dictlock.release()
|
||||
# now write data, assuming, the object is constant during write operation
|
||||
if "write_to_hdf" in dir(value):
|
||||
if hasattr(value, "write_to_hdf"):
|
||||
try:
|
||||
value.write_to_hdf(hdffile=dump_file,
|
||||
where=dump_dir,
|
||||
@@ -170,16 +171,255 @@ class DataPool(collections.abc.MutableMapping):
|
||||
traceback.print_tb(sys.exc_info()[2], None, traceback_file)
|
||||
print("detailed traceback: %s\n"%str(e)+traceback_file.getvalue())
|
||||
traceback_file=None
|
||||
elif isinstance(value, (numpy.ndarray, int, float, str, bytes)):
|
||||
try:
|
||||
obj_to_save = value
|
||||
if isinstance(value, str):
|
||||
obj_to_save = value.encode('utf-8')
|
||||
dump_file.create_array(dump_dir,
|
||||
name=group_keyname,
|
||||
obj=obj_to_save,
|
||||
title=key)
|
||||
except Exception as e:
|
||||
print("failed to write data_pool[\"%s\"]: %s"%(key,str(e)))
|
||||
else:
|
||||
print("don't know how to store data_pool[\"%s\"]"%key)
|
||||
print("don't know how to store data_pool[\"%s\"] (type: %s)"%(key, type(value)))
|
||||
value=None
|
||||
|
||||
finally:
|
||||
dump_group=None
|
||||
if type(hdffile) is bytes:
|
||||
if isinstance(hdffile, (bytes, str)):
|
||||
dump_file.close()
|
||||
dump_file=None
|
||||
|
||||
def read_from_hdf(self, hdffile, where="/data_pool"):
|
||||
"""
|
||||
Read data from HDF5 file and populate the DataPool.
|
||||
|
||||
Parameters:
|
||||
- hdffile: HDF5 file object or filename (bytes/str)
|
||||
- where: Location within HDF5 file to read from (default: "/data_pool")
|
||||
"""
|
||||
# Handle both filename and file object
|
||||
if type(hdffile) is bytes or isinstance(hdffile, str):
|
||||
hdf_file = tables.open_file(hdffile, mode="r")
|
||||
close_file = True
|
||||
elif isinstance(hdffile, tables.File):
|
||||
hdf_file = hdffile
|
||||
close_file = False
|
||||
else:
|
||||
raise Exception("expecting hdffile or string")
|
||||
|
||||
try:
|
||||
# Navigate to the data pool group
|
||||
if where not in hdf_file:
|
||||
raise Exception(f"HDF5 location '{where}' not found")
|
||||
|
||||
data_pool_group = hdf_file.get_node(where)
|
||||
|
||||
# Recursively process the group structure
|
||||
self._read_hdf_group(data_pool_group, "")
|
||||
|
||||
finally:
|
||||
if close_file:
|
||||
hdf_file.close()
|
||||
|
||||
def _read_hdf_group(self, group, path_prefix):
|
||||
"""
|
||||
Recursively read HDF5 group and populate DataPool.
|
||||
|
||||
Parameters:
|
||||
- group: HDF5 group object
|
||||
- path_prefix: Current path in DataPool hierarchy
|
||||
"""
|
||||
# Process all nodes in this group
|
||||
for node in group:
|
||||
# Get the node name (last part of the path)
|
||||
node_name = node._v_name
|
||||
|
||||
# Try to get the original name from title
|
||||
title = node._v_title
|
||||
|
||||
if title:
|
||||
# For DAMARIS objects, the title is usually the full path
|
||||
# For intermediate groups, it's just the part name
|
||||
if "/" in title:
|
||||
full_path = title
|
||||
else:
|
||||
if path_prefix:
|
||||
full_path = f"{path_prefix}/{title}"
|
||||
else:
|
||||
full_path = title
|
||||
else:
|
||||
# Strip dir_ and dict_ prefixes from node name
|
||||
clean_name = self._clean_hdf_name(node_name)
|
||||
|
||||
# Build the full path in DataPool hierarchy
|
||||
if path_prefix:
|
||||
full_path = f"{path_prefix}/{clean_name}"
|
||||
else:
|
||||
full_path = clean_name
|
||||
|
||||
full_path = full_path.replace("//", "/")
|
||||
|
||||
if isinstance(node, (tables.Group, tables.Table)):
|
||||
# Check if this is a known DAMARIS object type
|
||||
if hasattr(node._v_attrs, 'damaris_type'):
|
||||
damaris_type = node._v_attrs.damaris_type
|
||||
|
||||
# Handle different DAMARIS object types
|
||||
if damaris_type == "Accumulation":
|
||||
from damaris.data.Accumulation import read_from_hdf
|
||||
obj = read_from_hdf(node)
|
||||
if obj is not None:
|
||||
self[full_path] = obj
|
||||
|
||||
elif damaris_type == "ADC_Result":
|
||||
from damaris.data.ADC_Result import read_from_hdf
|
||||
obj = read_from_hdf(node)
|
||||
if obj is not None:
|
||||
self[full_path] = obj
|
||||
|
||||
elif damaris_type == "MeasurementResult":
|
||||
from damaris.data.MeasurementResult import read_from_hdf
|
||||
obj = read_from_hdf(node)
|
||||
if obj is not None:
|
||||
self[full_path] = obj
|
||||
|
||||
# Skip further processing of this group since we handled it
|
||||
continue
|
||||
|
||||
# Recursively process sub-groups
|
||||
if isinstance(node, tables.Group):
|
||||
self._read_hdf_group(node, full_path)
|
||||
|
||||
elif isinstance(node, (tables.Array, tables.CArray, tables.EArray, tables.Table)):
|
||||
# Handle simple array data
|
||||
try:
|
||||
data = node.read()
|
||||
# If it's a scalar array of bytes, decode it to string
|
||||
if hasattr(data, 'decode'):
|
||||
try:
|
||||
data = data.decode('utf-8')
|
||||
except:
|
||||
pass
|
||||
elif isinstance(data, numpy.ndarray) and data.dtype.kind in ('S', 'V'):
|
||||
# For numpy byte arrays, try to decode if they are scalar or small
|
||||
if data.size == 1:
|
||||
try:
|
||||
decoded = data.item().decode('utf-8')
|
||||
data = decoded
|
||||
except:
|
||||
pass
|
||||
self[full_path] = data
|
||||
except Exception as e:
|
||||
print(f"Warning: Could not read array {full_path}: {e}")
|
||||
|
||||
# Note: Other node types (Table, etc.) would be handled here as needed
|
||||
|
||||
def _clean_hdf_name(self, name):
|
||||
"""
|
||||
Clean HDF5 node name by removing dir_ and dict_ prefixes.
|
||||
|
||||
Parameters:
|
||||
- name: HDF5 node name (may have dir_ or dict_ prefix)
|
||||
|
||||
Returns:
|
||||
- Cleaned name with prefixes removed
|
||||
"""
|
||||
if name.startswith("dir_"):
|
||||
return name[4:] # Remove "dir_" prefix
|
||||
elif name.startswith("dict_"):
|
||||
return name[5:] # Remove "dict_" prefix
|
||||
else:
|
||||
return name
|
||||
|
||||
@classmethod
|
||||
def load_hdf5(cls, filename):
|
||||
"""
|
||||
Load a DAMARIS HDF5 file and return DataPool and metadata.
|
||||
|
||||
Parameters:
|
||||
- filename: Path to HDF5 file
|
||||
|
||||
Returns:
|
||||
- tuple: (DataPool instance, metadata dict)
|
||||
|
||||
metadata dict contains:
|
||||
- 'scripts': dict with experiment scripts and spool info
|
||||
- 'log': log data if available
|
||||
- 'timeline': timeline data if available
|
||||
"""
|
||||
# Validate file
|
||||
if not tables.is_pytables_file(filename):
|
||||
raise Exception(f"File {filename} is not a valid PyTables HDF5 file")
|
||||
|
||||
# Create DataPool and load data
|
||||
data_pool = cls()
|
||||
|
||||
try:
|
||||
hdf_file = tables.open_file(filename, mode="r")
|
||||
|
||||
# Load main data pool
|
||||
if "/data_pool" in hdf_file:
|
||||
data_pool.read_from_hdf(hdf_file, where="/data_pool")
|
||||
|
||||
# Extract metadata
|
||||
metadata = {}
|
||||
|
||||
# Load scripts
|
||||
if "/scripts" in hdf_file:
|
||||
scripts_group = hdf_file.get_node("/scripts")
|
||||
scripts_data = {}
|
||||
# Use _v_children to get node names safely
|
||||
if hasattr(scripts_group, '_v_children'):
|
||||
for node_name, node in scripts_group._v_children.items():
|
||||
try:
|
||||
if hasattr(node, 'read'):
|
||||
data = node.read()
|
||||
# Handle numpy string arrays and other types
|
||||
if hasattr(data, 'decode'):
|
||||
# bytes or numpy bytes
|
||||
scripts_data[node_name] = data.decode('utf-8', errors='replace')
|
||||
elif hasattr(data, 'tolist'):
|
||||
# numpy array
|
||||
scripts_data[node_name] = str(data.tolist())
|
||||
elif isinstance(data, (str, bytes)):
|
||||
# string or bytes
|
||||
if isinstance(data, bytes):
|
||||
scripts_data[node_name] = data.decode('utf-8', errors='replace')
|
||||
else:
|
||||
scripts_data[node_name] = data
|
||||
else:
|
||||
scripts_data[node_name] = str(data)
|
||||
except Exception as e:
|
||||
print(f"Warning: Could not read script {node_name}: {e}")
|
||||
metadata['scripts'] = scripts_data
|
||||
|
||||
# Load log
|
||||
if "/log" in hdf_file:
|
||||
try:
|
||||
log_node = hdf_file.get_node("/log")
|
||||
metadata['log'] = log_node.read()
|
||||
except Exception as e:
|
||||
print(f"Warning: Could not read log: {e}")
|
||||
|
||||
# Load timeline
|
||||
if "/timeline" in hdf_file:
|
||||
try:
|
||||
timeline_node = hdf_file.get_node("/timeline")
|
||||
metadata['timeline'] = timeline_node.read()
|
||||
except Exception as e:
|
||||
print(f"Warning: Could not read timeline: {e}")
|
||||
|
||||
return data_pool, metadata
|
||||
|
||||
except Exception as e:
|
||||
raise Exception(f"Failed to load HDF5 file {filename}: {e}")
|
||||
finally:
|
||||
if 'hdf_file' in locals():
|
||||
hdf_file.close()
|
||||
|
||||
def register_listener(self, listening_function):
|
||||
self.__registered_listeners.append(listening_function)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user