Skip to content

Tools Module

tools.report_generator

PDF report generation for SLAM benchmark results.

This module uses ReportLab to generate comprehensive PDF reports with: - Summary tables comparing multiple runs - Visualization charts (ATE, Coverage, IoU, CPU/RAM) - Map comparison images - Trajectory plots - Automated performance analysis

SLAMReportGenerator

Generator for comprehensive SLAM benchmark PDF reports.

Creates multi-page PDF reports with tables, charts, and images comparing multiple SLAM algorithm runs.

Parameters:

Name Type Description Default
runs_data

List of dicts containing run metrics and metadata

required
output_path

Path where the PDF will be saved

required
plot_path

Optional path to a trajectory plot image

None
Source code in tools/report_generator.py
 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
class SLAMReportGenerator:
    """Generator for comprehensive SLAM benchmark PDF reports.

    Creates multi-page PDF reports with tables, charts, and images comparing
    multiple SLAM algorithm runs.

    Args:
        runs_data: List of dicts containing run metrics and metadata
        output_path: Path where the PDF will be saved
        plot_path: Optional path to a trajectory plot image
    """
    def __init__(self, runs_data, output_path, plot_path=None):
        self.runs_data = runs_data
        self.output_path = output_path
        self.plot_path = plot_path
        self.styles = getSampleStyleSheet()
        # Use landscape for better width handling
        self.doc = SimpleDocTemplate(str(output_path), pagesize=landscape(A4))
        self.elements = []

        # Custom styles
        self.styles.add(ParagraphStyle(name='CenterTitle', parent=self.styles['Title'], alignment=1))
        self.styles.add(ParagraphStyle(name='SubTitle', parent=self.styles['Heading2'], color=colors.HexColor("#6366f1")))

    def add_header(self):
        title = Paragraph("SLAM Benchmarking Analysis Report", self.styles['CenterTitle'])
        self.elements.append(title)

        date_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        info = Paragraph(f"Generated on: {date_str}", self.styles['Normal'])
        self.elements.append(info)
        self.elements.append(Spacer(1, 0.2 * inch))


    def add_metrics_charts(self):
        if not self.runs_data:
            return

        self.elements.append(PageBreak())
        self.elements.append(Paragraph("Metrics Visualization", self.styles['Title']))

        # Prepare data
        names = []
        for i, r in enumerate(self.runs_data):
            label = r.get('slam', 'Unknown')
            if names.count(label) > 0 or any(x == label for x in names):
                label = f"{label} ({i+1})"
            names.append(label)

        ate = [r.get('ate') if r.get('ate') is not None else 0 for r in self.runs_data]

        coverage = [r.get('coverage') if r.get('coverage') is not None else 0 for r in self.runs_data]
        acc_cov = [r.get('accessible_coverage') if r.get('accessible_coverage') is not None else 0 for r in self.runs_data]
        iou = [r.get('occupancy_iou') if r.get('occupancy_iou') is not None else 0 for r in self.runs_data]

        cpu = [r.get('cpu') if r.get('cpu') is not None else 0 for r in self.runs_data]
        ram = [r.get('ram') if r.get('ram') is not None else 0 for r in self.runs_data]

        plt.style.use('ggplot')

        # --- Section 1: Trajectory (ATE) ---
        self.elements.append(Paragraph("1. Trajectory Precision", self.styles['Heading2']))
        self.elements.append(Spacer(1, 0.1 * inch))

        fig1, ax1 = plt.subplots(figsize=(8, 4))
        ax1.bar(names, ate, color='#ef4444')
        ax1.set_title('ATE RMSE (m) - Lower is Better')
        ax1.tick_params(axis='x', rotation=15)
        plt.tight_layout()

        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp1:
            plt.savefig(tmp1.name, dpi=120)
            self.elements.append(Image(tmp1.name, width=7*inch, height=3.5*inch))
        plt.close(fig1)
        self.elements.append(Spacer(1, 0.2 * inch))


        # --- Section 2: Map Quality ---
        self.elements.append(Paragraph("2. Map Quality", self.styles['Heading2']))
        self.elements.append(Spacer(1, 0.1 * inch))

        ssim = [r.get('ssim') if r.get('ssim') is not None else 0 for r in self.runs_data]
        thick = [r.get('wall_thick') if r.get('wall_thick') is not None else 0 for r in self.runs_data]

        # Use GridSpec for 3 top, 2 bottom layout
        fig2 = plt.figure(figsize=(12, 8))
        gs = fig2.add_gridspec(2, 6) # 2 rows, 6 columns grid for flexibility

        # Row 1: Coverage, AccCov, IoU (each takes 2 columns)
        ax1 = fig2.add_subplot(gs[0, 0:2])
        ax2 = fig2.add_subplot(gs[0, 2:4])
        ax3 = fig2.add_subplot(gs[0, 4:6])

        # Row 2: SSIM, Thickness (centered, each takes 2 columns, offset by 1)
        ax4 = fig2.add_subplot(gs[1, 1:3])
        ax5 = fig2.add_subplot(gs[1, 3:5])

        # Plot Data
        ax1.bar(names, coverage, color='#22c55e')
        ax1.set_title('Global Coverage (%)')
        ax1.tick_params(axis='x', rotation=30, labelsize=8)

        ax2.bar(names, acc_cov, color='#10b981')
        ax2.set_title('Accessible Coverage (%)')
        ax2.tick_params(axis='x', rotation=30, labelsize=8)

        ax3.bar(names, iou, color='#8b5cf6')
        ax3.set_title('Occupancy IoU (0-1)')
        ax3.tick_params(axis='x', rotation=30, labelsize=8)

        ax4.bar(names, ssim, color='#d8b4fe')
        ax4.set_title('Map SSIM (0-1)')
        ax4.tick_params(axis='x', rotation=30, labelsize=8)

        ax5.bar(names, thick, color='#fbbf24')
        ax5.set_title('Wall Thickness (cm)')
        ax5.tick_params(axis='x', rotation=30, labelsize=8)

        plt.tight_layout()

        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp2:
            plt.savefig(tmp2.name, dpi=120)
            self.elements.append(Image(tmp2.name, width=9*inch, height=6*inch))
        plt.close(fig2)
        self.elements.append(Spacer(1, 0.2 * inch))


        # --- Section 3: System Consumption ---
        self.elements.append(Paragraph("3. System Consumption", self.styles['Heading2']))
        self.elements.append(Spacer(1, 0.1 * inch))

        fig3, axs3 = plt.subplots(1, 2, figsize=(10, 4))

        axs3[0].bar(names, cpu, color='#3b82f6')
        axs3[0].set_title('Max CPU (%)')
        axs3[0].tick_params(axis='x', rotation=30, labelsize=8)

        axs3[1].bar(names, ram, color='#f59e0b')
        axs3[1].set_title('Max RAM (MB)')
        axs3[1].tick_params(axis='x', rotation=30, labelsize=8)

        plt.tight_layout()

        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp3:
            plt.savefig(tmp3.name, dpi=120)
            self.elements.append(Image(tmp3.name, width=8*inch, height=3.2*inch))
        plt.close(fig3)
        self.elements.append(Spacer(1, 0.3 * inch))


    def add_summary_table(self, runs_data):
        self.elements.append(Paragraph("Comparison Summary", self.styles['SubTitle']))
        self.elements.append(Spacer(1, 0.1 * inch))

        col_count = len(runs_data)


        # If too many runs, switch to Row-based layout (Metrics as columns)
        if col_count > 5:
            # Table Header
            table_header = ["Run ID", "SLAM", "Dataset", "Duration (s)", "ATE (m)", "Cov (%)", "AccCov(%)", "IoU", "Status"]
            data = [table_header]

            for i, r in enumerate(runs_data):
                 # ID handling
                run_id = r.get('name', r.get('id', f'Run {i+1}'))
                # Truncate ID if too long
                if len(run_id) > 20: run_id = run_id[:8] + "..." + run_id[-8:]

                row = [
                    run_id,
                    r.get('slam', '-'),
                    r.get('dataset', '-'),
                    f"{r.get('duration', 0):.2f}" if r.get('duration') is not None else "-",
                    f"{r.get('ate', 0):.3f}" if r.get('ate') is not None else "-",
                    f"{r.get('coverage', 0):.1f}" if r.get('coverage') is not None else "-",
                    f"{r.get('accessible_coverage', 0):.1f}" if r.get('accessible_coverage') is not None else "-",
                    f"{r.get('occupancy_iou', 0):.3f}" if r.get('occupancy_iou') is not None else "-",
                    r.get('status', '-')
                ]
                data.append(row)

            col_widths = [1.8*inch, 1.1*inch, 1.4*inch, 0.9*inch, 0.7*inch, 0.7*inch, 0.8*inch, 0.6*inch, 0.8*inch]
            t = Table(data, hAlign='LEFT', colWidths=col_widths)

        else:
            # Pivot Layout (Comparison style) for few runs
            header = ["Metric", "Run 1", "Run 2", "Run 3"]
            table_header = ["Metric"] + [f"Run {i+1}" for i in range(col_count)]

            # Helper for formatting
            def fmt(key, val):
                if val is None or val == "-": return "-"
                try:
                    v = float(val)
                    if key in ["ate", "occupancy_iou", "ssim", "lidar_noise"]: return f"{v:.4f}"
                    if key in ["coverage", "accessible_coverage", "cpu", "ram", "lidar_range"]: return f"{v:.1f}"
                    if key in ["wall_thick", "duration"]: return f"{v:.2f}"
                    if key in ["speed_scale"]: return f"{v:.0f}"
                except: pass
                return str(val)

            data = [
                table_header,
                ["Health Status"] + [r.get('status', '-') for r in runs_data],
                ["SLAM Algorithm"] + [r.get('slam', '-') for r in runs_data],
                ["Dataset"] + [r.get('dataset', '-') for r in runs_data],
                ["Duration (s)"] + [fmt('duration', r.get('duration')) for r in runs_data],
                ["ATE RMSE (m)"] + [fmt('ate', r.get('ate')) for r in runs_data],
                ["Coverage (%)"] + [fmt('coverage', r.get('coverage')) for r in runs_data],
                ["Acc. Coverage (%)"] + [fmt('accessible_coverage', r.get('accessible_coverage')) for r in runs_data],
                ["Occupancy IoU"] + [fmt('occupancy_iou', r.get('occupancy_iou')) for r in runs_data],
                ["Structural Similarity"] + [fmt('ssim', r.get('ssim')) for r in runs_data],
                ["Wall Thick. (cm)"] + [fmt('wall_thick', r.get('wall_thick')) for r in runs_data],
                ["Max RAM (MB)"] + [fmt('ram', r.get('ram')) for r in runs_data],
                ["Max CPU (%)"] + [fmt('cpu', r.get('cpu')) for r in runs_data],
                ["Lidar Noise (std)"] + [fmt('lidar_noise', r.get('lidar_noise')) for r in runs_data],
                ["Max Range (m)"] + [fmt('lidar_range', r.get('lidar_range')) for r in runs_data],
                ["Speed Scale (%)"] + [fmt('speed_scale', r.get('speed_scale')) for r in runs_data],
            ]
            t = Table(data, hAlign='LEFT')

        t.setStyle(TableStyle([
            ('BACKGROUND', (0, 0), (-1, 0), colors.HexColor("#1e293b")),
            ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
            ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
            ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
            ('BOTTOMPADDING', (0, 0), (-1, 0), 12),
            ('BACKGROUND', (0, 1), (-1, -1), colors.whitesmoke),
            ('GRID', (0, 0), (-1, -1), 1, colors.grey),
        ]))

        self.elements.append(t)
        self.elements.append(Spacer(1, 0.3 * inch))

    def add_plot(self, plot_path):
        if os.path.exists(plot_path):
            self.elements.append(Paragraph("Trajectory Visualization", self.styles['SubTitle']))
            self.elements.append(Spacer(1, 0.1 * inch))
            # Adjust image size for landscape
            img = Image(plot_path, width=8*inch, height=5*inch)
            self.elements.append(img)
            self.elements.append(Spacer(1, 0.2 * inch))

    def add_map_comparison(self, runs_data):
        self.elements.append(PageBreak())
        self.elements.append(Paragraph("Map Reconstruction Comparison", self.styles['SubTitle']))
        self.elements.append(Spacer(1, 0.2 * inch))

        # Display GT Map (Use the first one found)
        gt_path = None
        for r in runs_data:
            if r.get('gt_map_image_path') and os.path.exists(r.get('gt_map_image_path')):
                gt_path = r.get('gt_map_image_path')
                break

        if gt_path:
            self.elements.append(Paragraph("Ground Truth Map", self.styles['Heading3']))
            self.elements.append(Image(gt_path, width=4*inch, height=4*inch, kind='proportional'))
            self.elements.append(Spacer(1, 0.2 * inch))

        # Display Generated Maps in a grid (2 per row)
        self.elements.append(Paragraph("Generated Maps", self.styles['Heading3']))
        self.elements.append(Spacer(1, 0.1 * inch))

        map_images = []
        for i, r in enumerate(runs_data):
            path = r.get('map_image_path')
            title = f"{r.get('slam', 'Run '+str(i+1))} (IoU: {r.get('occupancy_iou', 0):.2f})"
            if path and os.path.exists(path):
                img = Image(path, width=3.5*inch, height=3.5*inch, kind='proportional')
                map_images.append([img, Paragraph(title, self.styles['Normal'])])
            else:
                map_images.append([Paragraph("Map Image Not Found", self.styles['Normal']), Paragraph(title, self.styles['Normal'])])

        # Batch into rows of 3 for Landscape
        row_data = []
        current_row = []
        for item in map_images:
            # Create a mini table for Image + Title
            t_cell = Table([[item[0]], [item[1]]], colWidths=[3.6*inch], rowHeights=[3.6*inch, 0.4*inch])
            t_cell.setStyle(TableStyle([('ALIGN', (0,0), (-1,-1), 'CENTER'), ('VALIGN', (0,0), (-1,-1), 'MIDDLE')]))
            current_row.append(t_cell)

            if len(current_row) == 3:
                row_data.append(current_row)
                current_row = []

        if current_row:
            # Pad with empty cells
            while len(current_row) < 3:
                current_row.append("")
            row_data.append(current_row)

        if row_data:
            t = Table(row_data, colWidths=[3.8*inch, 3.8*inch, 3.8*inch])
            t.setStyle(TableStyle([
                ('ALIGN', (0,0), (-1,-1), 'CENTER'),
                ('VALIGN', (0,0), (-1,-1), 'TOP'),
            ]))
            self.elements.append(t)



    def add_trajectory_comparison(self, runs_data):
        self.elements.append(PageBreak())
        self.elements.append(Paragraph("Trajectory Analysis (GT vs Estimate)", self.styles['SubTitle']))
        self.elements.append(Spacer(1, 0.2 * inch))

        # Display Trajectory Plots in a grid
        traj_images = []
        for i, r in enumerate(runs_data):
            path = r.get('ate_image_path')
            # Fallback if ate_image_path missing (legacy support)
            if not path and r.get('map_image_path'):
                # Try to guess
                maybe_path = Path(r.get('map_image_path')).parent / "ate_plot.png"
                if maybe_path.exists():
                    path = str(maybe_path)

            title = f"{r.get('slam', 'Run '+str(i+1))} (ATE: {r.get('ate', 0):.3f}m)"
            if path and os.path.exists(path):
                img = Image(path, width=3.5*inch, height=3.5*inch, kind='proportional')
                traj_images.append([img, Paragraph(title, self.styles['Normal'])])
            else:
                traj_images.append([Paragraph("Trajectory Plot Not Found", self.styles['Normal']), Paragraph(title, self.styles['Normal'])])

        # Batch into rows
        row_data = []
        current_row = []
        for item in traj_images:
            t_cell = Table([[item[0]], [item[1]]], colWidths=[3.6*inch], rowHeights=[3.6*inch, 0.4*inch])
            t_cell.setStyle(TableStyle([('ALIGN', (0,0), (-1,-1), 'CENTER'), ('VALIGN', (0,0), (-1,-1), 'MIDDLE')]))
            current_row.append(t_cell)

            if len(current_row) == 3:
                row_data.append(current_row)
                current_row = []

        if current_row:
            while len(current_row) < 3:
                current_row.append("")
            row_data.append(current_row)

        if row_data:
            t = Table(row_data, colWidths=[3.8*inch, 3.8*inch, 3.8*inch])
            t.setStyle(TableStyle([
                ('ALIGN', (0,0), (-1,-1), 'CENTER'),
                ('VALIGN', (0,0), (-1,-1), 'TOP'),
            ]))
            self.elements.append(t)


    def add_metric_explanations(self):
        self.elements.append(Paragraph("Guide to Metrics", self.styles['Heading2']))
        self.elements.append(Spacer(1, 0.1 * inch))

        definitions = [
            ("<b>ATE RMSE (Absolute Trajectory Error)</b>: Measures the global consistency of the trajectory. Lower is better. Values < 0.1m indicate high precision; > 0.5m suggest significant drift or SLAM failure.", 
             "#ef4444"), # Red
            ("<b>Coverage (%)</b>: Percentage of the Ground Truth free space that was successfully mapped as free or occupied. Higher is better.", 
             "#22c55e"), # Green
            ("<b>Acc. Coverage (%)</b>: Coverage restricted to the area actually visited/approached by the robot. Useful to judge exploration efficiency independent of mission completeness.", 
             "#22c55e"),
            ("<b>Occupancy IoU (Intersection over Union)</b>: Measures how well the estimated obstacles match the Ground Truth walls. Ranges 0 to 1. 1.0 is perfect alignment. < 0.5 usually implies map distortion or offset.", 
             "#6366f1"), # Indigo
            ("<b>SSIM (Structural Similarity)</b>: Visual similarity between generated map and Ground Truth. 1.0 is identical. Captures general structure better than pixel-wise metrics.", 
             "#8b5cf6"), # Purple
            ("<b>Wall Thickness</b>: Average thickness of mapped walls. Thicker walls than reality (e.g. > 15cm) indicate 'blur' or uncertainty in the map.", 
             "#f59e0b"), # Amber
            ("<b>Duration (s)</b>: Total time to map the environment. Lower is better for efficiency, assuming map quality is maintained.", 
             "#64748b"), # Slate
            ("<b>Max CPU (%)</b>: Peak processor usage during the run. Lower indicates better computational efficiency, critical for onboard operations.", 
             "#ef4444"), # Red
            ("<b>Max RAM (MB)</b>: Peak memory usage. Lower is better, especially for constrained hardware platforms.", 
             "#ef4444")  # Red
        ]

        for text, color in definitions:
            p = Paragraph(text, self.styles['Normal'])
            p.textColor = colors.HexColor(color)
            self.elements.append(p)
            self.elements.append(Spacer(1, 0.05 * inch))

        self.elements.append(Spacer(1, 0.2 * inch))

    def add_run_details(self, runs_data):
        for i, run in enumerate(runs_data):
            self.elements.append(PageBreak())
            self.elements.append(Paragraph(f"Detailed Analysis: Run {i+1}", self.styles['SubTitle']))
            run_id = run.get('name', run.get('id', 'Unknown'))
            self.elements.append(Paragraph(f"ID: {run_id}", self.styles['Code']))
            self.elements.append(Spacer(1, 0.15 * inch))

            # --- Automated Interpretation ---
            slam = run.get('slam', 'Unknown SLAM')
            ate = run.get('ate')
            cov = run.get('coverage')
            iou = run.get('occupancy_iou')

            analysis_text = f"<b>Performance Summary:</b><br/>"
            analysis_text += f"The algorithm <b>{slam}</b> "

            # ATE Analysis
            if ate is not None:
                if ate < 0.1: analysis_text += "demonstrated <b>excellent localization accuracy</b> with negligible drift. "
                elif ate < 0.5: analysis_text += "showed <b>acceptable accuracy</b>, though some minor drift occurred. "
                else: analysis_text += "suffered from <b>significant trajectory drift</b>, suggesting localization failure. "

            # Coverage Analysis
            if cov is not None:
                if cov > 90: analysis_text += "Exploration was <b>highly complete</b>, covering most of the environment. "
                elif cov > 50: analysis_text += "Exploration was <b>partial</b>, leaving significant areas unmapped. "
                else: analysis_text += "Exploration was <b>minimal</b>. "

            # IoU Analysis
            if iou is not None:
                if iou > 0.7: analysis_text += "The resulting map structure is <b>highly accurate</b> aligned with Ground Truth. "
                elif iou > 0.4: analysis_text += "The map captures the general layout but contains <b>noticeable distortions</b> or ghost obstacles. "
                else: analysis_text += "The map quality is <b>poor</b>, with little overlap with the actual environment. "

            p = Paragraph(analysis_text, self.styles['Normal'])
            self.elements.append(p)
            self.elements.append(Spacer(1, 0.1 * inch))
            # -------------------------------

            if run.get('reasons'):
                self.elements.append(Spacer(1, 0.1 * inch))
                self.elements.append(Paragraph("Detected Anomalies:", self.styles['Heading4']))
                for reason in run['reasons']:
                    self.elements.append(Paragraph(f"• {reason}", self.styles['Normal']))

    def generate(self):
        self.add_header()
        self.add_summary_table(self.runs_data)
        self.add_metric_explanations() # Added Guide
        self.add_metrics_charts()
        if self.plot_path:
            self.add_plot(self.plot_path)

        self.add_trajectory_comparison(self.runs_data)
        self.add_map_comparison(self.runs_data)
        self.add_run_details(self.runs_data)
        self.doc.build(self.elements)
        return self.output_path

