IO

io

PulseSeqFile

PulseSeqFile(
    file_path: str = None,
    sequence: Sequence = None,
    version: str = None,
)

Read, write, and convert Pulseq .seq files.

PulseSeqFile can be constructed from either a Pulseq file path or a CMRseq :class:~cmrseq.Sequence. Export defaults to Pulseq 1.5.0. Pass version="1.4.0" when compatibility with older Pulseq tooling is required. Pulseq 1.5.0 output includes RF center/use fields and trigger extensions.

Create a Pulseq adapter from a file or CMRseq sequence.

Parameters:

Name	Type	Description	Default
file_path	str	Path to an existing Pulseq .seq file to parse.	None
sequence	Sequence	Sequence object to convert into Pulseq tables.	None
version	str	Pulseq version to write when sequence is provided. Defaults to "1.5.0".	None

Raises:

Type	Description
ValueError	If neither source is provided, both sources are provided, or a write version is supplied for file parsing.

Source code in cmrseq/io/_pulseq.py

def __init__(self, file_path: str = None, sequence: Sequence = None, version: str = None):
    r"""Create a Pulseq adapter from a file or CMRseq sequence.

    Parameters
    ----------
    file_path
        Path to an existing Pulseq ``.seq`` file to parse.
    sequence
        Sequence object to convert into Pulseq tables.
    version
        Pulseq version to write when ``sequence`` is provided. Defaults to ``"1.5.0"``.

    Raises
    ------
    ValueError
        If neither source is provided, both sources are provided, or a write version is supplied
        for file parsing.
    """
    if (file_path is not None and sequence is not None) or \
            (file_path is None and sequence is None):
        raise ValueError("Exactly one of the input sources must be specified.")

    if file_path is not None and version is not None:
        raise ValueError("If a file_path is specified, the version is set by the file itself.")

    if version is None:
        self.version = "1.5.0"
    else:
        self.version = version

    self.shape_table = OrderedDict()  # hash_val: (id, n_samples, 1d-arr)
    self.shape_hash_table = OrderedDict()
    self.rf_table = OrderedDict()  # hash_val: (id, amp, mag_id, phs_id, time_id, delay, freq, phase)
    self.rf_hash_table = OrderedDict()
    self.traps_table = OrderedDict()
    self.traps_hash_table = OrderedDict()
    self.grads_table = OrderedDict()
    self.grads_hash_table = OrderedDict()
    self.adc_table = OrderedDict()
    self.adc_hash_table = OrderedDict()
    self.ext_table = OrderedDict()
    self.ext_hash_table = OrderedDict()
    self.labelinc_table = OrderedDict()
    self.labelinc_hash_table = OrderedDict()
    self.labelset_table = OrderedDict()
    self.labelset_hash_table = OrderedDict()
    self.trigger_table = OrderedDict()
    self.trigger_hash_table = OrderedDict()

    if file_path is not None:
        self.from_pulseq_file(file_path)
    elif sequence is not None:
        self.from_sequence(sequence)

SECTION_HEADERS class-attribute instance-attribute

SECTION_HEADERS: Tuple[str, ...] = (
    "[VERSION]",
    "[DEFINITIONS]",
    "[BLOCKS]",
    "[GRADIENTS]",
    "[RF]",
    "[TRAP]",
    "[ADC]",
    "[EXTENSIONS]",
    "[SHAPES]",
    "[SIGNATURE]",
)

REQUIRED_DEFINITIONS class-attribute instance-attribute

REQUIRED_DEFINITIONS: Tuple[str, ...] = (
    "AdcRasterTime",
    "BlockDurationRaster",
    "GradientRasterTime",
    "RadiofrequencyRasterTime",
)

version instance-attribute

version: str

raster_times instance-attribute

raster_times: dict

additional_defs instance-attribute

additional_defs: dict

block_array instance-attribute

block_array: ndarray

shape_table instance-attribute

shape_table: OrderedDict = OrderedDict()

rf_table instance-attribute

rf_table: OrderedDict = OrderedDict()

adc_table instance-attribute

adc_table: OrderedDict = OrderedDict()

traps_table instance-attribute

traps_table: OrderedDict = OrderedDict()

grads_table instance-attribute

grads_table: OrderedDict = OrderedDict()

ext_table instance-attribute

ext_table: OrderedDict = OrderedDict()

labelset_table instance-attribute

labelset_table: OrderedDict = OrderedDict()

labelinc_table instance-attribute

labelinc_table: OrderedDict = OrderedDict()

shape_hash_table instance-attribute

shape_hash_table = OrderedDict()

rf_hash_table instance-attribute

rf_hash_table = OrderedDict()

traps_hash_table instance-attribute

