Spaces:
Sleeping
Sleeping
File size: 3,124 Bytes
a6cc5f0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | class L2ContextMixin:
"""Reusable helper for hierarchical L1/L2 frame alignment."""
def _init_l2_context(
self,
*,
num_l2_context=0,
l2_frame_rate=1.0,
l1_context_frames=1,
require_l2_context=False,
):
self.num_l2_context = int(num_l2_context)
self.l1_context_frames = int(l1_context_frames)
if self.num_l2_context < 0:
raise ValueError(f"num_l2_context must be non-negative, got {self.num_l2_context}.")
if self.l1_context_frames <= 0:
raise ValueError(f"l1_context_frames must be positive, got {self.l1_context_frames}.")
if require_l2_context and self.num_l2_context <= 0:
raise ValueError("num_l2_context must be positive for hierarchical L1/L2 loaders.")
self.l2_frame_interval = None
self._l2_anchor_offset = (self.l1_context_frames - 1) * self.frame_interval
if self.num_l2_context == 0:
return
ratio = float(self.stored_data_frame_rate) / float(l2_frame_rate)
rounded_ratio = round(ratio)
if abs(ratio - rounded_ratio) > 1e-8:
raise ValueError(
"stored_data_frame_rate must be an integer multiple of l2_frame_rate, "
f"got stored_data_frame_rate={self.stored_data_frame_rate}, "
f"l2_frame_rate={l2_frame_rate}."
)
self.l2_frame_interval = int(rounded_ratio)
@property
def l2_context_enabled(self):
return self.num_l2_context > 0
def get_required_l1_start_offset(self):
if not self.l2_context_enabled:
return 0
return max(
0,
(self.num_l2_context - 1) * self.l2_frame_interval - self._l2_anchor_offset,
)
def has_l2_context_for_start(self, start_frame):
if not self.l2_context_enabled:
return True
l2_anchor = start_frame + self._l2_anchor_offset
oldest_l2 = l2_anchor - self.l2_frame_interval * (self.num_l2_context - 1)
return oldest_l2 >= 0
def filter_index_map_with_l2_headroom(self, index_map, start_frame_idx=-1):
if not self.l2_context_enabled:
return list(index_map)
filtered = []
for item in index_map:
start_frame = item[start_frame_idx]
if self.has_l2_context_for_start(start_frame):
filtered.append(item)
return filtered
def get_l1_indices(self, start_frame, num_frames):
return list(range(
start_frame,
start_frame + num_frames * self.frame_interval,
self.frame_interval,
))
def get_l2_indices(self, start_frame):
if not self.l2_context_enabled:
return []
l2_end = start_frame + self._l2_anchor_offset
l2_start = l2_end - (self.num_l2_context - 1) * self.l2_frame_interval
return list(range(l2_start, l2_end + 1, self.l2_frame_interval))
def get_l1_and_l2_indices(self, start_frame, num_frames):
return self.get_l1_indices(start_frame, num_frames), self.get_l2_indices(start_frame)
|