Spaces:
Sleeping
Sleeping
| from typing import List | |
| from backend.schemas import Segment | |
| class TemporalAggregator: | |
| def __init__(self, fps: int = 5, threshold: float = 0.5): | |
| self.fps = fps | |
| self.threshold = threshold | |
| def aggregate(self, frame_scores: List[float]) -> List[Segment]: | |
| """ | |
| Converts a list of frame-level probabilities (e.g. [0.1, 0.9, 0.9]) | |
| into temporal segments (Start=00:00:01, End=00:00:05). | |
| """ | |
| segments = [] | |
| in_segment = False | |
| start_frame_idx = 0 | |
| # Add a dummy safe frame at the end to close any open segments | |
| extended_scores = frame_scores + [0.0] | |
| for i, score in enumerate(extended_scores): | |
| is_fake = score > self.threshold | |
| if is_fake and not in_segment: | |
| # Start of a new fake segment | |
| in_segment = True | |
| start_frame_idx = i | |
| elif not is_fake and in_segment: | |
| # End of a fake segment | |
| in_segment = False | |
| end_frame_idx = i - 1 # Previous frame was the last fake one | |
| # Convert frame indices to timestamps | |
| start_time = self._frame_to_timestamp(start_frame_idx) | |
| end_time = self._frame_to_timestamp(end_frame_idx) | |
| # Calculate average confidence for this segment | |
| segment_scores = frame_scores[start_frame_idx : end_frame_idx + 1] | |
| avg_conf = sum(segment_scores) / len(segment_scores) if segment_scores else 0.0 | |
| segments.append(Segment( | |
| start_time=start_time, | |
| end_time=end_time, | |
| confidence=round(avg_conf, 2) | |
| )) | |
| return segments | |
| def _frame_to_timestamp(self, frame_idx: int) -> str: | |
| total_seconds = frame_idx / self.fps | |
| mins = int(total_seconds // 60) | |
| secs = int(total_seconds % 60) | |
| return f"00:{mins:02d}:{secs:02d}" | |