"""INI scaffolding helpers for new datasets."""
import configparser
import datetime
import os
import platform
import psutil
import sys
from coseda.logging_utils import log_print, log_start
from coseda.io import handle_input, read_config
from coseda.importers.gatan_metareader import extract_info_from_gatan_metadata
from coseda.importers.import_velox import extract_instrument_data
[docs]
def create_insfiles(input_path, input_type, target=None):
"""Dispatch creation of INI scaffolds based on raw input type (DM4 vs EMD)."""
if input_type == 'dm4_folder':
create_insfiles_gatan(input_path)
elif input_type == 'emd' or input_type == '.emd':
result = create_insfiles_emd(input_path, target)
return result
else:
create_insfiles_emd(input_path, target)
[docs]
def create_insfiles_emd(input_path, target):
"""
Create per-file INI scaffolding for .emd inputs (folder, list, or file).
Builds timestamped run folders, initializes plots/log files, and fills
AcquisitionDetails/Paths from Velox metadata when available.
"""
'''
# Check if input_path is a string (folder path) or a list (specific file paths)
if isinstance(input_path, str):
# List all .emd files in the input folder
originalfiles = [file for file in os.listdir(input_path) if file.endswith(input_type)]
originalfiles = [os.path.join(input_path, file) for file in originalfiles]
elif isinstance(input_path, list):
# Filter out only .emd files from the list
originalfiles = [file for file in input_path if file.endswith(input_type)]
log_print(originalfiles)
else:
log_print("Invalid input. Please provide a folder path or a list of file paths.")
return False'''
input_type = 'emd'
if isinstance(input_path, str):
if os.path.isdir(input_path):
# Input is a folder path, collect all .emd files in the folder
originalfiles = [
os.path.join(input_path, file)
for file in os.listdir(input_path)
if file.endswith(f'.{input_type}')
]
elif os.path.isfile(input_path) and input_path.endswith(f'.{input_type}'):
# Input is a single .emd file
originalfiles = [input_path]
else:
raise ValueError(
f"The input must be a valid .{input_type} file path or a folder containing .{input_type} files."
)
elif isinstance(input_path, list):
# Input is a list of .emd files, validate all paths
originalfiles = [
file for file in input_path
if file.endswith(f'.{input_type}') and os.path.isfile(file)
]
else:
raise ValueError("Invalid input. Please provide a folder path, a single file path, or a list of file paths.")
if not originalfiles:
raise ValueError(f"No .{input_type} files found in the provided input.")
for file_path in originalfiles:
if ' ' in os.path.basename(file_path):
log_print(
"You're trying to use files with spaces in their names. Considering you might need to work on "
"this data in a Unix system, this is stupid and I refuse to do this! Remove the spaces and try "
"again."
)
return False
newfiles = []
for originalfile_path in originalfiles:
# Create the configfile and folder structure
directory = os.path.dirname(originalfile_path)
filename, _ = os.path.splitext(os.path.basename(originalfile_path))
timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
# Create output project folder for HDF5 and log
outputfolder = f"{filename}_run_{timestamp}"
outputfolder_path = os.path.join(directory, outputfolder)
os.makedirs(outputfolder_path, exist_ok=True)
os.makedirs(os.path.join(outputfolder_path, "plots"), exist_ok=True)
# Define INS file inside the project folder
configfile = f"{filename}_run_{timestamp}.ini"
configfile_path = os.path.join(outputfolder_path, configfile)
framepath = 'entry/data/images'
log_print('configuration file created for ' + originalfile_path)
instrument_data = extract_instrument_data(input_type, originalfile_path)
with open(configfile_path, 'w') as configfile_f:
pass
log_print(f"configuration will be written to {configfile_path}")
newfiles.append(configfile_path)
log_print(f"results will be saved in {outputfolder_path}")
# Create log file
logfile = f"{filename}_run_{timestamp}.log"
logfile_path = os.path.join(outputfolder_path, logfile)
# Open log file, add some structure and basic information
config = read_config(configfile_path)
if not config.has_section('Paths'):
config.add_section('Paths')
config.set('Paths', 'framepath', framepath)
config.set('Paths', 'originalfile', originalfile_path)
log_start(logfile_path, f'original file is {originalfile_path}')
if instrument_data is not None:
if not config.has_section('AcquisitionDetails'):
config.add_section('AcquisitionDetails')
config.set('AcquisitionDetails', 'acquisition_start', str(instrument_data.get('AcquisitionStart', '')))
config.set('AcquisitionDetails', 'acquisition_end', str(instrument_data.get('AcquisitionEnd', '')))
config.set('AcquisitionDetails', 'instrument_manufacturer', str(instrument_data.get('InstrumentManufacturer', '')))
config.set('AcquisitionDetails', 'instrument_model', str(instrument_data.get('InstrumentModel', '')))
config.set('AcquisitionDetails', 'instrument_id', str(instrument_data.get('InstrumentId', '')))
config.set('AcquisitionDetails', 'detector_used', str(instrument_data.get('DetectorUsed', '')))
config.set('AcquisitionDetails', 'acceleration_voltage', str(instrument_data.get('AccelerationVoltage', '')))
config.set('AcquisitionDetails', 'camera_length', str(instrument_data.get('CameraLength', '')))
config.set('AcquisitionDetails', 'pixel_width', str(instrument_data.get('PixelWidth', '')))
config.set('AcquisitionDetails', 'pixel_height', str(instrument_data.get('PixelHeight', '')))
config.set('AcquisitionDetails', 'binning_width', str(instrument_data.get('BinningWidth', '')))
config.set('AcquisitionDetails', 'binning_height', str(instrument_data.get('BinningHeight', '')))
resolution = instrument_data.get('Resolution', ('', ''))
config.set('AcquisitionDetails', 'resolution_width', str(resolution[0]))
config.set('AcquisitionDetails', 'resolution_height', str(resolution[1]))
config.set('AcquisitionDetails', 'exposure_time', str(instrument_data.get('ExposureTime', '')))
config.set('AcquisitionDetails', 'pixel_unit', str(instrument_data.get('PixelUnit', '')))
config.set('AcquisitionDetails', 'pixel_offset_x', str(instrument_data.get('PixelOffsetX', '')))
config.set('AcquisitionDetails', 'pixel_offset_y', str(instrument_data.get('PixelOffsetY', '')))
log_print('successfully extracted acquisition details from original file')
if not config.has_section('Parameters'):
config.add_section('Parameters')
if not config.has_section('Output'):
config.add_section('Output')
with open(configfile_path, 'w') as cfgfile:
config.write(cfgfile)
log_start(logfile_path, '.ini file created')
log_start(logfile_path, f'system: {platform.system()} {platform.architecture()}, version {platform.version()}')
log_start(
logfile_path,
f'hardware: {psutil.cpu_count(logical=False)} physical cores, '
f'{psutil.cpu_count(logical=True)} logical cores, '
f'{round(psutil.virtual_memory().total / (1024 ** 3))} GB total memory'
)
log_start(logfile_path, f'python version {sys.version}')
log_start(logfile_path, f'data acquisition details: {str(instrument_data)}')
return newfiles
[docs]
def create_plain_insfile(hdf5_file_path, output_folder, acquisition_details, parameters):
"""
Create a plain .ins file for an existing HDF5 file with manually provided inputs.
Parameters:
- hdf5_file_path (str): Path to the existing HDF5 file.
- output_folder (str): Path to the folder where the .ins file will be saved.
- acquisition_details (dict): Dictionary containing the acquisition details.
- parameters (dict): Dictionary containing the parameters.
Returns:
- str: Path to the created .ins file.
"""
if not os.path.isfile(hdf5_file_path):
raise ValueError(
f"The provided HDF5 file path '{hdf5_file_path}' does not exist or is not a file."
)
if not os.path.isdir(output_folder):
os.makedirs(output_folder)
# Extract the filename without extension
filename = os.path.splitext(os.path.basename(hdf5_file_path))[0]
timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
# Create output folder structure
outputfolder = f"{filename}_run_{timestamp}"
outputfolder_path = os.path.join(output_folder, outputfolder)
os.makedirs(outputfolder_path)
os.makedirs(os.path.join(outputfolder_path, "plots"))
# Define the .ins file path inside the project folder
insfile_path = os.path.join(outputfolder_path, f"{filename}_run_{timestamp}.ini")
# Create the .ins file
config = configparser.ConfigParser()
# Add the Paths section
config['Paths'] = {
'h5file': f"{filename}_run_{timestamp}.h5",
'originalfile_path': hdf5_file_path,
# Store the absolute path so downstream helpers (which already know the
# INI directory) don't append the folder name a second time.
'outputfolder': outputfolder_path
}
# Record the INI path in the INI
config.set('Paths', 'configfile_path', insfile_path)
# Add the provided acquisition details
if not config.has_section('AcquisitionDetails'):
config.add_section('AcquisitionDetails')
for key, value in acquisition_details.items():
config.set('AcquisitionDetails', key, str(value))
# Add the provided parameters
if not config.has_section('Parameters'):
config.add_section('Parameters')
for key, value in parameters.items():
config.set('Parameters', key, str(value))
# Create log file
logfile = f"{filename}_run_{timestamp}.log"
logfile_path = os.path.join(outputfolder_path, logfile)
# Add log file path to the config
config.set('Paths', 'logfile', logfile)
config.set('Paths', 'framepath', 'entry/data/images')
# Write the .ins file
with open(insfile_path, 'w') as cfgfile:
config.write(cfgfile)
# Log the creation of the .ins file
with open(logfile_path, 'w') as log:
log.write(f".ins file created at: {insfile_path}\n")
log.write(f"System: {platform.system()} {platform.architecture()}, version {platform.version()}\n")
log.write(
f"Hardware: {psutil.cpu_count(logical=False)} physical cores, "
f"{psutil.cpu_count(logical=True)} logical cores, "
f"{round(psutil.virtual_memory().total / (1024 ** 3))} GB total memory\n"
)
log.write(f"Python version: {sys.version}\n")
log.write(f"Acquisition details: {acquisition_details}\n")
log.write(f"Parameters: {parameters}\n")
log_print(f"Plain .ins file created at: {insfile_path}")
return insfile_path
[docs]
def create_insfiles_gatan(input_path, target=None):
filename = os.path.basename(input_path)
directory = os.path.dirname(input_path)
timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
# Create project folder
outputfolder = f"{filename}_run_{timestamp}"
if target is None:
outputfolder_path = os.path.join(directory, outputfolder)
else:
outputfolder_path = os.path.join(target, outputfolder)
os.makedirs(outputfolder_path, exist_ok=True)
os.makedirs(os.path.join(outputfolder_path, "plots"), exist_ok=True)
# Define INS file inside the project folder
configfile = f"{filename}_run_{timestamp}.ini"
configfile_path = os.path.join(outputfolder_path, configfile)
framepath = 'entry/data/images'
newfiles = configfile_path
log_print('configuration file created for ' + input_path)
with open(configfile_path, 'w') as configfile:
pass
log_print(f"configuration will be written to {configfile_path}")
log_print(f"results will be saved in {outputfolder_path}")
# Create log file
logfile = f"{filename}_run_{timestamp}.log"
logfile_path = os.path.join(outputfolder_path, logfile)
# Open log file, add some structure and basic information
config = read_config(configfile_path)
if not config.has_section('Paths'):
config.add_section('Paths')
config.set('Paths', 'framepath', framepath)
if not config.has_section('AcquisitionDetails'):
config.add_section('AcquisitionDetails')
try:
(
exposure_time,
resolution_height,
resolution_width,
binning_height,
binning_width,
acceleration_voltage,
pixel_height,
pixel_width,
pixel_unit,
camera_length,
detector_used,
instrument_id,
instrument_manufacturer,
instrument_model,
acquisition_start,
acquisition_end,
) = extract_info_from_gatan_metadata(input_path)
except Exception:
log_print('No valid metadata found in file')
config.set('AcquisitionDetails', 'acquisition_start', str(acquisition_start))
config.set('AcquisitionDetails', 'acquisition_end', str(acquisition_end))
config.set('AcquisitionDetails', 'instrument_manufacturer', str(instrument_manufacturer))
config.set('AcquisitionDetails', 'instrument_model', str(instrument_model))
config.set('AcquisitionDetails', 'instrument_id', str(instrument_id))
config.set('AcquisitionDetails', 'detector_used', str(detector_used))
config.set('AcquisitionDetails', 'acceleration_voltage', str(acceleration_voltage))
config.set('AcquisitionDetails', 'camera_length', str(camera_length))
config.set('AcquisitionDetails', 'pixel_width', str(pixel_width))
config.set('AcquisitionDetails', 'pixel_height', str(pixel_height))
config.set('AcquisitionDetails', 'binning_width', str(binning_width))
config.set('AcquisitionDetails', 'binning_height', str(binning_height))
config.set('AcquisitionDetails', 'resolution_width', str(resolution_width))
config.set('AcquisitionDetails', 'resolution_height', str(resolution_height))
config.set('AcquisitionDetails', 'exposure_time', str(exposure_time))
config.set('AcquisitionDetails', 'pixel_unit', str(pixel_unit))
if not config.has_section('Parameters'):
config.add_section('Parameters')
if not config.has_section('Output'):
config.add_section('Output')
with open(configfile_path, 'w') as cfgfile:
config.write(cfgfile)
log_start(logfile_path, '.ini file created')
log_start(logfile_path, f'system: {platform.system()} {platform.architecture()}, version {platform.version()}')
log_start(
logfile_path,
f'hardware: {psutil.cpu_count(logical=False)} physical cores, '
f'{psutil.cpu_count(logical=True)} logical cores, '
f'{round(psutil.virtual_memory().total / (1024 ** 3))} GB total memory'
)
log_start(logfile_path, f'python version {sys.version}')
log_start(
logfile_path,
'data acquisition details: InstrumentManufacturer: '
f'{str(instrument_manufacturer)}, InstrumentModel: {str(instrument_model)}, '
f'InstrumentId: {str(instrument_id)}, DetectorUsed: {str(detector_used)}, '
f'AccelerationVoltage: {str(acceleration_voltage)}, CameraLength: {camera_length}, '
f'PixelWidth: {pixel_width} PixelHeight: {pixel_height}, BinningWidth: {binning_width}, '
f'BinningHeight: {binning_height}, Resolution: ({resolution_width}, {resolution_height}), '
f'ExposureTime: {exposure_time}, PixelUnit: {pixel_unit}, AcquisitionStart: '
f'{acquisition_start}, AcquisitionEnd: {acquisition_end}'
)
return True, newfiles