80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
import os
|
|
import sys
|
|
#root_dir = os.path.abspath(os.curdir)
|
|
#sys.path.append(root_dir)
|
|
|
|
from instruments.readers.xps_ibw_reader import read_xps_ibw_file_as_dict
|
|
from instruments.readers.g5505_text_reader import read_txt_files_as_dict
|
|
|
|
|
|
file_extensions = ['.ibw','.txt','.dat','.h5','.TXT','.csv']
|
|
|
|
# Define the instruments directory (modify this as needed or set to None)
|
|
default_instruments_dir = None # or provide an absolute path
|
|
|
|
file_readers = {
|
|
'ibw': lambda a1: read_xps_ibw_file_as_dict(a1),
|
|
'txt': lambda a1: read_txt_files_as_dict(a1, instruments_dir=default_instruments_dir, work_with_copy=False),
|
|
'TXT': lambda a1: read_txt_files_as_dict(a1, instruments_dir=default_instruments_dir, work_with_copy=False),
|
|
'dat': lambda a1: read_txt_files_as_dict(a1, instruments_dir=default_instruments_dir, work_with_copy=False),
|
|
#'ACSM_TOFWARE_txt': lambda a1: read_txt_files_as_dict(a1, instruments_dir=default_instruments_dir, work_with_copy=False),
|
|
#'ACSM_TOFWARE_csv': lambda a1: read_txt_files_as_dict(a1, instruments_dir=default_instruments_dir, work_with_copy=False)
|
|
}
|
|
|
|
# Add new "instrument reader (Data flagging app data)"
|
|
|
|
from instruments.readers.acsm_tofware_reader import read_acsm_files_as_dict
|
|
file_extensions.append('.txt')
|
|
file_readers.update({'ACSM_TOFWARE_txt' : lambda x: read_acsm_files_as_dict(x, instruments_dir=default_instruments_dir, work_with_copy=False)})
|
|
|
|
file_extensions.append('.csv')
|
|
file_readers.update({'ACSM_TOFWARE_csv' : lambda x: read_acsm_files_as_dict(x, instruments_dir=default_instruments_dir, work_with_copy=False)})
|
|
|
|
from instruments.readers.flag_reader import read_jsonflag_as_dict
|
|
file_extensions.append('.json')
|
|
file_readers.update({'ACSM_TOFWARE_flags_json' : lambda x: read_jsonflag_as_dict(x)})
|
|
|
|
def compute_filereader_key_from_path(hdf5_file_path):
|
|
"""Constructs the key 'instrumentname_ext' based on hdf5_file_path, structured as
|
|
/instrumentname/to/filename.ext, which access the file reader that should be used to read such a file.
|
|
|
|
Parameters
|
|
----------
|
|
hdf5_file_path : str
|
|
_description_
|
|
|
|
Returns
|
|
-------
|
|
_type_
|
|
_description_
|
|
"""
|
|
|
|
parts = hdf5_file_path.strip('/').split('/')
|
|
|
|
# Extract the filename and its extension
|
|
filename, file_extension = os.path.splitext(parts[-1])
|
|
|
|
# Extract the first directory directly under the root directory '/' in the hdf5 file
|
|
subfolder_name = parts[0] if len(parts) > 1 else ""
|
|
|
|
# Remove leading dot from the file extension
|
|
file_extension = file_extension.lstrip('.')
|
|
|
|
# Construct the resulting string
|
|
full_string = f"{subfolder_name}_{file_extension}"
|
|
|
|
return full_string, file_extension
|
|
|
|
def select_file_reader(path):
|
|
full_string, extension = compute_filereader_key_from_path(path)
|
|
|
|
# First, try to match the full string
|
|
if full_string in file_readers:
|
|
return file_readers[full_string]
|
|
|
|
# If no match, try to match the reader using only the extension
|
|
if extension in file_readers:
|
|
return file_readers[extension]
|
|
|
|
# Default case if no reader is found
|
|
return lambda x : None |