sonogram_utility.py

import cv2
import random
import copy
from pyannote.core import Annotation, Segment
import numpy as np
import soundfile as sf
import torch
import torchaudio
import pandas as pd
import datetime as dt

def colors(n):
    '''
    Creates a list size n of distinctive colors
    
    Creates an arbitrary amount of distinctive colors in RGB format, evenly divided among hues (see HSV). 
    In practice, this proves to be fairly resistant to colorblindness as well.
    
    Parameters
    ----------
    n : int
        Number of distinctive colors required

    Returns
    -------
    ret : list
        List of colors in (BGR) format
    '''
    if n == 0:
        return []
    ret = []
    # Random starting place
    h = int(random.random() * 180)
    # Calculate step size based on number needed.  OpenCV supports Hue from 0-179
    step = 180 / n
    # Iterate across hue dimension to generate colors
    for i in range(n):
        h += step
        h = int(h) % 180
        hsv = np.uint8([[[h,200,200]]])
        bgr = cv2.cvtColor(hsv,cv2.COLOR_HSV2BGR)
        ret.append((bgr[0][0][0].item()/255,bgr[0][0][1].item()/255,bgr[0][0][2].item()/255))
    return ret

def colorsCSS(n):
    '''
    Creates a list size n of distinctive colors
    
    Creates an arbitrary amount of distinctive colors in CSS format, evenly divided among hues (see HSV). 
    In practice, this proves to be fairly resistant to colorblindness as well.
    
    Parameters
    ----------
    n : int
        Number of distinctive colors required

    Returns
    -------
    ret : list
        List of colors in CSS format
    '''
    if n == 0:
        return []
    ret = []
    # Random starting place
    h = int(random.random() * 180)
    # Calculate step size based on number needed.  OpenCV supports Hue from 0-179
    step = 180 / n
    # Iterate across hue dimension to generate colors
    for i in range(n):
        h += step
        h = int(h) % 180
        hsv = np.uint8([[[h,200,200]]])
        bgr = cv2.cvtColor(hsv,cv2.COLOR_HSV2BGR)
        b = f'{bgr[0][0][0].item():02x}'
        g = f'{bgr[0][0][1].item():02x}'
        r = f'{bgr[0][0][2].item():02x}'
        ret.append('#'+b+g+r)
    return ret
    
def extendSpeakers(mySpeakerList, fileLabel = 'NONE', maximumSecondDifference = 1, minimumSecondDuration = 0):
    '''
    (DEPRECATED)
    Extends speaker Segments for Instructor/Audience split data stored as a list
    '''
    mySpeakerAnnotations = Annotation(uri=fileLabel)
    newSpeakerList = [[],[]]
    # Iterate through individual speakers
    for i, speaker in enumerate(mySpeakerList):
        # Rearrange times in chronological order
        speaker.sort()
        lastEnd = -1
        tempSection = None
        # Iterate through sections
        for section in speaker:
            if lastEnd == -1:
                tempSection = copy.deepcopy(section)
                lastEnd = section[0] + section[1]
            else:
                if section[0] - lastEnd <= maximumSecondDifference:
                    tempSection = (tempSection[0],max(section[0] + section[1] - tempSection[0],tempSection[1]))
                    lastEnd = tempSection[0] + tempSection[1]
                else:
                    if tempSection[1] >= minimumSecondDuration:
                        newSpeakerList[i].append(tempSection)
                        mySpeakerAnnotations[Segment(tempSection[0],lastEnd)] = i
                    tempSection = copy.deepcopy(section)
                    lastEnd = section[0] + section[1]
        if tempSection is not None:
            # Add the last section back in
            if tempSection[1] >= minimumSecondDuration:
                newSpeakerList[i].append(tempSection)
                mySpeakerAnnotations[Segment(tempSection[0],lastEnd)] = i
    return newSpeakerList,mySpeakerAnnotations

