-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdemo.py
More file actions
9356 lines (8545 loc) · 430 KB
/
Copy pathdemo.py
File metadata and controls
9356 lines (8545 loc) · 430 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
demo.py — Unified pygame host for Project Spyder Ascend.
Layered as follows:
Section 1 (lines below) — verbatim copy of Configurer.py lines 437+.
SimBus + TeleopSim + matplotlib `demo` helper +
the _run_pygame_teleop launcher + main entry.
Provides the full reference teleop sandbox.
Section 2 (further below) — Phase A additions:
Planner ABC, GBNNBasePlanner, NavController,
ReconfigSequencer, PlannerRegistry, EventHandler,
LMB-drag → GBNN coverage with heatmap overlay,
headless test suite, SB3 PPO load spike.
Configurer.py keeps its full standalone teleop sandbox; this file is the
mode-1 host that Phases B–E adapters will plug into.
"""
# ---------------------------------------------------------------------------
# Module-level imports (mirroring Configurer.py's top-of-file imports)
# ---------------------------------------------------------------------------
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Dict, List, Optional, Tuple
import math
import random
import sys
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.patches as patches
# Pure FSM core lives in Configurer.py — import the dataclasses + enum + class
# the copied SimBus / TeleopSim code references.
from configurer.open_configurer import Configurer, Twist, Pose, FSMState
# pygame is imported lazily inside _run_pygame_teleop() — matches Configurer.
# ===========================================================================
# Section 0 (Phase A) — Contracts, GBNN adapter, helper imports
# ===========================================================================
#
# These types + classes are the integration surface for Phases B–E. They
# sit at module scope so the verbatim TeleopSim copy below can call into
# them with minimal surgery. Configurer.py is NOT modified.
# ===========================================================================
from abc import ABC, abstractmethod
from common.obstacles import (
Obstacle, ObstacleKind, ObstacleManager, OccupancyGrid,
pathfind_astar, pathfind_dijkstra, smooth_path,
MDA_ARM_RADIUS, MDA_ARM_REACH_M, MDA_ARM_FOV_DEG, MDA_MOUNT_RANGE_M,
)
from common.replicated_gbnn import GBNN
from gbnnh.open_gbnnh import GBNN_H, RoIFrame, AccessPoint
# ---- type aliases ----
XY = Tuple[float, float]
Waypoint = Tuple[float, float]
Pose2D = Tuple[float, float, float]
Cell = Tuple[int, int]
StepIdx = int
@dataclass
class Rect:
"""Axis-aligned rectangle in world coordinates (metres)."""
x0: float
y0: float
x1: float
y1: float
def normalized(self) -> "Rect":
return Rect(
min(self.x0, self.x1), min(self.y0, self.y1),
max(self.x0, self.x1), max(self.y0, self.y1),
)
def contains(self, x: float, y: float) -> bool:
n = self.normalized()
return n.x0 <= x <= n.x1 and n.y0 <= y <= n.y1
@property
def width(self) -> float:
return abs(self.x1 - self.x0)
@property
def height(self) -> float:
return abs(self.y1 - self.y0)
class PlannerMode(Enum):
"""Active planner mode (mode 1 = the verbatim Configurer demo)."""
MANUAL = 1 # WASD + LMB-click P2P + LMB-drag GBNN coverage
INTERSTAR = 2 # Phase B
GBNNH = 5 # Phase E
@dataclass
class RobotState:
robot_id: int
pose: Pose
fsm_state: FSMState
host_id: int
n: int
footprint_m: float = 0.70
@dataclass
class GoalSpec:
"""Tagged union — exactly one field is non-None."""
point: Optional[XY] = None # point-to-point navigation
points: Optional[List[XY]] = None # fission target list
area: Optional[Rect] = None # coverage RoI / mode-1 GBNN drag-rect
@dataclass
class ReconfigCommand:
"""Demo-side wrapper around the paper-faithful rcfg triple (Configurer paper, Table I)."""
recipient_id: int
rcfg: Tuple[int, int, int] # [neighbour_id, size, split_command_code]
when: StepIdx = 0
@dataclass
class PlanResult:
assignments: Dict[int, List[Waypoint]] = field(default_factory=dict)
reconfig: List[ReconfigCommand] = field(default_factory=list)
algo_starts: Dict[int, Pose2D] = field(default_factory=dict)
extras: Dict = field(default_factory=dict)
metrics: Dict[str, float] = field(default_factory=dict)
@dataclass
class WorldSpec:
"""Canonical description of the simulation world for adapters."""
bounds: Tuple[float, float, float, float]
cell_m: float
obs_mgr: "ObstacleManager"
robots: List[RobotState]
@dataclass
class PlannerQuery:
mode: PlannerMode
world: WorldSpec
robots: List[RobotState]
selected_ids: List[int]
goal: GoalSpec
class Planner(ABC):
"""Base interface for all planners. Adapters subclass this."""
@abstractmethod
def plan(self, query: PlannerQuery) -> PlanResult:
"""One-shot plan call. Returns a PlanResult; may be empty."""
def reset(self) -> None: ...
def step(self) -> Optional[PlanResult]:
"""Optional iterative advance — return per-step diff or None."""
return None
def is_done(self) -> bool: return True
def render_state(self) -> Dict: return {}
# ---------------------------------------------------------------------------
# DefaultPlanner — wraps the existing pathfind_astar for completeness.
# (The verbatim TeleopSim already uses A*/Dijkstra directly; this adapter
# exists so Phases B–E can call DefaultPlanner.plan() uniformly.)
# ---------------------------------------------------------------------------
class DefaultPlanner(Planner):
def __init__(self, cell_size: float = 0.20, inflate_radius: float = 0.40):
self.cell_size = cell_size
self.inflate_radius = inflate_radius
self._last_path: List[Waypoint] = []
def plan(self, query: PlannerQuery) -> PlanResult:
if not query.selected_ids or query.goal.point is None:
return PlanResult()
rid = query.selected_ids[0]
robot = next((r for r in query.robots if r.robot_id == rid), None)
if robot is None:
return PlanResult()
positions = [(r.pose.x, r.pose.y, r.footprint_m / 2)
for r in query.robots]
excl = next((i for i, r in enumerate(query.robots)
if r.robot_id == rid), None)
grid = query.world.obs_mgr.build_occupancy_grid(
world_bounds = query.world.bounds,
cell_size = self.cell_size,
inflate_radius = self.inflate_radius,
robot_positions= positions,
exclude_rid = excl,
)
path = pathfind_astar(grid,
(robot.pose.x, robot.pose.y),
query.goal.point)
if path is None or len(path) < 2:
self._last_path = []
return PlanResult(metrics={"plan_failed": 1.0})
self._last_path = list(path)
return PlanResult(
assignments={rid: list(path)},
metrics={"path_len_cells": float(len(path))},
)
def render_state(self) -> Dict:
return {"path": list(self._last_path)}
# ---------------------------------------------------------------------------
# GBNNBasePlanner — wraps GBNN.py for mode-1 LMB-drag area coverage.
# Single robot, single RoI, holonomic, dynamic obstacles re-pushed each tick.
# ---------------------------------------------------------------------------
class GBNNBasePlanner(Planner):
"""LMB-drag → rectangular RoI → GBNN coverage for the active robot.
Rasterises the drawn rectangle to a grid at the robot's footprint cell
size (under-bite via math.floor — never spills outside the rect).
If the active robot is inside the RoI, GBNN starts from its current
cell; otherwise the planner returns an A* approach path back to the
nearest free RoI cell, and only when the robot arrives does GBNN take
over. Dynamic obstacles are re-pushed every step via set_occupancy().
"""
def __init__(
self,
footprint_m: float = 0.70,
approach_cell_size: float = 0.20,
approach_inflate: float = 0.40,
) -> None:
self.footprint_m = footprint_m
self.approach_cell_size = approach_cell_size
self.approach_inflate = approach_inflate
self._gbnn: Optional[GBNN] = None
self._roi: Optional[Rect] = None
self._cell_size: float = 0.0
self._origin: XY = (0.0, 0.0)
self._grid_shape: Tuple[int, int] = (0, 0)
self._active_rid: Optional[int] = None
self._approach_path: List[Waypoint] = []
self._world: Optional[WorldSpec] = None
# Robot-visited cells. Tracked separately from GBNN's grid so that
# transient obstacles (humans walking through, doors closing) don't
# cause already-covered cells to revert to "unvisited" the moment
# the obstacle clears — which would make coverage uncompletable.
self._visited_cells: set = set()
def _cell_to_world(self, cell: Cell) -> XY:
r, c = cell
return (self._origin[0] + (c + 0.5) * self._cell_size,
self._origin[1] + (r + 0.5) * self._cell_size)
def _world_to_cell(self, wx: float, wy: float) -> Cell:
cs = self._cell_size
c = int((wx - self._origin[0]) / cs)
r = int((wy - self._origin[1]) / cs)
rows, cols = self._grid_shape
return (max(0, min(rows - 1, r)), max(0, min(cols - 1, c)))
def _external_robot_positions(
self, world: WorldSpec,
) -> List[Tuple[float, float, float]]:
"""Robot positions list with the ACTIVE FORMATION excluded.
GBNN is planning coverage for a single planning unit — whether
that unit is a split singleton (n=1) or a fused singleton
(n≥2). In the fused case, every member of the active formation
moves rigidly with the host, so treating those members as
obstacles would (a) pre-block the robot's own cells and (b)
introduce ghost occupancy that wanders with the formation.
Same-host-id robots must therefore be excluded from the
occupancy grid — only robots outside this formation count as
dynamic obstacles to route around.
"""
# Find host_id of active robot
active_host = None
for r in world.robots:
if r.robot_id == self._active_rid:
active_host = r.host_id
break
if active_host is None:
# Fallback: only exclude the active robot itself by identity
return [(r.pose.x, r.pose.y, r.footprint_m / 2)
for r in world.robots
if r.robot_id != self._active_rid]
return [(r.pose.x, r.pose.y, r.footprint_m / 2)
for r in world.robots
if r.host_id != active_host]
# Inflation margin used when testing obstacle-vs-cell overlap.
# Treats each obstacle as if its OBB / caster circles / robot radius
# were expanded by this amount: any cell whose interior overlaps the
# inflated zone is marked blocked. Small enough to avoid over-block
# on rotated obstacles, large enough to keep robots from grazing.
GBNN_CELL_INFLATE_M: float = 0.20
@staticmethod
def _obb_overlaps_aabb(
obs, x_min: float, x_max: float,
y_min: float, y_max: float, inflate: float,
) -> bool:
"""SAT — true geometric overlap between the obstacle's OBB
(inflated outward by `inflate` along each local axis) and the
axis-aligned rectangle [x_min, x_max] x [y_min, y_max]."""
hw = obs.half_w + inflate
hh = obs.half_h + inflate
cos_y = math.cos(obs.yaw)
sin_y = math.sin(obs.yaw)
obb = [
(obs.x + cos_y * (-hw) - sin_y * (-hh),
obs.y + sin_y * (-hw) + cos_y * (-hh)),
(obs.x + cos_y * ( hw) - sin_y * (-hh),
obs.y + sin_y * ( hw) + cos_y * (-hh)),
(obs.x + cos_y * ( hw) - sin_y * ( hh),
obs.y + sin_y * ( hw) + cos_y * ( hh)),
(obs.x + cos_y * (-hw) - sin_y * ( hh),
obs.y + sin_y * (-hw) + cos_y * ( hh)),
]
cell = [
(x_min, y_min), (x_max, y_min),
(x_max, y_max), (x_min, y_max),
]
axes = (
(1.0, 0.0), # AABB x
(0.0, 1.0), # AABB y
(cos_y, sin_y), # OBB local x
(-sin_y, cos_y), # OBB local y
)
for (ax, ay) in axes:
cmin = min(p[0] * ax + p[1] * ay for p in cell)
cmax = max(p[0] * ax + p[1] * ay for p in cell)
omin = min(p[0] * ax + p[1] * ay for p in obb)
omax = max(p[0] * ax + p[1] * ay for p in obb)
if cmax < omin or omax < cmin:
return False
return True
def _cell_blocked(
self,
cell_x: float, cell_y: float, cs: float,
obstacles: list, robot_positions: list,
) -> bool:
"""
Check whether a GBNN cell at world-frame (cell_x, cell_y) with
side `cs` is blocked by any obstacle or external robot.
Uses true geometric overlap — Separating Axis Theorem for OBB
obstacles, point-clamp distance for caster circles and robot
occupancy circles — with each obstacle inflated outward by
GBNN_CELL_INFLATE_M. ANY overlap between the inflated obstacle
zone and the cell's axis-aligned rectangle marks the cell as
blocked, with no sampling false-negatives.
"""
x_min, x_max = cell_x, cell_x + cs
y_min, y_max = cell_y, cell_y + cs
inflate = self.GBNN_CELL_INFLATE_M
for obs in obstacles:
if obs.is_caster_trolley:
for (wcx, wcy, cr) in obs.caster_circles():
# Circle-AABB overlap with inflation
clamp_x = max(x_min, min(x_max, wcx))
clamp_y = max(y_min, min(y_max, wcy))
r_inf = cr + inflate
if (wcx - clamp_x) ** 2 + (wcy - clamp_y) ** 2 \
< r_inf * r_inf:
return True
else:
if self._obb_overlaps_aabb(
obs, x_min, x_max, y_min, y_max, inflate):
return True
for (rx, ry, r_occ) in robot_positions:
clamp_x = max(x_min, min(x_max, rx))
clamp_y = max(y_min, min(y_max, ry))
r_inf = r_occ + inflate
if (rx - clamp_x) ** 2 + (ry - clamp_y) ** 2 \
< r_inf * r_inf:
return True
return False
def _build_roi_grid(self, world: WorldSpec, rect: Rect) -> np.ndarray:
"""Rasterise — under-bite (math.floor) so the grid never spills past
the user-drawn rectangle. Excess margin is split evenly between
the two sides of each axis."""
rect = rect.normalized()
cs = self.footprint_m
self._cell_size = cs
cols = max(1, int(math.floor(rect.width / cs)))
rows = max(1, int(math.floor(rect.height / cs)))
pad_x = (rect.width - cols * cs) / 2.0
pad_y = (rect.height - rows * cs) / 2.0
self._origin = (rect.x0 + pad_x, rect.y0 + pad_y)
self._grid_shape = (rows, cols)
g = np.ones((rows, cols), dtype=float)
# Exclude the active formation's members — they move with the
# host, so they mustn't appear as static obstacles.
positions = self._external_robot_positions(world)
obstacles = [obs for obs in world.obs_mgr.obstacles.values()
if not obs.is_mounted]
for r in range(rows):
for c in range(cols):
cell_x = self._origin[0] + c * cs
cell_y = self._origin[1] + r * cs
if self._cell_blocked(cell_x, cell_y, cs,
obstacles, positions):
g[r, c] = -1.0
return g
def _live_mask(self) -> np.ndarray:
"""Rebuild obstacle mask each tick so dynamic obstacles propagate.
Excludes every same-formation robot (same host_id) — split or
fused, GBNN must never see its own planning unit as obstacles.
"""
rows, cols = self._grid_shape
cs = self._cell_size
rect = self._roi
assert rect is not None and self._world is not None
positions = self._external_robot_positions(self._world)
obstacles = [obs for obs in self._world.obs_mgr.obstacles.values()
if not obs.is_mounted]
mask = np.zeros((rows, cols), dtype=bool)
for r in range(rows):
for c in range(cols):
cell_x = self._origin[0] + c * cs
cell_y = self._origin[1] + r * cs
if self._cell_blocked(cell_x, cell_y, cs,
obstacles, positions):
mask[r, c] = True
return mask
@staticmethod
def _nearest_free_cell(grid: np.ndarray, seed: Cell) -> Optional[Cell]:
from collections import deque
rows, cols = grid.shape
if grid[seed[0], seed[1]] != -1.0:
return seed
q, seen = deque([seed]), {seed}
while q:
r, c = q.popleft()
for dr in (-1, 0, 1):
for dc in (-1, 0, 1):
if dr == 0 and dc == 0: continue
nr, nc = r + dr, c + dc
if (0 <= nr < rows and 0 <= nc < cols
and (nr, nc) not in seen):
seen.add((nr, nc))
if grid[nr, nc] != -1.0:
return (nr, nc)
q.append((nr, nc))
return None
def plan(self, query: PlannerQuery) -> PlanResult:
if query.goal.area is None or not query.selected_ids:
return PlanResult()
rid = query.selected_ids[0]
robot = next((r for r in query.robots if r.robot_id == rid), None)
if robot is None:
return PlanResult()
self._active_rid = rid
self._roi = query.goal.area.normalized()
self._world = query.world
grid = self._build_roi_grid(query.world, self._roi)
if self._roi.contains(robot.pose.x, robot.pose.y):
start_cell = self._world_to_cell(robot.pose.x, robot.pose.y)
if grid[start_cell[0], start_cell[1]] == -1.0:
start_cell = self._nearest_free_cell(grid, start_cell)
self._approach_path = []
else:
best, best_d = None, float('inf')
rows, cols = self._grid_shape
for r in range(rows):
for c in range(cols):
if grid[r, c] == -1.0: continue
wx, wy = self._cell_to_world((r, c))
d = math.hypot(wx - robot.pose.x, wy - robot.pose.y)
if d < best_d:
best_d, best = d, (r, c)
if best is None:
return PlanResult(metrics={"plan_failed": 1.0})
start_cell = best
# Approach A* must also see the active formation as "self",
# not as an obstacle (otherwise a fused singleton can't find
# a path to any RoI cell that its own members are standing
# on or next to). Reuse the formation-wide exclusion helper.
positions = self._external_robot_positions(query.world)
approach_grid = query.world.obs_mgr.build_occupancy_grid(
world_bounds=query.world.bounds,
cell_size=self.approach_cell_size,
inflate_radius=self.approach_inflate,
robot_positions=positions, exclude_rid=None,
)
self._approach_path = pathfind_astar(
approach_grid, (robot.pose.x, robot.pose.y),
self._cell_to_world(start_cell),
) or []
if start_cell is None or grid[start_cell[0], start_cell[1]] == -1.0:
return PlanResult(metrics={"plan_failed": 1.0})
self._gbnn = GBNN()
self._gbnn.reset(grid, start_cell)
# Seed visited tracker with start cell (it was reset by GBNN to a
# smooth covered-residue value, not +1). Future step() calls will
# add each cell the robot moves into.
self._visited_cells = {tuple(start_cell)}
assignments: Dict[int, List[Waypoint]] = {}
if self._approach_path:
assignments[rid] = list(self._approach_path)
return PlanResult(
assignments=assignments,
algo_starts={rid: (*self._cell_to_world(start_cell), 0.0)},
extras=self.render_state(),
metrics={"coverage_pct": self._gbnn.coverage_pct},
)
def step(self) -> Optional[PlanResult]:
if self._gbnn is None or self._world is None:
return None
if self._gbnn.is_done():
return PlanResult(extras=self.render_state(),
metrics={"coverage_pct": 1.0, "done": 1.0})
# Push current obstacle mask into GBNN — this can reset cells that
# were obstacle-then-cleared back to +1 (unvisited). Restore the
# visited-state for any cell the robot has actually been to so the
# task remains completable when humans / doors come and go.
self._gbnn.set_occupancy(self._live_mask())
for (vr, vc) in self._visited_cells:
if (0 <= vr < self._gbnn._grid.shape[0]
and 0 <= vc < self._gbnn._grid.shape[1]
and self._gbnn._grid[vr, vc] == 1.0):
# Cell was visited but got reset to 1.0 by set_occupancy
# because a transient obstacle just left it. Mark it
# visited again (0.0 = covered, no longer attractor).
self._gbnn._grid[vr, vc] = 0.0
new_cell = self._gbnn.step()
# Track the cell the robot is now committed to visit
self._visited_cells.add(tuple(new_cell))
new_wxy = self._cell_to_world(new_cell)
return PlanResult(
assignments={self._active_rid: [new_wxy]} if self._active_rid else {},
extras=self.render_state(),
metrics={"coverage_pct": self._gbnn.coverage_pct},
)
def is_done(self) -> bool:
return bool(self._gbnn.is_done()) if self._gbnn is not None else True
def reset(self) -> None:
self._gbnn = None
self._roi = None
self._active_rid = None
self._approach_path = []
self._world = None
self._visited_cells = set()
def render_state(self) -> Dict:
if self._gbnn is None or self._roi is None:
return {}
pos = self._gbnn.position
return {
"roi": self._roi,
"origin": self._origin,
"cell_size": self._cell_size,
"grid_shape": self._grid_shape,
"activity_grid": self._gbnn.activity_grid,
"position_cell": pos,
"position_world": self._cell_to_world(pos) if pos else None,
"path_cells": self._gbnn.path,
"coverage_pct": self._gbnn.coverage_pct,
"iterations": self._gbnn.iterations,
# Active robot id → demo.py uses this to tint cell shading
# with the robot's own colour, so multi-robot coverage scenes
# stay visually disambiguated.
"active_rid": self._active_rid,
}
# ---------------------------------------------------------------------------
# InterstarPlanner — mode-2 multi-robot fusion / fission via Inter-Star A*
# ---------------------------------------------------------------------------
# Auto-dispatch policy (from B5):
# * ≥2 robots selected (via Ctrl+click) + LMB click = FUSION
# — all selected robots converge on the click point
# * 1 fused singleton (n>1) selected + n LMB clicks = FISSION
# — divergence from the singleton's pose to the n click points
#
# Returns per-robot world-frame paths + (k-1) pairwise FUSE / FISSION
# ReconfigCommands scheduled at arrival ticks, and metrics:
# expansions, baseline_expansions (vs standard A*), expansions_ratio,
# shared_segment (world-frame polyline for the amber render overlay).
from interstar.open_interstar import Interstar as _Interstar
class InterstarPlanner(Planner):
"""Inter-Star adapter — multi-robot fusion (convergence) or fission
(divergence) with shared-path exploitation.
Per Phase B B1, the Interstar class now accepts explicit starts +
fission_start + mode + numpy grid and returns shared-path metrics via
its `plan()` classmethod. This adapter is a thin shim that:
1. Rasterises the world obstacles to a numpy grid
2. Converts live robot poses + goal(s) to grid cells
3. Calls Interstar.plan()
4. Converts the resulting (row, col) paths back to world waypoints
5. Builds a PlanResult with per-robot paths + (n-1) FUSEs (fusion)
or (n-1) FISSIONs (fission) + shared-segment extras + metrics
"""
def __init__(
self,
cell_size: float = 0.20,
inflate_radius: float = 0.40,
) -> None:
self.cell_size = cell_size
self.inflate_radius = inflate_radius
self._last_paths: Dict[int, List[Waypoint]] = {}
self._last_metrics: Dict[str, float] = {}
self._last_shared: List[Waypoint] = [] # world-frame amber segment
self._last_mode: str = "" # "fusion" / "fission"
# ---- helpers ----
def _world_to_cell(self, grid, wx: float, wy: float) -> Cell:
"""Convert a world (x, y) to an OccupancyGrid (row, col) tuple."""
c, r = grid.world_to_cell(wx, wy)
# Clamp into bounds
r = max(0, min(grid.rows - 1, r))
c = max(0, min(grid.cols - 1, c))
return (r, c)
def _cell_to_world(self, grid, cell: Cell) -> XY:
r, c = cell
return grid.cell_to_world(c, r)
# ---- Planner API ----
def plan(self, query: PlannerQuery) -> PlanResult:
if not query.selected_ids:
return PlanResult()
# Snapshot world: occupancy grid with NON-PARTICIPANT robots
# treated as inflated obstacles. Selected participants AND
# any same-formation members of those participants are
# excluded — they're the planning agents and shouldn't block
# each other (and won't, since the cursor-based driver
# serialises arrivals at shared cells). Mirrors A*'s
# nav_members exclusion in `_compute_nav_cmd_for`, which
# keeps the two layers consistent: cells Inter-Star plans
# through are exactly the cells A* would consider free.
participants = set(query.selected_ids)
other_robot_pos: List[Tuple[float, float, float]] = []
for r in query.robots:
if r.robot_id in participants:
continue
# Same-formation members of any participant move with the
# host and shouldn't block the participant's own grid.
if r.host_id in participants:
continue
other_robot_pos.append(
(r.pose.x, r.pose.y, r.footprint_m / 2))
occ = query.world.obs_mgr.build_occupancy_grid(
world_bounds = query.world.bounds,
cell_size = self.cell_size,
inflate_radius = self.inflate_radius,
robot_positions = other_robot_pos if other_robot_pos else None,
exclude_rid = None,
)
# Build a numpy grid 0 = free, 1 = obstacle (Inter-Star's encoding)
g = np.zeros((occ.rows, occ.cols), dtype=int)
for r in range(occ.rows):
for c in range(occ.cols):
if not occ.is_free(c, r):
g[r, c] = 1
# Build starts / goal based on dispatch type:
# FUSION — multiple selected robots converge on query.goal.point
# FISSION — single selected robot with multiple goal points
sel = list(query.selected_ids)
if query.goal.points and len(sel) == 1:
# FISSION: 1 robot, n goal points
self._last_mode = "fission"
robot = next((r for r in query.robots if r.robot_id == sel[0]),
None)
if robot is None:
return PlanResult()
fission_start_cell = self._world_to_cell(
occ, robot.pose.x, robot.pose.y)
# Free the start cell so A* can proceed
g[fission_start_cell[0], fission_start_cell[1]] = 0
goal_cells = []
for (gx, gy) in query.goal.points:
gc = self._world_to_cell(occ, gx, gy)
g[gc[0], gc[1]] = 0
goal_cells.append(gc)
cell_paths, metrics = _Interstar.plan(
starts = [],
goal = goal_cells,
grid = g,
mode = "fission",
fission_start = fission_start_cell,
render = False,
)
# paths[k] corresponds to fission goal k; there's no direct
# robot_id mapping (only one robot is splitting). We assign
# each cell-path to one of the existing split-off robot ids;
# demo.py's _dispatch will ultimately feed them to the host.
# For the adapter, we just return indexed paths under the
# single selected rid — the caller interprets them as an
# n-way fission of that formation.
world_paths: Dict[int, List[Waypoint]] = {}
# With fission, paths[0], paths[1], ... are the n divergence
# trajectories. We emit them under synthetic keys -1, -2, -3
# so the caller can disambiguate — demo.py's dispatcher will
# emit matching FISSION commands and route the paths to the
# freshly-split members.
for k, cp in enumerate(cell_paths):
world_paths[-(k + 1)] = [self._cell_to_world(occ, tuple(c))
for c in cp]
reconfig = [
ReconfigCommand(
recipient_id = sel[0],
rcfg = (0, 0, -1), # split_command_code = -1
when = 0,
)
]
self._last_paths = world_paths
self._last_metrics = metrics
self._last_shared = []
return PlanResult(
assignments = world_paths,
reconfig = reconfig,
extras = {"shared_segment": [], "mode": "fission"},
metrics = {
"expansions": float(metrics["expansions"]),
"baseline": float(metrics["baseline_expansions"]),
"expansions_ratio": float(metrics["expansions_ratio"]),
},
)
# ---- FUSION ----
if query.goal.point is None:
return PlanResult()
self._last_mode = "fusion"
start_cells = []
for rid in sel:
robot = next((r for r in query.robots if r.robot_id == rid), None)
if robot is None:
continue
sc = self._world_to_cell(occ, robot.pose.x, robot.pose.y)
# Free start cell in case obstacle inflation covered it
g[sc[0], sc[1]] = 0
start_cells.append(sc)
goal_cell = self._world_to_cell(occ, *query.goal.point)
g[goal_cell[0], goal_cell[1]] = 0
if len(start_cells) < 2:
return PlanResult(metrics={"plan_failed": 1.0})
cell_paths, metrics = _Interstar.plan(
starts = start_cells,
goal = goal_cell,
grid = g,
mode = "fusion",
render = False,
)
# Convert to world-frame paths keyed by robot_id (order-preserving
# with `sel` — the Interstar call preserves input ordering).
world_paths = {}
for rid, cp in zip(sel, cell_paths):
if not cp:
continue
world_paths[rid] = [self._cell_to_world(occ, tuple(c)) for c in cp]
shared_world = [
self._cell_to_world(occ, tuple(c))
for c in metrics.get("shared_segment", [])
]
# Generate (n-1) FUSE ReconfigCommands. The first-arriving robot
# (index 0 in sel) becomes the initial host; each subsequent robot
# fuses with the growing formation pair-by-pair. The NavController
# in demo.py drives each robot along its path; when a robot arrives
# at the fusion point, the sequencer fires the next pairwise FUSE.
reconfig: List[ReconfigCommand] = []
if len(sel) >= 2:
host_so_far = min(sel) # smallest-id becomes canonical host
for idx, rid in enumerate(sel):
if rid == host_so_far:
continue
# Two-sided handshake: each side publishes its tag to the
# other. Sequencer fires them together (same `when`).
reconfig.append(ReconfigCommand(
recipient_id = host_so_far,
rcfg = (rid, 0, 0),
when = idx,
))
reconfig.append(ReconfigCommand(
recipient_id = rid,
rcfg = (host_so_far, 0, 0),
when = idx,
))
self._last_paths = world_paths
self._last_metrics = metrics
self._last_shared = shared_world
return PlanResult(
assignments = world_paths,
reconfig = reconfig,
extras = {
"shared_segment": shared_world,
"mode": "fusion",
},
metrics = {
"expansions": float(metrics["expansions"]),
"baseline": float(metrics["baseline_expansions"]),
"expansions_ratio": float(metrics["expansions_ratio"]),
},
)
def render_state(self) -> Dict:
return {
"paths": dict(self._last_paths),
"shared_segment": list(self._last_shared),
"metrics": dict(self._last_metrics),
"mode": self._last_mode,
}
# ---------------------------------------------------------------------------
# ReconfigSequencer — FIFO of ReconfigCommands fired into Configurer FSMs
# ---------------------------------------------------------------------------
# Phase A does not produce reconfig commands (mode 1 = single-robot teleop /
# coverage); future adapters may emit them. This scaffold class lets the demo
# accept a `List[ReconfigCommand]` from any adapter and dispatch them as fast
# as the pairwise FSM handshake allows — paper-faithful per Configurer paper, Table I.
class ReconfigSequencer:
"""FIFO queue of ReconfigCommand, fired one-at-a-time as each
recipient's FSM clears (returns to CONFIG with rcfg inbox == [0,0,0]).
No artificial pacing — the next command fires the instant the previous
pairwise handshake completes. N-robot fusion is dispatched as (N-1)
pairwise FUSEs scheduled in order via the `when` field.
Phase A: instantiated by TeleopSim, idle (queue stays empty).
Future adapters extend the queue via `enqueue` / `extend`.
"""
def __init__(self, bus) -> None:
self._bus = bus
self._queue: List[ReconfigCommand] = []
self._step_idx: int = 0
def enqueue(self, cmd: ReconfigCommand) -> None:
self._queue.append(cmd)
self._queue.sort(key=lambda c: c.when)
def extend(self, cmds: List[ReconfigCommand]) -> None:
for c in cmds:
self.enqueue(c)
def clear(self) -> None:
self._queue.clear()
def pending(self) -> int:
return len(self._queue)
def tick(self) -> None:
"""Fire any eligible commands whose recipient's FSM is ready."""
self._step_idx += 1
i = 0
while i < len(self._queue):
cmd = self._queue[i]
if cmd.when > self._step_idx:
i += 1
continue
cfg = self._bus.configurers.get(cmd.recipient_id)
if cfg is None:
self._queue.pop(i)
continue
if cfg.fsm_state == FSMState.CONFIG and cfg.rcfg == [0, 0, 0]:
cfg.ingest_rcfg(list(cmd.rcfg))
self._queue.pop(i)
# don't advance i — re-check the new head next iter
else:
i += 1
# ---------------------------------------------------------------------------
# PlannerRegistry — named-handle lookup of Planner adapter instances
# ---------------------------------------------------------------------------
# Used in Phase A by TeleopSim to hold the singletons of DefaultPlanner and
# GBNNBasePlanner. Phases B and E register InterstarPlanner and GBNNHPlanner
# under their own keys — the EventHandler then resolves "which planner do I
# dispatch to" via `registry.get(key)`.
class PlannerRegistry:
"""Central named-handle registry of Planner adapter instances.
Phase A keys (registered by TeleopSim.__init__):
"default" → DefaultPlanner (P2P A* — used by mode-1 LMB-click)
"gbnn" → GBNNBasePlanner (RoI coverage — used by mode-1 LMB-drag)
Phase B+ adds: "interstar", "gbnnh".
"""
def __init__(self) -> None:
self._planners: Dict[str, Planner] = {}
def register(self, key: str, planner: Planner) -> None:
self._planners[key] = planner
def get(self, key: str) -> Optional[Planner]:
return self._planners.get(key)
def keys(self) -> List[str]:
return list(self._planners.keys())
# ===========================================================================
# Section 1 — copy of Configurer.py lines 437 .. EOF (verbatim)
# ===========================================================================
# ============================================================================
# SIMBUS (demo-only helper: in-process ROS-topic stand-in)
# ============================================================================
class SimBus:
"""
In-process message bus for running N Configurer instances on one machine.
Replaces ROS2 topics for headless / Spyder testing. Not used in
production -- ROS2 wrapper will replace this with real publishers.
Responsibilities
----------------
* Registry of robots (id -> Configurer).
* Route rcfg publishes to the target Configurer's ingest_rcfg().
* Demo-time physics: integrate each robot's xfm_vel into a simulated
ground-truth pose (Euler step), feed back via ingest_pose().
* Owns the matplotlib visualisation (one fresh figure per frame,
Spyder-compatible -- matches Interstar / GBNN+H pattern).
"""
# Colour palette
COLORS = [
'blue', 'green', 'red', 'cyan', 'magenta',
'orange', 'purple', 'brown', 'pink', 'olive',
]
def __init__(self, dt: float = 0.1, visualize: bool = True):
self.configurers: Dict[int, Configurer] = {}
self.poses: Dict[int, Pose] = {}
self.dt: float = dt
self.visualize: bool = visualize
self._step_ix: int = 0
self._xfm_log: Dict[int, List[Twist]] = {}
# ------------------------------------------------------------------
# Registration + transport
# ------------------------------------------------------------------
def register(self, configurer: Configurer, pose: Optional[Pose] = None) -> None:
"""Add a Configurer to the bus. pose = initial ground-truth pose."""
rid = configurer.robot_id
self.configurers[rid] = configurer
self.poses[rid] = pose if pose is not None else Pose()
self._xfm_log[rid] = []
def publish_rcfg(self, target_id: int, rcfg: List[int]) -> None:
"""Route an rcfg publish from one Configurer to another."""
target = self.configurers.get(target_id)
if target is None:
return # target not registered -- silently drop (matches flaky topic)
target.ingest_rcfg(rcfg)
# ------------------------------------------------------------------
# Commands (joystick / teleop equivalents)
# ------------------------------------------------------------------
def send_cmd_vel(self, robot_id: int, cmd: Twist) -> None:
cfg = self.configurers.get(robot_id)
if cfg is not None:
cfg.ingest_cmd_vel(cmd)
def send_fusion_command(self, host_id: int, neighbour_id: int) -> None: