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

merging diverged

Browse Source
This commit is contained in:
weygoldt
2023-01-25 10:19:19 +01:00
parent 0b109e8c5e
commit a6b7ed2c6c
6 changed files with 270 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
code/modules/behaviour_handling.py
View File

@@ -1,8 +1,7 @@
import numpy as np
import os
import os
import numpy as np
from IPython import embed
@@ -19,46 +18,60 @@ class Behavior:
Attributes
----------
behavior: 0: chasing onset, 1: chasing offset, 2: physical contact
behavior_type:
behavioral_category:
comment_start:
comment_stop:
dataframe: pandas dataframe with all the data
duration_s:
media_file:
observation_date:
observation_id:
start_s: start time of the event in seconds
stop_s: stop time of the event in seconds
total_length:
behavior_type:
behavioral_category:
comment_start:
comment_stop:
dataframe: pandas dataframe with all the data
duration_s:
media_file:
observation_date:
observation_id:
start_s: start time of the event in seconds
stop_s: stop time of the event in seconds
total_length:
"""
def __init__(self, folder_path: str) -> None:
LED_on_time_BORIS = np.load(os.path.join(folder_path, 'LED_on_time.npy'), allow_pickle=True)
LED_on_time_BORIS = np.load(os.path.join(
folder_path, 'LED_on_time.npy'), allow_pickle=True)
csv_filename = [f for f in os.listdir(folder_path) if f.endswith('.csv')][0]
csv_filename = os.path.split(folder_path[:-1])[-1]
csv_filename = '-'.join(csv_filename.split('-')[:-1]) + '.csv'
# embed()
# csv_filename = [f for f in os.listdir(
# folder_path) if f.endswith('.csv')][0]
logger.info(f'CSV file: {csv_filename}')
self.dataframe = read_csv(os.path.join(folder_path, csv_filename))
self.chirps = np.load(os.path.join(folder_path, 'chirps.npy'), allow_pickle=True)
self.chirps_ids = np.load(os.path.join(folder_path, 'chirp_ids.npy'), allow_pickle=True)
self.chirps = np.load(os.path.join(
folder_path, 'chirps.npy'), allow_pickle=True)
self.chirps_ids = np.load(os.path.join(
folder_path, 'chirp_ids.npy'), allow_pickle=True)
self.ident = np.load(os.path.join(folder_path, 'ident_v.npy'), allow_pickle=True)
self.idx = np.load(os.path.join(folder_path, 'idx_v.npy'), allow_pickle=True)
self.freq = np.load(os.path.join(folder_path, 'fund_v.npy'), allow_pickle=True)
self.time = np.load(os.path.join(folder_path, "times.npy"), allow_pickle=True)
self.spec = np.load(os.path.join(folder_path, "spec.npy"), allow_pickle=True)
self.ident = np.load(os.path.join(
folder_path, 'ident_v.npy'), allow_pickle=True)
self.idx = np.load(os.path.join(
folder_path, 'idx_v.npy'), allow_pickle=True)
self.freq = np.load(os.path.join(
folder_path, 'fund_v.npy'), allow_pickle=True)
self.time = np.load(os.path.join(
folder_path, "times.npy"), allow_pickle=True)
self.spec = np.load(os.path.join(
folder_path, "spec.npy"), allow_pickle=True)
for k, key in enumerate(self.dataframe.keys()):
key = key.lower()
key = key.lower()
if ' ' in key:
key = key.replace(' ', '_')
if '(' in key:
key = key.replace('(', '')
key = key.replace(')', '')
setattr(self, key, np.array(self.dataframe[self.dataframe.keys()[k]]))
setattr(self, key, np.array(
self.dataframe[self.dataframe.keys()[k]]))
last_LED_t_BORIS = LED_on_time_BORIS[-1]
real_time_range = self.time[-1] - self.time[0]
factor = 1.034141
@@ -68,16 +81,17 @@ class Behavior:
def correct_chasing_events(
category: np.ndarray,
category: np.ndarray,
timestamps: np.ndarray
) -> tuple[np.ndarray, np.ndarray]:
) -> tuple[np.ndarray, np.ndarray]:
onset_ids = np.arange(
len(category))[category == 0]
offset_ids = np.arange(
len(category))[category == 1]
wrong_bh = np.arange(len(category))[category!=2][:-1][np.diff(category[category!=2])==0]
wrong_bh = np.arange(len(category))[
category != 2][:-1][np.diff(category[category != 2]) == 0]
if onset_ids[0] > offset_ids[0]:
offset_ids = np.delete(offset_ids, 0)
help_index = offset_ids[0]
@@ -95,21 +109,22 @@ def correct_chasing_events(
logger.info(f'Offsets are greater than onsets by {len_diff}')
elif len(onset_ids) == len(offset_ids):
logger.info('Chasing events are equal')
return category, timestamps
def event_triggered_chirps(
event: np.ndarray,
chirps:np.ndarray,
event: np.ndarray,
chirps: np.ndarray,
time_before_event: int,
time_after_event: int,
dt: float,
width: float,
)-> tuple[np.ndarray, np.ndarray, np.ndarray]:
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
event_chirps = [] # chirps that are in specified window around event
centered_chirps = [] # timestamps of chirps around event centered on the event timepoint
# timestamps of chirps around event centered on the event timepoint
centered_chirps = []
for event_timestamp in event:
start = event_timestamp - time_before_event
@@ -118,18 +133,19 @@ def event_triggered_chirps(
event_chirps.append(chirps_around_event)
if len(chirps_around_event) == 0:
continue
else:
else:
centered_chirps.append(chirps_around_event - event_timestamp)
time = np.arange(-time_before_event, time_after_event, dt)
# Kernel density estimation with some if's
if len(centered_chirps) == 0:
centered_chirps = np.array([])
centered_chirps_convolved = np.zeros(len(time))
else:
centered_chirps = np.concatenate(centered_chirps, axis=0) # convert list of arrays to one array for plotting
centered_chirps_convolved = (acausal_kde1d(centered_chirps, time, width)) / len(event)
# convert list of arrays to one array for plotting
centered_chirps = np.concatenate(centered_chirps, axis=0)
centered_chirps_convolved = (acausal_kde1d(
centered_chirps, time, width)) / len(event)
return event_chirps, centered_chirps, centered_chirps_convolved