def twoClassExtendAnnotation(myAnnotation,maximumSecondDifference = 1, minimumSecondDuration = 0):
    '''
    (DEPRECATED)
    Extends speaker Segments for Instructor/Audience split data stored as an Annotation
    '''
    lecturerID = None
    lecturerLen = 0
    
    # Identify lecturer
    for speakerName in myAnnotation.labels():
        tempLen = len(myAnnotation.label_support(speakerName))
        if tempLen > lecturerLen:
            lecturerLen = tempLen
            lecturerID = speakerName

    tempSpeakerList = [[],[]]
    # Recreate speakerList as [[lecturer labels],[audience labels]]
    for speakerName in myAnnotation.labels():
        if speakerName != lecturerID:
            for segmentItem in myAnnotation.label_support(speakerName):
                tempSpeakerList[1].append((segmentItem.start,segmentItem.duration))
        else:
            for segmentItem in myAnnotation.label_support(speakerName):
                tempSpeakerList[0].append((segmentItem.start,segmentItem.duration))
                
    newList, newAnnotation = extendSpeakers(tempSpeakerList, fileLabel = myAnnotation.uri, maximumSecondDifference = maximumSecondDifference, minimumSecondDuration = minimumSecondDuration)

    return newList, newAnnotation

def loadAudioRTTM(sampleRTTM):
    '''
    Loads RTTM file in as list of (speaker times) and as Annotation

    ...

    Parameters
    ----------
    sampleRTTM : str
        Full path to RTTM file to read

    Returns
    -------
    speakerList : list
        List of speakers as (List of times).  Outer list represents speakers, inner list contains (start time, duration) speech segments
    prediction : pyannote.core.Annotation
        Annotation object containing RTTM data
    '''
    # Read in prediction data
    # Data in list form, for convenient plotting
    speakerList = []
    # Data in Annotation form, for convenient error rate calculation
    prediction = Annotation(uri=sampleRTTM)
    with open(sampleRTTM, "r") as rttm:
        # Process line by line
        for line in rttm:
            # Delimited by ' '
            speakerResult = line.split(' ')
            # Assume speaker is identified as number
            index = int(speakerResult[7][-2:])
            # Collect speech time start and end
            start = float(speakerResult[3])
            end = start + float(speakerResult[4])
            # Extend speakerList until a sublist exists for given speaker
            while len(speakerList) < index + 1:
                speakerList.append([])
            # Add to speaker list and Annotation objects
            speakerList[index].append((float(speakerResult[3]),float(speakerResult[4])))
            prediction[Segment(start,end)] = speakerResult[7]

    return speakerList, prediction

def loadAudioTXT(sampleTXT):
    '''
    Loads specially formatted TXT file in as list of (speaker times) and as Annotation

    File to be read should be formatted with rows as:
    (start time in seconds)\t(end time in seconds)\t(speaker ID)

    Parameters
    ----------
    sampleTXT : str
        Full path to specially formatted TXT file to read

    Returns
    -------
    [] : list (DEPRECATED)
        Empty list placeholder
    prediction : pyannote.core.Annotation
        Annotation object containing RTTM data
    '''
    prediction = Annotation(uri=sampleTXT)
    with open(sampleTXT, "r") as txt:
        # Iterate through rows
        for line in txt:
            # Delimited with tabs '\t'
            speakerResult = line.split('\t')
            # For debugging
            print(speakerResult)
            # Expect 3 columns
            if len(speakerResult) < 3:
                continue
            start = float(speakerResult[0])
            end = float(speakerResult[1])
            prediction[Segment(start,end)] = speakerResult[2]

    return [], prediction

def loadAudioCSV(sampleCSV):
    '''
    Loads specially formatted CSV file in as list of (speaker times) and as Annotation

    File to be read should be formatted with first row containing:
        Start,Finish,Resource
    These headers represent start time, end time, and speaker ID.

    Parameters
    ----------
    sampleCSV : str
        Full path to specially formatted CSV file to read

    Returns
    -------
    [] : list (DEPRECATED)
        Empty list placeholder
    prediction : pyannote.core.Annotation
        Annotation object containing RTTM data
    '''
    # Read in prediction data using dataframes
    df = pd.read_csv(sampleCSV)
    
    df = df.reset_index()  # make sure indexes pair with number of rows
    
    # Data in Annotation form, for convenient error rate calculation
    prediction = Annotation(uri=sampleCSV)
    
    for i, row in df.iterrows():
        index = row['Resource']
        start = row['Start']
        end = row['Finish']
        prediction[Segment(start,end)] = index

    return [], prediction
    
