Upload folder using huggingface_hub

data/generators/README.md

@@ -0,0 +1,23 @@
+# data/generators
+
+Synthetic dataset generation code for CityFlow cities and scenarios.
+
+## Main files
+
+- [generate_dataset.py](/Users/aditya/Developer/traffic-llm/data/generators/generate_dataset.py)
+  Primary dataset build script.
+- [city_generator.py](/Users/aditya/Developer/traffic-llm/data/generators/city_generator.py)
+  High-level city assembly.
+- [roadnet_generator.py](/Users/aditya/Developer/traffic-llm/data/generators/roadnet_generator.py)
+  Road network generation.
+- [district_generator.py](/Users/aditya/Developer/traffic-llm/data/generators/district_generator.py)
+  District assignments and relationships.
+- [flow_generator.py](/Users/aditya/Developer/traffic-llm/data/generators/flow_generator.py)
+  Vehicle flow generation.
+- [scenario_generator.py](/Users/aditya/Developer/traffic-llm/data/generators/scenario_generator.py)
+  Scenario-specific flow and config generation.
+
+## Notes
+
+- This folder is for offline dataset creation.
+- The current training pipeline consumes the generated files directly and does not regenerate them.

data/generators/__init__.py

@@ -0,0 +1,6 @@
+"""Synthetic CityFlow dataset generation package."""
+
+from .city_generator import CityGenerator
+from .schemas import DatasetGenerationConfig
+
+__all__ = ["CityGenerator", "DatasetGenerationConfig"]

data/generators/city_generator.py

@@ -0,0 +1,341 @@
+"""City-level orchestration: topology, districts, scenarios, flows, configs, validation."""
+
+from __future__ import annotations
+
+import random
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any
+
+from .config_generator import ConfigGenerator
+from .district_generator import DistrictGenerator
+from .flow_generator import FlowGenerator
+from .roadnet_generator import RoadnetGenerator
+from .scenario_generator import ScenarioGenerator
+from .schemas import DatasetGenerationConfig, TopologyType
+from .utils import (
+    build_road_index,
+    build_roadlink_index,
+    clamp,
+    compute_scenario_diagnostics,
+    ensure_dir,
+    summarize_route_validation,
+    validate_district_contiguity,
+    validate_district_exit_capacity,
+    validate_inter_district_connectivity,
+    validate_unique_ids,
+    write_json,
+)
+
+
+class CityGenerator:
+    """Generate one or many synthetic cities with scenario-specific CityFlow files."""
+
+    def __init__(self) -> None:
+        self.roadnet_generator = RoadnetGenerator()
+        self.district_generator = DistrictGenerator()
+        self.scenario_generator = ScenarioGenerator()
+        self.flow_generator = FlowGenerator()
+        self.config_generator = ConfigGenerator()
+
+    def generate_dataset(self, config: DatasetGenerationConfig) -> None:
+        ensure_dir(config.output_dir)
+        failures: list[tuple[str, str]] = []
+        for idx in range(config.num_cities):
+            city_id = f"city_{idx + 1:04d}"
+            city_seed = config.seed + idx * 10_003
+            try:
+                self.generate_city(
+                    city_id=city_id,
+                    output_dir=config.output_dir / city_id,
+                    config=config,
+                    city_seed=city_seed,
+                )
+            except Exception as exc:
+                failures.append((city_id, str(exc)))
+                if config.fail_fast:
+                    raise
+        if failures:
+            details = "; ".join(
+                f"{city}: {message}" for city, message in failures[:5]
+            )
+            raise RuntimeError(
+                f"Dataset generation failed for {len(failures)} city/cities. {details}"
+            )
+
+    def generate_city(
+        self,
+        city_id: str,
+        output_dir: Path,
+        config: DatasetGenerationConfig,
+        city_seed: int,
+    ) -> None:
+        ensure_dir(output_dir)
+        rng = random.Random(city_seed)
+        topology_pool: list[TopologyType] = list(config.topologies)
+        if not topology_pool:
+            raise ValueError("No topology families provided in configuration.")
+
+        attempts_per_topology = 10
+        ordered_topologies = topology_pool.copy()
+        rng.shuffle(ordered_topologies)
+        max_attempts = attempts_per_topology * len(ordered_topologies)
+        attempt_count = 0
+        city_graph = None
+        district_data = None
+        last_error: Exception | None = None
+        for topology in ordered_topologies:
+            for topology_attempt in range(attempts_per_topology):
+                attempt_count += 1
+                attempt_seed = city_seed + ((attempt_count - 1) * 1009)
+                target_intersections = clamp(
+                    rng.randint(
+                        config.min_districts * config.min_intersections_per_district,
+                        config.max_districts * config.max_intersections_per_district,
+                    ),
+                    low=config.min_districts * config.min_intersections_per_district,
+                    high=config.max_districts * config.max_intersections_per_district + 36,
+                )
+                try:
+                    city_graph = self.roadnet_generator.generate(
+                        city_id=city_id,
+                        seed=attempt_seed,
+                        topology=topology,
+                        target_intersections=target_intersections,
+                        ring_diagonal_keep_prob=config.ring_diagonal_keep_prob,
+                        ring_max_diagonal_fraction=config.ring_max_diagonal_fraction,
+                    )
+                    validate_unique_ids(city_graph)
+                    max_districts = max(
+                        config.min_districts,
+                        min(
+                            config.max_districts,
+                            max(
+                                2,
+                                len(
+                                    [
+                                        nid
+                                        for nid in city_graph.intersections
+                                        if nid not in city_graph.gateway_intersections
+                                    ]
+                                )
+                                // max(1, config.min_intersections_per_district),
+                            ),
+                        ),
+                    )
+                    if topology == "ring_road":
+                        max_districts = min(max_districts, max(6, target_intersections // 10))
+                        max_districts = max(config.min_districts, max_districts)
+                    num_districts = rng.randint(config.min_districts, max_districts)
+
+                    district_data = self.district_generator.generate(
+                        city_graph=city_graph,
+                        num_districts=num_districts,
+                        seed=attempt_seed + 17,
+                    )
+                    validate_district_contiguity(city_graph, district_data)
+                    validate_inter_district_connectivity(district_data)
+                    min_exit_roads = 2 if topology == "ring_road" else 3
+                    min_entry_roads = 2 if topology == "ring_road" else 3
+                    min_neighbor_districts = 1 if topology == "ring_road" else 2
+                    validate_district_exit_capacity(
+                        district_data=district_data,
+                        min_exit_roads=min_exit_roads,
+                        min_entry_roads=min_entry_roads,
+                        min_neighbor_districts=min_neighbor_districts,
+                    )
+                    print(
+                        f"[INFO] {city_id} attempt={attempt_count} "
+                        f"topology={topology} topology_try={topology_attempt + 1}/{attempts_per_topology} "
+                        "generated successfully"
+                    )
+                    break
+                except Exception as exc:
+                    message = str(exc)
+                    print(
+                        f"[WARN] {city_id} attempt={attempt_count} "
+                        f"topology={topology} topology_try={topology_attempt + 1}/{attempts_per_topology} "
+                        f"failed: {message}"
+                    )
+                    last_error = exc
+                    city_graph = None
+                    district_data = None
+                    continue
+
+            if city_graph is not None and district_data is not None:
+                break
+
+        if city_graph is None or district_data is None:
+            raise ValueError(
+                f"Unable to produce a structurally valid city after {max_attempts} attempts: {last_error}"
+            )
+
+        roadnet_path = output_dir / "roadnet.json"
+        write_json(roadnet_path, city_graph.roadnet)
+
+        district_map = {
+            "intersection_to_district": district_data.intersection_to_district,
+            "district_neighbors": district_data.district_neighbors,
+            "boundary_intersections": district_data.boundary_intersections,
+            "gateway_intersections": sorted(city_graph.gateway_intersections),
+            "gateway_roads": sorted(city_graph.gateway_roads),
+            "districts": [
+                {
+                    "id": d.id,
+                    "type": d.district_type,
+                    "intersections": d.intersections,
+                    "neighbors": d.neighbors,
+                    "boundary_intersections": d.boundary_intersections,
+                    "entry_roads": d.entry_roads,
+                    "exit_roads": d.exit_roads,
+                }
+                for d in district_data.districts.values()
+            ],
+        }
+        write_json(output_dir / "district_map.json", district_map)
+
+        metadata = self._city_metadata(
+            city_id=city_id,
+            topology=topology,
+            city_seed=city_seed,
+            city_graph=city_graph,
+            district_data=district_data,
+            config=config,
+        )
+        write_json(output_dir / "metadata.json", metadata)
+
+        print(f"[INFO] {city_id} generated: topology={topology}, districts={len(district_data.districts)}")
+
+        scenario_plans = self.scenario_generator.generate(
+            city_graph=city_graph,
+            district_data=district_data,
+            scenario_names=config.scenarios,
+            base_seed=city_seed + 1000,
+            config=config,
+        )
+        self._generate_scenarios(
+            output_dir=output_dir,
+            city_graph=city_graph,
+            district_data=district_data,
+            scenario_plans=scenario_plans,
+            config=config,
+            roadnet_path=roadnet_path,
+        )
+
+    def _generate_scenarios(
+        self,
+        output_dir: Path,
+        city_graph: Any,
+        district_data: Any,
+        scenario_plans: dict[str, Any],
+        config: DatasetGenerationConfig,
+        roadnet_path: Path,
+    ) -> None:
+        roads_by_id = build_road_index(city_graph.roadnet)
+        roadlinks_by_intersection = build_roadlink_index(city_graph.roadnet)
+        scenarios_dir = output_dir / "scenarios"
+        ensure_dir(scenarios_dir)
+
+        for scenario_name, plan in scenario_plans.items():
+            scenario_dir = scenarios_dir / scenario_name
+            ensure_dir(scenario_dir)
+            flows = self.flow_generator.generate(
+                city_graph=city_graph,
+                district_data=district_data,
+                scenario=plan,
+                simulation_steps=config.simulation_steps,
+            )
+            validation_summary = summarize_route_validation(
+                flow_entries=flows,
+                roads_by_id=roads_by_id,
+                roadlinks_by_intersection=roadlinks_by_intersection,
+            )
+            if validation_summary["invalid_routes"] > 0:
+                reasons = ", ".join(
+                    f"{reason}={count}"
+                    for reason, count in validation_summary["top_failure_reasons"]
+                )
+                raise ValueError(
+                    f"{scenario_name} contains invalid routes after generation: {reasons}"
+                )
+            write_json(scenario_dir / "flow.json", flows)
+            diagnostics = compute_scenario_diagnostics(
+                flow_entries=flows,
+                city_graph=city_graph,
+                district_data=district_data,
+            )
+
+            config_payload = self.config_generator.generate(
+                simulation_steps=config.simulation_steps,
+                interval=config.interval,
+                seed=plan.seed,
+                save_replay=config.save_replay,
+                roadnet_file=roadnet_path,
+                flow_file=scenario_dir / "flow.json",
+                scenario_dir=scenario_dir,
+            )
+            write_json(scenario_dir / "config.json", config_payload)
+            write_json(
+                scenario_dir / "scenario_metadata.json",
+                {
+                    "name": scenario_name,
+                    "intensity": plan.intensity,
+                    "seed": plan.seed,
+                    "trip_multiplier": plan.trip_multiplier,
+                    "trip_mix": asdict(plan.trip_mix),
+                    "blocked_roads": sorted(plan.blocked_roads),
+                    "penalized_roads": plan.penalized_roads,
+                    "event_district": plan.event_district,
+                    "overload_district": plan.overload_district,
+                    "diagnostics": diagnostics,
+                    "details": plan.metadata,
+                },
+            )
+
+    def _city_metadata(
+        self,
+        city_id: str,
+        topology: TopologyType,
+        city_seed: int,
+        city_graph: Any,
+        district_data: Any,
+        config: DatasetGenerationConfig,
+    ) -> dict[str, Any]:
+        total_lanes = sum(
+            road.num_lanes for road in city_graph.directed_roads.values()
+        )
+        district_types = {
+            did: district.district_type
+            for did, district in district_data.districts.items()
+        }
+        return {
+            "city_id": city_id,
+            "topology": topology,
+            "seed": city_seed,
+            "counts": {
+                "intersections": len(city_graph.intersections),
+                "roads": len(city_graph.directed_roads),
+                "lanes": total_lanes,
+                "districts": len(district_data.districts),
+            },
+            "district_types": district_types,
+            "district_adjacency_graph": district_data.district_neighbors,
+            "inter_district_connector_roads": district_data.inter_district_roads,
+            "arterial_roads": sorted(city_graph.arterial_roads),
+            "gateway_intersections": sorted(city_graph.gateway_intersections),
+            "gateway_connector_roads": sorted(city_graph.gateway_roads),
+            "generation_parameters": {
+                "min_districts": config.min_districts,
+                "max_districts": config.max_districts,
+                "min_intersections_per_district": config.min_intersections_per_district,
+                "max_intersections_per_district": config.max_intersections_per_district,
+                "simulation_steps": config.simulation_steps,
+                "interval": config.interval,
+                "intensity_levels": config.intensity_levels,
+                "global_demand_multiplier": config.global_demand_multiplier,
+                "intensity_distribution": config.intensity_distribution,
+                "scenario_demand_multipliers": config.scenario_demand_multipliers,
+                "ring_diagonal_keep_prob": config.ring_diagonal_keep_prob,
+                "ring_max_diagonal_fraction": config.ring_max_diagonal_fraction,
+            },
+        }

data/generators/config_generator.py

@@ -0,0 +1,50 @@
+"""CityFlow config generation for per-scenario simulation runs."""
+
+from __future__ import annotations
+
+from pathlib import Path
+import os
+from typing import Any
+
+
+class ConfigGenerator:
+    """Build CityFlow-compatible config.json payloads."""
+
+    def generate(
+        self,
+        simulation_steps: int,
+        interval: float,
+        seed: int,
+        save_replay: bool,
+        roadnet_file: Path,
+        flow_file: Path,
+        scenario_dir: Path,
+    ) -> dict[str, Any]:
+        # CityFlow expects a small signed integer for seed.
+        safe_seed = int(seed) & 0x7FFFFFFF
+        # Use absolute paths so CityFlow can resolve files regardless of working dir.
+        roadnet_path = roadnet_file.resolve()
+        flow_path = flow_file.resolve()
+        scenario_path = scenario_dir.resolve()
+        base_dir = roadnet_path.parent
+        roadnet_rel = os.path.relpath(roadnet_path, base_dir)
+        flow_rel = os.path.relpath(flow_path, base_dir)
+        flow_rel_dir = Path(flow_rel).parent
+        roadnet_log_rel = str(flow_rel_dir / "roadnetLogFile.json")
+        replay_log_rel = str(flow_rel_dir / "replay.txt")
+        dir_str = str(base_dir)
+        if not dir_str.endswith(os.sep):
+            dir_str = dir_str + os.sep
+        return {
+            "interval": interval,
+            "seed": safe_seed,
+            "dir": dir_str,
+            "roadnetFile": roadnet_rel,
+            "flowFile": flow_rel,
+            "rlTrafficLight": True,
+            "laneChange": False,
+            "saveReplay": save_replay,
+            "roadnetLogFile": roadnet_log_rel,
+            "replayLogFile": replay_log_rel,
+            "step": simulation_steps,
+        }

data/generators/district_generator.py

@@ -0,0 +1,367 @@
+"""District partitioning and district-level metadata generation."""
+
+from __future__ import annotations
+
+import random
+from collections import Counter, defaultdict, deque
+
+from .schemas import CityGraph, DistrictData, DistrictRecord, DistrictType
+from .utils import connected_components, euclidean
+
+
+class DistrictGenerator:
+    """Generate contiguous district partitions over the city intersection graph."""
+
+    DISTRICT_TYPE_WEIGHTS: dict[DistrictType, float] = {
+        "residential": 0.35,
+        "commercial": 0.25,
+        "industrial": 0.20,
+        "mixed": 0.20,
+    }
+
+    def generate(
+        self,
+        city_graph: CityGraph,
+        num_districts: int,
+        seed: int,
+    ) -> DistrictData:
+        rng = random.Random(seed)
+        node_ids = sorted(
+            n
+            for n in city_graph.intersections.keys()
+            if n not in city_graph.gateway_intersections
+        )
+        if len(node_ids) < 2:
+            raise ValueError("Insufficient non-gateway intersections for districting.")
+        if num_districts >= len(node_ids):
+            num_districts = max(2, len(node_ids) // 2)
+
+        local_coords = {nid: city_graph.intersections[nid] for nid in node_ids}
+        seeds = self._farthest_seeds(local_coords, num_districts, rng)
+        assignment = self._grow_contiguous_regions(
+            local_coords=local_coords,
+            adjacency=city_graph.adjacency,
+            seeds=seeds,
+            rng=rng,
+        )
+        assignment = self._enforce_contiguity(
+            assignment=assignment,
+            adjacency=city_graph.adjacency,
+            coords=local_coords,
+        )
+        assignment = self._fill_empty_districts(
+            assignment=assignment,
+            node_ids=node_ids,
+            adjacency=city_graph.adjacency,
+            district_ids=list(seeds.keys()),
+        )
+
+        district_neighbors: dict[str, set[str]] = {
+            did: set() for did in seeds.keys()
+        }
+        boundary: set[str] = set()
+        for a, neighbors in city_graph.adjacency.items():
+            if a not in assignment:
+                continue
+            da = assignment[a]
+            for b in neighbors:
+                if b not in assignment:
+                    continue
+                db = assignment[b]
+                if da != db:
+                    district_neighbors[da].add(db)
+                    boundary.add(a)
+
+        entry_roads: dict[str, list[str]] = defaultdict(list)
+        exit_roads: dict[str, list[str]] = defaultdict(list)
+        inter_district_roads: set[str] = set()
+        for road in city_graph.directed_roads.values():
+            if (
+                road.start_intersection not in assignment
+                or road.end_intersection not in assignment
+            ):
+                continue
+            ds = assignment[road.start_intersection]
+            de = assignment[road.end_intersection]
+            if ds != de:
+                inter_district_roads.add(road.id)
+                exit_roads[ds].append(road.id)
+                entry_roads[de].append(road.id)
+        city_graph.inter_district_roads = inter_district_roads
+
+        district_records: dict[str, DistrictRecord] = {}
+        type_values = list(self.DISTRICT_TYPE_WEIGHTS.keys())
+        type_weights = list(self.DISTRICT_TYPE_WEIGHTS.values())
+        for district_id in seeds:
+            members = sorted([n for n, d in assignment.items() if d == district_id])
+            d_boundary = sorted([n for n in members if n in boundary])
+            district_type = rng.choices(type_values, weights=type_weights, k=1)[0]
+            district_records[district_id] = DistrictRecord(
+                id=district_id,
+                district_type=district_type,
+                intersections=members,
+                neighbors=sorted(district_neighbors[district_id]),
+                boundary_intersections=d_boundary,
+                entry_roads=sorted(set(entry_roads[district_id])),
+                exit_roads=sorted(set(exit_roads[district_id])),
+            )
+
+        return DistrictData(
+            intersection_to_district=assignment,
+            districts=district_records,
+            district_neighbors={
+                k: sorted(v) for k, v in district_neighbors.items()
+            },
+            boundary_intersections=sorted(boundary),
+            inter_district_roads=sorted(inter_district_roads),
+        )
+
+    def _farthest_seeds(
+        self,
+        coords: dict[str, tuple[float, float]],
+        num_districts: int,
+        rng: random.Random,
+    ) -> dict[str, str]:
+        nodes = sorted(coords.keys())
+        first = rng.choice(nodes)
+        selected = [first]
+
+        while len(selected) < num_districts:
+            best_node = None
+            best_dist = -1.0
+            for node in nodes:
+                if node in selected:
+                    continue
+                nearest = min(
+                    euclidean(coords[node], coords[s]) for s in selected
+                )
+                if nearest > best_dist:
+                    best_dist = nearest
+                    best_node = node
+            if best_node is None:
+                break
+            selected.append(best_node)
+        return {f"d_{idx:02d}": node for idx, node in enumerate(selected)}
+
+    def _assign_nearest(
+        self,
+        coords: dict[str, tuple[float, float]],
+        seeds: dict[str, str],
+    ) -> dict[str, str]:
+        assignment: dict[str, str] = {}
+        for node, point in coords.items():
+            district = min(
+                seeds.keys(),
+                key=lambda did: euclidean(point, coords[seeds[did]]),
+            )
+            assignment[node] = district
+        return assignment
+
+    def _grow_contiguous_regions(
+        self,
+        local_coords: dict[str, tuple[float, float]],
+        adjacency: dict[str, set[str]],
+        seeds: dict[str, str],
+        rng: random.Random,
+    ) -> dict[str, str]:
+        districts = list(seeds.keys())
+        district_sizes = {district_id: 1 for district_id in districts}
+        assignment: dict[str, str] = {seed_node: district_id for district_id, seed_node in seeds.items()}
+        frontiers: dict[str, deque[str]] = {
+            district_id: deque([seed_node]) for district_id, seed_node in seeds.items()
+        }
+        remaining = set(local_coords.keys()) - set(assignment.keys())
+
+        if not remaining:
+            return assignment
+
+        target_avg = max(1, len(local_coords) // len(districts))
+        target_limits = {
+            district_id: target_avg + 2 for district_id in districts
+        }
+        overcap = 0
+
+        # Expand from multiple district frontiers to ensure contiguity by construction.
+        frontier_order = deque(districts)
+        while remaining:
+            if not frontier_order:
+                break
+            district_id = frontier_order.popleft()
+            current_frontier = frontiers[district_id]
+            if not current_frontier:
+                continue
+
+            source = current_frontier.popleft()
+            neighbors = [n for n in adjacency.get(source, set()) if n in remaining]
+            rng.shuffle(neighbors)
+
+            expanded = False
+            for neighbor in neighbors:
+                if neighbor not in remaining:
+                    continue
+                can_expand = (
+                    overcap > 3
+                    or district_sizes[district_id] < target_limits[district_id]
+                )
+                if not can_expand and overcap <= 3:
+                    continue
+                assignment[neighbor] = district_id
+                remaining.remove(neighbor)
+                current_frontier.append(neighbor)
+                district_sizes[district_id] += 1
+                frontier_order.append(district_id)
+                frontier_order.append(district_id)
+                expanded = True
+                break
+
+            if expanded:
+                continue
+
+            # If all districts reached targets, allow unrestricted growth to consume leftovers.
+            overcap += 1
+            for fallback_neighbor in neighbors:
+                if fallback_neighbor not in remaining:
+                    continue
+                assignment[fallback_neighbor] = district_id
+                remaining.remove(fallback_neighbor)
+                current_frontier.append(fallback_neighbor)
+                district_sizes[district_id] += 1
+                frontier_order.append(district_id)
+                break
+
+            if not expanded and not current_frontier:
+                # keep exploring this district only if it can still absorb nodes.
+                if all(size >= target_limits[d] for d, size in district_sizes.items()):
+                    continue
+
+            frontier_order.append(district_id)
+
+            if overcap > 10_000:
+                # Safety break for unexpected stalling.
+                break
+
+        # If anything remains because of local disconnectedness in the non-gateway subgraph,
+        # assign by nearest-seed fallback and rely on contiguity enforcement later.
+        if remaining:
+            fallback = self._assign_nearest(local_coords, seeds)
+            for node in remaining:
+                assignment[node] = fallback[node]
+        return assignment
+
+    def _enforce_contiguity(
+        self,
+        assignment: dict[str, str],
+        adjacency: dict[str, set[str]],
+        coords: dict[str, tuple[float, float]],
+    ) -> dict[str, str]:
+        district_ids = sorted(set(assignment.values()))
+        changed = True
+        while changed:
+            changed = False
+            for district_id in district_ids:
+                nodes = [n for n, d in assignment.items() if d == district_id]
+                if len(nodes) <= 1:
+                    continue
+                comps = connected_components(nodes, adjacency)
+                if len(comps) <= 1:
+                    continue
+                comps.sort(key=len, reverse=True)
+                keep = comps[0]
+                for comp in comps[1:]:
+                    for node in comp:
+                        reassigned = self._best_neighbor_district(
+                            node=node,
+                            assignment=assignment,
+                            adjacency=adjacency,
+                            coords=coords,
+                        )
+                        if reassigned != district_id:
+                            assignment[node] = reassigned
+                            changed = True
+        return assignment
+
+    def _best_neighbor_district(
+        self,
+        node: str,
+        assignment: dict[str, str],
+        adjacency: dict[str, set[str]],
+        coords: dict[str, tuple[float, float]],
+    ) -> str:
+        neighbors = [n for n in adjacency[node] if n in assignment]
+        if not neighbors:
+            return assignment[node]
+        counts = Counter(assignment[n] for n in neighbors)
+        best = counts.most_common(1)[0][0]
+        if len(counts) == 1:
+            return best
+        best_score = float("inf")
+        best_district = best
+        for district in counts.keys():
+            district_nodes = [n for n, d in assignment.items() if d == district]
+            if not district_nodes:
+                continue
+            centroid_x = sum(coords[n][0] for n in district_nodes) / len(district_nodes)
+            centroid_y = sum(coords[n][1] for n in district_nodes) / len(district_nodes)
+            dist = euclidean(coords[node], (centroid_x, centroid_y))
+            if dist < best_score:
+                best_score = dist
+                best_district = district
+        return best_district
+
+    def _fill_empty_districts(
+        self,
+        assignment: dict[str, str],
+        node_ids: list[str],
+        adjacency: dict[str, set[str]],
+        district_ids: list[str],
+    ) -> dict[str, str]:
+        # This method now mainly keeps a lower bound on singleton-heavy districts.
+        # Keep existing behavior if an empty district somehow appears.
+        counts = Counter(assignment.values())
+        empty = [d for d in district_ids if counts[d] == 0]
+        if not empty:
+            return assignment
+
+        for empty_id in empty:
+            largest = max(district_ids, key=lambda d: counts[d])
+            donor_candidates = [n for n in node_ids if assignment[n] == largest]
+            if not donor_candidates:
+                continue
+            pivot = donor_candidates[0]
+            for candidate in donor_candidates:
+                if any(
+                    assignment[n] != largest and assignment[n] != empty_id
+                    for n in adjacency.get(candidate, set())
+                    if n in assignment
+                ):
+                    pivot = candidate
+                    break
+            assignment[pivot] = empty_id
+            counts[largest] -= 1
+            counts[empty_id] += 1
+
+        # For any missing district, steal a boundary node from the largest district.
+        for district in district_ids:
+            if counts[district] > 1:
+                continue
+            root = next((n for n in node_ids if assignment[n] == district), None)
+            if root is None:
+                continue
+            queue = deque([root])
+            while queue and counts[district] < 2:
+                current = queue.popleft()
+                for candidate in node_ids:
+                    if assignment[candidate] == district:
+                        continue
+                    if candidate not in adjacency.get(current, set()):
+                        continue
+                    old = assignment[candidate]
+                    if counts[old] <= 2:
+                        continue
+                    assignment[candidate] = district
+                    counts[old] -= 1
+                    counts[district] += 1
+                    queue.append(candidate)
+                    if counts[district] >= 2:
+                        break
+        return assignment

data/generators/flow_generator.py

@@ -0,0 +1,662 @@
+"""Flow generation with district-aware O-D pressure and turn-feasible routing."""
+
+from __future__ import annotations
+
+import heapq
+import math
+import random
+from collections import defaultdict
+from typing import Any
+
+from .schemas import CityGraph, DistrictData, ScenarioPlan
+from .utils import (
+    build_road_index,
+    build_roadlink_index,
+    choose_weighted,
+    summarize_route_validation,
+    validate_route_with_reasons,
+)
+
+
+class FlowGenerator:
+    """Create high-pressure CityFlow flow entries for each scenario."""
+
+    DISTRICT_BASE_PRODUCTION = {
+        "residential": 1.70,
+        "commercial": 0.85,
+        "industrial": 0.95,
+        "mixed": 1.15,
+    }
+    DISTRICT_BASE_ATTRACTION = {
+        "residential": 0.90,
+        "commercial": 1.85,
+        "industrial": 1.55,
+        "mixed": 1.25,
+    }
+
+    INTENSITY_SCALE = {
+        "normal": 1.0,
+        "moderate_rush": 1.22,
+        "heavy_rush": 1.45,
+        "overload": 1.75,
+        "accident_overload": 2.0,
+    }
+
+    def generate(
+        self,
+        city_graph: CityGraph,
+        district_data: DistrictData,
+        scenario: ScenarioPlan,
+        simulation_steps: int,
+    ) -> list[dict[str, Any]]:
+        rng = random.Random(scenario.seed)
+        district_ids = sorted(district_data.districts.keys())
+        if len(district_ids) < 2:
+            raise ValueError("Flow generation requires at least 2 districts.")
+
+        base_density = int(scenario.metadata.get("base_demand_per_intersection", 36))
+        trip_total = int(
+            len(city_graph.intersections)
+            * base_density
+            * scenario.trip_multiplier
+        )
+        trip_total = max(260, min(42000, trip_total))
+
+        routing_state = self._build_routing_state(city_graph, scenario)
+        district_features = self._build_district_features(city_graph, district_data)
+        connector_counts = self._build_connector_counts(city_graph, district_data)
+        production, attraction = self._district_weights(
+            city_graph=city_graph,
+            district_data=district_data,
+            district_features=district_features,
+            scenario=scenario,
+        )
+        gateway_context = self._build_gateway_context(city_graph, district_data)
+        external_share = self._external_trip_share(scenario, has_gateways=bool(gateway_context["gateways"]))
+
+        flows: list[dict[str, Any]] = []
+        max_global_sampling_attempts = max(5000, trip_total * 8)
+        global_attempts = 0
+        while len(flows) < trip_total and global_attempts < max_global_sampling_attempts:
+            global_attempts += 1
+            route = None
+            max_attempts = 45
+            for _ in range(max_attempts):
+                external_mode = self._sample_external_mode(
+                    rng=rng,
+                    external_share=external_share,
+                )
+                category = self._sample_trip_category(rng, scenario.trip_mix)
+                origin_node: str
+                destination_node: str
+                origin_district: str
+                destination_district: str
+
+                if external_mode == "inbound":
+                    gateway = self._sample_gateway_for_inbound(
+                        rng=rng,
+                        gateway_context=gateway_context,
+                        attraction_weights=attraction,
+                    )
+                    if gateway is None:
+                        continue
+                    destination_district = self._sample_origin_district(
+                        rng=rng,
+                        district_ids=district_ids,
+                        weights=attraction,
+                    )
+                    origin_node = gateway
+                    destination_node = self._sample_intersection_in_district(
+                        rng=rng,
+                        district_data=district_data,
+                        district_id=destination_district,
+                        favor_boundary=False,
+                        excluded_nodes=gateway_context["anchor_nodes"],
+                    )
+                elif external_mode == "outbound":
+                    origin_district = self._sample_origin_district(
+                        rng=rng,
+                        district_ids=district_ids,
+                        weights=production,
+                    )
+                    gateway = self._sample_gateway_for_outbound(
+                        rng=rng,
+                        origin_district=origin_district,
+                        gateway_context=gateway_context,
+                        connector_counts=connector_counts,
+                    )
+                    if gateway is None:
+                        continue
+                    origin_node = self._sample_intersection_in_district(
+                        rng=rng,
+                        district_data=district_data,
+                        district_id=origin_district,
+                        favor_boundary=True,
+                        excluded_nodes=gateway_context["anchor_nodes"],
+                    )
+                    destination_node = gateway
+                else:
+                    origin_district = self._sample_origin_district(
+                        rng=rng,
+                        district_ids=district_ids,
+                        weights=production,
+                    )
+                    destination_district = self._sample_destination_district(
+                        rng=rng,
+                        city_graph=city_graph,
+                        origin_district=origin_district,
+                        category=category,
+                        district_data=district_data,
+                        district_features=district_features,
+                        district_ids=district_ids,
+                        attraction_weights=attraction,
+                        connector_counts=connector_counts,
+                    )
+                    origin_node = self._sample_intersection_in_district(
+                        rng=rng,
+                        district_data=district_data,
+                        district_id=origin_district,
+                        favor_boundary=(category != "intra"),
+                        excluded_nodes=set(),
+                    )
+                    destination_node = self._sample_intersection_in_district(
+                        rng=rng,
+                        district_data=district_data,
+                        district_id=destination_district,
+                        favor_boundary=(category == "long"),
+                        excluded_nodes=set(),
+                    )
+                if origin_node == destination_node:
+                    continue
+
+                route = self._find_turn_feasible_route(
+                    start_intersection=origin_node,
+                    end_intersection=destination_node,
+                    road_lookup=routing_state["road_lookup"],
+                    start_roads_by_intersection=routing_state["start_roads_by_intersection"],
+                    transitions=routing_state["transitions"],
+                    road_cost=routing_state["road_cost"],
+                )
+                if not route:
+                    continue
+                reasons = validate_route_with_reasons(
+                    route=route,
+                    roads_by_id=routing_state["road_lookup"],
+                    roadlinks_by_intersection=routing_state["roadlinks_by_intersection"],
+                )
+                if reasons:
+                    route = None
+                    continue
+                break
+
+            if route:
+                start_time = self._sample_departure(rng, scenario, simulation_steps)
+                flows.append(
+                    {
+                        "vehicle": {
+                            "length": 5.0,
+                            "width": 2.0,
+                            "maxPosAcc": 2.2,
+                            "maxNegAcc": 4.6,
+                            "usualPosAcc": 2.0,
+                            "usualNegAcc": 4.2,
+                            "minGap": 2.2,
+                            "maxSpeed": 13.89,
+                            "headwayTime": 1.4,
+                        },
+                        "route": route,
+                        "interval": 1.0,
+                        "startTime": start_time,
+                        "endTime": start_time,
+                    }
+                )
+
+        completion_ratio = len(flows) / max(1, trip_total)
+        min_completion_ratio = (
+            0.55 if scenario.name in {"accident", "construction"} else 0.70
+        )
+        if completion_ratio < min_completion_ratio:
+            raise ValueError(
+                f"Scenario {scenario.name} produced too few valid flows "
+                f"({len(flows)}/{trip_total}, completion={completion_ratio:.3f})."
+            )
+
+        if not flows:
+            raise ValueError(f"Scenario {scenario.name} produced no valid flows.")
+        summary = summarize_route_validation(
+            flow_entries=flows,
+            roads_by_id=routing_state["road_lookup"],
+            roadlinks_by_intersection=routing_state["roadlinks_by_intersection"],
+        )
+        if summary["invalid_routes"] > 0:
+            reasons = ", ".join(
+                f"{reason}={count}" for reason, count in summary["top_failure_reasons"]
+            )
+            raise ValueError(
+                f"Scenario {scenario.name} has invalid routes after regeneration: {reasons}"
+            )
+        return flows
+
+    def _build_gateway_context(
+        self,
+        city_graph: CityGraph,
+        district_data: DistrictData,
+    ) -> dict[str, Any]:
+        assignment = district_data.intersection_to_district
+        gateways = sorted(city_graph.gateway_intersections)
+        anchors_by_gateway: dict[str, str] = {}
+        district_by_gateway: dict[str, str] = {}
+        gateways_by_district: dict[str, list[str]] = defaultdict(list)
+
+        for gateway in gateways:
+            neighbors = [
+                n for n in city_graph.adjacency.get(gateway, set()) if n in assignment
+            ]
+            if not neighbors:
+                continue
+            anchor = sorted(neighbors)[0]
+            district_id = assignment[anchor]
+            anchors_by_gateway[gateway] = anchor
+            district_by_gateway[gateway] = district_id
+            gateways_by_district[district_id].append(gateway)
+
+        return {
+            "gateways": sorted(anchors_by_gateway.keys()),
+            "anchors_by_gateway": anchors_by_gateway,
+            "district_by_gateway": district_by_gateway,
+            "gateways_by_district": gateways_by_district,
+            "anchor_nodes": set(anchors_by_gateway.values()),
+        }
+
+    def _external_trip_share(
+        self,
+        scenario: ScenarioPlan,
+        has_gateways: bool,
+    ) -> float:
+        if not has_gateways:
+            return 0.0
+        base = {
+            "normal": 0.12,
+            "morning_rush": 0.16,
+            "evening_rush": 0.16,
+            "accident": 0.20,
+            "construction": 0.18,
+            "event_spike": 0.18,
+            "district_overload": 0.19,
+        }[scenario.name]
+        intensity_boost = {
+            "normal": 0.00,
+            "moderate_rush": 0.02,
+            "heavy_rush": 0.04,
+            "overload": 0.07,
+            "accident_overload": 0.10,
+        }.get(scenario.intensity, 0.0)
+        return min(0.42, base + intensity_boost)
+
+    def _sample_external_mode(
+        self,
+        rng: random.Random,
+        external_share: float,
+    ) -> str:
+        if external_share <= 0.0:
+            return "none"
+        if rng.random() >= external_share:
+            return "none"
+        return "inbound" if rng.random() < 0.5 else "outbound"
+
+    def _sample_gateway_for_inbound(
+        self,
+        rng: random.Random,
+        gateway_context: dict[str, Any],
+        attraction_weights: dict[str, float],
+    ) -> str | None:
+        gateways = gateway_context["gateways"]
+        if not gateways:
+            return None
+        district_by_gateway = gateway_context["district_by_gateway"]
+        weights: list[float] = []
+        for gateway in gateways:
+            district_id = district_by_gateway[gateway]
+            weights.append(1.0 + attraction_weights.get(district_id, 1.0))
+        return choose_weighted(rng, gateways, weights)
+
+    def _sample_gateway_for_outbound(
+        self,
+        rng: random.Random,
+        origin_district: str,
+        gateway_context: dict[str, Any],
+        connector_counts: dict[tuple[str, str], int],
+    ) -> str | None:
+        gateways = gateway_context["gateways"]
+        if not gateways:
+            return None
+        district_by_gateway = gateway_context["district_by_gateway"]
+        weights: list[float] = []
+        for gateway in gateways:
+            gateway_district = district_by_gateway[gateway]
+            connector_bonus = 1.0 + connector_counts.get(
+                (origin_district, gateway_district), 0
+            )
+            same_district_bonus = 2.0 if gateway_district == origin_district else 1.0
+            weights.append(connector_bonus * same_district_bonus)
+        return choose_weighted(rng, gateways, weights)
+
+    def _build_routing_state(
+        self,
+        city_graph: CityGraph,
+        scenario: ScenarioPlan,
+    ) -> dict[str, Any]:
+        road_lookup = build_road_index(city_graph.roadnet)
+        roadlinks_by_intersection = build_roadlink_index(city_graph.roadnet)
+        start_roads_by_intersection: dict[str, list[str]] = defaultdict(list)
+        road_cost: dict[str, float] = {}
+        available_roads: set[str] = set()
+
+        for road in city_graph.directed_roads.values():
+            if road.id in scenario.blocked_roads:
+                continue
+            cost = max(1.0, road.length / max(road.speed_limit, 1.0))
+            if road.id in scenario.penalized_roads:
+                cost *= scenario.penalized_roads[road.id]
+            road_cost[road.id] = cost
+            available_roads.add(road.id)
+            start_roads_by_intersection[road.start_intersection].append(road.id)
+
+        transitions: dict[str, list[str]] = defaultdict(list)
+        for pairs in roadlinks_by_intersection.values():
+            for start_road, end_road in pairs:
+                if start_road not in available_roads or end_road not in available_roads:
+                    continue
+                transitions[start_road].append(end_road)
+        return {
+            "road_lookup": road_lookup,
+            "roadlinks_by_intersection": roadlinks_by_intersection,
+            "start_roads_by_intersection": start_roads_by_intersection,
+            "transitions": transitions,
+            "road_cost": road_cost,
+        }
+
+    def _find_turn_feasible_route(
+        self,
+        start_intersection: str,
+        end_intersection: str,
+        road_lookup: dict[str, dict[str, Any]],
+        start_roads_by_intersection: dict[str, list[str]],
+        transitions: dict[str, list[str]],
+        road_cost: dict[str, float],
+    ) -> list[str] | None:
+        if start_intersection == end_intersection:
+            return None
+        start_roads = start_roads_by_intersection.get(start_intersection, [])
+        if not start_roads:
+            return None
+
+        queue: list[tuple[float, str]] = []
+        dist: dict[str, float] = {}
+        prev: dict[str, str | None] = {}
+
+        for road_id in start_roads:
+            if road_id not in road_cost:
+                continue
+            cost = road_cost[road_id]
+            dist[road_id] = cost
+            prev[road_id] = None
+            heapq.heappush(queue, (cost, road_id))
+
+        best_terminal: str | None = None
+        while queue:
+            current_cost, current_road = heapq.heappop(queue)
+            if current_cost > dist.get(current_road, float("inf")):
+                continue
+
+            current_end = road_lookup[current_road]["endIntersection"]
+            if current_end == end_intersection:
+                best_terminal = current_road
+                break
+
+            for next_road in transitions.get(current_road, []):
+                next_cost = current_cost + road_cost[next_road]
+                if next_cost < dist.get(next_road, float("inf")):
+                    dist[next_road] = next_cost
+                    prev[next_road] = current_road
+                    heapq.heappush(queue, (next_cost, next_road))
+
+        if best_terminal is None:
+            return None
+
+        route: list[str] = []
+        cursor: str | None = best_terminal
+        while cursor is not None:
+            route.append(cursor)
+            cursor = prev[cursor]
+        route.reverse()
+        return route
+
+    def _build_district_features(
+        self,
+        city_graph: CityGraph,
+        district_data: DistrictData,
+    ) -> dict[str, dict[str, float]]:
+        features: dict[str, dict[str, float]] = {}
+        for did, district in district_data.districts.items():
+            members = district.intersections
+            size = len(members)
+            cx = sum(city_graph.intersections[n][0] for n in members) / max(1, size)
+            cy = sum(city_graph.intersections[n][1] for n in members) / max(1, size)
+            features[did] = {
+                "size": float(size),
+                "neighbors": float(len(district.neighbors)),
+                "exits": float(len(district.exit_roads)),
+                "boundary": float(len(district.boundary_intersections)),
+                "cx": cx,
+                "cy": cy,
+            }
+        return features
+
+    def _build_connector_counts(
+        self,
+        city_graph: CityGraph,
+        district_data: DistrictData,
+    ) -> dict[tuple[str, str], int]:
+        connector_counts: dict[tuple[str, str], int] = defaultdict(int)
+        assignment = district_data.intersection_to_district
+        for a, neighbors in city_graph.adjacency.items():
+            if a not in assignment:
+                continue
+            da = assignment[a]
+            for b in neighbors:
+                if b not in assignment:
+                    continue
+                db = assignment[b]
+                if da == db:
+                    continue
+                connector_counts[(da, db)] += 1
+        return connector_counts
+
+    def _district_weights(
+        self,
+        city_graph: CityGraph,
+        district_data: DistrictData,
+        district_features: dict[str, dict[str, float]],
+        scenario: ScenarioPlan,
+    ) -> tuple[dict[str, float], dict[str, float]]:
+        production: dict[str, float] = {}
+        attraction: dict[str, float] = {}
+        intensity = self.INTENSITY_SCALE.get(scenario.intensity, 1.0)
+
+        for did, district in district_data.districts.items():
+            feature = district_features[did]
+            size_factor = max(0.85, min(2.1, math.sqrt(feature["size"]) / 2.0))
+            connector_factor = 1.0 + min(1.6, feature["exits"] / 7.0)
+            base_prod = self.DISTRICT_BASE_PRODUCTION[district.district_type]
+            base_attr = self.DISTRICT_BASE_ATTRACTION[district.district_type]
+            production[did] = base_prod * size_factor * (0.60 + 0.40 * connector_factor)
+            attraction[did] = base_attr * size_factor * (0.62 + 0.38 * connector_factor)
+
+        if scenario.name == "morning_rush":
+            self._scale_by_type(district_data, production, "residential", 3.0 * intensity)
+            self._scale_by_type(district_data, production, "mixed", 1.3 * intensity)
+            self._scale_by_type(district_data, attraction, "commercial", 3.2 * intensity)
+            self._scale_by_type(district_data, attraction, "industrial", 2.8 * intensity)
+            self._scale_by_type(district_data, attraction, "residential", 0.58)
+        elif scenario.name == "evening_rush":
+            self._scale_by_type(district_data, production, "commercial", 3.1 * intensity)
+            self._scale_by_type(district_data, production, "industrial", 2.7 * intensity)
+            self._scale_by_type(district_data, attraction, "residential", 3.0 * intensity)
+            self._scale_by_type(district_data, attraction, "commercial", 0.62)
+        elif scenario.name == "event_spike" and scenario.event_district:
+            attraction[scenario.event_district] *= 3.8 * intensity
+            production[scenario.event_district] *= 1.9 * intensity
+        elif scenario.name == "district_overload" and scenario.overload_district:
+            production[scenario.overload_district] *= 3.2 * intensity
+            attraction[scenario.overload_district] *= 3.0 * intensity
+
+        if scenario.name in {"accident", "construction"}:
+            impacted = self._impacted_districts(city_graph, district_data, scenario)
+            for did in impacted:
+                attraction[did] *= 1.6 * intensity
+                production[did] *= 1.45 * intensity
+                for neighbor in district_data.district_neighbors.get(did, []):
+                    attraction[neighbor] *= 1.18
+                    production[neighbor] *= 1.16
+
+        return production, attraction
+
+    def _impacted_districts(
+        self,
+        city_graph: CityGraph,
+        district_data: DistrictData,
+        scenario: ScenarioPlan,
+    ) -> set[str]:
+        impacted: set[str] = set()
+        assignment = district_data.intersection_to_district
+        for road_id in set(scenario.blocked_roads) | set(scenario.penalized_roads.keys()):
+            road = city_graph.directed_roads.get(road_id)
+            if road is None:
+                continue
+            if road.start_intersection in assignment:
+                impacted.add(assignment[road.start_intersection])
+            if road.end_intersection in assignment:
+                impacted.add(assignment[road.end_intersection])
+        return impacted
+
+    def _scale_by_type(
+        self,
+        district_data: DistrictData,
+        weights: dict[str, float],
+        district_type: str,
+        factor: float,
+    ) -> None:
+        for did, district in district_data.districts.items():
+            if district.district_type == district_type:
+                weights[did] *= factor
+
+    def _sample_trip_category(self, rng: random.Random, trip_mix: Any) -> str:
+        labels = ["intra", "adjacent", "long"]
+        weights = [
+            trip_mix.intra_district,
+            trip_mix.adjacent_district,
+            trip_mix.long_distance,
+        ]
+        return choose_weighted(rng, labels, weights)
+
+    def _sample_origin_district(
+        self,
+        rng: random.Random,
+        district_ids: list[str],
+        weights: dict[str, float],
+    ) -> str:
+        values = district_ids
+        scalar = [weights[d] for d in district_ids]
+        return choose_weighted(rng, values, scalar)
+
+    def _sample_destination_district(
+        self,
+        rng: random.Random,
+        city_graph: CityGraph,
+        origin_district: str,
+        category: str,
+        district_data: DistrictData,
+        district_features: dict[str, dict[str, float]],
+        district_ids: list[str],
+        attraction_weights: dict[str, float],
+        connector_counts: dict[tuple[str, str], int],
+    ) -> str:
+        if category == "intra":
+            return origin_district
+
+        if category == "adjacent":
+            neighbors = district_data.district_neighbors.get(origin_district, [])
+            if neighbors:
+                weights = []
+                for neighbor in neighbors:
+                    connector = 1.0 + connector_counts.get((origin_district, neighbor), 0)
+                    weights.append(attraction_weights[neighbor] * connector)
+                return choose_weighted(rng, neighbors, weights)
+
+        origin_feature = district_features[origin_district]
+        candidates = [d for d in district_ids if d != origin_district]
+        if category == "long":
+            candidates = [
+                did
+                for did in candidates
+                if did not in district_data.district_neighbors.get(origin_district, [])
+            ] or [d for d in district_ids if d != origin_district]
+
+        weights: list[float] = []
+        for candidate in candidates:
+            feature = district_features[candidate]
+            dx = feature["cx"] - origin_feature["cx"]
+            dy = feature["cy"] - origin_feature["cy"]
+            distance = math.hypot(dx, dy)
+            normalized_distance = max(1.0, distance / 260.0)
+            corridor_bonus = 1.0 + min(
+                1.5,
+                (
+                    feature["exits"] + feature["neighbors"]
+                ) / 10.0,
+            )
+            if category == "long":
+                weight = attraction_weights[candidate] * normalized_distance * corridor_bonus
+            else:
+                weight = attraction_weights[candidate] * (0.85 + 0.15 * corridor_bonus)
+            weights.append(weight)
+        return choose_weighted(rng, candidates, weights)
+
+    def _sample_intersection_in_district(
+        self,
+        rng: random.Random,
+        district_data: DistrictData,
+        district_id: str,
+        favor_boundary: bool,
+        excluded_nodes: set[str],
+    ) -> str:
+        district = district_data.districts[district_id]
+        values = [node for node in district.intersections if node not in excluded_nodes]
+        if not values:
+            values = district.intersections
+        boundary = set(district.boundary_intersections)
+        weights: list[float] = []
+        for node in values:
+            if node in boundary:
+                weights.append(2.3 if favor_boundary else 1.15)
+            else:
+                weights.append(0.95 if favor_boundary else 1.2)
+        return choose_weighted(rng, values, weights)
+
+    def _sample_departure(
+        self,
+        rng: random.Random,
+        scenario: ScenarioPlan,
+        simulation_steps: int,
+    ) -> int:
+        windows = scenario.departure_windows
+        labels = list(range(len(windows)))
+        weights = [w for _, _, w in windows]
+        selected = choose_weighted(rng, [str(i) for i in labels], weights)
+        window = windows[int(selected)]
+        start = int(window[0] * simulation_steps)
+        end = int(window[1] * simulation_steps)
+        if end <= start:
+            end = start + 1
+        return rng.randint(start, max(start, end - 1))

data/generators/generate_dataset.py

@@ -0,0 +1,179 @@
+"""CLI entrypoint for synthetic CityFlow dataset generation."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from .city_generator import CityGenerator
+from .schemas import DatasetGenerationConfig, DemandIntensity
+
+DEFAULT_TOPOLOGIES = ["irregular_grid"]
+DEFAULT_SCENARIOS = [
+    "normal",
+    "morning_rush",
+    "evening_rush",
+    "accident",
+    "construction",
+    "event_spike",
+    "district_overload",
+]
+DEFAULT_INTENSITY_LEVELS: list[DemandIntensity] = [
+    "normal",
+    "moderate_rush",
+    "heavy_rush",
+    "overload",
+    "accident_overload",
+]
+DEFAULT_INTENSITY_DISTRIBUTION: dict[DemandIntensity, float] = {
+    "normal": 0.20,
+    "moderate_rush": 0.42,
+    "heavy_rush": 0.24,
+    "overload": 0.10,
+    "accident_overload": 0.04,
+}
+DEFAULT_SCENARIO_DEMAND_MULTIPLIERS: dict[str, float] = {
+    "normal": 1.15,
+    "morning_rush": 1.35,
+    "evening_rush": 1.35,
+    "accident": 1.75,
+    "construction": 1.55,
+    "event_spike": 1.65,
+    "district_overload": 1.70,
+}
+
+
+def _parse_csv_list(raw: str | None) -> list[str] | None:
+    if raw is None:
+        return None
+    values = [part.strip() for part in raw.split(",")]
+    return [v for v in values if v]
+
+
+def _parse_key_value_floats(raw: str | None) -> dict[str, float]:
+    if not raw:
+        return {}
+    result: dict[str, float] = {}
+    for token in raw.split(","):
+        token = token.strip()
+        if not token:
+            continue
+        if "=" not in token:
+            raise ValueError(f"Expected key=value pair, got '{token}'.")
+        key, value = token.split("=", 1)
+        result[key.strip()] = float(value.strip())
+    return result
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Generate synthetic CityFlow cities with district-aware scenarios."
+    )
+    parser.add_argument("--num-cities", type=int, default=100)
+    parser.add_argument("--output-dir", type=Path, default=Path("data/generated"))
+    parser.add_argument("--seed", type=int, default=42)
+    parser.add_argument("--min-districts", type=int, default=6)
+    parser.add_argument("--max-districts", type=int, default=20)
+    parser.add_argument("--min-intersections-per-district", type=int, default=4)
+    parser.add_argument("--max-intersections-per-district", type=int, default=10)
+    parser.add_argument(
+        "--topologies",
+        type=str,
+        default=None,
+        help="Comma-separated list of topologies.",
+    )
+    parser.add_argument(
+        "--scenarios",
+        type=str,
+        default=None,
+        help="Comma-separated list of scenarios.",
+    )
+    parser.add_argument("--simulation-steps", type=int, default=3600)
+    parser.add_argument("--interval", type=float, default=1.0)
+    parser.add_argument(
+        "--intensity-levels",
+        type=str,
+        default=None,
+        help="Comma-separated intensity levels.",
+    )
+    parser.add_argument(
+        "--intensity-distribution",
+        type=str,
+        default=None,
+        help="Comma-separated key=value weights for intensities.",
+    )
+    parser.add_argument(
+        "--global-demand-multiplier",
+        type=float,
+        default=1.25,
+        help="Global demand multiplier across all scenarios.",
+    )
+    parser.add_argument(
+        "--scenario-demand-multipliers",
+        type=str,
+        default=None,
+        help="Comma-separated key=value multipliers by scenario name.",
+    )
+    parser.add_argument(
+        "--ring-diagonal-keep-prob",
+        type=float,
+        default=0.07,
+        help="Keep probability for optional ring-road interior diagonals.",
+    )
+    parser.add_argument(
+        "--ring-max-diagonal-fraction",
+        type=float,
+        default=0.03,
+        help="Maximum fraction of optional diagonals retained in ring-road topology.",
+    )
+    parser.add_argument("--save-replay", action="store_true")
+    parser.add_argument("--fail-fast", action="store_true")
+    return parser
+
+
+def main() -> None:
+    parser = build_parser()
+    args = parser.parse_args()
+    topologies = _parse_csv_list(args.topologies)
+    scenarios = _parse_csv_list(args.scenarios)
+    intensity_levels = _parse_csv_list(args.intensity_levels)
+    intensity_distribution = _parse_key_value_floats(args.intensity_distribution)
+    scenario_demand_multipliers = _parse_key_value_floats(
+        args.scenario_demand_multipliers
+    )
+    config = DatasetGenerationConfig(
+        num_cities=args.num_cities,
+        output_dir=args.output_dir,
+        seed=args.seed,
+        min_districts=args.min_districts,
+        max_districts=args.max_districts,
+        min_intersections_per_district=args.min_intersections_per_district,
+        max_intersections_per_district=args.max_intersections_per_district,
+        topologies=topologies if topologies is not None else DEFAULT_TOPOLOGIES,
+        scenarios=scenarios if scenarios is not None else DEFAULT_SCENARIOS,
+        intensity_levels=(
+            intensity_levels if intensity_levels is not None else DEFAULT_INTENSITY_LEVELS
+        ),
+        intensity_distribution=(
+            intensity_distribution
+            if intensity_distribution
+            else DEFAULT_INTENSITY_DISTRIBUTION
+        ),
+        global_demand_multiplier=args.global_demand_multiplier,
+        scenario_demand_multipliers=(
+            scenario_demand_multipliers
+            if scenario_demand_multipliers
+            else DEFAULT_SCENARIO_DEMAND_MULTIPLIERS
+        ),
+        ring_diagonal_keep_prob=args.ring_diagonal_keep_prob,
+        ring_max_diagonal_fraction=args.ring_max_diagonal_fraction,
+        simulation_steps=args.simulation_steps,
+        interval=args.interval,
+        save_replay=args.save_replay,
+        fail_fast=args.fail_fast,
+    )
+    CityGenerator().generate_dataset(config)
+
+
+if __name__ == "__main__":
+    main()

data/generators/main.py

@@ -0,0 +1,60 @@
+"""Preconfigured dataset generation harness (no CLI arguments)."""
+
+from __future__ import annotations
+
+import secrets
+from pathlib import Path
+
+from .city_generator import CityGenerator
+from .schemas import DatasetGenerationConfig
+
+
+# -----------------------------------------------------------------------------
+# Edit these values to control the default batch generation run.
+# -----------------------------------------------------------------------------
+NUM_CITIES: int = 100
+OUTPUT_DIR: Path = Path("data/generated")
+BASE_SEED: int | None = 42  # set to an int for reproducible runs
+TOPOLOGIES: list[str] = [
+    "rectangular_grid",
+    "irregular_grid",
+    "arterial_local",
+    "ring_road",
+    "mixed",
+]
+MIN_DISTRICTS: int = 6
+MAX_DISTRICTS: int = 20
+MIN_INTERSECTIONS_PER_DISTRICT: int = 4
+MAX_INTERSECTIONS_PER_DISTRICT: int = 10
+SIMULATION_STEPS: int = 3600
+INTERVAL: float = 1.0
+FAIL_FAST: bool = False
+
+
+def main() -> None:
+    """Run deterministic-configured dataset generation with pre-set defaults."""
+    base_seed = BASE_SEED if BASE_SEED is not None else secrets.randbits(63)
+
+    config = DatasetGenerationConfig(
+        num_cities=NUM_CITIES,
+        output_dir=OUTPUT_DIR,
+        seed=base_seed,
+        topologies=TOPOLOGIES,
+        min_districts=MIN_DISTRICTS,
+        max_districts=MAX_DISTRICTS,
+        min_intersections_per_district=MIN_INTERSECTIONS_PER_DISTRICT,
+        max_intersections_per_district=MAX_INTERSECTIONS_PER_DISTRICT,
+        simulation_steps=SIMULATION_STEPS,
+        interval=INTERVAL,
+        fail_fast=FAIL_FAST,
+    )
+
+    print(f"Generating {config.num_cities} cities into {config.output_dir}")
+    print(f"Base seed: {config.seed}")
+    print(f"Topologies: {', '.join(TOPOLOGIES)}")
+
+    CityGenerator().generate_dataset(config)
+
+
+if __name__ == "__main__":
+    main()

data/generators/roadnet_generator.py

@@ -0,0 +1,1034 @@
+"""Road network topology generation and CityFlow roadnet assembly."""
+
+from __future__ import annotations
+
+import math
+import random
+from collections import defaultdict
+from itertools import product
+
+from .schemas import CityGraph, RoadRecord, TopologyType
+from .utils import euclidean
+
+
+class RoadnetGenerator:
+    """Generate city intersection graph and convert to CityFlow roadnet format."""
+
+    def generate(
+        self,
+        city_id: str,
+        seed: int,
+        topology: TopologyType,
+        target_intersections: int,
+        ring_diagonal_keep_prob: float = 0.07,
+        ring_max_diagonal_fraction: float = 0.03,
+    ) -> CityGraph:
+        rng = random.Random(seed)
+        coords, undirected_edges, arterial_pairs = self._build_topology(
+            topology=topology,
+            target_nodes=target_intersections,
+            rng=rng,
+            ring_diagonal_keep_prob=ring_diagonal_keep_prob,
+            ring_max_diagonal_fraction=ring_max_diagonal_fraction,
+        )
+        (
+            coords,
+            undirected_edges,
+            gateway_pairs,
+            gateway_nodes,
+        ) = self._augment_with_perimeter_gateways(
+            coords=coords,
+            undirected_edges=undirected_edges,
+            rng=rng,
+        )
+        adjacency = self._to_adjacency(coords, undirected_edges)
+        directed_roads, arterial_road_ids, gateway_road_ids = self._build_directed_roads(
+            coords=coords,
+            undirected_edges=undirected_edges,
+            arterial_pairs=arterial_pairs,
+            gateway_pairs=gateway_pairs,
+        )
+        roadnet = self._build_roadnet(
+            coords=coords,
+            adjacency=adjacency,
+            directed_roads=directed_roads,
+        )
+        return CityGraph(
+            city_id=city_id,
+            topology=topology,
+            seed=seed,
+            intersections=coords,
+            adjacency=adjacency,
+            directed_roads=directed_roads,
+            roadnet=roadnet,
+            arterial_roads=arterial_road_ids,
+            gateway_intersections=gateway_nodes,
+            gateway_roads=gateway_road_ids,
+        )
+
+    def _build_topology(
+        self,
+        topology: TopologyType,
+        target_nodes: int,
+        rng: random.Random,
+        ring_diagonal_keep_prob: float,
+        ring_max_diagonal_fraction: float,
+    ) -> tuple[
+        dict[str, tuple[float, float]],
+        list[tuple[str, str]],
+        set[frozenset[str]],
+    ]:
+        if topology == "rectangular_grid":
+            return self._rectangular_grid(target_nodes, rng)
+        if topology == "irregular_grid":
+            return self._irregular_grid(target_nodes, rng)
+        if topology == "arterial_local":
+            return self._arterial_local(target_nodes, rng)
+        if topology == "ring_road":
+            return self._ring_road(
+                target_nodes=target_nodes,
+                rng=rng,
+                ring_diagonal_keep_prob=ring_diagonal_keep_prob,
+                ring_max_diagonal_fraction=ring_max_diagonal_fraction,
+            )
+        return self._mixed(target_nodes, rng)
+
+    def _dimensions(self, target_nodes: int) -> tuple[int, int]:
+        cols = max(3, int(round(math.sqrt(target_nodes))))
+        rows = max(3, int(math.ceil(target_nodes / cols)))
+        return rows, cols
+
+    def _grid_coords(
+        self,
+        rows: int,
+        cols: int,
+        spacing: float,
+        jitter: float,
+        rng: random.Random,
+    ) -> dict[str, tuple[float, float]]:
+        coords: dict[str, tuple[float, float]] = {}
+        idx = 0
+        for r, c in product(range(rows), range(cols)):
+            x = c * spacing + rng.uniform(-jitter, jitter)
+            y = r * spacing + rng.uniform(-jitter, jitter)
+            coords[f"i_{idx:04d}"] = (x, y)
+            idx += 1
+        return coords
+
+    def _smooth_axis_offsets(
+        self,
+        size: int,
+        max_offset: float,
+        step_limit: float,
+        rng: random.Random,
+    ) -> list[float]:
+        offsets = [0.0] * size
+        drift = 0.0
+        for idx in range(1, size - 1):
+            drift += rng.uniform(-step_limit, step_limit)
+            drift = max(-max_offset, min(max_offset, drift))
+            offsets[idx] = drift
+
+        if size > 2:
+            mean_mid = sum(offsets[1:-1]) / (size - 2)
+            for idx in range(1, size - 1):
+                centered = offsets[idx] - mean_mid
+                offsets[idx] = max(-max_offset, min(max_offset, centered))
+
+        offsets[0] = 0.0
+        offsets[-1] = 0.0
+        return offsets
+
+    def _boundary_stability_weight(self, idx: int, size: int) -> float:
+        distance = min(idx, size - 1 - idx)
+        if distance <= 0:
+            return 0.0
+        if distance == 1:
+            return 0.2
+        if distance == 2:
+            return 0.45
+        if distance == 3:
+            return 0.72
+        return 1.00
+
+    def _axis_positions(
+        self,
+        size: int,
+        spacing: float,
+        gap_variation: float,
+        rng: random.Random,
+    ) -> list[float]:
+        if size <= 1:
+            return [0.0]
+
+        gap_profile = self._smooth_drop_profile(size - 1, rng)
+        positions = [0.0]
+        min_gap = spacing * (1.0 - gap_variation)
+        max_gap = spacing * (1.0 + gap_variation)
+
+        for gap_idx in range(size - 1):
+            centered = (gap_profile[gap_idx] - 0.5) * 2.0
+            edge_distance = min(gap_idx, (size - 2) - gap_idx)
+            if edge_distance <= 0:
+                edge_weight = 0.32
+            elif edge_distance == 1:
+                edge_weight = 0.55
+            elif edge_distance == 2:
+                edge_weight = 0.8
+            else:
+                edge_weight = 1.0
+
+            local_noise = rng.uniform(-0.22, 0.22) * gap_variation * edge_weight
+            gap = spacing * (1.0 + (centered * gap_variation * edge_weight) + local_noise)
+            gap = max(min_gap, min(max_gap, gap))
+            positions.append(positions[-1] + gap)
+
+        nominal_span = spacing * (size - 1)
+        actual_span = positions[-1]
+        if actual_span > 1e-9:
+            scale = nominal_span / actual_span
+            positions = [p * scale for p in positions]
+        return positions
+
+    def _irregular_grid_coords(
+        self,
+        rows: int,
+        cols: int,
+        spacing: float,
+        rng: random.Random,
+    ) -> dict[str, tuple[float, float]]:
+        # Keep intersections mostly on row/column lines while varying block size.
+        row_positions = self._axis_positions(rows, spacing, gap_variation=0.16, rng=rng)
+        col_positions = self._axis_positions(cols, spacing, gap_variation=0.16, rng=rng)
+
+        # Small line-wise drift and very small local jitter.
+        max_row_offset = spacing * 0.018
+        max_col_offset = spacing * 0.018
+        row_step = spacing * 0.006
+        col_step = spacing * 0.006
+        local_jitter = spacing * 0.0045
+
+        row_offsets = self._smooth_axis_offsets(rows, max_row_offset, row_step, rng)
+        col_offsets = self._smooth_axis_offsets(cols, max_col_offset, col_step, rng)
+
+        coords: dict[str, tuple[float, float]] = {}
+        idx = 0
+        for r, c in product(range(rows), range(cols)):
+            # Keep perimeter nodes stable while allowing interior irregularity.
+            perimeter_weight = min(
+                self._boundary_stability_weight(r, rows),
+                self._boundary_stability_weight(c, cols),
+            )
+            jitter_weight = perimeter_weight * perimeter_weight
+            x = col_positions[c] + (col_offsets[c] * perimeter_weight)
+            y = row_positions[r] + (row_offsets[r] * perimeter_weight)
+            x += rng.uniform(-local_jitter, local_jitter) * jitter_weight
+            y += rng.uniform(-local_jitter, local_jitter) * jitter_weight
+            coords[f"i_{idx:04d}"] = (x, y)
+            idx += 1
+        return coords
+
+    def _grid_edges(self, rows: int, cols: int) -> list[tuple[int, int]]:
+        edges: list[tuple[int, int]] = []
+        for r in range(rows):
+            for c in range(cols):
+                idx = r * cols + c
+                if c + 1 < cols:
+                    edges.append((idx, idx + 1))
+                if r + 1 < rows:
+                    edges.append((idx, idx + cols))
+        return edges
+
+    def _estimate_spacing(
+        self,
+        coords: dict[str, tuple[float, float]],
+        undirected_edges: list[tuple[str, str]],
+    ) -> float:
+        if not undirected_edges:
+            return 120.0
+        lengths = [
+            euclidean(coords[a], coords[b])
+            for a, b in undirected_edges
+            if a in coords and b in coords
+        ]
+        if not lengths:
+            return 120.0
+        lengths.sort()
+        return lengths[len(lengths) // 2]
+
+    def _select_spread_nodes(
+        self,
+        candidates: list[str],
+        coords: dict[str, tuple[float, float]],
+        count: int,
+        axis: str,
+    ) -> list[str]:
+        if count <= 0 or not candidates:
+            return []
+        if len(candidates) <= count:
+            return candidates
+        axis_idx = 0 if axis == "x" else 1
+        ordered = sorted(
+            candidates,
+            key=lambda nid: (coords[nid][axis_idx], nid),
+        )
+        selected: list[str] = []
+        for i in range(count):
+            pos = int(round((i * (len(ordered) - 1)) / max(1, count - 1)))
+            selected.append(ordered[pos])
+        deduped = sorted(set(selected), key=lambda nid: ordered.index(nid))
+        if len(deduped) >= count:
+            return deduped[:count]
+        for node in ordered:
+            if node in deduped:
+                continue
+            deduped.append(node)
+            if len(deduped) >= count:
+                break
+        return deduped
+
+    def _augment_with_perimeter_gateways(
+        self,
+        coords: dict[str, tuple[float, float]],
+        undirected_edges: list[tuple[str, str]],
+        rng: random.Random,
+    ) -> tuple[
+        dict[str, tuple[float, float]],
+        list[tuple[str, str]],
+        set[frozenset[str]],
+        set[str],
+    ]:
+        if not coords:
+            return coords, undirected_edges, set(), set()
+
+        min_x = min(x for x, _ in coords.values())
+        max_x = max(x for x, _ in coords.values())
+        min_y = min(y for _, y in coords.values())
+        max_y = max(y for _, y in coords.values())
+        spacing = self._estimate_spacing(coords, undirected_edges)
+        threshold = max(4.0, spacing * 0.08)
+        per_side_target = 2
+
+        side_to_candidates: dict[str, list[str]] = {
+            "west": [],
+            "east": [],
+            "south": [],
+            "north": [],
+        }
+        for node_id, (x, y) in coords.items():
+            distances = {
+                "west": abs(x - min_x),
+                "east": abs(x - max_x),
+                "south": abs(y - min_y),
+                "north": abs(y - max_y),
+            }
+            side = min(distances, key=distances.get)
+            if distances[side] <= threshold:
+                side_to_candidates[side].append(node_id)
+
+        selected_anchors: list[tuple[str, str]] = []
+        used_anchors: set[str] = set()
+        for side in ("west", "east", "south", "north"):
+            candidates = [n for n in side_to_candidates[side] if n not in used_anchors]
+            axis = "y" if side in {"west", "east"} else "x"
+            chosen = self._select_spread_nodes(
+                candidates=candidates,
+                coords=coords,
+                count=per_side_target,
+                axis=axis,
+            )
+            for anchor in chosen:
+                if anchor in used_anchors:
+                    continue
+                used_anchors.add(anchor)
+                selected_anchors.append((side, anchor))
+
+        if not selected_anchors:
+            return coords, undirected_edges, set(), set()
+
+        offset = max(45.0, spacing * 0.82)
+        gateway_pairs: set[frozenset[str]] = set()
+        gateway_nodes: set[str] = set()
+        next_idx = 0
+        for side, anchor in selected_anchors:
+            ax, ay = coords[anchor]
+            if side == "west":
+                gx, gy = min_x - offset, ay + rng.uniform(-spacing * 0.03, spacing * 0.03)
+            elif side == "east":
+                gx, gy = max_x + offset, ay + rng.uniform(-spacing * 0.03, spacing * 0.03)
+            elif side == "south":
+                gx, gy = ax + rng.uniform(-spacing * 0.03, spacing * 0.03), min_y - offset
+            else:
+                gx, gy = ax + rng.uniform(-spacing * 0.03, spacing * 0.03), max_y + offset
+
+            gateway_id = f"g_{next_idx:04d}"
+            next_idx += 1
+            coords[gateway_id] = (gx, gy)
+            undirected_edges.append((anchor, gateway_id))
+            gateway_pairs.add(frozenset((anchor, gateway_id)))
+            gateway_nodes.add(gateway_id)
+
+        return coords, undirected_edges, gateway_pairs, gateway_nodes
+
+    def _arterial_indices(self, size: int) -> list[int]:
+        candidates = {size // 3, size // 2, (2 * size) // 3}
+        selected = sorted(i for i in candidates if 0 < i < size - 1)
+        if not selected and size > 2:
+            selected = [size // 2]
+        return selected
+
+    def _smooth_drop_profile(self, size: int, rng: random.Random) -> list[float]:
+        values: list[float] = []
+        state = rng.uniform(-0.25, 0.25)
+        for _ in range(size):
+            state = 0.78 * state + 0.22 * rng.uniform(-1.0, 1.0)
+            values.append(state)
+        low = min(values)
+        high = max(values)
+        if abs(high - low) < 1e-6:
+            return [0.5] * size
+        return [(value - low) / (high - low) for value in values]
+
+    def _edge_orientation(
+        self,
+        a: int,
+        b: int,
+        cols: int,
+    ) -> tuple[str, int, int]:
+        ra, ca = divmod(a, cols)
+        rb, cb = divmod(b, cols)
+        if ra == rb:
+            return ("horizontal", ra, min(ca, cb))
+        return ("vertical", ca, min(ra, rb))
+
+    def _is_edge_protected(
+        self,
+        a: int,
+        b: int,
+        rows: int,
+        cols: int,
+        arterial_rows: set[int],
+        arterial_cols: set[int],
+    ) -> tuple[bool, bool]:
+        ra, ca = divmod(a, cols)
+        rb, cb = divmod(b, cols)
+        if ra == rb:
+            on_perimeter = ra in {0, rows - 1}
+            on_arterial = ra in arterial_rows
+        else:
+            on_perimeter = ca in {0, cols - 1}
+            on_arterial = ca in arterial_cols
+        return on_perimeter or on_arterial, on_arterial
+
+    def _has_path_without_edge(
+        self,
+        start: int,
+        goal: int,
+        adjacency: dict[int, set[int]],
+        edge: tuple[int, int],
+    ) -> bool:
+        blocked_u, blocked_v = edge
+        stack = [start]
+        visited = {start}
+        while stack:
+            node = stack.pop()
+            if node == goal:
+                return True
+            for nxt in adjacency[node]:
+                if (
+                    (node == blocked_u and nxt == blocked_v)
+                    or (node == blocked_v and nxt == blocked_u)
+                ):
+                    continue
+                if nxt in visited:
+                    continue
+                visited.add(nxt)
+                stack.append(nxt)
+        return False
+
+    def _is_connected_without_str_edge(
+        self,
+        start: str,
+        goal: str,
+        adjacency: dict[str, set[str]],
+        edge: tuple[str, str],
+    ) -> bool:
+        blocked_u, blocked_v = edge
+        stack = [start]
+        visited = {start}
+        while stack:
+            node = stack.pop()
+            if node == goal:
+                return True
+            for nxt in adjacency[node]:
+                if (
+                    (node == blocked_u and nxt == blocked_v)
+                    or (node == blocked_v and nxt == blocked_u)
+                ):
+                    continue
+                if nxt in visited:
+                    continue
+                visited.add(nxt)
+                stack.append(nxt)
+        return False
+
+    def _rectangular_grid(
+        self, target_nodes: int, rng: random.Random
+    ) -> tuple[dict[str, tuple[float, float]], list[tuple[str, str]], set[frozenset[str]]]:
+        rows, cols = self._dimensions(target_nodes)
+        coords = self._grid_coords(rows, cols, spacing=120.0, jitter=6.0, rng=rng)
+        edges_raw = self._grid_edges(rows, cols)
+        id_list = list(coords.keys())
+        edges = [(id_list[a], id_list[b]) for a, b in edges_raw]
+        return coords, edges, set()
+
+    def _irregular_grid(
+        self, target_nodes: int, rng: random.Random
+    ) -> tuple[dict[str, tuple[float, float]], list[tuple[str, str]], set[frozenset[str]]]:
+        rows, cols = self._dimensions(int(target_nodes * 1.08))
+        coords = self._irregular_grid_coords(rows, cols, spacing=115.0, rng=rng)
+        edges_raw = self._grid_edges(rows, cols)
+        filtered: set[tuple[int, int]] = set(edges_raw)
+        adjacency: dict[int, set[int]] = defaultdict(set)
+        for a, b in edges_raw:
+            adjacency[a].add(b)
+            adjacency[b].add(a)
+
+        arterial_rows = set(self._arterial_indices(rows))
+        arterial_cols = set(self._arterial_indices(cols))
+        row_profile = self._smooth_drop_profile(rows, rng)
+        col_profile = self._smooth_drop_profile(cols, rng)
+        base_drop_prob = 0.11
+
+        interior_rows = [r for r in range(2, rows - 2) if r not in arterial_rows]
+        interior_cols = [c for c in range(2, cols - 2) if c not in arterial_cols]
+        row_gap_rows = set(
+            rng.sample(interior_rows, k=min(max(1, rows // 7), len(interior_rows)))
+        ) if interior_rows else set()
+        col_gap_cols = set(
+            rng.sample(interior_cols, k=min(max(1, cols // 7), len(interior_cols)))
+        ) if interior_cols else set()
+
+        arterial_pairs: set[frozenset[str]] = set()
+        removable: list[tuple[int, int]] = []
+        for a, b in edges_raw:
+            protected, arterial = self._is_edge_protected(
+                a=a,
+                b=b,
+                rows=rows,
+                cols=cols,
+                arterial_rows=arterial_rows,
+                arterial_cols=arterial_cols,
+            )
+            if arterial:
+                ida = f"i_{a:04d}"
+                idb = f"i_{b:04d}"
+                arterial_pairs.add(frozenset((ida, idb)))
+            if not protected:
+                removable.append((a, b))
+
+        rng.shuffle(removable)
+        for a, b in removable:
+            orientation, major_idx, minor_idx = self._edge_orientation(a, b, cols)
+            if orientation == "horizontal":
+                drop_prob = base_drop_prob + (0.11 * row_profile[major_idx]) + (
+                    0.08 * col_profile[minor_idx]
+                )
+                if major_idx in row_gap_rows:
+                    drop_prob += 0.16
+            else:
+                drop_prob = base_drop_prob + (0.11 * col_profile[major_idx]) + (
+                    0.08 * row_profile[minor_idx]
+                )
+                if major_idx in col_gap_cols:
+                    drop_prob += 0.16
+
+            # Keep perimeter-adjacent links denser to avoid sparse fringes.
+            ra, ca = divmod(a, cols)
+            rb, cb = divmod(b, cols)
+            boundary_distance = min(
+                ra,
+                rb,
+                rows - 1 - ra,
+                rows - 1 - rb,
+                ca,
+                cb,
+                cols - 1 - ca,
+                cols - 1 - cb,
+            )
+            if boundary_distance <= 1:
+                drop_prob *= 0.45
+            elif boundary_distance == 2:
+                drop_prob *= 0.75
+
+            if rng.random() > min(0.46, max(0.0, drop_prob)):
+                continue
+            if len(adjacency[a]) <= 2 or len(adjacency[b]) <= 2:
+                continue
+            if not self._has_path_without_edge(a, b, adjacency, (a, b)):
+                continue
+
+            adjacency[a].remove(b)
+            adjacency[b].remove(a)
+            filtered.remove((a, b))
+
+        id_list = list(coords.keys())
+        edges = [(id_list[a], id_list[b]) for a, b in sorted(filtered)]
+        return coords, edges, arterial_pairs
+
+    def _arterial_local(
+        self, target_nodes: int, rng: random.Random
+    ) -> tuple[dict[str, tuple[float, float]], list[tuple[str, str]], set[frozenset[str]]]:
+        rows, cols = self._dimensions(target_nodes)
+        coords = self._grid_coords(rows, cols, spacing=130.0, jitter=8.0, rng=rng)
+        edges_raw = self._grid_edges(rows, cols)
+        arterial_rows = {rows // 3, (2 * rows) // 3}
+        arterial_cols = {cols // 3, (2 * cols) // 3}
+        id_list = list(coords.keys())
+        edges: list[tuple[str, str]] = []
+        arterial_pairs: set[frozenset[str]] = set()
+
+        for a, b in edges_raw:
+            ra, ca = divmod(a, cols)
+            rb, cb = divmod(b, cols)
+            ida, idb = id_list[a], id_list[b]
+            edges.append((ida, idb))
+            if (
+                ra in arterial_rows
+                or rb in arterial_rows
+                or ca in arterial_cols
+                or cb in arterial_cols
+            ):
+                arterial_pairs.add(frozenset((ida, idb)))
+            elif rng.random() < 0.06:
+                arterial_pairs.add(frozenset((ida, idb)))
+        return coords, edges, arterial_pairs
+
+    def _ring_road(
+        self,
+        target_nodes: int,
+        rng: random.Random,
+        ring_diagonal_keep_prob: float,
+        ring_max_diagonal_fraction: float,
+    ) -> tuple[dict[str, tuple[float, float]], list[tuple[str, str]], set[frozenset[str]]]:
+        ring_nodes = max(14, int(target_nodes * 0.22))
+        inner_nodes = max(24, target_nodes - ring_nodes)
+        rows, cols = self._dimensions(inner_nodes)
+        inner_coords = self._grid_coords(rows, cols, spacing=90.0, jitter=10.0, rng=rng)
+        inner_ids = list(inner_coords.keys())
+        inner_index = {node: idx for idx, node in enumerate(inner_ids)}
+        min_x = min(x for x, _ in inner_coords.values())
+        max_x = max(x for x, _ in inner_coords.values())
+        min_y = min(y for _, y in inner_coords.values())
+        max_y = max(y for _, y in inner_coords.values())
+        center = ((min_x + max_x) / 2.0, (min_y + max_y) / 2.0)
+        radius = max(max_x - min_x, max_y - min_y) * 0.70
+
+        def norm_edge(a: str, b: str) -> tuple[str, str]:
+            return (a, b) if a < b else (b, a)
+
+        coords: dict[str, tuple[float, float]] = dict(inner_coords)
+        edge_set: set[tuple[str, str]] = set()
+        arterial_pairs: set[frozenset[str]] = set()
+
+        for a, b in self._grid_edges(rows, cols):
+            edge_set.add(norm_edge(inner_ids[a], inner_ids[b]))
+
+        ring_ids: list[str] = []
+        start_idx = len(coords)
+        for i in range(ring_nodes):
+            angle = (2.0 * math.pi * i) / ring_nodes
+            x = center[0] + radius * math.cos(angle)
+            y = center[1] + radius * math.sin(angle)
+            rid = f"i_{start_idx + i:04d}"
+            ring_ids.append(rid)
+            coords[rid] = (x, y)
+
+        for i, rid in enumerate(ring_ids):
+            nxt = ring_ids[(i + 1) % ring_nodes]
+            edge = norm_edge(rid, nxt)
+            edge_set.add(edge)
+            arterial_pairs.add(frozenset(edge))
+
+        boundary_inner_nodes = [
+            nid
+            for nid in inner_ids
+            if ((inner_index[nid] // cols) in {0, rows - 1})
+            or ((inner_index[nid] % cols) in {0, cols - 1})
+        ]
+        spokes = max(6, ring_nodes // 3)
+        anchor_pool = boundary_inner_nodes[:] if boundary_inner_nodes else inner_ids
+        anchor_ids = [
+            anchor_pool[(idx * len(anchor_pool)) // spokes]
+            for idx in range(spokes)
+        ]
+
+        protected_spokes: set[tuple[str, str]] = set()
+        for i in range(spokes):
+            ring_node = ring_ids[(i * ring_nodes) // spokes]
+            inner_node = anchor_ids[i]
+            edge = norm_edge(ring_node, inner_node)
+            edge_set.add(edge)
+            arterial_pairs.add(frozenset(edge))
+            protected_spokes.add(edge)
+
+        # Optional non-primary diagonals: sparse extra radials + sparse interior diagonals.
+        diagonal_candidates: list[tuple[tuple[str, str], float]] = []
+        extra_spokes = max(2, min(5, ring_nodes // 4))
+        for i in range(extra_spokes):
+            ring_node = ring_ids[(i * ring_nodes) // extra_spokes]
+            inner_node = anchor_pool[(i * len(anchor_pool)) // extra_spokes]
+            edge = norm_edge(ring_node, inner_node)
+            if edge in edge_set or edge in protected_spokes:
+                continue
+            score = 1.2 + (0.003 * euclidean(coords[ring_node], coords[inner_node]))
+            diagonal_candidates.append((edge, score))
+
+        def _orientation(
+            p: tuple[float, float],
+            q: tuple[float, float],
+            r: tuple[float, float],
+        ) -> int:
+            value = ((q[1] - p[1]) * (r[0] - q[0])) - ((q[0] - p[0]) * (r[1] - q[1]))
+            if abs(value) < 1e-9:
+                return 0
+            return 1 if value > 0 else 2
+
+        def _on_segment(
+            p: tuple[float, float],
+            q: tuple[float, float],
+            r: tuple[float, float],
+        ) -> bool:
+            return (
+                min(p[0], r[0]) <= q[0] <= max(p[0], r[0])
+                and min(p[1], r[1]) <= q[1] <= max(p[1], r[1])
+            )
+
+        def _segments_intersect(
+            p1: tuple[float, float],
+            q1: tuple[float, float],
+            p2: tuple[float, float],
+            q2: tuple[float, float],
+        ) -> bool:
+            o1 = _orientation(p1, q1, p2)
+            o2 = _orientation(p1, q1, q2)
+            o3 = _orientation(p2, q2, p1)
+            o4 = _orientation(p2, q2, q1)
+            if o1 != o2 and o3 != o4:
+                return True
+            if o1 == 0 and _on_segment(p1, p2, q1):
+                return True
+            if o2 == 0 and _on_segment(p1, q2, q1):
+                return True
+            if o3 == 0 and _on_segment(p2, p1, q2):
+                return True
+            if o4 == 0 and _on_segment(p2, q1, q2):
+                return True
+            return False
+
+        def _has_nonendpoint_intersection(
+            edge: tuple[str, str],
+            other_edges: set[tuple[str, str]],
+        ) -> bool:
+            a, b = edge
+            p1, q1 = coords[a], coords[b]
+            for u, v in other_edges:
+                # Shared endpoint is expected at valid junctions.
+                if len({a, b, u, v}) < 4:
+                    continue
+                p2, q2 = coords[u], coords[v]
+                if _segments_intersect(p1, q1, p2, q2):
+                    return True
+            return False
+
+        for r in range(rows - 1):
+            for c in range(cols - 1):
+                if rng.random() > 0.05:
+                    continue
+                tl = inner_ids[(r * cols) + c]
+                tr = inner_ids[(r * cols) + c + 1]
+                bl = inner_ids[((r + 1) * cols) + c]
+                br = inner_ids[((r + 1) * cols) + c + 1]
+                a, b = (tl, br) if rng.random() < 0.5 else (tr, bl)
+                edge = norm_edge(a, b)
+                if edge in edge_set:
+                    continue
+                boundary_bonus = 0.30 if (r in {0, rows - 2} or c in {0, cols - 2}) else 0.0
+                diagonal_candidates.append((edge, 1.0 + boundary_bonus))
+
+        keep_prob = max(0.0, min(1.0, ring_diagonal_keep_prob))
+        kept_candidates: list[tuple[tuple[str, str], float]] = []
+        geometry_edges = set(edge_set)
+        for edge, score in diagonal_candidates:
+            if rng.random() > keep_prob:
+                continue
+            if _has_nonendpoint_intersection(edge, geometry_edges):
+                continue
+            kept_candidates.append((edge, score))
+            geometry_edges.add(edge)
+        if diagonal_candidates and not kept_candidates:
+            for edge, score in sorted(diagonal_candidates, key=lambda item: item[1], reverse=True):
+                if not _has_nonendpoint_intersection(edge, edge_set):
+                    kept_candidates.append((edge, score))
+                    break
+
+        for edge, _ in kept_candidates:
+            edge_set.add(edge)
+
+        # Prune weaker optional diagonals while preserving connectivity.
+        max_fraction = max(0.0, min(1.0, ring_max_diagonal_fraction))
+        max_kept = max(1, int(round(max_fraction * max(1, len(diagonal_candidates)))))
+        hard_cap = max(1, min(3, len(inner_ids) // 60))
+        max_kept = min(max_kept, hard_cap)
+        if len(kept_candidates) > max_kept:
+            adjacency: dict[str, set[str]] = {nid: set() for nid in coords}
+            for u, v in edge_set:
+                adjacency[u].add(v)
+                adjacency[v].add(u)
+            kept = len(kept_candidates)
+            for edge, _ in sorted(kept_candidates, key=lambda item: item[1]):
+                if kept <= max_kept:
+                    break
+                u, v = edge
+                if edge not in edge_set:
+                    continue
+                if len(adjacency[u]) <= 1 or len(adjacency[v]) <= 1:
+                    continue
+                if not self._is_connected_without_str_edge(u, v, adjacency, edge):
+                    continue
+                adjacency[u].remove(v)
+                adjacency[v].remove(u)
+                edge_set.remove(edge)
+                kept -= 1
+
+        edges = sorted(edge_set)
+        return coords, edges, arterial_pairs
+
+    def _mixed(
+        self, target_nodes: int, rng: random.Random
+    ) -> tuple[dict[str, tuple[float, float]], list[tuple[str, str]], set[frozenset[str]]]:
+        coords, edges, arterial_pairs = self._arterial_local(target_nodes, rng)
+        ids = list(coords.keys())
+        for _ in range(max(3, len(ids) // 20)):
+            a, b = rng.sample(ids, 2)
+            if euclidean(coords[a], coords[b]) < 220.0:
+                edge = (a, b) if a < b else (b, a)
+                if edge not in edges:
+                    edges.append(edge)
+                    if rng.random() < 0.4:
+                        arterial_pairs.add(frozenset(edge))
+        return coords, edges, arterial_pairs
+
+    def _to_adjacency(
+        self,
+        coords: dict[str, tuple[float, float]],
+        undirected_edges: list[tuple[str, str]],
+    ) -> dict[str, set[str]]:
+        adjacency: dict[str, set[str]] = {nid: set() for nid in coords}
+        for a, b in undirected_edges:
+            if a == b:
+                continue
+            adjacency[a].add(b)
+            adjacency[b].add(a)
+        return adjacency
+
+    def _build_directed_roads(
+        self,
+        coords: dict[str, tuple[float, float]],
+        undirected_edges: list[tuple[str, str]],
+        arterial_pairs: set[frozenset[str]],
+        gateway_pairs: set[frozenset[str]],
+    ) -> tuple[dict[str, RoadRecord], set[str], set[str]]:
+        roads: dict[str, RoadRecord] = {}
+        arterial_ids: set[str] = set()
+        gateway_ids: set[str] = set()
+        for a, b in undirected_edges:
+            for start, end in ((a, b), (b, a)):
+                is_arterial = frozenset((a, b)) in arterial_pairs
+                is_gateway = frozenset((a, b)) in gateway_pairs
+                if is_gateway:
+                    speed = 12.0
+                    lanes = 2
+                else:
+                    speed = 14.0 if is_arterial else 11.0
+                    lanes = 3 if is_arterial else 2
+                rid = f"r_{start}_{end}"
+                points = [
+                    {"x": round(coords[start][0], 3), "y": round(coords[start][1], 3)},
+                    {"x": round(coords[end][0], 3), "y": round(coords[end][1], 3)},
+                ]
+                record = RoadRecord(
+                    id=rid,
+                    start_intersection=start,
+                    end_intersection=end,
+                    length=euclidean(coords[start], coords[end]),
+                    speed_limit=speed,
+                    num_lanes=lanes,
+                    points=points,
+                    is_arterial=is_arterial,
+                )
+                roads[rid] = record
+                if is_arterial:
+                    arterial_ids.add(rid)
+                if is_gateway:
+                    gateway_ids.add(rid)
+        return roads, arterial_ids, gateway_ids
+
+    def _build_roadnet(
+        self,
+        coords: dict[str, tuple[float, float]],
+        adjacency: dict[str, set[str]],
+        directed_roads: dict[str, RoadRecord],
+    ) -> dict[str, list[dict[str, object]]]:
+        in_roads_by_node: dict[str, list[RoadRecord]] = defaultdict(list)
+        out_roads_by_node: dict[str, list[RoadRecord]] = defaultdict(list)
+        for road in directed_roads.values():
+            out_roads_by_node[road.start_intersection].append(road)
+            in_roads_by_node[road.end_intersection].append(road)
+
+        min_x = min(x for x, _ in coords.values())
+        max_x = max(x for x, _ in coords.values())
+        min_y = min(y for _, y in coords.values())
+        max_y = max(y for _, y in coords.values())
+        border_eps = 3.0
+
+        intersections: list[dict[str, object]] = []
+        for nid in sorted(coords):
+            x, y = coords[nid]
+            degree = len(adjacency[nid])
+            is_border = (
+                abs(x - min_x) < border_eps
+                or abs(x - max_x) < border_eps
+                or abs(y - min_y) < border_eps
+                or abs(y - max_y) < border_eps
+            )
+            # Keep boundary intersections non-virtual when they are part of the street grid.
+            # Mark only true stubs/dead-ends as virtual.
+            virtual = degree <= 1
+
+            road_links: list[dict[str, object]] = []
+            incoming = sorted(in_roads_by_node[nid], key=lambda r: r.id)
+            outgoing = sorted(out_roads_by_node[nid], key=lambda r: r.id)
+            for in_road in incoming:
+                for out_road in outgoing:
+                    if out_road.end_intersection == in_road.start_intersection:
+                        continue
+                    lane_links = []
+                    lane_count = min(in_road.num_lanes, out_road.num_lanes)
+                    for lane_idx in range(lane_count):
+                        lane_links.append(
+                            {
+                                "startLaneIndex": lane_idx,
+                                "endLaneIndex": lane_idx,
+                                "points": [
+                                    dict(in_road.points[-1]),
+                                    dict(out_road.points[0]),
+                                ],
+                            }
+                        )
+                    road_links.append(
+                        {
+                            "type": self._movement_type(
+                                coords[in_road.start_intersection],
+                                coords[nid],
+                                coords[out_road.end_intersection],
+                            ),
+                            "startRoad": in_road.id,
+                            "endRoad": out_road.id,
+                            "laneLinks": lane_links,
+                        }
+                    )
+
+            lightphases = self._light_phases(nid, coords, incoming, road_links)
+            connected_roads = sorted(
+                {road.id for road in incoming + outgoing}
+            )
+            intersections.append(
+                {
+                    "id": nid,
+                    "point": {"x": round(x, 3), "y": round(y, 3)},
+                    "width": 0,
+                    "roads": connected_roads,
+                    "virtual": virtual,
+                    "roadLinks": road_links,
+                    "trafficLight": {
+                        "roadLinkIndices": list(range(len(road_links))),
+                        "lightphases": lightphases,
+                    },
+                }
+            )
+
+        roads: list[dict[str, object]] = []
+        for rid in sorted(directed_roads):
+            road = directed_roads[rid]
+            roads.append(
+                {
+                    "id": road.id,
+                    "startIntersection": road.start_intersection,
+                    "endIntersection": road.end_intersection,
+                    "points": road.points,
+                    "lanes": [
+                        {"maxSpeed": road.speed_limit, "width": 3.2}
+                        for _ in range(road.num_lanes)
+                    ],
+                }
+            )
+        return {"intersections": intersections, "roads": roads}
+
+    def _movement_type(
+        self,
+        in_start: tuple[float, float],
+        center: tuple[float, float],
+        out_end: tuple[float, float],
+    ) -> str:
+        v1 = (center[0] - in_start[0], center[1] - in_start[1])
+        v2 = (out_end[0] - center[0], out_end[1] - center[1])
+        cross = (v1[0] * v2[1]) - (v1[1] * v2[0])
+        dot = (v1[0] * v2[0]) + (v1[1] * v2[1])
+        angle = math.atan2(cross, dot)
+        if abs(angle) < (math.pi / 4):
+            return "go_straight"
+        if angle > 0:
+            return "turn_left"
+        return "turn_right"
+
+    def _light_phases(
+        self,
+        node_id: str,
+        coords: dict[str, tuple[float, float]],
+        incoming: list[RoadRecord],
+        road_links: list[dict[str, object]],
+    ) -> list[dict[str, object]]:
+        if not road_links:
+            return [{"time": 30, "availableRoadLinks": []}]
+
+        vertical_incoming: set[str] = set()
+        horizontal_incoming: set[str] = set()
+        center = coords[node_id]
+        for road in incoming:
+            source = coords[road.start_intersection]
+            vx = center[0] - source[0]
+            vy = center[1] - source[1]
+            if abs(vy) >= abs(vx):
+                vertical_incoming.add(road.id)
+            else:
+                horizontal_incoming.add(road.id)
+
+        vertical_links: list[int] = []
+        horizontal_links: list[int] = []
+        for idx, link in enumerate(road_links):
+            start_road = str(link["startRoad"])
+            if start_road in vertical_incoming:
+                vertical_links.append(idx)
+            else:
+                horizontal_links.append(idx)
+
+        if not vertical_links or not horizontal_links:
+            return [{"time": 35, "availableRoadLinks": list(range(len(road_links)))}]
+
+        return [
+            {"time": 30, "availableRoadLinks": vertical_links},
+            {"time": 5, "availableRoadLinks": []},
+            {"time": 30, "availableRoadLinks": horizontal_links},
+            {"time": 5, "availableRoadLinks": []},
+        ]

data/generators/scenario_generator.py

@@ -0,0 +1,443 @@
+"""Scenario plan generation for demand intensity and network disturbances."""
+
+from __future__ import annotations
+
+import random
+from collections import defaultdict
+
+from .schemas import (
+    CityGraph,
+    DatasetGenerationConfig,
+    DemandIntensity,
+    DistrictData,
+    ScenarioPlan,
+    ScenarioType,
+    TripMix,
+)
+
+
+class ScenarioGenerator:
+    """Generate scenario-specific modifiers, bottlenecks, and demand intensity."""
+
+    INTENSITY_MULTIPLIER: dict[DemandIntensity, float] = {
+        "normal": 1.0,
+        "moderate_rush": 1.45,
+        "heavy_rush": 2.1,
+        "overload": 2.9,
+        "accident_overload": 3.6,
+    }
+
+    SCENARIO_BASE_MULTIPLIER: dict[ScenarioType, float] = {
+        "normal": 1.15,
+        "morning_rush": 1.35,
+        "evening_rush": 1.35,
+        "accident": 1.85,
+        "construction": 1.65,
+        "event_spike": 1.75,
+        "district_overload": 1.80,
+    }
+
+    BASE_DEMAND_PER_INTERSECTION: dict[ScenarioType, int] = {
+        "normal": 42,
+        "morning_rush": 52,
+        "evening_rush": 52,
+        "accident": 60,
+        "construction": 56,
+        "event_spike": 62,
+        "district_overload": 64,
+    }
+
+    INTENSITY_ALLOWED_BY_SCENARIO: dict[ScenarioType, list[DemandIntensity]] = {
+        "normal": ["normal", "moderate_rush", "heavy_rush"],
+        "morning_rush": ["moderate_rush", "heavy_rush", "overload"],
+        "evening_rush": ["moderate_rush", "heavy_rush", "overload"],
+        "accident": ["heavy_rush", "overload", "accident_overload"],
+        "construction": ["moderate_rush", "heavy_rush", "overload", "accident_overload"],
+        "event_spike": ["moderate_rush", "heavy_rush", "overload", "accident_overload"],
+        "district_overload": ["moderate_rush", "heavy_rush", "overload", "accident_overload"],
+    }
+
+    def generate(
+        self,
+        city_graph: CityGraph,
+        district_data: DistrictData,
+        scenario_names: list[ScenarioType],
+        base_seed: int,
+        config: DatasetGenerationConfig,
+    ) -> dict[str, ScenarioPlan]:
+        plans: dict[str, ScenarioPlan] = {}
+        for idx, name in enumerate(scenario_names):
+            seed = base_seed + (idx * 101)
+            rng = random.Random(seed)
+            intensity = self._sample_intensity(name, rng, config)
+            trip_multiplier = self._trip_multiplier(name, intensity, config)
+            trip_mix = self._trip_mix(name, intensity)
+            departure_windows = self._departure_windows(name, intensity)
+            base_demand = self.BASE_DEMAND_PER_INTERSECTION[name]
+
+            if name == "normal":
+                plans[name] = ScenarioPlan(
+                    name=name,
+                    intensity=intensity,
+                    seed=seed,
+                    trip_multiplier=trip_multiplier,
+                    trip_mix=trip_mix,
+                    departure_windows=departure_windows,
+                    metadata={
+                        "description": "Balanced baseline traffic with sampled intensity.",
+                        "base_demand_per_intersection": base_demand,
+                    },
+                )
+            elif name == "morning_rush":
+                plans[name] = ScenarioPlan(
+                    name=name,
+                    intensity=intensity,
+                    seed=seed,
+                    trip_multiplier=trip_multiplier,
+                    trip_mix=trip_mix,
+                    departure_windows=departure_windows,
+                    metadata={
+                        "description": "Strong residential outbound and work-district inbound pressure.",
+                        "base_demand_per_intersection": base_demand,
+                    },
+                )
+            elif name == "evening_rush":
+                plans[name] = ScenarioPlan(
+                    name=name,
+                    intensity=intensity,
+                    seed=seed,
+                    trip_multiplier=trip_multiplier,
+                    trip_mix=trip_mix,
+                    departure_windows=departure_windows,
+                    metadata={
+                        "description": "Strong work-district outbound and residential inbound pressure.",
+                        "base_demand_per_intersection": base_demand,
+                    },
+                )
+            elif name == "accident":
+                blocked, penalized = self._accident_impairments(
+                    city_graph, district_data, intensity, rng
+                )
+                plans[name] = ScenarioPlan(
+                    name=name,
+                    intensity=intensity,
+                    seed=seed,
+                    trip_multiplier=trip_multiplier,
+                    trip_mix=trip_mix,
+                    departure_windows=departure_windows,
+                    blocked_roads=blocked,
+                    penalized_roads=penalized,
+                    metadata={
+                        "description": "Severe disruption on connector/arterial corridors.",
+                        "base_demand_per_intersection": base_demand,
+                        "accident_roads": sorted(blocked),
+                        "bottleneck_penalties": {k: v for k, v in penalized.items() if k not in blocked},
+                    },
+                )
+            elif name == "construction":
+                blocked, penalized = self._construction_impairments(
+                    city_graph, district_data, intensity, rng
+                )
+                plans[name] = ScenarioPlan(
+                    name=name,
+                    intensity=intensity,
+                    seed=seed,
+                    trip_multiplier=trip_multiplier,
+                    trip_mix=trip_mix,
+                    departure_windows=departure_windows,
+                    blocked_roads=blocked,
+                    penalized_roads=penalized,
+                    metadata={
+                        "description": "Localized but severe construction bottlenecks.",
+                        "base_demand_per_intersection": base_demand,
+                        "construction_roads": sorted(blocked),
+                        "bottleneck_penalties": {k: v for k, v in penalized.items() if k not in blocked},
+                    },
+                )
+            elif name == "event_spike":
+                event_district = self._pick_high_pressure_district(district_data, rng)
+                plans[name] = ScenarioPlan(
+                    name=name,
+                    intensity=intensity,
+                    seed=seed,
+                    trip_multiplier=trip_multiplier,
+                    trip_mix=trip_mix,
+                    departure_windows=departure_windows,
+                    event_district=event_district,
+                    metadata={
+                        "description": "Pre-event surge and outbound release around event district.",
+                        "base_demand_per_intersection": base_demand,
+                        "event_district": event_district,
+                    },
+                )
+            elif name == "district_overload":
+                overload = self._pick_high_pressure_district(district_data, rng)
+                plans[name] = ScenarioPlan(
+                    name=name,
+                    intensity=intensity,
+                    seed=seed,
+                    trip_multiplier=trip_multiplier,
+                    trip_mix=trip_mix,
+                    departure_windows=departure_windows,
+                    overload_district=overload,
+                    metadata={
+                        "description": "One district receives amplified production and attraction.",
+                        "base_demand_per_intersection": base_demand,
+                        "overload_district": overload,
+                    },
+                )
+            else:
+                raise ValueError(f"Unsupported scenario: {name}")
+        return plans
+
+    def _sample_intensity(
+        self,
+        scenario: ScenarioType,
+        rng: random.Random,
+        config: DatasetGenerationConfig,
+    ) -> DemandIntensity:
+        allowed = self.INTENSITY_ALLOWED_BY_SCENARIO[scenario]
+        candidates = [i for i in config.intensity_levels if i in allowed]
+        if not candidates:
+            candidates = allowed
+        scenario_bias: dict[ScenarioType, dict[DemandIntensity, float]] = {
+            "normal": {"normal": 1.35},
+            "morning_rush": {"heavy_rush": 1.4, "overload": 1.45},
+            "evening_rush": {"heavy_rush": 1.4, "overload": 1.45},
+            "accident": {"overload": 1.8, "accident_overload": 2.2},
+            "construction": {"heavy_rush": 1.3, "overload": 1.7, "accident_overload": 2.0},
+            "event_spike": {"heavy_rush": 1.35, "overload": 1.65, "accident_overload": 1.8},
+            "district_overload": {"heavy_rush": 1.35, "overload": 1.75, "accident_overload": 1.95},
+        }
+        bias = scenario_bias.get(scenario, {})
+        weights = [
+            max(0.0, config.intensity_distribution.get(i, 0.0)) * bias.get(i, 1.0)
+            for i in candidates
+        ]
+        if sum(weights) <= 0.0:
+            weights = [1.0] * len(candidates)
+        return rng.choices(candidates, weights=weights, k=1)[0]
+
+    def _trip_multiplier(
+        self,
+        scenario: ScenarioType,
+        intensity: DemandIntensity,
+        config: DatasetGenerationConfig,
+    ) -> float:
+        base = self.SCENARIO_BASE_MULTIPLIER[scenario]
+        intensity_scale = self.INTENSITY_MULTIPLIER[intensity]
+        per_scenario = config.scenario_demand_multipliers.get(scenario, 1.0)
+        return base * intensity_scale * config.global_demand_multiplier * per_scenario
+
+    def _trip_mix(
+        self,
+        scenario: ScenarioType,
+        intensity: DemandIntensity,
+    ) -> TripMix:
+        base: dict[ScenarioType, tuple[float, float, float]] = {
+            "normal": (0.44, 0.34, 0.22),
+            "morning_rush": (0.34, 0.38, 0.28),
+            "evening_rush": (0.34, 0.38, 0.28),
+            "accident": (0.28, 0.40, 0.32),
+            "construction": (0.30, 0.40, 0.30),
+            "event_spike": (0.24, 0.40, 0.36),
+            "district_overload": (0.26, 0.42, 0.32),
+        }
+        intra, adjacent, long = base[scenario]
+        intensity_shift = {
+            "normal": 0.00,
+            "moderate_rush": 0.03,
+            "heavy_rush": 0.06,
+            "overload": 0.09,
+            "accident_overload": 0.12,
+        }[intensity]
+        intra = max(0.14, intra - intensity_shift)
+        adjacent = min(0.56, adjacent + (0.55 * intensity_shift))
+        long = min(0.44, long + (0.45 * intensity_shift))
+        norm = intra + adjacent + long
+        return TripMix(
+            intra_district=intra / norm,
+            adjacent_district=adjacent / norm,
+            long_distance=long / norm,
+        )
+
+    def _departure_windows(
+        self,
+        scenario: ScenarioType,
+        intensity: DemandIntensity,
+    ) -> list[tuple[float, float, float]]:
+        compression = {
+            "normal": 1.00,
+            "moderate_rush": 0.82,
+            "heavy_rush": 0.62,
+            "overload": 0.46,
+            "accident_overload": 0.35,
+        }[intensity]
+
+        if scenario == "morning_rush":
+            peak_width = 0.34 * compression
+            peak_start = max(0.07, 0.26 - peak_width / 2.0)
+            peak_end = min(0.58, peak_start + peak_width)
+            return [(0.0, peak_start, 0.12), (peak_start, peak_end, 0.76), (peak_end, 1.0, 0.12)]
+        if scenario == "evening_rush":
+            peak_width = 0.34 * compression
+            peak_end = min(0.95, 0.74 + peak_width / 2.0)
+            peak_start = max(0.34, peak_end - peak_width)
+            return [(0.0, peak_start, 0.10), (peak_start, peak_end, 0.76), (peak_end, 1.0, 0.14)]
+        if scenario in {"event_spike", "district_overload"}:
+            peak_width = 0.42 * compression
+            peak_start = max(0.20, 0.52 - peak_width / 2.0)
+            peak_end = min(0.90, peak_start + peak_width)
+            return [(0.0, peak_start, 0.10), (peak_start, peak_end, 0.74), (peak_end, 1.0, 0.16)]
+        if scenario in {"accident", "construction"}:
+            peak_width = 0.45 * compression
+            peak_start = max(0.16, 0.48 - peak_width / 2.0)
+            peak_end = min(0.88, peak_start + peak_width)
+            return [(0.0, peak_start, 0.16), (peak_start, peak_end, 0.68), (peak_end, 1.0, 0.16)]
+        return [(0.0, 0.28, 0.22), (0.28, 0.72, 0.56), (0.72, 1.0, 0.22)]
+
+    def _road_importance_scores(
+        self,
+        city_graph: CityGraph,
+        district_data: DistrictData,
+    ) -> dict[str, float]:
+        boundary_nodes = set(district_data.boundary_intersections)
+        scores: dict[str, float] = {}
+        for road_id, road in city_graph.directed_roads.items():
+            score = 1.0
+            if road_id in city_graph.arterial_roads:
+                score += 3.0
+            if road_id in district_data.inter_district_roads:
+                score += 2.5
+            if road.start_intersection in boundary_nodes or road.end_intersection in boundary_nodes:
+                score += 1.8
+            score += 0.30 * len(city_graph.adjacency[road.start_intersection])
+            score += 0.30 * len(city_graph.adjacency[road.end_intersection])
+            scores[road_id] = score
+        return scores
+
+    def _weighted_sample_without_replacement(
+        self,
+        candidates: list[str],
+        weights: dict[str, float],
+        k: int,
+        rng: random.Random,
+    ) -> list[str]:
+        remaining = list(candidates)
+        picked: list[str] = []
+        k = min(k, len(remaining))
+        while remaining and len(picked) < k:
+            probs = [max(0.01, weights.get(c, 0.01)) for c in remaining]
+            choice = rng.choices(remaining, weights=probs, k=1)[0]
+            picked.append(choice)
+            remaining.remove(choice)
+        return picked
+
+    def _accident_impairments(
+        self,
+        city_graph: CityGraph,
+        district_data: DistrictData,
+        intensity: DemandIntensity,
+        rng: random.Random,
+    ) -> tuple[set[str], dict[str, float]]:
+        importance = self._road_importance_scores(city_graph, district_data)
+        ranked = sorted(importance.keys(), key=lambda rid: importance[rid], reverse=True)
+        block_count = {
+            "normal": 1,
+            "moderate_rush": 2,
+            "heavy_rush": 2,
+            "overload": 3,
+            "accident_overload": 4,
+        }[intensity]
+        blocked = set(self._weighted_sample_without_replacement(ranked[:120], importance, block_count, rng))
+        severity = {
+            "normal": 7.0,
+            "moderate_rush": 8.5,
+            "heavy_rush": 10.5,
+            "overload": 12.5,
+            "accident_overload": 14.5,
+        }[intensity]
+        penalized: dict[str, float] = {rid: severity for rid in blocked}
+
+        # Expand impairment to adjacent connector roads to create spillback.
+        by_intersection: dict[str, set[str]] = defaultdict(set)
+        for road_id, road in city_graph.directed_roads.items():
+            by_intersection[road.start_intersection].add(road_id)
+            by_intersection[road.end_intersection].add(road_id)
+        for rid in blocked:
+            road = city_graph.directed_roads[rid]
+            nearby = by_intersection[road.start_intersection] | by_intersection[road.end_intersection]
+            for neighbor in nearby:
+                if neighbor in blocked:
+                    continue
+                if neighbor in district_data.inter_district_roads or neighbor in city_graph.arterial_roads:
+                    penalized[neighbor] = max(penalized.get(neighbor, 0.0), severity * 0.78)
+        return blocked, penalized
+
+    def _construction_impairments(
+        self,
+        city_graph: CityGraph,
+        district_data: DistrictData,
+        intensity: DemandIntensity,
+        rng: random.Random,
+    ) -> tuple[set[str], dict[str, float]]:
+        candidate_district = self._pick_high_pressure_district(district_data, rng)
+        members = set(district_data.districts[candidate_district].intersections)
+        localized = [
+            rid
+            for rid, road in city_graph.directed_roads.items()
+            if road.start_intersection in members or road.end_intersection in members
+        ]
+        if not localized:
+            localized = sorted(city_graph.directed_roads.keys())
+
+        importance = self._road_importance_scores(city_graph, district_data)
+        block_count = {
+            "normal": 1,
+            "moderate_rush": 1,
+            "heavy_rush": 2,
+            "overload": 3,
+            "accident_overload": 4,
+        }[intensity]
+        penalize_count = {
+            "normal": 10,
+            "moderate_rush": 16,
+            "heavy_rush": 22,
+            "overload": 30,
+            "accident_overload": 36,
+        }[intensity]
+        blocked = set(
+            self._weighted_sample_without_replacement(localized, importance, block_count, rng)
+        )
+        penalize_candidates = self._weighted_sample_without_replacement(
+            localized, importance, penalize_count, rng
+        )
+        severity = {
+            "normal": 5.0,
+            "moderate_rush": 6.5,
+            "heavy_rush": 8.5,
+            "overload": 10.5,
+            "accident_overload": 12.5,
+        }[intensity]
+        penalized: dict[str, float] = {}
+        for rid in penalize_candidates:
+            factor = severity
+            if rid in blocked:
+                factor = severity * 1.35
+            penalized[rid] = max(penalized.get(rid, 0.0), factor)
+        for rid in blocked:
+            penalized[rid] = max(penalized.get(rid, 0.0), severity * 1.55)
+        return blocked, penalized
+
+    def _pick_high_pressure_district(
+        self,
+        district_data: DistrictData,
+        rng: random.Random,
+    ) -> str:
+        district_ids = sorted(district_data.districts.keys())
+        weights: list[float] = []
+        for did in district_ids:
+            district = district_data.districts[did]
+            w = 1.0 + 0.20 * len(district.neighbors) + 0.06 * len(district.boundary_intersections)
+            weights.append(w)
+        return rng.choices(district_ids, weights=weights, k=1)[0]

data/generators/schemas.py

@@ -0,0 +1,177 @@
+"""Typed schemas and dataclasses for synthetic CityFlow generation."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Literal
+
+TopologyType = Literal[
+    "rectangular_grid",
+    "irregular_grid",
+    "arterial_local",
+    "ring_road",
+    "mixed",
+]
+
+DistrictType = Literal["residential", "commercial", "industrial", "mixed"]
+DemandIntensity = Literal[
+    "normal",
+    "moderate_rush",
+    "heavy_rush",
+    "overload",
+    "accident_overload",
+]
+ScenarioType = Literal[
+    "normal",
+    "morning_rush",
+    "evening_rush",
+    "accident",
+    "construction",
+    "event_spike",
+    "district_overload",
+]
+
+
+@dataclass(slots=True, frozen=True)
+class TripMix:
+    """Trip category distribution."""
+
+    intra_district: float = 0.5
+    adjacent_district: float = 0.3
+    long_distance: float = 0.2
+
+
+@dataclass(slots=True)
+class DatasetGenerationConfig:
+    """Top-level CLI / generation configuration."""
+
+    num_cities: int
+    output_dir: Path
+    seed: int = 42
+    min_districts: int = 6
+    max_districts: int = 20
+    min_intersections_per_district: int = 4
+    max_intersections_per_district: int = 10
+    topologies: list[TopologyType] = field(
+        default_factory=lambda: ["irregular_grid"]
+    )
+    scenarios: list[ScenarioType] = field(
+        default_factory=lambda: [
+            "normal",
+            "morning_rush",
+            "evening_rush",
+            "accident",
+            "construction",
+            "event_spike",
+            "district_overload",
+        ]
+    )
+    intensity_levels: list[DemandIntensity] = field(
+        default_factory=lambda: [
+            "normal",
+            "moderate_rush",
+            "heavy_rush",
+            "overload",
+            "accident_overload",
+        ]
+    )
+    intensity_distribution: dict[DemandIntensity, float] = field(
+        default_factory=lambda: {
+            "normal": 0.20,
+            "moderate_rush": 0.42,
+            "heavy_rush": 0.24,
+            "overload": 0.10,
+            "accident_overload": 0.04,
+        }
+    )
+    global_demand_multiplier: float = 1.25
+    scenario_demand_multipliers: dict[str, float] = field(
+        default_factory=lambda: {
+            "normal": 1.15,
+            "morning_rush": 1.35,
+            "evening_rush": 1.35,
+            "accident": 1.75,
+            "construction": 1.55,
+            "event_spike": 1.65,
+            "district_overload": 1.70,
+        }
+    )
+    ring_diagonal_keep_prob: float = 0.07
+    ring_max_diagonal_fraction: float = 0.03
+    simulation_steps: int = 3600
+    interval: float = 1.0
+    save_replay: bool = False
+    fail_fast: bool = False
+
+
+@dataclass(slots=True, frozen=True)
+class RoadRecord:
+    """Directed road edge record."""
+
+    id: str
+    start_intersection: str
+    end_intersection: str
+    length: float
+    speed_limit: float
+    num_lanes: int
+    points: list[dict[str, float]]
+    is_arterial: bool
+
+
+@dataclass(slots=True)
+class CityGraph:
+    """Intermediate graph representation for generation."""
+
+    city_id: str
+    topology: TopologyType
+    seed: int
+    intersections: dict[str, tuple[float, float]]
+    adjacency: dict[str, set[str]]
+    directed_roads: dict[str, RoadRecord]
+    roadnet: dict[str, Any]
+    arterial_roads: set[str]
+    gateway_intersections: set[str] = field(default_factory=set)
+    gateway_roads: set[str] = field(default_factory=set)
+    inter_district_roads: set[str] = field(default_factory=set)
+
+
+@dataclass(slots=True)
+class DistrictRecord:
+    """District-level metadata."""
+
+    id: str
+    district_type: DistrictType
+    intersections: list[str]
+    neighbors: list[str]
+    boundary_intersections: list[str]
+    entry_roads: list[str]
+    exit_roads: list[str]
+
+
+@dataclass(slots=True)
+class DistrictData:
+    """District overlay output."""
+
+    intersection_to_district: dict[str, str]
+    districts: dict[str, DistrictRecord]
+    district_neighbors: dict[str, list[str]]
+    boundary_intersections: list[str]
+    inter_district_roads: list[str]
+
+
+@dataclass(slots=True)
+class ScenarioPlan:
+    """Scenario-specific route demand and impairment configuration."""
+
+    name: ScenarioType
+    intensity: DemandIntensity
+    seed: int
+    trip_multiplier: float
+    trip_mix: TripMix
+    departure_windows: list[tuple[float, float, float]]
+    blocked_roads: set[str] = field(default_factory=set)
+    penalized_roads: dict[str, float] = field(default_factory=dict)
+    event_district: str | None = None
+    overload_district: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)

data/generators/utils.py

@@ -0,0 +1,377 @@
+"""Utilities for deterministic generation, IO, graph operations, and validation."""
+
+from __future__ import annotations
+
+import heapq
+import json
+import math
+import random
+from collections import Counter, defaultdict, deque
+from pathlib import Path
+from typing import Any, Iterable
+
+from .schemas import CityGraph, DistrictData
+
+
+def ensure_dir(path: Path) -> None:
+    path.mkdir(parents=True, exist_ok=True)
+
+
+def write_json(path: Path, payload: Any) -> None:
+    ensure_dir(path.parent)
+    path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+
+
+def euclidean(a: tuple[float, float], b: tuple[float, float]) -> float:
+    return math.hypot(a[0] - b[0], a[1] - b[1])
+
+
+def clamp(value: int, low: int, high: int) -> int:
+    return max(low, min(high, value))
+
+
+def choose_weighted(rng: random.Random, values: list[str], weights: list[float]) -> str:
+    total = sum(weights)
+    if total <= 0:
+        return values[rng.randrange(len(values))]
+    cutoff = rng.random() * total
+    cursor = 0.0
+    for value, weight in zip(values, weights):
+        cursor += weight
+        if cursor >= cutoff:
+            return value
+    return values[-1]
+
+
+def connected_components(nodes: Iterable[str], adjacency: dict[str, set[str]]) -> list[set[str]]:
+    pending = set(nodes)
+    components: list[set[str]] = []
+    while pending:
+        root = pending.pop()
+        comp = {root}
+        queue = deque([root])
+        while queue:
+            cur = queue.popleft()
+            for nxt in adjacency[cur]:
+                if nxt in pending:
+                    pending.remove(nxt)
+                    comp.add(nxt)
+                    queue.append(nxt)
+        components.append(comp)
+    return components
+
+
+def dijkstra_shortest_path(
+    start: str,
+    end: str,
+    graph: dict[str, list[tuple[str, float, str]]],
+) -> list[str] | None:
+    """Return road-id path from start intersection to end intersection."""
+    if start == end:
+        return []
+    queue: list[tuple[float, str]] = [(0.0, start)]
+    dist: dict[str, float] = {start: 0.0}
+    prev: dict[str, tuple[str, str]] = {}
+
+    while queue:
+        cost, node = heapq.heappop(queue)
+        if node == end:
+            break
+        if cost > dist[node]:
+            continue
+        for nxt, edge_cost, road_id in graph.get(node, []):
+            candidate = cost + edge_cost
+            if candidate < dist.get(nxt, float("inf")):
+                dist[nxt] = candidate
+                prev[nxt] = (node, road_id)
+                heapq.heappush(queue, (candidate, nxt))
+
+    if end not in prev:
+        return None
+
+    route: list[str] = []
+    cursor = end
+    while cursor != start:
+        parent, road_id = prev[cursor]
+        route.append(road_id)
+        cursor = parent
+    route.reverse()
+    return route
+
+
+def validate_unique_ids(city_graph: CityGraph) -> None:
+    intersection_ids = set(city_graph.intersections.keys())
+    if len(intersection_ids) != len(city_graph.intersections):
+        raise ValueError("Duplicate intersection IDs found.")
+    road_ids = set(city_graph.directed_roads.keys())
+    if len(road_ids) != len(city_graph.directed_roads):
+        raise ValueError("Duplicate road IDs found.")
+
+
+def validate_district_contiguity(city_graph: CityGraph, district_data: DistrictData) -> None:
+    by_district: dict[str, set[str]] = defaultdict(set)
+    for intersection_id, district_id in district_data.intersection_to_district.items():
+        by_district[district_id].add(intersection_id)
+
+    for district_id, members in by_district.items():
+        if not members:
+            raise ValueError(f"District {district_id} has no intersections.")
+        components = connected_components(members, city_graph.adjacency)
+        if len(components) != 1:
+            raise ValueError(f"District {district_id} is not contiguous.")
+
+
+def validate_inter_district_connectivity(district_data: DistrictData) -> None:
+    connected = sum(1 for roads in district_data.district_neighbors.values() if roads)
+    if connected == 0:
+        raise ValueError("No inter-district connections found.")
+
+
+def validate_district_exit_capacity(
+    district_data: DistrictData,
+    min_exit_roads: int = 2,
+    min_entry_roads: int = 2,
+    min_neighbor_districts: int = 1,
+) -> None:
+    underconnected_exit: list[str] = []
+    underconnected_entry: list[str] = []
+    underconnected_neighbors: list[str] = []
+    for district_id, district in district_data.districts.items():
+        if len(district.exit_roads) < min_exit_roads:
+            underconnected_exit.append(
+                f"{district_id}:{len(district.exit_roads)}"
+            )
+        if len(district.entry_roads) < min_entry_roads:
+            underconnected_entry.append(
+                f"{district_id}:{len(district.entry_roads)}"
+            )
+        if len(district.neighbors) < min_neighbor_districts:
+            underconnected_neighbors.append(
+                f"{district_id}:{len(district.neighbors)}"
+            )
+    if underconnected_exit or underconnected_entry or underconnected_neighbors:
+        parts: list[str] = []
+        if underconnected_exit:
+            parts.append(
+                f"exit<{min_exit_roads}: " + ", ".join(underconnected_exit[:8])
+            )
+        if underconnected_entry:
+            parts.append(
+                f"entry<{min_entry_roads}: " + ", ".join(underconnected_entry[:8])
+            )
+        if underconnected_neighbors:
+            parts.append(
+                f"neighbors<{min_neighbor_districts}: "
+                + ", ".join(underconnected_neighbors[:8])
+            )
+        raise ValueError(
+            "District external connectivity too low: " + " | ".join(parts)
+        )
+
+
+def validate_routes(
+    flow_entries: list[dict[str, Any]],
+    roads_by_id: dict[str, Any],
+) -> None:
+    if not flow_entries:
+        raise ValueError("Scenario flow is empty.")
+    for idx, vehicle in enumerate(flow_entries):
+        route = vehicle.get("route", [])
+        if not route:
+            raise ValueError(f"Flow entry {idx} has empty route.")
+        for road_id in route:
+            if road_id not in roads_by_id:
+                raise ValueError(f"Flow entry {idx} references missing road {road_id}.")
+        for left, right in zip(route, route[1:]):
+            left_end = roads_by_id[left]["endIntersection"]
+            right_start = roads_by_id[right]["startIntersection"]
+            if left_end != right_start:
+                raise ValueError(
+                    f"Invalid route transition: {left} -> {right} in entry {idx}."
+                )
+
+
+def build_road_index(roadnet: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    return {road["id"]: road for road in roadnet.get("roads", [])}
+
+
+def build_roadlink_index(
+    roadnet: dict[str, Any],
+) -> dict[str, set[tuple[str, str]]]:
+    roadlinks_by_intersection: dict[str, set[tuple[str, str]]] = defaultdict(set)
+    for intersection in roadnet.get("intersections", []):
+        if intersection.get("virtual", False):
+            # CityFlow ignores roadLinks on virtual intersections.
+            continue
+        iid = intersection["id"]
+        for road_link in intersection.get("roadLinks", []):
+            pair = (road_link["startRoad"], road_link["endRoad"])
+            roadlinks_by_intersection[iid].add(pair)
+    return roadlinks_by_intersection
+
+
+def validate_route_with_reasons(
+    route: list[str],
+    roads_by_id: dict[str, dict[str, Any]],
+    roadlinks_by_intersection: dict[str, set[tuple[str, str]]],
+) -> list[str]:
+    reasons: list[str] = []
+    if not route:
+        return ["empty_route"]
+    if len(route) < 2:
+        return ["route_too_short"]
+
+    for rid in route:
+        if rid not in roads_by_id:
+            reasons.append(f"missing_road:{rid}")
+            return reasons
+
+    for left, right in zip(route, route[1:]):
+        left_road = roads_by_id[left]
+        right_road = roads_by_id[right]
+        shared_intersection = left_road["endIntersection"]
+        if shared_intersection != right_road["startIntersection"]:
+            reasons.append("mismatched_intersection_transition")
+            continue
+        if (left, right) not in roadlinks_by_intersection.get(shared_intersection, set()):
+            reasons.append("missing_roadlink_transition")
+    return reasons
+
+
+def summarize_route_validation(
+    flow_entries: list[dict[str, Any]],
+    roads_by_id: dict[str, dict[str, Any]],
+    roadlinks_by_intersection: dict[str, set[tuple[str, str]]],
+) -> dict[str, Any]:
+    reason_counter: Counter[str] = Counter()
+    total = len(flow_entries)
+    valid = 0
+    invalid = 0
+    for flow in flow_entries:
+        reasons = validate_route_with_reasons(
+            route=flow.get("route", []),
+            roads_by_id=roads_by_id,
+            roadlinks_by_intersection=roadlinks_by_intersection,
+        )
+        if reasons:
+            invalid += 1
+            reason_counter.update(reasons)
+        else:
+            valid += 1
+    return {
+        "total_routes": total,
+        "valid_routes": valid,
+        "invalid_routes": invalid,
+        "top_failure_reasons": reason_counter.most_common(10),
+    }
+
+
+def compute_scenario_diagnostics(
+    flow_entries: list[dict[str, Any]],
+    city_graph: CityGraph,
+    district_data: DistrictData,
+) -> dict[str, Any]:
+    roads_by_id = build_road_index(city_graph.roadnet)
+    assignment = district_data.intersection_to_district
+    road_usage: Counter[str] = Counter()
+    origin_counter: Counter[str] = Counter()
+    destination_counter: Counter[str] = Counter()
+    corridor_counter: Counter[str] = Counter()
+    external_origin = 0
+    external_destination = 0
+
+    for flow in flow_entries:
+        route = flow.get("route", [])
+        if not route:
+            continue
+        first = roads_by_id.get(route[0])
+        last = roads_by_id.get(route[-1])
+        if first:
+            origin_key = assignment.get(first["startIntersection"], "external")
+            origin_counter[origin_key] += 1
+            if origin_key == "external":
+                external_origin += 1
+        if last:
+            destination_key = assignment.get(last["endIntersection"], "external")
+            destination_counter[destination_key] += 1
+            if destination_key == "external":
+                external_destination += 1
+        for road_id in route:
+            road_usage[road_id] += 1
+            road = roads_by_id[road_id]
+            ds = assignment.get(road["startIntersection"], "external")
+            de = assignment.get(road["endIntersection"], "external")
+            if ds != de:
+                corridor_counter[f"{ds}->{de}"] += 1
+
+    total_routes = len(flow_entries)
+    total_road_traversals = sum(road_usage.values())
+    total_roads = len(roads_by_id)
+    used_roads = len(road_usage)
+    unused_roads = total_roads - used_roads
+    boundary_roads = set(district_data.inter_district_roads)
+    gateway_roads = set(city_graph.gateway_roads)
+    boundary_usage = sum(
+        count for road_id, count in road_usage.items() if road_id in boundary_roads
+    )
+    gateway_usage = sum(
+        count for road_id, count in road_usage.items() if road_id in gateway_roads
+    )
+    top_road_usage = road_usage.most_common(15)
+    top_corridors = corridor_counter.most_common(10)
+    total_lanes = sum(road.num_lanes for road in city_graph.directed_roads.values())
+    demand_per_lane = total_routes / max(1.0, float(total_lanes))
+    avg_route_len = total_road_traversals / max(1.0, float(total_routes))
+    concentration = (
+        sum(count for _, count in top_road_usage[:10]) / max(1.0, float(total_road_traversals))
+    )
+    boundary_share = boundary_usage / max(1.0, float(total_road_traversals))
+    congestion_score = min(
+        100.0,
+        1.9 * demand_per_lane
+        + 2.3 * avg_route_len
+        + 46.0 * concentration
+        + 34.0 * boundary_share,
+    )
+    if congestion_score < 30.0:
+        congestion_level = "manageable"
+    elif congestion_score < 52.0:
+        congestion_level = "moderate"
+    elif congestion_score < 72.0:
+        congestion_level = "heavy"
+    else:
+        congestion_level = "extreme"
+
+    return {
+        "total_vehicles": total_routes,
+        "vehicles_by_origin_district": dict(origin_counter),
+        "vehicles_by_destination_district": dict(destination_counter),
+        "vehicles_from_external": external_origin,
+        "vehicles_to_external": external_destination,
+        "roads_used": used_roads,
+        "unused_roads": unused_roads,
+        "boundary_road_usage": boundary_usage,
+        "gateway_road_usage": gateway_usage,
+        "boundary_road_share": round(boundary_share, 4),
+        "top_used_roads": [
+            {
+                "road_id": road_id,
+                "traversals": count,
+                "is_arterial": road_id in city_graph.arterial_roads,
+                "is_inter_district": road_id in boundary_roads,
+                "is_gateway": road_id in gateway_roads,
+            }
+            for road_id, count in top_road_usage
+        ],
+        "top_used_corridors": [
+            {"corridor": corridor, "traversals": count}
+            for corridor, count in top_corridors
+        ],
+        "estimated_congestion_intensity": {
+            "score": round(congestion_score, 3),
+            "level": congestion_level,
+            "demand_per_lane": round(demand_per_lane, 3),
+            "avg_route_length": round(avg_route_len, 3),
+            "road_usage_concentration": round(concentration, 4),
+        },
+    }

data/generators/validate_routes.py

@@ -0,0 +1,89 @@
+"""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())