Spaces:
Sleeping
Sleeping
File size: 2,429 Bytes
9bed109 | 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 | import logging
from typing import Any, cast
from shared.schemas.planner_output import PlannerOutput
from shared.schemas.voice_segments import VoiceSegmentTimestamps
logger = logging.getLogger(__name__)
def validate_sync_duration(
video_duration: float, audio_duration: float, threshold: float = 0.5
) -> dict[str, Any]:
"""
Checks if the generated video duration matches the expected audio duration.
"""
if video_duration is None or audio_duration is None:
return {
"video_duration": video_duration,
"audio_duration": audio_duration,
"diff": 0,
"sync_issue": False,
"error": "Missing duration data",
}
diff = abs(video_duration - audio_duration)
is_issue = diff > threshold
return {
"video_duration": round(video_duration, 3),
"audio_duration": round(audio_duration, 3),
"diff": round(diff, 3),
"sync_issue": is_issue,
"threshold": threshold,
}
def align_beats_to_audio(plan: PlannerOutput, timestamps: VoiceSegmentTimestamps) -> dict[str, Any]:
"""
Core sync logic: Maps narrative beats from the execution plan to physical audio timestamps.
Returns a structured 'sync_segments' dictionary.
"""
segments = []
# Version 2 voice timestamps provide paragraph-level 'segments'
ts_segments = timestamps.segments
for i, beat in enumerate(plan.beats):
if i < len(ts_segments):
ts = ts_segments[i]
segments.append(
{
"step_label": beat.step_label,
"start": ts.start,
"end": ts.end,
"duration": round(ts.end - ts.start, 3),
"text_ref": ts.text,
}
)
else:
# Fallback if plan has more beats than voice segments
prev_end = cast(float, segments[-1]["end"]) if segments else 0.0
segments.append(
{
"step_label": beat.step_label,
"start": prev_end,
"end": prev_end + 2.0,
"duration": 2.0,
"text_ref": "[fallback/unvoiced]",
}
)
return {
"version": "1",
"beats": segments,
"total_audio_duration": round(ts_segments[-1].end if ts_segments else 0.0, 3),
}
|