from coseda.logging_utils import log_print
import os
import platform
import h5py
import datetime
import configparser
import psutil
import sys
import re
import numpy as np
from PIL import Image
from coseda.io import get_dir_size, get_free_space_windows, get_free_space_unix, handle_input, parse_config
from coseda.logging_utils import log_start, log_result
from coseda.initialize import create_insfiles
from coseda.nexus.process import write_beam_incident_energy, write_nxprocess_import
from coseda.nexus.images import ensure_image_nxdata
from coseda.nexus.indices import ensure_image_key
from coseda.nexus.logs import ensure_dense_logs
from coseda.nexus.goniometer import ensure_goniometer_transforms, get_goniometer_transform_order
from coseda.nexus.groups import ensure_nexus_parents
from coseda.nexus.detector import write_detector_geometry
TIFF_EXTENSIONS = (".tif", ".tiff")
def _natural_sort_key(filename):
"""Sort filenames in human order, e.g. frame_2 before frame_10."""
parts = re.split(r'(\d+)', filename.casefold())
key = []
for part in parts:
if part.isdigit():
key.append((1, int(part), part))
else:
key.append((0, part))
return key
[docs]
def find_tiff_frame_files(tiff_folder):
"""Return TIFF filenames in a stable natural order."""
return sorted(
(
filename
for filename in os.listdir(tiff_folder)
if os.path.isfile(os.path.join(tiff_folder, filename))
and filename.casefold().endswith(TIFF_EXTENSIONS)
),
key=_natural_sort_key,
)
def _legacy_consecutive_frame_files(tiff_folder, start_frame, extension):
frame_files = []
frame_number = start_frame
while frame_number is not None:
filename = f"{frame_number}{extension}"
if not os.path.exists(os.path.join(tiff_folder, filename)):
break
frame_files.append(filename)
frame_number += 1
return frame_files
[docs]
def create_insfiles_tiff(input_path, target):
filename = os.path.basename(input_path)
directory = os.path.dirname(input_path)
timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
outputfolder = f"{filename}_run_{timestamp}"
framepath = 'entry/data/images'
# Determine the output folder path
if target is None:
outputfolder_parent = directory
else:
outputfolder_parent = target
outputfolder_path = os.path.join(outputfolder_parent, outputfolder)
# Create the output folder and subfolder
os.makedirs(outputfolder_path, exist_ok=True)
os.makedirs(os.path.join(outputfolder_path, "plots"), exist_ok=True)
# Create the config file path in the parent folder of the new folder
configfile = f"{filename}_run_{timestamp}.ini"
configfile_path = os.path.join(outputfolder_parent, configfile)
# Create an empty configuration file
with open(configfile_path, 'w') as configfile_handle:
pass
log_print(f"Configuration file created for {input_path}")
log_print(f"Configuration will be written to {configfile_path}")
log_print(f"Results will be saved in {outputfolder_path}")
# Create log file path
logfile = f"{filename}_run_{timestamp}.log"
logfile_path = os.path.join(outputfolder_path, logfile)
# Generate HDF5 file name
h5file = f"{filename}_run_{timestamp}.h5"
# Open log file, add some structure and basic information
config = configparser.ConfigParser()
config.read(configfile_path)
if not config.has_section('Paths'):
config.add_section('Paths')
config.set('Paths', 'configfile_path', configfile_path)
config.set('Paths', 'outputfolder', outputfolder_path)
config.set('Paths', 'originalfile', filename)
config.set('Paths', 'originalfile_path', input_path) # Set full original file path
config.set('Paths', 'logfile', os.path.basename(logfile_path))
config.set('Paths', 'framepath', framepath)
config.set('Paths', 'h5file', h5file) # Write the HDF5 filename
if not config.has_section('AcquisitionDetails'):
config.add_section('AcquisitionDetails')
if not config.has_section('Parameters'):
config.add_section('Parameters')
if not config.has_section('Output'):
config.add_section('Output')
# Write the configuration to the file
with open(configfile_path, 'w') as configfile_handle:
config.write(configfile_handle)
# Log basic system details
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, {psutil.cpu_count(logical=True)} logical cores, {round(psutil.virtual_memory().total / (1024 ** 3))} GB total memory')
log_start(logfile_path, f'python version {sys.version}')
# Log the HDF5 filename
log_start(logfile_path, f'HDF5 file will be created as {h5file}')
set_tiffconversion_parameters(configfile_path)
return True, configfile_path
[docs]
def set_tiffconversion_parameters(ini_file_path, compression=None, compression_level=None):
# Ensure the .ini file exists
if not os.path.exists(ini_file_path):
raise FileNotFoundError(f"The .ini file '{ini_file_path}' does not exist.")
# Read the .ini file
config = configparser.ConfigParser()
config.read(ini_file_path)
# Extract the TIFF folder from the ini file
if not config.has_section('Paths') or not config.has_option('Paths', 'originalfile_path'):
raise ValueError(f"The .ini file '{ini_file_path}' must contain 'originalfile_path' in the 'Paths' section.")
tiff_folder = config.get('Paths', 'originalfile_path')
# Extract the logfile path from the ini file
if not config.has_option('Paths', 'logfile'):
raise ValueError(f"The .ini file '{ini_file_path}' must contain 'logfile' in the 'Paths' section.")
logfile_name = config.get('Paths', 'logfile')
if not config.has_option('Paths', 'outputfolder'):
raise ValueError(f"The .ini file '{ini_file_path}' must contain 'outputfolder' in the 'Paths' section.")
outputfolder = config.get('Paths', 'outputfolder')
logfile_path = os.path.join(outputfolder, logfile_name)
# Ensure the folder exists
if not os.path.exists(tiff_folder):
log_start(logfile_path, f"Error: TIFF folder '{tiff_folder}' does not exist.")
raise FileNotFoundError(f"The TIFF folder '{tiff_folder}' specified in the .ini file does not exist.")
# Find TIFF filenames in the folder (.tif/.tiff, any case) and preserve
# their natural filename order for conversion.
tiff_files = find_tiff_frame_files(tiff_folder)
if not tiff_files:
log_start(logfile_path, f"No valid TIFF files found in folder: {tiff_folder}")
raise ValueError(f"No TIFF files found in '{tiff_folder}'.")
# Keep the legacy start_frame setting when names are plain numbers.
numeric_match = re.fullmatch(r'(\d+)\.(tif|tiff)', tiff_files[0], re.IGNORECASE)
start_frame = int(numeric_match.group(1)) if numeric_match else None
# Use the extension (with original casing) from the first file
extension = os.path.splitext(tiff_files[0])[1]
if not extension:
extension = ".tiff"
# Ensure the 'Parameters' section exists
if not config.has_section('Parameters'):
config.add_section('Parameters')
# Set the parameters in the Parameters section
config.set('Parameters', 'tiffconversion_start_frame', str(start_frame) if start_frame is not None else "None")
config.set('Parameters', 'tiffconversion_compression', compression if compression else "None")
config.set('Parameters', 'tiffconversion_compression_level', str(compression_level) if compression_level else "None")
config.set('Parameters', 'tiffconversion_extension', extension)
config.set('Parameters', 'tiffconversion_first_frame', tiff_files[0])
config.set('Parameters', 'tiffconversion_frame_count', str(len(tiff_files)))
# Write back the changes to the .ini file
with open(ini_file_path, 'w') as configfile:
config.write(configfile)
# Log the updates
log_start(logfile_path, f"recognized {tiff_files[0]} as first frame")
log_start(logfile_path, f"recognized {len(tiff_files)} TIFF frames")
if compression is None:
log_start(logfile_path, f"no compression used")
else:
log_start(logfile_path, f"using {compression} compression level {compression_level}")
[docs]
def stupid_tiff_folder_conversion(
tiffparentfolder,
outputfolder,
logfile_path,
start_frame=1000000,
frame_files=None,
chunk_size=(1000, 512, 512),
compression=None,
compression_level=4,
extension=".tiff",
acceleration_voltage=None,
camera_length=None,
camera_length_correction=1.0,
goniometer_transform_order=None,
):
try:
# Define the HDF5 file path within the outputfolder
h5file = os.path.basename(outputfolder) + ".h5"
h5file_path = os.path.join(outputfolder, h5file)
# Calculate the total size of the TIFF input directory
file_size = get_dir_size(tiffparentfolder)
# Determine free space on the output drive
if platform.system() == 'Windows':
free_space = get_free_space_windows(outputfolder)
else:
free_space = get_free_space_unix(outputfolder)
# Check if free space is at least 1.1 times the input size
required_space = 1.1 * file_size
if free_space < required_space:
error_message = (
f"Insufficient disk space. Required: {required_space / (1024 ** 3):.2f} GB, "
f"Available: {free_space / (1024 ** 3):.2f} GB."
)
log_start(logfile_path, error_message)
return error_message, None
else:
log_start(logfile_path,
f"Sufficient disk space available. Required: {required_space / (1024 ** 3):.2f} GB, "
f"Available: {free_space / (1024 ** 3):.2f} GB.")
new_framepath = 'entry/data/images'
new_indexpath = 'entry/data/index'
with h5py.File(h5file_path, 'w') as new_file:
log_start(logfile_path, "HDF5 file created")
initial_shape = (0, chunk_size[1], chunk_size[2])
maxshape = (None, chunk_size[1], chunk_size[2])
# Create dataset based on compression settings
if compression is None:
# No compression
new_dataset = new_file.create_dataset(
new_framepath,
shape=initial_shape,
maxshape=maxshape,
chunks=chunk_size,
dtype='int16'
)
elif compression == "gzip":
# GZIP compression with compression level
new_dataset = new_file.create_dataset(
new_framepath,
shape=initial_shape,
maxshape=maxshape,
chunks=chunk_size,
dtype='int16',
compression=compression,
compression_opts=compression_level
)
else:
# Other compression (e.g., LZF, which doesn't require compression options)
new_dataset = new_file.create_dataset(
new_framepath,
shape=initial_shape,
maxshape=maxshape,
chunks=chunk_size,
dtype='int16',
compression=compression
)
index_dataset = new_file.create_dataset(
new_indexpath,
shape=(0,),
maxshape=(None,),
dtype='i4'
)
frame_count = 0
save_interval = 10000
if frame_files is None:
frame_files = _legacy_consecutive_frame_files(tiffparentfolder, start_frame, extension)
if not frame_files:
frame_files = find_tiff_frame_files(tiffparentfolder)
if not frame_files:
error_message = f"No TIFF files found in folder: {tiffparentfolder}"
log_start(logfile_path, error_message)
return error_message, None
for framenamefull in frame_files:
currentpath = os.path.join(tiffparentfolder, framenamefull)
try:
frame = Image.open(currentpath) # Using PIL to open TIFF
frame = np.array(frame) # Convert the image to a NumPy array
# Resize datasets for the new frame
new_dataset.resize(new_dataset.shape[0] + 1, axis=0)
new_dataset[-1] = frame
index_dataset.resize(index_dataset.shape[0] + 1, axis=0)
index_dataset[-1] = frame_count
frame_count += 1
if frame_count % 1000 == 0:
log_print(f'{frame_count} frames processed')
if frame_count % save_interval == 0:
new_file.flush() # Flush data to disk
log_start(logfile_path, f"Saved after processing {frame_count} frames")
except Exception as frame_error:
log_start(logfile_path, f"Error processing frame {framenamefull}: {str(frame_error)}")
continue
log_print(f"Conversion complete. Processed {frame_count} frames.")
ensure_nexus_parents(new_file)
ensure_image_nxdata(new_file)
ensure_image_key(new_file)
ensure_dense_logs(new_file)
ensure_goniometer_transforms(new_file, goniometer_transform_order)
write_nxprocess_import(
new_file,
program="coseda.importers.import_tiff",
input_path=tiffparentfolder,
output_path=h5file_path,
parameters={
"start_frame": start_frame,
"first_frame": frame_files[0],
"frame_count": len(frame_files),
"chunk_size": chunk_size,
"compression": compression or "none",
"compression_level": compression_level,
"extension": extension,
},
)
write_beam_incident_energy(new_file, acceleration_voltage)
write_detector_geometry(new_file, camera_length, camera_length_correction)
log_start(logfile_path, "Dataset created and data copied")
return None, h5file
except Exception as e:
return f"Error during file conversion: {str(e)}", None
[docs]
def tiff_conversion(ini_file_path):
# Ensure the .ini file exists
if not os.path.exists(ini_file_path):
raise FileNotFoundError(f"The .ini file '{ini_file_path}' does not exist")
# Read the .ini file
config = configparser.ConfigParser()
config.read(ini_file_path)
# Extract necessary paths and parameters
if not config.has_section('Paths') or not config.has_option('Paths', 'originalfile_path'):
raise ValueError(f"The .ini file '{ini_file_path}' must contain 'originalfile_path' in the 'Paths' section")
tiffparentfolder = config.get('Paths', 'originalfile_path')
if not config.has_option('Paths', 'outputfolder'):
raise ValueError(f"The .ini file '{ini_file_path}' must contain 'outputfolder' in the 'Paths' section")
outputfolder = config.get('Paths', 'outputfolder')
if not config.has_option('Paths', 'logfile'):
raise ValueError(f"The .ini file '{ini_file_path}' must contain 'logfile' in the 'Paths' section")
logfile_path = os.path.join(os.path.dirname(ini_file_path), config.get('Paths', 'logfile'))
if not config.has_section('Parameters'):
raise ValueError(f"The .ini file '{ini_file_path}' must contain a 'Parameters' section")
# Read legacy `tiffconversion_start_frame`, allowing None.
start_frame_raw = config.get('Parameters', 'tiffconversion_start_frame', fallback=None)
start_frame = int(start_frame_raw) if start_frame_raw not in (None, "None") else None
# Read `tiffconversion_compression`, allowing None
compression = config.get('Parameters', 'tiffconversion_compression', fallback=None)
compression = None if compression == "None" else compression
# Read `tiffconversion_compression_level`, allowing None
compression_level_raw = config.get('Parameters', 'tiffconversion_compression_level', fallback=None)
compression_level = int(compression_level_raw) if compression_level_raw not in (None, "None") else 4 # Default to 4 if not set
extension = config.get('Parameters', 'tiffconversion_extension', fallback='.tiff')
if not extension.startswith('.'):
extension = f".{extension}"
configured_first_frame = config.get('Parameters', 'tiffconversion_first_frame', fallback=None)
if configured_first_frame not in (None, "", "None"):
frame_files = find_tiff_frame_files(tiffparentfolder)
else:
frame_files = _legacy_consecutive_frame_files(tiffparentfolder, start_frame, extension)
if not frame_files:
frame_files = find_tiff_frame_files(tiffparentfolder)
if not frame_files:
log_start(logfile_path, f"No TIFF files found in folder: {tiffparentfolder}")
raise ValueError(f"No TIFF files found in '{tiffparentfolder}'.")
# Determine frame size dynamically from the first TIFF file.
first_frame_path = os.path.join(tiffparentfolder, frame_files[0])
frame = Image.open(first_frame_path) # Using PIL to open TIFF
frame_array = np.array(frame) # Convert the image to a NumPy array
frame_height, frame_width = frame_array.shape
# Default chunk size
chunk_size = (1000, frame_height, frame_width)
# Log the determined frame size and chunk size
log_start(logfile_path, f"First frame determined as {frame_files[0]}")
log_start(logfile_path, f"Found {len(frame_files)} TIFF frames")
log_start(logfile_path, f"Frame size determined as {frame_height}x{frame_width}")
log_start(logfile_path, f"Using chunk size {chunk_size}")
# Perform the TIFF-to-HDF5 conversion
try:
acceleration_voltage = float(config.get('AcquisitionDetails', 'acceleration_voltage'))
except Exception:
acceleration_voltage = None
try:
camera_length = float(config.get('AcquisitionDetails', 'camera_length'))
except Exception:
camera_length = None
try:
camera_length_correction = float(config.get('AcquisitionDetails', 'camera_length_correction'))
except Exception:
camera_length_correction = 1.0
goniometer_transform_order = get_goniometer_transform_order(config)
error_message, _ = stupid_tiff_folder_conversion(
tiffparentfolder=tiffparentfolder,
outputfolder=outputfolder,
logfile_path=logfile_path,
start_frame=start_frame,
frame_files=frame_files,
chunk_size=chunk_size,
compression=compression,
compression_level=compression_level,
extension=extension,
acceleration_voltage=acceleration_voltage,
camera_length=camera_length,
camera_length_correction=camera_length_correction,
goniometer_transform_order=goniometer_transform_order,
)
if error_message:
raise RuntimeError(error_message)
[docs]
def tiff_conversion_batch(input_path):
# Use handle_input to retrieve a list of valid .ini files and the new input path
configfiles, _ = handle_input(input_path)
# Process each .ini file
for configfile in configfiles:
log_print(f"Processing {configfile}...")
try:
# Parse the .ini file to retrieve necessary parameters
config, outputfolder, originalfile, logfile, path, outputfolder_path, originalfile_path, logfile_path, framepath, h5file, h5file_path = parse_config(configfile)
# Log the start of the process
log_start(logfile_path, f"starting TIFF-to-HDF5 conversion for {configfile}")
# Perform the conversion
tiff_conversion(configfile)
# Log successful completion
log_start(logfile_path, f"successfully completed TIFF-to-HDF5 conversion for {configfile}")
except Exception as e:
# Log any errors that occur during the process
error_message = f"Error processing {configfile}: {str(e)}"
log_print(error_message)
if 'logfile_path' in locals():
log_start(logfile_path, error_message)