Sync DemoService 347d491b75d7ea205a93c78d594520ed50d1d31a

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

@@ -11,6 +11,7 @@ from fastapi import APIRouter, HTTPException, status
 from pydantic import BaseModel, Field, field_validator
 
 from app.core.config import settings
+from app.services.gpx_shape_similarity import best_gpx_shape_match
 from app.services.walking_route_fallback import generate_walking_fallback
 
 router = APIRouter(prefix="/ai", tags=["ai"])
@@ -422,7 +423,8 @@ def _run_fast_road_pipeline(
     # A small safety margin is enough for the live service. Generating a
     # larger pool on the CPU-only Space makes a five-route request look like a
     # failed request before it can return any usable road loop.
-    candidate_pool = min(8, max(route_count, route_count + 1))
+    examples = getattr(module, "GPX_SHAPE_TEMPLATES", {}).get(selected_label, [])
+    candidate_pool = min(16 if examples else 8, max(route_count, route_count + (11 if examples else 1)))
     candidates = build_candidates(
         graph=graph,
         route_count=candidate_pool,
@@ -456,13 +458,14 @@ def _run_fast_road_pipeline(
         raise RuntimeError("No walkable road routes were generated near this location.")
 
     routes: list[dict[str, Any]] = []
-    for route in candidates[:route_count]:
+    for route in candidates:
         # Simplified OSM edges can bend between intersections. Preserve those
         # bends rather than drawing straight lines across blocks or water.
         coords = _route_nodes_to_coords(graph, route["nodes"])
         if len(coords) < 2:
             continue
         shape_quality = float(route.get("shape_quality", 0.0))
+        similarity, example_index = best_gpx_shape_match(coords, examples) if examples else (0.0, None)
         route_label = selected_label or "nearby_route"
         maneuvers = maneuvers_for_route(graph, route["nodes"]) if callable(maneuvers_for_route) else []
         routes.append(
@@ -478,6 +481,7 @@ def _run_fast_road_pipeline(
                 "is_loop": bool(route.get("is_loop")),
                 "route_quality": {
                     "art_ready": generation_mode == "fast_walk_roads" and bool(selected_label),
+                    **({"gpx_similarity": similarity, "gpx_example_index": example_index} if examples else {}),
                     "target_shape": {
                         "score": shape_quality,
                         "route_coverage": shape_quality,
@@ -492,6 +496,12 @@ def _run_fast_road_pipeline(
             }
         )
 
+    if examples:
+        routes.sort(key=lambda route: (
+            -route["route_quality"]["gpx_similarity"],
+            abs(route["length_km"] - target_distance_km),
+        ))
+        routes = routes[:route_count]
     if not routes:
         raise RuntimeError("Walkable road routes did not contain usable map coordinates.")
 
@@ -723,6 +733,7 @@ def generate_routes(payload: RouteGenerateRequest) -> dict[str, Any]:
                     route_count=route_count, target_distance_km=target_distance_km,
                     selected_label=selected_label, place=payload.place,
                     template=template_fn(selected_label) if callable(template_fn) else None,
+                    templates=getattr(module, "GPX_SHAPE_TEMPLATES", {}).get(selected_label, []),
                     generation_id=payload.generationId,
                 )
                 if real_roads:

api/ai-service/app/services/gpx_shape_similarity.py

@@ -0,0 +1,59 @@
+"""Compare a real road trace with location-free GPX outline examples."""
+
+from __future__ import annotations
+
+import math
+
+
+def _sample_outline(points: list[tuple[float, float]], count: int = 24) -> list[complex]:
+    if len(points) < 3:
+        return []
+    stroke = [complex(x, y) for x, y in points]
+    if abs(stroke[0] - stroke[-1]) > 1e-9:
+        stroke.append(stroke[0])
+    lengths = [abs(end - start) for start, end in zip(stroke, stroke[1:])]
+    perimeter = sum(lengths)
+    if perimeter <= 1e-9:
+        return []
+    sampled = []
+    segment = 0
+    passed = 0.0
+    for index in range(count):
+        target = perimeter * index / count
+        while segment < len(lengths) - 1 and passed + lengths[segment] < target:
+            passed += lengths[segment]
+            segment += 1
+        fraction = (target - passed) / lengths[segment] if lengths[segment] else 0.0
+        sampled.append(stroke[segment] + (stroke[segment + 1] - stroke[segment]) * fraction)
+    center = sum(sampled) / count
+    sampled = [point - center for point in sampled]
+    size = math.sqrt(sum(abs(point) ** 2 for point in sampled))
+    return [point / size for point in sampled] if size > 1e-9 else []
+
+
+def best_gpx_shape_match(
+    route_coords: list[tuple[float, float]] | list[list[float]],
+    templates: list[list[tuple[float, float]]],
+) -> tuple[float, int | None]:
+    """Return rotation/scale/start-point invariant similarity and example index."""
+    if len(route_coords) < 4 or not templates:
+        return 0.0, None
+    center_lat = sum(point[0] for point in route_coords) / len(route_coords)
+    longitude_scale = math.cos(math.radians(center_lat))
+    road = _sample_outline([(point[1] * longitude_scale, -point[0]) for point in route_coords])
+    if not road:
+        return 0.0, None
+    best_score, best_index = 0.0, None
+    for index, template in enumerate(templates):
+        example = _sample_outline(template)
+        if not example:
+            continue
+        for direction in (example, list(reversed(example))):
+            for shift in range(len(road)):
+                # Magnitude of the complex dot product is the best score after
+                # rotating either outline. Both traces have unit RMS size.
+                score = abs(sum(road[i] * direction[(i + shift) % len(road)].conjugate()
+                                for i in range(len(road))))
+                if score > best_score:
+                    best_score, best_index = score, index
+    return min(1.0, best_score), best_index

api/ai-service/app/services/walking_route_fallback.py

@@ -11,6 +11,8 @@ from typing import Any
 
 import requests
 
+from app.services.gpx_shape_similarity import best_gpx_shape_match
+
 PROVIDER_URL = os.getenv("WALK_ROUTE_PROVIDER_URL", "https://valhalla1.openstreetmap.de/route")
 HEADERS = {"User-Agent": "RunTheShape/1.0 (+https://github.com/hwangtutu/Tranimal_Run)"}
 _CACHE: dict[tuple, tuple[float, dict]] = {}
@@ -58,6 +60,7 @@ def generate_walking_fallback(
     *, center_lat: float, center_lon: float, route_count: int,
     target_distance_km: float, selected_label: str | None, place: str | None,
     template: list[tuple[float, float]] | None = None,
+    templates: list[list[tuple[float, float]]] | None = None,
     generation_id: str | None = None,
 ) -> dict[str, Any] | None:
     key = (round(center_lat, 5), round(center_lon, 5), route_count, target_distance_km, selected_label, generation_id)
@@ -66,24 +69,29 @@ def generate_walking_fallback(
     if cached and time.monotonic() - cached[0] < 6 * 3600:
         return {**cached[1], "place": place or cached[1]["place"]}
 
-    outline = list(template or [(0, 0), (1, 0), (1, 1), (0, 1), (0, 0)])
-    if outline[0] != outline[-1]:
-        outline.append(outline[0])
-    # Keep public requests small. Every segment is routed by the pedestrian
-    # engine; these waypoints are never returned as a fabricated road line.
-    # The public pedestrian endpoint accepts at most ten locations, including
-    # both endpoints. Longer drawing outlines must be sampled before routing.
-    if len(outline) > 10:
-        outline = [outline[round(i * (len(outline)-1) / 9)] for i in range(10)]
-    perimeter = sum(math.hypot(b[0]-a[0], b[1]-a[1]) for a, b in zip(outline, outline[1:]))
-    if perimeter <= 0:
-        return None
-    scale = target_distance_km * 1000 / perimeter / 1.2
+    examples = list(templates or [])
+    if examples:
+        # Each live request explores different GPX examples before the road
+        # provider's strict time budget expires.
+        offset = int(generation_angle(generation_id) / math.tau * len(examples))
+        examples = examples[offset:] + examples[:offset]
+    else:
+        examples = [template or [(0, 0), (1, 0), (1, 1), (0, 1), (0, 0)]]
     deadline = time.monotonic() + 16
     routes, seen = [], set()
     for attempt in range(min(route_count + 3, 8)):
-        if time.monotonic() >= deadline or len(routes) >= route_count:
+        if time.monotonic() >= deadline or (not templates and len(routes) >= route_count):
             break
+        outline = list(examples[attempt % len(examples)])
+        if outline[0] != outline[-1]:
+            outline.append(outline[0])
+        # The pedestrian provider accepts at most ten locations.
+        if len(outline) > 10:
+            outline = [outline[round(i * (len(outline)-1) / 9)] for i in range(10)]
+        perimeter = sum(math.hypot(b[0]-a[0], b[1]-a[1]) for a, b in zip(outline, outline[1:]))
+        if perimeter <= 0:
+            continue
+        scale = target_distance_km * 1000 / perimeter / 1.2
         angle = generation_angle(generation_id) + attempt * math.pi * (3 - math.sqrt(5))
         # Shorter retries also work near a coast or on a small island.
         size = scale * (1 if attempt < 3 else 0.5 if attempt < 5 else 0.2)
@@ -123,11 +131,13 @@ def generate_walking_fallback(
                 continue
             seen.add(signature)
             label = selected_label or "nearby_route"
+            similarity, example_index = best_gpx_shape_match(coords, templates) if templates else (0.0, None)
             routes.append({"image_path":"","top1":label,"top1_score":0.0,"top1_shape":0.0,
                 "selected_label":selected_label,"selected_label_score":None,"is_loop":True,
                 "length_km":length,"route_coords":coords,"center":coords[0],"maneuvers":[],
                 "estimated_route":False,"results":[{"label":label,"score":0.0,"shape":0.0}],
-                "route_quality":{"art_ready":False}})
+                "route_quality":{"art_ready":False,
+                    **({"gpx_similarity":similarity,"gpx_example_index":example_index} if templates else {})}})
         except requests.RequestException as exc:
             if not isinstance(exc, requests.HTTPError) or (exc.response is not None and exc.response.status_code in {429,502,503,504}):
                 break
@@ -135,7 +145,9 @@ def generate_walking_fallback(
             continue
     if not routes:
         return None
-    routes.sort(key=lambda route: abs(route["length_km"]-target_distance_km))
+    routes.sort(key=lambda route: (-route["route_quality"].get("gpx_similarity", 0.0),
+                                   abs(route["length_km"]-target_distance_km)))
+    routes = routes[:route_count]
     result = {"place":place or f"{center_lat:.6f},{center_lon:.6f}","source":"point",
         "candidate_count":len(routes),"requested_count":route_count,"generated_count":len(routes),
         "target_distance_km":target_distance_km,"labels":[selected_label] if selected_label else [],

api/ai-service/tests/test_gpx_shape_similarity.py

@@ -0,0 +1,26 @@
+import math
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from app.services.gpx_shape_similarity import best_gpx_shape_match
+
+
+def test_finds_closest_gpx_outline_independent_of_rotation_and_location():
+    star = [(math.cos(i * math.tau * 2 / 5), math.sin(i * math.tau * 2 / 5)) for i in range(5)]
+    star.append(star[0])
+    square = [(-1, -1), (1, -1), (1, 1), (-1, 1), (-1, -1)]
+    angle = 0.7
+    route = []
+    for x, y in star:
+        east = x * math.cos(angle) - y * math.sin(angle)
+        south = x * math.sin(angle) + y * math.cos(angle)
+        route.append([37.5 - south / 111320, 127.0 + east / (111320 * math.cos(math.radians(37.5)))])
+    score, index = best_gpx_shape_match(route, [square, star])
+    assert index == 1
+    assert score > 0.99
+
+
+def test_empty_or_short_route_has_no_match():
+    assert best_gpx_shape_match([[37.5, 127.0]], [[(0, 0), (1, 0), (0, 1)]]) == (0.0, None)

api/ai-service/tests/test_route_generation_recovery.py

@@ -113,6 +113,26 @@ def test_real_independent_roads_are_preferred_to_estimated_preview(pipeline):
     assert result == expected
 
 
+def test_gpx_guided_fast_routes_are_ranked_by_actual_road_similarity(monkeypatch):
+    candidates = [
+        {"nodes": [1, 2], "length_m": 5900, "shape_quality": 0.8, "is_loop": True},
+        {"nodes": [3, 4], "length_m": 6100, "shape_quality": 0.5, "is_loop": True},
+    ]
+    module = SimpleNamespace(
+        GPX_SHAPE_TEMPLATES={"dog": [[(0, 0), (1, 0), (0, 1)]]},
+        _build_or_get_walk_graph=lambda *_: (object(), "test"),
+        generate_candidate_routes=lambda **_: 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, _examples: (0.9 if coords[0][0] == 3 else 0.2, 0))
+    result = routes._run_fast_road_pipeline(module, place=None, center_lat=37, center_lon=127,
+        radius_km=2, route_count=2, target_distance_km=6, selected_label="dog")
+    assert [route["route_quality"]["gpx_similarity"] for route in result["routes"]] == [0.9, 0.2]
+    assert result["best_route"] == result["routes"][0]
+
+
 def test_place_request_returns_all_100_routes(pipeline):
     pipeline.setattr(routes, "_run_fast_road_pipeline", Mock(side_effect=RuntimeError("All Overpass endpoints failed")))
     result = routes.generate_routes(routes.RouteGenerateRequest(place="서귀포시", fallbackLatitude=33.25, fallbackLongitude=126.56, routeCount=100))

api/ai-service/tests/test_walking_route_fallback.py

@@ -105,3 +105,27 @@ def test_new_generation_queries_new_waypoints_but_network_retry_reuses_result(mo
     next_locations = json.loads(get.call_args.kwargs['params']['json'])['locations']
     assert next_locations[0] == first_locations[0], 'Keep the selected starting point'
     assert next_locations[1:] != first_locations[1:], 'Actually explore different roads'
+
+
+def test_gpx_fallback_checks_multiple_examples_and_returns_best_match(monkeypatch):
+    import json
+    def different_route(*_args, **_kwargs):
+        index = different_route.calls
+        different_route.calls += 1
+        shape = encode([[37.5, 127.0], [37.501 + index * 0.001, 127.002], [37.5, 127.0]])
+        return SimpleNamespace(raise_for_status=lambda: None,
+            json=lambda: {'trip': {'summary': {'length': 0.5, 'has_ferry': False},
+                'legs': [{'shape': shape}]}})
+    different_route.calls = 0
+    get = Mock(side_effect=different_route)
+    monkeypatch.setattr(requests, 'get', get)
+    scores = iter([(0.2, 0), (0.9, 1), (0.3, 0), (0.4, 1)])
+    monkeypatch.setattr(walking, 'best_gpx_shape_match', lambda *_: next(scores))
+    examples = [[(0, 0), (1, 0), (1, 1), (0, 0)],
+                [(0, 0), (0, 1), (1, 1), (0, 0)]]
+    result = walking.generate_walking_fallback(**options(), templates=examples, generation_id='check')
+    assert get.call_count == 4
+    locations = [json.loads(call.kwargs['params']['json'])['locations'] for call in get.call_args_list]
+    assert locations[0] != locations[1]
+    assert result['best_route']['route_quality']['gpx_similarity'] == 0.9
+    assert result['generated_count'] == 1

router/ai/scripts/clip_osm_region_test.py

@@ -786,8 +786,9 @@ def guided_geo_waypoints(
     graph: nx.MultiDiGraph,
     label: str,
     target_distance_m: float,
+    template: list[tuple[float, float]] | None = None,
 ) -> list[tuple[float, float]]:
-    template = guided_waypoints_for_label(label)
+    template = template if template is not None else guided_waypoints_for_label(label)
     if not template:
         return []
     template = diversify_guided_waypoints(template)
@@ -2408,12 +2409,15 @@ def generate_candidate_routes(
     guided_template = guided_waypoints_for_label(selected_label)
     if guided_template:
         guided_attempts = min(max(route_count * LABEL_GUIDED_ATTEMPT_MULTIPLIER, 36), 180)
-        for _ in range(guided_attempts):
+        examples = GPX_SHAPE_TEMPLATES.get(selected_label, [])
+        shuffled_examples = random.sample(examples, len(examples)) if examples else []
+        for attempt in range(guided_attempts):
             try:
                 geo_points = guided_geo_waypoints(
                     graph=graph,
                     label=str(selected_label),
                     target_distance_m=target_distance_m,
+                    template=shuffled_examples[attempt % len(shuffled_examples)] if shuffled_examples else None,
                 )
                 waypoint_nodes = nearest_nodes_for_geo_points(graph, geo_points)
                 route_nodes = build_guided_route_nodes(graph, waypoint_nodes)