traps_hash_table = OrderedDict()

grads_hash_table instance-attribute

grads_hash_table = OrderedDict()

adc_hash_table instance-attribute

adc_hash_table = OrderedDict()

ext_hash_table instance-attribute

ext_hash_table = OrderedDict()

labelinc_hash_table instance-attribute

labelinc_hash_table = OrderedDict()

labelset_hash_table instance-attribute

labelset_hash_table = OrderedDict()

trigger_table instance-attribute

trigger_table = OrderedDict()

trigger_hash_table instance-attribute

trigger_hash_table = OrderedDict()

from_pulseq_file

from_pulseq_file(filepath: str)

Load a .seq file and parse supported Pulseq sections.

Parameters:

Name	Type	Description	Default
filepath	str	path to a file of type .seq	required

Raises:

Type	Description
ValueError	If filepath does not exist.

Source code in cmrseq/io/_pulseq.py

def from_pulseq_file(self, filepath: str):
    r"""Load a ``.seq`` file and parse supported Pulseq sections.

    Parameters
    ----------
    filepath
        path to a file of type .seq

    Raises
    ------
    ValueError
        If ``filepath`` does not exist.
    """

    if not os.path.exists(filepath):
        raise ValueError(f"No pulseq file found at specified location:\n\t{filepath}")

    with open(filepath, "r") as seqfile:
        all_lines = seqfile.read().splitlines()
    all_lines = [re.sub(r'\s+', ' ', line.strip()) for line in all_lines]

    # Find section starts and calculate number of lines per section
    sections = self._subdivide_sections(all_lines)

    # Parse Meta information (version and definitions)
    self.version = self._parse_version(sections["[VERSION]"])
    self.raster_times, self.additional_defs = self._parse_definitions(sections["[DEFINITIONS]"])

    # Parse block definitions
    self.block_array = np.genfromtxt(sections["[BLOCKS]"], comments="#",
                                     delimiter=" ", dtype=int)

    # Parse lookup tables for block definitions
    shape_table, rf_table, adc_table, traps_table, grads_table, ext_table = [{} for _ in
                                                                             range(6)]
    if "[SHAPES]" in sections.keys():
        shape_table = self._parse_shapes(sections["[SHAPES]"])
    self.shape_table = shape_table

    if "[RF]" in sections.keys():
        # If any RF is specified, SHAPES must be present in definitons as well
        rf_table = self._parse_rf(sections["[RF]"], self.version)
    self.rf_table = rf_table

    if "[ADC]" in sections.keys():
        adc_table = self._parse_adc(sections["[ADC]"], self.version)
    self.adc_table = adc_table

    if "[TRAP]" in sections.keys():
        traps_table = self._parse_traps(sections["[TRAP]"])
    self.traps_table = traps_table

    if "[GRADIENTS]" in sections.keys():
        grads_table = self._parse_gradients(sections["[GRADIENTS]"], self.version)
    self.grads_table = grads_table

    if "[EXTENSIONS]" in sections.keys():
        print(2)
    self.ext_table = ext_table

write

write(filepath: str, compress_shapes: bool = False)

Writes the current sequence definition to a file in the Pulseq format.

Parameters:

Name	Type	Description	Default
filepath	str	Output .seq path.	required
compress_shapes	bool	If True, write compressed shape definitions.	False

Returns:

Type	Description
None

Raises:

Type	Description
ValueError	if file at specified location already exists

Source code in cmrseq/io/_pulseq.py

def write(self, filepath: str, compress_shapes: bool = False):
    r"""Writes the current sequence definition to a file in the Pulseq format.

    Parameters
    ----------
    filepath
        Output ``.seq`` path.
    compress_shapes
        If ``True``, write compressed shape definitions.

    Returns
    -------
    None

    Raises
    ------
    ValueError
        if file at specified location already exists
    """
    # if os.path.exists(filepath):
    #     raise ValueError("File at specified location already exists")

    version_sec = self._format_version(self.version)
    definition_sec = self._format_definitions(self.raster_times, self.additional_defs)
    block_sec = self._format_blocks_def(self.block_array)
    rf_sec = self._format_rf(self.rf_table, self.version)
    grad_sec = self._format_gradients(self.grads_table, self.version)
    trap_sec = self._format_traps(self.traps_table)
    adc_sec = self._format_adc(self.adc_table, self.version)
    ext_sec = self._format_ext(self.ext_table)
    inc_sec = self._format_inclabels(self.labelinc_table)
    set_sec = self._format_setlabels(self.labelset_table)
    trig_sec = self._format_triggers(self.trigger_table)
    shape_sec = self._format_shapes(self.shape_table, compress_shapes)

    total = "\n".join([version_sec, definition_sec, block_sec, rf_sec,
                       grad_sec, trap_sec, adc_sec, ext_sec, inc_sec, set_sec, trig_sec, shape_sec])

    total = self._sign_definition(total)

    with open(filepath, "w+") as wfile:
        wfile.write(total)

