|
|
import numpy as np
|
|
|
|
|
|
from sortedcontainers import SortedList
|
|
|
|
|
|
PYANNOTE_SEGMENT = 'segment'
|
|
|
|
|
|
|
|
|
class Timeline:
|
|
|
@classmethod
|
|
|
def from_df(cls, df, uri = None):
|
|
|
return cls(segments=list(df[PYANNOTE_SEGMENT]), uri=uri)
|
|
|
|
|
|
def __init__(self, segments = None, uri = None):
|
|
|
if segments is None: segments = ()
|
|
|
segments_set = set([segment for segment in segments if segment])
|
|
|
|
|
|
self.segments_set_ = segments_set
|
|
|
self.segments_list_ = SortedList(segments_set)
|
|
|
self.segments_boundaries_ = SortedList((boundary for segment in segments_set for boundary in segment))
|
|
|
self.uri = uri
|
|
|
|
|
|
def __len__(self):
|
|
|
return len(self.segments_set_)
|
|
|
|
|
|
def __nonzero__(self):
|
|
|
return self.__bool__()
|
|
|
|
|
|
def __bool__(self):
|
|
|
return len(self.segments_set_) > 0
|
|
|
|
|
|
def __iter__(self):
|
|
|
return iter(self.segments_list_)
|
|
|
|
|
|
def __getitem__(self, k):
|
|
|
return self.segments_list_[k]
|
|
|
|
|
|
def __eq__(self, other):
|
|
|
return self.segments_set_ == other.segments_set_
|
|
|
|
|
|
def __ne__(self, other):
|
|
|
return self.segments_set_ != other.segments_set_
|
|
|
|
|
|
def index(self, segment):
|
|
|
return self.segments_list_.index(segment)
|
|
|
|
|
|
def add(self, segment):
|
|
|
segments_set_ = self.segments_set_
|
|
|
if segment in segments_set_ or not segment: return self
|
|
|
|
|
|
segments_set_.add(segment)
|
|
|
self.segments_list_.add(segment)
|
|
|
|
|
|
segments_boundaries_ = self.segments_boundaries_
|
|
|
segments_boundaries_.add(segment.start)
|
|
|
segments_boundaries_.add(segment.end)
|
|
|
|
|
|
return self
|
|
|
|
|
|
def remove(self, segment):
|
|
|
segments_set_ = self.segments_set_
|
|
|
if segment not in segments_set_: return self
|
|
|
|
|
|
segments_set_.remove(segment)
|
|
|
self.segments_list_.remove(segment)
|
|
|
|
|
|
segments_boundaries_ = self.segments_boundaries_
|
|
|
segments_boundaries_.remove(segment.start)
|
|
|
segments_boundaries_.remove(segment.end)
|
|
|
|
|
|
return self
|
|
|
|
|
|
def discard(self, segment):
|
|
|
return self.remove(segment)
|
|
|
|
|
|
def __ior__(self, timeline):
|
|
|
return self.update(timeline)
|
|
|
|
|
|
def update(self, timeline):
|
|
|
segments_set = self.segments_set_
|
|
|
segments_set |= timeline.segments_set_
|
|
|
|
|
|
self.segments_list_ = SortedList(segments_set)
|
|
|
self.segments_boundaries_ = SortedList((boundary for segment in segments_set for boundary in segment))
|
|
|
|
|
|
return self
|
|
|
|
|
|
def __or__(self, timeline):
|
|
|
return self.union(timeline)
|
|
|
|
|
|
def union(self, timeline):
|
|
|
return Timeline(segments=self.segments_set_ | timeline.segments_set_, uri=self.uri)
|
|
|
|
|
|
def co_iter(self, other):
|
|
|
for segment in self.segments_list_:
|
|
|
temp = Segment(start=segment.end, end=segment.end)
|
|
|
|
|
|
for other_segment in other.segments_list_.irange(maximum=temp):
|
|
|
if segment.intersects(other_segment): yield segment, other_segment
|
|
|
|
|
|
def crop_iter(self, support, mode = 'intersection', returns_mapping = False):
|
|
|
if mode not in {'loose', 'strict', 'intersection'}: raise ValueError
|
|
|
if not isinstance(support, (Segment, Timeline)): raise TypeError
|
|
|
|
|
|
if isinstance(support, Segment):
|
|
|
support = Timeline(segments=([support] if support else []), uri=self.uri)
|
|
|
|
|
|
for yielded in self.crop_iter(support, mode=mode, returns_mapping=returns_mapping):
|
|
|
yield yielded
|
|
|
|
|
|
return
|
|
|
|
|
|
support = support.support()
|
|
|
|
|
|
if mode == 'loose':
|
|
|
for segment, _ in self.co_iter(support):
|
|
|
yield segment
|
|
|
|
|
|
return
|
|
|
|
|
|
if mode == 'strict':
|
|
|
for segment, other_segment in self.co_iter(support):
|
|
|
if segment in other_segment: yield segment
|
|
|
|
|
|
return
|
|
|
|
|
|
for segment, other_segment in self.co_iter(support):
|
|
|
mapped_to = segment & other_segment
|
|
|
if not mapped_to: continue
|
|
|
|
|
|
if returns_mapping: yield segment, mapped_to
|
|
|
else: yield mapped_to
|
|
|
|
|
|
def crop(self, support, mode = 'intersection', returns_mapping = False):
|
|
|
if mode == 'intersection' and returns_mapping:
|
|
|
segments, mapping = [], {}
|
|
|
|
|
|
for segment, mapped_to in self.crop_iter(support, mode='intersection', returns_mapping=True):
|
|
|
segments.append(mapped_to)
|
|
|
mapping[mapped_to] = mapping.get(mapped_to, list()) + [segment]
|
|
|
|
|
|
return Timeline(segments=segments, uri=self.uri), mapping
|
|
|
|
|
|
return Timeline(segments=self.crop_iter(support, mode=mode), uri=self.uri)
|
|
|
|
|
|
def overlapping(self, t):
|
|
|
return list(self.overlapping_iter(t))
|
|
|
|
|
|
def overlapping_iter(self, t):
|
|
|
for segment in self.segments_list_.irange(maximum=Segment(start=t, end=t)):
|
|
|
if segment.overlaps(t): yield segment
|
|
|
|
|
|
def get_overlap(self):
|
|
|
overlaps_tl = Timeline(uri=self.uri)
|
|
|
|
|
|
for s1, s2 in self.co_iter(self):
|
|
|
if s1 == s2: continue
|
|
|
|
|
|
overlaps_tl.add(s1 & s2)
|
|
|
|
|
|
return overlaps_tl.support()
|
|
|
|
|
|
def extrude(self, removed, mode = 'intersection'):
|
|
|
if isinstance(removed, Segment): removed = Timeline([removed])
|
|
|
|
|
|
if mode == "loose": mode = "strict"
|
|
|
elif mode == "strict": mode = "loose"
|
|
|
|
|
|
return self.crop(removed.gaps(support=Timeline([self.extent()], uri=self.uri)), mode=mode)
|
|
|
|
|
|
def __str__(self):
|
|
|
n = len(self.segments_list_)
|
|
|
string = "["
|
|
|
|
|
|
for i, segment in enumerate(self.segments_list_):
|
|
|
string += str(segment)
|
|
|
string += "\n " if i + 1 < n else ""
|
|
|
|
|
|
string += "]"
|
|
|
return string
|
|
|
|
|
|
def __repr__(self):
|
|
|
return "<Timeline(uri=%s, segments=%s)>" % (self.uri, list(self.segments_list_))
|
|
|
|
|
|
def __contains__(self, included):
|
|
|
if isinstance(included, Segment): return included in self.segments_set_
|
|
|
elif isinstance(included, Timeline): return self.segments_set_.issuperset(included.segments_set_)
|
|
|
else: raise TypeError
|
|
|
|
|
|
def empty(self):
|
|
|
return Timeline(uri=self.uri)
|
|
|
|
|
|
def covers(self, other):
|
|
|
gaps = self.gaps(support=other.extent())
|
|
|
|
|
|
for _ in gaps.co_iter(other):
|
|
|
return False
|
|
|
|
|
|
return True
|
|
|
|
|
|
def copy(self, segment_func = None):
|
|
|
if segment_func is None: return Timeline(segments=self.segments_list_, uri=self.uri)
|
|
|
return Timeline(segments=[segment_func(s) for s in self.segments_list_], uri=self.uri)
|
|
|
|
|
|
def extent(self):
|
|
|
if self.segments_set_:
|
|
|
segments_boundaries_ = self.segments_boundaries_
|
|
|
return Segment(start=segments_boundaries_[0], end=segments_boundaries_[-1])
|
|
|
|
|
|
return Segment(start=0.0, end=0.0)
|
|
|
|
|
|
def support_iter(self, collar = 0.0):
|
|
|
if not self: return
|
|
|
|
|
|
new_segment = self.segments_list_[0]
|
|
|
|
|
|
for segment in self:
|
|
|
possible_gap = segment ^ new_segment
|
|
|
|
|
|
if not possible_gap or possible_gap.duration < collar: new_segment |= segment
|
|
|
else:
|
|
|
yield new_segment
|
|
|
new_segment = segment
|
|
|
|
|
|
yield new_segment
|
|
|
|
|
|
def support(self, collar = 0.):
|
|
|
return Timeline(segments=self.support_iter(collar), uri=self.uri)
|
|
|
|
|
|
def duration(self):
|
|
|
return sum(s.duration for s in self.support_iter())
|
|
|
|
|
|
def gaps_iter(self, support = None):
|
|
|
if support is None: support = self.extent()
|
|
|
if not isinstance(support, (Segment, Timeline)): raise TypeError
|
|
|
|
|
|
if isinstance(support, Segment):
|
|
|
end = support.start
|
|
|
|
|
|
for segment in self.crop(support, mode='intersection').support():
|
|
|
gap = Segment(start=end, end=segment.start)
|
|
|
if gap: yield gap
|
|
|
|
|
|
end = segment.end
|
|
|
|
|
|
gap = Segment(start=end, end=support.end)
|
|
|
if gap: yield gap
|
|
|
elif isinstance(support, Timeline):
|
|
|
for segment in support.support():
|
|
|
for gap in self.gaps_iter(support=segment):
|
|
|
yield gap
|
|
|
|
|
|
def gaps(self, support = None):
|
|
|
return Timeline(segments=self.gaps_iter(support=support), uri=self.uri)
|
|
|
|
|
|
def segmentation(self):
|
|
|
support = self.support()
|
|
|
timestamps = set([])
|
|
|
|
|
|
for (start, end) in self:
|
|
|
timestamps.add(start)
|
|
|
timestamps.add(end)
|
|
|
|
|
|
timestamps = sorted(timestamps)
|
|
|
if len(timestamps) == 0: return Timeline(uri=self.uri)
|
|
|
|
|
|
segments = []
|
|
|
start = timestamps[0]
|
|
|
|
|
|
for end in timestamps[1:]:
|
|
|
segment = Segment(start=start, end=end)
|
|
|
|
|
|
if segment and support.overlapping(segment.middle): segments.append(segment)
|
|
|
start = end
|
|
|
|
|
|
return Timeline(segments=segments, uri=self.uri)
|
|
|
|
|
|
def _iter_uem(self):
|
|
|
uri = self.uri if self.uri else "<NA>"
|
|
|
|
|
|
for segment in self:
|
|
|
yield f"{uri} 1 {segment.start:.3f} {segment.end:.3f}\n"
|
|
|
|
|
|
def to_uem(self):
|
|
|
return "".join([line for line in self._iter_uem()])
|
|
|
|
|
|
def write_uem(self, file):
|
|
|
for line in self._iter_uem():
|
|
|
file.write(line)
|
|
|
|
|
|
def _repr_png_(self):
|
|
|
return None
|
|
|
|
|
|
class Segment:
|
|
|
def __init__(self, start, end):
|
|
|
self.start = start
|
|
|
self.end = end
|
|
|
|
|
|
@staticmethod
|
|
|
def set_precision(ndigits = None):
|
|
|
global AUTO_ROUND_TIME, SEGMENT_PRECISION
|
|
|
|
|
|
if ndigits is None:
|
|
|
AUTO_ROUND_TIME = False
|
|
|
SEGMENT_PRECISION = 1e-6
|
|
|
else:
|
|
|
AUTO_ROUND_TIME = True
|
|
|
SEGMENT_PRECISION = 10 ** (-ndigits)
|
|
|
|
|
|
def __bool__(self):
|
|
|
return bool((self.end - self.start) > SEGMENT_PRECISION)
|
|
|
|
|
|
def __post_init__(self):
|
|
|
if AUTO_ROUND_TIME:
|
|
|
object.__setattr__(self, 'start', int(self.start / SEGMENT_PRECISION + 0.5) * SEGMENT_PRECISION)
|
|
|
object.__setattr__(self, 'end', int(self.end / SEGMENT_PRECISION + 0.5) * SEGMENT_PRECISION)
|
|
|
|
|
|
@property
|
|
|
def duration(self):
|
|
|
return self.end - self.start if self else 0.
|
|
|
|
|
|
@property
|
|
|
def middle(self):
|
|
|
return .5 * (self.start + self.end)
|
|
|
|
|
|
def __iter__(self):
|
|
|
yield self.start
|
|
|
yield self.end
|
|
|
|
|
|
def copy(self):
|
|
|
return Segment(start=self.start, end=self.end)
|
|
|
|
|
|
def __contains__(self, other):
|
|
|
return (self.start <= other.start) and (self.end >= other.end)
|
|
|
|
|
|
def __and__(self, other):
|
|
|
return Segment(start=max(self.start, other.start), end=min(self.end, other.end))
|
|
|
|
|
|
def intersects(self, other):
|
|
|
return (self.start < other.start and other.start < self.end - SEGMENT_PRECISION) or (self.start > other.start and self.start < other.end - SEGMENT_PRECISION) or (self.start == other.start)
|
|
|
|
|
|
def overlaps(self, t):
|
|
|
return self.start <= t and self.end >= t
|
|
|
|
|
|
def __or__(self, other):
|
|
|
if not self: return other
|
|
|
if not other: return self
|
|
|
|
|
|
return Segment(start=min(self.start, other.start), end=max(self.end, other.end))
|
|
|
|
|
|
def __xor__(self, other):
|
|
|
if (not self) or (not other): raise ValueError
|
|
|
|
|
|
return Segment(start=min(self.end, other.end), end=max(self.start, other.start))
|
|
|
|
|
|
def _str_helper(self, seconds):
|
|
|
from datetime import timedelta
|
|
|
|
|
|
negative = seconds < 0
|
|
|
td = timedelta(seconds=abs(seconds))
|
|
|
|
|
|
hours, remainder = divmod(td.seconds + 86400 * td.days, 3600)
|
|
|
minutes, seconds = divmod(remainder, 60)
|
|
|
|
|
|
return '%s%02d:%02d:%02d.%03d' % ('-' if negative else ' ', hours, minutes, seconds, td.microseconds / 1000)
|
|
|
|
|
|
def __str__(self):
|
|
|
if self: return '[%s --> %s]' % (self._str_helper(self.start), self._str_helper(self.end))
|
|
|
return '[]'
|
|
|
|
|
|
def __repr__(self):
|
|
|
return '<Segment(%g, %g)>' % (self.start, self.end)
|
|
|
|
|
|
def _repr_png_(self):
|
|
|
return None
|
|
|
|
|
|
class SlidingWindow:
|
|
|
def __init__(self, duration=0.030, step=0.010, start=0.000, end=None):
|
|
|
if duration <= 0: raise ValueError
|
|
|
self.__duration = duration
|
|
|
if step <= 0: raise ValueError
|
|
|
|
|
|
self.__step = step
|
|
|
self.__start = start
|
|
|
|
|
|
if end is None: self.__end = np.inf
|
|
|
else:
|
|
|
if end <= start: raise ValueError
|
|
|
self.__end = end
|
|
|
|
|
|
self.__i = -1
|
|
|
|
|
|
@property
|
|
|
def start(self):
|
|
|
return self.__start
|
|
|
|
|
|
@property
|
|
|
def end(self):
|
|
|
return self.__end
|
|
|
|
|
|
@property
|
|
|
def step(self):
|
|
|
return self.__step
|
|
|
|
|
|
@property
|
|
|
def duration(self):
|
|
|
return self.__duration
|
|
|
|
|
|
def closest_frame(self, t):
|
|
|
return int(np.rint((t - self.__start - .5 * self.__duration) / self.__step))
|
|
|
|
|
|
def samples(self, from_duration, mode = 'strict'):
|
|
|
if mode == 'strict': return int(np.floor((from_duration - self.duration) / self.step)) + 1
|
|
|
elif mode == 'loose': return int(np.floor((from_duration + self.duration) / self.step))
|
|
|
elif mode == 'center': return int(np.rint((from_duration / self.step)))
|
|
|
|
|
|
def crop(self, focus, mode = 'loose', fixed = None, return_ranges = False):
|
|
|
if not isinstance(focus, (Segment, Timeline)): raise TypeError
|
|
|
|
|
|
if isinstance(focus, Timeline):
|
|
|
if fixed is not None: raise ValueError
|
|
|
|
|
|
if return_ranges:
|
|
|
ranges = []
|
|
|
|
|
|
for i, s in enumerate(focus.support()):
|
|
|
rng = self.crop(s, mode=mode, fixed=fixed, return_ranges=True)
|
|
|
|
|
|
if i == 0 or rng[0][0] > ranges[-1][1]: ranges += rng
|
|
|
else: ranges[-1][1] = rng[0][1]
|
|
|
|
|
|
return ranges
|
|
|
|
|
|
return np.unique(np.hstack([self.crop(s, mode=mode, fixed=fixed, return_ranges=False) for s in focus.support()]))
|
|
|
|
|
|
if mode == 'loose':
|
|
|
i = int(np.ceil((focus.start - self.duration - self.start) / self.step))
|
|
|
|
|
|
if fixed is None:
|
|
|
j = int(np.floor((focus.end - self.start) / self.step))
|
|
|
rng = (i, j + 1)
|
|
|
else:
|
|
|
n = self.samples(fixed, mode='loose')
|
|
|
rng = (i, i + n)
|
|
|
elif mode == 'strict':
|
|
|
i = int(np.ceil((focus.start - self.start) / self.step))
|
|
|
|
|
|
if fixed is None:
|
|
|
j = int(np.floor((focus.end - self.duration - self.start) / self.step))
|
|
|
rng = (i, j + 1)
|
|
|
else:
|
|
|
n = self.samples(fixed, mode='strict')
|
|
|
rng = (i, i + n)
|
|
|
elif mode == 'center':
|
|
|
i = self.closest_frame(focus.start)
|
|
|
|
|
|
if fixed is None:
|
|
|
j = self.closest_frame(focus.end)
|
|
|
rng = (i, j + 1)
|
|
|
else:
|
|
|
n = self.samples(fixed, mode='center')
|
|
|
rng = (i, i + n)
|
|
|
else: raise ValueError
|
|
|
|
|
|
if return_ranges: return [list(rng)]
|
|
|
return np.array(range(*rng), dtype=np.int64)
|
|
|
|
|
|
def segmentToRange(self, segment):
|
|
|
return self.segment_to_range(segment)
|
|
|
|
|
|
def segment_to_range(self, segment):
|
|
|
return self.closest_frame(segment.start), int(segment.duration / self.step) + 1
|
|
|
|
|
|
def rangeToSegment(self, i0, n):
|
|
|
return self.range_to_segment(i0, n)
|
|
|
|
|
|
def range_to_segment(self, i0, n):
|
|
|
start = self.__start + (i0 - .5) * self.__step + .5 * self.__duration
|
|
|
|
|
|
if i0 == 0: start = self.start
|
|
|
return Segment(start, start + (n * self.__step))
|
|
|
|
|
|
def samplesToDuration(self, nSamples):
|
|
|
return self.samples_to_duration(nSamples)
|
|
|
|
|
|
def samples_to_duration(self, n_samples):
|
|
|
return self.range_to_segment(0, n_samples).duration
|
|
|
|
|
|
def durationToSamples(self, duration):
|
|
|
return self.duration_to_samples(duration)
|
|
|
|
|
|
def duration_to_samples(self, duration):
|
|
|
return self.segment_to_range(Segment(0, duration))[1]
|
|
|
|
|
|
def __getitem__(self, i):
|
|
|
start = self.__start + i * self.__step
|
|
|
if start >= self.__end: return None
|
|
|
|
|
|
return Segment(start=start, end=start + self.__duration)
|
|
|
|
|
|
def next(self):
|
|
|
return self.__next__()
|
|
|
|
|
|
def __next__(self):
|
|
|
self.__i += 1
|
|
|
window = self[self.__i]
|
|
|
|
|
|
if window: return window
|
|
|
else: raise StopIteration()
|
|
|
|
|
|
def __iter__(self):
|
|
|
self.__i = -1
|
|
|
return self
|
|
|
|
|
|
def __len__(self):
|
|
|
if np.isinf(self.__end): raise ValueError
|
|
|
i = self.closest_frame(self.__end)
|
|
|
|
|
|
while (self[i]):
|
|
|
i += 1
|
|
|
|
|
|
length = i
|
|
|
return length
|
|
|
|
|
|
def copy(self):
|
|
|
return self.__class__(duration=self.duration, step=self.step, start=self.start, end=self.end)
|
|
|
|
|
|
def __call__(self, support, align_last = False):
|
|
|
if isinstance(support, Timeline): segments = support
|
|
|
elif isinstance(support, Segment): segments = Timeline(segments=[support])
|
|
|
else: raise TypeError
|
|
|
|
|
|
for segment in segments:
|
|
|
if segment.duration < self.duration: continue
|
|
|
|
|
|
for s in SlidingWindow(duration=self.duration, step=self.step, start=segment.start, end=segment.end):
|
|
|
if s in segment:
|
|
|
yield s
|
|
|
last = s
|
|
|
|
|
|
if align_last and last.end < segment.end: yield Segment(start=segment.end - self.duration, end=segment.end) |
