Sync DemoService 607dcedd38c0c5513b842dc6f9d1ccec13360b91

api/ai-service/app/api/v1/endpoints/route_generation.py

@@ -434,12 +434,21 @@ def _run_fast_road_pipeline(
         # Explore multiple physical sizes on the same road graph. The UI did
         # not request a distance; these are search scales, not length targets.
         candidates = []
+        rank_projections = getattr(module, "rank_road_template_projections", None)
         for search_distance_km in (6.0, 11.0, 18.0):
+            template_kwargs = {}
+            if selected_label == "symbol_star" and callable(rank_projections):
+                # Screen every supplied star outline against this road graph;
+                # only the best fits enter the expensive shortest-path stage.
+                template_kwargs["template_examples"] = rank_projections(
+                    graph, examples, search_distance_km * 1000.0,
+                )
             candidates.extend(build_candidates(
                 graph=graph,
                 route_count=min(6, candidate_pool),
                 target_distance_m=search_distance_km * 1000.0,
                 selected_label=selected_label,
+                **template_kwargs,
             ))
     else:
         candidates = build_candidates(

api/ai-service/tests/test_gpx_shape_templates.py

@@ -10,6 +10,7 @@ TEMPLATES = ROOT / "router" / "ai" / "data" / "gpx_shape_templates.json"
 APPROVED_TEMPLATES = ROOT / "router" / "ai" / "data" / "approved_road_shape_templates.json"
 TSHIRT_TEMPLATES = ROOT / "router" / "ai" / "data" / "tshirt_shape_templates.json"
 STAR_TEMPLATES = ROOT / "router" / "ai" / "data" / "star_shape_templates.json"
+ALL_STAR_TEMPLATES = ROOT / "router" / "ai" / "data" / "star_shape_templates_all.json"
 
 
 def test_dog_and_star_gpx_examples_are_normalized_and_closed():
@@ -63,6 +64,31 @@ def test_new_star_gpx_examples_are_available_for_road_generation():
     exec(compile(ast.Module(body=[loader], type_ignores=[]), str(script), "exec"), scope)
     loaded = scope["_load_gpx_shape_templates"]()
     assert all([tuple(point) for point in outline] in loaded["symbol_star"] for outline in outlines)
+    all_outlines = json.loads(ALL_STAR_TEMPLATES.read_text(encoding="utf-8"))["symbol_star"]
+    assert len(all_outlines) == 992
+    assert all([tuple(point) for point in outline] in loaded["symbol_star"] for outline in all_outlines)
+
+
+def test_every_star_outline_is_screened_before_road_pathfinding():
+    import math
+    import networkx as nx
+    import numpy as np
+    from scipy.spatial import cKDTree
+
+    script = ROOT / "router" / "ai" / "scripts" / "clip_osm_region_test.py"
+    tree = ast.parse(script.read_text(encoding="utf-8-sig"))
+    names = {"guided_template_perimeter", "rank_road_template_projections"}
+    functions = ast.Module(body=[node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name in names], type_ignores=[])
+    scope = {"math": math, "np": np, "cKDTree": cKDTree, "nx": nx, "list": list}
+    exec(compile(functions, str(script), "exec"), scope)
+    graph = nx.MultiDiGraph()
+    for index in range(11):
+        for other in range(11):
+            graph.add_node(index * 11 + other, x=127 + index * 0.001, y=37.5 + other * 0.001)
+    nearby = [(0, 0), (0.02, 0), (0.02, 0.02), (0, 0.02), (0, 0)]
+    distant = [(0.4, 0.4), (0.48, 0.4), (0.48, 0.48), (0.4, 0.48), (0.4, 0.4)]
+    ranked = scope["rank_road_template_projections"](graph, [distant, nearby], 1000, limit=1)
+    assert ranked == [nearby]
 
 
 def test_road_checked_examples_are_packaged_for_generation():

api/ai-service/tests/test_route_generation_recovery.py

@@ -157,6 +157,36 @@ def test_shape_priority_searches_multiple_lengths_and_ranks_by_shape(monkeypatch
     assert result["target_distance_km"] == 0
 
 
+def test_star_search_screens_all_outlines_before_road_routing(monkeypatch):
+    examples = [[(index, 0), (index, 1), (index + 1, 1), (index, 0)] for index in range(1000)]
+    screened = []
+    routed = []
+
+    def screen(_graph, templates, distance):
+        screened.append((len(templates), distance))
+        return templates[-36:]
+
+    def candidates(**kwargs):
+        routed.append(kwargs["template_examples"])
+        return [{"nodes": [1, 2], "length_m": kwargs["target_distance_m"],
+                 "shape_quality": 0.6, "is_loop": True}]
+
+    module = SimpleNamespace(
+        GPX_SHAPE_TEMPLATES={"symbol_star": examples},
+        _build_or_get_walk_graph=lambda *_: (object(), "test"),
+        rank_road_template_projections=screen,
+        generate_candidate_routes=candidates,
+        route_points=lambda *_: [], simplify_coords=lambda points: points,
+    )
+    monkeypatch.setattr(routes, "_route_nodes_to_coords", lambda _graph, nodes: [[nodes[0], 0], [nodes[1], 1]])
+    monkeypatch.setattr(routes, "best_gpx_shape_match", lambda _coords, templates: (0.9, len(templates) - 1))
+    routes._run_fast_road_pipeline(module, place=None, center_lat=37, center_lon=127,
+        radius_km=5, route_count=2, target_distance_km=6, selected_label="symbol_star", shape_priority=True)
+    assert screened == [(1000, 6000), (1000, 11000), (1000, 18000)]
+    assert len(routed) == 3
+    assert all(len(templates) == 36 and templates[-1] == examples[-1] for templates in routed)
+
+
 def test_shape_priority_uses_guided_outline_for_labels_without_gpx(monkeypatch):
     module = SimpleNamespace(
         GPX_SHAPE_TEMPLATES={},

router/ai/scripts/build_star_shape_templates.py

@@ -1,6 +1,6 @@
 """Package a bounded set of location-free outlines from a multi-track star GPX.
 
-Usage: python build_star_shape_templates.py STAR_ALL.gpx OUTPUT_JSON
+Usage: python build_star_shape_templates.py STAR_ALL.gpx OUTPUT_JSON [--all]
 """
 
 from __future__ import annotations
@@ -17,7 +17,7 @@ from build_gpx_shape_templates import normalized_outline_from_track
 MAX_TEMPLATES = 64
 
 
-def build(source: Path) -> dict[str, list[list[list[float]]]]:
+def build(source: Path, max_templates: int | None = MAX_TEMPLATES) -> dict[str, list[list[list[float]]]]:
     outlines = []
     seen = set()
     root = ET.parse(source).getroot()
@@ -34,14 +34,16 @@ def build(source: Path) -> dict[str, list[list[list[float]]]]:
             outlines.append(outline)
     if not outlines:
         raise ValueError("No star outlines found")
-    return {"symbol_star": random.Random(42).sample(outlines, min(MAX_TEMPLATES, len(outlines)))}
+    if max_templates is None:
+        return {"symbol_star": outlines}
+    return {"symbol_star": random.Random(42).sample(outlines, min(max_templates, len(outlines)))}
 
 
 if __name__ == "__main__":
-    if len(sys.argv) != 3:
-        raise SystemExit("Usage: build_star_shape_templates.py STAR_ALL.gpx OUTPUT_JSON")
+    if len(sys.argv) not in (3, 4) or (len(sys.argv) == 4 and sys.argv[3] != "--all"):
+        raise SystemExit("Usage: build_star_shape_templates.py STAR_ALL.gpx OUTPUT_JSON [--all]")
     output = Path(sys.argv[2])
-    templates = build(Path(sys.argv[1]))
+    templates = build(Path(sys.argv[1]), max_templates=None if len(sys.argv) == 4 else MAX_TEMPLATES)
     output.parent.mkdir(parents=True, exist_ok=True)
     output.write_text(json.dumps(templates, separators=(",", ":")) + "\n", encoding="utf-8")
     print({label: len(items) for label, items in templates.items()})

router/ai/scripts/clip_osm_region_test.py

@@ -18,6 +18,7 @@ import networkx as nx
 import numpy as np
 import osmnx as ox
 import requests
+from scipy.spatial import cKDTree
 from PIL import Image
 import psycopg
 from transformers import CLIPModel, CLIPProcessor
@@ -146,7 +147,7 @@ def _load_gpx_shape_templates() -> dict[str, list[list[tuple[float, float]]]]:
     data_dir = Path(__file__).resolve().parents[1] / "data"
     templates: dict[str, list[list[tuple[float, float]]]] = {"dog": [], "symbol_star": [], "t_shirt": []}
     seen: dict[str, set[tuple[tuple[float, float], ...]]] = {label: set() for label in templates}
-    for filename in ("gpx_shape_templates.json", "approved_road_shape_templates.json", "tshirt_shape_templates.json", "star_shape_templates.json"):
+    for filename in ("gpx_shape_templates.json", "approved_road_shape_templates.json", "tshirt_shape_templates.json", "star_shape_templates.json", "star_shape_templates_all.json"):
         try:
             payload = json.loads((data_dir / filename).read_text(encoding="utf-8"))
         except (OSError, ValueError, TypeError):
@@ -786,6 +787,53 @@ def guided_template_perimeter(points: list[tuple[float, float]]) -> float:
     return max(total, 1e-6)
 
 
+def rank_road_template_projections(
+    graph: nx.MultiDiGraph,
+    templates: list[list[tuple[float, float]]],
+    target_distance_m: float,
+    limit: int = 36,
+) -> list[list[tuple[float, float]]]:
+    """Project every outline cheaply, then reserve road pathfinding for the best fits.
+
+    The score measures how far outline waypoints would snap to walkable road
+    nodes at eight orientations. It is a screening score, not a routed result.
+    """
+    nodes = list(graph.nodes(data=True))
+    if not nodes or not templates:
+        return []
+    center_lat = float(np.median([float(data["y"]) for _, data in nodes]))
+    center_lon = float(np.median([float(data["x"]) for _, data in nodes]))
+    meters_per_lon = max(1.0, 111_320.0 * math.cos(math.radians(center_lat)))
+    road_xy = np.array([
+        ((float(data["x"]) - center_lon) * meters_per_lon,
+         (center_lat - float(data["y"])) * 111_320.0)
+        for _, data in nodes
+    ])
+    tree = cKDTree(road_xy)
+    ranked = []
+    for index, template in enumerate(templates):
+        body = template[:-1] if len(template) > 1 and template[0] == template[-1] else template
+        if len(body) < 4:
+            continue
+        points = np.asarray(body, dtype=float)
+        perimeter = guided_template_perimeter(template)
+        points *= target_distance_m / perimeter
+        best = float("inf")
+        for angle_index in range(8):
+            angle = angle_index * math.pi / 4
+            cos_angle, sin_angle = math.cos(angle), math.sin(angle)
+            rotated = np.column_stack((
+                points[:, 0] * cos_angle - points[:, 1] * sin_angle,
+                points[:, 0] * sin_angle + points[:, 1] * cos_angle,
+            ))
+            distances, nearest = tree.query(rotated)
+            repeated = 1.0 - len(set(nearest.tolist())) / len(nearest)
+            best = min(best, float(np.mean(distances)) + 150.0 * repeated)
+        ranked.append((best, index))
+    ranked.sort()
+    return [templates[index] for _, index in ranked[:limit]]
+
+
 def guided_geo_waypoints(
     graph: nx.MultiDiGraph,
     label: str,
@@ -2389,6 +2437,7 @@ def generate_candidate_routes(
     target_distance_m: float,
     tolerance_ratio: float = 0.35,
     selected_label: str | None = None,
+    template_examples: list[list[tuple[float, float]]] | None = None,
 ) -> list[dict[str, Any]]:
     nodes = list(graph.nodes)
     if len(nodes) < 8:
@@ -2410,10 +2459,10 @@ def generate_candidate_routes(
     candidate_edge_sets: list[set[tuple[int, int]]] = []
 
     # 1) label-guided candidates: project a rough label outline onto the OSM graph first.
-    guided_template = guided_waypoints_for_label(selected_label)
+    guided_template = template_examples[0] if template_examples else guided_waypoints_for_label(selected_label)
     if guided_template:
         guided_attempts = min(max(route_count * LABEL_GUIDED_ATTEMPT_MULTIPLIER, 36), 180)
-        examples = GPX_SHAPE_TEMPLATES.get(selected_label, [])
+        examples = template_examples if template_examples is not None else GPX_SHAPE_TEMPLATES.get(selected_label, [])
         shuffled_examples = random.sample(examples, len(examples)) if examples else []
         for attempt in range(guided_attempts):
             try: