LogalDeveloper/Owlbot
0
Fork 0
Code Issues 13 Actions Packages Releases 3 Wiki Activity

Refactored clips module into layered architecture.
CI / Formatting (push) Successful in 5s
Details
CI / Linting (push) Successful in 6s
Details
CI / Tests (Python 3.12) (push) Successful in 15s
Details
CI / Tests (Python 3.13) (push) Successful in 18s
Details
CI / Tests (Python 3.14) (push) Successful in 12s
Details
CI / Type Checking (push) Successful in 10s
Details
CI / Spelling (push) Successful in 5s
Details

Browse Source
This commit is contained in:
LogalDeveloper
2026-04-13 13:11:20 -04:00
parent 5e712d2059
commit d3e3460ce8
10 changed files with 1142 additions and 909 deletions
Download Patch File Download Diff File Expand all files Collapse all files
owlbot/builtin_modules/clips/cache.py
+279 -376
View File
@@ -12,10 +12,11 @@
# See the License for the specific language governing permissions and
# limitations under the License.
"""HLS stream caching subsystem for the clips module.
"""HLS stream caching collaborator for the clips module.
Polls the Owncast HLS variant playlist, downloads new segments, and manages
a rolling window of cached chunks on disk.
a rolling window of cached chunks on disk. Has no dependency on the Owlbot
module framework; all external resources are passed in explicitly.
"""
from __future__ import annotations
@@ -30,15 +31,13 @@ from pathlib import Path
from typing import TYPE_CHECKING
from urllib.parse import urljoin
import aiohttp
import m3u8 # type: ignore[import-untyped]
if TYPE_CHECKING:
import logging
from collections.abc import AsyncIterator
from owlbot.api import ModuleContext
from .types import get_state
_SEGMENT_URI_RE = re.compile(r"stream-([A-Za-z0-9]+)-(\d+)\.(ts|m4s)$")
_MAX_RETRIES = 10
@@ -55,113 +54,56 @@ class ChunkCache:
def __init__(
self,
http: aiohttp.ClientSession,
cache_dir: Path,
cache_duration: int,
min_clip_length: int,
segment_id: str,
extension: str,
target_duration: float,
variant_url: str,
logger: logging.Logger,
init_path: Path | None = None,
) -> None:
"""Initialize the chunk cache.
:param http: HTTP client session for polling.
:param cache_dir: Directory to store cached segment files.
:param cache_duration: Maximum cache window in seconds.
:param min_clip_length: Minimum clip duration in seconds.
:param segment_id: Unique stream segment identifier (e.g. ``jdofFGg``).
:param extension: Segment file extension without dot (e.g. ``ts`` or ``m4s``).
:param target_duration: HLS target segment duration in seconds.
:param variant_url: URL to the variant m3u8 playlist.
:param logger: Logger instance.
:param init_path: Path to fMP4 initialization segment, if applicable.
"""
self._http = http
self._cache_dir = cache_dir
self._cache_duration = cache_duration
self._min_clip_length = min_clip_length
self._segment_id = segment_id
self._extension = extension
self._target_duration = target_duration
self._variant_url = variant_url
self._logger = logger
self._init_path = init_path
self._sequences: set[int] = set()
self._target_duration: float = 0.0
self._prune_guard_count: int = 0
self._init_path: Path | None = None
def __len__(self) -> int:
"""Return the number of cached segments."""
return len(self._sequences)
self._task: asyncio.Task[None] | None = None
@property
def total_duration(self) -> float:
def buffered_duration(self) -> float:
"""Return total duration of cached chunks in seconds."""
return len(self._sequences) * self._target_duration
@property
def cache_dir(self) -> Path:
"""Return the cache directory path."""
return self._cache_dir
@property
def sequences(self) -> set[int]:
"""Return the set of cached sequence numbers."""
return self._sequences.copy()
def set_target_duration(self, duration: float) -> None:
"""Set the target segment duration from the HLS playlist.
:param duration: Target segment duration in seconds.
"""
self._target_duration = duration
def set_init_path(self, path: Path) -> None:
"""Set the path to the fMP4 initialization segment.
:param path: Path to the downloaded init segment.
"""
self._init_path = path
def segment_path(self, sequence: int) -> Path:
"""Return the file path for a segment by sequence number.
:param sequence: HLS media sequence number.
:return: Path where the segment file should be stored.
"""
def _segment_path(self, sequence: int) -> Path:
"""Return the file path for a segment by sequence number."""
return (
self._cache_dir / f"stream-{self._segment_id}-{sequence}.{self._extension}"
)
def has_chunk(self, sequence: int) -> bool:
"""Check if a segment is in the cache.
:param sequence: HLS media sequence number.
:return: True if the segment is cached.
"""
return sequence in self._sequences
def add_chunk(self, sequence: int) -> None:
"""Add a segment to the cache.
:param sequence: HLS media sequence number.
"""
self._sequences.add(sequence)
def prune(self) -> bool:
"""Remove oldest segments until total duration is within the cache window.
:return: True if any segments were pruned, False otherwise.
"""
if self._prune_guard_count > 0:
return False
pruned = False
while self.total_duration > self._cache_duration and len(self._sequences) > 1:
oldest = min(self._sequences)
self._sequences.discard(oldest)
path = self.segment_path(oldest)
if path.exists():
path.unlink()
pruned = True
return pruned
@asynccontextmanager
async def suppress_pruning(self) -> AsyncIterator[None]:
"""Context manager that suppresses pruning while reading from the cache.
Use as:
async with cache.suppress_pruning():
# ffmpeg reads from cache here
Supports nested calls via reference counting.
"""
self._prune_guard_count += 1
@@ -170,24 +112,6 @@ class ChunkCache:
finally:
self._prune_guard_count -= 1
@property
def enough_for_clipping(self) -> bool:
"""Check if cached data is sufficient for clipping.
Requires at least 2x the configured min clip length to ensure enough
buffered data.
:return: True if total cached duration >= 2 * min_clip_length.
"""
return self.total_duration >= self._min_clip_length * 2
def sorted_sequences(self) -> list[int]:
"""Return cached sequence numbers in ascending order.
:return: Sorted list of sequence numbers.
"""
return sorted(self._sequences)
def concat_protocol_string(self) -> str:
"""Build the ffmpeg concat protocol input string.
@@ -198,276 +122,66 @@ class ChunkCache:
parts: list[str] = []
if self._init_path is not None:
parts.append(str(self._init_path))
parts.extend(str(self.segment_path(seq)) for seq in self.sorted_sequences())
parts.extend(str(self._segment_path(seq)) for seq in sorted(self._sequences))
return "|".join(parts)
def cleanup(self) -> None:
"""Remove the cache directory and all its contents."""
if self._cache_dir.exists():
shutil.rmtree(self._cache_dir)
def start(self) -> None:
"""Start the background polling task."""
if self._task is not None and not self._task.done():
return
self._task = asyncio.create_task(
self._polling_loop(), name="Clips Module - HLS polling loop"
)
async def stop(self) -> None:
"""Cancel the internal polling task and clean up cached files."""
if self._task is not None:
self._task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._task
self._task = None
await asyncio.to_thread(shutil.rmtree, self._cache_dir, True)
self._sequences.clear()
async def _download_segment(self, segment_url: str, seq: int) -> bool:
"""Download a single HLS segment and add it to the cache.
async def start_caching(ctx: ModuleContext) -> None:
"""Start the HLS caching background task.
Creates a cache directory and launches the background polling loop
that downloads new stream segments.
:param ctx: The module context.
"""
module_state = get_state(ctx)
if module_state.cache is not None:
ctx.logger.debug("Caching already active, skipping start.")
return
owncast_url = ctx.owncast_client.base_url
master_url = f"{owncast_url.rstrip('/')}/hls/stream.m3u8"
cache_duration = int(ctx.config.get("cache_duration"))
min_clip_length = int(ctx.config.get("min_clip_length"))
cache_dir_path = Path(tempfile.mkdtemp(prefix="owlbot-clips-"))
task = asyncio.create_task(
_polling_loop(ctx, master_url, cache_dir_path, cache_duration, min_clip_length),
)
module_state.cache_task = task
ctx.logger.info(f"HLS caching started. Cache dir: {cache_dir_path}")
async def stop_caching(ctx: ModuleContext) -> None:
"""Stop the HLS caching background task and clean up temp files.
:param ctx: The module context.
"""
module_state = get_state(ctx)
if module_state.cache_task is not None:
module_state.cache_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await module_state.cache_task
module_state.cache_task = None
if module_state.cache is not None:
module_state.cache.cleanup()
module_state.cache = None
ctx.logger.info("HLS cache stopped and cleaned up.")
async def _fetch_playlist(
ctx: ModuleContext,
url: str,
label: str,
) -> m3u8.M3U8 | None:
"""Fetch and parse an HLS playlist with retry logic.
Attempts up to `_MAX_RETRIES` fetches with a
`_RETRY_DELAY`-second backoff between failures.
Propagates `~asyncio.CancelledError` immediately.
:param ctx: The module context.
:param url: URL of the m3u8 playlist.
:param label: Human-readable label for log messages.
:return: Parsed playlist, or ``None`` if all attempts are exhausted.
"""
for attempt in range(1, _MAX_RETRIES + 1):
reason: str | None = None
exc_info = False
:param segment_url: URL to download the segment from.
:param seq: HLS media sequence number.
:return: True if the segment was downloaded successfully, False otherwise.
"""
try:
async with ctx.http.get(url) as resp:
if resp.status != 200:
reason = f"HTTP {resp.status}"
else:
content = await resp.text()
return m3u8.loads(content)
async with self._http.get(segment_url) as seg_resp:
if seg_resp.status != 200:
return False
data = await seg_resp.read()
except asyncio.CancelledError:
raise
except Exception as e:
reason = str(e)
exc_info = True
ctx.logger.warning(
f"{label} fetch failed (attempt {attempt}/{_MAX_RETRIES}): "
f"{reason}. Retrying in {_RETRY_DELAY:.0f}s.",
exc_info=exc_info,
)
await asyncio.sleep(_RETRY_DELAY)
ctx.logger.error(f"{label} failed after {_MAX_RETRIES} attempts.")
return None
except aiohttp.ClientError:
self._logger.debug(
"Network error downloading segment %d.", seq, exc_info=True
)
return False
chunk_path = self._segment_path(seq)
await asyncio.to_thread(chunk_path.write_bytes, data)
async def _download_segment(
ctx: ModuleContext,
cache: ChunkCache,
segment_url: str,
seq: int,
) -> bool:
"""Download a single HLS segment and add it to the cache.
:param ctx: The module context.
:param cache: The ChunkCache to add the segment to.
:param segment_url: URL to download the .ts segment from.
:param seq: HLS media sequence number.
:return: True if the segment was downloaded successfully, False otherwise.
"""
try:
async with ctx.http.get(segment_url) as seg_resp:
if seg_resp.status != 200:
return False
data = await seg_resp.read()
except Exception:
ctx.logger.debug("Network error downloading segment %d.", seq, exc_info=True)
return False
chunk_path = cache.segment_path(seq)
await asyncio.to_thread(chunk_path.write_bytes, data)
cache.add_chunk(seq)
ctx.logger.debug("Cached segment %d (%d bytes).", seq, len(data))
return True
async def _download_init_segment(
ctx: ModuleContext,
cache: ChunkCache,
init_url: str,
filename: str,
) -> bool:
"""Download the fMP4 initialization segment with retry logic.
Attempts up to `_MAX_RETRIES` downloads with a
`_RETRY_DELAY`-second backoff between failures.
Propagates `~asyncio.CancelledError` immediately.
:param ctx: The module context.
:param cache: The ChunkCache to set the init path on.
:param init_url: URL to download the init segment from.
:param filename: Original filename to save as.
:return: True if downloaded successfully, False otherwise.
"""
for attempt in range(1, _MAX_RETRIES + 1):
reason: str | None = None
exc_info = False
try:
async with ctx.http.get(init_url) as resp:
if resp.status != 200:
reason = f"HTTP {resp.status}"
else:
data = await resp.read()
init_path = cache.cache_dir / filename
await asyncio.to_thread(init_path.write_bytes, data)
cache.set_init_path(init_path)
ctx.logger.debug("Cached init segment (%d bytes).", len(data))
return True
except asyncio.CancelledError:
raise
except Exception as e:
reason = str(e)
exc_info = True
ctx.logger.warning(
f"Init segment fetch failed (attempt {attempt}/{_MAX_RETRIES}): "
f"{reason}. Retrying in {_RETRY_DELAY:.0f}s.",
exc_info=exc_info,
)
await asyncio.sleep(_RETRY_DELAY)
ctx.logger.error(f"Init segment failed after {_MAX_RETRIES} attempts.")
return False
async def _polling_loop(
ctx: ModuleContext,
master_url: str,
cache_dir: Path,
cache_duration: int,
min_clip_length: int,
) -> None:
"""Background task that polls the HLS variant playlist for new segments.
Constructs the :class:`ChunkCache` after the first successful playlist
fetch, since the segment identifier and extension are not known until then.
:param ctx: The module context.
:param master_url: URL to the master m3u8 playlist.
:param cache_dir: Directory for cached segment files.
:param cache_duration: Maximum cache window in seconds.
:param min_clip_length: Minimum clip duration in seconds.
"""
module_state = get_state(ctx)
try:
master_playlist = await _fetch_playlist(ctx, master_url, "Master playlist")
if master_playlist is None or not master_playlist.playlists:
ctx.logger.error("Could not fetch master playlist. Caching disabled.")
shutil.rmtree(cache_dir, ignore_errors=True)
module_state.cache_task = None
return
best = max(
master_playlist.playlists,
key=lambda p: p.stream_info.bandwidth or 0,
)
variant_url = urljoin(master_url, best.uri)
ctx.logger.debug(
f"Selected variant: {variant_url} "
f"(bandwidth={best.stream_info.bandwidth}, "
f"resolution={best.stream_info.resolution})"
)
variant_playlist = await _fetch_playlist(ctx, variant_url, "Variant playlist")
if variant_playlist is None or not variant_playlist.segments:
ctx.logger.error("Could not fetch variant playlist. Caching disabled.")
shutil.rmtree(cache_dir, ignore_errors=True)
module_state.cache_task = None
return
first_match = _SEGMENT_URI_RE.search(variant_playlist.segments[0].uri)
if first_match is None:
ctx.logger.error("Could not parse segment URI. Caching disabled.")
shutil.rmtree(cache_dir, ignore_errors=True)
module_state.cache_task = None
return
segment_id, extension = first_match.group(1), first_match.group(3)
cache = ChunkCache(
cache_dir=cache_dir,
cache_duration=cache_duration,
min_clip_length=min_clip_length,
segment_id=segment_id,
extension=extension,
)
module_state.cache = cache
poll_interval = (
float(variant_playlist.target_duration)
if variant_playlist.target_duration
else 2.0
)
cache.set_target_duration(poll_interval)
ctx.logger.debug("Initial poll interval set to %.1fs.", poll_interval)
if variant_playlist.segment_map:
init_uri = variant_playlist.segment_map[0].uri
if init_uri:
init_url = urljoin(variant_url, init_uri)
init_filename = init_uri.rsplit("/", 1)[-1]
if not await _download_init_segment(
ctx, cache, init_url, init_filename
):
ctx.logger.error(
"Could not download init segment. Caching disabled."
)
cache.cleanup()
module_state.cache = None
module_state.cache_task = None
return
self._sequences.add(seq)
self._logger.debug("Cached segment %d (%d bytes).", seq, len(data))
return True
async def _polling_loop(self) -> None:
"""Background task that polls the HLS variant playlist for new segments."""
pending_retries: set[int] = set()
while True:
try:
async with ctx.http.get(variant_url) as resp:
async with self._http.get(self._variant_url) as resp:
if resp.status != 200:
ctx.logger.warning(
f"Variant playlist returned HTTP {resp.status}."
self._logger.warning(
"Variant playlist returned HTTP %d.", resp.status
)
await asyncio.sleep(poll_interval)
await asyncio.sleep(self._target_duration)
continue
content = await resp.text()
@@ -485,48 +199,237 @@ async def _polling_loop(
if seq not in playlist_segments:
abandoned.add(seq)
continue
segment_url = urljoin(variant_url, playlist_segments[seq])
if await _download_segment(ctx, cache, segment_url, seq):
segment_url = urljoin(self._variant_url, playlist_segments[seq])
if await self._download_segment(segment_url, seq):
pending_retries.discard(seq)
ctx.logger.info(f"Segment {seq} recovered after retry.")
self._logger.info("Segment %d recovered after retry.", seq)
for seq in abandoned:
pending_retries.discard(seq)
ctx.logger.warning(
f"Segment {seq} no longer in playlist, giving up."
self._logger.warning(
"Segment %d no longer in playlist, giving up.", seq
)
for seq, uri in playlist_segments.items():
if cache.has_chunk(seq):
if seq in self._sequences:
continue
if seq in pending_retries:
continue
segment_url = urljoin(variant_url, uri)
if not await _download_segment(ctx, cache, segment_url, seq):
segment_url = urljoin(self._variant_url, uri)
if not await self._download_segment(segment_url, seq):
pending_retries.add(seq)
ctx.logger.warning(
f"Segment {seq} download failed, queued for retry."
self._logger.warning(
"Segment %d download failed, queued for retry.", seq
)
pre_prune_count = len(cache)
pruned = cache.prune()
if pruned:
ctx.logger.debug(
f"Pruned {pre_prune_count - len(cache)} segment(s). "
f"Cache: {len(cache)} chunks, "
f"{cache.total_duration:.1f}s total."
)
if self._prune_guard_count == 0:
pre_prune = len(self._sequences)
while (
self.buffered_duration > self._cache_duration
and len(self._sequences) > 1
):
oldest = min(self._sequences)
self._sequences.discard(oldest)
path = self._segment_path(oldest)
await asyncio.to_thread(path.unlink, True)
if len(self._sequences) < pre_prune:
self._logger.debug(
"Pruned %d segment(s). Cache: %d chunks, %.1fs total.",
pre_prune - len(self._sequences),
len(self._sequences),
self.buffered_duration,
)
except asyncio.CancelledError:
raise
except Exception:
ctx.logger.error("Error in HLS polling loop.", exc_info=True)
self._logger.error("Error in HLS polling loop.", exc_info=True)
await asyncio.sleep(poll_interval)
await asyncio.sleep(self._target_duration)
except asyncio.CancelledError:
if module_state.cache is None:
shutil.rmtree(cache_dir, ignore_errors=True)
async def start_caching(
http: aiohttp.ClientSession,
base_url: str,
cache_duration: int,
logger: logging.Logger,
) -> ChunkCache:
"""Start HLS caching and return the running ChunkCache.
Fetches the master and variant playlists, constructs the cache, downloads
the init segment if needed, and launches the background polling task.
:param http: Shared HTTP client session.
:param base_url: Base Owncast URL (e.g. ``http://localhost:8080``).
:param cache_duration: Maximum cache window in seconds.
:param logger: Logger instance.
:return: A running ChunkCache with its background polling task active.
:raises RuntimeError: If initial playlist fetching or setup fails.
"""
master_url = f"{base_url.rstrip('/')}/hls/stream.m3u8"
cache_dir = Path(tempfile.mkdtemp(prefix="owlbot-clips-"))
try:
master_playlist = await _fetch_playlist(
http, master_url, "Master playlist", logger
)
if master_playlist is None or not master_playlist.playlists:
msg = "Could not fetch master playlist."
raise RuntimeError(msg)
best = max(
master_playlist.playlists,
key=lambda p: p.stream_info.bandwidth or 0,
)
variant_url = urljoin(master_url, best.uri)
logger.debug(
"Selected variant: %s (bandwidth=%s, resolution=%s)",
variant_url,
best.stream_info.bandwidth,
best.stream_info.resolution,
)
variant_playlist = await _fetch_playlist(
http, variant_url, "Variant playlist", logger
)
if variant_playlist is None or not variant_playlist.segments:
msg = "Could not fetch variant playlist."
raise RuntimeError(msg)
first_match = _SEGMENT_URI_RE.search(variant_playlist.segments[0].uri)
if first_match is None:
msg = "Could not parse segment URI."
raise RuntimeError(msg)
segment_id, extension = first_match.group(1), first_match.group(3)
poll_interval = (
float(variant_playlist.target_duration)
if variant_playlist.target_duration
else 2.0
)
logger.debug("Initial poll interval set to %.1fs.", poll_interval)
init_path: Path | None = None
if variant_playlist.segment_map:
init_uri = variant_playlist.segment_map[0].uri
if init_uri:
init_url = urljoin(variant_url, init_uri)
init_filename = init_uri.rsplit("/", 1)[-1]
init_path = await _download_init_segment(
http, cache_dir, init_url, init_filename, logger
)
if init_path is None:
msg = "Could not download init segment."
raise RuntimeError(msg)
cache = ChunkCache(
http=http,
cache_dir=cache_dir,
cache_duration=cache_duration,
segment_id=segment_id,
extension=extension,
target_duration=poll_interval,
variant_url=variant_url,
logger=logger,
init_path=init_path,
)
cache.start()
logger.info("HLS caching started. Cache dir: %s", cache_dir)
return cache
except BaseException:
shutil.rmtree(cache_dir, ignore_errors=True)
raise
async def _fetch_playlist(
http: aiohttp.ClientSession,
url: str,
label: str,
logger: logging.Logger,
) -> m3u8.M3U8 | None:
"""Fetch and parse an HLS playlist with retry logic.
:param http: HTTP client session.
:param url: URL of the m3u8 playlist.
:param label: Human-readable label for log messages.
:param logger: Logger instance.
:return: Parsed playlist, or ``None`` if all attempts are exhausted.
"""
for attempt in range(1, _MAX_RETRIES + 1):
reason: str | None = None
exc_info = False
try:
async with http.get(url) as resp:
if resp.status != 200:
reason = f"HTTP {resp.status}"
else:
content = await resp.text()
return m3u8.loads(content)
except asyncio.CancelledError:
raise
except aiohttp.ClientError as e:
reason = str(e)
exc_info = True
logger.warning(
"%s fetch failed (attempt %d/%d): %s. Retrying in %.0fs.",
label,
attempt,
_MAX_RETRIES,
reason,
_RETRY_DELAY,
exc_info=exc_info,
)
await asyncio.sleep(_RETRY_DELAY)
logger.error("%s failed after %d attempts.", label, _MAX_RETRIES)
return None
async def _download_init_segment(
http: aiohttp.ClientSession,
cache_dir: Path,
init_url: str,
filename: str,
logger: logging.Logger,
) -> Path | None:
"""Download the fMP4 initialization segment with retry logic.
:param http: HTTP client session.
:param cache_dir: Directory to save the init segment into.
:param init_url: URL to download the init segment from.
:param filename: Original filename to save as.
:param logger: Logger instance.
:return: Path to the downloaded init segment, or None on failure.
"""
for attempt in range(1, _MAX_RETRIES + 1):
reason: str | None = None
exc_info = False
try:
async with http.get(init_url) as resp:
if resp.status != 200:
reason = f"HTTP {resp.status}"
else:
data = await resp.read()
init_path = cache_dir / filename
await asyncio.to_thread(init_path.write_bytes, data)
logger.debug("Cached init segment (%d bytes).", len(data))
return init_path
except asyncio.CancelledError:
raise
except aiohttp.ClientError as e:
reason = str(e)
exc_info = True
logger.warning(
"Init segment fetch failed (attempt %d/%d): %s. Retrying in %.0fs.",
attempt,
_MAX_RETRIES,
reason,
_RETRY_DELAY,
exc_info=exc_info,
)
await asyncio.sleep(_RETRY_DELAY)
logger.error("Init segment failed after %d attempts.", _MAX_RETRIES)
return None