It should be the same question as before.
Use a machine to label phone call recordings. Each recording is 100 seconds long. There are N labelers and only 3 labels [human, machine, silence]. The result is determined by consensus (> N / 2), and the result is also 100 seconds long. If there is any stretch of time with no consensus, you need to raise an error.
input:
labelers =
[
[('0.0', '10.5', 'human'), ('10.5', '20.0', 'machine'), ('20.0', '100.0', 'silence')],
[('0.0', '9.0', 'human'), ('9.0', '20.0', 'silence'), ('20.0', '100.0', 'human')],
[('0.0', '20.0', 'silence'), ('20.0', '100.0', 'human')]
]
output:
labels = [('0.0', '9.0', 'human'), ('9.0', '20.0', 'silence'), ('20.0', '100.0', 'human')]
Good luck!
Update (2026-10-30 13:02 +08:00): I failed. Leaving a working solution here.
from collections import Counter
def consensus_segments(labelers):
"""Pass 1: repeatedly take the earliest head end, vote on that stretch, then advance the heads.
Works directly on the input: each labeler's list is consumed from the front,
so the lists are empty when this returns.
"""
pieces = []
while True:
heads = [labeler[0] for labeler in labelers if labeler]
if not heads:
break
start = min(h[0] for h in heads)
end = min(h[1] for h in heads) # earliest end among the heads
votes = [h[2] for h in heads]
label, count = Counter(votes).most_common(1)[0]
winner = label if count > len(labelers) / 2 else None # needs more than half of the labelers
pieces.append((start, end, winner))
for labeler in labelers:
if labeler:
s, e, lab = labeler[0]
if e == end:
labeler.pop(0) # head fully used up, drop it
else:
labeler[0] = (end, e, lab) # head is longer, shrink it from the left
return pieces
def merge_adjacent(pieces):
"""Pass 2: join touching segments that have the same label."""
if not pieces: return []
merged = []
cur_start, cur_end, cur_label = pieces[0] # start with the first piece
for start, end, label in pieces[1:]: # go through the remaining pieces
if start == cur_end and label == cur_label:
cur_end = end # touching and same label: extend current segment
else:
merged.append((cur_start, cur_end, cur_label)) # close the current segment
cur_start, cur_end, cur_label = start, end, label
merged.append((cur_start, cur_end, cur_label)) # don't forget the last one
return merged
def merge_labelers(labelers):
return merge_adjacent(consensus_segments(labelers))
if __name__ == "__main__":
labelers = [
[(0.0, 10.5, 'human'), (10.5, 20.0, 'machine'), (20.0, 100.0, 'silence')],
[(0.0, 9.0, 'human'), (9.0, 20.0, 'silence'), (20.0, 100.0, 'human')],
[(0.0, 20.0, 'silence'), (20.0, 100.0, 'human')],
]
pieces = consensus_segments(labelers)
print(pieces)
labels = merge_adjacent(pieces)
print(labels)
assert labels == [(0.0, 9.0, 'human'), (9.0, 20.0, 'silence'), (20.0, 100.0, 'human')]
# no label has more than half of the votes -> no winner
no_majority = [
[(0.0, 10.0, 'human')],
[(0.0, 10.0, 'machine')],
[(0.0, 10.0, 'silence')],
]
result = merge_labelers(no_majority)
print(result)
assert result == [(0.0, 10.0, None)]
Discussion
Loading comments…