80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
import sys
|
|
import os
|
|
|
|
try:
|
|
thisFilePath = os.path.abspath(__file__)
|
|
except NameError:
|
|
print("Error: __file__ is not available. Ensure the script is being run from a file.")
|
|
print("[Notice] Path to DIMA package may not be resolved properly.")
|
|
thisFilePath = os.getcwd() # Use current directory or specify a default
|
|
|
|
dimaPath = os.path.normpath(os.path.join(thisFilePath, "..",'..','..')) # Move up to project root
|
|
|
|
if dimaPath not in sys.path: # Avoid duplicate entries
|
|
sys.path.insert(0,dimaPath)
|
|
|
|
import pandas as pd
|
|
import collections
|
|
import yaml
|
|
import h5py
|
|
import argparse
|
|
import logging
|
|
import utils.g5505_utils as utils
|
|
import src.hdf5_ops as hdf5_ops
|
|
import instruments.filereader_registry as filereader_registry
|
|
|
|
|
|
|
|
|
|
def hdf5_file_reader(dest_file_obj_or_path, src_file_path=None, dest_group_name=None, work_with_copy: bool = True):
|
|
"""
|
|
Reads an HDF5 file and copies its contents to a destination group.
|
|
If an HDF5 file object is provided, it skips reading from a file path.
|
|
"""
|
|
# Determine if dest_file_obj_or_path is a file path or an HDF5 file object
|
|
if isinstance(dest_file_obj_or_path, h5py.File):
|
|
dest_file_obj = dest_file_obj_or_path
|
|
else:
|
|
dest_file_obj = h5py.File(dest_file_obj_or_path, mode='r+', track_order=True)
|
|
|
|
try:
|
|
if work_with_copy:
|
|
tmp_src_file_path = utils.make_file_copy(src_file_path)
|
|
else:
|
|
tmp_src_file_path = src_file_path
|
|
|
|
# Open source HDF5 file
|
|
with h5py.File(tmp_src_file_path, 'r') as src_file:
|
|
dest_file_obj.copy(source=src_file['/'], dest=dest_group_name)
|
|
|
|
# Remove temporary file if created
|
|
if 'tmp_files' in tmp_src_file_path:
|
|
os.remove(tmp_src_file_path)
|
|
|
|
print(f'Completed transfer for /{dest_group_name}')
|
|
|
|
finally:
|
|
if not isinstance(dest_file_obj_or_path, h5py.File):
|
|
dest_file_obj.close()
|
|
|
|
return 0
|
|
|
|
if __name__ == "__main__":
|
|
|
|
# Set up argument parsing
|
|
parser = argparse.ArgumentParser(description="Data ingestion process to HDF5 files.")
|
|
parser.add_argument('dst_file_path', type=str, help="Path to the target HDF5 file.")
|
|
parser.add_argument('src_file_path', type=str, help="Path to source HDF5 file to be saved to target HDF5 file.")
|
|
parser.add_argument('dst_group_name', type=str, help="Group name '/instFolder/[category]/fileName' in the target HDF5 file.")
|
|
|
|
args = parser.parse_args()
|
|
|
|
dst_file_path = args.dst_file_path
|
|
src_file_path = args.src_file_path
|
|
dst_group_name = args.dst_group_name
|
|
default_mode = 'r+'
|
|
|
|
status = hdf5_file_reader(dst_file_path, src_file_path, dst_group_name)
|
|
|
|
print('Return status: {status}')
|