Skip to content

Commit

Permalink
adjust align_starting_times to set_aligned_starting_times
Browse files Browse the repository at this point in the history
  • Loading branch information
CodyCBakerPhD committed Jun 7, 2023
1 parent bc258f2 commit d983d1c
Show file tree
Hide file tree
Showing 10 changed files with 126 additions and 102 deletions.
6 changes: 3 additions & 3 deletions src/neuroconv/basetemporalalignmentinterface.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,18 +55,18 @@ def set_aligned_timestamps(self, aligned_timestamps: np.ndarray):
"The protocol for synchronizing the timestamps of this interface has not been specified!"
)

def align_starting_time(self, starting_time: float):
def set_aligned_starting_time(self, aligned_starting_time: float):
"""
Align the starting time for this interface relative to the common session start time.
Must be in units seconds relative to the common 'session_start_time'.
Parameters
----------
starting_time : float
aligned_starting_time : float
The starting time for all temporal data in this interface.
"""
self.set_aligned_timestamps(aligned_timestamps=self.get_timestamps() + starting_time)
self.set_aligned_timestamps(aligned_timestamps=self.get_timestamps() + aligned_starting_time)

def align_by_interpolation(self, unaligned_timestamps: np.ndarray, aligned_timestamps: np.ndarray):
"""
Expand Down
32 changes: 16 additions & 16 deletions src/neuroconv/datainterfaces/behavior/audio/audiointerface.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,51 +98,51 @@ def get_timestamps(self) -> Optional[np.ndarray]:
def set_aligned_timestamps(self, aligned_timestamps: List[np.ndarray]):
raise NotImplementedError("The AudioInterface does not yet support timestamps.")

def align_starting_time(self, starting_time: float):
def set_aligned_starting_time(self, aligned_starting_time: float):
"""
Align all starting times for all audio files in this interface relative to the common session start time.
Must be in units seconds relative to the common 'session_start_time'.
Parameters
----------
starting_time : float
aligned_starting_time : float
The common starting time for all temporal data in this interface.
Applies to all segments if there are multiple file paths used by the interface.
"""
if self._segment_starting_times is None and self._number_of_audio_files == 1:
self._segment_starting_times = [starting_time]
self._segment_starting_times = [aligned_starting_time]
elif self._segment_starting_times is not None and self._number_of_audio_files > 1:
self._segment_starting_times = [
segment_starting_time + starting_time for segment_starting_time in self._segment_starting_times
segment_starting_time + aligned_starting_time for segment_starting_time in self._segment_starting_times
]
else:
raise ValueError(
"There are no segment starting times to shift by a common value! "
"Please set them using 'align_segment_starting_times'."
"Please set them using 'set_aligned_segment_starting_times'."
)

def align_segment_starting_times(self, segment_starting_times: List[float]):
def set_aligned_segment_starting_times(self, aligned_segment_starting_times: List[float]):
"""
Align the individual starting time for each audio file in this interface relative to the common session start time.
Must be in units seconds relative to the common 'session_start_time'.
Parameters
----------
segment_starting_times : list of floats
aligned_segment_starting_times : list of floats
The relative starting times of each audio file (segment).
"""
segment_starting_times_length = len(segment_starting_times)
assert isinstance(segment_starting_times, list) and all(
[isinstance(x, float) for x in segment_starting_times]
), "Argument 'segment_starting_times' must be a list of floats."
assert segment_starting_times_length == self._number_of_audio_files, (
f"The number of entries in 'segment_starting_times' ({segment_starting_times_length}) must be equal to the number of "
f"audio file paths ({self._number_of_audio_files})."
aligned_segment_starting_times_length = len(aligned_segment_starting_times)
assert isinstance(aligned_segment_starting_times, list) and all(
[isinstance(x, float) for x in aligned_segment_starting_times]
), "Argument 'aligned_segment_starting_times' must be a list of floats."
assert aligned_segment_starting_times_length == self._number_of_audio_files, (
f"The number of entries in 'aligned_segment_starting_times' ({aligned_segment_starting_times_length}) "
f"must be equal to the number of audio file paths ({self._number_of_audio_files})."
)

self._segment_starting_times = segment_starting_times
self._segment_starting_times = aligned_segment_starting_times

def align_by_interpolation(self, unaligned_timestamps: np.ndarray, aligned_timestamps: np.ndarray):
raise NotImplementedError("The AudioInterface does not yet support timestamps.")
Expand Down Expand Up @@ -200,7 +200,7 @@ def run_conversion(
if self._number_of_audio_files > 1 and self._segment_starting_times is None:
raise ValueError(
"If you have multiple audio files, then you must specify each starting time by calling "
"'.align_segment_starting_times(segment_starting_times=...)'!"
"'.set_aligned_segment_starting_times(...)'!"
)
starting_times = self._segment_starting_times or [0.0]

Expand Down
20 changes: 10 additions & 10 deletions src/neuroconv/datainterfaces/behavior/video/videodatainterface.py
Original file line number Diff line number Diff line change
Expand Up @@ -157,15 +157,15 @@ def set_aligned_timestamps(self, aligned_timestamps: List[np.ndarray]):
), "If setting both timestamps and starting times, please set the timestamps first so they can be shifted by the starting times."
self._timestamps = aligned_timestamps

def align_starting_time(self, starting_time: float, stub_test: bool = False):
def set_aligned_starting_time(self, aligned_starting_time: float, stub_test: bool = False):
"""
Align all starting times for all videos in this interface relative to the common session start time.
Must be in units seconds relative to the common 'session_start_time'.
Parameters
----------
starting_time : float
aligned_starting_time : float
The common starting time for all segments of temporal data in this interface.
stub_test : bool, default: False
If timestamps have not been set to this interface, it will attempt to retrieve them
Expand All @@ -177,17 +177,17 @@ def align_starting_time(self, starting_time: float, stub_test: bool = False):
if self._timestamps is not None:
self.set_aligned_timestamps(
aligned_timestamps=[
timestamps + starting_time for timestamps in self.get_timestamps(stub_test=stub_test)
timestamps + aligned_starting_time for timestamps in self.get_timestamps(stub_test=stub_test)
]
)
elif self._segment_starting_times is not None:
self._segment_starting_times = [
starting_time + starting_time for starting_time in self._segment_starting_times
segment_starting_time + aligned_starting_time for segment_starting_time in self._segment_starting_times
]
else:
raise ValueError("There are no timestamps or starting times set to shift by a common value!")