def splitIntoTimeSegments(testFile,maxDurationInSeconds=60):
    '''
    Read audio file and split into specified chunks of time

    Reads in audio file and batches audio waveform based on time provided. Useful if the entire audio cannot be loaded simultaneously, as it can be processed in batches.

    Parameters
    ----------
    testFile : str
        Full path to audio file
    maxDurationInSeconds : float or int
        The max length of time for each chunk. Keep in mind that the final chunk will usually be smaller

    Returns
    -------
    audioSegments : list
        List of waveform values, chunked to time specified
    sample_rate : int
        Sample rate of the audio file
    '''
    # Read in data
    data, sample_rate = sf.read(testFile, dtype="float32", always_2d=True)
    # Extract waveform data
    waveform = torch.from_numpy(data.T)  # shape: [channels, samples]
    
    audioSegments = []
    outOfBoundsIndex = waveform.shape[-1]
    currentStart = 0
    # Determine the end of the current chunk being processed
    currentEnd = min(maxDurationInSeconds * sample_rate,outOfBoundsIndex)
    done = False
    while(not done):
        # Chunk waveform and store
        waveformSegment = waveform[:,currentStart:currentEnd]
        audioSegments.append(waveformSegment)
        # Check for end of audio
        if currentEnd >= outOfBoundsIndex:
            done = True
            break
        else:
            # Move to next chunk
            currentStart = currentEnd
            currentEnd = min(currentStart + maxDurationInSeconds * sample_rate,outOfBoundsIndex)
    return audioSegments, sample_rate

def audioNormalize(waveform,sampleRate,stepSizeInSeconds = 2,dbThreshold = -50,dbTarget = -5):
    '''
    Normalize audio loudness based on decibels

    ...

    Parameters
    ----------
    waveform : np.array
        Audio waveform
    sampleRate : int
        Sample rate of source audio file
    stepSizeInSeconds : float or int
        Window to apply normalization to
    dbThreshold : int
        Minimum decibel level to consider below 80
    dbTarget : int
        Maximum decibel level for normalization below 80

    Returns
    -------
    copyWaveform : np.array
        Normalized audio waveform
    '''
    print("In audioNormalize")
    # Create copy of waveform and detach from CPU if necessary
    copyWaveform = waveform.clone().detach()
    print("Waveform copy made")
    # Create transformation from waveform amplitude to decibel
    transform = torchaudio.transforms.AmplitudeToDB(stype="amplitude", top_db=80)
    # Prepare start and end of each normalization chunk
    currStart = 0
    currEnd = int(min(currStart + stepSizeInSeconds * sampleRate, len(copyWaveform[0])-1))
    done = False
    while(not done):
        # Create decibel representation of target chunk
        copyWaveform_db = waveform[:,currStart:currEnd].clone().detach()
        copyWaveform_db = transform(copyWaveform_db)
        if currStart == 0:
            print("First DB level calculated")
            
        # Check first channel to see if above threshold for loudness enhancement
        if torch.max(copyWaveform_db[0]).item() > dbThreshold:
            # Determine how much gain is required
            gain = torch.min(dbTarget - copyWaveform_db[0])
            adjustGain = torchaudio.transforms.Vol(gain,'db')
            # Apply gain increase
            copyWaveform[0][currStart:currEnd] = adjustGain(copyWaveform[0][currStart:currEnd])
        # Check second channel (when applicable) to see if above threshold for loudness enhancement
        if len(copyWaveform_db) > 1:
            if torch.max(copyWaveform_db[1]).item() > dbThreshold:
                # Determine how much gain is required
                gain = torch.min(dbTarget - copyWaveform_db[1])
                adjustGain = torchaudio.transforms.Vol(gain,'db')
                # Apply gain increase
                copyWaveform[1][currStart:currEnd] = adjustGain(copyWaveform[1][currStart:currEnd])
        # Move to next chunk to process
        currStart += int(stepSizeInSeconds * sampleRate)
        if currStart > currEnd:
            done = True
        else:
            currEnd = int(min(currStart + stepSizeInSeconds * sampleRate, len(copyWaveform[0])-1))
    print("Waveform enhanced")
    return copyWaveform