to_cmrseq

to_cmrseq(
    system_specs: SystemSpec,
    block_indices: Iterable[int] = None,
) -> List[Sequence]

Converts the parsed file into a list of cmrseq.Sequence objects.

Parameters:

Name	Type	Description	Default
system_specs	SystemSpec	System limits used to construct CMRseq blocks.	required
block_indices	Iterable[int]	Iterable[int] specifiying which blocks to convert if None all blocks are converted	None

Returns:

Type	Description
List of cmrseq.Sequence each representing one block of the pulseseq definition

Source code in cmrseq/io/_pulseq.py

def to_cmrseq(self, system_specs: SystemSpec, block_indices: Iterable[int] = None) -> List[
    Sequence]:
    r"""Converts the parsed file into a list of cmrseq.Sequence objects.

    Parameters
    ----------
    system_specs
        System limits used to construct CMRseq blocks.
    block_indices
        Iterable[int] specifiying which blocks to convert if None all blocks are converted

    Returns
    -------
    List of cmrseq.Sequence each representing one block of the pulseseq definition
    """
    if block_indices is None:
        block_indices = range(self.block_array.shape[0])

    sequence_objects = []
    # Assumption: each block can contain one each of the classes (RF, GX, GY, GZ, ADC, EXT)
    for idx in tqdm(block_indices, desc="Converting block definitons to CMRseq objects"):
        sequence_blocks = []
        block_def = self.block_array[idx]

        # Construct RF
        rf_def = self.rf_table.get(block_def[2], None)
        if rf_def is not None:
            rf_object = self._rfdef_to_block(rf_def, system_specs, name=f"rf_id_{block_def[2]}")
            sequence_blocks.append(rf_object)

        # Construct Gradients
        gradients_per_dir = self._graddef_to_block(block_def, system_specs)
        sequence_blocks.extend(gradients_per_dir)

        # Construct ADC
        adc_def = self.adc_table.get(block_def[6], None)
        if adc_def is not None:
            adc_object = bausteine.SymmetricADC(system_specs=system_specs,
                                                num_samples=adc_def['num_samples'],
                                                dwell=adc_def['dwell'],
                                                delay=adc_def['delay'],
                                                frequency_offset=adc_def['frequency_offset'],
                                                phase_offset=adc_def['phase_offset'])
            sequence_blocks.append(adc_object)

        # Only block duration is specified --> Delay
        if len(sequence_blocks) == 0:
            sequence_blocks.append(bausteine.Delay(system_specs=system_specs,
                                                   duration=float(block_def[1]) *
                                                            self.raster_times["blocks"]))

        # Pulseq files can add a delay by specifying a block duration that is longer
        # than all contained events. In this case a padding with a delay is necessary
        target_block_dur = block_def[1] * self.raster_times["blocks"].to("ms")
        max_block_dur = Quantity(max([b.tmax.m_as("ms") for b in sequence_blocks]), "ms")
        if target_block_dur.m_as("ms") - max_block_dur.m_as("ms") > 1e-6:
            sequence_blocks.append(bausteine.Delay(system_specs=system_specs,
                                                   duration=target_block_dur - max_block_dur,
                                                   delay=max_block_dur))

        sequence_objects.append(Sequence(sequence_blocks, system_specs=system_specs))
    return sequence_objects

from_sequence

from_sequence(sequence: Sequence)

Creates a pulseq-style sequence definition from a cmrseq.sequence object.

Source code in cmrseq/io/_pulseq.py

def from_sequence(self, sequence: Sequence):
    r"""Creates a pulseq-style sequence definition from a cmrseq.sequence object.

    """

    # Initialize label counters to zero
    self.counters = dict(kspace_encode_step_1=0,
                          kspace_encode_step_2=0,
                          average=0,
                          slice=0,
                          contrast=0,
                          phase=0,
                          repetition=0,
                          set=0,
                          segment=0)

    # Conversion dictionary from cmrseq(ISMRMRD) label names to pulseq label names
    self.pulseq_labels = dict(kspace_encode_step_1="LIN",
                         kspace_encode_step_2="PAR",
                         slice="SLC",
                         segment="SEG",
                         repetition="REP",
                         average="AVG",
                         set="SET",
                         contrast="ECO",
                         phase="PHS")

    self.raster_times = dict(rf=sequence._system_specs.rf_raster_time,
                             grad=sequence._system_specs.grad_raster_time,
                             adc=sequence._system_specs.adc_raster_time)
    self.raster_times["blocks"] = max(self.raster_times.values())

    self.additional_defs = {}

    # self.additional_defs = {k: sequence._system_specs.__getattribute__(k) for k in
    #                         ("rf_dead_time", "rf_ringdown_time", "rf_lead_time",
    #                          "adc_dead_time", "rf_peak_power", "max_grad", "max_slew")}

    rf_dead_time = np.round(sequence._system_specs.rf_dead_time.m_as("ms"), decimals=8)
    adc_dead_time = np.round(sequence._system_specs.adc_dead_time.m_as("ms"), decimals=8)

    self.block_array = []

    # Create a set of double ended cues to 'stream'-interleave the event definitions, based
    # on the start-time of the blocks
    t, (gx, gy, gz) = sequence.combined_gradients()

    gx_que = deque(cmrseq.utils.find_gradient_blocks(t, gx)
                   + [("dummy", (sys.float_info.max,))])
    gy_que = deque(cmrseq.utils.find_gradient_blocks(t, gy)
                   + [("dummy", (sys.float_info.max,))])
    gz_que = deque(cmrseq.utils.find_gradient_blocks(t, gz)
                   + [("dummy", (sys.float_info.max,))])
    rf_que = deque(sequence.get_block(typedef=cmrseq.bausteine.RFPulse)
                   + [SimpleNamespace(tmin=Quantity(sys.float_info.max, "ms"))])
    adc_que = deque(sequence.get_block(typedef=cmrseq.bausteine.ADC)
                    + [SimpleNamespace(tmin=Quantity(sys.float_info.max, "ms"))])
    all_queues: dict[str: deque] = dict(RF=rf_que, GX=gx_que, GY=gy_que, GZ=gz_que, ADC=adc_que)
    key_list = list(all_queues.keys())

    # Combine queues as (blocktype, tmin, tmax, block_def)
    combined_que: deque[(str, float, float, Any)] = deque()
    while any([len(_) > 1 for _ in all_queues.values()]):
        # Dead time is accounted for during block sorting
        min_idx = min(range(5), key=lambda x: [np.round(rf_que[0].tmin.m_as("ms") - rf_dead_time, decimals=8),
                                               gx_que[0][1][0],
                                               gy_que[0][1][0],
                                               gz_que[0][1][0],
                                               np.round(adc_que[0].tmin.m_as("ms") - adc_dead_time, decimals=8)][x])
        blocktype = key_list[min_idx]
        block_def = all_queues[blocktype].popleft()

        if blocktype in ["RF", "ADC"]:
            block_min = block_def.tmin.m_as("ms")
            block_max = block_def.tmax.m_as("ms")
            # Special case for trigger labels.
            # if block_def.label._labels["triggerWait"] is not None or block_def.label._labels["triggerSend"] is not None:
            #     block_min = np.round(block_def.tmin_block.m_as("ms"), decimals=8)

            # Dead times can be added here, to ensure that the dead times are obeyed by the splitting algorithm
            if blocktype == "RF":
                # RF dead time only at the start of the block.
                block_max += rf_dead_time
                block_min -= rf_dead_time
            else:
                # ADC dead time on either side
                block_max += adc_dead_time
                block_min -= adc_dead_time

            block_min = np.round(block_min, decimals=8)
            block_max = np.round(block_max, decimals=8)

        else:
            block_min = block_def[1][0]
            block_max = block_def[1][-1]

        # # round min max to raster
        # block_min = np.around(np.floor(
        #         np.around(block_min / self.raster_times["blocks"].m_as("ms"), decimals=6)
        #         ) * self.raster_times["blocks"].m_as("ms"), decimals=6)

        # block_max = np.around(np.floor(
        #         np.around(block_max / self.raster_times["blocks"].m_as("ms"), decimals=6)
        #         ) * self.raster_times["blocks"].m_as("ms"), decimals=6)

        # A negative block min might happen if a sequence is badly constructed (starts with an RF pulse)
        # If this is the case we throw an error
        if block_min < 0:
            raise ValueError(f"Encountered negative block start time trigger by {blocktype}. Ensure the sequence does not start with an RF or ADC without an appropriate delay.")
        combined_que.append((blocktype, block_min, block_max, block_def))
    combined_que.append(("GX",sys.float_info.max,sys.float_info.max, ("dummy", (sys.float_info.max,))))
    total_number_of_blocks = len(combined_que) - 1

    # In a loop add event definitions into the current block entry, and finalize the block
    # if block borders are detected (consecutive events on the same channel)
    current_block_entry = {"RF": None, "GX": None, "GY": None, "GZ": None,
                           "ADC": None, "EXT": None}
    progress_bar = tqdm(range(total_number_of_blocks),
                        desc="Converting to Pulseq blocks")

    tmp_type_que = deque()
    tmp_def_que = deque()

    # Before starting, need to take care of pure delays at the start
    delay_time = combined_que[0][1]
    delay_time = np.around(np.floor(
        np.around(delay_time / self.raster_times["blocks"].m_as("ms"), decimals=6)
        ) * self.raster_times["blocks"].m_as("ms"), decimals=6)
    if delay_time > 0:
        self._register_delay_block(delay_time)

    while len(combined_que) > 1:
        progress_bar.n = total_number_of_blocks - len(combined_que)
        progress_bar.refresh()
        ## Pull blocks until a collision by type is found
        while combined_que[0][0] not in tmp_type_que and len(combined_que) > 1:
            curr_type, new_block_min, new_block_max, block_def = combined_que.popleft()
            tmp_type_que.append(curr_type)
            tmp_def_que.append((new_block_min, new_block_max, block_def))

        # End of queue reached
        if len(combined_que) == 1:
            current_block_entry.update({btype: bdef for btype, (_, _, bdef)
                                        in zip(tmp_type_que, tmp_def_que)})
            block_start = tmp_def_que[0][0]
            block_end = np.max([b[1] for b in tmp_def_que])
            # Add some extra raster to block_end to account for possible ADC dead time
            # This is ok, since it is the end of the sequence?????
            block_end = block_end + self.raster_times["blocks"].m_as('ms')
            self._register_block(current_block_entry, block_start, block_end - block_start,
                                 sequence._system_specs)
            current_block_entry = {"RF": None, "GX": None, "GY": None, "GZ": None,
                                   "ADC": None, "EXT": None}
            break

        # Sometimes the set of blocks can be split into two groups, if there is no overlap
        # While this increases the number of blocks, it produces a more intuitive splitting
        blocks_starts = np.around(np.array([b[0] for b in tmp_def_que]), decimals=8)
        blocks_ends = np.around(np.array([b[1] for b in tmp_def_que]), decimals=8)
        max_block_end = blocks_ends[0]
        for idx in range(1,len(blocks_starts)):
            if blocks_starts[idx]<max_block_end:
                # this block is in the keep group
                if max_block_end<blocks_ends[idx]:
                    max_block_end = blocks_ends[idx]

        while tmp_def_que[-1][0] >= max_block_end:
            combined_que.appendleft((tmp_type_que.pop(), *tmp_def_que.pop()))

        # This is the start of the next block in the queue
        coliding_block_min = combined_que[0][1]

        # Easiest case: # All block are ending before the next one starts:
        if all([b[1] <= coliding_block_min for b in tmp_def_que]):
            current_block_entry.update({btype: bdef for btype, (_, _, bdef)
                                        in zip(tmp_type_que, tmp_def_que)})
            block_start = tmp_def_que[0][0]
            # If these blocks all end before the next one starts, we can add a delay
            blocks_ends = np.around(np.array([b[1] for b in tmp_def_que]), decimals=8)
            block_end = np.max(blocks_ends)
            block_end = np.around(np.ceil(
                np.around(block_end / self.raster_times["blocks"].m_as("ms"), decimals=6)
                ) * self.raster_times["blocks"].m_as("ms"), decimals=6)
            # Add a delay block to fill the gap
            # Needs to be on block raster
            delay_time = coliding_block_min - block_end
            delay_time = np.around(np.floor(
                np.around(delay_time / self.raster_times["blocks"].m_as("ms"), decimals=6)
                ) * self.raster_times["blocks"].m_as("ms"), decimals=6)

            self._register_block(current_block_entry, block_start, block_end - block_start,
                                 sequence._system_specs)
            # This delay happens after the block
            if delay_time > 0:
                self._register_delay_block(delay_time)

            current_block_entry = {"RF": None, "GX": None, "GY": None, "GZ": None,
                                   "ADC": None, "EXT": None}
            tmp_def_que.clear(), tmp_type_que.clear()
            continue


        # Collision can't be solved by poping blocks from queue, hence splitting the gradients
        # in the current block definition at the colliding definition start time.
        # Assumptions for this block collision:
        # 1. RF and ADCs have absolute priority for definining block borders, they are
        # always fully contained in a block.
        # 2. Gradients are allowed to be split. Hence, when a block collision occurs,
        # All gradient channels of the previous block are split such that the second part is
        # included into the next block as arbitrary waveform.
        else:
            current_block_entry.update(
                {btype: bdef for btype, (_, _, bdef) in zip(tmp_type_que, tmp_def_que)})
            block_start = tmp_def_que[0][0]
            block_end = coliding_block_min
            splitting_time = np.around(np.floor(
                np.around(block_end / self.raster_times["blocks"].m_as("ms"), decimals=6)
            ) * self.raster_times["blocks"].m_as("ms"), decimals=6)

            # There is an edge case, where multiple gradient blocks occur on the same channel during the RF pulse or ADC
            # In this case, these blocks need to be merged and split at the end of the RF pulse
            if current_block_entry["RF"] is not None:
                # Update splitting time if RF pulse finishes after the collision
                splitting_time = np.maximum(current_block_entry["RF"].tmax.m_as("ms")+rf_dead_time,splitting_time)
            if current_block_entry["ADC"] is not None:
                # Update splitting time if ADC finishes after the collision
                splitting_time = np.maximum(current_block_entry["ADC"].tmax.m_as("ms")+adc_dead_time,splitting_time)
            # Splitting time must sit on block rater
            # There are edge cases when ADCs are on a raster lower than block raster and they are too close
            # But in this case, no sequence can be written.
            splitting_time = np.around(np.ceil(np.around(splitting_time / self.raster_times["blocks"].m_as("ms"), decimals=6)) *
                                       self.raster_times["blocks"].m_as("ms"), decimals=6)

            # Go through que, popping objects that start before the splitting time
            while (combined_que[0][1] < splitting_time) and len(combined_que) > 1:

                btype = combined_que[0][0]
                if btype == "RF":
                    if current_block_entry["RF"] is not None:
                        # RF pulse already in block, meaning there is a problem
                        # Either due to overlap of RF pulses, or an ADC object overlapping 2 RF pulses
                        raise ValueError("RF pulse overlap (splitting time: {}, colliding block start time: {})".format(splitting_time, combined_que[0][1]))
                    else:
                        # The next block is a RF pulse, and overlaps with the current splitting time
                        # We add it to the current block entry, and update the splitting time
                        curr_type, new_block_min, new_block_max, block_def = combined_que.popleft()
                        tmp_type_que.append(curr_type)
                        tmp_def_que.append((new_block_min, new_block_max, block_def))
                        current_block_entry.update({btype: bdef for btype, (_, _, bdef) in zip(tmp_type_que, tmp_def_que)})
                        splitting_time = np.maximum(current_block_entry["RF"].tmax.m_as("ms"),splitting_time)

                # Similar to above, but for ADC
                elif btype == "ADC":
                    if current_block_entry["ADC"] is not None:
                        # ADC already in block, meaning there is a problem
                        # Either due to overlap of ADCs, or a RF object overlapping 2 ADCs
                        raise ValueError("ADC overlap")
                    else:
                        # The next block is an ADC, and overlaps with the current splitting time
                        # We add it to the current block entry, and update the splitting time
                        curr_type, new_block_min, new_block_max, block_def = combined_que.popleft()
                        tmp_type_que.append(curr_type)
                        tmp_def_que.append((new_block_min, new_block_max, block_def))
                        current_block_entry.update({btype: bdef for btype, (_, _, bdef) in zip(tmp_type_que, tmp_def_que)})
                        splitting_time = np.maximum(current_block_entry["ADC"].tmax.m_as("ms"),splitting_time)

                else: # The next block is a gradient
                    if current_block_entry[btype] is None:
                        # Gradient not in block, meaning there is no problem
                        # We add it to the current block entry, and update the splitting time
                        curr_type, new_block_min, new_block_max, block_def = combined_que.popleft()
                        tmp_type_que.append(curr_type)
                        tmp_def_que.append((new_block_min, new_block_max, block_def))
                        current_block_entry.update({btype: bdef for btype, (_, _, bdef) in zip(tmp_type_que, tmp_def_que)})
                        # In some cases, this gradient block starts only a few samples before the splitting time
                        # This can result in < 2 samples of gradient, which may not be handled well by pulseq
                        # Hence, we check this case and if needed increase the splitting time
                        if splitting_time - new_block_min < self.raster_times["grad"].m_as("ms") * 2:
                            ext_time = int(np.ceil((splitting_time - new_block_min)/ self.raster_times["grad"].m_as("ms")))
                            ext_time = ext_time * self.raster_times["grad"].m_as("ms")
                            splitting_time = new_block_min + ext_time
                    else:
                        # Gradient already in block, so they need to be merged
                        _, _, _, block_def = combined_que.popleft()
                        block_merged = self._merge_gradient_events(current_block_entry[btype], block_def)
                        # Get index in que
                        idx = tmp_type_que.index(btype)
                        # Update the block definition
                        tmp_def_que[idx] = (block_merged[1][0],block_merged[1][-1], block_merged)
                        current_block_entry.update({btype: bdef for btype, (_, _, bdef) in zip(tmp_type_que, tmp_def_que)})
                        # We do not update splitting time, since we can split gradients

            for btype, (bmin, bmax, bdef) in zip(tmp_type_que, tmp_def_que):
                if btype in ("GX", "GY", "GZ") and bmax > splitting_time:
                    # If point is already contained in the definition, use it as index
                    if np.any(is_close := np.isclose(bdef[1], splitting_time,rtol=0)):
                        insertion_index = np.squeeze(np.argwhere(is_close))
                        insertion_val = bdef[2][insertion_index]
                        tmp_wf = bdef[2]
                        tmp_t = bdef[1]
                    # Else interpolate the waveform, insert the point and split at this location
                    else:
                        insertion_index = np.searchsorted(bdef[1], splitting_time)
                        insertion_val = np.interp(splitting_time, bdef[1], bdef[2])
                        tmp_wf = np.insert(bdef[2], insertion_index, insertion_val)
                        tmp_t = np.insert(bdef[1], insertion_index, splitting_time)
                    current_block_entry[btype] = ("arbitrary", tmp_t[:insertion_index + 1],
                                                  tmp_wf[:insertion_index + 1])
                    combined_que.appendleft((btype, tmp_t[insertion_index], tmp_t[-1],
                                             ("arbitrary", tmp_t[insertion_index:],
                                              tmp_wf[insertion_index:])))

            self._register_block(current_block_entry, block_start, splitting_time - block_start,
                                 sequence._system_specs)
            current_block_entry = {"RF": None, "GX": None, "GY": None, "GZ": None,
                                   "ADC": None, "EXT": None}
            tmp_def_que.clear(), tmp_type_que.clear()

    self.block_array = np.stack(self.block_array)

shift_definition staticmethod

shift_definition(waveform: ndarray)

Source code in cmrseq/io/_pulseq.py

@staticmethod
def shift_definition(waveform: np.ndarray):
    x_old = np.arange(0, waveform.shape[0], dtype=np.float64)
    x_new = x_old[:-1] + 0.5
    return np.interp(x_new, x_old, waveform)

check_add_shape

check_add_shape(arr: ndarray) -> int

Checks if specified array is already in self.shape_table (if not adds it to the table) and returns the corresponding shape_id.

Lookup is performed by computing the hash value of the array which serves as key of the shape_table dictionary.

Parameters:

Name	Type	Description	Default
arr	ndarray		required

Returns:

Type	Description
shape_id int

Source code in cmrseq/io/_pulseq.py

def check_add_shape(self, arr: np.ndarray) -> int:
    r"""Checks if specified array is already in self.shape_table (if not adds it to the table)
    and returns the corresponding shape_id.

    Lookup is performed by computing the hash value of the array which serves as key of the
    shape_table dictionary.

    Parameters
    ----------
    arr

    Returns
    -------
    shape_id int
    """
    # round value to nearest 1e-6 to avoid floating point issues
    arr = np.around(arr + 0.0, decimals=6)
    hash_value = hash(arr.tobytes())
    existing_entry = self.shape_hash_table.get(hash_value, None)
    if existing_entry is None:
        shape_id = len(self.shape_table) + 1  # this should be O(1)
        self.shape_hash_table[hash_value] = shape_id
        self.shape_table[shape_id] = arr.flatten()
    else:
        shape_id = existing_entry
    return shape_id

check_add_def staticmethod

check_add_def(
    def_tuple: tuple,
    table: dict[int:tuple],
    last_id: int = None,
) -> int

Checks if definition tuple is already in self.rf_table (if not adds it to the table) and returns the corresponding rf_id.

Lookup is performed by computing the hash value of the stringified definition tuple which serves as key of the rf_table dictionary.

Parameters:

Name	Type	Description	Default
def_tuple	tuple	RF definition as specified in the Pulseq package (amp, magnitude_id, phase_id, time_id, delay, frequency_offset, phase_offset)	required
table	dict[int:tuple]	One of the following lookup dictionaries [file.rf_table, file.adc_table, file.trap_table, file.arb_table]	required
last_id	int	if specified, the new entry is inserted at last_id + 1 otherwise the last id is computed as the length of table.	None

Returns:

Type	Description
definition id is corresponding table

Source code in cmrseq/io/_pulseq.py

@staticmethod
def check_add_def(def_tuple: tuple, table: dict[int: tuple], last_id: int = None) -> int:
    r"""Checks if definition tuple is already in self.rf_table (if not adds it to the table)
    and returns the corresponding rf_id.

    Lookup is performed by computing the hash value of the stringified definition tuple
    which serves as key of the rf_table dictionary.


    Parameters
    ----------
    def_tuple
        RF definition as specified in the Pulseq package (amp, magnitude_id, phase_id, time_id, delay, frequency_offset, phase_offset)
    table
        One of the following lookup dictionaries [file.rf_table, file.adc_table, file.trap_table, file.arb_table]
    last_id
        if specified, the new entry is inserted at last_id + 1 otherwise the last id is computed as the length of table.

    Returns
    -------
    definition id is corresponding table
    """
    # deal with rounding issues and negative zeroes
    def_tuple = [np.round(x + 0.0, 6) if not isinstance(x, str) else x for x in def_tuple]
    hash_value = hash(str(def_tuple))

    existing_entry = table.get(hash_value, None)
    if existing_entry is None:
        if last_id is None:
            def_id = len(table) + 1  # this should be O(1)
        else:
            def_id = last_id + 1
        table[hash_value] = (def_id, *def_tuple)
    else:
        def_id = existing_entry[0]
    return def_id

