soundbay/utils/metadata_processing.py

import os
from collections import namedtuple
from pathlib import Path
import pandas as pd
import numpy as np
import re
import soundfile as sf
from typing import List


# TODO add tests to the utils in this file

def load_n_adapt_raven_annotation_table_to_dv_dataset_requirements(annotation_file_path: str,
                                                                   annotation_filename_dict: dict = None,
                                                                   filename_suffix: str = ".Table.1.selections.txt"
                                                                   ) -> pd.DataFrame:
    # todo: decide whether to add annotation treatment
    df_annotations = pd.read_csv(annotation_file_path, sep="\t")
    if annotation_filename_dict is not None:
        try:
            df_annotations['filename'] = annotation_filename_dict[os.path.basename(annotation_file_path)].replace('.txt', '')
        except KeyError:
            print(f"KeyError: {os.path.basename(annotation_file_path)}. Using default filename.")
            df_annotations['filename'] = os.path.basename(annotation_file_path).replace(filename_suffix, '')
    else:
        df_annotations['filename'] = os.path.basename(annotation_file_path).replace(filename_suffix, '')
    df_annotations = df_annotations.rename(columns={'Begin Time (s)': 'begin_time', 'End Time (s)': 'end_time'})
    df_annotations['call_length'] = df_annotations['end_time'] - df_annotations['begin_time']
    return df_annotations


# <<<<<<< feature/EDA_script
def raven_to_df_annotations(annotations_path: str,
                            recording_path: str,
                            positive_tag_names: list = ['w', 'sc']):
    """
    Takes annotation files (selection table) created in Raven and turns it to a compatible annotations csv.
    """
    # create dataframe
    annotations = Path(annotations_path)
    recording = Path(recording_path)
    # ignore irrelevent files
    filelist = list(annotations.glob('*selections.txt'))
    metadata = []
    for file in filelist:
        dfTemp = pd.read_csv(file, sep="\t")
        dfTemp['filename'] = re.search(
            "\.Table.1.selections", file.as_posix()).group()
        dfTemp['FirstCallTime'] = np.amin(dfTemp['Begin Time (s)'])
        dfTemp['LastCallTime'] = np.amax(dfTemp['End Time (s)'])
        metadata.append(dfTemp)

    metadata = pd.concat(metadata)
    metadata.rename(columns={'Begin Time (s)': 'begin_time', 'End Time (s)': 'end_time'}, inplace=True)
    print('Number of Labels:', metadata.shape[0])
    for tag in positive_tag_names:
        metadata['Annotation'] = metadata['Annotation'].replace(np.nan, tag, regex=True)
    #add recording length to dataframe
    wav_filelist = list(recording.glob('*.wav'))
    wav_filedict = {re.search("\.Table.1.selections", file.as_posix()).group(): {'path': file} for file in wav_filelist
                    if re.search("\.Table.1.selections", file.as_posix())}
    for key, value in wav_filedict.items():
        record_length = sf.info(value['path']).duration
        value.update({'length': record_length})
        wav_filedict[key] = value
    annotation_lengths = []
    for filename_txt in metadata['filename']:
        annotation_lengths.append(wav_filedict[filename_txt]['length'])
    metadata['TotalRecordLength'] = annotation_lengths
    # filter metadata to contain only desired call types
    filters = positive_tag_names
    metadata_calls = metadata[metadata.Annotation.isin(filters)]
    unique_files, idx = np.unique(metadata['filename'], return_index=True)
    # Find true length of call sequences ( get rid of over lapping-sequences)
    non_overlap_all = non_overlap_df(metadata)
    non_overlap_calls = non_overlap_df(metadata_calls)
    # label background segments
    bg_segments = []
    for file in (unique_files):
        file_df = non_overlap_all[non_overlap_all['filename'] == file]
        begin = np.array(file_df['begin_time'])
        end = np.array(file_df['end_time'])
        for item in end:
            next_beginning = begin[begin > item]
            if next_beginning.size == 0:
                break
            next_beginning = np.min(next_beginning)
            bg_segments.append([item, next_beginning, file])
    bg_segments = pd.DataFrame(bg_segments, columns=['begin_time', 'end_time', 'filename'])
    bg_segments['call_length'] = bg_segments['end_time'] - bg_segments['begin_time']
    bg_segments.sort_values(by=['call_length'])
    # add labels: 0 for background and 1 for call. TODO: modify if there are different call types
    bg_segments['label'] = np.zeros(bg_segments.shape[0], dtype=int)
    non_overlap_calls['label'] = np.ones(non_overlap_calls.shape[0], dtype=int)
    # combine to a csv
    combined_annotations = pd.concat([bg_segments, non_overlap_calls])
    return combined_annotations


def annotations_df_to_csv(annotations_dataset, dataset_name: str = 'recordings_2018_filtered'):
    filename = 'combined_annotations_' + dataset_name + '.csv'
    annotations_dataset.to_csv(filename, index=False)

def get_overlap_pct(ref_start: float, ref_end: float, curr_start: float, curr_end: float) -> float:
    """
    Calculates the percentage of overlap between two time intervals.

    Ensures `ref_start` and `ref_end` represent the earlier interval.

    Parameters:
    - ref_start (float): Start time of the reference interval (earlier interval).
    - ref_end (float): End time of the reference interval.
    - curr_start (float): Start time of the current interval (may start later).
    - curr_end (float): End time of the current interval.

    Returns:
    - float: The overlap percentage, computed as:
      (overlap duration) / (shorter interval length).
      Returns a negative percentage if there is a gap between intervals.
    """
    # Ensure `ref_start` is earlier than `curr_start`
    assert ref_start <= ref_end, 'Reference interval should start before it ends.'
    assert curr_start <= curr_end, 'Current interval should start before it ends.'
    if curr_start < ref_start:
        ref_start, ref_end, curr_start, curr_end = curr_start, curr_end, ref_start, ref_end

    shorter_interval = min(ref_end - ref_start, curr_end - curr_start)

    # Compute overlap as negative if there's no overlap (ref_end < curr_start)
    overlap_duration = min(ref_end, curr_end) - curr_start

    # If the intervals don't overlap, the overlap_duration will be negative
    return overlap_duration / shorter_interval if shorter_interval != 0 else 0


def merge_calls(sorted_df: pd.DataFrame, overlap_pct_th: float = 0) -> List[pd.Series]:
    """
    Args:
        sorted_df: DataFrame with sorted calls by begin_time field
        overlap_pct_th: determines the min [%] overlap between two calls to merge them:
                        * if no overlap - do nothing
                        * if overlap [%] >= overlap_pct_th - merge two calls
                        * if overlap [%] < overlap_pct_th - split equally the overlapping part between cals

    Returns: List of non-overlapping merged calls from the DataFrame

    example:
    >>> df = pd.DataFrame({'begin_time': [5, 8, 10, 18], 'end_time': [6, 16, 20, 27]})
    >>> merge_calls(sorted_df=df, overlap_pct_th=0.5)
    output:
    [
        begin_time    end_time
        5             6          ,
        begin_time    end_time
        8             19         ,
        begin_time    end_time
        19            27
    ]
    """
    if 'call_type' in sorted_df.columns:
        assert sorted_df.call_type.nunique() == 1, 'The function is designed for a single call type.'
    if 'label' in sorted_df.columns:
        assert sorted_df.label.nunique() <= 2, 'The function is designed for a binary label.'

    merged = [sorted_df.iloc[0].copy()]
    for _, higher in sorted_df.iterrows():
        lower = merged[-1].copy()
        overlap_duration = lower.end_time - higher.begin_time
        # test for intersection between lower and higher:
        # we know via sorting that lower[0] <= higher[0]
        if overlap_duration >= 0:
            max_end_time = max(lower.end_time, higher.end_time)
            merged[-1].end_time = max_end_time  # replace by merged interval
            overlap_pct = get_overlap_pct(lower.begin_time, lower.end_time, higher.begin_time, higher.end_time)
            if overlap_pct < overlap_pct_th:
                merged = _split_calls_with_low_overlap(merged, higher, lower, overlap_duration)
        else:
            merged.append(higher.copy())
    return merged

def _split_calls_with_low_overlap(
        merged: List[pd.Series],
        higher: pd.Series,
        lower: pd.Series,
        overlap_duration: float) -> List[pd.Series]:
    """
    Split the overlapping duration equally between two calls.

    merged: list of non-overlapping calls.
    higher: the call that overlaps with the last call in merged.
    lower: the last call in merged.
    overlap_duration: the duration of the overlap between the last call in merged and higher.

    """
    merged[-1].end_time = lower.end_time - overlap_duration / 2
    higher.begin_time += overlap_duration / 2
    merged.append(higher.copy())
    return merged


def non_overlap_df(input_df: pd.DataFrame, overlap_pct_th: float = 0) -> pd.DataFrame:
    """
    Args:
        input_df: DataFrame with possibly overlapping calls

    Returns: a DataFrame object with non-overlapping calls (after merge).
    """
    non_overlap = []
    for file_name, file_df in input_df.groupby(by='filename'):
        file_df.sort_values(by='begin_time', inplace=True)
        merged = merge_calls(file_df, overlap_pct_th)
        non_overlap.extend(merged)
    non_overlap = pd.DataFrame(non_overlap)
    non_overlap['call_length'] = non_overlap['end_time'] - non_overlap['begin_time']

    return non_overlap


def reorder_columns_to_default_view(df: pd.DataFrame):
    """
    Args:
        df: dataframe of the annotations metadata

    Returns: a dataframe with reordered column, so the default order view will be kept
    """
    def get_metadata_fields():
        return ['begin_time', 'end_time', 'filename', 'call_length', 'label']

    orig_cols = df.columns.tolist()
    default_cols = get_metadata_fields()
    remaining = list(set(orig_cols) - set(default_cols))
    new_cols = default_cols + remaining
    return df[new_cols]


