from __future__ import annotations import argparse import logging from pathlib import Path from .coordinates import extract_coordinate_crops, score_coordinates_for_sheet from .detector_cv import detect_sheet, draw_overlay from .export_yolo import export_candidates_to_yolo from .inventory import discover_original_assets, download_assets, read_manifest_csv, write_manifest_csv from .report import build_report from .train_yolo import train_yolo from .utils import load_yaml, setup_logging LOG = logging.getLogger(__name__) def cmd_inventory(args): cfg = load_yaml(args.config) base_url = args.base_url or cfg["source"]["base_url"] assets = discover_original_assets(base_url=base_url, include_100k=args.include_100k) if args.limit: assets = assets[: args.limit] write_manifest_csv(assets, args.out) def cmd_download(args): assets = read_manifest_csv(args.manifest) selected = download_assets(assets, args.out_dir, limit=args.limit, overwrite=args.overwrite) write_manifest_csv(selected, args.out_manifest) def cmd_detect(args): cfg = load_yaml(args.config) detect_sheet(args.map, args.tif, args.sheet_id, cfg, args.out_dir) def cmd_overlay(args): draw_overlay(args.tif, args.candidates, args.out) def cmd_score_coords(args): cfg = load_yaml(args.config) score_coordinates_for_sheet( coord_csv=args.coordinates, candidates_csv=args.candidates, map_path=args.map, tif_path=args.tif, sheet_id=args.sheet_id, cfg=cfg, out_dir=args.out_dir, coord_crs=args.coord_crs, ) def cmd_crops(args): extract_coordinate_crops( coord_scores_csv=args.scores, map_path=args.map, tif_path=args.tif, out_dir=args.out_dir, crop_size=args.crop_size, ) def cmd_export_yolo(args): cfg = load_yaml(args.config) export_candidates_to_yolo( tif_path=args.tif, candidates_csv=args.candidates, out_dir=args.out_dir, cfg=cfg, sheet_id=args.sheet_id, tile_size=args.tile_size, overlap=args.overlap, val_fraction=args.val_fraction, ) def cmd_train_yolo(args): train_yolo(args.data_yaml, model=args.model, imgsz=args.imgsz, epochs=args.epochs, batch=args.batch, device=args.device) def cmd_report(args): build_report([Path(p) for p in args.candidates], [Path(p) for p in args.overlays], args.out) def build_parser(): p = argparse.ArgumentParser(prog="bgtopo-bluebox", description="BGtopoVJ blue rectangle/square PoC pipeline") p.add_argument("--verbose", action="store_true") sub = p.add_subparsers(dest="cmd", required=True) s = sub.add_parser("inventory", help="Crawl original raster directory and create manifest") s.add_argument("--config", default="configs/blue_detector.yaml") s.add_argument("--base-url", default=None) s.add_argument("--out", default="data/manifest.csv") s.add_argument("--limit", type=int, default=None) s.add_argument("--include-100k", action="store_true") s.set_defaults(func=cmd_inventory) s = sub.add_parser("download", help="Download .map/.tif pairs from manifest") s.add_argument("--manifest", default="data/manifest.csv") s.add_argument("--out-dir", default="data/raw") s.add_argument("--out-manifest", default="data/manifest_downloaded.csv") s.add_argument("--limit", type=int, default=2) s.add_argument("--overwrite", action="store_true") s.set_defaults(func=cmd_download) s = sub.add_parser("detect", help="Detect blue rectangle/square candidates on one sheet") s.add_argument("--config", default="configs/blue_detector.yaml") s.add_argument("--sheet-id", required=True) s.add_argument("--map", default=None) s.add_argument("--tif", required=True) s.add_argument("--out-dir", default="data/interim/candidates") s.set_defaults(func=cmd_detect) s = sub.add_parser("overlay", help="Draw candidate overlay for manual QA") s.add_argument("--tif", required=True) s.add_argument("--candidates", required=True) s.add_argument("--out", required=True) s.set_defaults(func=cmd_overlay) s = sub.add_parser("score-coords", help="Score coordinate CSV against candidates from one sheet") s.add_argument("--config", default="configs/blue_detector.yaml") s.add_argument("--sheet-id", required=True) s.add_argument("--coordinates", required=True) s.add_argument("--candidates", required=True) s.add_argument("--map", default=None) s.add_argument("--tif", required=True) s.add_argument("--out-dir", default="data/interim/coordinate_scores") s.add_argument("--coord-crs", default="EPSG:4326") s.set_defaults(func=cmd_score_coords) s = sub.add_parser("crops", help="Extract review crops around scored coordinates") s.add_argument("--scores", required=True) s.add_argument("--map", default=None) s.add_argument("--tif", required=True) s.add_argument("--out-dir", default="data/interim/crops") s.add_argument("--crop-size", type=int, default=256) s.set_defaults(func=cmd_crops) s = sub.add_parser("export-yolo", help="Export weak candidates to YOLO dataset format") s.add_argument("--config", default="configs/blue_detector.yaml") s.add_argument("--sheet-id", required=True) s.add_argument("--tif", required=True) s.add_argument("--candidates", required=True) s.add_argument("--out-dir", default="data/yolo/bluebox") s.add_argument("--tile-size", type=int, default=1024) s.add_argument("--overlap", type=int, default=128) s.add_argument("--val-fraction", type=float, default=0.20) s.set_defaults(func=cmd_export_yolo) s = sub.add_parser("train-yolo", help="Train YOLO on exported dataset") s.add_argument("--data-yaml", required=True) s.add_argument("--model", default="yolov8s.pt") s.add_argument("--imgsz", type=int, default=1024) s.add_argument("--epochs", type=int, default=80) s.add_argument("--batch", type=int, default=4) s.add_argument("--device", default="0") s.set_defaults(func=cmd_train_yolo) s = sub.add_parser("report", help="Build HTML QA report") s.add_argument("--candidates", nargs="+", required=True) s.add_argument("--overlays", nargs="*", default=[]) s.add_argument("--out", default="reports/poc_report.html") s.set_defaults(func=cmd_report) return p def main(): parser = build_parser() args = parser.parse_args() setup_logging(args.verbose) args.func(args) if __name__ == "__main__": main()