class equalizeVolume(torch.nn.Module):
    '''
    Torch Module wrapper for equalization
    '''
    def forward(self, waveform,sampleRate,stepSizeInSeconds,dbThreshold,dbTarget):
        print("In equalizeVolume")
        waveformDifference = audioNormalize(waveform,sampleRate,stepSizeInSeconds,dbThreshold,dbTarget)
        return waveformDifference

def combineWaveforms(waveformList):
    '''
    Combines waveform that has been split into batches (see splitIntoTimeSegments())

    Parameters
    ----------
    waveformList : list
        List of waveform segments to merge

    Returns
    -------
    : np.array
        Concatenated waveform
    '''
    return torch.cat(waveformList,1)

def annotationToSpeakerList(myAnnotation):
    '''
    Converts pyannote.core.Annotation object into List of speakers with times for easy processing of matplotlib charts.

    Parameters
    ----------
    myAnnotation : pyannote.core.Annotation
        Diarization object

    Returns
    -------
    tempSpeakerList : list
        List of speakers with (list of (start time, duration)). Outer list represents speakers, inner list contains time start and end.
    '''
    tempSpeakerList = []
    tempSpeakerNames = []
    # Iterate through all speakers
    for speakerName in myAnnotation.labels():
        speakerIndex = None
        # If never before seen speaker, add to both lists
        if speakerName not in tempSpeakerNames:
            # Speaker ID is new index
            speakerIndex = len(tempSpeakerNames)
            tempSpeakerNames.append(speakerName)
            tempSpeakerList.append([])
        else:
            # Lookup speaker ID based on name
            speakerIndex = tempSpeakerNames.index(speakerName)

        # Iterate through Segments and add to speaker list
        for segmentItem in myAnnotation.label_support(speakerName):
            tempSpeakerList[speakerIndex].append((segmentItem.start,segmentItem.duration))
    return tempSpeakerList