def correct_call_times_with_duration(df: pd.DataFrame, audio_files_path: str):
    """
    Args:
        df: dataframe of the annotations metadata
        audio_files_path: str indicates the path to the folder of wav files (given flat hierarchy of audio files)

    Returns:
        df with 'end_time' no longer than the file duration
        it also removes the calls with 'begin_time' longer than duration and prints out a warning
    """

    audio_lengths = [sf.info(f'{audio_files_path}/{file}.wav').duration for file in df['filename']]
    df['audio_length'] = audio_lengths

    end_time_to_long_ind = df['end_time'] > df['audio_length']
    begin_time_to_long_ind = df['begin_time'] > df['audio_length']

    df.loc[end_time_to_long_ind, 'end_time'] = df.loc[end_time_to_long_ind, 'audio_length']
    df.loc[end_time_to_long_ind, 'call_length'] = \
        df.loc[end_time_to_long_ind, 'end_time'] - df.loc[end_time_to_long_ind, 'begin_time']

    if begin_time_to_long_ind.sum() > 0:
        df = df[~begin_time_to_long_ind]
        print(f'removed {begin_time_to_long_ind.sum()} files with begin_time > duration, verify annotations please!')

    return df.drop('audio_length', axis=1)


def bg_from_non_overlap_calls(df: pd.DataFrame):
    """
    Args:
        df: a dataframe of the annotations metadata, with calls that don't overlap

    Returns: a dataframe with bg calls taken from the gaps of the positive calls in a given file
    """
    bg_calls = []
    # Sort by begin time to avoid negative call lengths and erroneously reverse times
    df = df.sort_values(by='begin_time', ascending=True)
    for _, df_per_file in df.groupby(by='filename'):
        # df_per_file is already sorted by begin_time!
        df_per_file_copy = df_per_file.iloc[1:].copy()
        if len(df_per_file_copy) < 1:
            continue
        bg_begin_times = np.array(df_per_file['end_time'])[:-1]
        bg_call_len = np.array(df_per_file['begin_time'])[1:] - bg_begin_times
        df_per_file_copy['begin_time'] = bg_begin_times
        df_per_file_copy['call_length'] = bg_call_len
        df_per_file_copy['end_time'] = df_per_file_copy['begin_time'] + df_per_file_copy['call_length']
        bg_calls.append(df_per_file_copy)
    bg_df = pd.concat(bg_calls)
    bg_df['label'] = 0
    return pd.concat([bg_df, df], ignore_index=True)


def multi_target_from_time_intervals_df(
        df: pd.DataFrame,
        n_classes: int,
        overlap_threshold_pct: float = 0.0,
        noise_class_value: int = 0) -> pd.Series:
    """
    Args:
        df: a dataframe with the columns: 'begin_time', 'end_time', 'label'.
        n_classes: the number of classes in the multi-label target not including noise class.
        overlap_threshold_pct: the minimum overlap between two calls to be considered as a true overlap.
        noise_class_value: the value of the noise class, e.g. 0.

    Returns: a pd.Series of the multi-label target with the df original index.

    example:
        >>> start_times = np.random.uniform(0, 10, 3)
        >>> end_times = start_times + 1
        >>> labels = np.random.choice([1,2], 3)
        >>> df = pd.DataFrame({'begin_time': start_times, 'end_time': end_times, 'label': labels})
        >>> df
                begin_time	end_time	label
            0	4.051811	6.051811	2
            1	8.789995	9.789995	2
            2	5.861857	6.861857	1
    >>> multi_target_from_time_intervals_df(df, overlap_threshold_pct=0, noise_class_value=0)
        0    [1, 1]
        1    [0, 1]
        2    [1, 1]

    If df contains multiple files use it with groupby:
    >>> df.groupby('filename').apply(multi_target_from_time_intervals_df, n_classes=2, overlap_threshold_pct=0, noise_class_value=0)
    """
    assert 0 <= overlap_threshold_pct <= 1, 'overlap_threshold_pct should be in the range [0, 1]'
    assert pd.api.types.is_integer_dtype(df.label), 'label should be an integer type'
    assert n_classes > 0, 'n_classes should be greater than 0'

    Interval = namedtuple('Interval', ['start', 'end', 'label'])
    overlaps = {idx: [0] * n_classes for idx in df.index}
    reference_intervals = {}

    # Process intervals in chronological order
    for idx, row in df.query(f'label != {noise_class_value}').sort_values('begin_time').iterrows():
        interval = Interval(row.begin_time, row.end_time, row.label)

        # Mark the interval as overlapping with itself
        overlaps[idx][int(interval.label) - 1] = 1

        # Remove expired previous intervals (end time < current interval start time)
        reference_intervals = {idx: reference_interval for idx, reference_interval in reference_intervals.items()
                               if reference_interval.end >= interval.start}

        # Check overlaps with reference intervals (overlap >= min_overlap_threshold) and update overlaps
        for ref_idx, ref in reference_intervals.items():
            overlap_pct = get_overlap_pct(ref.start, ref.end, interval.start, interval.end)
            if overlap_pct >= overlap_threshold_pct:
                overlaps[idx][int(ref.label) - 1] = 1
                overlaps[ref_idx][int(interval.label) - 1] = 1

        reference_intervals[idx] = interval

    return pd.Series(overlaps, name='label')