lf_mri_platform/services/reconstructor/reconsrtuctionApp.py

import json
import h5py
import sys
# sys.path.append("C:/Users/iuliia/recoUI/serv")
from traj import *
from reco import *
from seqData import SequenceDataFrame
from service import ReconstructionService


class ReconstructionApp():

    def __init__(self, name, digit, shift):
        super().__init__()
        self.name = name  # type seq (linear|non|epi|radial)
        # print("self.name", self.name)
        self.digit = digit  # 2|3 d
        self.phase_shift = shift  # bool

    def start_reconstruction(self, path_raw_data, path_np_data_json, path_order_json):
        try:
            service = ReconstructionService()

            nX, nY, nZ, d_scans, nContr, spoil, ETL, N_TE, phi_wing, N_wings = self.read_consts(path_np_data_json)
            # print(nX, nY, nZ, d_scans, nContr, spoil)
            service.path_raw_data = path_raw_data

            raw_data = service.read_mat_file()
            sequence_data = SequenceDataFrame()
            sequence_data.read_raw_data(raw_data)
            sequence_data.read_dots(int(nX))
            sequence_name = self.name  # Получаем введенное имя последовательности

            # print(sequence_name, path_order_json)

            if sequence_name == 'nonlinear_decart(tse)' or sequence_name == 'radial_propeller':
                print(path_order_json)

                k_space_order = self.read_order_from_json(path_order_json)
                print(f"Загружен порядок k-space: {k_space_order}")

                self.k_space_order = k_space_order

            arr_mat = np.array(sequence_data.mat_df['dot'])
            # print("len_arr_raw_data: ", len(arr_mat))

            # print(sequence_name)
            digit = self.digit  # Получаем 2d или 3d
            # print("digit:", self.digit)

            if not path_raw_data or not path_np_data_json:
                raise ValueError("Пожалуйста, выберите файлы и введите данные перед началом реконструкции.")

            else:
                json_data = path_np_data_json

                if digit == '2d':
                    # print("@ @ 2d @ @")
                    self.reconstruct_2d(sequence_name, arr_mat, json_data)
                elif digit == '3d':
                    # print("@ @ 3d @ @")
                    self.reconstruct_3d(sequence_name, arr_mat, json_data)

        except ValueError as ve:
            raise ValueError(str(ve))

        except NotImplementedError as nie:
            raise NotImplementedError(str(nie))

        except Exception as e:
            raise RuntimeError(f"Произошла ошибка в процессе реконструкции: {e}")

    def read_order_from_json(self, path_order_json):
        with open(path_order_json, 'r') as f:
            data = json.load(f)

        # Извлечение массива из ключа "k_space_order"
        k_space_order = data.get("k_space_order", [])
        return k_space_order

    def read_mat_file(self, path):
        with h5py.File(path, 'r') as f:
            key = list(f.keys())[0]

        with h5py.File(path, 'r') as f:
            data = f[key][:]

        data = np.array(data)

        return data

    def get_value_or_default(self, json_obj, key, default_value=1):
        return int(json_obj.get(key, default_value))

    def read_consts(self, path):
        with open(path, 'r') as file:
            data = json.load(file)

        # d_scans = int(data['D_scans'])
        d_scans = self.get_value_or_default(data, "D_scans", 1)
        Np = int(data['Np'])
        Nf = int(data['Nf'])
        sl_nb = int(data['sl_nb'])
        # nContrast = int(data['contrasts'])
        nContrast = self.get_value_or_default(data, "contrasts", 1)
        # spoil = float(data['RF_spoil'])
        spoil = self.get_value_or_default(data, "RF_spoil", 0)

        ETL = self.get_value_or_default(data, 'ETL', 0)  # TSE
        N_TE = self.get_value_or_default(data, 'N_TE', 0)  # TSE

        phi_wing = self.get_value_or_default(data, "phi_wing", 0)  # radial
        N_wings = self.get_value_or_default(data, "N_wings", 0)  # radial

        return Np, Nf, sl_nb, d_scans, nContrast, spoil, ETL, N_TE, phi_wing, N_wings

    def read_consts_3d(self, path):
        with open(path, 'r') as file:
            data = json.load(file)

        d_scans = self.get_value_or_default(data, "D_scans", 1)
        # d_scans = int(data['D_scans'])
        Np = int(data['Np'])
        Nf = int(data['Nf'])
        # sl_nb = self.get_value_or_default(data, "Nss", 1)
        sl_nb = int(data['Nss'])
        if sl_nb % 2 == 1:
            sl_nb -= 1
        # nContrast = int(data['contrasts'])
        nContrast = self.get_value_or_default(data, "contrasts", 1)
        # spoil = float(data['RF_spoil'])
        spoil = self.get_value_or_default(data, "RF_spoil", 0)

        return Np, Nf, sl_nb, d_scans, nContrast, spoil

    def reconstruct_2d(self, sequence_name, mat_data, json_data):
        try:
            nX, nY, nZ, d_scans, nContr, spoil, ETL, N_TE, phi_wing, N_wings = self.read_consts(json_data)
            # print(spoil)

            if self.phase_shift == False:
                spoil = 0

            '''
                all_data = k_space[x, y, z, contrast, coil] - собрано в правильном порядке
            '''
            if sequence_name == "linear_decart":
                # print("ll")
                all_data = gather_data_along_trajectory(mat_data, nX, nY, nZ, nContr)
                # print("all_data")

                save_ffft(all_data, spoil)

            elif sequence_name == "nonlinear_decart(tse)":
                all_data = gather_data_along_trajectory_nonlinear(mat_data, self.k_space_order, nX, nY, nZ, nContr)
                save_ffft(all_data, spoil)

            elif sequence_name == "linear_epi":
                all_data = gather_data_along_trajectory_linear_epi(mat_data, nX, nY, nZ, nContr)
                save_ffft(all_data, spoil)

            elif sequence_name == "radial_propeller":
                print("RADIAL")
                all_data = gather_data_along_trajectory_radial(mat_data, self.k_space_order, nX, nY, nZ, phi_wing,
                                                               N_wings)
                save_ffft(all_data, spoil)

            else:
                raise NotImplementedError("Для данного типа ИП еще не добавлена функция.")
                # QMessageBox.critical(self, "Ошибка: для данного типа ИП еще не добавлена функция")
        except ValueError as ve:
            raise ValueError(str(ve))

        except NotImplementedError as nie:
            raise NotImplementedError(str(nie))

        except Exception as e:
            raise RuntimeError(f"Произошла ошибка в процессе реконструкции: {e}")

    def reconstruct_3d(self, sequence_name, mat_data, json_data):
        try:
            nX, nY, nZ, d_scans, nContr, spoil = self.read_consts_3d(json_data)  # Nss = nSlice

            if self.phase_shift == False:
                spoil = 0

            '''
                all_data = [x, y, z, contrast, coil]
            '''

            if sequence_name == "linear_decart":
                all_data = gather_data_along_trajectory(mat_data, nX, nY, nZ, nContr)
                save_ffft_3d(all_data, spoil)

            elif sequence_name == "nonlinear_decart(tse)":
                all_data = gather_data_along_trajectory_nonlinear(mat_data, self.k_space_order, nX, nY, nZ, nContr)
                save_ffft_3d(all_data, spoil)

            else:
                raise NotImplementedError(
                    "Для данного типа ИП еще не добавлена функция.")  # TODO уже можно добавлять функцию, так как она работает
                # QMessageBox.critical(self, "Ошибка: для данного типа ИП еще не добавлена функция")

        except ValueError as ve:
            raise ValueError(str(ve))

        except NotImplementedError as nie:
            raise NotImplementedError(str(nie))

        except Exception as e:
            raise RuntimeError(f"Произошла ошибка в процессе реконструкции: {e}")