Source code for maad.features.composite_soundscape_descriptors

""" 
Utility functions for computing composite soundscape descriptors using data from multiple files at the same sampling site.
"""
import os
from concurrent.futures import ProcessPoolExecutor
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from maad import sound, util, rois

#%% Function argument validation
def _input_validation(data_input):
    """ Validate dataframe or path input argument """
    if isinstance(data_input, pd.DataFrame):
        df = data_input

    elif isinstance(data_input, str):
        if os.path.isdir(data_input):
            print('Collecting metadata from directory path...')
            df = util.get_metadata_dir(data_input)
            df['time'] = df.date.dt.hour
            print('Done!')
        elif os.path.isfile(data_input) and data_input.lower().endswith(".csv"):
            print('Loading metadata from csv file')
            try:
                # Attempt to read all wav data from the provided file path.
                df = pd.read_csv(data_input, dtype={'time': str}) 
                df['date'] = pd.to_datetime(df.date)
                df['time'] = df.date.dt.hour
            except FileNotFoundError:
                raise FileNotFoundError(f"File not found: {data_input}")
            print('Done!')
    
    else:
        raise ValueError("Input 'data' must be either a Pandas DataFrame or a file path string")
    
    return df

def _validate_n_jobs(n_jobs):
    """ Validate number of jobs """
    if not isinstance(n_jobs, int):
        raise ValueError("n_jobs must be an integer.")
    
    if n_jobs <= 0 or n_jobs == -1:
        # Use all available CPUs
        n_jobs = os.cpu_count()
        print(f"Using all available CPUs: {n_jobs}")
    else:
        # Validate that the number is not larger than available CPUs
        available_cpus = os.cpu_count()
        if n_jobs > available_cpus:
            # Set n_jobs to the maximum number of available CPUs
            n_jobs = available_cpus
            print(f"Adjusted n_jobs to maximum available CPUs: {n_jobs}")
        else:
            print(f"Using specified number of CPUs: {n_jobs}")

    return n_jobs

#%%
def _spectral_peak_density(
        path_audio, target_fs, nperseg, noverlap, db_range, min_distance, threshold_abs):
    """
    Computes the spectral peak density for an audio file, representing the number of peaks per time step within each frequency bin.

    Parameters
    ----------
    path_audio : pandas DataFrame
        A Pandas DataFrame containing information about the audio files.
    target_fs : int
        The target sample rate to resample the audio signal if needed.
    nperseg : int
        Window length of each segment to compute the spectrogram.
    noverlap : int
        Number of samples to overlap between segments to compute the spectrogram.
    db_range : float
        Dynamic range of the computed spectrogram.
    min_distance : int
        Minimum number of indices separating peaks.
    threshold_abs : float
        Minimum amplitude threshold for peak detection in decibels.

    Returns
    -------
    peak_density : pandas Dataframe
        The peak density representation of the audio per frequency bin.

    """

    filename = os.path.basename(path_audio)
    print(f'Processing file {filename}', end='\r')

    # Load data
    s, fs = sound.load(path_audio)
    s = sound.resample(s, fs, target_fs, res_type="scipy_poly")
    Sxx, tn, fn, ext = sound.spectrogram(s, target_fs, nperseg=nperseg, noverlap=noverlap)
    Sxx_db = util.power2dB(Sxx, db_range=db_range)

    # Compute local max
    _, peak_freq = rois.spectrogram_local_max(
        Sxx_db, tn, fn, ext,
        min_distance, 
        threshold_abs)

    # Compute peak density (number of peaks / time steps)
    freq_idx, count_freq = np.unique(peak_freq, return_counts=True)
    count_peak = np.zeros(fn.shape)
    bool_index = np.isin(fn, freq_idx)
    indices = np.where(bool_index)[0]
    count_peak[indices] = count_freq / len(tn)
    peak_density = pd.Series(index=fn, data=count_peak)

    # Normalize per time
    peak_density.name = filename
    return peak_density.to_frame().T

