# data pool collects data from data handling script # provides data to experiment script and display import sys import tables import collections import threading import traceback import io import numpy class DataPool(collections.abc.MutableMapping): """ dictionary with sending change events """ # supports tranlation from dictionary keys to pytables hdf node names # taken from: Python Ref Manual Section 2.3: Identifiers and keywords # things are always prefixed by "dir_" or "dict_" translation_table="" for i in range(256): c=chr(i) if (c>="a" and c<="z") or \ (c>="A" and c<="Z") or \ (c>="0" and c<="9"): translation_table+=c else: translation_table+="_" class Event: access=0 updated_value=1 new_key=2 deleted_key=3 destroy=4 def __init__(self, what, subject="", origin=None): self.what=what self.subject=subject self.origin=origin def __repr__(self): return ""%(self.origin, self.what,self.subject) def copy(self): return DataPool.Event(self.what+0, self.subject+"", self.origin) def __init__(self): self.__mydict={} self.__dictlock=threading.Lock() self.__registered_listeners=[] def __getitem__(self, name): try: self.__dictlock.acquire() return self.__mydict[name] finally: self.__dictlock.release() def __setitem__(self, name, value): try: self.__dictlock.acquire() if name in self.__mydict: e=DataPool.Event(DataPool.Event.updated_value,name,self) else: e=DataPool.Event(DataPool.Event.new_key, name,self) self.__mydict[name]=value finally: self.__dictlock.release() self.__send_event(e) def __delitem__(self, name): try: self.__dictlock.acquire() del self.__mydict[name] finally: self.__dictlock.release() self.__send_event(DataPool.Event(DataPool.Event.deleted_key,name,self)) def __iter__(self): try: self.__dictlock.acquire() return iter(self.__mydict) finally: self.__dictlock.release() def __len__(self): try: self.__dictlock.acquire() return len(self.__mydict) finally: self.__dictlock.release() def keys(self): try: self.__dictlock.acquire() return list(self.__mydict.keys()) finally: self.__dictlock.release() def __send_event(self, _event): for listeners in self.__registered_listeners: listeners(_event.copy()) def __del__(self): self.__send_event(DataPool.Event(DataPool.Event.destroy)) self.__registered_listeners=None def write_hdf5(self,hdffile,where="/",name="data_pool", complib=None, complevel=None): """ Writes the content of a DataPool object to an HDF5 file using the `tables` library. This method serializes the entire content of the data pool into an HDF5 file, creating appropriate groups and arrays based on the keys and values stored in the data pool. Nested storage paths are supported, and group or array names are converted into valid HDF5 names. Paths are prepended with "dir_", results with "dict_". If duplicate names arise, extensions (like `_001`, `_002`) are appended to ensure uniqueness. Parameters: hdffile : str | bytes | tables.File Path to the HDF5 file or an already open `tables.File` object where the data should be written. Strings or bytes are interpreted as file paths. where : str, optional The root HDF5 group path under which the data pool will be stored. Defaults to "/". name : str, optional The name of the root HDF5 group under which the data pool is saved. Defaults to "data_pool". complib : str, optional Compression library to be used when creating HDF5 arrays. Optional, defaults to None, meaning no compression is applied. complevel : int, optional Compression level to use when applying compression. Has no effect if `complib` is set to None. Raises: Exception If `hdffile` is neither a string nor a valid `tables.File` object. Notes: - This method assumes that the content of the data pool does not change during the write operation. - Keys in the data pool starting with "__" (double underscores) are skipped. - If a key contains nested paths (e.g., "group1/subgroup2/item"), intermediate groups are created as necessary. - Group or array names are converted to valid HDF5 names, and name conflicts are resolved by appending extensions (e.g., "_001"). - Errors during the writing process are logged to the console, and traceback information is displayed for debugging purposes. """ if isinstance(hdffile, (bytes, str)): dump_file=tables.open_file(hdffile, mode="a") elif isinstance(hdffile,tables.File): dump_file=hdffile else: raise Exception("expecting hdffile or string") dump_group=dump_file.create_group(where, name, "DAMARIS data pool") self.__dictlock.acquire() dict_keys=list(self.__mydict.keys()) self.__dictlock.release() try: for key in dict_keys: if key.startswith("__"): continue dump_dir=dump_group # walk along the given path and create groups if necessary namelist = key.split("/") for part in namelist[:-1]: dir_part="dir_"+str(part).translate(DataPool.translation_table) if dir_part not in dump_dir: dump_dir=dump_file.create_group(dump_dir,name=dir_part,title=part) else: if dump_dir._v_children[dir_part]._v_title==part: dump_dir=dump_dir._v_children[dir_part] else: extension_count=0 while dir_part+"_%03d"%extension_count in dump_dir: extension_count+=1 dump_dir=dump_file.create_group(dump_dir, name=dir_part+"_%03d"%extension_count, title=part) # convert last part of key to a valid name group_keyname="dict_"+str(namelist[-1]).translate(DataPool.translation_table) # avoid double names by adding number extension if group_keyname in dump_dir: extension_count=0 while group_keyname+"_%03d"%extension_count in dump_dir: extension_count+=1 group_keyname+="_%03d"%extension_count self.__dictlock.acquire() if key not in self.__mydict: # outdated ... self.__dictlock.release() continue value=self.__mydict[key] self.__dictlock.release() # now write data, assuming, the object is constant during write operation if hasattr(value, "write_to_hdf"): try: value.write_to_hdf(hdffile=dump_file, where=dump_dir, name=group_keyname, title=key, complib=complib, complevel=complevel) except Exception as e: print("failed to write data_pool[\"%s\"]: %s"%(key,str(e))) traceback_file=io.StringIO() 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\"] (type: %s)"%(key, type(value))) value=None finally: dump_group=None if isinstance(hdffile, (bytes, str)): dump_file.close() dump_file=None def read_from_hdf(self, hdffile, where="/data_pool"): """ Read specified 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 complete DAMARIS HDF5 file and return DataPool and metadata, including scripts. 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) def unregister_listener(self, listening_function): if listening_function in self.__registered_listeners: self.__registered_listeners.remove(listening_function)