Many tracks, one output
Each record is (start, end, priority), and the priority doubles as the track id. Wherever records overlap, the most important one plays. Press play to watch a sweep line build the merged track and catch every second where it switches.
t = 0
At this event
Heap (most important first)
How it works ›Outlined entries have ended but still sit in the heap. They are only thrown out when they reach the top (lazy deletion). Tap to see the heap as a tree.
Edit the records
One record per line: start end priority. Repeat a priority to play the same track more than once.
Every switch
Tap a row to jump the sweep line there.
| Second | From | To | Why |
|---|
Airtime in the output
How the algorithm works
- Turn each record into two events: +1 at its start and −1 at its end. Ranges are half-open,
[start, end), so a record ending at 6 and another starting at 6 hand over cleanly. - Sort the events by time and walk them one distinct moment at a time.
- Keep a counter per track (how many of its records are playing right now) and a heap of tracks ordered by importance.
- An ending record only decrements its counter. Removing it from the middle of a heap would be slow, so it stays there until it floats to the top, and is popped then.
- After applying all events at time t, the top of the heap is the winner. If it is a different track than before, close the current segment at t and open a new one. That moment is a switch.
Sorting costs O(n log n), and each record is pushed and popped at most once, so the whole thing is O(n log n) time and O(n) memory for n records.
- start
- playback begins after silence
- preempt
- a more important track starts and interrupts the one playing
- fallback
- the playing track ends and a less important one takes over for the first time
- resume
- the same, but the track taking over was interrupted earlier and now continues
- end
- nothing is playing any more (a gap, or the very end)
Python
The core function. The full module, with Record, Segment, switches() and airtime(), is here: merge_tracks.py.
def merge(records: Iterable[Record], higher_wins: bool = True) -> list[Segment]:
"""Return the merged single track as non-overlapping segments, in time order.
Gaps where nothing plays are simply absent from the result.
Set higher_wins=False if a smaller number means a more important track.
"""
sign = -1 if higher_wins else 1 # heapq is a min-heap
events: list[tuple[Number, int, int]] = [] # (time, delta, track)
for r in records:
if r.end < r.start:
raise ValueError(f"end before start: {r}")
if r.end == r.start:
continue # zero-length, never audible
events.append((r.start, +1, r.priority))
events.append((r.end, -1, r.priority))
events.sort(key=lambda e: e[0])
active: dict[int, int] = defaultdict(int)
heap: list[int] = [] # sign * track
out: list[Segment] = []
current: Optional[int] = None
opened_at: Number = 0
for t, group in groupby(events, key=lambda e: e[0]):
for _, delta, track in group:
active[track] += delta
if delta > 0 and active[track] == 1:
heapq.heappush(heap, sign * track)
while heap and active[sign * heap[0]] == 0: # lazy deletion
heapq.heappop(heap)
winner = sign * heap[0] if heap else None
if winner != current:
if current is not None and t > opened_at:
out.append(Segment(opened_at, t, current))
current, opened_at = winner, t
return out
segments = merge([Record(0, 10, 1), Record(3, 6, 5), Record(5, 8, 3)])
# [Segment(0, 3, 1), Segment(3, 6, 5), Segment(6, 8, 3), Segment(8, 10, 1)]
for s in switches(segments):
print(s.at, s.from_track, "→", s.to_track, s.reason)
# 0 None → 1 start
# 3 1 → 5 preempt
# 6 5 → 3 fallback
# 8 3 → 1 resume
# 10 1 → None end