Moved functions and updated to use Annotations

sonogram_utility.py

@@ -193,4 +193,287 @@ def speakerListToDataFrame(speakerList):
             dtEnd = dt.datetime.combine(dt.date.today(), time1)
             dataList.append(dict(Task=f"Speaker {j}.{k}", Start=dtStart, Finish=dtEnd, Resource=f"Speaker {j+1}"))
     df = pd.DataFrame(dataList)
-    return df
+    return df
+
+def removeOverlap(timeSegment,overlap):
+    times = []
+    if timeSegment.start < overlap.start:
+        times.append(Segment(timeSegment.start,min(overlap.start,timeSegment.end)))
+    if timeSegment.end > overlap.end:
+        times.append(Segment(max(timeSegment.start,overlap.end),timeSegment.end))
+    return times
+
+def checkForOverlap(time1, time2):
+    overlap = time1 & time2
+    if overlap:
+        return overlap
+    else:
+        return None
+    
+
+def calcCategories(annotation,maxTime):
+    noVoice = [Segment(0,maxTime)]
+    oneVoice = []
+    multiVoice = []
+    # TBD Clean this up!!!
+    rawData = {}
+    for speakerName in annotation.labels():
+        if speakerName not in rawData.keys():
+            rawData[speakerName] = []
+        for segmentItem in annotation.label_support(speakerName):
+            rawData[speakerName].append(segmentItem)
+    for speaker in rawData.keys():
+        timesToProcess = []
+        for timeSlot in rawData[speaker]:
+            timesToProcess.append((speaker,timeSlot))
+        while len(timesToProcess) > 0:
+            currID, currTime = timesToProcess[0]
+            timesToProcess.remove(timesToProcess[0])
+            resetCheck = False
+            # Check in multi
+            for compareID,timeSlot in multiVoice:
+                overlapTime = checkForOverlap(currTime,timeSlot)
+                if overlapTime is None:
+                    continue
+                else:
+                    compareID.append(currID)
+                    newTimes = removeOverlap(currTime,timeSlot)#+removeOverlap(timeSlot,currTime)
+                    for i in range(len(newTimes)):
+                        newTimes[i] = (currID,newTimes[i])
+                    timesToProcess += newTimes
+                    resetCheck = True
+                    break
+            if resetCheck:
+                continue
+            # Check in one voice
+            for timeSlot in oneVoice:
+                tID = timeSlot[0]
+                tTime = timeSlot[1]
+                overlapTime = checkForOverlap(currTime,tTime)
+                if overlapTime is None:
+                    continue
+                else:
+                    oneVoice.remove(timeSlot)
+                    # Add back non overlap
+                    newTimes = removeOverlap(tTime,currTime)
+                    for i in range(len(newTimes)):
+                        newTimes[i] = (tID,newTimes[i])
+                    oneVoice += newTimes
+                    # Add overlap time to multivoice
+                    multiVoice.append(([tID,currID],overlapTime))
+                    # Add new times back to process
+                    newTimes = removeOverlap(currTime,tTime)
+                    for i in range(len(newTimes)):
+                        newTimes[i] = (currID,newTimes[i])
+                    timesToProcess += newTimes
+                    resetCheck = True
+                    break
+            if resetCheck:
+                continue
+            # Add to one voice
+            oneVoice.append((currID,currTime))
+    for _,timeSlot in multiVoice:
+        copyOfNo = copy.deepcopy(noVoice)
+        for emptySlot in noVoice:
+            if checkForOverlap(timeSlot,emptySlot) is None:
+                continue
+            else:
+                copyOfNo.remove(emptySlot)
+                copyOfNo += removeOverlap(emptySlot,timeSlot)
+        noVoice = copyOfNo
+    for _,timeSlot in oneVoice:
+        copyOfNo = copy.deepcopy(noVoice)
+        for emptySlot in noVoice:
+            if checkForOverlap(timeSlot,emptySlot) is None:
+                continue
+            else:
+                copyOfNo.remove(emptySlot)
+                copyOfNo += removeOverlap(emptySlot,timeSlot)
+        noVoice = copyOfNo
+    return noVoice, oneVoice, multiVoice, rawData 
+
+def sumTimes(annotation):
+    return annotation.get_timeline(False).duration()
+    
+def sumTimesPerSpeaker(timeSlotList):
+    speakerList = []
+    timeList = []
+    for speaker,timeSlot in timeSlotList:
+        if speaker not in speakerList:
+            speakerList.append(speaker)
+            timeList.append(0)
+        timeList[speakerList.index(speaker)] += timeSlot.duration
+    return speakerList, timeList
+    
+def sumMultiTimesPerSpeaker(timeSlotList):
+    speakerList = []
+    timeList = []
+    sList,tList = sumTimesPerSpeaker(timeSlotList)
+    for i,speakerGroup in enumerate(sList):
+        for speaker in speakerGroup:
+            if speaker not in speakerList:
+                speakerList.append(speaker)
+                timeList.append(0)
+            timeList[speakerList.index(speaker)] += tList[i]
+    return speakerList, timeList
+
+def annotationToDataFrame(myAnnotation):
+    dataList = []
+    speakerDict = {}
+    for currSpeaker in myAnnotation.labels():
+        if currSpeaker not in speakerDict.keys():
+            speakerDict[currSpeaker] = []
+        for currSegment in myAnnotation.subset([speaker]).itersegments():
+            speakerDict[currSpeaker].append(currSegment)
+
+    timeSummary = {}
+    for key in speakerDict.keys():
+        if key not in timeSummary.keys():
+            timeSummary[key] = 0
+        for speakingSegment in speakerDict[key]:
+            timeSummary[key] += speakingSegment.duration
+    
+    for key in speakerDict.keys():
+        for k, speakingSegment in enumerate(speakerDict[key]):
+            speakerName = key
+            startPoint = speakingSegment.start
+            endPoint = speakingSegment.end
+            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)
+            dtStart = dt.datetime.combine(dt.date.today(), time0)
+            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)
+            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 calcCategories(rawData,categories):
+    categorySlots = []
+    extraCategories = []
+    for category in categories:
+        categorySlots.append([])
+    for speaker in rawData.keys():
+        targetCategory = None
+        for i, category in enumerate(categories):
+            if speaker in category:
+                targetCategory = i
+        if targetCategory is None:
+            targetCategory = len(categorySlots)
+            categorySlots.append([])
+            extraCategories.append(speaker)
+            
+        for timeSlot in rawData[speaker]:
+            categorySlots[targetCategory].append((speaker,timeSlot))
+    # Clean up categories
+    cleanCategories = []
+    for category in categorySlots:
+        newCategory = []
+        catSorted = copy.deepcopy(sorted(category,key=lambda slot: slot[1][0]))
+        currID, currTime = None, None
+        if len(catSorted) > 0:
+            currID, currTime = catSorted[0]
+        for sp, timeSlot in catSorted[1:]:
+            overlapTime = checkForOverlap(currTime,timeSlot)
+            if overlapTime is None:
+                newCategory.append((currID,currTime))
+                currID = sp
+                currTime = timeSlot
+            else:
+                currID = currID + "+" + sp
+                currTime[1] = max(currTime[1],timeSlot[1])
+        if currTime is not None:
+            newCategory.append((currID,currTime))
+        cleanCategories.append(newCategory)
+    return cleanCategories,extraCategories
+
+def calcSpeakingTypes(myAnnotation,maxTime):
+    noVoice = [[0,maxTime]]
+    oneVoice = []
+    multiVoice = []
+    for speaker in myAnnotation.labels():
+        timesToProcess = []
+        for timeSegment in myAnnotation.subset([speaker]).itersegments():
+            timesToProcess.append((speaker,timeSegment))
+        while len(timesToProcess) > 0:
+            currID, currSegment = timesToProcess[0]
+            timesToProcess.remove(timesToProcess[0])
+            resetCheck = False
+            # Check in multi
+            for compareID,timeSegment in multiVoice:
+                overlapTime = checkForOverlap(currSegment,timeSegment)
+                if overlapTime is None:
+                    continue
+                else:
+                    compareID.append(currID)
+                    newTimes = removeOverlap(currSegment,timeSegment)
+                    for i in range(len(newTimes)):
+                        newTimes[i] = (currID,newTimes[i])
+                    timesToProcess += newTimes
+                    resetCheck = True
+                    break
+            if resetCheck:
+                continue
+            # Check in one voice
+            for timeSlot in oneVoice:
+                tID = timeSlot[0]
+                tSegment = timeSlot[1]
+                overlapTime = checkForOverlap(currSegment,tSegment)
+                if overlapTime is None:
+                    continue
+                else:
+                    oneVoice.remove(timeSlot)
+                    # Add back non overlap
+                    newTimes = removeOverlap(tSegment,currSegment)
+                    for i in range(len(newTimes)):
+                        newTimes[i] = (tID,newTimes[i])
+                    oneVoice += newTimes
+                    # Add overlap time to multivoice
+                    multiVoice.append(([tID,currID],overlapTime))
+                    # Add new times back to process
+                    newTimes = removeOverlap(currSegment,tSegment)
+                    for i in range(len(newTimes)):
+                        newTimes[i] = (currID,newTimes[i])
+                    timesToProcess += newTimes
+                    resetCheck = True
+                    break
+            if resetCheck:
+                continue
+            # Add to one voice
+            oneVoice.append((currID,currSegment))
+    for _,timeSlot in multiVoice:
+        copyOfNo = copy.deepcopy(noVoice)
+        for emptySlot in noVoice:
+            if checkForOverlap(timeSlot,emptySlot) is None:
+                continue
+            else:
+                copyOfNo.remove(emptySlot)
+                copyOfNo += removeOverlap(emptySlot,timeSlot)
+        noVoice = copyOfNo
+    for _,timeSlot in oneVoice:
+        copyOfNo = copy.deepcopy(noVoice)
+        for emptySlot in noVoice:
+            if checkForOverlap(timeSlot,emptySlot) is None:
+                continue
+            else:
+                copyOfNo.remove(emptySlot)
+                copyOfNo += removeOverlap(emptySlot,timeSlot)
+        noVoice = copyOfNo
+    return noVoice, oneVoice, multiVoice
+
+def timeToString(timeInSeconds):
+    if isinstance(timeInSeconds,list):
+        return [timeToString(t) for t in timeInSeconds]
+    else:
+        h = int(timeInSeconds//3600)
+        m = int(timeInSeconds%3600//60)
+        s = timeInSeconds%60
+        return f'{h:02d}::{m:02d}::{s:02.2f}'