Upload clashcr/cli.py with huggingface_hub

clashcr/cli.py

@@ -0,0 +1,340 @@
+"""ClashCR CLI entry point.
+
+Commands:
+    sync-cards      Sync card registry from APIs
+    record-battle   Record emulator gameplay
+    label-recording Interactive label editor for recordings
+    train-event-model  (placeholder) Train vision models
+    evaluate-recording Evaluate predictions vs labels
+    run             Live tracking run
+"""
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+import time
+from pathlib import Path
+
+import yaml
+
+from clashcr.utils.card_registry import CardRegistry
+from clashcr.core.evaluator import OfflineEvaluator
+
+logger = logging.getLogger("clashcr")
+
+
+def setup_logging(verbose: bool = False) -> None:
+    level = logging.DEBUG if verbose else logging.INFO
+    logging.basicConfig(
+        level=level,
+        format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+
+def load_config(path: str) -> dict:
+    with open(path, "r", encoding="utf-8") as f:
+        return yaml.safe_load(f)
+
+
+def cmd_sync_cards(args) -> int:
+    config = load_config(args.config) if args.config else {}
+    registry = CardRegistry(cache_path=config.get("card_registry_cache"))
+    token = config.get("clash_royale_api_token")
+    try:
+        meta = registry.sync(api_token=token, fail_on_stale=True)
+        print(f"Synced {meta.card_count} cards at {meta.timestamp}")
+        print(f"  RoyaleAPI source: {meta.source_card_count_royaleapi}")
+        print(f"  Official API source: {meta.source_card_count_official}")
+        for w in meta.warnings:
+            print(f"  WARNING: {w}")
+        registry.to_yaml(str(Path(registry.cache_path).with_suffix(".yaml")))
+    except RuntimeError as e:
+        print(f"SYNC FAILED (stale): {e}")
+        return 1
+    return 0
+
+
+def cmd_record_battle(args) -> int:
+    setup_logging(verbose=args.verbose)
+    config = load_config(args.config) if args.config else {}
+    output_dir = args.output
+    seconds = args.seconds
+    fps = args.fps
+
+    from clashcr.core.capture import EmulatorCapture
+    from clashcr.core.battle_gating import BattleGater
+    from clashcr.core.recorder import EvidenceRecorder, FrameMetadata
+
+    cap_config = config.get("capture", {})
+    capture = EmulatorCapture(
+        target_title_substring=cap_config.get("title_substring"),
+        target_class_name=cap_config.get("class_name"),
+    )
+
+    gater = BattleGater(
+        battle_threshold=config.get("battle_threshold", 0.6),
+        consecutive_frames=config.get("battle_consecutive_frames", 3),
+    )
+
+    recorder = EvidenceRecorder(
+        output_dir=output_dir,
+        save_frames=True,
+        max_fps=fps,
+    )
+
+    with recorder:
+        start = time.monotonic()
+        frame_idx = 0
+        while time.monotonic() - start < seconds:
+            frame = capture.capture(save_raw_path=None)
+            if frame is None:
+                time.sleep(0.1)
+                continue
+
+            battle = gater.process(frame)
+            info = capture.get_info()
+
+            meta = FrameMetadata(
+                timestamp=time.monotonic(),
+                frame_idx=frame_idx,
+                window_title=info.title if info else "",
+                window_hwnd=info.hwnd if info else 0,
+                window_width=info.client_width if info else 0,
+                window_height=info.client_height if info else 0,
+                dpi_scale=info.dpi_scale if info else 1.0,
+                battle_state=battle.state.name,
+                battle_confidence=battle.confidence,
+                mode_hint=battle.mode_hint,
+                arena_roi=battle.arena_roi,
+                own_roi=battle.own_roi,
+                opponent_roi=battle.opponent_roi,
+            )
+            recorder.record_frame(frame, meta)
+            frame_idx += 1
+            time.sleep(max(0, 1.0 / fps - 0.01))
+
+    print(f"Recording saved to {output_dir}")
+    return 0
+
+
+def cmd_label_recording(args) -> int:
+    print("Label recording: open labels.csv in the recording directory and add rows:")
+    print("  timestamp,frame_idx,side,card_key,confidence,manual_note,source")
+    print(f"  Recording: {args.recording}")
+    return 0
+
+
+def cmd_train_event_model(args) -> int:
+    print("Training placeholder: implement YOLO training on recorded+labeled data.")
+    print(f"Data dir: {args.data}")
+    return 0
+
+
+def cmd_evaluate_recording(args) -> int:
+    evaluator = OfflineEvaluator(timing_tolerance_seconds=args.tolerance)
+    result = evaluator.evaluate_recording_dir(args.recording)
+    print(f"Evaluation Results for {args.recording}")
+    print(f"  Precision: {result.precision:.3f}")
+    print(f"  Recall:    {result.recall:.3f}")
+    print(f"  F1:        {result.f1:.3f}")
+    print(f"  FP/min:    {result.false_positives_per_minute:.3f}")
+    print(f"  Missed:    {result.missed_events}")
+    print(f"  Mean timing error: {result.mean_timing_error_seconds:.2f}s")
+    print(f"  Predictions: {result.total_predictions}, Labels: {result.total_labels}, Correct: {result.correct}")
+    return 0
+
+
+def cmd_run(args) -> int:
+    setup_logging(verbose=args.verbose)
+    config = load_config(args.config) if args.config else {}
+
+    # Deferred imports to avoid loading cv2 for non-vision commands
+    from clashcr.core.capture import EmulatorCapture
+    from clashcr.core.battle_gating import BattleGater
+    from clashcr.core.event_detector import EventDetector
+    from clashcr.core.recorder import EvidenceRecorder, FrameMetadata
+    from clashcr.models.evidence_model import EvidenceModel
+    from clashcr.models.card_resolver import CardResolver
+    from clashcr.game.elixir_tracker import ElixirTracker
+    from clashcr.game.deck_tracker import DeckTracker
+
+    # Initialize components
+    registry = CardRegistry(cache_path=config.get("card_registry_cache"))
+    if not registry.load():
+        logger.error("Card registry not loaded. Run sync-cards first.")
+        return 1
+
+    cap_config = config.get("capture", {})
+    capture = EmulatorCapture(
+        target_title_substring=cap_config.get("title_substring"),
+        target_class_name=cap_config.get("class_name"),
+    )
+
+    gater = BattleGater(
+        battle_threshold=config.get("battle_threshold", 0.6),
+        consecutive_frames=config.get("battle_consecutive_frames", 3),
+    )
+
+    model_config = config.get("model", {})
+    evidence_model = EvidenceModel(
+        model_path=model_config.get("yolo_path"),
+        conf_threshold=model_config.get("conf_threshold", 0.5),
+        spell_enabled=model_config.get("spell_enabled", True),
+        device=model_config.get("device", "cpu"),
+    )
+
+    resolver = CardResolver(registry=registry)
+    elixir = ElixirTracker(registry=registry)
+    deck = DeckTracker()
+
+    detector = EventDetector(
+        diff_threshold=config.get("diff_threshold", 25),
+        min_event_area=config.get("min_event_area", 200),
+        debug_dir=args.debug_dir if args.debug_dir else None,
+    )
+
+    recorder = None
+    if args.record_dir:
+        recorder = EvidenceRecorder(output_dir=args.record_dir, save_frames=args.raw, max_fps=8.0)
+        recorder.open()
+
+    logger.info("Starting live run...")
+    frame_idx = 0
+    try:
+        while True:
+            frame = capture.capture(save_raw_path=args.debug_frame if args.debug_frame and frame_idx % 30 == 0 else None)
+            if frame is None:
+                time.sleep(0.05)
+                continue
+
+            battle = gater.process(frame)
+
+            if recorder:
+                info = capture.get_info()
+                meta = FrameMetadata(
+                    timestamp=time.monotonic(),
+                    frame_idx=frame_idx,
+                    window_title=info.title if info else "",
+                    window_hwnd=info.hwnd if info else 0,
+                    window_width=info.client_width if info else 0,
+                    window_height=info.client_height if info else 0,
+                    dpi_scale=info.dpi_scale if info else 1.0,
+                    battle_state=battle.state.name,
+                    battle_confidence=battle.confidence,
+                    mode_hint=battle.mode_hint,
+                    arena_roi=battle.arena_roi,
+                    own_roi=battle.own_roi,
+                    opponent_roi=battle.opponent_roi,
+                )
+                recorder.record_frame(frame, meta)
+
+            if not gater.is_battle():
+                frame_idx += 1
+                time.sleep(0.05)
+                continue
+
+            # Detect events
+            events = detector.process(frame, battle)
+            for ev in events:
+                if ev.suppressed or ev.side != "opponent":
+                    continue
+
+                # Extract opponent ROI crop
+                x, y, w, h = ev.bbox
+                crop = frame[y:y+h, x:x+w]
+                if crop.size == 0:
+                    continue
+
+                bundle = evidence_model.process(crop, side="opponent")
+                resolved = resolver.resolve(bundle)
+
+                if resolved.is_ambiguous:
+                    logger.info("[AMBIGUOUS] %s (reason: %s)", resolved.card_key, resolved.ambiguity_reason)
+                    if recorder:
+                        recorder.add_prediction(ev.timestamp, frame_idx, "opponent",
+                                                resolved.card_key, resolved.confidence,
+                                                str(bundle.possible_cards), resolved.reasoning)
+                    continue
+
+                if resolver.deduplicate(resolved, cooldown_seconds=2.0):
+                    continue
+
+                elixir.deduct(resolved.card_key, resolved.confidence)
+                deck.register_play(resolved.card_key, resolved.confidence)
+
+                logger.info("[EVENT] %s (conf=%.2f, elixir=%.1f, deck=%s)",
+                            resolved.card_key, resolved.confidence,
+                            elixir.get_state().current, deck.state.deck)
+
+                if recorder:
+                    recorder.add_prediction(ev.timestamp, frame_idx, "opponent",
+                                            resolved.card_key, resolved.confidence,
+                                            str(bundle.possible_cards), resolved.reasoning)
+
+            frame_idx += 1
+            time.sleep(0.05)
+    except KeyboardInterrupt:
+        logger.info("Stopping live run.")
+    finally:
+        if recorder:
+            recorder.close()
+
+    return 0
+
+
+def main(argv: Optional[list] = None) -> int:
+    parser = argparse.ArgumentParser(prog="clashcr", description="Clash Royale opponent card tracker")
+    parser.add_argument("--verbose", "-v", action="store_true", help="Verbose logging")
+    sub = parser.add_subparsers(dest="command")
+
+    p_sync = sub.add_parser("sync-cards", help="Sync card registry")
+    p_sync.add_argument("--config", default="config.yaml", help="Config file")
+
+    p_rec = sub.add_parser("record-battle", help="Record battle footage")
+    p_rec.add_argument("--config", default="config.yaml")
+    p_rec.add_argument("--output", required=True, help="Output directory")
+    p_rec.add_argument("--seconds", type=int, default=180, help="Recording duration")
+    p_rec.add_argument("--fps", type=float, default=8.0, help="Capture FPS")
+
+    p_label = sub.add_parser("label-recording", help="Label a recording")
+    p_label.add_argument("--recording", required=True)
+
+    p_train = sub.add_parser("train-event-model", help="Train event detection model")
+    p_train.add_argument("--config", default="config.yaml")
+    p_train.add_argument("--data", required=True)
+
+    p_eval = sub.add_parser("evaluate-recording", help="Evaluate recording predictions")
+    p_eval.add_argument("--config", default="config.yaml")
+    p_eval.add_argument("--recording", required=True)
+    p_eval.add_argument("--labels", default=None)
+    p_eval.add_argument("--tolerance", type=float, default=3.0, help="Timing tolerance in seconds")
+
+    p_run = sub.add_parser("run", help="Live tracking run")
+    p_run.add_argument("--config", default="config.yaml")
+    p_run.add_argument("--raw", action="store_true", help="Save raw frames")
+    p_run.add_argument("--debug-frame", default=None, help="Save debug frame path")
+    p_run.add_argument("--debug-dir", default=None, help="Save debug event crops")
+    p_run.add_argument("--record-dir", default=None, help="Record session to directory")
+
+    args = parser.parse_args(argv)
+    if not args.command:
+        parser.print_help()
+        return 1
+
+    setup_logging(verbose=args.verbose)
+
+    handlers = {
+        "sync-cards": cmd_sync_cards,
+        "record-battle": cmd_record_battle,
+        "label-recording": cmd_label_recording,
+        "train-event-model": cmd_train_event_model,
+        "evaluate-recording": cmd_evaluate_recording,
+        "run": cmd_run,
+    }
+    return handlers[args.command](args)
+
+
+if __name__ == "__main__":
+    sys.exit(main())