File size: 3,568 Bytes
7344bef
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
85
from __future__ import annotations

import math
import re

import gradio as gr

TIMED_PROMPT_EXAMPLE = "00:00\nA calm cinematic opening shot.\n\n00:30\nThe mood becomes tense and dramatic."
TIMED_PROMPT_TIMESTAMP_RE = re.compile(r"^\d{1,2}:\d{2}(?::\d{2})?(?:\.\d+)?$")


def parse_time_input(value, *, label: str, allow_empty: bool) -> float | None:
    if value is None:
        return None if allow_empty else 0.0
    if isinstance(value, (int, float)):
        if not math.isfinite(float(value)):
            raise gr.Error(f"{label} must be a finite time value.")
        return max(0.0, float(value))
    text = str(value).strip()
    if len(text) == 0:
        return None if allow_empty else 0.0
    if ":" not in text:
        try:
            seconds = float(text)
            if not math.isfinite(seconds):
                raise ValueError
            return max(0.0, seconds)
        except ValueError as exc:
            raise gr.Error(f"{label} must be a number of seconds, MM:SS(.xx), or HH:MM:SS(.xx).") from exc
    parts = text.split(":")
    if len(parts) not in (2, 3):
        raise gr.Error(f"{label} must be a number of seconds, MM:SS(.xx), or HH:MM:SS(.xx).")
    try:
        if len(parts) == 2:
            minutes = int(parts[0])
            seconds = float(parts[1])
            if not math.isfinite(seconds):
                raise ValueError
            return max(0.0, minutes * 60.0 + seconds)
        hours = int(parts[0])
        minutes = int(parts[1])
        seconds = float(parts[2])
        if not math.isfinite(seconds):
            raise ValueError
        return max(0.0, hours * 3600.0 + minutes * 60.0 + seconds)
    except ValueError as exc:
        raise gr.Error(f"{label} must be a number of seconds, MM:SS(.xx), or HH:MM:SS(.xx).") from exc


def parse_prompt_schedule(prompt_text: str) -> list[tuple[float, str]]:
    text = str(prompt_text or "").replace("\r\n", "\n").replace("\r", "\n").strip()
    if len(text) == 0:
        return [(0.0, "")]
    blocks = [block.strip() for block in re.split(r"\n\s*\n", text) if len(block.strip()) > 0]
    first_line = text.split("\n", 1)[0].strip()
    if len(blocks) <= 1 and not TIMED_PROMPT_TIMESTAMP_RE.fullmatch(first_line):
        return [(0.0, text)]
    schedule: list[tuple[float, str]] = []
    for block in blocks:
        lines = block.split("\n")
        timestamp_line = lines[0].strip()
        if not TIMED_PROMPT_TIMESTAMP_RE.fullmatch(timestamp_line):
            raise gr.Error(
                "Timed prompts must be separated by blank lines, and each block must start with a timestamp like MM:SS(.xx) or HH:MM:SS(.xx).\n\n"
                f"Example:\n{TIMED_PROMPT_EXAMPLE}"
            )
        prompt_body = "\n".join(lines[1:]).strip()
        if len(prompt_body) == 0:
            raise gr.Error(
                "Each timed prompt block must contain prompt text after its timestamp.\n\n"
                f"Example:\n{TIMED_PROMPT_EXAMPLE}"
            )
        schedule.append((float(parse_time_input(timestamp_line, label="Timed prompt timestamp", allow_empty=False) or 0.0), prompt_body))
    return sorted(schedule, key=lambda item: item[0])


def resolve_prompt_for_chunk(prompt_schedule: list[tuple[float, str]], chunk_start_seconds: float, default_prompt: str) -> str:
    prompt_text = str(default_prompt or "")
    for start_seconds, scheduled_prompt in prompt_schedule:
        if float(start_seconds) <= float(chunk_start_seconds) + 1e-9:
            prompt_text = scheduled_prompt
        else:
            break
    return prompt_text