Fxns: Initial reach & drop trials - #705
Conversation
| continue | ||
|
|
||
| # 1) Find peaks in speed and get their heights | ||
| peak_idx, peak_properties = find_peaks(speed, height=speed_threshold) |
There was a problem hiding this comment.
i think you need to include distance=(0.4*samplerate) to be consistent with the paper. this will make sure the peaks are all +/- 200ms apart, eliminating the smaller peaks first.
you can also include prominence=prominence_fraction here so you don't have to separately compute it later.
| # 2) Compute prominences for those peaks | ||
| prominences, left_bases, right_bases = peak_prominences(speed, peak_idx) | ||
|
|
||
| # 3) Keep peaks with prominence >= (prominence_fraction * peak_height) | ||
| peak_heights = peak_properties["peak_heights"] | ||
| keep = ( prominences >= (prominence_fraction * peak_heights) ) | ||
| peak_idx = peak_idx[keep] | ||
| if peak_idx.size == 0: | ||
| split_indices.append(len(traj)) | ||
| continue |
There was a problem hiding this comment.
can be simplified by passing the prominence argument to find_peaks
| # 4) Find troughs (local minima) in speed | ||
| trough_idx = find_peaks(-speed)[0] # peaks of -speed are minima of speed | ||
| if trough_idx.size == 0: | ||
| split_indices.append(len(traj)) | ||
| continue | ||
|
|
||
| # 5) For each qualifying peak in chronological order, find the first trough after it. | ||
| peak_idx = np.sort(peak_idx) | ||
| no_initial_peak = True | ||
| for p in peak_idx: | ||
| # first trough strictly after the peak | ||
| after = trough_idx[trough_idx > p] | ||
| if after.size == 0: | ||
| continue | ||
| t = after[0] | ||
|
|
||
| # Check the "initial peak" criterion: the trough position is >= distance_threshold | ||
| if np.linalg.norm(trajectory[t]) >= distance_threshold: | ||
| split_indices.append(int(t)) | ||
| no_initial_peak = False | ||
| break |
There was a problem hiding this comment.
i'm not sure i understand this part. what are you trying to do here? maybe this can be accomplished with distance parameter in find_peaks?
| nans = np.any(np.isnan(trial), axis=1) | ||
|
|
||
| max_consecutive = 0 | ||
| current_consecutive = 0 | ||
|
|
||
| for nan in nans: | ||
|
|
||
| if nan: | ||
| current_consecutive += 1 | ||
| max_consecutive = max(max_consecutive, current_consecutive) | ||
| else: | ||
| current_consecutive = 0 | ||
|
|
||
| if (max_consecutive / samplerate) > max_gap: | ||
| drop.append(True) | ||
| else: | ||
| drop.append(False) |
There was a problem hiding this comment.
this looks good. it maybe slow if there are a lot of trials, i suggest using the existing function from utils if you add from .. import utils somewhere at the top, then:
| nans = np.any(np.isnan(trial), axis=1) | |
| max_consecutive = 0 | |
| current_consecutive = 0 | |
| for nan in nans: | |
| if nan: | |
| current_consecutive += 1 | |
| max_consecutive = max(max_consecutive, current_consecutive) | |
| else: | |
| current_consecutive = 0 | |
| if (max_consecutive / samplerate) > max_gap: | |
| drop.append(True) | |
| else: | |
| drop.append(False) | |
| max_consecutive = utils.max_repeated_nans(np.mean(trial, axis=1)) | |
| drop.append(max_consecutive / samplerate > max_gap) |
|
|
||
| Args: | ||
| trajectories (list of numpy.ndarray): List of N trial trajectories, where each element | ||
| is an array of shape (T, D) with T time points and D spatial dimensions. |
There was a problem hiding this comment.
we've been using nt and nch to describe time and spatial dimensions.
| Based on method by Rouse et al., 2022 | ||
| https://pmc.ncbi.nlm.nih.gov/articles/PMC9014981/ |
There was a problem hiding this comment.
i think briefly describing the approach would be helpful as well as citing the paper you based it on
|
left some comments / suggestions! let me know if you have questions |
2 participants
Add functions to 1) identify trials that have gaps in data over a specified time threshold; and 2) identify the index where an initial reach ends using peak-finding on movement speed. Add human center-out test data for these functions.