File size: 2,770 Bytes
9d09c45 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 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 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | """Validate generated CityFlow routes and print summary statistics."""
from __future__ import annotations
import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any
from .utils import build_road_index, build_roadlink_index, summarize_route_validation
def _load_json(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
def _resolve_from_config(config_path: Path) -> tuple[Path, Path]:
config = _load_json(config_path)
base_dir = Path(config["dir"])
roadnet_path = (base_dir / config["roadnetFile"]).resolve()
flow_path = (base_dir / config["flowFile"]).resolve()
return roadnet_path, flow_path
def _print_summary(summary: dict[str, Any]) -> None:
print(f"total routes: {summary['total_routes']}")
print(f"valid routes: {summary['valid_routes']}")
print(f"invalid routes: {summary['invalid_routes']}")
if summary["top_failure_reasons"]:
formatted = ", ".join(
f"{reason}={count}" for reason, count in summary["top_failure_reasons"]
)
print(f"top failure reasons: {formatted}")
else:
print("top failure reasons: none")
def main() -> int:
parser = argparse.ArgumentParser(description="Validate CityFlow flow routes.")
parser.add_argument(
"--config",
type=Path,
default=None,
help="Path to scenario config.json.",
)
parser.add_argument(
"--roadnet",
type=Path,
default=None,
help="Path to roadnet.json (required if --config not provided).",
)
parser.add_argument(
"--flow",
type=Path,
default=None,
help="Path to flow.json (required if --config not provided).",
)
args = parser.parse_args()
if args.config is not None:
roadnet_path, flow_path = _resolve_from_config(args.config.resolve())
else:
if args.roadnet is None or args.flow is None:
raise ValueError("Provide --config OR both --roadnet and --flow.")
roadnet_path = args.roadnet.resolve()
flow_path = args.flow.resolve()
if not roadnet_path.exists():
raise FileNotFoundError(os.fspath(roadnet_path))
if not flow_path.exists():
raise FileNotFoundError(os.fspath(flow_path))
roadnet = _load_json(roadnet_path)
flows = _load_json(flow_path)
roads_by_id = build_road_index(roadnet)
roadlinks_by_intersection = build_roadlink_index(roadnet)
summary = summarize_route_validation(
flow_entries=flows,
roads_by_id=roads_by_id,
roadlinks_by_intersection=roadlinks_by_intersection,
)
_print_summary(summary)
return 1 if summary["invalid_routes"] > 0 else 0
if __name__ == "__main__":
sys.exit(main())
|