OpenFlowFrames/cli.py at main · ZeroHackz/OpenFlowFrames · GitHub

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
86
87
88
89
90
91
92
93
"""Headless CLI for OpenFlowFrames.

Drives the same core engine as the GUI, so interpolation can run without a
display — from scripts, CI, or agent plugins (e.g. the DeepSeek Harness
`dsh-openflowframes` bundle).

Usage:
    python cli.py --probe --input C:\\path\\to\\video.mp4
    python cli.py --input C:\\path\\to\\video.mp4 --factor 2 --model "RIFE 4.9" --output out.mp4
    python cli.py --input C:\\path\\to\\frames --fps 24 --factor 4 --out-mode png --output out_frames
"""

import argparse
import json
import sys
from pathlib import Path

from core import engine


def _probe_json(src: Path, fps: float) -> str:
    if src.is_dir():
        info = engine.probe_image_dir(src, fps)
    else:
        info = engine.probe(src)
    return json.dumps({
        "path": str(info.path),
        "width": info.width,
        "height": info.height,
        "fps": info.fps_float,
        "frame_count": info.frame_count,
        "has_audio": info.has_audio,
        "is_frames": info.is_frames,
    })


def _pick_model(name: str | None):
    models = engine.load_models()
    if name:
        for m in models:
            if name.lower() in m["name"].lower():
                return m
        sys.exit(f"Model '{name}' not found. Available: "
                 f"{', '.join(m['name'] for m in models)}")
    return engine.default_model(models)


def main():
    ap = argparse.ArgumentParser(prog="openflowframes",
                                 description="Headless RIFE interpolation")
    ap.add_argument("--input", required=True,
                    help="Video file or directory of PNG/JPG/WebP frames")
    ap.add_argument("--output", default=None,
                    help="Output file (mp4) or folder (png). Default: next to the input")
    ap.add_argument("--factor", type=int, default=2, help="Interpolation factor (default 2)")
    ap.add_argument("--model", default=None, help="Model name or substring (default: repo default)")
    ap.add_argument("--crf", type=int, default=17, help="H.264 quality (default 17)")
    ap.add_argument("--fps", type=float, default=24.0,
                    help="Input framerate when --input is a frame directory")
    ap.add_argument("--out-mode", choices=["mp4", "png"], default="mp4",
                    help="mp4 video or png frame sequence (default mp4)")
    ap.add_argument("--probe", action="store_true",
                    help="Print input metadata as JSON and exit")
    args = ap.parse_args()

    src = Path(args.input)
    if not src.exists():
        sys.exit(f"Input not found: {src}")

    if args.probe:
        print(_probe_json(src, args.fps))
        return

    if src.is_dir():
        video = engine.probe_image_dir(src, args.fps)
    else:
        video = engine.probe(src)
    model = _pick_model(args.model)
    out = (Path(args.output) if args.output
           else engine.make_out_path(video, args.factor, args.out_mode))

    def report(ev: engine.Progress):
        if ev.message:
            print(f"[{ev.stage} {ev.fraction * 100:.0f}%] {ev.message}", flush=True)

    job = engine.InterpolationJob(video, model, args.factor, out,
                                  crf=args.crf, out_mode=args.out_mode)
    job.run(report)
    print(f"Saved: {out}", flush=True)


if __name__ == "__main__":
    main()