scripts/_mediaskills_common.py
"""Shared helpers for mediaskills portable scripts (vendored into each skill)."""
from __future__ import annotations
import argparse
import json
import os
import random
import re
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from shutil import which
from typing import Any
EXIT_OK = 0
EXIT_BAD_ARGS = 1
EXIT_MISSING_DEP = 2
EXIT_PROCESSING = 3
def emit_success(op: str, data: dict[str, Any], outputs: list[str] | None = None) -> None:
print(json.dumps({"ok": True, "op": op, "data": data, "output_paths": outputs or []}))
def emit_error(op: str, err: str, *, code: int = EXIT_PROCESSING) -> None:
print(json.dumps({"ok": False, "op": op, "error": err}), file=sys.stderr)
sys.exit(code)
def emit_progress(stage: str, pct: float) -> None:
print(json.dumps({"progress": pct, "stage": stage}), file=sys.stderr)
def require_cmd(cmd: str, op: str) -> None:
if which(cmd) is None:
emit_error(
op,
f"{cmd} not found on PATH. Install via install-media-tools skill "
"(scripts/install.sh) or your system package manager.",
code=EXIT_MISSING_DEP,
)
def run(cmd: list[str], op: str) -> subprocess.CompletedProcess[str]:
try:
return subprocess.run(cmd, check=True, capture_output=True, text=True)
except subprocess.CalledProcessError as e:
err = (e.stderr or e.stdout or str(e))[:800]
emit_error(op, f"Command failed: {err}", code=EXIT_PROCESSING)
def run_bytes(cmd: list[str], op: str) -> subprocess.CompletedProcess[bytes]:
try:
return subprocess.run(cmd, check=True, capture_output=True)
except subprocess.CalledProcessError as e:
err = (e.stderr or b"").decode("utf-8", errors="replace")[:800]
emit_error(op, f"Command failed: {err}", code=EXIT_PROCESSING)
def workspace_root() -> Path:
"""Directory containing `.agents/skills` (repo / workspace root)."""
here = Path(__file__).resolve().parent
for parent in (here, *here.parents):
if (parent / ".agents" / "skills").is_dir():
return parent
return Path.cwd()
def mediaskills_dir() -> Path:
data_dir = os.environ.get("MEDIASKILLS_DATA_DIR")
if data_dir and data_dir.startswith("/"):
return Path(data_dir)
return workspace_root() / ".mediaskills"
def generated_dir() -> Path:
out = mediaskills_dir() / "generated"
out.mkdir(parents=True, exist_ok=True)
return out
def resolve_output(input_path: str | None, suffix: str, explicit: str | None = None) -> Path:
if explicit:
return Path(explicit)
stem = Path(input_path).stem if input_path else "output"
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
token = f"{ts}_{random.randint(1000, 9999)}"
if suffix.startswith("."):
name = f"{stem}_{token}{suffix}"
else:
label = suffix[1:] if suffix.startswith("_") else suffix
if "." in label:
base, ext = label.rsplit(".", 1)
name = f"{stem}_{base}_{token}.{ext}"
else:
name = f"{stem}_{label}_{token}"
return generated_dir() / name
def is_truthy(value: Any) -> bool:
if value is True:
return True
if isinstance(value, str):
return value.strip().lower() in {"1", "true", "yes", "on"}
return False
def ffprobe_json(path: str, op: str) -> dict[str, Any]:
require_cmd("ffprobe", op)
result = run(
[
"ffprobe",
"-v",
"quiet",
"-print_format",
"json",
"-show_format",
"-show_streams",
path,
],
op,
)
return json.loads(result.stdout)
def probe_duration(path: str) -> float:
out = subprocess.check_output(
[
"ffprobe",
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
path,
],
text=True,
).strip()
return float(out)
def summarize_probe(data: dict[str, Any]) -> dict[str, Any]:
fmt = data.get("format", {})
streams = data.get("streams", [])
video = next((s for s in streams if s.get("codec_type") == "video"), None)
audio = [s for s in streams if s.get("codec_type") == "audio"]
return {
"format": fmt.get("format_name"),
"duration": fmt.get("duration"),
"size": fmt.get("size"),
"video": (
{
"codec": video.get("codec_name"),
"width": video.get("width"),
"height": video.get("height"),
}
if video
else None
),
"audio_codecs": [s.get("codec_name") for s in audio],
}
def compare_probe_summaries(a: dict[str, Any], b: dict[str, Any]) -> dict[str, Any]:
def summary(d: dict[str, Any]) -> dict[str, Any]:
fmt = d.get("format", {})
video = next((s for s in d.get("streams", []) if s.get("codec_type") == "video"), None)
return {
"duration": fmt.get("duration"),
"size": fmt.get("size"),
"video": f"{video.get('width')}x{video.get('height')}" if video else None,
}
return {"a": summary(a), "b": summary(b)}
def format_srt_ts(seconds: float) -> str:
if seconds < 0:
seconds = 0.0
ms = int(round(seconds * 1000))
h, rem = divmod(ms, 3_600_000)
m, rem = divmod(rem, 60_000)
s, milli = divmod(rem, 1000)
return f"{h:02d}:{m:02d}:{s:02d},{milli:03d}"
def format_vtt_ts(seconds: float) -> str:
return format_srt_ts(seconds).replace(",", ".")
def parse_srt_ts(value: str) -> float:
value = value.strip().replace(",", ".")
h, m, rest = value.split(":")
s = float(rest)
return int(h) * 3600 + int(m) * 60 + s
def parse_srt(text: str) -> list[dict[str, Any]]:
blocks = re.split(r"\n\s*\n", text.strip(), flags=re.MULTILINE)
cues: list[dict[str, Any]] = []
for block in blocks:
lines = [ln.strip("\ufeff") for ln in block.splitlines() if ln.strip()]
if len(lines) < 2:
continue
idx = 1 if re.fullmatch(r"\d+", lines[0]) else 0
if idx >= len(lines) or "-->" not in lines[idx]:
continue
start_s, end_s = [p.strip() for p in lines[idx].split("-->")]
start_s = start_s.split()[0]
end_s = end_s.split()[0]
body = "\n".join(lines[idx + 1 :]).strip()
if not body:
continue
cues.append(
{
"start": parse_srt_ts(start_s),
"end": parse_srt_ts(end_s),
"text": body,
}
)
return cues
def cues_to_srt(cues: list[dict[str, Any]]) -> str:
lines: list[str] = []
for i, cue in enumerate(cues, 1):
lines.append(str(i))
lines.append(f"{format_srt_ts(cue['start'])} --> {format_srt_ts(cue['end'])}")
lines.append(cue["text"])
lines.append("")
return "\n".join(lines)
def cues_to_vtt(cues: list[dict[str, Any]]) -> str:
lines = ["WEBVTT", ""]
for cue in cues:
lines.append(f"{format_vtt_ts(cue['start'])} --> {format_vtt_ts(cue['end'])}")
lines.append(cue["text"])
lines.append("")
return "\n".join(lines)
def text_to_cues(text: str, duration: float = 5.0) -> list[dict[str, Any]]:
chunks = [c.strip() for c in re.split(r"(?<=[.!?])\s+|\n+", text) if c.strip()]
if not chunks:
chunks = [text.strip() or "[empty]"]
each = max(1.5, duration / max(len(chunks), 1))
cues = []
t = 0.0
for chunk in chunks:
cues.append({"start": t, "end": t + each, "text": chunk})
t += each
return cues
def tc_to_frames(tc: str, fps: float) -> int:
parts = tc.replace(";", ":").split(":")
if len(parts) != 4:
raise ValueError(f"Invalid timecode: {tc}")
h, m, s, f = (int(p) for p in parts)
fps_i = int(round(fps))
return ((h * 3600 + m * 60 + s) * fps_i) + f
def frames_to_tc(frames: int, fps: float) -> str:
fps_i = max(1, int(round(fps)))
if frames < 0:
frames = 0
f = frames % fps_i
total_s = frames // fps_i
s = total_s % 60
total_m = total_s // 60
m = total_m % 60
h = total_m // 60
return f"{h:02d}:{m:02d}:{s:02d}:{f:02d}"
def tc_to_seconds(tc: str, fps: float) -> float:
return tc_to_frames(tc, fps) / float(fps)
def seconds_to_tc(seconds: float, fps: float) -> str:
return frames_to_tc(int(round(seconds * fps)), fps)
def parse_time_arg(value: str) -> float:
"""Parse seconds (float) or HH:MM:SS[.mmm] or HH:MM:SS:FF timecode."""
if re.fullmatch(r"-?\d+(\.\d+)?", value):
return float(value)
if ":" in value:
parts = value.replace(",", ".").split(":")
if len(parts) == 3:
h, m, s = parts
return int(h) * 3600 + int(m) * 60 + float(s)
if len(parts) == 4:
return tc_to_seconds(value, 30.0)
raise ValueError(f"Unrecognized time format: {value}")
def add_input_arg(parser: argparse.ArgumentParser, *, required: bool = True) -> None:
parser.add_argument(
"--input",
"-i",
required=required,
help="Path to input media file",
)
def add_output_arg(parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"--output",
"-o",
help="Output path (default: workspace .mediaskills/generated/)",
)
def validate_input_path(path: str, op: str) -> Path:
p = Path(path)
if not p.is_file():
emit_error(op, f"Input file not found: {path}", code=EXIT_BAD_ARGS)
return p
def main_wrapper(main_fn: Any) -> None:
try:
main_fn()
except SystemExit:
raise
except ValueError as e:
emit_error("unknown", str(e), code=EXIT_BAD_ARGS)
except subprocess.CalledProcessError as e:
detail = e.stderr or e.stdout or str(e)
if isinstance(detail, bytes):
detail = detail.decode("utf-8", errors="replace")
emit_error("unknown", str(detail).strip()[:1200], code=EXIT_PROCESSING)
except RuntimeError as e:
emit_error("unknown", str(e), code=EXIT_PROCESSING)
scripts/ask.py
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Return probe data for an agent to answer a question about a file."""
from __future__ import annotations
import argparse
from _mediaskills_common import (
add_input_arg,
emit_error,
emit_success,
ffprobe_json,
main_wrapper,
validate_input_path,
EXIT_BAD_ARGS,
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=__doc__,
epilog="The agent reads probe data and answers --question in natural language.",
)
add_input_arg(parser)
parser.add_argument("--question", required=True, help="Question about the media file")
return parser
def main() -> None:
args = build_parser().parse_args()
op = "inspect.ask"
if not args.question.strip():
emit_error(op, "--question must not be empty", code=EXIT_BAD_ARGS)
path = validate_input_path(args.input, op)
probe = ffprobe_json(str(path), op)
emit_success(
op,
{
"input_path": str(path),
"question": args.question,
"probe": probe,
},
)
if __name__ == "__main__":
main_wrapper(main)
scripts/batch_probe.py
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Probe multiple files and return a table of metadata."""
from __future__ import annotations
import argparse
from pathlib import Path
from _mediaskills_common import (
emit_error,
emit_success,
ffprobe_json,
main_wrapper,
summarize_probe,
EXIT_BAD_ARGS,
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--paths",
nargs="+",
required=True,
help="One or more media file paths",
)
return parser
def main() -> None:
args = build_parser().parse_args()
op = "inspect.batch_probe"
rows: list[dict] = []
for raw in args.paths:
path = Path(raw)
if not path.is_file():
rows.append({"path": raw, "filename": path.name, "error": "file not found"})
continue
try:
probe = ffprobe_json(str(path), op)
summary = summarize_probe(probe)
fmt = probe.get("format", {})
rows.append(
{
"path": str(path),
"filename": path.name,
"duration_seconds": float(fmt.get("duration", 0) or 0),
"size_bytes": int(fmt.get("size", 0) or 0),
"format": summary.get("format"),
"video": summary.get("video"),
}
)
except SystemExit:
raise
except Exception as e:
rows.append({"path": str(path), "filename": path.name, "error": str(e)})
emit_success(op, {"rows": rows})
if __name__ == "__main__":
main_wrapper(main)
scripts/compare.py
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Compare two media files."""
from __future__ import annotations
import argparse
from pathlib import Path
from _mediaskills_common import (
compare_probe_summaries,
emit_error,
emit_success,
ffprobe_json,
main_wrapper,
EXIT_BAD_ARGS,
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--input-a", required=True, help="First media file")
parser.add_argument("--input-b", required=True, help="Second media file")
return parser
def main() -> None:
args = build_parser().parse_args()
op = "inspect.compare"
a, b = Path(args.input_a), Path(args.input_b)
if not a.is_file():
emit_error(op, f"Input file not found: {a}", code=EXIT_BAD_ARGS)
if not b.is_file():
emit_error(op, f"Input file not found: {b}", code=EXIT_BAD_ARGS)
probe_a = ffprobe_json(str(a), op)
probe_b = ffprobe_json(str(b), op)
emit_success(op, compare_probe_summaries(probe_a, probe_b))
if __name__ == "__main__":
main_wrapper(main)
scripts/describe.py
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Human-readable summary of a media file."""
from __future__ import annotations
import argparse
from _mediaskills_common import (
add_input_arg,
emit_success,
ffprobe_json,
main_wrapper,
summarize_probe,
validate_input_path,
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
add_input_arg(parser)
return parser
def main() -> None:
args = build_parser().parse_args()
op = "inspect.describe"
path = validate_input_path(args.input, op)
probe = ffprobe_json(str(path), op)
emit_success(op, summarize_probe(probe))
if __name__ == "__main__":
main_wrapper(main)
scripts/duration.py
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Get duration of a media file in seconds."""
from __future__ import annotations
import argparse
from _mediaskills_common import (
add_input_arg,
emit_success,
main_wrapper,
probe_duration,
require_cmd,
validate_input_path,
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
add_input_arg(parser)
return parser
def main() -> None:
args = build_parser().parse_args()
op = "inspect.duration"
path = validate_input_path(args.input, op)
require_cmd("ffprobe", op)
duration = probe_duration(str(path))
emit_success(op, {"duration_seconds": duration})
if __name__ == "__main__":
main_wrapper(main)
scripts/info.py
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Return inspect skill capabilities and requirements."""
from __future__ import annotations
import argparse
from _mediaskills_common import emit_success, main_wrapper
def build_parser() -> argparse.ArgumentParser:
return argparse.ArgumentParser(
description=__doc__,
epilog="Example: uv run info.py",
)
def main() -> None:
build_parser().parse_args()
emit_success(
"inspect.info",
{
"read_only": True,
"binaries": ["ffprobe"],
"scripts": [
"probe.py",
"describe.py",
"duration.py",
"resolution.py",
"compare.py",
"batch_probe.py",
"ask.py",
],
"outputs": "JSON metadata only; no file writes except optional compare tables",
"notes": "Run before destructive video/audio/image operations.",
},
)
if __name__ == "__main__":
main_wrapper(main)
scripts/probe.py
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Probe a media file with ffprobe and return structured metadata."""
from __future__ import annotations
import argparse
import json
from _mediaskills_common import (
EXIT_BAD_ARGS,
add_input_arg,
emit_error,
emit_success,
ffprobe_json,
main_wrapper,
validate_input_path,
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=__doc__,
epilog="Example: uv run probe.py --input video.mp4",
)
add_input_arg(parser)
return parser
def main() -> None:
args = build_parser().parse_args()
op = "inspect.probe"
path = validate_input_path(args.input, op)
data = ffprobe_json(str(path), op)
emit_success(op, data)
if __name__ == "__main__":
main_wrapper(main)
scripts/resolution.py
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Get video resolution of a media file."""
from __future__ import annotations
import argparse
from _mediaskills_common import (
add_input_arg,
emit_success,
main_wrapper,
require_cmd,
run,
validate_input_path,
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
add_input_arg(parser)
return parser
def main() -> None:
args = build_parser().parse_args()
op = "inspect.resolution"
path = validate_input_path(args.input, op)
require_cmd("ffprobe", op)
result = run(
[
"ffprobe",
"-v",
"error",
"-select_streams",
"v:0",
"-show_entries",
"stream=width,height",
"-of",
"csv=s=x:p=0",
str(path),
],
op,
)
wh = (result.stdout.strip().split("x") + ["", ""])[:2]
width = int(wh[0]) if wh[0] else None
height = int(wh[1]) if wh[1] else None
emit_success(op, {"width": width, "height": height})
if __name__ == "__main__":
main_wrapper(main)
SKILL.md
---
name: inspect
description: Probe, describe, compare, and batch-inspect video, audio, and image files using ffprobe. Use when you need metadata (duration, resolution, codecs, size), want to compare two files before/after processing, or need structured probe data to answer questions about a media file without modifying it.
license: MIT
compatibility: Requires ffprobe on PATH. Scripts are Python 3.11+, run via `uv run` (no separate install step).
metadata:
mediaskills-category: inspect
mediaskills-binaries: ffprobe
---
# Inspect
Read-only media inspection. Use this skill when you need **metadata without changing files**.
## When to use
- **This skill** — ffprobe metadata, duration, resolution, batch comparison tables, structured JSON for agent reasoning.
- **`audio` / `image` / `video-transformation`** — when you need to **transform** media (trim, transcode, resize). Run `inspect` first to confirm codecs and duration before destructive operations.
- **Raw ffprobe** — fine for one-off queries, but these scripts return consistent JSON the agent can parse reliably.
## Gotchas
- **Container vs stream duration** — ffprobe `format.duration` is authoritative for most files; individual stream durations can differ when audio/video lengths don't match (bad mux, trailing silence). Prefer `format.duration` for edit boundaries.
- **No video stream** — audio-only files return empty resolution; `resolution.py` reports `width`/`height` as null, not an error.
- **Variable frame rate (VFR)** — `avg_frame_rate` in probe JSON may differ from `r_frame_rate`. For frame-accurate edits, inspect both before assuming CFR.
- **Corrupt or partial files** — `moov atom not found` means an incomplete MP4 (common with interrupted downloads). Re-download or remux; probing cannot recover metadata.
- **Image "video" streams** — some still formats appear as a single-frame video stream; duration may be `N/A` or `0.04` seconds.
## Running scripts
This skill's scripts live in `scripts/` alongside this file. Before running any script, resolve its **absolute path** from *this skill's own directory* — do not assume the shell cwd is the skill folder, the repo root, or the workspace. Stay in the workspace (so media paths resolve) and pass the absolute script path to `uv run`:
```bash
uv run <skill_directory>/scripts/describe.py --input "<path-to-media>"
```
`<skill_directory>` is a **placeholder**. Never pass those angle brackets to the shell — the command will fail. Replace it with a real absolute directory before invoking `uv run`.
How to get that directory (do this yourself; do not ask the user first; do not skip the skill):
1. If your tools showed a filesystem path when loading this `SKILL.md`, strip `SKILL.md` and use that folder.
2. If you already ran another mediaskills script (for example `.../audio/scripts/extract.py`), this skill is a **sibling** of that folder: same parent, directory name `inspect`.
3. Otherwise search standard install locations and confirm the script exists:
```bash
NAME=inspect
for d in \
"$HOME/.agents/skills/$NAME" \
"$HOME/.cursor/skills/$NAME" \
"$HOME/.codex/skills/$NAME" \
"$HOME/.claude/skills/$NAME" \
"$PWD/.agents/skills/$NAME" \
"$PWD/.cursor/skills/$NAME" \
"$PWD/skills/$NAME"
do
if [ -f "$d/SKILL.md" ] && [ -e "$d/scripts/describe.py" ]; then
SKILL_DIR=$(cd "$d" && pwd)
break
fi
done
[ -n "$SKILL_DIR" ] || { echo "mediaskills: $NAME not found" >&2; exit 1; }
uv run "$SKILL_DIR/scripts/describe.py" --input "<path-to-media>"
```
Prefer absolute paths for `--input` / `--output` when cwd is uncertain.
## Missing binaries vs permissions
Before running any script, treat these two failures differently:
- If a required binary (`uv`, `ffprobe`) reports `No such file or directory` or `command not found`: this is a missing or unreachable binary, **not** a permissions issue. Escalating sandbox permissions will not fix it. Run `command -v uv` and `command -v ffprobe` to see whether they resolve at all. If not, install or re-link them (`install-media-tools` doctor/install) rather than retrying with wider permissions.
- Only escalate sandbox permissions if the error is explicitly a permission denial (e.g. `Operation not permitted`, `Permission denied`).
## Recipes
### Quick metadata check
```bash
uv run <skill_directory>/scripts/describe.py --input /path/to/clip.mp4
```
Example result:
```json
{"ok": true, "op": "inspect.describe", "data": {"format": "mov,mp4,m4a,3gp,3g2,mj2", "duration": "12.500000", "size": "1048576", "video": {"codec": "h264", "width": 1920, "height": 1080}, "audio_codecs": ["aac"]}, "output_paths": []}
```
### Full structured probe (for programmatic use)
```bash
uv run <skill_directory>/scripts/probe.py --input /path/to/clip.mp4
```
Returns complete ffprobe JSON in `data` — all streams, tags, bit rates.
### Compare before/after a transcode
```bash
uv run <skill_directory>/scripts/compare.py --input-a original.mp4 --input-b transcoded.mp4
```
Check `data.a` vs `data.b` for duration drift, resolution change, or size reduction.
### Batch inventory a folder
```bash
uv run <skill_directory>/scripts/batch_probe.py --paths clip1.mp4 clip2.mp4 clip3.wav
```
Use `data.rows` for a table. Rows with `error` indicate unreadable paths.
### Answer a specific question (agent workflow)
```bash
uv run <skill_directory>/scripts/ask.py --input clip.mp4 --question "Does this file have an audio track?"
```
Returns full probe JSON plus your question — the agent reads `data.probe.streams` and answers.
## Troubleshooting
| Error / symptom | Likely cause | Action |
| --- | --- | --- |
| `ffprobe not found` | Binary missing | Run `install-media-tools` doctor/install |
| `Invalid data found when processing input` | Not a media file or corrupt | Verify file type with `file` command |
| `width`/`height` null | Audio-only or subtitle-only | Use `probe.py` to list streams |
| Duration `N/A` | Still image or broken header | Try `ffprobe -show_format` manually |
## Available scripts
| Script | Purpose |
| --- | --- |
| `scripts/probe.py` | Full ffprobe JSON |
| `scripts/describe.py` | Compact human/agent summary |
| `scripts/duration.py` | Duration in seconds |
| `scripts/resolution.py` | Video width × height |
| `scripts/compare.py` | Side-by-side summary of two files |
| `scripts/batch_probe.py` | Table of metadata for many files |
| `scripts/ask.py` | Probe + question for agent reasoning |
## Acceptance checks (agent must pass before delivery)
1. Contract: exit 0, `ok: true`.
2. Spot-check: duration/streams/codecs answer the user's question; use `compare.py` when verifying a transform result against its source.
3. On failure: re-probe or escalate; do not invent metadata.
## Do not use for
- Modifying or transcoding media (use `audio`, `image`, or `video-transformation`)
- Downloading from URLs (use `download`)
- Generating captions from speech (use `speech-captions`)
## Related skills
- `install-media-tools` — install ffprobe/ffmpeg
- `audio`, `video-transformation`, `image` — processing after inspection
tests/test_inspect.py
"""Tests for inspect skill scripts."""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parents[3]
from tests.helpers import parse_json_stdout, run_script # noqa: E402
SKILL_DIR = Path(__file__).resolve().parents[1]
SCRIPTS = SKILL_DIR / "scripts"
@pytest.fixture(scope="module", autouse=True)
def sync_common():
import subprocess
subprocess.run(
[sys.executable, str(REPO_ROOT / "scripts" / "sync_shared_libs.py")],
check=True,
cwd=REPO_ROOT,
)
def test_info():
result = run_script(SCRIPTS / "info.py")
assert result.returncode == 0, result.stderr
data = parse_json_stdout(result)
assert data["op"] == "inspect.info"
assert data["data"]["read_only"] is True
def test_probe(sample_video: Path):
result = run_script(SCRIPTS / "probe.py", "--input", str(sample_video))
assert result.returncode == 0, result.stderr
data = parse_json_stdout(result)
assert data["ok"] is True
assert data["op"] == "inspect.probe"
assert "streams" in data["data"]
def test_describe(sample_video: Path):
result = run_script(SCRIPTS / "describe.py", "--input", str(sample_video))
assert result.returncode == 0, result.stderr
data = parse_json_stdout(result)
assert data["ok"] is True
assert data["data"]["video"]["width"] == 320
def test_duration(sample_video: Path):
result = run_script(SCRIPTS / "duration.py", "--input", str(sample_video))
assert result.returncode == 0, result.stderr
data = parse_json_stdout(result)
assert 1.5 <= data["data"]["duration_seconds"] <= 2.5
def test_resolution(sample_video: Path):
result = run_script(SCRIPTS / "resolution.py", "--input", str(sample_video))
assert result.returncode == 0, result.stderr
data = parse_json_stdout(result)
assert data["data"]["width"] == 320
assert data["data"]["height"] == 240
def test_compare(sample_video: Path, tmp_path: Path):
copy = tmp_path / "copy.mp4"
copy.write_bytes(sample_video.read_bytes())
result = run_script(
SCRIPTS / "compare.py",
"--input-a",
str(sample_video),
"--input-b",
str(copy),
)
assert result.returncode == 0, result.stderr
data = parse_json_stdout(result)
assert "a" in data["data"] and "b" in data["data"]
def test_batch_probe(sample_video: Path, sample_audio: Path):
result = run_script(
SCRIPTS / "batch_probe.py",
"--paths",
str(sample_video),
str(sample_audio),
)
assert result.returncode == 0, result.stderr
data = parse_json_stdout(result)
assert len(data["data"]["rows"]) == 2
def test_ask(sample_video: Path):
result = run_script(
SCRIPTS / "ask.py",
"--input",
str(sample_video),
"--question",
"What is the resolution?",
)
assert result.returncode == 0, result.stderr
data = parse_json_stdout(result)
assert data["data"]["question"] == "What is the resolution?"