Shorten checked coastal onward waypoint runs

Author: woensug-choi

README.ko.md

@@ -61,6 +61,14 @@ path = planner.route(start, goal)
 구간의 끝에서는 다음 구간이 준비될 때까지 기다립니다. 날짜 변경선에서도
 경도가 이어지도록 처리합니다.
 
+해안 경로는 다음 목표나 지역 지형의 출구까지 검증된 수역을 따라갑니다.
+끝점에는 만과 곶을 돌아가는 후속 경로가 이어집니다. 일반 해안 경유점은
+진행 방향을 안내하며, 항구와 필수 통과점은 도착 조건을 유지합니다.
+
+후속 경유점은 검증된 구간의 끝을 표시합니다. 차량이 지나가는 전체 폭의
+수심과 지도상 육지를 확인한 뒤 직선 구간의 중간 지점을 줄입니다.
+출구와 해안 선호 경로는 유지합니다.
+
 | 기준 | 기본값 |
 |---|---|
 | 수상 차량 로컬 수심 여유 | 1.5 m |

README.md

@@ -66,6 +66,10 @@ exit from the regional grid. Its endpoint has an onward route around bays and
 headlands. Ordinary coastal course vertices guide progress; ports and required
 passage points retain their arrival checks.
 
+Onward waypoints mark checked segment ends. Straight grid runs are shortened
+when the complete swept corridor passes depth and mapped-land checks. The
+shortened route preserves its exit and coastline preference.
+
 | Policy | Default |
 |---|---|
 | Surface local depth clearance | 1.5 m |

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "posim-routing"
-version = "0.1.3"
+version = "0.1.4"
 description = "Global maritime and terrain checked coastal route planning"
 requires-python = ">=3.10"
 license = {text = "Apache-2.0"}

src/posim_routing/__init__.py

@@ -1,2 +1,2 @@
 """POSIM-ROUTING: reusable ocean simulation components."""
-__version__ = "0.1.3"
+__version__ = "0.1.4"

src/posim_routing/coastal.py

@@ -259,25 +259,36 @@ def xy(node):
         nodes = nodes[:end+1]
         # Shorten the grid path without erasing its coastal preference. Long
         # offshore shortcuts across bays have a higher coast-weighted cost.
-        points, index = [[*origin, 0.]], 0
-        while index < len(nodes)-1:
-            chosen = index+1
-            for end in range(min(len(nodes)-1, index+10), index, -1):
-                a, b = xy(nodes[index]), xy(nodes[end])
-                length = math.dist(a, b)
-                count = max(2, math.ceil(length/min(dx, dy)*3))
-                ix = np.rint(np.linspace(nodes[index][1], nodes[end][1], count)).astype(int)
-                iy = np.rint(np.linspace(nodes[index][0], nodes[end][0], count)).astype(int)
-                weighted = length*float(np.mean(penalty[iy, ix]))
-                old = costs[nodes[end]]-costs[nodes[index]]
-                if safe[iy, ix].all() and weighted <= old*1.08:
-                    chosen = end
-                    break
-            points.append(xy(nodes[chosen]))
-            index = chosen
+        def shorten(route, lookahead=None):
+            points, index = [xy(route[0])], 0
+            while index < len(route)-1:
+                chosen = index+1
+                last = min(len(route)-1, index+lookahead) if lookahead else len(route)-1
+                for candidate in range(last, index, -1):
+                    if time.monotonic() > deadline:
+                        raise RuntimeError("coastal_planning_budget_exceeded")
+                    a, b = xy(route[index]), xy(route[candidate])
+                    length = math.dist(a, b)
+                    count = max(2, math.ceil(length/min(dx, dy)*3))
+                    ix = np.rint(np.linspace(route[index][1], route[candidate][1], count)).astype(int)
+                    iy = np.rint(np.linspace(route[index][0], route[candidate][0], count)).astype(int)
+                    weighted = length*float(np.mean(penalty[iy, ix]))
+                    old = costs[route[candidate]]-costs[route[index]]
+                    if (safe[iy, ix].all() and weighted <= old*1.08
+                            and validator.segment(a, b)):
+                        chosen = candidate
+                        break
+                points.append(xy(route[chosen]))
+                index = chosen
+            return points
+
+        points = shorten(nodes, lookahead=10)
+        points[0] = [*origin, 0.]
+        onward = shorten([endpoint, *onward_nodes])
         if sum(math.dist(a, b) for a, b in zip(points, points[1:])) > 40000:
             continue
