Skip to content

cli

Modules

fastvideo.entrypoints.cli.cli_types

Classes

fastvideo.entrypoints.cli.cli_types.CLISubcommand

Base class for CLI subcommands

Functions
fastvideo.entrypoints.cli.cli_types.CLISubcommand.cmd 
cmd(args: Namespace) -> None

Execute the command with the given arguments

Source code in fastvideo/entrypoints/cli/cli_types.py 
14
15
16
def cmd(self, args: argparse.Namespace) -> None:
    """Execute the command with the given arguments"""
    raise NotImplementedError
fastvideo.entrypoints.cli.cli_types.CLISubcommand.subparser_init 
subparser_init(subparsers: _SubParsersAction) -> FlexibleArgumentParser

Initialize the subparser for this command

Source code in fastvideo/entrypoints/cli/cli_types.py 
22
23
24
25
26
def subparser_init(
        self,
        subparsers: argparse._SubParsersAction) -> FlexibleArgumentParser:
    """Initialize the subparser for this command"""
    raise NotImplementedError
fastvideo.entrypoints.cli.cli_types.CLISubcommand.validate 
validate(args: Namespace) -> None

Validate the arguments for this command

Source code in fastvideo/entrypoints/cli/cli_types.py 
18
19
20
def validate(self, args: argparse.Namespace) -> None:
    """Validate the arguments for this command"""
    pass

fastvideo.entrypoints.cli.generate

Classes

fastvideo.entrypoints.cli.generate.GenerateSubcommand 
GenerateSubcommand()

Bases: CLISubcommand

The generate subcommand for the FastVideo CLI

Source code in fastvideo/entrypoints/cli/generate.py 
23
24
25
26
27
def __init__(self) -> None:
    self.name = "generate"
    super().__init__()
    self.init_arg_names = self._get_init_arg_names()
    self.generation_arg_names = self._get_generation_arg_names()
Functions
fastvideo.entrypoints.cli.generate.GenerateSubcommand.validate 
validate(args: Namespace) -> None

Validate the arguments for this command

Source code in fastvideo/entrypoints/cli/generate.py 
93
94
95
96
97
98
99
def validate(self, args: argparse.Namespace) -> None:
    """Validate the arguments for this command"""
    if args.num_gpus is not None and args.num_gpus <= 0:
        raise ValueError("Number of gpus must be positive")

    if args.config and not os.path.exists(args.config):
        raise ValueError(f"Config file not found: {args.config}")

Functions

fastvideo.entrypoints.cli.main

Classes

Functions

fastvideo.entrypoints.cli.main.cmd_init 
cmd_init() -> list[CLISubcommand]

Initialize all commands from separate modules

Source code in fastvideo/entrypoints/cli/main.py 
 9
10
11
12
13
def cmd_init() -> list[CLISubcommand]:
    """Initialize all commands from separate modules"""
    commands = []
    commands.extend(generate_cmd_init())
    return commands

fastvideo.entrypoints.cli.utils

Functions

fastvideo.entrypoints.cli.utils.launch_distributed 
launch_distributed(num_gpus: int, args: list[str], master_port: int | None = None) -> int

Launch a distributed job with the given arguments

Parameters:

Name Type Description Default
num_gpus int

Number of GPUs to use

 required
args list[str]

Arguments to pass to v1_fastvideo_inference.py (defaults to sys.argv[1:])

 required
master_port int | None

Port for the master process (default: random)

 None
Source code in fastvideo/entrypoints/cli/utils.py 
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
def launch_distributed(num_gpus: int,
                       args: list[str],
                       master_port: int | None = None) -> int:
    """
    Launch a distributed job with the given arguments

    Args:
        num_gpus: Number of GPUs to use
        args: Arguments to pass to v1_fastvideo_inference.py (defaults to sys.argv[1:])
        master_port: Port for the master process (default: random)
    """

    current_env = os.environ.copy()
    python_executable = sys.executable
    project_root = os.path.abspath(
        os.path.join(os.path.dirname(__file__), "../../../.."))
    main_script = os.path.join(project_root,
                               "fastvideo/sample/v1_fastvideo_inference.py")

    cmd = [
        python_executable, "-m", "torch.distributed.run",
        f"--nproc_per_node={num_gpus}"
    ]

    if master_port is not None:
        cmd.append(f"--master_port={master_port}")

    cmd.append(main_script)
    cmd.extend(args)

    logger.info("Running inference with %d GPU(s)", num_gpus)
    logger.info("Launching command: %s", " ".join(cmd))

    current_env["PYTHONIOENCODING"] = "utf-8"
    process = subprocess.Popen(cmd,
                               env=current_env,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.STDOUT,
                               universal_newlines=True,
                               bufsize=1,
                               encoding='utf-8',
                               errors='replace')

    if process.stdout:
        for line in iter(process.stdout.readline, ''):
            print(line.strip())

    return process.wait()