Added functionality to remove duplicate events from the event libraries

- Added `remove_duplicates` in `event_lib`
- Added `remove_duplicates` in `Sequence`
- Changed `write_seq` to call `remove_duplicates` before writing
- Changed `read_seq` to call `remove_duplicates` after reading

pypulseq/Sequence/read_seq.py

@@ -13,7 +13,7 @@
 from pypulseq.supported_labels_rf_use import get_supported_labels
 
 
-def read(self, path: str, detect_rf_use: bool = False) -> None:
+def read(self, path: str, detect_rf_use: bool = False, remove_duplicates: bool = True) -> None:
  """
  Load sequence from file - read the given filename and load sequence data into sequence object.
 
@@ -26,6 +26,8 @@ def read(self, path: str, detect_rf_use: bool = False) -> None:
  detect_rf_use : bool, default=False
  Boolean flag to let the function infer the currently missing flags concerning the intended use of the RF pulses
  (excitation, refocusing, etc). These are important for the k-space trajectory calculation.
+ remove_duplicates: bool, default=True
+ Remove duplicate events from the sequence after reading
 
  Raises
  ------
@@ -357,7 +359,12 @@ def read(self, path: str, detect_rf_use: bool = False) -> None:
  for block_counter,events in self.block_events.items():
  if events[1] == k:
  del self.block_cache[block_counter]
-
+
+ # When removing duplicates, remove and remap events in the sequence without
+ # creating a copy.
+ if remove_duplicates:
+ self.remove_duplicates(in_place=True)
+
 
 def __read_definitions(input_file) -> Dict[str, str]:
  """

pypulseq/Sequence/sequence.py

@@ -5,6 +5,13 @@
 from typing import Tuple, List
 from typing import Union
 from warnings import warn
+from copy import deepcopy
+
+try:
+ from typing import Self
+except ImportError:
+ from typing import TypeVar
+ Self = TypeVar('Self', bound='Sequence')
 
 import matplotlib as mpl
 import numpy as np
