#!/usr/bin/env python2
# -*- coding: utf-8 -*-
"""Module that measures each continuous CF and FM segment with either
inbuilt or user-defined functions.
"""
from itsfm.signal_processing import *
from itsfm.sanity_checks import make_sure_its_negative
import itsfm.measurement_functions as measurefuncs
from datetime import datetime
import numpy as np
import pandas as pd
from scipy import ndimage
[docs]def measure_hbc_call(call, fs, cf, fm, **kwargs):
'''Performs common or unique measurements on each of the Cf
and FM segments detected.
Parameters
----------
audio : np.array
fs : float>0.
Frequency of sampling in Hz.
cf : np.array
Boolean array with True indicating samples that define the CF
fm : np.array
Boolean array with True indicating samples that define the FM
measurements : list, optional
List with measurement functions
Returns
--------
measurement_values : pd.DataFrame
A wide format dataframe with one row corresponding to all
the measured values for a CF or FM segment
See Also
--------
itsfm.measurement_functions
Example
-------
Create a call with fs and make fake CF and FM segments
>>> fs = 1.0
>>> call = np.random.normal(0,1,100)
>>> cf = np.concatenate((np.tile(0, 50), np.tile(1,50))).astype('bool')
>>> fm = np.invert(cf)
Get the default measurements by not specifying any measurements explicitly.
>>> sound_segments, measures = measure_hbc_call(call, fs,
cf, fm )
>>> print(measures)
And here's an example with some custom functions.The default measurements
will appear in addition to the custom measurements.
>>> from itsfm.measurement_functions import measure_peak_amplitude, measure_peak_frequency
>>> custom_measures = [peak_frequency, measure_peak_amplitude]
>>> sound_segments, measures = measure_hbc_call(call, fs,
cf, fm,
measurements=custom_measures)
'''
all_cf_fm_segments = parse_cffm_segments(cf, fm)
if len(all_cf_fm_segments)==0:
raise ValueError('No CF or FM segments were found -- please re-check')
if kwargs.get('measurements') is not None:
all_measurements = common_measurements() + kwargs['measurements']
else:
all_measurements = common_measurements()
measurement_values = []
for segment in all_cf_fm_segments:
segment_id, segment_indices = segment
segment_measurements = perform_segment_measurements(call, fs,
segment,
all_measurements, **kwargs)
measurement_values.append(segment_measurements)
measurement_values = pd.concat(measurement_values).reset_index(drop=True)
return measurement_values
[docs]def parse_cffm_segments(cf, fm):
'''Recognises continuous stretches of Cf and FM segments,
organises them into separate 'objects' and orders them in time.
Parameters
----------
cf, fm : np.array
Boolean arrays indicating which samples are CF/FM.
Returns
-------
cffm_regions_numbered : np.array with tuples.
Each tuple corresponds to one CF or FM region in the audio.
The tuple has two entries 1) the region identifier, eg. 'fm1'
and 2) the indices that correspond to the region eg. slice(1,50)
Example
-------
# an example sound with two cfs and an fm in the middle
>>> cf = np.array([0,1,1,0,0,0,1,1,0]).astype('bool')
>>> fm = np.array([0,0,0,1,1,1,0,0,0]).astype('bool')
>>> ordered_regions = parse_cffm_segments(cf, fm)
>>> print(ordered_regions)
[['cf1', slice(1, 3, None)], ['fm1', slice(3, 6, None)],
['cf2', slice(6, 8, None)]]
'''
cf_regions, fm_regions = find_regions(cf), find_regions(fm)
cf_fm_regions_ordered = combine_and_order_regions(cf_regions, fm_regions)
cffm_regions_numbered = assign_cffm_regionids(cf_fm_regions_ordered, cf_regions,
fm_regions)
return cffm_regions_numbered
[docs]def find_regions(X):
'''
'''
region_ids, num_regions = ndimage.label(X.flatten())
region_locations = np.array(ndimage.find_objects(region_ids)).flatten()
return region_locations
[docs]def combine_and_order_regions(cf_slices, fm_slices):
'''
'''
cffm_regions = np.sort(np.concatenate((cf_slices, fm_slices)))
return cffm_regions
[docs]def assign_cffm_regionids(cffm, cf_regions, fm_regions):
'''
'''
cf_counter = 1
fm_counter = 1
assigned_ids = []
for i, region in enumerate(cffm):
if region in cf_regions:
regiontype = 'cf'
region_number = str(cf_counter)
cf_counter += 1
elif region in fm_regions:
regiontype = 'fm'
region_number = str(fm_counter)
fm_counter += 1
else:
raise ValueError('Could not find the current regions, please check the region', region)
region_id = regiontype + region_number
assigned_ids.append([region_id, region])
return assigned_ids
[docs]def common_measurements():
'''Loads the default common measurement set
for any region.
'''
common_funcs = [ getattr(measurefuncs, each) for each in ['start', 'stop',
'duration'] ]
return common_funcs