Source code for deepmr.io.header.api

"""Acquisition Header API."""

__all__ = ["read_acqheader", "write_acqheader"]

import copy
import os
import time

import numpy as np

from . import base as _base

# from . import bart as _bart
from . import matlab as _matlab
from . import mrd as _mrd


[docs]def read_acqheader(filepath, *args, device="cpu", verbose=False, **kwargs): """ Read acquisition header from file. The header info (e.g., k-space trajectory, shape) can be used to simulate acquisitions or to inform raw data loading (e.g., via ordering) to reshape from acquisition to reconstruction ordering and image post-processing (transposition, flipping) and exporting. Parameters ---------- filepath : str Path to acquisition header file. Supports wildcard (e.g., ``/path-to-data-folder/*.h5``). *args Variable length argument list passed to the specific subroutines for the different datatypes (see ``Keyword Arguments``). device : str, optional Computational device for internal attributes. The default is ``cpu``. verbose : int, optional Verbosity level ``(0=Silent, 1=Less, 2=More)``. The default is ``0``. Keyword Arguments ----------------- dcfpath : str, optional Path to the dcf file (:func:`deepmr.io.matlab.read_matlab_acqhead`). The default is None. methodpath : str, optional Path to the schedule description file (:func:`deepmr.io.matlab.read_matlab_acqhead`). The default is None. sliceprofpath : str, optional Path to the slice profile file (:func:`deepmr.io.matlab.read_matlab_acqhead`). The default is None. Returns ------- head : Header Deserialized acquisition header. Notes ----- The returned ``head`` (:func:`deepmr.Header`) is a structure with the following fields: * shape (torch.Tensor): This is the expected image size of shape ``(nz, ny, nx)``. * resolution (torch.Tensor): This is the expected image resolution in mm of shape ``(dz, dy, dx)``. * t (torch.Tensor): This is the readout sampling time ``(0, t_read)`` in ``ms``. with shape ``(nsamples,)``. * traj (torch.Tensor): This is the k-space trajectory normalized as ``(-0.5 * shape, 0.5 * shape)`` with shape ``(ncontrasts, nviews, nsamples, ndims)``. * dcf (torch.Tensor): This is the k-space sampling density compensation factor with shape ``(ncontrasts, nviews, nsamples)``. * FA (torch.Tensor, float): This is either the acquisition flip angle in degrees or the list of flip angles of shape ``(ncontrasts,)`` for each image in the series. * TR (torch.Tensor, float): This is either the repetition time in ms or the list of repetition times of shape ``(ncontrasts,)`` for each image in the series. * TE (torch.Tensor, float): This is either the echo time in ms or the list of echo times of shape ``(ncontrasts,)`` for each image in the series. * TI (torch.Tensor, float): This is either the inversion time in ms or the list of inversion times of shape ``(ncontrasts,)`` for each image in the series. * user (dict): User parameters. Some examples are: * ordering (torch.Tensor): Indices for reordering (acquisition to reconstruction) of acquired k-space data, shaped ``(3, nslices * ncontrasts * nview)``, whose rows are ``contrast_index``, ``slice_index`` and ``view_index``, respectively. * mode (str): Acquisition mode (``2Dcart``, ``3Dcart``, ``2Dnoncart``, ``3Dnoncart``). * separable (bool): Whether the acquisition can be decoupled by fft along ``slice`` / ``readout`` directions (3D stack-of-noncartesian / 3D cartesian, respectively) or not (3D noncartesian and 2D acquisitions). * slice_profile (torch.Tensor): Flip angle scaling along slice profile of shape ``(nlocs,)``. * basis (torch.Tensor): Low rank subspace basis for subspace reconstruction of shape ``(ncontrasts, ncoeff)``. * affine (np.ndarray): Affine matrix describing image spacing, orientation and origin of shape ``(4, 4)``. * ref_dicom (pydicom.Dataset): Template dicom for image export. * flip (list): List of spatial axis to be flipped after image reconstruction. The default is an empty list (no flipping). * transpose (list): Permutation of image dimensions after reconstruction, depending on acquisition mode: * **2Dcart:** reconstructed image has ``(nslices, ncontrasts, ny, nx) -> transpose = [1, 0, 2, 3]`` * **2Dnoncart:** reconstructed image has ``(nslices, ncontrasts, ny, nx) -> transpose = [1, 0, 2, 3]`` * **3Dcart:** reconstructed image has ``(ncontrasts, nz, ny, nx) -> transpose = [0, 1, 2, 3]`` * **3Dnoncart:** reconstructed image has ``(nx, ncontrasts, nz, ny) -> transpose = [1, 2, 3, 0]`` The default is an empty list (no transposition). """ tstart = time.time() if verbose >= 1: print(f"Reading acquisition header from file {filepath}...", end="\t") done = False # mrd if filepath.endswith(".h5"): try: head = _mrd.read_mrd_acqhead(filepath) done = True except Exception as e: msg0 = e else: msg0 = "" # matfile if filepath.endswith(".mat") and not (done): try: head = _matlab.read_matlab_acqhead(filepath, *args, **kwargs) done = True except Exception as e: msg1 = _get_error(e) else: msg1 = "" # bart # if filepath.endswith(".cfl") and not(done): # try: # head = _bart.read_bart_acqhead(filepath) # done = True # except Exception: # pass # fallback if filepath.endswith(".h5") and not (done): try: done = True head = _base.read_base_acqheader(filepath) except Exception as e: msg2 = _get_error(e) else: msg2 = "" # check if we loaded data if not (done): raise RuntimeError( f"File (={filepath}) not recognized! Error:\nMRD {msg0}\nMATLAB {msg1}\nBase {msg2}" ) # normalize trajectory if head.traj is not None: ndim = head.traj.shape[-1] traj_max = ((head.traj**2).sum(axis=-1) ** 0.5).max() head.traj = head.traj / (2 * traj_max) # normalize to (-0.5, 0.5) head.traj = head.traj * head.shape[-ndim:] # cast head.torch(device) # final report if verbose == 2: print(f"Readout time: {round(float(head.t[-1]), 2)} ms") if head.traj is not None: print(f"Trajectory range: ({head.traj.min()},{head.traj.max()})") print( f"Trajectory shape: (ncontrasts={head.traj.shape[0]}, nviews={head.traj.shape[1]}, nsamples={head.traj.shape[2]}, ndim={head.traj.shape[-1]})" ) if head.dcf is not None: print( f"DCF shape: (ncontrasts={head.dcf.shape[0]}, nviews={head.dcf.shape[1]}, nsamples={head.dcf.shape[2]})" ) if head.FA is not None: if len(np.unique(head.FA)) > 1: if verbose == 2: print(f"Flip Angle train length: {len(head.FA)}") else: FA = float(np.unique(head.FA)[0]) head.FA = FA if verbose == 2: print(f"Constant FA: {round(abs(FA), 2)} deg") if head.TR is not None: if len(np.unique(head.TR)) > 1: if verbose == 2: print(f"TR train length: {len(head.TR)}") else: TR = float(np.unique(head.TR)[0]) head.TR = TR if verbose == 2: print(f"Constant TR: {round(TR, 2)} ms") if head.TE is not None: if len(np.unique(head.TE)) > 1: if verbose == 2: print(f"Echo train length: {len(head.TE)}") else: TE = float(np.unique(head.TE)[0]) head.TE = TE if verbose == 2: print(f"Constant TE: {round(TE, 2)} ms") if head.TI is not None and np.allclose(head.TI, 0.0) is False: if len(np.unique(head.TI)) > 1: if verbose == 2: print(f"Inversion train length: {len(head.TI)}") else: TI = float(np.unique(head.TI)[0]) head.TI = TI if verbose == 2: print(f"Constant TI: {round(TI, 2)} ms") tend = time.time() if verbose >= 1: print(f"done! Elapsed time: {round(tend-tstart, 2)} s...") return head
[docs]def write_acqheader(filename, head, filepath="./", dataformat="hdf5"): """ Write acquisition header to file. Supported output format are ``hdf5`` (faster i/o) and ``mrd``. Parameters ---------- filename : str Name of the file. head: Header Structure containing trajectory of shape ``(ncontrasts, nviews, npts, ndim)`` and meta information (shape, resolution, spacing, etc). filepath : str, optional Path to file. The default is ``./``. dataformat : str, optional Available formats (``mrd`` or ``hdf5``). The default is ``hdf5``. """ head = copy.deepcopy(head) head.ref_dicom = None if dataformat == "hdf5": _base.write_base_acqheader(head, os.path.join(filepath, filename)) elif dataformat == "mrd": _mrd.write_mrd_acqhead(head, os.path.join(filepath, filename)) else: raise RuntimeError( f"Data format = {dataformat} not recognized! Please use 'mrd' or 'hdf5'" )
# %% sub routines def _get_error(ex): trace = [] tb = ex.__traceback__ while tb is not None: trace.append( { "filename": tb.tb_frame.f_code.co_filename, "name": tb.tb_frame.f_code.co_name, "lineno": tb.tb_lineno, } ) tb = tb.tb_next return str({"type": type(ex).__name__, "message": str(ex), "trace": trace})