@@ -1145,7 +1152,7 @@ def plot(
  if plot_now:
  plt.show()
 
- def read(self, file_path: str, detect_rf_use: bool = False) -> None:
+ def read(self, file_path: str, detect_rf_use: bool = False, remove_duplicates: bool = True) -> None:
  """
  Read `.seq` file from `file_path`.
 
@@ -1154,11 +1161,13 @@ def read(self, file_path: str, detect_rf_use: bool = False) -> None:
  detect_rf_use
  file_path : str
  Path to `.seq` file to be read.
+ remove_duplicates : bool, default=True
+ Remove duplicate events from the sequence after reading.
  """
  if self.use_block_cache:
  self.block_cache.clear()
 
- read(self, path=file_path, detect_rf_use=detect_rf_use)
+ read(self, path=file_path, detect_rf_use=detect_rf_use, remove_duplicates=remove_duplicates)
 
  # Initialize next free block ID
  self.next_free_block_ID = (max(self.block_events) + 1) if self.block_events else 1
@@ -1177,6 +1186,76 @@ def register_label_event(self, event: SimpleNamespace) -> int:
  def register_rf_event(self, event: SimpleNamespace) -> Tuple[int, List[int]]:
  return block.register_rf_event(self, event)
 
+ def remove_duplicates(self, in_place: bool = False) -> Self:
+ """
+ Removes duplicate events from the shape and event libraries contained
+ in this sequence.
+
+ Parameters
+ ----------
+ in_place : bool, optional
+ If true, removes the duplicates from the current sequence.
+ Otherwise, a copy is created. The default is False.
+
+ Returns
+ -------
+ seq_copy : Sequence
+ If `in_place`, returns self. Otherwise returns a copy of the
+ sequence.
+ """
+ if in_place:
+ seq_copy = self
+ else:
+ # Avoid copying block_cache for performance
+ tmp = self.block_cache
+ self.block_cache = {}
+ seq_copy = deepcopy(self)
+ self.block_cache = tmp
+
+ # Find duplicate in shape library
+ seq_copy.shape_library, mapping = seq_copy.shape_library.remove_duplicates(9)
+
+ # Remap shape IDs of arbitrary gradient events
+ for x in seq_copy.grad_library.data:
+ if seq_copy.grad_library.type[x] == 'g':
+ data = seq_copy.grad_library.data[x]
+ new_data = (data[0],) + (mapping[data[1]], mapping[data[2]]) + data[3:]
+ if data != new_data:
+ seq_copy.grad_library.update(x, None, new_data)
+
+ # Remap shape IDs of RF events
+ for x in seq_copy.rf_library.data:
+ data = seq_copy.rf_library.data[x]
+ new_data = (data[0],) + (mapping[data[1]], mapping[data[2]], mapping[data[3]]) + data[4:]
+ if data != new_data:
+ seq_copy.rf_library.update(x, None, new_data)
+
+ # Filter duplicates in gradient library
+ seq_copy.grad_library, mapping = seq_copy.grad_library.remove_duplicates((6, 6, 6, 6, 6, 6))
+
+ # Remap gradient event IDs
+ for x in seq_copy.block_events:
+ seq_copy.block_events[x][2] = mapping[seq_copy.block_events[x][2]]
+ seq_copy.block_events[x][3] = mapping[seq_copy.block_events[x][3]]
+ seq_copy.block_events[x][4] = mapping[seq_copy.block_events[x][4]]
+
+ # Filter duplicates in RF library
+ seq_copy.rf_library, mapping = seq_copy.rf_library.remove_duplicates((12, 0, 0, 0, 6, 6, 6))
+
+ # Remap RF event IDs
+ for x in seq_copy.block_events:
+ seq_copy.block_events[x][1] = mapping[seq_copy.block_events[x][1]]
+
+ # Filter duplicates in ADC library
+ seq_copy.adc_library, mapping = seq_copy.adc_library.remove_duplicates((0, 9, 6, 6, 6, 6))
+
+ # Remap ADC event IDs
+ for x in seq_copy.block_events:
+ seq_copy.block_events[x][5] = mapping[seq_copy.block_events[x][5]]
+
+ return seq_copy
+
+
  def rf_from_lib_data(self, lib_data: list, use: str = str()) -> SimpleNamespace:
  """
  Construct RF object from `lib_data`.
@@ -1754,7 +1833,7 @@ def waveforms_export(self, time_range=(0, np.inf)) -> dict:
 
  return all_waveforms
 
- def write(self, name: str, create_signature: bool = True) -> None:
+ def write(self, name: str, create_signature: bool = True, remove_duplicates: bool = True) -> None:
  """
  Write the sequence data to the given filename using the open file format for MR sequences.
 
@@ -1766,5 +1845,7 @@ def write(self, name: str, create_signature: bool = True) -> None:
  Filename of `.seq` file to be written to disk.
  create_signature : bool, default=True
  Boolean flag to indicate if the file has to be signed.
+ remove_duplicates : bool, default=True
+ Remove duplicate events from the sequence before writing
  """
- write_seq(self, name, create_signature)
+ write_seq(self, name, create_signature, remove_duplicates)

pypulseq/Sequence/write_seq.py

@@ -7,7 +7,7 @@
 from pypulseq.supported_labels_rf_use import get_supported_labels
 
 
-def write(self, file_name: Union[str, Path], create_signature) -> None:
+def write(self, file_name: Union[str, Path], create_signature, remove_duplicates=True) -> None:
  """
  Write the sequence data to the given filename using the open file format for MR sequences.
 
@@ -18,6 +18,9 @@ def write(self, file_name: Union[str, Path], create_signature) -> None:
  file_name : str or Path
  File name of `.seq` file to be written to disk.
  create_signature : bool
+ remove_duplicates : bool
+ Before writing, remove and remap events that would be duplicates after
+ the rounding done during writing
 
  Raises
  ------
@@ -30,7 +33,12 @@ def write(self, file_name: Union[str, Path], create_signature) -> None:
  if file_name.suffix != '.seq':
  # Append .seq suffix
  file_name = file_name.with_suffix(file_name.suffix + '.seq')
-
+
+ # If removing duplicates, make a copy of the sequence with the duplicate
+ # events removed.
+ if remove_duplicates:
+ self = self.remove_duplicates()
+
  with open(file_name, "w") as output_file:
  output_file.write("# Pulseq sequence file\n")
  output_file.write("# Created by PyPulseq\n\n")

pypulseq/event_lib.py

@@ -1,6 +1,12 @@
 from types import SimpleNamespace
-from typing import Tuple
-
+from typing import Tuple, Union
+try:
+ from typing import Self
+except ImportError:
+ from typing import TypeVar
+ Self = TypeVar('Self', bound='EventLibrary')
+
+import math
 import numpy as np
 
 
@@ -219,10 +225,6 @@ def update(
  data_type : str, default=str()
  """
  if key_id in self.data:
- # TODO: When reading a sequence file we can end up with duplicate
- # events due to rounding to 6 digits. update() calls to both
- # duplicates would give a key error here on the second
- # duplicate. Ideally no duplicates should exist though...
  if self.data[key_id] in self.keymap:
  del self.keymap[self.data[key_id]]
 
@@ -244,3 +246,67 @@ def update_data(
  data_type : str
  """
  self.update(key_id, old_data, new_data, data_type)
+
+ def remove_duplicates(self, digits: Union[int, Tuple[int]]) -> Tuple[Self, dict]:
+ """
+ Remove duplicate events from this event library by rounding the data
+ according to the significant `digits` specification, and then removing
+ duplicate events.
+ Returns a new event library, leaving the current one intact.
+
+ Parameters
+ ----------
+ digits : Union[int, List[int]]
+ For libraries with `numpy_data == True`:
+ A single number specifying the number of significant digits
+ after rounding.
+ Otherwise:
+ A tuple of numbers specifying the number of significant digits
+ after rounding for each entry in the event data tuple.
+
+ Returns
+ -------
+ new_library : EventLibrary
+ Event library with the duplicate events removed
+ mapping : dict
+ Dictionary containing a mapping of IDs in the old library to IDs
+ in the new library.
+ """
+ def round_data(data: Tuple[float], digits: Tuple[int]) -> Tuple[float]:
+ """
+ Round the data tuple to a specified number of significant digits,
+ specified by `digits`. Rounding behaviour is similar to the {.Ng}
+ format specifier if N > 0, and similar to {.0f} otherwise.
+ """
+ return tuple(round(d, dig - int(math.ceil(math.log10(abs(d) + 1e-12))) if dig > 0 else 0) for d, dig in zip(data, digits))
+
+ def round_data_numpy(data: np.ndarray, digits: int) -> np.ndarray:
+ """
+ Round the data array to a specified number of significant digits,
+ specified by `digits`. Rounding behaviour is similar to the {.Ng}
+ format specifier if N > 0, and similar to {.0f} otherwise.
+ """
+ mags = 10 ** (digits - (np.ceil(np.log10(abs(data) + 1e-12))) if digits > 0 else 0)
+ result = np.round(data * mags) / mags
+ result.flags.writeable = False
+ return result
+
+ # Round library data bbased on `digits` specification
+ if self.numpy_data:
+ rounded_data = {x:round_data_numpy(self.data[x], digits) for x in self.data}
+ else:
+ rounded_data = {x:round_data(self.data[x], digits) for x in self.data}
+
+ # Initialize filtered library
+ new_library = EventLibrary(numpy_data=self.numpy_data)
+
+ # Initialize ID mapping. Always include 0:0 to allow the mapping dict
+ # to be used for mapping block_events (which can contain 0, i.e. no
+ # event)
+ mapping = {0:0}
+
+ # Recreate library using rounded values
+ for k,v in rounded_data.items():
+ mapping[k], _ = new_library.find_or_insert(v, self.type[k] if k in self.type else str())
+
+ return new_library, mapping