sequence_to_json

sequence_to_json(
    sequence: Sequence, file_name: str = None
) -> OrderedDict

Converts a cmrseq.Sequence object to an ordered dictionary containing a JSON compatible representation of the sequence. The first node contains the system specifications and subsequent nodes contain the block definitions. After converting the sequence to a dict, this function serializes it to json format and saves it to the specified location (if not None).

Parameters:

Name	Type	Description	Default
sequence	Sequence		required
file_name	str	str defaults to None. Specifies the saving location. If not specified, the sequence representation is not saved but only returned as json compatible dictionary.	None

Returns:

Type	Description
object

Source code in cmrseq/io/_json.py

def sequence_to_json(sequence: cmrseq.Sequence, file_name: str = None) -> OrderedDict:
    r"""Converts a cmrseq.Sequence object to an ordered dictionary containing a JSON compatible
    representation of the sequence. The first node contains the system specifications and subsequent
    nodes contain the block definitions. After converting the sequence to a dict, this function
    serializes it to json format and saves it to the specified location (if not None).

    Parameters
    ----------
    sequence
    file_name
        str defaults to None. Specifies the saving location. If not specified, the sequence representation is not saved but only returned as json compatible dictionary.

    Returns
    -------
    object
    """
    sequence.validate()
    block_names = sequence.blocks
    blocks = [sequence.get_block(bn) for bn in block_names]
    save_dict = OrderedDict(system_specs=_specs_to_dict(sequence._system_specs))  # pylint: disable=W0212
    save_dict.update({bn: _block_to_dict(b) for bn, b in zip(block_names, blocks)})

    with open(f"{file_name}.json", "w+", encoding="utf-8") as filep:
        json.dump(save_dict, filep)
    return save_dict

