This commit is contained in:
CroneKorkN 2025-06-01 18:33:56 +02:00
parent e126ca829d
commit d502b91cc8
Signed by: cronekorkn
SSH key fingerprint: SHA256:v0410ZKfuO1QHdgKBsdQNF64xmTxOF8osF1LIqwTcVw

27
process
View file

@ -18,11 +18,10 @@ DETECT_FREQUENCY_TOLERANCE = 2 # Hz
DETECT_FREQUENCY_FROM = DETECT_FREQUENCY - DETECT_FREQUENCY_TOLERANCE # Hz DETECT_FREQUENCY_FROM = DETECT_FREQUENCY - DETECT_FREQUENCY_TOLERANCE # Hz
DETECT_FREQUENCY_TO = DETECT_FREQUENCY + DETECT_FREQUENCY_TOLERANCE # Hz DETECT_FREQUENCY_TO = DETECT_FREQUENCY + DETECT_FREQUENCY_TOLERANCE # Hz
ADJACENCY_FACTOR = 2 # area to look for noise around the target frequency ADJACENCY_FACTOR = 2 # area to look for noise around the target frequency
AMPLITUDE_THRESHOLD = 200 # relative DB (rDB) (because not calibrated)
BLOCK_SECONDS = 3 # seconds (longer means more frequency resolution, but less time resolution) BLOCK_SECONDS = 3 # seconds (longer means more frequency resolution, but less time resolution)
DETECTION_DISTANCE = 30 # seconds (minimum time between detections) DETECTION_DISTANCE = 30 # seconds (minimum time between detections)
BLOCK_OVERLAP_FACTOR = 0.8 # overlap between blocks (0.8 means 80% overlap) BLOCK_OVERLAP_FACTOR = 0.9 # overlap between blocks (0.2 means 20% overlap)
MAX_NOISE = 0.1 # maximum noise level (relative DB) to consider a detection valid MIN_SIGNAL_QUALITY = 1000.0 # maximum noise level (relative DB) to consider a detection valid
def process_recording(filename): def process_recording(filename):
print('processing', filename) print('processing', filename)
@ -41,12 +40,12 @@ def process_recording(filename):
current_event = None current_event = None
# read blocks of audio data with overlap from sound variable # read blocks of audio data with overlap from sound variable
block_num = 0 sample_num = 0
for block in soundfile.blocks(path, blocksize=samples_per_block, overlap=overlapping_samples): for block in soundfile.blocks(path, blocksize=samples_per_block, overlap=overlapping_samples):
block_num += 1 sample_num += samples_per_block - overlapping_samples
# get block date and calculate FFT # get block date and calculate FFT
block_date = recording_date + datetime.timedelta(seconds=block_num * (samples_per_block - overlapping_samples) / samplerate) block_date = recording_date + datetime.timedelta(seconds=sample_num / samplerate)
labels = rfftfreq(len(block), d=1/samplerate) labels = rfftfreq(len(block), d=1/samplerate)
complex_amplitudes = rfft(block) complex_amplitudes = rfft(block)
amplitudes = np.abs(complex_amplitudes) amplitudes = np.abs(complex_amplitudes)
@ -61,19 +60,17 @@ def process_recording(filename):
max_freq = search_labels[max_amplitude_index] max_freq = search_labels[max_amplitude_index]
# get the average amplitude of the search frequencies # get the average amplitude of the search frequencies
adjacent_amplitudes = search_amplitudes[(search_labels < DETECT_FREQUENCY_FROM) | (search_labels > DETECT_FREQUENCY_TO)] adjacent_amplitudes = amplitudes[(labels < DETECT_FREQUENCY_FROM) | (labels > DETECT_FREQUENCY_TO)]
noise = np.mean(adjacent_amplitudes)/max_amplitude signal_quality = max_amplitude/np.mean(adjacent_amplitudes)
# check for detection criteria # check for detection criteria
max_freq_detected = DETECT_FREQUENCY_FROM <= max_freq <= DETECT_FREQUENCY_TO max_freq_detected = DETECT_FREQUENCY_FROM <= max_freq <= DETECT_FREQUENCY_TO
amplitude_detected = max_amplitude > AMPLITUDE_THRESHOLD good_signal_quality = signal_quality > MIN_SIGNAL_QUALITY
low_noise_detected = noise < MAX_NOISE
# conclude detection # conclude detection
if ( if (
max_freq_detected and max_freq_detected and
amplitude_detected and good_signal_quality
low_noise_detected
): ):
# detecting an event # detecting an event
if not current_event: if not current_event:
@ -90,7 +87,7 @@ def process_recording(filename):
'end_freq': max_freq, 'end_freq': max_freq,
'max_amplitude': max(max_amplitude, current_event['max_amplitude']), 'max_amplitude': max(max_amplitude, current_event['max_amplitude']),
}) })
print(f'- {block_date.strftime('%Y-%m-%d %H:%M:%S')}: {max_amplitude:.1f}rDB @ {max_freq:.1f}Hz (noise {noise:.3f}rDB)') print(f'- {block_date.strftime('%Y-%m-%d %H:%M:%S')}: {max_amplitude:.1f}rDB @ {max_freq:.1f}Hz (signal {signal_quality:.3f}x)')
else: else:
# not detecting an event # not detecting an event
if current_event: if current_event:
@ -101,9 +98,9 @@ def process_recording(filename):
write_plot() write_plot()
current_event = None current_event = None
block_num += (DETECTION_DISTANCE // BLOCK_SECONDS) * samples_per_block #block_num += (DETECTION_DISTANCE // BLOCK_SECONDS) * samples_per_block
block_num += 1 #block_num += 1
def write_clip(): def write_clip():