How the Three-Phase Pipeline (Prepare, Run, Evaluate) Coordinates Workflows in LingBot-Map

The three-phase pipeline coordinates through shared ConfigManager and BSSManager abstractions that ensure prepare.py, run.py, and evaluate.py operate on identical dataset-method selections, workspace paths, and artifact completion states without requiring external orchestration tools.

LingBot-Map (available at Robbyant/lingbot-map) implements a deterministic benchmarking workflow that transforms raw sensor datasets into quantitative metrics. Rather than relying on complex workflow engines, the system achieves coordination through centralized configuration management and a unified storage structure, allowing each phase to execute independently while maintaining strict consistency.

The Three Phases and Their Responsibilities

The pipeline consists of three sequential stages implemented in separate modules within the benchmark/ directory.

Prepare Phase (benchmark/prepare.py)

The prepare phase converts raw sensor recordings into the standardized Benchmark Storage Structure (BSS). It loads raw frames in parallel, extracts per-frame RGB, depth, and pose data, and writes global metadata including camera intrinsics and timestamps. The phase uses ClassLoader (from benchmark/core/registry.py) to instantiate dataset classes and BSSSaver (from benchmark/core/saver.py) to persist data, finally invoking gt_artifact.mark_complete(metadata=...) to signal readiness.

Run Phase (benchmark/run.py)

The run phase executes one or more methods (such as neural network models) against the prepared BSS artifacts. For each dataset-method combination, it loads ground truth via BSSLoader (from benchmark/core/loader.py) and calls method.process_scene. The phase supports two execution modes: in-process execution for pure Python methods, or isolated subprocess execution via run_via_subprocess (lines 91-112) when the method configuration specifies an env field. Predictions are saved using BSSSaver, and completion is tracked via pred_artifact.mark_complete.

Evaluate Phase (benchmark/evaluate.py)

The evaluate phase computes quantitative metrics and aggregates results across three hierarchical layers. Layer 1 contains per-scene evaluation JSON files produced by evaluators like AUCEvaluator and TrajectoryEvaluator; Layer 2 aggregates scene-level cross-method results; Layer 3 produces dataset-level summaries. The aggregate_eval function (lines 25-92) orchestrates this by calling bss_manager.save_scene_eval for Layer 2 and bss_manager.save_dataset_eval for Layer 3, using _average_dicts (lines 94-102) for numeric aggregation.

Coordination Through Shared Abstractions

The phases remain synchronized through two primary shared objects defined in benchmark/core/.

ConfigManager Unifies Selection

All three scripts instantiate ConfigManager (benchmark/core/config.py) from the same YAML configuration file passed via --config. Identical method calls to get_selected_dataset_names() and get_selected_method_names() (visible at lines 31-36 in prepare.py and lines 21-26 in run.py) guarantee that each phase processes the exact same dataset-method pairs. This eliminates configuration drift between preparation and execution.

BSSManager Provides Storage Coherence

BSSManager (benchmark/core/storage.py) creates a consistent view of the workspace directory obtained via ConfigManager.get_workspace(). It provides get_artifact(dataset, scene, name) to locate data folders and list_scenes(dataset) to enumerate contents. Because all phases instantiate BSSManager(workspace) using the same root path, a scene prepared by prepare.py is immediately visible to run.py and evaluate.py without path translation.

Artifact Completion Protocol

Phases communicate state through artifact metadata flags. After successful completion, prepare.py writes the complete flag using gt_artifact.mark_complete(metadata=...), while run.py uses pred_artifact.mark_complete. Subsequent phases check these flags implicitly through BSSLoader or list_methods to skip reprocessing unless the user supplies --force.

Method Dispatch and Execution Isolation

The run phase handles heterogeneous execution requirements through conditional dispatch logic defined in benchmark/run.py.

In-process execution occurs when the method configuration lacks an env field, allowing direct method invocation within the main Python process. This path is implemented in the run_scene function (line 31).

Subprocess isolation activates when env is defined. The run_via_subprocess function constructs a conda run command (lines 91-112) targeting benchmark/run_worker.py. This design isolates conflicting GPU dependencies or heavy compute workloads while maintaining the same BSS I/O interface.

Running the Complete Pipeline

Execute the three phases sequentially using the same configuration file to maintain coordination:

# Phase 1: Convert raw datasets to BSS format

python benchmark/prepare.py --config configs/base.yaml

# Phase 2: Run methods on prepared data

python benchmark/run.py --config configs/base.yaml

# Phase 3: Compute and aggregate metrics

python benchmark/evaluate.py --config configs/base.yaml

The following excerpt from benchmark/run.py demonstrates how the coordination objects are instantiated and used:

from pathlib import Path
from benchmark.core.config import ConfigManager
from benchmark.core.storage import BSSManager
from benchmark.core.registry import ClassLoader

# Shared configuration handling

config_manager = ConfigManager(Path(args.config))
workspace = Path(config_manager.get_workspace())
bss_manager = BSSManager(workspace)

# Consistent dataset and method selection

dataset_names = config_manager.get_selected_dataset_names()
method_names = config_manager.get_selected_method_names()

for dataset_name in dataset_names:
    for method_name in method_names:
        # Load method with potential environment isolation

        method_cfg = config_manager.get_method_config(method_name)
        method_cls = ClassLoader.load_method(method_cfg['class'])
        method = method_cls(logger=logger, **method_cfg.get('params', {}))
        
        # Resolve ground truth artifact via BSSManager

        gt_artifact = bss_manager.get_artifact(dataset_name, scene, 'gt')
        
        # Choose execution path based on environment config

        if method_cfg.get('env'):
            run_via_subprocess(method_cfg['env'], args.config, method_name,
                               dataset_name, scene, args.force, logger)
        else:
            run_scene(method, gt_artifact, pred_artifact, logger)

Summary

• ConfigManager ensures all phases use identical dataset-method selections by parsing a single YAML configuration file.
• BSSManager provides a unified workspace view via benchmark/core/storage.py, allowing seamless data handoff between prepare, run, and evaluate phases.
• Artifact completion flags written by mark_complete() prevent redundant processing while supporting forced recomputation via --force.
• Dual execution paths in the run phase support both in-process methods and isolated conda environments via run_via_subprocess without breaking the storage abstraction.
• Three-layer aggregation in the evaluate phase enables detailed per-scene analysis alongside dataset-level benchmarking through aggregate_eval.

Frequently Asked Questions

How does LingBot-Map ensure all three phases process the same datasets?

All phases instantiate ConfigManager from benchmark/core/config.py using the same --config argument. Identical calls to get_selected_dataset_names() and get_selected_method_names() return the same lists in prepare.py (lines 31-36) and run.py (lines 21-26), guaranteeing consistency without external orchestration.

What prevents the run phase from executing before data is prepared?

The BSSManager checks artifact completion flags written by gt_artifact.mark_complete(metadata=...) in prepare.py. The run phase loads artifacts via BSSLoader, which respects these flags, and the list_methods function filters out incomplete artifacts unless the --force flag is provided.

Can methods with conflicting dependencies run in the same pipeline?

Yes. The run phase in benchmark/run.py supports environment isolation through the env configuration field. When specified, run_via_subprocess (lines 91-112) dispatches the method via conda run in a separate process executing benchmark/run_worker.py, while methods without env run in-process within the same Python interpreter.

How are evaluation metrics aggregated across multiple scenes?

The evaluate phase uses a three-layer hierarchy. Scene-level evaluators generate Layer 1 JSON files. The aggregate_eval function (lines 25-92 in evaluate.py) collects these into Layer 2 scene-level results via bss_manager.save_scene_eval, then computes Layer 3 dataset-level aggregates via bss_manager.save_dataset_eval using helpers like _average_dicts (lines 94-102).

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}