def speakerListToDataFrame(speakerList):
    '''
    Convert speaker list to pandas.DataFrame object

    ...

    Parameters
    ----------
    speakerList : list
        List of speakers with (list of (start time, duration)). Outer list represents speakers, inner list contains time start and end.

    Returns
    -------
    df : pandas.DataFrame
        DataFrame representation of input
    '''
    dataList = []
    # Iterate through speakers
    for j, row in enumerate(speakerList):
        # Iterate through times
        for k, speakingPoint in enumerate(row):
            # Convert start time into HH:MM:SS:MS format
            h0 = int(speakingPoint[0]//3600)
            m0 = int(speakingPoint[0]%3600//60)
            s0 = int(speakingPoint[0]%60)
            ms0 = int(speakingPoint[0]*1000000%1000000)
            time0 = dt.time(h0,m0,s0,ms0)
            # Set day as today, because plotly needs full datetime
            dtStart = dt.datetime.combine(dt.date.today(), time0)
            # Convert end time into HH:MM:SS:MS format
            endPoint = speakingPoint[0] + speakingPoint[1]
            h1 = int(endPoint//3600)
            m1 = int(endPoint%3600//60)
            s1 = int(endPoint%60)
            ms1 = int(endPoint*1000000%1000000)
            time1 = dt.time(h1,m1,s1,ms1)
            # Set day as today, because plotly needs full datetime
            dtEnd = dt.datetime.combine(dt.date.today(), time1)
            # Add to formatted list for DataFrame
            dataList.append(dict(Task=f"Speaker {j}.{k}", Start=dtStart, Finish=dtEnd, Resource=f"Speaker {j+1}"))
    df = pd.DataFrame(dataList)
    return df

def removeOverlap(timeSegment,overlap):
    '''
    Removes overlap (if any) from two segments of time

    ...

    Parameters
    ----------
    timeSegment : pyannote.core.Segment
        Segment to remove overlap from
    overlap : pyannote.core.Segment
        Segment to apply as overlap mask

    Returns
    -------
    times : list
        List of up to two Segments
    '''
    times = []
    # If first Segment begins before overlap
    if timeSegment.start < overlap.start:
        # Create new Segment which starts at first Segment but ends based on overlap
        # Visual
        # First       ----------------
        # Overlap          -------
        # Result      -----
        times.append(Segment(timeSegment.start,min(overlap.start,timeSegment.end)))
    # If first Segment ends after overlap
    if timeSegment.end > overlap.end:
        # Create new Segment which starts based on overlap but ends when first Segment ends
        # Visual
        # First       ----------------
        # Overlap          -------
        # Result                  ----
        times.append(Segment(max(timeSegment.start,overlap.end),timeSegment.end))
    return times

def checkForOverlap(time1, time2):
    '''
    Checks for overlap of two pyannote.core.Segments

    ...

    Parameters
    ----------
    time1 : pyannote.core.Segment
        First Segment to check
    time2 : pyannote.core.Segment
        Second Segment to check

    Returns
    -------
    overlap : Segment
        Overlapping Segment, or None if none exists
    '''
    overlap = time1 & time2
    if overlap:
        return overlap
    else:
        return None
    
def sumSegments(segmentList):
    '''
    Adds up all durations of provided Segments in list

    ...
    
    Parameters
    ----------
    segmentList : list
        List of pyannote.core.Segment

    Returns
    -------
    total : float or int
        Total duration of all Segments
    '''
    total = 0
    for s in segmentList:
        total += s.duration
    return total

def sumTimes(myAnnotation):
    '''
    Calculates duration of pyannote.core.Annotation

    ...

    Parameters
    ----------
    myAnnotation : pyannote.core.Annotation
        Target Annotation

    Returns
    -------
    : float
        Duration in seconds of Annotation
    '''
    return myAnnotation.get_timeline(False).duration()
    
def sumTimesPerSpeaker(myAnnotation):
    '''
    Calculates duration of each speaker in pyannote.core.Annotation

    ...

    Parameters
    ----------
    myAnnotation : pyannote.core.Annotation
        Target Annotation

    Returns
    -------
    speakerList : list
        List of speakers
    timeList : list
        List of times matching speakerList
    '''
    speakerList = []
    timeList = []
    # Iterate through speakers
    for speaker in myAnnotation.labels():
        # If new speaker, then add to list
        if speaker not in speakerList:
            speakerList.append(speaker)
            timeList.append(0)
        # Get duration of speaker
        timeList[speakerList.index(speaker)] += sumTimes(myAnnotation.subset([speaker]))
    return speakerList, timeList
    
def sumMultiTimesPerSpeaker(myAnnotation):
    '''
    Calculates duration of each speaker in pyannote.core.Annotation, including multi-speaker labels

    Multi-speaker labels can be identified as a str delimited with '+' for each speaker

    Parameters
    ----------
    myAnnotation : pyannote.core.Annotation
        Target Annotation

    Returns
    -------
    speakerList : list
        List of speakers
    timeList : list
        List of times matching speakerList
    '''
    speakerList = []
    timeList = []
    # Get top-level view of durations for speakers
    sList,tList = sumTimesPerSpeaker(myAnnotation)
    # Iterate through speakers
    for i,speakerGroup in enumerate(sList):
        # Split multi-group speakers, normal speakers are treated as list of 1
        speakerSplit = speakerGroup.split('+')
        # For each speaker with associated duration
        for speaker in speakerSplit:
            # If a new speaker, then add to list
            if speaker not in speakerList:
                speakerList.append(speaker)
                timeList.append(0)
            # Add individual speaker duration (not group)
            timeList[speakerList.index(speaker)] += tList[i]
    return speakerList, timeList

def annotationToDataFrame(myAnnotation):
    '''
    Convert pyannote.core.Annotation to specially formatted pandas.DataFrame object

    ...

    Parameters
    ----------
    myAnnotation : pyannote.core.Annotation
        Diarization representation

    Returns
    -------
    df : pandas.DataFrame
        DataFrame representation of input
    timeSummary : dict
        Maps speakers to duration spoken
    '''
    dataList = []
    speakerDict = {}
    # Iterate through speakers
    for currSpeaker in myAnnotation.labels():
        # If new speaker, then create entry
        if currSpeaker not in speakerDict.keys():
            speakerDict[currSpeaker] = []
        # Collect individual segments for speaker
        for currSegment in myAnnotation.subset([currSpeaker]).itersegments():
            speakerDict[currSpeaker].append(currSegment)

    timeSummary = {}
    # Iterate through speakers
    for key in speakerDict.keys():
        # If new speaker (for time calculations), then create entry
        if key not in timeSummary.keys():
            timeSummary[key] = 0
        # Add duration of all segments for speaker
        for speakingSegment in speakerDict[key]:
            timeSummary[key] += speakingSegment.duration

    # Iterate through speakers
    for key in speakerDict.keys():
        # Iterate through segments
        for k, speakingSegment in enumerate(speakerDict[key]):
            # Create specially formatted DataFrame entry
            speakerName = key
            startPoint = speakingSegment.start
            endPoint = speakingSegment.end
            # Convert to HH:MM:SS:MS format
            h0 = int(startPoint//3600)
            m0 = int(startPoint%3600//60)
            s0 = int(startPoint%60)
            ms0 = int(startPoint*1000000%1000000)
            time0 = dt.time(h0,m0,s0,ms0)
            # Set day as today, because plotly needs full datetime
            dtStart = dt.datetime.combine(dt.date.today(), time0)
            # Convert to HH:MM:SS:MS format
            h1 = int(endPoint//3600)
            m1 = int(endPoint%3600//60)
            s1 = int(endPoint%60)
            ms1 = int(endPoint*1000000%1000000)
            time1 = dt.time(h1,m1,s1,ms1)
            # Set day as today, because plotly needs full datetime
            dtEnd = dt.datetime.combine(dt.date.today(), time1)
            dataList.append(dict(Task=speakerName + f".{k}", Start=dtStart, Finish=dtEnd, Resource=speakerName))
    df = pd.DataFrame(dataList)
    return df, timeSummary
    
def annotationToSimpleDataFrame(myAnnotation):
    '''
    Convert pyannote.core.Annotation directly to pandas.DataFrame object

    ...

    Parameters
    ----------
    myAnnotation : pyannote.core.Annotation
        Diarization representation

    Returns
    -------
    df : pandas.DataFrame
        DataFrame representation of input
    timeSummary : dict
        Maps speakers to duration spoken
    '''
    dataList = []
    speakerDict = {}
    # Iterate through speakers
    for currSpeaker in myAnnotation.labels():
        # If new speaker, then add entry
        if currSpeaker not in speakerDict.keys():
            speakerDict[currSpeaker] = []
        # Collect Segments for speaker
        for currSegment in myAnnotation.subset([currSpeaker]).itersegments():
            speakerDict[currSpeaker].append(currSegment)

    timeSummary = {}
    # Iterate through speakers
    for key in speakerDict.keys():
        # If new speaker, then add entry
        if key not in timeSummary.keys():
            timeSummary[key] = 0
        # Calculate duration by summing all durations of Segments
        for speakingSegment in speakerDict[key]:
            timeSummary[key] += speakingSegment.duration

    # Iterate through speakers
    for key in speakerDict.keys():
        # Iterate through Segments
        for k, speakingSegment in enumerate(speakerDict[key]):
            # Create simplified DataFrame entry
            speakerName = key
            startPoint = speakingSegment.start
            endPoint = speakingSegment.end
            dataList.append(dict(Task=speakerName + f".{k}", Start=startPoint, Finish=endPoint, Resource=speakerName))
    df = pd.DataFrame(dataList)
    return df, timeSummary

def calcCategories(myAnnotation,categories):
    '''
    Combines speakers based on categories

    ...

    Parameters
    ----------
    myAnnotation : pyannote.core.Annotation
        Target Annotation
    categories : list
        List of known categories, which contain a list of speakers. List(List(speaker))

    Returns
    -------
    cleanCategories : List
        List of all categories, which contains a list of (speaker,pyannote.core.Segment) pairs. List(List(speaker,Segment)).
        Outer list length = categories + len(extraCategories)
    extraCategories : List
        List of speakers which fit in no category
    '''
    categorySlots = []
    extraCategories = []
    # Initialize categories
    for category in categories:
        categorySlots.append([])
    # Iterate through speakers
    for speaker in myAnnotation.labels():
        # Identify which category speaker belongs to
        targetCategory = None
        for i, category in enumerate(categories):
            if speaker in category:
                targetCategory = i
        # If no category found, then add as "extra category"
        if targetCategory is None:
            targetCategory = len(categorySlots)
            categorySlots.append([])
            extraCategories.append(speaker)
        # Add (speaker,Segment) pair to associated category
        for timeSegment in myAnnotation.subset([speaker]).itersegments():
            categorySlots[targetCategory].append((speaker,timeSegment))
            
    # Clean up categories by merging Segments as necessary
    cleanCategories = []
    # Iterate through categories + extra categories
    for category in categorySlots:
        newCategory = []
        # Copy and sort current category based on start time of Segments
        catSorted = copy.deepcopy(sorted(category,key=lambda cSegment: cSegment[1].start))
        currID, currSegment = None, None
        # If any Segments exist, start at the beginning
        if len(catSorted) > 0:
            currID, currSegment = catSorted[0]
        # Iterate through remaining Segments
        for sp, segmentSlot in catSorted[1:]:
            # Find overlaps
            overlapTime = checkForOverlap(currSegment,segmentSlot)
            # If no overlap with previous Segment, add as normal
            if overlapTime is None:
                newCategory.append((currID,currSegment))
                currID = sp
                currTime = segmentSlot
            # If overlapping previous Segment, then combine into one Segment
            else:
                # Combine names
                currID = currID + "+" + sp
                # Union of segments
                currTime[1] = currSegment | segmentSlot
        # If any Segments existed, then add "clean" category
        if currSegment is not None:
            newCategory.append((currID,currSegment))
        cleanCategories.append(newCategory)
    return cleanCategories,extraCategories

def calcSpeakingTypes(pipeline,myAnnotation,maxTime):
    '''
    Calculates no voice, one voice, and multi voice for a given Annotation

    ...

    Parameters
    ----------
    pipeline : sonogram.Sonogram
        Model object to use for analysis call
    myAnnotation : pyannote.core.Annotation
        Target Annotation
    maxTime : float
        The duration of the audio file. Note that Annotation does NOT strictly provide this.

    Returns
    -------
    nvAnnotation : pyannote.core.Annotation
        Annotation containing only 'no voice' labels
    ovAnnotation : pyannote.core.Annotation
        Annotation containing only 'one voice' labels
    mvAnnotation : pyannote.core.Annotation
        Annotation containing only 'multi voice' labels
    '''
    # Create 3 new Annotations to hold no voice, one voice, and multi voice
    nvAnnotation = Annotation()
    ovAnnotation = Annotation()
    mvAnnotation = Annotation()

    # Generate categories
    categorySegmentList, timeSteps = pipeline.annotationToNoiseList(myAnnotation,maxTime)
    # [group,individual,silence], each as (start,duration)
    print("MultiVoice")
    # Iterate through (speaker,Segment) pairs for multi voice
    for seg in categorySegmentList[0]:
        # Rename 'group' to 'unclear' since group is implied already
        if 'group' in seg[0] or seg[0] is None:
            print(f'unclear : {seg[1]}')
            mvAnnotation[seg[1]] = 'unclear'
        else:
            print(f'{seg[0]} : {seg[1]}')
            mvAnnotation[seg[1]] = seg[0]
    print("OneVoice")
    # Iterate through (speaker,Segment) pairs for one voice
    for seg in categorySegmentList[1]:
        print(f'{seg[0]} : {seg[1]}')
        ovAnnotation[seg[1]] = seg[0]
    print("NoVoice")
    # Iterate through (speaker,Segment) pairs for no voice
    for seg in categorySegmentList[2]:
        print(f'{seg[0]} : {seg[1]}')
        # Name speaker as 'silence' instead of None
        nvAnnotation[seg[1]] = 'silence'
    return nvAnnotation, ovAnnotation, mvAnnotation

def timeToString(timeInSeconds):
    '''
    Convert time(s) into HH:MM:SS.MS format

    ...

    Parameters
    ----------
    timeInSeconds : float or int or list
        Time to convert (in seconds).  May contain a list of times to convert recursively
    '''
    # If list, then format time for each entry
    if isinstance(timeInSeconds,list):
        return [timeToString(t) for t in timeInSeconds]
    else:
        # Format time
        h = int(timeInSeconds//3600)
        m = int(timeInSeconds%3600//60)
        s = timeInSeconds%60
        return f'{h:02d}::{m:02d}::{s:02.2f}'