110. Writing custom plugins

When no existing plugin reads your data or produces the output you need, write your own. Your plugin imports base classes from the installed evaluation_pipeline package. Implement your plugin in a separate module and point the flow at it. Your plugin does not modify or recompile the pipeline itself — it is loaded dynamically at runtime.

This section assumes you are familiar with selecting and configuring existing plugins (see Using Evaluation Pipeline plugins).

110.1 Plugin types

Each plugin type extends a base class and implements one or more abstract methods.

110.2 Dynamic loading

1. The pipeline reads the fully qualified class name from a library fragment (for example, my_plugins.parsers.CsvParser).
2. If the fragment lists sys_path entries, those directories are prepended to sys.path, making your code importable.
3. The pipeline imports and instantiates the class with the config dictionary, then calls its abstract method in-process.

To make a plugin available, add a YAML fragment to a plugin library and point FTX_EVALUATE_<LIBRARY>_PLUGIN_LIB at that library's directory.

110.3 Step by step: a custom parser

The parser is the most common plugin to write — it converts raw recording data into a Foretellix protobuf object list.

110.3.1 Step 1. Write the plugin module

# File: /home/user/my_plugins/parsers/csv_parser.py

from pathlib import Path
from evaluation_pipeline.plugins.parsers import BaseParser


class CsvParser(BaseParser):
    """Convert CSV recording data to a protobuf object list."""

    def parse(self, input_path: Path, output_dir: Path) -> Path:
        output_file = output_dir / "recording_ol.pb"

        delimiter = self.config.get("delimiter", ",")
        # ... your conversion logic here ...

        return output_file

110.3.2 Step 2. Add a library fragment

Create a plugin library directory and point an environment variable at it — for example, set FTX_EVALUATE_MY_PLUGIN_LIB to /home/user/my_plugin_lib. Then drop a fragment in the stage's subdirectory:

# /home/user/my_plugin_lib/parsers/csv.yaml
class: my_plugins.parsers.csv_parser.CsvParser
sys_path:
  - /home/user            # parent of the my_plugins package
config:
  delimiter: ","

The sys_path entry points at the parent directory of your plugin package so the class is importable.

110.3.3 Step 3. Reference the plugin in the flow

stages: [explore, parse, denoise, ingest, match, post_match]

user_top_file: $FTX/path/to/user_config_top.osc

explore:
  plugin: common.pandaset_explorer

parse:
  plugin: my.csv

denoise:
  config: /home/user/configs/denoiser_config.yaml

110.3.4 Step 4. Test the plugin

$FTX_EVALUATE_EXE run-single \
    --drive_data /data/recordings/recording_001.csv \
    --output_path /tmp/test_parser \
    --flow /path/to/flow_config.yaml \
    --keep_intermediate_results raw \
    --debug

Use --keep_intermediate_results raw to inspect every stage's output and --debug for verbose logging.

110.4 More examples

110.4.1 Custom explorer

from pathlib import Path
from evaluation_pipeline.plugins.explorers import BaseExplorer, ExploreResult


class RosbagExplorer(BaseExplorer):
    """Discover data files from a recording directory."""

    required_fields = frozenset({"map_data"})

    def explore(self, data_path: Path) -> ExploreResult:
        map_file = self.discover_sibling(data_path, "*.xodr")
        if map_file is None:
            raise FileNotFoundError(f"No .xodr map found next to {data_path}")
        return ExploreResult(drive_data=str(data_path), map_data=map_file)

# /home/user/my_plugin_lib/explorers/rosbag.yaml
class: my_plugins.explorers.rosbag_explorer.RosbagExplorer
sys_path:
  - /home/user            # parent of the my_plugins package

110.4.2 Custom external matcher

from pathlib import Path
from evaluation_pipeline.plugins.ext_matchers import BaseExtMatch


class MyExtMatch(BaseExtMatch):
    """Enrich run results with external match intervals."""

    def ext_match(self, run_dir: Path, matches_path: Path, output_dir: Path) -> Path:
        self.copy_run_dir(run_dir, output_dir)
        for rd_file in self.find_run_data_files(output_dir):
            enrich_run_data(rd_file, matches_path, **self.config)
        return output_dir

# /home/user/my_plugin_lib/ext_matchers/my.yaml
class: my_plugins.ext_matchers.my_ext_match.MyExtMatch
sys_path:
  - /home/user            # parent of the my_plugins package
config:
  threshold: 0.8

110.4.3 Custom post-processor

from pathlib import Path
from evaluation_pipeline.plugins.post_processors import BasePostProcess


