# SPDX-FileCopyrightText: 2021 Division of Intelligent Medical Systems, DKFZ
# SPDX-FileCopyrightText: 2021 Janek Groehl
# SPDX-License-Identifier: MIT
import numpy as np
import os
import uuid
from simpa.log import Logger
from simpa.io_handling import load_data_field, load_hdf5
from simpa.core.device_digital_twins import DigitalDeviceTwinBase, PhotoacousticDevice
from simpa.utils import Settings, Tags
from pacfish import BaseAdapter, MetaDatum, DeviceMetaDataCreator, DetectionElementCreator, MetadataAcquisitionTags
from pacfish.iohandler import write_data as write_ipasc_data
[docs]class IpascSimpaAdapter(BaseAdapter):
"""
This class contains the logic to extract the needed meta data from the SIMPA simulation files to
fulfill the requirements of the IPASC standard data format.
"""
def __init__(self, hdf5_file_path: str, device: DigitalDeviceTwinBase, settings: Settings = None):
self.logger = Logger()
raise RuntimeError("Please do not use this adapter currently! It is still under construction!")
# Input validation with descriptive error messages
if not os.path.exists(hdf5_file_path):
self.logger.error(f"The given file path ({hdf5_file_path}) does not exist.")
raise AssertionError(f"The given file path ({hdf5_file_path}) does not exist.")
if not os.path.isfile(hdf5_file_path):
self.logger.error(f"The given file path ({hdf5_file_path}) does not point to a file.")
raise AssertionError(f"The given file path ({hdf5_file_path}) does not point to a file.")
if not hdf5_file_path.endswith(".hdf5"):
self.logger.error(f"The given file path must point to an hdf5 file that ends with '.hdf5'")
raise AssertionError(f"The given file path must point to an hdf5 file that ends with '.hdf5'")
self.simpa_hdf5_file_path = hdf5_file_path
self.ipasc_hdf5_file_path = hdf5_file_path.replace(".hdf5", "_ipasc.hdf5")
# checking SIMPA settings dictionary
if settings is None:
settings = load_hdf5(hdf5_file_path)
if Tags.SETTINGS not in settings:
self.logger.error("Unable to recover settings dictionary. Please supply a valid settings dictionary for a "
"successful export.")
settings = settings[Tags.SETTINGS]
if settings is None or not isinstance(settings, Settings):
self.logger.error("No settings found at Tags.SETTINGS in the loaded HDF5 file. "
"Please supply a valid settings dictionary for a successful export.")
self.settings = settings
# checking given photoacoustic device
if device is None or not isinstance(device, PhotoacousticDevice):
self.logger.error("Given device was not a photoacoustic device.")
raise AssertionError("Given device was not a photoacoustic device.")
self.device = device
if Tags.WAVELENGTHS not in settings:
self.logger.error("Tags.WAVELENGTHS was not defined in the settings dictionary. Aborting IPASC file export.")
self.wavelengths = self.settings[Tags.WAVELENGTHS]
# Load the data for the first wavelength just to get the number of elements and number of time steps
try:
num_elements, num_time_steps = np.shape(load_data_field(self.simpa_hdf5_file_path,
Tags.DATA_FIELD_TIME_SERIES_DATA, self.wavelengths[0]))
except KeyError as e:
self.logger.error(e)
raise AssertionError(e)
self.time_series_data = np.zeros(shape=(num_elements, num_time_steps, len(self.wavelengths), 1))
for wl_idx, wavelength in enumerate(self.wavelengths):
self.time_series_data[:, :, wl_idx, 0] = load_data_field(self.simpa_hdf5_file_path,
Tags.DATA_FIELD_TIME_SERIES_DATA, wavelength)
self.time_series_data = self.time_series_data.astype(np.float32)
super(IpascSimpaAdapter, self).__init__()
[docs] def generate_binary_data(self) -> np.ndarray:
return self.time_series_data
[docs]def export_to_ipasc(hdf5_file_path: str, device: DigitalDeviceTwinBase, settings: Settings = None):
"""
This function exports parts of the SIMPA simulation results of the given hdf5_file_path into the IPASC
standard data format.
:param hdf5_file_path: A string with the path to an HDF5 file containing a SIMPA simulation result
:param device: A PhotoacousticDevice that describes the digital device twin
:param settings: The settings dictionary used for the simulation. if not given, it is attempted to recover the
settings dictionary from the HDF5 file.
:return: None
"""
try:
ipasc_adapter = IpascSimpaAdapter(hdf5_file_path, device, settings)
pa_data = ipasc_adapter.generate_pa_data()
write_ipasc_data(ipasc_adapter.ipasc_hdf5_file_path, pa_data)
except Exception as e:
Logger().error(e)