refactor(cli): migrate from argparse to typer for command-line interface

This commit is contained in:
2026-03-06 11:41:35 +08:00
parent 33f2b8f542
commit 7dbd704d6b
8 changed files with 195 additions and 91 deletions

View File

@@ -0,0 +1,6 @@
from .train import train
from .benchmark import benchmark
from .visualize import visualize
from .generate import generate
__all__ = ["train", "benchmark", "visualize", "generate"]

View File

@@ -0,0 +1,53 @@
from typing import cast
import typer
def benchmark(
ctx: typer.Context,
model_path: str = typer.Option(
None, "--model", "-m", help="Path to compressor model weights"
),
):
import torch
from benchmarks import run_benchmark
from compressors import DinoCompressor
from configs import cfg_manager
from transformers import AutoImageProcessor, BitImageProcessorFast
from utils import get_device
config = cfg_manager.get()
benchmark_cfg = config.benchmark
if not benchmark_cfg.enabled:
typer.echo(
"Benchmark is not enabled. Set benchmark.enabled=true in config.yaml",
err=True,
)
raise typer.Exit(code=1)
device = get_device()
model_cfg = config.model
processor = cast(
BitImageProcessorFast,
AutoImageProcessor.from_pretrained(model_cfg.dino_model, device_map=device),
)
model = DinoCompressor().to(device)
if model_path:
from compressors import HashCompressor
compressor = HashCompressor(
input_dim=model_cfg.compression_dim,
hash_bits=model_cfg.compression_dim,
)
compressor.load_state_dict(torch.load(model_path))
model.compressor = compressor
run_benchmark(
model=model,
processor=processor,
config=benchmark_cfg,
model_name="dinov2",
)

View File

@@ -0,0 +1,25 @@
import typer
def generate(ctx: typer.Context):
from configs import cfg_manager
from data_loading.synthesizer import ImageSynthesizer
config = cfg_manager.get()
dataset_cfg = config.dataset
synthesizer = ImageSynthesizer(
dataset_root=dataset_cfg.dataset_root,
output_dir=dataset_cfg.output_dir,
num_objects_range=dataset_cfg.num_objects_range,
num_scenes=dataset_cfg.num_scenes,
object_scale_range=dataset_cfg.object_scale_range,
rotation_range=dataset_cfg.rotation_range,
overlap_threshold=dataset_cfg.overlap_threshold,
seed=dataset_cfg.seed,
)
generated_files = synthesizer.generate()
typer.echo(
f"Generated {len(generated_files)} synthesized images in {dataset_cfg.output_dir}"
)

View File

@@ -0,0 +1,20 @@
import typer
def train(
ctx: typer.Context,
epoch_size: int = typer.Option(10, "--epoch", "-e", help="Number of epochs"),
batch_size: int = typer.Option(64, "--batch", "-b", help="Batch size"),
lr: float = typer.Option(1e-4, "--lr", "-l", help="Learning rate"),
checkpoint_path: str = typer.Option(
"hash_checkpoint.pt", "--checkpoint", "-c", help="Checkpoint path"
),
):
from compressors import train as train_module
train_module(
epoch_size=epoch_size,
batch_size=batch_size,
lr=lr,
checkpoint_path=checkpoint_path,
)

View File

@@ -0,0 +1,12 @@
import typer
def visualize(
ctx: typer.Context,
host: str = typer.Option("127.0.0.1", "--host", help="Server host"),
port: int = typer.Option(8050, "--port", "-p", help="Server port"),
debug: bool = typer.Option(True, "--debug/--no-debug", help="Enable debug mode"),
):
from visualizer import app as dash_app
dash_app.run(host=host, port=port, debug=debug)

View File

@@ -1,89 +1,16 @@
import argparse
import typer
from commands import benchmark, generate, train, visualize
app = typer.Typer(
name="mini-nav",
help="Mini-Nav: A vision-language navigation system",
add_completion=False,
)
app.command(name="train")(train)
app.command(name="benchmark")(benchmark)
app.command(name="visualize")(visualize)
app.command(name="generate")(generate)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument(
"action",
choices=["train", "benchmark", "visualize", "generate"],
help="Action to perform: train, benchmark, visualize, or generate",
)
args = parser.parse_args()
if args.action == "train":
from compressors import train
train(
epoch_size=10, batch_size=64, lr=1e-4, checkpoint_path="hash_checkpoint.pt"
)
elif args.action == "benchmark":
from typing import cast
import torch
from benchmarks import run_benchmark
from compressors import DinoCompressor
from configs import cfg_manager
from transformers import AutoImageProcessor, BitImageProcessorFast
from utils import get_device
config = cfg_manager.get()
benchmark_cfg = config.benchmark
if not benchmark_cfg.enabled:
print("Benchmark is not enabled. Set benchmark.enabled=true in config.yaml")
exit(1)
device = get_device()
# Load model and processor based on config
model_cfg = config.model
processor = cast(
BitImageProcessorFast,
AutoImageProcessor.from_pretrained(model_cfg.dino_model, device_map=device),
)
# Load compressor weights if specified in model config
model = DinoCompressor().to(device)
if model_cfg.compressor_path is not None:
from compressors import HashCompressor
compressor = HashCompressor(
input_dim=model_cfg.compression_dim,
output_dim=model_cfg.compression_dim,
)
compressor.load_state_dict(torch.load(model_cfg.compressor_path))
# Wrap with compressor if path is specified
model.compressor = compressor
# Run benchmark
run_benchmark(
model=model,
processor=processor,
config=benchmark_cfg,
model_name="dinov2",
)
elif args.action == "visualize":
from visualizer import app
app.run(debug=True)
else: # generate
from configs import cfg_manager
from data_loading.synthesizer import ImageSynthesizer
config = cfg_manager.get()
dataset_cfg = config.dataset
synthesizer = ImageSynthesizer(
dataset_root=dataset_cfg.dataset_root,
output_dir=dataset_cfg.output_dir,
num_objects_range=dataset_cfg.num_objects_range,
num_scenes=dataset_cfg.num_scenes,
object_scale_range=dataset_cfg.object_scale_range,
rotation_range=dataset_cfg.rotation_range,
overlap_threshold=dataset_cfg.overlap_threshold,
seed=dataset_cfg.seed,
)
generated_files = synthesizer.generate()
print(
f"Generated {len(generated_files)} synthesized images in {dataset_cfg.output_dir}"
)
app()