class ReportGenerator(BasePostProcess):
    """Generate an HTML report from pipeline results."""

    def post_process(self, context_dir: Path, output_dir: Path) -> None:
        post_match_dir = context_dir / "post_match"
        report = generate_html_report(post_match_dir, **self.config)
        (output_dir / "report.html").write_text(report)

# /home/user/my_plugin_lib/post_processors/report.yaml
class: my_plugins.post_processors.report_generator.ReportGenerator
sys_path:
  - /home/user            # parent of the my_plugins package
config:
  format: html
  include_charts: true

110.4.4 Running post-processing after a failed run

By default, post_process runs only when every upstream stage succeeds — if an earlier stage fails, the pipeline stops and post_process is skipped. To run a post-processor at the end of the run even when an upstream stage failed — for example, to emit a failure report or notify an external system — enable execute_on_failed_run.

Set it in the flow YAML's post_process block to control the behavior per flow without editing plugin code:

post_process:
  class: my_plugins.report_generator.ReportGenerator
  execute_on_failed_run: true   # run even if an upstream stage failed

Alternatively, set the execute_on_failed_run class attribute on the plugin to make it the default for every flow that uses the plugin:

class ReportGenerator(BasePostProcess):
    execute_on_failed_run = True   # run even if an upstream stage failed

    def post_process(self, context_dir: Path, output_dir: Path) -> None:
        ...

When both are set, the YAML setting takes precedence over the class attribute. A plugin library fragment can also set execute_on_failed_run; like any fragment value, the main flow YAML overrides it, and it overrides the class attribute.

The run is still reported as failed — the flag only guarantees the post-processor gets a chance to run. Inspect self.stage_results to see which stages succeeded or failed, and self.last_successful_stage for the last stage that completed. Because an upstream stage failed, the artifacts your post-processor expects may be missing, so guard against absent stage directories.

110.5 Plugin configuration

The config dictionary from the flow YAML is passed to your plugin's constructor and is accessible as self.config. You can add any key-value pairs in the config dictionary. The pipeline does not validate the contents, it forwards the dictionary to your class.

110.6 Parser attributes

The pipeline sets several attributes on a parser instance before calling parse(). Read them to adapt your parser's behavior.

If your parser discovers a map or video during parsing, set self.map_data or self.video_data and the pipeline propagates those values to downstream stages. Likewise, set self.projection to expose the recording's coordinate reference system to the map stage and other downstream consumers.

110.7 Quality gates

If a parser determines the input data quality is too poor to produce reliable downstream results, it can raise evaluation_pipeline.core.errors.QualityGateError. The pipeline catches it, records the failure type as QUALITY_GATE, and halts the run. The same mechanism is available to denoiser and map-generation implementations.

110.8 Available Python libraries

The ftx_evaluate binary is a frozen executable. Your plugin code is loaded dynamically at runtime, but any package it imports must already be bundled in the binary. The following libraries are available:

Standard library modules (json, pathlib, subprocess, argparse, xml, csv, and so on) are always available.

110.8.1 Unbundled dependencies

If your plugin requires a Python package that is not bundled in the ftx_evaluate binary, you cannot import it directly — the import fails with ImportError.

To use an unbundled package, compile your module into a standalone executable (for example, with PyInstaller) that carries all the dependencies it needs, and call it from your plugin as a subprocess:

import subprocess
from pathlib import Path
from evaluation_pipeline.plugins.parsers import BaseParser


class MyParser(BaseParser):
    """Parser that delegates to a separately compiled binary."""

    def parse(self, input_path: Path, output_dir: Path) -> Path:
        output_pb = output_dir / "recording_ol.pb"
        binary = self.config["converter_binary"]

        result = subprocess.run(
            [binary, "--input", str(input_path), "--output", str(output_pb)],
            capture_output=True,
            text=True,
        )
        if result.returncode != 0:
            raise RuntimeError(
                f"Converter failed (exit {result.returncode}):\n{result.stderr}"
            )
        return output_pb

# /home/user/my_plugin_lib/parsers/binary.yaml
class: my_plugins.parsers.my_parser.MyParser
sys_path:
  - /home/user            # parent of the my_plugins package
config:
  converter_binary: /home/user/bin/my_converter

This keeps the plugin itself lightweight — only the standard library plus bundled packages — while delegating the processing to an external process that has its own dependency environment.

110.9 Notes for plugin authors

• No recompilation needed. Plugins are loaded dynamically. Place your code anywhere on disk and point to it with sys_path.
• sys_path is local only. Cloud storage URLs are not supported in sys_path entries.
• Output is captured. The pipeline redirects standard output and error to the stage log file during plugin execution, so print() and standard logging are captured automatically.
• Signal failure with an exception. Raise an exception to fail the stage — the pipeline records the error in the run checkpoint and marks the stage as failed.

110.10 Troubleshooting custom plugins