"""Write processing parameters to INI files."""
from coseda.logging_utils import log_start, shoutout
from coseda.io import handle_input, read_config, config_to_paths
[docs]
def write_h5conversionsettings(input_path, limit_frames, bin_factor, sample_class):
"""Persist HDF5 conversion limits/binning and sample class into each INI."""
configfiles, _ = handle_input(input_path)
for configfile in configfiles:
shoutout(configfile)
config = read_config(configfile)
# Set parameters
config.set('Parameters', 'sample_class', f'{sample_class}')
config.set('Parameters', 'h5conversion_limit_frames', f'{limit_frames}')
config.set('Parameters', 'h5conversion_bin_factor', f'{bin_factor}')
with open(configfile, 'w') as cfgfile:
config.write(cfgfile)
_, _, _, logfile_path, _, _ = config_to_paths(configfile)
log_start(logfile_path, f'sample class set to: {sample_class}')
log_start(
logfile_path,
f'HDF5 conversion parameters set, limit_frames = {limit_frames}, bin_factor = {bin_factor}'
)
[docs]
def write_peakfindersettings(
input_path,
threshold,
min_snr,
min_pix_count,
max_pix_count,
local_bg_radius,
min_res,
max_res,
x0=None,
y0=None,
num_threads=None,
):
"""Write Diffractem peakfinder_8 parameters and geometry hints into each INI."""
configfiles, _ = handle_input(input_path)
for configfile in configfiles:
shoutout(configfile)
config = read_config(configfile)
# Set parameters
parameters_to_set = {
'peakfinding_threshold': threshold,
'peakfinding_min_snr': min_snr,
'peakfinding_min_pix_count': min_pix_count,
'peakfinding_max_pix_count': max_pix_count,
'peakfinding_local_bg_radius': local_bg_radius,
'peakfinding_min_res': min_res,
'peakfinding_max_res': max_res,
}
# Add 'num_threads' if provided
if num_threads is not None:
parameters_to_set['peakfinding_num_threads'] = num_threads
# Set x0 and y0, defaulting to 'None' if not provided
parameters_to_set['peakfinding_x0'] = x0 if x0 is not None else 'None'
parameters_to_set['peakfinding_y0'] = y0 if y0 is not None else 'None'
for param, value in parameters_to_set.items():
config.set('Parameters', param, str(value))
# Write configfile and log
with open(configfile, 'w') as cfgfile:
config.write(cfgfile)
_, _, _, logfile_path, _, _ = config_to_paths(configfile)
# Prepare log messages
log_message = (
f'peakfinder settings set, threshold = {threshold}, '
f'min_snr = {min_snr}, min_pix_count = {min_pix_count}, '
f'max_pix_count = {max_pix_count}, local_bg_radius = {local_bg_radius}, '
f'min_res = {min_res}, max_res = {max_res}'
)
if num_threads is not None:
log_message += f', num_threads = {num_threads}'
else:
log_message += ', num_threads not set'
log_start(logfile_path, log_message)
geometry_message = (
f'detector geometry set, x0 = {x0 if x0 is not None else "None"}, '
f'y0 = {y0 if y0 is not None else "None"}'
)
log_start(logfile_path, geometry_message)
# Parameter sensibility checks
if max_pix_count <= 2 * min_pix_count:
log_start(
logfile_path,
'Warning: max_pix_count is close or equal to min_pix_count, '
'attempting to find very specific peak sizes may lead to a small number of identified peaks.'
)
if max_res <= 2 * min_res:
log_start(
logfile_path,
'Warning: max_res is close or equal to min_res, this might prevent peakfinder8 from identifying any peaks.'
)
[docs]
def write_centerfindersettings(
input_path,
tolerance,
min_peaks,
resolution_limit,
min_samples_fraction,
dbscan_eps=None,
force_linear_fit=False,
skip_linear_fit=False,
x0=None,
y0=None,
):
"""Persist center-finding thresholds and initial center guess into each INI."""
configfiles, input_path = handle_input(input_path)
# Ensure mutually exclusive flags: skip overrides force
if skip_linear_fit:
force_linear_fit = False
for configfile in configfiles:
shoutout(configfile)
config = read_config(configfile)
# Set parameters
config.set('Parameters', 'centerfinding_tolerance', f'{tolerance}')
config.set('Parameters', 'centerfinding_min_peaks', f'{min_peaks}')
config.set('Parameters', 'centerfinding_resolution_limit', f'{resolution_limit}')
config.set('Parameters', 'centerfinding_min_samples_fraction', f'{min_samples_fraction}')
if dbscan_eps is not None:
config.set('Parameters', 'centerfinding_dbscan_eps', f'{dbscan_eps}')
config.set('Parameters', 'centerfinding_x0', f'{x0}')
config.set('Parameters', 'centerfinding_y0', f'{y0}')
config.set('Parameters', 'centerfinding_force_linear_fit', str(force_linear_fit))
config.set('Parameters', 'centerfinding_skip_linear_fit', str(skip_linear_fit))
# Write the file with 'utf-8' encoding
with open(configfile, 'w') as cfgfile:
config.write(cfgfile)
_, _, _, logfile_path, _, _ = config_to_paths(configfile)
log_start(
logfile_path,
f'parameters set for center finding, tolerance = {tolerance}, min_peaks = {min_peaks}, '
f'resolution_limit = {resolution_limit}, min_samples_fraction = {min_samples_fraction}, '
f'dbscan_eps = {dbscan_eps if dbscan_eps is not None else "default"}, '
f'force_linear_fit = {force_linear_fit}, skip_linear_fit = {skip_linear_fit}, x0 = {x0}, y0 = {y0}'
)
[docs]
def write_centerrefinementsettings(
input_path,
tolerance,
min_peaks,
resolution_limit,
min_samples_fraction=None,
max_iterations=None,
convergence_threshold=None,
deviation_aggregation=None,
lowess_frac=None,
lowess_window=None,
):
"""Persist iterative center refinement settings (tolerance, samples, LOWESS) into INIs."""
configfiles, input_path = handle_input(input_path)
if max_iterations is None or convergence_threshold is None:
raise ValueError("max_iterations and convergence_threshold must be provided for center refinement settings.")
for configfile in configfiles:
shoutout(configfile)
config = read_config(configfile)
# Set parameters
config.set('Parameters', 'centerrefinement_tolerance', f'{tolerance}')
config.set('Parameters', 'centerrefinement_min_peaks', f'{min_peaks}')
config.set('Parameters', 'centerrefinement_resolution_limit', f'{resolution_limit}')
config.set('Parameters', 'centerrefinement_max_iterations', f'{max_iterations}')
config.set('Parameters', 'centerrefinement_convergence_threshold', f'{convergence_threshold}')
if deviation_aggregation is not None:
config.set('Parameters', 'centerrefinement_deviation_aggregation', f'{deviation_aggregation}')
if lowess_frac is not None:
config.set('Parameters', 'centerrefinement_lowess_frac', f'{lowess_frac}')
elif config.has_option('Parameters', 'centerrefinement_lowess_frac') and lowess_frac is None:
# leave existing value if not provided explicitly
pass
if lowess_window is not None and lowess_window > 0:
config.set('Parameters', 'centerrefinement_lowess_window', f'{int(lowess_window)}')
elif config.has_option('Parameters', 'centerrefinement_lowess_window'):
config.remove_option('Parameters', 'centerrefinement_lowess_window')
# Write the file with 'utf-8' encoding
with open(configfile, 'w') as cfgfile:
config.write(cfgfile)
_, _, _, logfile_path, _, _ = config_to_paths(configfile)
message = (
f'parameters set for refinement of center, tolerance = {tolerance}, '
f'min_peaks = {min_peaks}, resolution_limit = {resolution_limit}, '
f'max_iterations = {max_iterations}, convergence_threshold = {convergence_threshold}'
)
if deviation_aggregation is not None:
message += f', deviation_aggregation = {deviation_aggregation}'
if lowess_frac is not None:
message += f', lowess_frac = {lowess_frac}'
if lowess_window is not None:
message += f', lowess_window = {lowess_window}'
log_start(logfile_path, message)
[docs]
def write_gandalfiteratorsettings(
configfile,
geomfile_path=None,
cellfile_path=None,
output_file_base=None,
threads=None,
max_radius=None,
step=None,
peakfinder_method=None,
peakfinder_params=None,
min_peaks=None,
cell_tolerance=None,
sampling_pitch=None,
grad_desc_iterations=None,
xgandalf_tolerance=None,
int_radius=None,
other_flags=None,
):
"""
Update the parent INI file with Gandalf-iterator settings under [Parameters].
Does NOT require input folder (list file is used instead).
"""
shoutout(configfile)
outputfolder, outputfolder_path, logfile, logfile_path, h5file, h5file_path = config_to_paths(configfile)
config = read_config(configfile)
# Ensure 'Paths' section exists and set geometry, cell, output_base
if not config.has_section('Paths'):
config.add_section('Paths')
if geomfile_path is not None:
config.set('Paths', 'geomfile_path', geomfile_path)
if cellfile_path is not None:
config.set('Paths', 'cellfile_path', cellfile_path)
if output_file_base is not None:
config.set('Paths', 'output_file_base', output_file_base)
# Prepare parameters to set under [Parameters]
params = {
'gandalfiterations_threads': threads,
'gandalfiterations_max_radius': max_radius,
'gandalfiterations_step': step,
'gandalfiterations_peakfinder_method': peakfinder_method,
'gandalfiterations_min_peaks': min_peaks,
'gandalfiterations_cell_tolerance': cell_tolerance,
'gandalfiterations_sampling_pitch': sampling_pitch,
'gandalfiterations_grad_desc_iterations': grad_desc_iterations,
'gandalfiterations_xgandalf_tolerance': xgandalf_tolerance,
'gandalfiterations_int_radius': int_radius,
}
# Add peakfinder_params and other_flags as multiline if provided
if peakfinder_params is not None:
params['gandalfiterations_peakfinder_params'] = "\n".join(peakfinder_params)
if other_flags is not None:
params['gandalfiterations_other_flags'] = "\n".join(other_flags)
if not config.has_section('Parameters'):
config.add_section('Parameters')
# Write all parameters, only writing those that are provided (not None)
for key, value in params.items():
if value is not None:
config.set('Parameters', key, str(value))
# If value is None, skip writing the key (leave any existing value untouched)
# Write the updated config file
with open(configfile, 'w', encoding='utf-8') as cfgfile:
config.write(cfgfile)
# Log the settings updated
log_message = 'Gandalf iterator settings updated: ' + ', '.join(
f'{k} = {config.get("Parameters", k)}' for k in params.keys()
)
log_start(logfile_path, log_message)
# Log Paths settings for clarity
paths_keys = ['geomfile_path', 'cellfile_path', 'output_file_base']
paths_settings = [
f'{key} = {config.get("Paths", key)}' for key in paths_keys if config.has_option('Paths', key)
]
if paths_settings:
paths_message = 'Gandalf iterator paths updated: ' + ', '.join(paths_settings)
log_start(logfile_path, paths_message)
[docs]
def write_stripsettings(input, strip_threshold, strip_force):
"""Persist strip-empty-frame settings."""
if isinstance(input, list):
configfiles = list(input)
else:
configfiles, _ = handle_input(input)
for configfile in configfiles:
config = read_config(configfile)
shoutout(configfile)
# Ensure 'Parameters' section exists
if not config.has_section('Parameters'):
config.add_section('Parameters')
# Set parameters
config.set('Parameters', 'strip_threshold', f'{strip_threshold}')
config.set('Parameters', 'strip_force', f'{strip_force}')
# Write the updated config file with 'utf-8' encoding
with open(configfile, 'w', encoding='utf-8') as f:
config.write(f)
outputfolder, outputfolder_path, logfile, logfile_path, h5file, h5file_path = config_to_paths(configfile)
log_start(
logfile_path,
f"Parameters set for stripping: strip_threshold = {strip_threshold}, strip_force = {strip_force}"
)
[docs]
def write_mergingsettings(
input_path,
hkl_filename,
pointgroup,
min_res,
iterations,
threads=None,
model="offset",
polarisation="none",
min_measurements=2,
max_adu="inf",
push_res="inf",
no_Bscale=True,
output_every_cycle=True,
save_partialator_logs=False,
unique_axis=None,
streamfile=None,
):
"""
Write merging parameters into the INI file.
'hkl_filename' is saved under [Paths], all other parameters under [Parameters]
with keys prefixed by 'merging_'.
"""
# Determine list of config files (INI) to update
configfiles, base_path = handle_input(input_path)
for configfile in configfiles:
shoutout(configfile)
config = read_config(configfile)
# Ensure [Paths] section exists and set HKL filename
if not config.has_section('Paths'):
config.add_section('Paths')
config.set('Paths', 'hkl_file', hkl_filename)
# Write streamfile if provided
if streamfile is not None:
config.set('Paths', 'streamfile', streamfile)
# Prepare merging parameters under [Parameters]
params = {
'merging_pointgroup': pointgroup.replace(" ", ""),
'merging_min_res': min_res,
'merging_iterations': iterations,
'merging_model': model,
'merging_polarisation': polarisation,
'merging_min_measurements': min_measurements,
'merging_max_adu': max_adu,
'merging_push_res': push_res,
'merging_no_Bscale': str(no_Bscale),
'merging_output_every_cycle': str(output_every_cycle),
'merging_save_partialator_logs': str(save_partialator_logs),
}
if threads is not None:
params['merging_threads'] = threads
if unique_axis is not None:
params['merging_unique_axis'] = unique_axis
# Ensure [Parameters] section exists
if not config.has_section('Parameters'):
config.add_section('Parameters')
# Write each merging parameter if provided
for key, value in params.items():
if value is not None:
config.set('Parameters', key, str(value))
# Save back to INI file
with open(configfile, 'w', encoding='utf-8') as cfgfile:
config.write(cfgfile)
_, _, _, logfile_path, _, _ = config_to_paths(configfile)
log_start(logfile_path, 'Merging settings: writing to INI')
log_message = (
f"Merging settings set: hkl_file = {hkl_filename}, "
f"pointgroup = {pointgroup}, min_res = {min_res}, "
f"iterations = {iterations}, model = {model}, polarisation = {polarisation}, "
f"min_measurements = {min_measurements}, max_adu = {max_adu}, "
f"push_res = {push_res}, no_Bscale = {no_Bscale}, "
f"output_every_cycle = {output_every_cycle}, "
f"save_partialator_logs = {save_partialator_logs}"
)
if threads is not None:
log_message += f", threads = {threads}"
if unique_axis is not None:
log_message += f", unique_axis = {unique_axis}"
log_start(logfile_path, log_message)
[docs]
def write_xgandalfsettings(
input_path,
tolerance,
sampling_pitch,
min_lattice_vector_length,
max_lattice_vector_length,
grad_desc_iterations,
tolerance_5d,
fix_profile_radius,
):
"""
Writes Xgandalf settings to the [Parameters] section of each config file determined by handle_input.
Logs the new settings if [Paths] logfile is present.
"""
configfiles, _ = handle_input(input_path)
for configfile in configfiles:
config = read_config(configfile)
if not config.has_section('Parameters'):
config.add_section('Parameters')
config.set('Parameters', 'xgandalf_tolerance', str(tolerance))
config.set('Parameters', 'xgandalf_sampling_pitch', str(sampling_pitch))
config.set('Parameters', 'xgandalf_min_lattice_vector_length', str(min_lattice_vector_length))
config.set('Parameters', 'xgandalf_max_lattice_vector_length', str(max_lattice_vector_length))
config.set('Parameters', 'xgandalf_grad_desc_iterations', str(grad_desc_iterations))
config.set('Parameters', 'tolerance_5d', str(tolerance_5d))
config.set('Parameters', 'fix_profile_radius', str(fix_profile_radius))
with open(configfile, 'w', encoding='utf-8') as cfgfile:
config.write(cfgfile)
# Resolve logfile path from project folder
_, _, _, logfile_path, _, _ = config_to_paths(configfile)
log_start(
logfile_path,
f'Xgandalf settings set: tolerance={tolerance}, sampling_pitch={sampling_pitch}, '
f'min_lattice_vector_length={min_lattice_vector_length}, '
f'max_lattice_vector_length={max_lattice_vector_length}, '
f'grad_desc_iterations={grad_desc_iterations}, tolerance_5d={tolerance_5d}, '
f'fix_profile_radius={fix_profile_radius}'
)
[docs]
def write_icisettings(input_path, **params):
"""
Write ICI orchestrator key parameters into the [Parameters] section of each config file.
Keys are stored with the 'ici_' prefix to avoid collisions.
"""
configfiles, _ = handle_input(input_path)
for configfile in configfiles:
config = read_config(configfile)
if not config.has_section('Parameters'):
config.add_section('Parameters')
if config.has_option('Parameters', 'ici_flags'):
config.remove_option('Parameters', 'ici_flags')
for key, value in params.items():
if value is None:
continue
config.set('Parameters', f'ici_{key}', str(value))
with open(configfile, 'w', encoding='utf-8') as cfgfile:
config.write(cfgfile)
_, _, _, logfile_path, _, _ = config_to_paths(configfile)
try:
log_message = 'ICI settings set: ' + ', '.join(
f'ici_{k}={v}' for k, v in params.items() if v is not None
)
log_start(logfile_path, log_message)
except Exception:
pass