timeline_utils.py

"""Validation helpers for ARDY's continuous action timeline."""

from __future__ import annotations

import math
from typing import Any


MIN_SEGMENT_SECONDS = 2.0
MAX_SEGMENT_SECONDS = 4.0
MAX_TOTAL_SECONDS = 8.0
MAX_SEGMENTS = 8
MAX_PROMPT_CHARS = 500


def _rows(value: Any) -> list:
    if value is None:
        return []
    if hasattr(value, "values"):
        return value.values.tolist()
    if hasattr(value, "tolist") and not isinstance(value, list):
        return value.tolist()
    return list(value)


def normalize_timeline(value: Any, *, fps: float) -> list[dict]:
    """Normalize Gradio table rows and calculate exact frame ranges."""
    if not math.isfinite(float(fps)) or float(fps) <= 0:
        raise ValueError("FPS must be greater than zero.")

    clean_rows = []
    for row in _rows(value):
        if row is None:
            continue
        row = list(row)
        prompt = " ".join(str(row[0] if row else "").split())
        duration_value = row[1] if len(row) > 1 else None
        if not prompt and duration_value in (None, ""):
            continue
        if not prompt:
            raise ValueError("Every timeline row must contain an action prompt.")
        if len(prompt) > MAX_PROMPT_CHARS:
            raise ValueError(
                f"Each action prompt must contain at most {MAX_PROMPT_CHARS} characters."
            )
        try:
            duration = float(duration_value)
        except (TypeError, ValueError) as exc:
            raise ValueError(
                f"Duration for '{prompt}' must be a number."
            ) from exc
        if not math.isfinite(duration) or not MIN_SEGMENT_SECONDS <= duration <= MAX_SEGMENT_SECONDS:
            raise ValueError(
                f"Each segment must be between {MIN_SEGMENT_SECONDS:g} "
                f"and {MAX_SEGMENT_SECONDS:g} seconds."
            )
        clean_rows.append((prompt, duration))

    if not clean_rows:
        raise ValueError("Add at least one action to the timeline.")
    if len(clean_rows) > MAX_SEGMENTS:
        raise ValueError(f"The timeline supports at most {MAX_SEGMENTS} segments.")
    if sum(duration for _, duration in clean_rows) > MAX_TOTAL_SECONDS + 1e-6:
        raise ValueError(
            f"The combined timeline may not exceed {MAX_TOTAL_SECONDS:g} seconds."
        )

    segments = []
    start_frame = 0
    for prompt, duration in clean_rows:
        frame_count = max(1, int(round(duration * float(fps))))
        end_frame = start_frame + frame_count
        segments.append(
            {
                "prompt": prompt,
                "duration_seconds": frame_count / float(fps),
                "start_frame": start_frame,
                "end_frame": end_frame,
            }
        )
        start_frame = end_frame
    return segments