#%%
[docs] def graphical_soundscape( data, threshold_abs, path_audio='path_audio', time='time', target_fs=48000, nperseg=256, noverlap=128, db_range=80, min_distance=1, n_jobs=1): """ Computes a graphical soundscape from a given DataFrame of audio files. This function is a variant of the original graphical soundscapes introduced by Campos-Cerqueira et al. The peaks are detected on the spectrogram instead of detecting peaks on the spectrum. Results are similar but not equal to the ones computed using seewave in R. References: - Campos‐Cerqueira, M., et al., 2020. How does FSC forest certification affect the acoustically active fauna in Madre de Dios, Peru? Remote Sensing in Ecology and Conservation 6, 274–285. https://doi.org/10.1002/rse2.120 - Furumo, P.R., Aide, T.M., 2019. Using soundscapes to assess biodiversity in Neotropical oil palm landscapes. Landscape Ecology 34, 911–923. - Campos-Cerqueira, M., Aide, T.M., 2017. Changes in the acoustic structure and composition along a tropical elevational gradient. JEA 1, 1–1. https://doi.org/10.22261/JEA.PNCO7I Parameters ---------- data : pandas DataFrame A Pandas DataFrame containing information about the audio files. If a string is passed with a directory location or a csv file, parameters 'path_audio' and 'time' will be set as default and can't be customized. threshold_abs : float Minimum amplitude threshold for peak detection in decibels. path_audio : str Column name where the full path of audio is provided. time : str Column name where the time is provided as a string using the format 'HHMMSS'. target_fs : int The target sample rate to resample the audio signal if needed. nperseg : int Window length of each segment to compute the spectrogram. noverlap : int Number of samples to overlap between segments to compute the spectrogram. db_range : float Dynamic range of the computed spectrogram. min_distance : int Minimum number of indices separating peaks. n_jobs : int Number of processes to use for parallel computing. Default is 1. Returns ------- res : pandas DataFrame A Pandas DataFrame containing the graphical representation of the soundscape. Examples -------- >>> from maad.util import get_metadata_dir >>> from maad.features import graphical_soundscape, plot_graph >>> df = get_metadata_dir('../../data/indices') >>> df['hour'] = df.date.dt.hour >>> gs = graphical_soundscape(data=df, threshold_abs=-80, time='hour') >>> plot_graph(gs) """ df = _input_validation(data) df.sort_values(by=path_audio, inplace=True) total_files = len(df) print(f'{total_files} files found to process...') flist = df[path_audio].to_list() if n_jobs==1: # Use sequential processing results = [] for path_audio in flist: result = _spectral_peak_density(path_audio, target_fs, nperseg, noverlap, db_range, min_distance, threshold_abs) results.append(result) else: # Use parallel processing _validate_n_jobs(n_jobs) with ProcessPoolExecutor() as executor: futures = [executor.submit(_spectral_peak_density, path_audio, target_fs, nperseg, noverlap, db_range, min_distance, threshold_abs) for path_audio in flist] # Wait for all tasks to complete results = [future.result() for future in futures] res = pd.concat(results) res['time'] = df[time].values print('\nComputation completed!') return res.groupby('time').mean()
#%%
[docs] def plot_graph(graph, ax=None, savefig=False, fname=None): """ Plots a graphical soundscape Parameters ---------- graph : pandas.Dataframe A graphical soundscape as pandas dataframe with index as time and frequency as columns ax : matplotlib.axes, optional Axes for subplots. If not provided it creates a new figure, by default None. Returns ------- ax Axes of the figure """ if ax == None: fig, ax = plt.subplots() ax.imshow(graph.values.T, aspect='auto', origin='lower', extent=[int(graph.index[0]), int(graph.index[-1]), graph.columns[0], graph.columns[-1]]) ax.set_xlabel('Time (h)') ax.set_ylabel('Frequency (Hz)') if savefig: plt.savefig(fname, bbox_inches='tight') return ax