tools.simulator_manager

Simulator Manager - Central manager for all simulators

SimulatorManager

Manages multiple simulator implementations

Source code in tools/simulator_manager.py
class SimulatorManager:
    """Manages multiple simulator implementations"""

    def __init__(self):
        self.simulators: Dict[str, BaseSimulator] = {
            'gazebo': GazeboSimulator(),
            'o3de': O3DESimulator()
        }
        self._active_processes = {}

    def get_simulator(self, name: str) -> Optional[BaseSimulator]:
        """Get simulator by name"""
        return self.simulators.get(name)

    def list_simulators(self) -> Dict[str, Dict[str, any]]:
        """Get info about all simulators"""
        info = {}
        for name, sim in self.simulators.items():
            info[name] = {
                'installed': sim.is_installed(),
                'version': sim.get_version(),
                'install_size_mb': sim.get_install_size_mb(),
                'dependencies': sim.verify_dependencies()
            }
        return info

    def ensure_installed(self, sim_name: str, progress_callback=None) -> bool:
        """Ensure simulator is installed, install if needed

        Args:
            sim_name: Name of simulator ('gazebo' or 'o3de')
            progress_callback: Optional callback(message, percent)

        Returns:
            True if simulator is ready to use
        """
        sim = self.get_simulator(sim_name)
        if not sim:
            if progress_callback:
                progress_callback(f"Unknown simulator: {sim_name}", 0)
            return False

        if sim.is_installed():
            if progress_callback:
                progress_callback(f"{sim_name} already installed", 100)
            return True

        # Try to install
        return sim.install(progress_callback)

    def cleanup_all(self):
        """Cleanup all simulator processes"""
        for sim in self.simulators.values():
            try:
                sim.cleanup()
            except Exception:
                pass

cleanup_all()

Cleanup all simulator processes

Source code in tools/simulator_manager.py
def cleanup_all(self):
    """Cleanup all simulator processes"""
    for sim in self.simulators.values():
        try:
            sim.cleanup()
        except Exception:
            pass

ensure_installed(sim_name, progress_callback=None)

Ensure simulator is installed, install if needed

Parameters:

Name Type Description Default
sim_name str

Name of simulator ('gazebo' or 'o3de')

required
progress_callback

Optional callback(message, percent)

None

Returns:

Type Description
bool

True if simulator is ready to use

Source code in tools/simulator_manager.py
def ensure_installed(self, sim_name: str, progress_callback=None) -> bool:
    """Ensure simulator is installed, install if needed

    Args:
        sim_name: Name of simulator ('gazebo' or 'o3de')
        progress_callback: Optional callback(message, percent)

    Returns:
        True if simulator is ready to use
    """
    sim = self.get_simulator(sim_name)
    if not sim:
        if progress_callback:
            progress_callback(f"Unknown simulator: {sim_name}", 0)
        return False

    if sim.is_installed():
        if progress_callback:
            progress_callback(f"{sim_name} already installed", 100)
        return True

    # Try to install
    return sim.install(progress_callback)

get_simulator(name)

Get simulator by name

