You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
54 lines
1.8 KiB
54 lines
1.8 KiB
from __future__ import annotations |
|
|
|
from decimal import Decimal, InvalidOperation |
|
from typing import Iterable |
|
|
|
from voice_transcriptor.models import ChunkBoundary |
|
|
|
|
|
def _decimal(value: Decimal | str | int | float) -> Decimal: |
|
try: |
|
result = Decimal(str(value)) |
|
except (InvalidOperation, ValueError) as exc: |
|
raise ValueError("Time values must be finite decimal numbers.") from exc |
|
if not result.is_finite(): |
|
raise ValueError("Time values must be finite decimal numbers.") |
|
return result |
|
|
|
|
|
def calculate_chunk_boundaries( |
|
total_duration: Decimal | str | int | float, |
|
chunk_duration: Decimal | str | int | float = Decimal("900"), |
|
overlap: Decimal | str | int | float = Decimal("15"), |
|
) -> tuple[ChunkBoundary, ...]: |
|
total = _decimal(total_duration) |
|
chunk = _decimal(chunk_duration) |
|
shared = _decimal(overlap) |
|
if total <= 0 or chunk <= 0: |
|
raise ValueError("Duration and chunk duration must be greater than zero.") |
|
if shared < 0 or shared >= chunk: |
|
raise ValueError("Overlap must be non-negative and smaller than chunk duration.") |
|
|
|
result: list[ChunkBoundary] = [] |
|
start = Decimal("0") |
|
stride = chunk - shared |
|
while start < total: |
|
end = min(start + chunk, total) |
|
result.append(ChunkBoundary(len(result), start, end)) |
|
if end == total: |
|
break |
|
start += stride |
|
return tuple(result) |
|
|
|
|
|
def total_chunk_duration(boundaries: Iterable[ChunkBoundary]) -> Decimal: |
|
return sum((item.duration_seconds for item in boundaries), Decimal("0")) |
|
|
|
|
|
def source_timestamp( |
|
boundary: ChunkBoundary, local_seconds: Decimal | str | int | float |
|
) -> Decimal: |
|
local = _decimal(local_seconds) |
|
if local < 0 or local > boundary.duration_seconds: |
|
raise ValueError("Local timestamp is outside the chunk.") |
|
return boundary.start_seconds + local
|
|
|