#!/usr/bin/env python3 """Sprite generation pipeline CLI for Magic Civilization. Usage: python3 tools/sprite-generation/cli.py scan python3 tools/sprite-generation/cli.py status python3 tools/sprite-generation/cli.py generate --category terrain --variants 8 python3 tools/sprite-generation/cli.py poll --watch python3 tools/sprite-generation/cli.py review --port 5801 python3 tools/sprite-generation/cli.py install """ from __future__ import annotations import argparse import json import sys from pathlib import Path TOOL_DIR = Path(__file__).resolve().parent PROJECT = TOOL_DIR.parent.parent sys.path.insert(0, str(TOOL_DIR)) CONFIG_PATH = TOOL_DIR / "sprite-config.json" DB_PATH = TOOL_DIR / "spritegen.db" RAW_DIR = TOOL_DIR / "raw" VARIANTS_DIR = TOOL_DIR / "variants" HEX_MASK_PATH = TOOL_DIR / "hex_mask.png" LOCAL_DATA = PROJECT / "public" / "games" / "age-of-dwarves" / "data" DEMO_DATA = TOOL_DIR / "demo-data" ASSETS_DIR = PROJECT / "public" / "games" / "age-of-dwarves" / "assets" def _data_dir(args: argparse.Namespace) -> Path: if hasattr(args, "data_dir") and args.data_dir: return Path(args.data_dir) if hasattr(args, "demo") and args.demo: return DEMO_DATA return LOCAL_DATA def _load_config() -> dict: return json.loads(CONFIG_PATH.read_text()) def _registry(): from engine.registry import SpriteRegistry return SpriteRegistry(DB_PATH) def cmd_scan(args: argparse.Namespace) -> None: from engine.scanner import SpriteScanner reg = _registry() data = _data_dir(args) print(f"Scanning data from: {data}") demo = hasattr(args, "demo") and args.demo sprite_type = getattr(args, "sprite_type", None) scanner = SpriteScanner(data_dir=data, assets_dir=ASSETS_DIR, registry=reg) if demo: report = scanner.scan_all(skip_biome_grid=True, skip_ui=True, sprite_type=sprite_type) else: report = scanner.scan_all(sprite_type=sprite_type) print(f"\nBy category:") for cat, count in sorted(report.categories.items()): print(f" {cat:20s} {count:4d} sprites") print(f"\n {'TOTAL':20s} {report.new_sprites:4d} new, {report.existing_sprites} existing") print(f" {'DIMENSIONS':20s} {report.new_dimensions:4d} new, {report.existing_dimensions} existing") def cmd_status(args: argparse.Namespace) -> None: reg = _registry() stats = reg.get_stats() if not stats["by_category"]: print("No sprites in registry. Run 'scan' first.") return print(f"\n{'Category':<20s} {'needed':>8s} {'generating':>12s} {'review':>8s} {'approved':>10s} {'installed':>10s} {'rejected':>10s} {'skip':>6s} {'TOTAL':>8s}") print("─" * 102) for cat in sorted(stats["by_category"]): row = stats["by_category"][cat] total = sum(row.values()) print( f"{cat:<20s} {row.get('needed', 0):>8d} {row.get('generating', 0):>12d} " f"{row.get('review', 0):>8d} {row.get('approved', 0):>10d} {row.get('installed', 0):>10d} " f"{row.get('rejected', 0):>10d} {row.get('skip', 0):>6d} {total:>8d}" ) total_row = stats["total"] grand = sum(total_row.values()) print("─" * 102) print( f"{'TOTAL':<20s} {total_row.get('needed', 0):>8d} {total_row.get('generating', 0):>12d} " f"{total_row.get('review', 0):>8d} {total_row.get('approved', 0):>10d} {total_row.get('installed', 0):>10d} " f"{total_row.get('rejected', 0):>10d} {total_row.get('skip', 0):>6d} {grand:>8d}" ) def cmd_generate(args: argparse.Namespace) -> None: """Submit generation requests to model-boss queue. Returns immediately.""" import asyncio from engine.generator import SpriteGenerator reg = _registry() config = _load_config() gen = SpriteGenerator(config=config, registry=reg, raw_dir=RAW_DIR) sprites = reg.get_sprites( category=args.category, status="needed", limit=args.max or 10000, ) if args.sprite: sprites = [s for s in sprites if s["id"] == args.sprite] if not sprites: print("No sprites in 'needed' status matching filters.") return sprite_ids = [s["id"] for s in sprites] if args.dry_run: print(f"Would generate {len(sprite_ids)} sprites x {args.variants} variants = {len(sprite_ids) * args.variants} jobs\n") for sid in sprite_ids[:50]: print(f" {sid}") if len(sprite_ids) > 50: print(f" ... and {len(sprite_ids) - 50} more") return submitted = asyncio.run(gen.submit_batch( sprite_ids=sprite_ids, variants_per=args.variants, priority=args.priority, )) print(f"\nSubmitted {submitted} requests. Run 'listen' to collect results.") def cmd_listen(args: argparse.Namespace) -> None: """Collect pending generation results via Redis pubsub.""" import asyncio from engine.generator import SpriteGenerator reg = _registry() config = _load_config() gen = SpriteGenerator(config=config, registry=reg, raw_dir=RAW_DIR) collected = asyncio.run(gen.collect_pending()) print(f"\nCollected {collected} images. Run 'rank' to score them.") def cmd_rank(args: argparse.Namespace) -> None: import asyncio from engine.ranker import SpriteRanker reg = _registry() ranker = SpriteRanker(registry=reg, raw_dir=RAW_DIR) if args.sprite: result = asyncio.run(ranker.rank_and_filter(args.sprite)) print(f"\n{args.sprite}: {result['good_count']}/{len(result['ranked'])} good variants") for r in result["ranked"]: scores = r["scores"] flag = "✓" if r["confidence"] >= 0.7 else "✗" dims = " ".join(f"{k}={v:.2f}" for k, v in scores.items()) print(f" {flag} variant {r['variant_id']} (seed {r['seed']}): confidence={r['confidence']:.2f} {dims}") if result["needs_regen"]: print(f"\n Needs {result['deficit']} more good variants — re-generate with higher priority") else: print(f"Ranking all sprites in 'review' status...") reg = _registry() # Always filter to units unless explicitly overridden category_filter = args.category if hasattr(args, "category") and args.category else "units" all_sprites = reg.get_sprites(category=category_filter) total_ready = 0 total_need_regen = 0 total_sprites = 0 for sprite in all_sprites: if sprite["status"] == "review": total_sprites += 1 result = asyncio.run(ranker.rank_and_filter(sprite["id"])) if result["good_count"] >= ranker.target_approved: total_ready += 1 else: total_need_regen += 1 print(f"\nSummary: {total_ready} ready, {total_need_regen} need re-generation (of {total_sprites} total)") def cmd_install(args: argparse.Namespace) -> None: from engine.installer import SpriteInstaller reg = _registry() installer = SpriteInstaller(assets_dir=ASSETS_DIR, registry=reg) count = installer.install_approved(category=args.category, dry_run=args.dry_run) print(f"\n{'Would install' if args.dry_run else 'Installed'}: {count} sprites") def cmd_approve(args: argparse.Namespace) -> None: from engine.pipeline import SpritePipeline reg = _registry() pipeline = SpritePipeline( registry=reg, raw_dir=RAW_DIR, variants_dir=VARIANTS_DIR, assets_dir=ASSETS_DIR, game_db_path=LOCAL_DATA / "sprites.db", ) result = pipeline.approve_and_install(args.variant, alt_name=args.alt) if result: print(f"\nShipped: {result}") else: print("\nFailed to install.") def cmd_reset(args: argparse.Namespace) -> None: reg = _registry() if not args.sprite: print("--sprite is required for reset") return reg.update_sprite_status(args.sprite, "needed") print(f"Reset {args.sprite} to 'needed'") def cmd_review(args: argparse.Namespace) -> None: print(f"Starting review GUI server on port {args.port}...") print(f"Open http://localhost:{args.port} in your browser") try: from server import create_app import uvicorn app = create_app( registry=_registry(), raw_dir=RAW_DIR, variants_dir=VARIANTS_DIR, ) uvicorn.run(app, host="0.0.0.0", port=args.port) except ImportError as e: print(f"Error: {e}") print("Install dependencies: pip install fastapi uvicorn") def cmd_run(args: argparse.Namespace) -> None: """Full pipeline: submit → collect → rank → regen loop. 1. Submit all needed sprites to model-boss queue 2. Collect results as they arrive via Redis pubsub 3. Score each completed variant immediately 4. Re-queue sprites that need more variants 5. Repeat until all sprites are in review or approved """ import asyncio from engine.generator import SpriteGenerator from engine.ranker import SpriteRanker reg = _registry() config = _load_config() gen = SpriteGenerator(config=config, registry=reg, raw_dir=RAW_DIR) ranker = SpriteRanker(registry=reg, raw_dir=RAW_DIR) variants_per = args.variants MAX_REGEN_ATTEMPTS = getattr(args, "max_attempts", 15) regen_counts: dict[str, int] = {} # Start GUI server in background thread import threading def _serve(): from server import create_app import uvicorn app = create_app(registry=reg, raw_dir=RAW_DIR, variants_dir=VARIANTS_DIR) uvicorn.run(app, host="0.0.0.0", port=args.port, log_level="warning") server_thread = threading.Thread(target=_serve, daemon=True) server_thread.start() print(f"Review GUI: http://localhost:{args.port}/?spriteTheater=true") async def _on_variant_complete(variant_id: int, sprite_id: str) -> None: """Called when a variant result arrives. Advance through scoring pipeline.""" await ranker.advance_sprite(sprite_id) async def _run_pipeline(): loop_count = 0 while True: loop_count += 1 # --- Phase 1: Submit all needed sprites --- needed = reg.get_sprites( category=args.category, status="needed", limit=100, ) # Find sprites that already have in-flight (submitted) variants — don't re-queue them in_flight_ids: set[str] = set() if needed: placeholders = ",".join("?" * len(needed)) ids = [s["id"] for s in needed] rows = reg.conn.execute( f"SELECT DISTINCT sprite_id FROM variants WHERE sprite_id IN ({placeholders}) AND job_status='submitted'", ids, ).fetchall() in_flight_ids = {r[0] for r in rows} to_submit = [] for candidate in (needed or []): sid = candidate["id"] if sid in in_flight_ids: continue # already has pending variants — don't double-queue or burn regen_counts if regen_counts.get(sid, 0) < MAX_REGEN_ATTEMPTS: to_submit.append(sid) regen_counts[sid] = regen_counts.get(sid, 0) + 1 else: reg.update_sprite_status(sid, "review") print(f" {sid}: max regen attempts, moving to review") if to_submit: print(f"\n[loop {loop_count}] Submitting {len(to_submit)} sprites x {variants_per} variants...") await gen.submit_batch( sprite_ids=to_submit, variants_per=variants_per, priority="high", ) # --- Phase 2: Collect pending results via pubsub --- collected = await gen.collect_pending(on_complete=_on_variant_complete) # --- Phase 3: Evaluate sprites and decide regen --- if collected > 0: # Check which sprites now have enough good variants review_sprites = reg.get_sprites( category=args.category, status="review", limit=1000, ) for sprite in (review_sprites or []): result = await ranker.rank_and_filter(sprite["id"]) good = result["good_count"] total = len(result["ranked"]) if result["needs_regen"]: attempt = regen_counts.get(sprite["id"], 0) if attempt < MAX_REGEN_ATTEMPTS: reg.update_sprite_status(sprite["id"], "needed") print(f" {sprite['id']}: {good}/{total} good — needs regen") else: if result["ranked"]: best = result["ranked"][0] gate_ok = best.get("gate_passed", True) status = "gates passed" if gate_ok else "gate failed" print(f" {sprite['id']}: {good}/{total} good — {status}, best={best['confidence']:.0%}") # --- Status --- stats = reg.get_stats() total_row = stats["total"] needed_count = total_row.get("needed", 0) review_count = total_row.get("review", 0) done_count = total_row.get("approved", 0) + total_row.get("installed", 0) total_count = sum(total_row.values()) queued = reg.conn.execute( "SELECT COUNT(*) FROM variants WHERE job_status = 'submitted'" ).fetchone()[0] print(f"\n[loop {loop_count}] {needed_count} needed | {queued} queued | {review_count} review | {done_count} done / {total_count} total") if needed_count == 0 and queued == 0: print("\nAll sprites processed. Ctrl+C to stop.") try: await asyncio.sleep(30) except asyncio.CancelledError: break else: await asyncio.sleep(2) try: asyncio.run(_run_pipeline()) except KeyboardInterrupt: print("\nStopping pipeline.") def cmd_test_prompt(args: argparse.Namespace) -> None: """Generate test images with current prompt formula — no DB writes.""" import asyncio import base64 from pathlib import Path import httpx from engine.prompts import compose_prompt, get_negative config = _load_config() model = config["model"] guidance = config["defaults"].get("guidance_scale", 7.5) steps = config["defaults"].get("steps", 25) entity = { "name": args.entity.replace("_", " ").title(), "description": "", "combat_type": args.combat_type, } dims = {"race": args.race, "gender": args.gender} prompt = compose_prompt("units", entity, dims) negative = get_negative("units", combat_type=args.combat_type) print(f"Model: {model} Guidance: {guidance} Steps: {steps}") print(f"Prompt: {prompt[:200]}...") print(f"Negative: {negative[:100]}...") print(f"Seeds: {args.seeds}") outdir = Path(args.outdir) outdir.mkdir(exist_ok=True) async def generate_one(seed: int) -> None: async with httpx.AsyncClient(timeout=120.0) as client: body = { "model": model, "prompt": prompt, "negative_prompt": negative, "width": 1024, "height": 1024, "steps": steps, "guidance_scale": guidance, "seed": seed, } resp = await client.post(f"{config['api_base']}/v1/images/generations", json=body) if resp.status_code != 200: print(f" seed={seed}: FAILED ({resp.status_code})") return data = resp.json() img_b64 = None if "data" in data and data["data"]: img_b64 = data["data"][0].get("b64_json") if not img_b64 and "images" in data: img_b64 = data["images"][0] if img_b64: name = f"{args.entity}_{args.race}_{args.gender[0]}_s{seed}.png" path = outdir / name path.write_bytes(base64.b64decode(img_b64)) print(f" seed={seed} → {path}") for seed in args.seeds: asyncio.run(generate_one(seed)) print(f"\nResults in {outdir}") def cmd_refresh_prompts(args: argparse.Namespace) -> None: """Rebuild all stored prompts from current templates.""" from engine.prompts import compose_prompt, get_negative reg = _registry() sprites = reg.get_sprites(category=args.category, limit=100000) updated = 0 for s in sprites: eid = s["entity_id"] sid = s["id"] # Parse race/gender from entity_id dims = {} parts = eid if parts.endswith("_m"): dims["gender"] = "male" parts = parts[:-2] elif parts.endswith("_f"): dims["gender"] = "female" parts = parts[:-2] for race in ["dwarves", "humans", "high_elves", "orcs"]: suffix = f"_{race}" if parts.endswith(suffix): dims["race"] = race parts = parts[:-len(suffix)] break # Get stored dimensions dim_rows = reg.conn.execute( "SELECT dimension_type, dimension_value FROM sprite_dimensions WHERE sprite_id=?", (sid,), ).fetchall() for r in dim_rows: dims[r["dimension_type"]] = r["dimension_value"] # Determine combat type from entity_id base = parts.lower() combat_type = "melee" if any(w in base for w in ["archers", "bowmen", "crossbow", "longbow", "musket", "rifle"]): combat_type = "ranged" elif any(w in base for w in ["cavalry", "heavy_cavalry"]): combat_type = "cavalry" elif any(w in base for w in ["fishing", "trawler", "deep_sea", "marine"]): combat_type = "marine" elif any(w in base for w in ["founder", "wanderer", "engineer"]): combat_type = "civilian" elif any(w in base for w in ["magician", "field_medic", "field_commander", "tactician", "siege_master"]): combat_type = "specialist" entity = { "name": parts.replace("_", " ").title(), "description": "", "combat_type": combat_type, "entity_id": eid, } prompt = compose_prompt(args.category, entity, dims) neg = get_negative(args.category, combat_type=combat_type) reg.conn.execute("UPDATE sprites SET prompt=?, negative_prompt=? WHERE id=?", (prompt, neg, sid)) updated += 1 if args.clear_scores: result = reg.conn.execute( "UPDATE variants SET notes=NULL, rating=NULL " "WHERE is_approved=0 AND rating != -1 AND sprite_id IN " "(SELECT id FROM sprites WHERE category=?)", (args.category,), ) print(f"Cleared {result.rowcount} scores") reg.conn.execute( "UPDATE sprites SET status='needed' WHERE category=? AND status='review'", (args.category,), ) reg.conn.commit() print(f"Updated {updated} {args.category} prompts") def cmd_export(args: argparse.Namespace) -> None: import csv import io reg = _registry() sprites = reg.get_sprites(limit=100000) if args.format == "json": print(json.dumps(sprites, indent=2)) else: if not sprites: print("No sprites in registry.") return writer = csv.DictWriter(sys.stdout, fieldnames=sprites[0].keys()) writer.writeheader() for s in sprites: writer.writerow(s) def cmd_monitor(args: argparse.Namespace) -> None: """Monitor tier progression — report when Opus has 3 candidates for user review.""" import sqlite3 import time from datetime import datetime conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row opus_ready = {} check_count = 0 print("\n" + "="*80) print("SPRITE PIPELINE MONITOR — Tier Progression Tracker") print("="*80) while True: # Get tier status for all sprites status = conn.execute(""" SELECT sprite_id, COUNT(*) as total_variants, SUM(CASE WHEN COALESCE(review_tier, 0) >= 0 THEN 1 ELSE 0 END) as tier0_scored, SUM(CASE WHEN COALESCE(review_tier, 0) >= 1 THEN 1 ELSE 0 END) as tier1_scored, SUM(CASE WHEN COALESCE(review_tier, 0) >= 2 THEN 1 ELSE 0 END) as tier2_scored, SUM(CASE WHEN COALESCE(review_tier, 0) >= 3 THEN 1 ELSE 0 END) as fully_approved FROM variants WHERE job_status='completed' AND raw_path IS NOT NULL GROUP BY sprite_id ORDER BY tier2_scored DESC, tier1_scored DESC, tier0_scored DESC """).fetchall() status = [dict(r) for r in status] # Group by tier in_vlm = [s for s in status if s['tier0_scored'] > 0 and s['tier1_scored'] == 0] in_haiku = [s for s in status if s['tier1_scored'] > 0 and s['tier2_scored'] == 0] in_opus = [s for s in status if s['tier2_scored'] > 0 and s['fully_approved'] == 0] timestamp = datetime.now().strftime("%H:%M:%S") check_count += 1 # Print summary print(f"\n[{timestamp}] Check #{check_count}") print(f" VLM (tier 0): {len(in_vlm):2d} sprites scoring") print(f" Haiku (tier 1): {len(in_haiku):2d} sprites scoring") print(f" Opus (tier 2): {len(in_opus):2d} sprites in review") # Check Opus progress if in_opus: print(f"\n Opus candidates in progress:") for s in in_opus: print(f" {s['sprite_id']}: {s['tier2_scored']:2d}/{s['total_variants']} in tier 2") # Check for 3-candidate ready sprites new_ready = [] for s in in_opus: candidates = conn.execute(""" SELECT id, seed, rating, notes, COALESCE(review_tier, 0) as review_tier FROM variants WHERE sprite_id = ? AND job_status = 'completed' AND raw_path IS NOT NULL AND COALESCE(review_tier, 0) >= 2 AND COALESCE(rating, 0) > 0 ORDER BY COALESCE(rating, 0) DESC LIMIT 3 """, (s['sprite_id'],)).fetchall() candidates = [dict(c) for c in candidates] if len(candidates) >= 3 and s['sprite_id'] not in opus_ready: opus_ready[s['sprite_id']] = candidates new_ready.append((s['sprite_id'], candidates)) # Report newly ready sprites if new_ready: print(f"\n{'='*80}") print(f"🎯 OPUS READY! {len(new_ready)} sprite(s) have 3 candidates for user final review:") print(f"{'='*80}") for sprite_id, candidates in new_ready: print(f"\n{sprite_id}:") for i, cand in enumerate(candidates, 1): print(f" {i}. variant #{cand['id']} (seed {cand['seed']}, rating {cand['rating']:.1f}/5)") if len(opus_ready) > 0: print(f"\n📊 Total ready for final review: {len(opus_ready)} sprite(s)") sys.stdout.flush() time.sleep(30) # Check every 30 seconds def main() -> None: parser = argparse.ArgumentParser( description="Sprite generation pipeline for Magic Civilization", prog="sprite-gen", ) parser.add_argument("--data-dir", type=str, help="Override game data directory path") parser.add_argument("--demo", action="store_true", help="Use minimal demo data (1 per domain)") sub = parser.add_subparsers(dest="command") # run — full pipeline orchestrator p = sub.add_parser("run", help="Run full pipeline: generate → rank → regen loop + GUI") p.add_argument("--port", type=int, default=5850, help="GUI server port (default: 5850)") p.add_argument("--category", type=str, help="Only process one category") p.add_argument("--variants", type=int, default=8, help="Variants per sprite per batch (default: 8)") p.add_argument("--max-attempts", type=int, default=15, dest="max_attempts", help="Max regen attempts per sprite (default: 15 = up to 120 total variants)") p.set_defaults(func=cmd_run) # start — launch GUI server only p = sub.add_parser("start", help="Launch review GUI server") p.add_argument("--port", type=int, default=5850, help="Server port (default: 5850)") p.set_defaults(func=cmd_review) # scan p = sub.add_parser("scan", help="Scan game data, populate sprite registry") p.add_argument("--sprite-type", dest="sprite_type", type=str, choices=["terrain", "biome_grid", "units", "buildings", "resources", "improvements", "spells", "ui"], help="Only scan this category (e.g. --sprite-type units)") p.set_defaults(func=cmd_scan) # status p = sub.add_parser("status", help="Show sprite counts by category and status") p.set_defaults(func=cmd_status) # generate — submit to queue (returns immediately) p = sub.add_parser("generate", help="Submit generation requests to model-boss queue (non-blocking)") p.add_argument("--category", type=str, help="Only generate one category") p.add_argument("--sprite", type=str, help="Generate single sprite by ID") p.add_argument("--variants", type=int, default=4, help="Variants per sprite (default: 4)") p.add_argument("--priority", choices=["low", "normal", "high"], default="normal") p.add_argument("--max", type=int, help="Max sprites to generate") p.add_argument("--dry-run", action="store_true", help="Show what would be generated") p.set_defaults(func=cmd_generate) # listen — collect results via pubsub p = sub.add_parser("listen", help="Collect pending generation results via Redis pubsub") p.set_defaults(func=cmd_listen) # rank p = sub.add_parser("rank", help="AI-rank variants using Haiku vision") p.add_argument("--category", type=str, help="Only rank one category") p.add_argument("--sprite", type=str, help="Rank single sprite by ID") p.set_defaults(func=cmd_rank) # install p = sub.add_parser("install", help="Copy approved sprites to game assets") p.add_argument("--category", type=str, help="Only install one category") p.add_argument("--dry-run", action="store_true") p.set_defaults(func=cmd_install) # approve p = sub.add_parser("approve", help="Approve variant → process → install → manifest") p.add_argument("variant", type=int, help="Variant ID to approve (e.g. 129)") p.add_argument("--alt", type=str, help="Install as alternate (e.g. --alt front)") p.set_defaults(func=cmd_approve) # reset p = sub.add_parser("reset", help="Reset sprite back to needed status") p.add_argument("--sprite", type=str, required=True, help="Sprite ID to reset") p.set_defaults(func=cmd_reset) # export p = sub.add_parser("export", help="Export registry as CSV or JSON") p.add_argument("--format", choices=["csv", "json"], default="csv") p.set_defaults(func=cmd_export) # test-prompt — rapid prompt iteration without touching the DB p = sub.add_parser("test-prompt", help="Generate test images with current prompt formula (no DB)") p.add_argument("--entity", type=str, default="spearmen", help="Unit entity_id to test") p.add_argument("--race", type=str, default="dwarves") p.add_argument("--gender", type=str, default="male") p.add_argument("--combat-type", type=str, default="melee") p.add_argument("--seeds", type=int, nargs="+", default=[42, 123, 777], help="Seeds to test") p.add_argument("--outdir", type=str, default="/tmp/prompt-test") p.set_defaults(func=cmd_test_prompt) # refresh-prompts — rebuild all stored prompts from current templates p = sub.add_parser("refresh-prompts", help="Rebuild all stored prompts with current templates") p.add_argument("--category", type=str, default="units") p.add_argument("--clear-scores", action="store_true", help="Also clear existing scores for re-ranking") p.set_defaults(func=cmd_refresh_prompts) # monitor — track tier progression and report when Opus has candidates p = sub.add_parser("monitor", help="Monitor tier progression; report when Opus has 3 candidates for user review") p.set_defaults(func=cmd_monitor) args = parser.parse_args() if args.command is None: args.port = 5850 cmd_review(args) else: args.func(args) if __name__ == "__main__": main()