sequence_from_json

sequence_from_json(file: str) -> cmrseq.Sequence

Loads json file from specified locations and reconstructs a Sequence object from it.

Parameters:

Name	Type	Description	Default
file	str	str file location containing the sequence definition	required

Returns:

Type	Description
cmrseq.Sequence object

Source code in cmrseq/io/_json.py

def sequence_from_json(file: str) -> 'cmrseq.Sequence':
    r"""Loads json file from specified locations and reconstructs a Sequence object from it.

    Parameters
    ----------
    file
        str file location containing the sequence definition

    Returns
    -------
    cmrseq.Sequence object
    """
    with open(file, "r", encoding="utf-8") as filep:
        sequence_dict = json.load(filep, object_pairs_hook=OrderedDict)
    specs = dict_to_specs(sequence_dict["system_specs"])
    blocks = [_dict_to_block(specs, b) for i, b in enumerate(sequence_dict.values()) if i > 0]
    return cmrseq.Sequence(blocks, system_specs=specs)

dict_to_specs

dict_to_specs(input_dict: dict) -> cmrseq.SystemSpec

Creates a new cmrseq.SystemSpecs object and writes all attributes from the specified input dictionary to the system specifications after converting each entry to a pint.Quantity

Parameters:

Name	Type	Description	Default
input_dict	dict		required

Returns:

Type	Description
object

Source code in cmrseq/io/_json.py

def dict_to_specs(input_dict: dict) -> cmrseq.SystemSpec:
    r"""Creates a new cmrseq.SystemSpecs object and writes all attributes from the specified input
    dictionary to the system specifications after converting each entry to a pint.Quantity

    Parameters
    ----------
    input_dict

    Returns
    -------
    object
    """
    specs = cmrseq.SystemSpec()
    for key, val in input_dict.items():
        if key == "enable_simulatenous_trasmit_receive":
            specs.enable_simulatenous_trasmit_receive = val
        else:
            specs.__dict__[key] = _dict_to_quantity(val)

    return specs