-        if not all(validator.segment(a, b, record=True) for a, b in zip(points, points[1:])):
+        if not all(validator.segment(a, b, record=True) for route in (points, onward)
+                   for a, b in zip(route, route[1:])):
             continue
         return dict(points=[list(inverse.transform(*point[:2])) for point in points[1:]],
                     index=0, mode="coastal", planning_seconds=time.monotonic()-started,
@@ -286,7 +297,7 @@ def xy(node):
                     endpoint_standoff_m=float(coast_distance[endpoint]),
                     maximum_preferred_standoff_m=MAXIMUM_STANDOFF_M,
                     progress_to_course_waypoint_m=float(progress[endpoint]),
-                    onward_points=[list(inverse.transform(*xy(node)[:2])) for node in onward_nodes],
+                    onward_points=[list(inverse.transform(*point[:2])) for point in onward[1:]],
                     onward_exit=list(inverse.transform(*xy(goal)[:2])),
                     onward_exit_progress_m=float(progress[goal]),
                     onward_verified=True,

tests/test_coastal_navigation.py

@@ -95,6 +95,17 @@ def test_closed_bay_requires_an_onward_water_exit(self):
         self.assertLess(end_y,-1000)
         self.assert_swept_safe(terrain,start,dict(points=plan['points']+plan['onward_points']))
 
+    def test_long_onward_runs_use_segment_ends_and_keep_the_exit(self):
+        terrain = grid()
+        start = geo(800, -4000)
+        plan = coastal_corridor(terrain, SPEC, start, geo(6500, 9000))
+        self.assertIsNotNone(plan)
+        self.assertGreater(distance(plan['points'][-1], plan['onward_exit']), 5000.)
+        self.assertGreater(len(plan['onward_points']), 0)
+        self.assertLessEqual(len(plan['onward_points']), 4)
+        np.testing.assert_allclose(plan['onward_points'][-1], plan['onward_exit'], atol=1e-10)
+        self.assert_swept_safe(terrain, start, dict(points=plan['points']+plan['onward_points']))
+
     def test_shallow_and_unknown_coast_is_not_a_destination(self):
         terrain = grid()
         heights = np.array(terrain['heights'])

tests/test_land_exclusion.py

@@ -62,6 +62,22 @@ def test_unknown_coverage_and_missing_dataset_are_rejected(self):
             with self.assertRaisesRegex(RuntimeError,'data_unavailable'):
                 guard_from_spec(SPEC)
 
+    def test_shortened_onward_segments_keep_a_required_mapped_land_turn(self):
+        axis=np.linspace(-10000.,10000.,257);xx,yy=np.meshgrid(axis,axis)
+        grid=dict(xs=axis.tolist(),ys=axis.tolist(),heights=np.where(xx<=0,20.,-20.).tolist())
+        # Both numeric terrain and unbuffered grid vertices miss this thin bank.
+        land=shapely.union_all([shapely.box(-11000,-11000,0,11000),
+                               shapely.box(3000,8000,3002,9500)])
+        guard=LandGuard(land,shapely.box(-11000,-11000,11000,11000))
+        start=list(INVERSE.transform(800,-4000));target=list(INVERSE.transform(6500,9000))
+        plan=coastal_corridor(grid,SPEC,start,target,guard=guard)
+        self.assertIsNotNone(plan)
+        onward=[[*FORWARD.transform(*p),0.] for p in [plan['points'][-1],*plan['onward_points']]]
+        self.assertFalse(guard.segment(onward[0],onward[-1],40.))
+        self.assertGreater(len(plan['onward_points']),1)
+        self.assertTrue(all(guard.segment(a,b,40.) for a,b in zip(onward,onward[1:])))
+        np.testing.assert_allclose(plan['onward_points'][-1],plan['onward_exit'],atol=1e-10)
+
     def test_nearer_disconnected_exits_do_not_hide_reachable_water_exit(self):
         axis=np.linspace(-10000.,10000.,257)
         grid=dict(xs=axis.tolist(),ys=axis.tolist(),heights=np.full((257,257),-20.).tolist())