Source code in tools/simulator_manager.py
def get_simulator(self, name: str) -> Optional[BaseSimulator]:
    """Get simulator by name"""
    return self.simulators.get(name)

list_simulators()

Get info about all simulators

Source code in tools/simulator_manager.py
def list_simulators(self) -> Dict[str, Dict[str, any]]:
    """Get info about all simulators"""
    info = {}
    for name, sim in self.simulators.items():
        info[name] = {
            'installed': sim.is_installed(),
            'version': sim.get_version(),
            'install_size_mb': sim.get_install_size_mb(),
            'dependencies': sim.verify_dependencies()
        }
    return info

tools.benchmark

read_bag_data(bag_path)

Reads a rosbag and extracts: - GT Poses: (timestamp, x, y) from /gazebo/model_states - TF Updates: list of (timestamp, TransformStamped)

Source code in tools/benchmark.py
def read_bag_data(bag_path):
    """
    Reads a rosbag and extracts:
    - GT Poses: (timestamp, x, y) from /gazebo/model_states
    - TF Updates: list of (timestamp, TransformStamped)
    """
    bag_path = str(bag_path)
    if not Path(bag_path).exists():
        raise FileNotFoundError(f"Bag not found: {bag_path}")

    # Try explicit sqlite3 first
    try:
        storage_options = StorageOptions(uri=bag_path, storage_id='sqlite3')
        converter_options = ConverterOptions(
            input_serialization_format='cdr',
            output_serialization_format='cdr')
        reader = SequentialReader()
        reader.open(storage_options, converter_options)
    except Exception as e1:
        print(f"Failed to open with sqlite3: {e1}")
        # Try auto-detection
        try:
            storage_options = StorageOptions(uri=bag_path, storage_id='')
            reader = SequentialReader()
            reader.open(storage_options, converter_options)
        except Exception as e2:
            print(f"Failed to open with auto-detect: {e2}")
            return [], [], []

    topic_types = reader.get_all_topics_and_types()
    type_map = {t.name: t.type for t in topic_types}

    gt_poses = []
    tf_data = []
    odom_data = []

    msg_count = 0
    while reader.has_next():
        (topic, data, t_ns) = reader.read_next()
        msg_type_str = type_map[topic]

        try:
            msg_type = get_message(msg_type_str)
            msg = deserialize_message(data, msg_type)
        except Exception as e:
            # print(f"Deserialization error for {topic}: {e}")
            continue

        if topic == '/gazebo/model_states':
            # msg is ModelStates
            try:
                idx = -1
                for i, name in enumerate(msg.name):
                    # Robust robot detection: look for standard TB3 names but ignore known static environments
                    # 'turtlebot3_house', 'turtlebot3_world' are static maps.
                    # 'turtlebot3_waffle', 'turtlebot3_burger', 'waffle', 'burger' are robots.

                    if ('turtlebot3' in name or 'waffle' in name or 'burger' in name):
                        if 'house' in name or 'world' in name or 'ground' in name:
                            continue # Skip static environment models

                        idx = i
                        break

                if idx != -1:
                    pose = msg.pose[idx]
                    gt_poses.append((t_ns, pose.position.x, pose.position.y))
            except Exception:
                pass

        elif topic == '/tf' or topic == '/tf_static':
            for transform in msg.transforms:
                # Normalize frame names
                transform.header.frame_id = transform.header.frame_id.strip().lstrip('/')
                transform.child_frame_id = transform.child_frame_id.strip().lstrip('/')
                tf_data.append((t_ns, transform))

        elif topic == '/odom':
            # msg is Odometry
            # Normalize frame names
            msg.header.frame_id = msg.header.frame_id.strip().lstrip('/')
            msg.child_frame_id = msg.child_frame_id.strip().lstrip('/')

            # Store as translation and quaternion
            trans = msg.pose.pose.position
            rot = msg.pose.pose.orientation
            odom_data.append((t_ns, np.array([trans.x, trans.y, trans.z]), np.array([rot.x, rot.y, rot.z, rot.w])))

        msg_count += 1

    print(f"Read: {len(gt_poses)} GT, {len(tf_data)} TF, {len(odom_data)} Odom.")
    return gt_poses, tf_data, odom_data