def align_segment_starting_times(self, segment_starting_times: List[float], stub_test: bool = False):
def set_aligned_segment_starting_times(self, aligned_segment_starting_times: List[float], stub_test: bool = False):
"""
Align the individual starting time for each video (segment) in this interface relative to the common session start time.
Expand All @@ -204,22 +204,22 @@ def align_segment_starting_times(self, segment_starting_times: List[float], stub
To limit that scan to a small number of frames, set `stub_test=True`.
"""
segment_starting_times_length = len(segment_starting_times)
assert segment_starting_times_length == self._number_of_files, (
f"The length of the 'segment_starting_times' list ({segment_starting_times_length}) does not match the "
aligned_segment_starting_times_length = len(aligned_segment_starting_times)
assert aligned_segment_starting_times_length == self._number_of_files, (
f"The length of the 'aligned_segment_starting_times' list ({aligned_segment_starting_times_length}) does not match the "
"number of video files ({self._number_of_files})!"
)
if self._timestamps is not None:
self.set_aligned_timestamps(
aligned_timestamps=[
timestamps + segment_starting_time
for timestamps, segment_starting_time in zip(
self.get_timestamps(stub_test=stub_test), segment_starting_times
self.get_timestamps(stub_test=stub_test), aligned_segment_starting_times
)
]
)
else:
self._segment_starting_times = segment_starting_times
self._segment_starting_times = aligned_segment_starting_times

def align_by_interpolation(self, unaligned_timestamps: np.ndarray, aligned_timestamps: np.ndarray):
raise NotImplementedError("The `align_by_interpolation` method has not been developed for this interface yet.")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -193,37 +193,40 @@ def set_aligned_segment_timestamps(self, aligned_segment_timestamps: List[np.nda
times=aligned_segment_timestamps[segment_index], segment_index=segment_index
)

def align_starting_time(self, starting_time: float):
def set_aligned_starting_time(self, aligned_starting_time: float):
if self._number_of_segments == 1:
self.set_aligned_timestamps(aligned_timestamps=self.get_timestamps() + starting_time)
self.set_aligned_timestamps(aligned_timestamps=self.get_timestamps() + aligned_starting_time)
else:
self.set_aligned_segment_timestamps(
aligned_segment_timestamps=[
segment_timestamps + starting_time for segment_timestamps in self.get_timestamps()
segment_timestamps + aligned_starting_time for segment_timestamps in self.get_timestamps()
]
)

def align_segment_starting_times(self, segment_starting_times: List[float]):
def set_aligned_segment_starting_times(self, aligned_segment_starting_times: List[float]):
"""
Align the starting time for each segment in this interface relative to the common session start time.
Must be in units seconds relative to the common 'session_start_time'.
Parameters
----------
segment_starting_times : list of floats
aligned_segment_starting_times : list of floats
The starting time for each segment of data in this interface.
"""
assert (
len(segment_starting_times) == self._number_of_segments
), f"The length of the starting_times ({len(segment_starting_times)}) does not match the number of segments ({self._number_of_segments})!"
assert len(aligned_segment_starting_times) == self._number_of_segments, (
f"The length of the starting_times ({len(aligned_segment_starting_times)}) does not match the "
"number of segments ({self._number_of_segments})!"
)

if self._number_of_segments == 1:
self.align_starting_time(starting_time=segment_starting_times[0])
self.set_aligned_starting_time(aligned_starting_time=aligned_segment_starting_times[0])
else:
aligned_segment_timestamps = [
segment_timestamps + segment_starting_time
for segment_timestamps, segment_starting_time in zip(self.get_timestamps(), segment_starting_times)
segment_timestamps + aligned_segment_starting_time
for segment_timestamps, aligned_segment_starting_time in zip(
self.get_timestamps(), aligned_segment_starting_times
)
]
self.set_aligned_segment_timestamps(aligned_segment_timestamps=aligned_segment_timestamps)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -166,47 +166,52 @@ def set_aligned_segment_timestamps(self, aligned_segment_timestamps: List[np.nda
times=aligned_segment_timestamps[segment_index], segment_index=segment_index
)

def align_starting_time(self, starting_time: float):
def set_aligned_starting_time(self, aligned_starting_time: float):
if self.sorting_extractor.has_recording():
if self._number_of_segments == 1:
self.set_aligned_timestamps(aligned_timestamps=self.get_timestamps() + starting_time)
self.set_aligned_timestamps(aligned_timestamps=self.get_timestamps() + aligned_starting_time)
else:
self.set_aligned_segment_timestamps(
aligned_timestamps=[
segment_timestamps + starting_time for segment_timestamps in self.get_timestamps()
segment_timestamps + aligned_starting_time for segment_timestamps in self.get_timestamps()
]
)
else:
for sorting_segment in self.sorting_extractor._sorting_segments:
sorting_segment._t_start += starting_time
sorting_segment._t_start += aligned_starting_time

def align_segment_starting_times(self, segment_starting_times: List[float]):
def set_aligned_segment_starting_times(self, aligned_segment_starting_times: List[float]):
"""
Align the starting time for each segment in this interface relative to the common session start time.
Must be in units seconds relative to the common 'session_start_time'.
Parameters
----------
segment_starting_times : list of floats
aligned_segment_starting_times : list of floats
The starting time for each segment of data in this interface.
"""
assert (
len(segment_starting_times) == self._number_of_segments
), f"The length of the starting_times ({len(segment_starting_times)}) does not match the number of segments ({self._number_of_segments})!"
assert len(aligned_segment_starting_times) == self._number_of_segments, (
f"The length of the starting_times ({len(aligned_segment_starting_times)}) does not match the number "
f"of segments ({self._number_of_segments})!"
)

if self.sorting_extractor.has_recording():
if self._number_of_segments == 1:
self.set_aligned_timestamps(aligned_timestamps=self.get_timestamps() + segment_starting_times[0])
self.set_aligned_starting_time(aligned_starting_time=aligned_segment_starting_times[0])
else:
aligned_segment_timestamps = [
segment_timestamps + segment_starting_time
for segment_timestamps, segment_starting_time in zip(self.get_timestamps(), segment_starting_times)
segment_timestamps + aligned_segment_starting_time
for segment_timestamps, aligned_segment_starting_time in zip(
self.get_timestamps(), aligned_segment_starting_times
)
]
self.set_aligned_segment_timestamps(aligned_segment_timestamps=aligned_segment_timestamps)
else:
for sorting_segment, starting_time in zip(self.sorting_extractor._sorting_segments, segment_starting_times):
sorting_segment._t_start = starting_time
for sorting_segment, aligned_segment_starting_time in zip(
self.sorting_extractor._sorting_segments, aligned_segment_starting_times
):
sorting_segment._t_start = aligned_segment_starting_time

def subset_sorting(self):
max_min_spike_time = max(
Expand Down
20 changes: 12 additions & 8 deletions src/neuroconv/datainterfaces/icephys/abf/abfdatainterface.py
Original file line number Diff line number Diff line change
Expand Up @@ -144,35 +144,39 @@ def get_metadata(self) -> dict:

return metadata

def align_starting_time(self, starting_time: float):
def align_starting_time(self, aligned_starting_time: float):
for reader in self.readers_list:
number_of_segments = reader.header["nb_segment"][0]
for segment_index in range(number_of_segments):
reader._t_starts[segment_index] += starting_time
reader._t_starts[segment_index] += aligned_starting_time

def align_segment_starting_times(self, segment_starting_times: List[List[float]], stub_test: bool = False):
def set_aligned_segment_starting_times(
self, aligned_segment_starting_times: List[List[float]], stub_test: bool = False
):
"""
Align the individual starting time for each video in this interface relative to the common session start time.
Must be in units seconds relative to the common 'session_start_time'.
Parameters
----------
segment_starting_times : list of list of floats
aligned_segment_starting_times : list of list of floats
The relative starting times of each video.
Outer list is over file paths (readers).
Inner list is over segments of each recording.
"""
number_of_files_from_starting_times = len(segment_starting_times)
number_of_files_from_starting_times = len(aligned_segment_starting_times)
assert number_of_files_from_starting_times == len(self.readers_list), (
f"The length of the outer list of 'starting_times' ({number_of_files_from_starting_times}) "
"does not match the number of files ({len(self.readers_list)})!"
)

for file_index, (reader, segment_starting_times) in enumerate(zip(self.readers_list, segment_starting_times)):
for file_index, (reader, aligned_segment_starting_times_by_file) in enumerate(
zip(self.readers_list, aligned_segment_starting_times)
):
number_of_segments = reader.header["nb_segment"][0]
assert number_of_segments == len(
segment_starting_times
aligned_segment_starting_times_by_file
), f"The length of starting times index {file_index} does not match the number of segments of that reader!"

reader._t_starts = segment_starting_times
reader._t_starts = aligned_segment_starting_times_by_file
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ def get_timestamps(self) -> np.ndarray:
def set_aligned_timestamps(self, aligned_timestamps: np.ndarray):
raise NotImplementedError("Icephys interfaces do not yet support timestamps.")

def align_starting_time(self, starting_time: float):
def set_aligned_starting_time(self, aligned_starting_time: float):
raise NotImplementedError("This icephys interface has not specified the method for aligning starting time.")

def align_by_interpolation(self, unaligned_timestamps: np.ndarray, aligned_timestamps: np.ndarray):
Expand Down
4 changes: 2 additions & 2 deletions src/neuroconv/datainterfaces/text/timeintervalsinterface.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,11 +63,11 @@ def get_timestamps(self, column: str) -> np.ndarray:

return self.dataframe[column].values

def align_starting_time(self, starting_time: float):
def set_aligned_starting_time(self, aligned_starting_time: float):
timing_columns = [column for column in self.dataframe.columns if column.endswith("_time")]

for column in timing_columns:
self.dataframe[column] += starting_time
self.dataframe[column] += aligned_starting_time

def set_aligned_timestamps(
self, aligned_timestamps: np.ndarray, column: str, interpolate_other_columns: bool = False
Expand Down
Loading

0 comments on commit d983d1c

Please sign in to comment.