diff --git a/docs/architecture/platform.md b/docs/architecture/platform.md index a0b03b29..dc74f81e 100644 --- a/docs/architecture/platform.md +++ b/docs/architecture/platform.md @@ -1090,9 +1090,11 @@ Initial curated tools: - `get_capability` - `list_media` - `get_media` +- `export_subtitles` - Streamable HTTP: `create_media_upload`, `get_media_upload` - Filesystem-accessible stdio: `ingest_local_media`, `get_media_ingestion` - `get_index_status` +- `plan_bulk_index` - `start_indexing` - `search_moments` - `query_video` diff --git a/docs/local-api.md b/docs/local-api.md index 737a3482..574904e6 100644 --- a/docs/local-api.md +++ b/docs/local-api.md @@ -70,6 +70,34 @@ do not share a filesystem. The [Premiere Pro extension preview](integrations/premiere-pro.md) uses this workflow for media already loaded in a Premiere project. +### Plan bulk indexing before submitting jobs + +A client can preview which registered media still need indexing without +submitting redundant durable jobs. Send `POST /api/v1/index/plan` with target +`modalities` and optional `media_ids`: + +```bash +curl -X POST http://127.0.0.1:32191/api/v1/index/plan \ + -H "Content-Type: application/json" \ + -d '{"modalities": ["scene"]}' +``` + +The returned plan categorizes media into pending and skipped targets, recording +the exact skip reason (such as already covered by the active snapshot with +matching content checksums). + +### Export transcribed subtitles + +A client can export speech transcripts and timed subtitles for speech-indexed +media items in `.srt`, `.vtt`, or JSON format: + +```bash +curl http://127.0.0.1:32191/api/v1/media/{media_id}/subtitles?format=srt +``` + +Pass `format=vtt` for WebVTT subtitles or `format=json` for machine-readable +cue timestamps and transcript text. + ## Connect a local AI assistant An assistant that can start a program on the same computer does not need the diff --git a/src/vidxp/api_routes/index.py b/src/vidxp/api_routes/index.py index 84ee004a..1f1c45e3 100644 --- a/src/vidxp/api_routes/index.py +++ b/src/vidxp/api_routes/index.py @@ -3,7 +3,11 @@ from fastapi import APIRouter, Depends from vidxp.api_routes.dependencies import context, read_principal -from vidxp.application_models import IndexStatus +from vidxp.application_models import ( + BulkIndexPlan, + IndexStatus, + PlanBulkIndexCommand, +) from vidxp.composition import HttpApplicationContext @@ -24,3 +28,20 @@ def get_index_status( service: Annotated[HttpApplicationContext, Depends(context)], ) -> IndexStatus: return service.application.index_status() + + +@router.post( + "/plan", + response_model=BulkIndexPlan, + operation_id="planBulkIndex", + summary="Plan bulk indexing", + description=( + "Decide which registered media still need indexing before submitting " + "durable indexing jobs." + ), +) +def plan_bulk_index( + command: PlanBulkIndexCommand, + service: Annotated[HttpApplicationContext, Depends(context)], +) -> BulkIndexPlan: + return service.application.plan_bulk_index(command) diff --git a/src/vidxp/api_routes/media.py b/src/vidxp/api_routes/media.py index 4ab31bfd..562ab645 100644 --- a/src/vidxp/api_routes/media.py +++ b/src/vidxp/api_routes/media.py @@ -16,12 +16,14 @@ ApplicationError, CreateUploadIntentCommand, ErrorCategory, + ExportSubtitlesCommand, ListMediaCommand, LocalMediaIngestionCommand, MediaAsset, MediaPage, MediaUploadSessionStatus, Principal, + SubtitleFormat, UploadIntent, UploadIntentId, UploadSessionId, @@ -326,3 +328,26 @@ def head_media_content( service: Annotated[HttpApplicationContext, Depends(context)], ) -> Response: return _content(media_id, request, service) + + +@router.get( + "/{media_id}/subtitles", + response_model=None, + operation_id="exportMediaSubtitles", + summary="Export transcribed subtitles for speech-indexed media", + dependencies=[Depends(read_principal)], +) +def export_media_subtitles( + media_id: MediaId, + service: Annotated[HttpApplicationContext, Depends(context)], + format: SubtitleFormat = SubtitleFormat.srt, +) -> Response: + result = service.application.export_subtitles( + ExportSubtitlesCommand(media_id=media_id, format=format) + ) + if format == SubtitleFormat.json: + return Response(content=result.content, media_type="application/json") + if format == SubtitleFormat.vtt: + return Response(content=result.content, media_type="text/vtt") + return Response(content=result.content, media_type="text/plain; charset=utf-8") + diff --git a/src/vidxp/application.py b/src/vidxp/application.py index 71b40d2d..5dede66f 100644 --- a/src/vidxp/application.py +++ b/src/vidxp/application.py @@ -2,6 +2,7 @@ from dataclasses import replace +import json from contextlib import contextmanager from pathlib import Path from shutil import which @@ -24,6 +25,7 @@ DependencyKind, DependencyUnavailableError, ErrorCategory, + ExportSubtitlesCommand, FusedSearchResult, IndexResult, ImportMediaCommand, @@ -43,7 +45,10 @@ EvidenceBoardJobRequest, EvidenceBoardResult, EvidenceDeliveryPolicy, + SubtitleExportResult, + SubtitleFormat, ) +from vidxp.subtitles import format_srt, format_vtt, records_to_cues from vidxp.capabilities.actor.schemas import ( ActorClusterSummary, ActorClustersOutput, @@ -945,6 +950,46 @@ def remove_from_index(self, command: RemoveIndexCommand) -> bool: command.media_id, ) + @application_boundary + def export_subtitles( + self, + command: ExportSubtitlesCommand, + ) -> SubtitleExportResult: + self.get_media(command.media_id) + config = self._active_config() + self._require_indexed_capability("speech", config) + snapshot = self._read_active_snapshot() + if snapshot is not None: + if command.media_id not in snapshot.generations: + raise ApplicationError( + "media_not_indexed", + ErrorCategory.validation, + f"Media '{command.media_id}' is not present in the active index snapshot.", + details={"media_id": command.media_id}, + ) + if "speech" not in snapshot.generations[command.media_id].modalities: + raise ApplicationError( + "speech_not_indexed", + ErrorCategory.validation, + f"Media '{command.media_id}' was indexed without speech transcription.", + details={"media_id": command.media_id}, + ) + with self.index_backend.open_store(config) as storage: + raw_records = storage.records("speech", video_id=command.media_id) + cues = records_to_cues(raw_records) + if command.format == SubtitleFormat.vtt: + content = format_vtt(cues) + elif command.format == SubtitleFormat.json: + content = json.dumps([cue.model_dump(mode="json") for cue in cues], indent=2) + "\n" + else: + content = format_srt(cues) + return SubtitleExportResult( + media_id=command.media_id, + format=command.format, + content=content, + cues=tuple(cues), + ) + def _base_config(self) -> IndexConfig: return IndexConfig.local( storage_directory=self.index_directory, diff --git a/src/vidxp/application_models.py b/src/vidxp/application_models.py index 257edcf8..fadd876c 100644 --- a/src/vidxp/application_models.py +++ b/src/vidxp/application_models.py @@ -729,6 +729,36 @@ class RemoveIndexCommand(ApplicationModel): media_id: MediaId +class SubtitleFormat(StrEnum): + srt = "srt" + vtt = "vtt" + json = "json" + + +class SubtitleCue(ApplicationModel): + start: float = Field(ge=0, description="Cue start time in seconds.") + end: float = Field(ge=0, description="Cue end time in seconds.") + text: str = Field(min_length=1, description="Transcribed speech text.") + + +class ExportSubtitlesCommand(ApplicationModel): + media_id: MediaId = Field(description="Cataloged media identifier.") + format: SubtitleFormat = Field( + default=SubtitleFormat.srt, + description="Subtitle output format.", + ) + + +class SubtitleExportResult(ApplicationModel): + media_id: MediaId + format: SubtitleFormat + content: str = Field(description="Formatted subtitle text.") + cues: tuple[SubtitleCue, ...] = Field( + default=(), + description="Structured cues with timestamps.", + ) + + class Artifact(ApplicationModel): schema_version: Literal[ARTIFACT_SCHEMA_VERSION] = ARTIFACT_SCHEMA_VERSION artifact_id: ArtifactId diff --git a/src/vidxp/cli_commands/media.py b/src/vidxp/cli_commands/media.py index 3905b4be..332306a8 100644 --- a/src/vidxp/cli_commands/media.py +++ b/src/vidxp/cli_commands/media.py @@ -7,7 +7,12 @@ from rich.console import Console from rich.table import Table -from vidxp.application_models import ImportMediaCommand, ListMediaCommand +from vidxp.application_models import ( + ExportSubtitlesCommand, + ImportMediaCommand, + ListMediaCommand, + SubtitleFormat, +) from vidxp.cli_support import ( OutputFormat, effective_output_format, @@ -141,3 +146,57 @@ def show_media( emit_json(payload) else: Console().print_json(data=payload) + + +@app.command("subtitles") +def export_subtitles( + ctx: typer.Context, + media_id: Annotated[ + str, + typer.Argument( + help="Stable media identifier returned by import or list." + ), + ], + format: Annotated[ + SubtitleFormat, + typer.Option( + "--format", + "-f", + help="Subtitle format: srt, vtt, or json.", + case_sensitive=False, + ), + ] = SubtitleFormat.srt, + output: Annotated[ + Path | None, + typer.Option( + "--output", + "-o", + help="Optional file path to write the subtitles to.", + ), + ] = None, + json_output: Annotated[ + bool, + typer.Option("--json", help="Emit machine-readable JSON."), + ] = False, +) -> None: + """Export transcribed speech subtitles (.srt, .vtt, or JSON) for indexed media.""" + + state = state_from_context(ctx) + result = state.service.export_subtitles( + ExportSubtitlesCommand(media_id=media_id, format=format) + ) + if effective_output_format(state, json_output) == OutputFormat.json: + emit_json(result.model_dump(mode="json")) + return + + if output is not None: + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(result.content, encoding="utf-8") + if not state.quiet: + typer.secho( + f"Exported {format.value.upper()} subtitles to {output}", + fg=typer.colors.GREEN, + ) + else: + typer.echo(result.content.rstrip("\n")) + diff --git a/src/vidxp/control_plane.py b/src/vidxp/control_plane.py index bc86329c..3161dbfd 100644 --- a/src/vidxp/control_plane.py +++ b/src/vidxp/control_plane.py @@ -18,6 +18,7 @@ ComponentReadiness, CreateIndexCommand, DependencyCheckResult, + ExportSubtitlesCommand, IndexStatus, Identifier, InvalidRequestError, @@ -27,6 +28,7 @@ ModelUnavailableError, ResourceNotFoundError, RuntimeReadiness, + SubtitleExportResult, WorkspaceCapability, WorkspaceMedia, WorkspaceMediaCapability, @@ -474,3 +476,10 @@ def runtime_readiness(self) -> RuntimeReadiness: components=components, dependencies=models, ) + + def export_subtitles( + self, + command: ExportSubtitlesCommand, + ) -> SubtitleExportResult: + raise NotImplementedError + diff --git a/src/vidxp/mcp.py b/src/vidxp/mcp.py index c359c335..f31320fd 100644 --- a/src/vidxp/mcp.py +++ b/src/vidxp/mcp.py @@ -37,6 +37,7 @@ Artifact, ArtifactDeliveryMode, ArtifactDownload, + BulkIndexPlan, CapabilityInfo, CapabilityList, CreateSnippetCommand, @@ -48,6 +49,7 @@ EvidenceDeliveryPolicy, EvidenceDeliveryResult, EvidenceBoardResult, + ExportSubtitlesCommand, DEFAULT_JOB_WAIT_SECONDS, FusedSearchResult, Identifier, @@ -69,6 +71,7 @@ MediaPage, MediaUploadSessionStatus, Principal, + PlanBulkIndexCommand, PrepareModelsCommand, QueryVideoCommand, QueryAnswer, @@ -76,6 +79,8 @@ SearchCommand, Sha256, SnippetProfile, + SubtitleExportResult, + SubtitleFormat, WorkspaceOverview, UploadSessionId, ) @@ -1397,6 +1402,29 @@ async def get_media(media_id: MediaId) -> MediaAsset: operation=lambda _actor: context.application.get_media(media_id), ) + @server.tool( + title="Export subtitles", + description=( + "Export speech transcript and timed subtitles (SRT, WebVTT, or JSON) " + "for an indexed media item." + ), + annotations=_READ_ONLY, + structured_output=True, + ) + async def export_subtitles( + media_id: MediaId, + format: SubtitleFormat = SubtitleFormat.srt, + ) -> SubtitleExportResult: + return await _invoke_async( + context, + default_principal=default_principal, + permission=RepositoryPermission.read, + operation=lambda _actor: context.application.export_subtitles( + ExportSubtitlesCommand(media_id=media_id, format=format) + ), + ) + + @server.tool( title="Create media upload", description=( @@ -1568,6 +1596,27 @@ async def get_index_status() -> IndexStatus: operation=lambda _actor: context.application.index_status(), ) + @server.tool( + title="Plan bulk indexing", + description=( + "Decide which registered media need indexing before submitting " + "indexing jobs. Pass media_ids to plan for specific videos, or " + "omit to plan across all registered media. Media already covered " + "by the active index with matching content and options is skipped." + ), + annotations=_READ_ONLY, + structured_output=True, + ) + async def plan_bulk_index( + command: PlanBulkIndexCommand, + ) -> BulkIndexPlan: + return await _invoke_async( + context, + default_principal=default_principal, + permission=RepositoryPermission.read, + operation=lambda _actor: context.application.plan_bulk_index(command), + ) + @server.tool( title="Start indexing", description=( diff --git a/src/vidxp/subtitles.py b/src/vidxp/subtitles.py new file mode 100644 index 00000000..81e45bf4 --- /dev/null +++ b/src/vidxp/subtitles.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from typing import Any, Iterable, Mapping, Sequence + +from vidxp.application_models import SubtitleCue + + +def format_srt_timestamp(seconds: float) -> str: + """Format seconds into SubRip timestamp format: HH:MM:SS,mmm.""" + total_ms = max(0, int(round(seconds * 1000))) + hours, remainder = divmod(total_ms, 3600_000) + minutes, remainder = divmod(remainder, 60_000) + secs, millis = divmod(remainder, 1000) + return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}" + + +def format_vtt_timestamp(seconds: float) -> str: + """Format seconds into WebVTT timestamp format: HH:MM:SS.mmm.""" + total_ms = max(0, int(round(seconds * 1000))) + hours, remainder = divmod(total_ms, 3600_000) + minutes, remainder = divmod(remainder, 60_000) + secs, millis = divmod(remainder, 1000) + return f"{hours:02d}:{minutes:02d}:{secs:02d}.{millis:03d}" + + +def format_srt(cues: Sequence[SubtitleCue]) -> str: + """Render a sequence of subtitle cues into SubRip (.srt) format.""" + if not cues: + return "" + blocks: list[str] = [] + for index, cue in enumerate(cues, start=1): + start = format_srt_timestamp(cue.start) + end = format_srt_timestamp(cue.end) + text = cue.text.strip() + blocks.append(f"{index}\n{start} --> {end}\n{text}") + return "\n\n".join(blocks) + "\n" + + +def format_vtt(cues: Sequence[SubtitleCue]) -> str: + """Render a sequence of subtitle cues into WebVTT (.vtt) format.""" + if not cues: + return "WEBVTT\n" + blocks: list[str] = ["WEBVTT"] + for index, cue in enumerate(cues, start=1): + start = format_vtt_timestamp(cue.start) + end = format_vtt_timestamp(cue.end) + text = cue.text.strip() + blocks.append(f"{index}\n{start} --> {end}\n{text}") + return "\n\n".join(blocks) + "\n" + + +def records_to_cues(records: Iterable[Mapping[str, Any]]) -> list[SubtitleCue]: + """Convert index storage records into sorted SubtitleCue items.""" + cues: list[SubtitleCue] = [] + for rec in records: + meta = rec.get("metadata") or {} + start_val = meta.get("start") + end_val = meta.get("end") + if start_val is None or end_val is None: + continue + start = float(start_val) + end = float(end_val) + text = str(meta.get("text", rec.get("document", ""))).strip() + if text and end >= start: + cues.append(SubtitleCue(start=start, end=end, text=text)) + cues.sort(key=lambda cue: (cue.start, cue.end)) + return cues diff --git a/tests/test_api.py b/tests/test_api.py index a5794f8f..3e73048b 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -19,6 +19,9 @@ from vidxp.application_models import ( ApplicationError, Artifact, + BulkIndexPlan, + BulkIndexTarget, + BulkIndexTargetState, ErrorCategory, ErrorDetail, EvidenceArtifact, @@ -37,13 +40,18 @@ JobQueue, JobState, JobWaitResult, + IndexOptions, IndexStatus, MediaAsset, MediaPage, MediaUploadSessionStatus, + PlanBulkIndexCommand, Principal, SearchCommand, SearchJobResult, + SubtitleCue, + SubtitleExportResult, + SubtitleFormat, QueryVideoCommand, UploadIntent, WorkspaceOverview, @@ -460,6 +468,43 @@ def test_workspace_passes_filters_to_application(self): self.assertEqual(command.filename, "batch") self.assertEqual(command.state, MediaState.pending) + def test_plan_bulk_index_delegates_to_application(self): + with TemporaryDirectory() as directory: + context = self.context(Path(directory)) + expected_plan = BulkIndexPlan( + targets=( + BulkIndexTarget( + media_id=MEDIA_ID, + original_filename="clip.mp4", + state=BulkIndexTargetState.pending, + ), + ), + modalities=("scene",), + options=IndexOptions( + modalities=("scene",), + frame_stride=1, + capability_options={}, + ), + ) + context.application.plan_bulk_index.return_value = expected_plan + with TestClient(create_app(context=context)) as client: + response = client.post( + "/api/v1/index/plan", + json={ + "media_ids": [MEDIA_ID], + "modalities": ["scene"], + }, + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.json(), + expected_plan.model_dump(mode="json"), + ) + command = context.application.plan_bulk_index.call_args.args[0] + self.assertEqual(command.media_ids, (MEDIA_ID,)) + self.assertEqual(command.modalities, ("scene",)) + def test_local_ingestion_api_delegates_to_durable_batch_workflow(self): with TemporaryDirectory() as directory: context = self.context(Path(directory), remote_uploads=True) @@ -641,6 +686,26 @@ def test_unexpected_errors_are_logged_and_masked(self): self.assertEqual(payload["error"]["code"], "internal_error") self.assertNotIn("secret-path", response.text) + def test_export_subtitles_delegates_to_application(self): + with TemporaryDirectory() as directory: + context = self.context(Path(directory)) + context.application.export_subtitles.return_value = ( + SubtitleExportResult( + media_id=MEDIA_ID, + format=SubtitleFormat.srt, + content="1\n00:00:01,000 --> 00:00:02,000\nHello world\n", + cues=(SubtitleCue(start=1.0, end=2.0, text="Hello world"),), + ) + ) + with TestClient(create_app(context=context)) as client: + response = client.get(f"/api/v1/media/{MEDIA_ID}/subtitles?format=srt") + + self.assertEqual(response.status_code, 200) + self.assertIn("00:00:01,000 --> 00:00:02,000\nHello world", response.text) + command = context.application.export_subtitles.call_args.args[0] + self.assertEqual(command.media_id, MEDIA_ID) + self.assertEqual(command.format, SubtitleFormat.srt) + def test_job_submission_is_thin_idempotent_delegation(self): with TemporaryDirectory() as directory: context = self.context(Path(directory)) diff --git a/tests/test_application.py b/tests/test_application.py index d60b3fe8..70d79982 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -19,6 +19,7 @@ DependencyUnavailableError, EvidenceBoardResult, EvidenceDeliveryMode, + ExportSubtitlesCommand, FusedSearchResult, IndexResult, IndexSnapshotReference, @@ -28,8 +29,10 @@ QueryAnswerMode, QueryVideoCommand, RemoveIndexCommand, + ResourceNotFoundError, SearchCommand, SearchHit, + SubtitleFormat, InitialEvidenceDeliveryPolicy, ) from vidxp.core.media import MediaUnavailableError @@ -1532,6 +1535,100 @@ def test_local_backend_injects_storage_for_generation_build(self): cleanup_storage.__exit__.assert_called_once() storage.__exit__.assert_called_once() + def test_export_subtitles_media_not_found(self): + with TemporaryDirectory() as directory: + app, backend = self.application(directory) + app.media.get.side_effect = ResourceNotFoundError( + f"media '{MEDIA_ID}'" + ) + with self.assertRaises(ResourceNotFoundError): + app.export_subtitles( + ExportSubtitlesCommand(media_id=MEDIA_ID) + ) + + def test_export_subtitles_media_not_indexed(self): + with TemporaryDirectory() as directory: + app, backend = self.application(directory) + app.media.get.return_value = Mock(media_id=MEDIA_ID) + mock_config = Mock(enabled_modalities=("speech", "scene")) + app._active_config = Mock(return_value=mock_config) + mock_snapshot = Mock(generations={}) + app._read_active_snapshot = Mock(return_value=mock_snapshot) + with self.assertRaises(ApplicationError) as ctx: + app.export_subtitles( + ExportSubtitlesCommand(media_id=MEDIA_ID) + ) + self.assertEqual(ctx.exception.code, "media_not_indexed") + + def test_export_subtitles_speech_not_indexed(self): + with TemporaryDirectory() as directory: + app, backend = self.application(directory) + app.media.get.return_value = Mock(media_id=MEDIA_ID) + mock_config = Mock(enabled_modalities=("speech", "scene")) + app._active_config = Mock(return_value=mock_config) + gen = Mock(modalities=("scene",)) + mock_snapshot = Mock(generations={MEDIA_ID: gen}) + app._read_active_snapshot = Mock(return_value=mock_snapshot) + with self.assertRaises(ApplicationError) as ctx: + app.export_subtitles( + ExportSubtitlesCommand(media_id=MEDIA_ID) + ) + self.assertEqual(ctx.exception.code, "speech_not_indexed") + + def test_export_subtitles_success(self): + with TemporaryDirectory() as directory: + app, backend = self.application(directory) + app.media.get.return_value = Mock(media_id=MEDIA_ID) + mock_config = Mock(enabled_modalities=("speech", "scene")) + app._active_config = Mock(return_value=mock_config) + gen = Mock(modalities=("speech", "scene")) + mock_snapshot = Mock(generations={MEDIA_ID: gen}) + app._read_active_snapshot = Mock(return_value=mock_snapshot) + + mock_storage = MagicMock() + mock_storage.records.return_value = [ + { + "metadata": {"start": 0.5, "end": 2.5, "text": "Hello world"}, + }, + { + "metadata": {"start": 3.0, "end": 5.0, "text": "VidXP subtitles"}, + }, + ] + manager = MagicMock() + manager.__enter__ = Mock(return_value=mock_storage) + manager.__exit__ = Mock(return_value=False) + backend.open_store.return_value = manager + + # SRT format + srt_res = app.export_subtitles( + ExportSubtitlesCommand(media_id=MEDIA_ID, format=SubtitleFormat.srt) + ) + self.assertEqual(srt_res.media_id, MEDIA_ID) + self.assertEqual(srt_res.format, SubtitleFormat.srt) + self.assertEqual(len(srt_res.cues), 2) + self.assertIn("00:00:00,500 --> 00:00:02,500\nHello world", srt_res.content) + self.assertIn("00:00:03,000 --> 00:00:05,000\nVidXP subtitles", srt_res.content) + + # VTT format + vtt_res = app.export_subtitles( + ExportSubtitlesCommand(media_id=MEDIA_ID, format=SubtitleFormat.vtt) + ) + self.assertEqual(vtt_res.format, SubtitleFormat.vtt) + self.assertTrue(vtt_res.content.startswith("WEBVTT\n\n")) + self.assertIn("00:00:00.500 --> 00:00:02.500\nHello world", vtt_res.content) + + # JSON format + json_res = app.export_subtitles( + ExportSubtitlesCommand(media_id=MEDIA_ID, format=SubtitleFormat.json) + ) + self.assertEqual(json_res.format, SubtitleFormat.json) + parsed = json.loads(json_res.content) + self.assertEqual(len(parsed), 2) + self.assertEqual(parsed[0]["text"], "Hello world") + self.assertEqual(parsed[0]["start"], 0.5) + if __name__ == "__main__": unittest.main() + + diff --git a/tests/test_cli.py b/tests/test_cli.py index afdebf86..b4f37100 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -52,6 +52,9 @@ SearchCommand, SearchJobResult, SearchMomentsPlanStep, + SubtitleCue, + SubtitleExportResult, + SubtitleFormat, ) from vidxp.core.artifacts import ArtifactKind, ArtifactState from vidxp.core.media import MediaState, MediaStream @@ -484,6 +487,41 @@ def test_media_list_passes_filters_to_service(self): self.assertEqual(command.filename, "clip.mp4") self.assertEqual(command.state, MediaState.ready) + def test_media_subtitles_prints_content(self): + self.service.export_subtitles.return_value = SubtitleExportResult( + media_id=MEDIA_ID, + format=SubtitleFormat.srt, + content="1\n00:00:01,000 --> 00:00:02,000\nHello world\n", + cues=(SubtitleCue(start=1.0, end=2.0, text="Hello world"),), + ) + result = self.invoke(["media", "subtitles", MEDIA_ID]) + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("00:00:01,000 --> 00:00:02,000\nHello world", result.output) + command = self.service.export_subtitles.call_args.args[0] + self.assertEqual(command.media_id, MEDIA_ID) + self.assertEqual(command.format, SubtitleFormat.srt) + + def test_media_subtitles_writes_file(self): + with TemporaryDirectory() as directory: + out_file = Path(directory) / "subs.vtt" + self.service.export_subtitles.return_value = SubtitleExportResult( + media_id=MEDIA_ID, + format=SubtitleFormat.vtt, + content="WEBVTT\n\n1\n00:00:01.000 --> 00:00:02.000\nHello world\n", + cues=(SubtitleCue(start=1.0, end=2.0, text="Hello world"),), + ) + result = self.invoke( + ["media", "subtitles", MEDIA_ID, "--format", "vtt", "--output", str(out_file)] + ) + self.assertEqual(result.exit_code, 0, result.output) + self.assertTrue(out_file.exists()) + self.assertEqual( + out_file.read_text(encoding="utf-8"), + "WEBVTT\n\n1\n00:00:01.000 --> 00:00:02.000\nHello world\n", + ) + command = self.service.export_subtitles.call_args.args[0] + self.assertEqual(command.format, SubtitleFormat.vtt) + def test_ui_share_uses_streamlit_wildcard_bind_and_warns(self): with ( patch( diff --git a/tests/test_mcp.py b/tests/test_mcp.py index b05268f4..00fe050d 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -29,6 +29,9 @@ from vidxp.application_models import ( ApplicationError, Artifact, + BulkIndexPlan, + BulkIndexTarget, + BulkIndexTargetState, EvidenceArtifact, EvidenceBoardCandidate, EvidenceBoardJobRequest, @@ -49,8 +52,10 @@ ErrorDetail, EvidenceDeliveryMode, EvidenceDeliveryPolicy, + IndexOptions, IndexStatus, InitialEvidenceDeliveryPolicy, + PlanBulkIndexCommand, Job, JobKind, JobPage, @@ -70,6 +75,9 @@ QueryJobResult, QueryPlan, SearchMomentsPlanStep, + SubtitleCue, + SubtitleExportResult, + SubtitleFormat, WorkspaceOverview, ) from vidxp.mcp_app import MCP_APP_MIME_TYPE, MCP_APP_RESOURCE_URI @@ -112,9 +120,11 @@ "get_runtime_readiness", "list_media", "get_media", + "export_subtitles", "create_media_upload", "get_media_upload", "get_index_status", + "plan_bulk_index", "start_indexing", "prepare_models", "search_moments", @@ -133,10 +143,10 @@ "cancel_job", ] STDIO_MCP_TOOL_NAMES = [ - *MCP_TOOL_NAMES[:6], + *MCP_TOOL_NAMES[:7], "ingest_local_media", "get_media_ingestion", - *MCP_TOOL_NAMES[8:], + *MCP_TOOL_NAMES[9:], ] @@ -1176,8 +1186,9 @@ def test_stdio_check_performs_handshake_and_tool_probe(self): rendered = output.getvalue() self.assertIn("OK VidXP MCP", rendered) self.assertIn("Index state: missing", rendered) - self.assertIn("Tools: 25", rendered) + self.assertIn("Tools: 27", rendered) self.assertIn("get_index_status", rendered) + self.assertIn("plan_bulk_index", rendered) async def test_server_info_exposes_vidxp_branding(self): with TemporaryDirectory() as directory: @@ -1304,6 +1315,87 @@ async def test_missing_models_fail_before_index_submission(self): ) context.jobs.submit_index.assert_called_once() + async def test_plan_bulk_index_delegates_to_application(self): + with TemporaryDirectory() as directory: + context = self.context(Path(directory)) + expected_plan = BulkIndexPlan( + targets=( + BulkIndexTarget( + media_id=MEDIA_ID, + original_filename="clip.mp4", + state=BulkIndexTargetState.pending, + ), + ), + modalities=("scene",), + options=IndexOptions( + modalities=("scene",), + frame_stride=1, + capability_options={}, + ), + ) + context.application.plan_bulk_index.return_value = expected_plan + server = create_mcp_server( + context, + default_principal=Principal( + subject="agent", + scopes=frozenset({"vidxp.read"}), + ), + ) + async with Client(server) as client: + result = await client.call_tool( + "plan_bulk_index", + { + "command": { + "media_ids": [MEDIA_ID], + "modalities": ["scene"], + }, + }, + ) + + self.assertFalse(result.is_error) + self.assertEqual( + result.structured_content, + expected_plan.model_dump(mode="json"), + ) + command = context.application.plan_bulk_index.call_args.args[0] + self.assertEqual(command.media_ids, (MEDIA_ID,)) + self.assertEqual(command.modalities, ("scene",)) + + async def test_export_subtitles_delegates_to_application(self): + with TemporaryDirectory() as directory: + context = self.context(Path(directory)) + expected = SubtitleExportResult( + media_id=MEDIA_ID, + format=SubtitleFormat.srt, + content="1\n00:00:01,000 --> 00:00:02,000\nHello world\n", + cues=(SubtitleCue(start=1.0, end=2.0, text="Hello world"),), + ) + context.application.export_subtitles.return_value = expected + server = create_mcp_server( + context, + default_principal=Principal( + subject="agent", + scopes=frozenset({"vidxp.read"}), + ), + ) + async with Client(server) as client: + result = await client.call_tool( + "export_subtitles", + { + "media_id": MEDIA_ID, + "format": "srt", + }, + ) + + self.assertFalse(result.is_error) + self.assertEqual( + result.structured_content, + expected.model_dump(mode="json"), + ) + command = context.application.export_subtitles.call_args.args[0] + self.assertEqual(command.media_id, MEDIA_ID) + self.assertEqual(command.format, SubtitleFormat.srt) + async def test_query_video_submits_the_shared_durable_command(self): with TemporaryDirectory() as directory: context = self.context(Path(directory)) @@ -2805,7 +2897,7 @@ async def test_streamable_http_works_with_the_official_remote_client(self): server.should_exit = True await serving - self.assertEqual(len(discovered.tools), 23) + self.assertEqual(len(discovered.tools), 25) self.assertNotIn( "create_media_upload", {tool.name for tool in discovered.tools}, diff --git a/tests/test_subtitles.py b/tests/test_subtitles.py new file mode 100644 index 00000000..d296d7d9 --- /dev/null +++ b/tests/test_subtitles.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import json +import unittest + +from vidxp.application_models import SubtitleCue +from vidxp.subtitles import ( + format_srt, + format_srt_timestamp, + format_vtt, + format_vtt_timestamp, + records_to_cues, +) + + +class SubtitlesTests(unittest.TestCase): + def test_format_srt_timestamp(self) -> None: + self.assertEqual(format_srt_timestamp(0.0), "00:00:00,000") + self.assertEqual(format_srt_timestamp(1.234), "00:00:01,234") + self.assertEqual(format_srt_timestamp(65.5), "00:01:05,500") + self.assertEqual(format_srt_timestamp(3661.042), "01:01:01,042") + self.assertEqual(format_srt_timestamp(-5.0), "00:00:00,000") + + def test_format_vtt_timestamp(self) -> None: + self.assertEqual(format_vtt_timestamp(0.0), "00:00:00.000") + self.assertEqual(format_vtt_timestamp(1.234), "00:00:01.234") + self.assertEqual(format_vtt_timestamp(65.5), "00:01:05.500") + self.assertEqual(format_vtt_timestamp(3661.042), "01:01:01.042") + self.assertEqual(format_vtt_timestamp(-5.0), "00:00:00.000") + + def test_format_srt_empty(self) -> None: + self.assertEqual(format_srt([]), "") + + def test_format_srt_cues(self) -> None: + cues = [ + SubtitleCue(start=1.0, end=3.5, text="Hello world!"), + SubtitleCue(start=4.0, end=6.2, text="This is VidXP subtitle export."), + ] + result = format_srt(cues) + expected = ( + "1\n" + "00:00:01,000 --> 00:00:03,500\n" + "Hello world!\n\n" + "2\n" + "00:00:04,000 --> 00:00:06,200\n" + "This is VidXP subtitle export.\n" + ) + self.assertEqual(result, expected) + + def test_format_vtt_empty(self) -> None: + self.assertEqual(format_vtt([]), "WEBVTT\n") + + def test_format_vtt_cues(self) -> None: + cues = [ + SubtitleCue(start=0.5, end=2.0, text="Welcome to VidXP."), + ] + result = format_vtt(cues) + expected = ( + "WEBVTT\n\n" + "1\n" + "00:00:00.500 --> 00:00:02.000\n" + "Welcome to VidXP.\n" + ) + self.assertEqual(result, expected) + + def test_records_to_cues(self) -> None: + records = [ + { + "metadata": {"start": 5.0, "end": 7.0, "text": "Second phrase"}, + }, + { + "metadata": {"start": 1.0, "end": 3.0, "text": "First phrase"}, + }, + { + # Missing metadata start/end should be skipped + "metadata": {"text": "Invalid phrase"}, + }, + { + # Empty text should be skipped + "metadata": {"start": 8.0, "end": 9.0, "text": " "}, + }, + { + # end < start should be skipped + "metadata": {"start": 10.0, "end": 9.0, "text": "Backwards"}, + }, + { + # Document fallback + "metadata": {"start": 12.0, "end": 14.0}, + "document": "Document phrase", + }, + ] + cues = records_to_cues(records) + self.assertEqual(len(cues), 3) + self.assertEqual(cues[0].text, "First phrase") + self.assertEqual(cues[0].start, 1.0) + self.assertEqual(cues[0].end, 3.0) + self.assertEqual(cues[1].text, "Second phrase") + self.assertEqual(cues[1].start, 5.0) + self.assertEqual(cues[2].text, "Document phrase") + self.assertEqual(cues[2].start, 12.0) + + +if __name__ == "__main__": + unittest.main()