raab/GP2023_chirp_detection
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'master' of https://whale.am28.uni-tuebingen.de/git/raab/GP2023_chirp_detection

Browse Source
This commit is contained in:
wendtalexander 2023-01-12 14:39:14 +01:00
commit 1077558868
1 changed files with 190 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

344
code/chirpdetection.py
View File

@ -29,8 +29,8 @@ class LoadData:
def __init__(self, datapath: str) -> None: def __init__(self, datapath: str) -> None:
# load raw data # load raw data
file = os.path.join(datapath, "traces-grid1.raw") self.file = os.path.join(datapath, "traces-grid1.raw")
self.data = DataLoader(file, 60.0, 0, channel=-1) self.data = DataLoader(self.file, 60.0, 0, channel=-1)
self.samplerate = self.data.samplerate self.samplerate = self.data.samplerate
# load wavetracker files # load wavetracker files
@ -40,6 +40,12 @@ class LoadData:
self.ident = np.load(datapath + "ident_v.npy", allow_pickle=True) self.ident = np.load(datapath + "ident_v.npy", allow_pickle=True)
self.ids = np.unique(self.ident[~np.isnan(self.ident)]) self.ids = np.unique(self.ident[~np.isnan(self.ident)])
def __repr__(self) -> str:
return f"LoadData({self.file})"
def __str__(self) -> str:
return f"LoadData({self.file})"
def instantaneos_frequency( def instantaneos_frequency(
signal: np.ndarray, samplerate: int signal: np.ndarray, samplerate: int
@ -62,7 +68,8 @@ def instantaneos_frequency(
# calculate instantaneos frequency with zero crossings # calculate instantaneos frequency with zero crossings
roll_signal = np.roll(signal, shift=1) roll_signal = np.roll(signal, shift=1)
time_signal = np.arange(len(signal)) / samplerate time_signal = np.arange(len(signal)) / samplerate
period_index = np.arange(len(signal))[(roll_signal < 0) & (signal >= 0)] period_index = np.arange(len(signal))[(
roll_signal < 0) & (signal >= 0)][1:-1]
upper_bound = np.abs(signal[period_index]) upper_bound = np.abs(signal[period_index])
lower_bound = np.abs(signal[period_index - 1]) lower_bound = np.abs(signal[period_index - 1])
@ -77,7 +84,7 @@ def instantaneos_frequency(
true_zero = lower_time + lower_ratio * time_delta true_zero = lower_time + lower_ratio * time_delta
# create new time array # create new time array
inst_freq_time = true_zero[1:] + 0.5 * np.diff(true_zero) inst_freq_time = true_zero[:-1] + 0.5 * np.diff(true_zero)
# compute frequency # compute frequency
inst_freq = gaussian_filter1d(1 / np.diff(true_zero), 5) inst_freq = gaussian_filter1d(1 / np.diff(true_zero), 5)
@ -167,175 +174,204 @@ def main(datapath: str) -> None:
# load wavetracker files # load wavetracker files
time = np.load(datapath + "times.npy", allow_pickle=True) time = np.load(datapath + "times.npy", allow_pickle=True)
freq = np.load(datapath + "fund_v.npy", allow_pickle=True) freq = np.load(datapath + "fund_v.npy", allow_pickle=True)
powers = np.load(datapath + "sign_v.npy", allow_pickle=True)
idx = np.load(datapath + "idx_v.npy", allow_pickle=True) idx = np.load(datapath + "idx_v.npy", allow_pickle=True)
ident = np.load(datapath + "ident_v.npy", allow_pickle=True) ident = np.load(datapath + "ident_v.npy", allow_pickle=True)
# set time window # <------------------------ Iterate through windows here # set time window # <------------------------ Iterate through windows here
window_duration = 60 * 5 * data.samplerate # 5 minutes window window_duration = 5 * data.samplerate # 5 seconds window
window_overlap = 30 * data.samplerate # 30 seconds overlap window_overlap = 0.5 * data.samplerate # 30 seconds overlap
raw_time = np.arange(data.shape[0]) raw_time = np.arange(data.shape[0])
window_starts = np.arange( t0 = (3 * 60 * 60 + 6 * 60 + 43.5) * data.samplerate
raw_time[0], raw_time[-1], window_duration - window_overlap / 2) dt = 60 * data.samplerate
window_starts = np.arange(
t0, t0 + dt, window_duration - window_overlap, dtype=int)
t0 = 3 * 60 * 60 + 6 * 60 + 43.5 for start_index in window_starts:
dt = 60
start_index = t0 * data.samplerate
stop_index = (t0 + dt) * data.samplerate
# load region of interest of raw data file
data_oi = data[start_index:stop_index, :]
# iterate through all fish # make t0 and dt
for track_id in np.unique(ident[~np.isnan(ident)])[:2]: t0 = start_index / data.samplerate
dt = window_duration / data.samplerate
# <------------------------------------------ Find best electrodes here # set index window
# <------------------------------------------ Iterate through electrodes stop_index = start_index + window_duration
# get indices for time array in time window
window_index = np.arange(len(idx))[
(ident == track_id) & (time[idx] >= t0) & (time[idx] <= (t0 + dt))
]
# filter baseline and above # t0 = 3 * 60 * 60 + 6 * 60 + 43.5
freq_temp = freq[window_index] # dt = 60
time_temp = time[idx[window_index]] # start_index = t0 * data.samplerate
# stop_index = (t0 + dt) * data.samplerate
electrode = 10 # load region of interest of raw data file
data_oi = data[start_index:stop_index, :]
# initialize plot
fig, axs = plt.subplots( fig, axs = plt.subplots(
7, 7,
1, 2,
figsize=(20 / 2.54, 12 / 2.54), figsize=(20 / 2.54, 12 / 2.54),
constrained_layout=True, constrained_layout=True,
sharex=True, sharex=True,
sharey='row',
) )
# plot spectrogram # iterate through all fish
plot_spectrogram(axs[0], data_oi[:, electrode], data.samplerate) for i, track_id in enumerate(np.unique(ident[~np.isnan(ident)])[:2]):
# plot wavetracker tracks to spectrogram # <------------------------------------------ Find best electrodes here
# for track_id in np.unique(ident): # <---------- Find freq gaps later # <------------------------------------------ Iterate through electrodes
# here # get indices for time array in time window
window_index = np.arange(len(idx))[
# # get indices for time array in time window (ident == track_id) & (time[idx] >= t0) & (
# window_index = np.arange(len(idx))[ time[idx] <= (t0 + dt))
# (ident == track_id) & ]
# (time[idx] >= t0) &
# (time[idx] <= (t0 + dt)) # get tracked frequencies and their times
# ] freq_temp = freq[window_index]
powers_temp = powers[window_index, :]
# freq_temp = freq[window_index] # time_temp = time[idx[window_index]]
# time_temp = time[idx[window_index]] track_samplerate = np.mean(1 / np.diff(time))
expected_duration = ((t0 + dt) - t0) * track_samplerate
# axs[0].plot(time_temp-t0, freq_temp, lw=2)
# axs[0].set_ylim(500, 1000) # check if tracked data available in this window
if len(freq_temp) < expected_duration * 0.9:
# track_id = ids continue
# frequency where second filter filters # get best electrode
search_freq = -50 electrode = np.argsort(np.nanmean(powers_temp, axis=0))[-1]
# electrode = best_electrodes[0]
# filter baseline and above
baseline, search = double_bandpass( # plot spectrogram
data_oi[:, electrode], data.samplerate, freq_temp, search_freq plot_spectrogram(axs[0, i], data_oi[:, electrode], data.samplerate)
)
# plot wavetracker tracks to spectrogram
# plot waveform of filtered signal # for track_id in np.unique(ident): # <---------- Find freq gaps later
axs[2].plot(np.arange(len(baseline)) / data.samplerate, baseline) # here
# plot instatneous frequency # # get indices for time array in time window
# broad_baseline = bandpass_filter(data_oi[:, electrode], data.samplerate, lowf=np.mean( # window_index = np.arange(len(idx))[
# freq_temp)-5, highf=np.mean(freq_temp)+200) # (ident == track_id) &
# (time[idx] >= t0) &
baseline_freq_time, baseline_freq = instantaneos_frequency( # (time[idx] <= (t0 + dt))
baseline, data.samplerate # ]
)
axs[1].plot(baseline_freq_time, baseline_freq) # freq_temp = freq[window_index]
# time_temp = time[idx[window_index]]
# plot waveform of filtered search signal
axs[3].plot(np.arange(len(baseline)) / data.samplerate, search) # axs[0].plot(time_temp-t0, freq_temp, lw=2)
# axs[0].set_ylim(500, 1000)
# compute envelopes
cutoff = 25 # track_id = ids
baseline_envelope = envelope(baseline, data.samplerate, cutoff)
axs[2].plot( # frequency where second filter filters
np.arange(len(baseline)) / data.samplerate, search_freq = -50
baseline_envelope,
c="orange", # filter baseline and above
) baseline, search = double_bandpass(
search_envelope = envelope(search, data.samplerate, cutoff) data_oi[:, electrode], data.samplerate, freq_temp, search_freq
axs[3].plot( )
np.arange(len(baseline)) / data.samplerate,
search_envelope, # plot waveform of filtered signal
c="orange", axs[2, i].plot(np.arange(len(baseline)) /
) data.samplerate, baseline, c="k")
# highpass filter envelopes # plot instatneous frequency
cutoff = 5 broad_baseline = bandpass_filter(data_oi[:, electrode], data.samplerate, lowf=np.mean(
baseline_envelope = highpass_filter( freq_temp)-5, highf=np.mean(freq_temp)+100)
baseline_envelope, data.samplerate, cutoff=cutoff
) baseline_freq_time, baseline_freq = instantaneos_frequency(
# search_envelope = highpass_filter( baseline, data.samplerate
# search_envelope, data.samplerate, cutoff=cutoff) )
axs[1, i].plot(baseline_freq_time, baseline_freq -
# envelopes of filtered envelope of filtered baseline np.median(baseline_freq), marker=".")
baseline_envelope = envelope(
np.abs(baseline_envelope), data.samplerate, cutoff # plot waveform of filtered search signal
) axs[3, i].plot(np.arange(len(baseline)) / data.samplerate, search)
# search_envelope = bandpass_filter( # compute envelopes
# search_envelope, data.samplerate, lowf=lowf, highf=highf) cutoff = 25
baseline_envelope = envelope(baseline, data.samplerate, cutoff)
# bandpass filter the instantaneous axs[2, i].plot(
inst_freq_filtered = bandpass_filter( np.arange(len(baseline)) / data.samplerate,
baseline_freq, data.samplerate, lowf=15, highf=8000 baseline_envelope,
) c="orange",
)
# plot filtered and rectified envelope axs[2, i].plot(
axs[4].plot( np.arange(len(baseline)) / data.samplerate,
np.arange(len(baseline)) / data.samplerate, baseline_envelope broad_baseline,
) c="green",
)
axs[5].plot(np.arange(len(baseline)) / search_envelope = envelope(search, data.samplerate, cutoff)
data.samplerate, search_envelope) axs[3, i].plot(
np.arange(len(baseline)) / data.samplerate,
axs[6].plot(baseline_freq_time, np.abs(inst_freq_filtered)) search_envelope,
c="orange",
# detect peaks baseline_enelope )
embed()
prominence = iqr(baseline_envelope) # highpass filter envelopes
baseline_peaks, _ = find_peaks( cutoff = 5
baseline_envelope, prominence=prominence) baseline_envelope = highpass_filter(
axs[4].scatter( baseline_envelope, data.samplerate, cutoff=cutoff
(np.arange(len(baseline)) / data.samplerate)[baseline_peaks], )
baseline_envelope[baseline_peaks], # search_envelope = highpass_filter(
c="red", # search_envelope, data.samplerate, cutoff=cutoff)
)
# envelopes of filtered envelope of filtered baseline
# detect peaks search_envelope baseline_envelope = envelope(
search_peaks, _ = find_peaks(search_envelope, height=0.0001) np.abs(baseline_envelope), data.samplerate, cutoff
axs[5].scatter( )
(np.arange(len(baseline)) / data.samplerate)[search_peaks],
search_envelope[search_peaks], # search_envelope = bandpass_filter(
c="red", # search_envelope, data.samplerate, lowf=lowf, highf=highf)
)
# bandpass filter the instantaneous
# detect peaks inst_freq_filtered inst_freq_filtered = bandpass_filter(
inst_freq_peaks, _ = find_peaks(np.abs(inst_freq_filtered), height=2) baseline_freq, data.samplerate, lowf=15, highf=8000
axs[6].scatter( )
baseline_freq_time[inst_freq_peaks],
np.abs(inst_freq_filtered)[inst_freq_peaks], # plot filtered and rectified envelope
c="red", axs[4, i].plot(
) np.arange(len(baseline)) / data.samplerate, baseline_envelope
)
axs[0].set_title("Spectrogram")
axs[1].set_title("Fitered baseline instanenous frequency") axs[5, i].plot(np.arange(len(baseline)) /
axs[2].set_title("Fitered baseline") data.samplerate, search_envelope)
axs[3].set_title("Fitered above")
axs[4].set_title("Filtered envelope of baseline envelope") axs[6, i].plot(baseline_freq_time, np.abs(inst_freq_filtered))
axs[5].set_title("Search envelope")
axs[6].set_title("Filtered absolute instantaneous frequency") # detect peaks baseline_enelope
prominence = iqr(baseline_envelope)
baseline_peaks, _ = find_peaks(
baseline_envelope, prominence=prominence)
axs[4, i].scatter(
(np.arange(len(baseline)) / data.samplerate)[baseline_peaks],
baseline_envelope[baseline_peaks],
c="red",
)
# detect peaks search_envelope
search_peaks, _ = find_peaks(search_envelope, height=0.0001)
axs[5, i].scatter(
(np.arange(len(baseline)) / data.samplerate)[search_peaks],
search_envelope[search_peaks],
c="red",
)
# detect peaks inst_freq_filtered
inst_freq_peaks, _ = find_peaks(
np.abs(inst_freq_filtered), height=2)
axs[6, i].scatter(
baseline_freq_time[inst_freq_peaks],
np.abs(inst_freq_filtered)[inst_freq_peaks],
c="red",
)
axs[0, i].set_title("Spectrogram")
axs[1, i].set_title("Fitered baseline instanenous frequency")
axs[2, i].set_title("Fitered baseline")
axs[3, i].set_title("Fitered above")
axs[4, i].set_title("Filtered envelope of baseline envelope")
axs[5, i].set_title("Search envelope")
axs[6, i].set_title("Filtered absolute instantaneous frequency")
plt.show() plt.show()