rfgen.core.registry

Plugin discovery and the metadata schema every plugin package declares. The registry is what turns “an installed package on the Python path” into “an emitter, channel, labeler, store, or command the framework can resolve from config.” A single concrete registry ships: EntryPointRegistry, which reads importlib.metadata.entry_points. An optional second primitive, pluggy, is gated behind the rfgen[plugin-hooks] extra for hook-based discovery.

Module summary

from rfgen.core.registry import EntryPointRegistry, PluginMetadata
from rfgen.waveforms.base import BaseEmitter

emitters = EntryPointRegistry[BaseEmitter]("rfgen.emitters")
emitters.discover()              # records names; does not import any plugin module
print(emitters.available())      # ('adsb', 'analog_fpv_video', 'apsk', 'ble', ...)
emitter = emitters.get("chirp_radar")  # this call triggers the lazy import

Discovery is two-phase by design: discover() walks the entry-point index and records the names; get(name) is the first call that imports the plugin module. Importing rfgen.core.registry itself never imports any third-party plugin.

Class index

Class	Kind	Notes
BaseRegistry	abc	Generic ABC; implements register, get, available, metadata_for.
EntryPointRegistry	concrete	Single shipped subclass; backed by importlib.metadata.entry_points.
PluginMetadata	datatype (Pydantic)	Frozen plugin descriptor.
BaseCommand	protocol	Dispatch contract for services callable from the CLI.

The constant EXIT_CODES_VALID = frozenset({0, 1, 2, 3, 130}) is also exported. It is the closed set of process exit codes a BaseCommand is allowed to return.

rfgen.core.registry.refresh_plugin_discovery

def refresh_plugin_discovery() -> tuple[str, ...]

Re-scan installed distributions for plugin entry points and return every discovered group name, sorted.

The entry-point index is read once per process, which is correct when sys.path is fixed before rfgen is imported. A distributed worker breaks that assumption: Spark appends a submission’s --py-files archives to sys.path around the time it imports the driver, so an index built first would omit exactly the plugins that submission shipped. Call this before the first registry lookup in that situation; ordinary local runs never need it.

See Shipping plugin packages for the remote path that depends on it.

---

class rfgen.core.registry.BaseRegistry

from abc import ABC, abstractmethod
from collections.abc import Callable
from typing import Generic, TypeVar

T = TypeVar("T")


class BaseRegistry(ABC, Generic[T]):
    """Generic ABC for plugin registries."""

    def register(
        self,
        name: str,
        factory: Callable[..., T],
        *,
        source: str = "<unknown>",
        metadata: PluginMetadata | None = None,
    ) -> None: ...

    def get(self, name: str) -> T: ...

    def available(self) -> tuple[str, ...]: ...

    def metadata_for(self, name: str) -> PluginMetadata | None: ...

    @abstractmethod
    def discover(self) -> None: ...

A generic ABC parameterized by the produced plugin type T. Subclasses implement discover(); the shared registration and lookup API lives on the base class so subclasses do not duplicate it.

register(name, factory, *, source, metadata)

Register a factory under a name.

• name: canonical name for the plugin entry.

• factory: zero-or-more-argument callable returning a T.

• source: human-readable origin of the registration (entry-point module path, file path, …). Used in conflict error messages.

• metadata: optional PluginMetadata describing the plugin. If provided and metadata.requires lists an extra whose marker package is not installed, registration raises BackendUnavailableError.

Raises:

• RegistryError: if name is already registered. The message names both source modules.

• BackendUnavailableError: if metadata.requires lists an extra whose marker package is not present.

get(name)

Resolve name and return the constructed plugin instance. Raises PluginNotFoundError if name is not registered; the message lists the available names for this registry.

available()

Return the sorted tuple of registered names. On EntryPointRegistry, this includes pending entry-point names that have been discovered but not yet loaded.

metadata_for(name)

Return the PluginMetadata registered for name, or None if no metadata was registered.

discover()

Abstract. Concrete subclasses populate the registry from the underlying source. Must be idempotent.

---

class rfgen.core.registry.EntryPointRegistry

class EntryPointRegistry(BaseRegistry[T]):
    """A BaseRegistry backed by importlib.metadata.entry_points."""

    def __init__(self, group: str) -> None: ...

    @property
    def group(self) -> str: ...

    def discover(self) -> None: ...
    def get(self, name: str) -> T: ...
    def available(self) -> tuple[str, ...]: ...

The single concrete BaseRegistry subclass shipped with rfgen. It is constructed with a single entry-point group name. The framework instantiates one registry per documented group:

• rfgen.emitters

• rfgen.channels

• rfgen.augmentations

• rfgen.radar_responses

• rfgen.scene_renderers

• rfgen.scene_composers

• rfgen.channel_plan_sources

• rfgen.time_placement

• rfgen.freq_placement

• rfgen.grid_sources

• rfgen.labelers

• rfgen.dataset_stores

• rfgen.annotators

• rfgen.annotation_backends

• rfgen.backend_capabilities

• rfgen.inference_clients

• rfgen.geometry_ingests

• rfgen.plan_exporters

• rfgen.metrics

• rfgen.executors

• rfgen.commands

Two more groups are declared in code rather than in pyproject.toml, and a plugin author publishing under either is loaded normally:

• rfgen.observation_projections (rfgen.observation.runtime.resolution)

• rfgen.annotation.credentials (rfgen.annotation.credentials)

The two scene-seam groups

Two of the groups above were added by the external-scene-seam work and their contracts are stated on their own pages; what follows is what a plugin author needs to know about the groups themselves.

rfgen.geometry_ingests registers implementations of the geometry ingestion contract: one name resolves to one class that declares which GeometryAssetKind values it reads. The refusal is the dispatcher’s rather than the class’s: it asserts the kind in hand against that declaration and raises geometry_format_not_ingestible by name, so one assertion covers an out-of-tree ingest as well as the in-repo one. The default is sionna_mitsuba, which is what an absent world_ingest means, so an absent value in stored provenance is itself a disclosure rather than a gap. A registered ingest is selected by name from configuration and the selection is asserted against the class’s own declaration, so a name that resolves to an ingest which cannot read the configured kind fails before any engine work. Contract: Engine.

rfgen.plan_exporters registers serializations of a ScenePlan into an external scene description: one name resolves to one class declaring a name, a filename suffix, and an export(plan, *, time_codes_per_second) returning text. The in-repo entry is usda. The group exists because the format an external engine wants is a property of that engine, so a second exporter, a .usdc binary writer or a glTF one, is a registration rather than a core change; rfgen export-plan --exporter NAME resolves through this group, so an out-of-tree exporter is reachable from the shipped command line the day its distribution is installed. Contract: Scene.

Lazy discovery

discover() walks the entry-point index and records every entry in the configured group, but does not call EntryPoint.load(). The first get(name) call for a pending name pops the entry, calls load(), and registers the resulting object as the factory under source="entry_point:<group>:<name>".

The full entry-point index is fetched once per process via a private _all_entry_points() helper wrapped in functools.lru_cache(maxsize=1). Entry points are statically declared in installed distribution metadata and do not change at runtime, so a single process-wide scan is safe. Across every registry rfgen constructs, this avoids one full metadata scan each.

Same-name conflict

Two installed packages declaring the same name in the same entry-point group raise RegistryError at registration time; the message names both source modules.

Missing-extra conflict

If a plugin’s PluginMetadata.requires lists an extra whose marker package is not installed, registration raises BackendUnavailableError, naming the missing extra. A legacy runtime error may spell pip install rfgen[<extra>], but that is not a supported source-checkout route; from the checkout root use uv pip install -e '.[<extra>]', or reinstall a built release wheel with its matching extra.

---

class rfgen.core.registry.PluginMetadata

from pydantic import BaseModel, ConfigDict, Field, field_validator
import packaging.version


class PluginMetadata(BaseModel):
    """Standardized descriptor for a plugin package."""

    model_config = ConfigDict(frozen=True)

    name: str = Field(min_length=1, pattern=r"^[A-Za-z][A-Za-z0-9_-]*$")
    version: str
    family: str | None = None
    summary: str = Field(default="", max_length=500)
    requires: list[str] = Field(default_factory=list)
    homepage: str | None = None
    license: str = "Apache-2.0"

A frozen Pydantic v2 model. Validates on construction and is the single source of truth for what a plugin claims to provide and what extras it needs. Frozen instances are read-only after construction; this matches the frozen-by-default posture of the dataclasses and prevents silent corruption of registry state by code that holds a shared reference.

Fields

Field	Type	Required	Default	Description
name	str	yes	none (required)	PyPI distribution name, e.g. "rfgen-myradar". Must match the documented identifier pattern.
version	str	yes	none (required)	SemVer / PEP 440 version of the plugin package. Validated via packaging.version.Version.
family	str | None	no	None	Family tag (emitter family, channel backend tag, …).
summary	str	no	""	Short human-readable description, max 500 chars.
requires	list[str]	no	[]	List of pip extras that must be installed for this plugin to operate. Registration fails with BackendUnavailableError if any required extra is missing.
homepage	str | None	no	None	Project URL, typically the GitHub repo.
license	str	no	"Apache-2.0"	SPDX license identifier.

Validation

• name is rejected unless it matches ^[A-Za-z][A-Za-z0-9_-]*$.

• version is rejected if packaging.version.Version(v) raises InvalidVersion.

• summary longer than 500 characters is rejected.

• The model is frozen; assigning to a field after construction raises pydantic.ValidationError.

Illustrative API-signature sketch (non-runnable)

from rfgen.core.registry import PluginMetadata

PLUGIN = PluginMetadata(
    name="rfgen-myradar",
    version="0.2.1",
    family="radar",
    summary="High-fidelity polyphase pulsed radar emitter for rfgen.",
    requires=["torchsig"],
    homepage="https://github.com/myorg/rfgen-myradar",
    license="Apache-2.0",
)

---

class rfgen.core.registry.BaseCommand

from typing import Protocol, runtime_checkable
from pydantic import BaseModel


@runtime_checkable
class BaseCommand(Protocol):
    """Dispatch contract for services callable from the CLI."""

    name: str
    summary: str
    params_schema: type[BaseModel]

    def run(
        self,
        *,
        params: BaseModel,
        registry: BaseRegistry[object],
        log_sink: object,
    ) -> int: ...

A runtime-checkable Protocol. Each service owns and registers its BaseCommand implementation under entry-point group rfgen.commands; the CLI dispatches to them by name through the registry, never through build-time imports. Generation is owned by rfgen.generation, annotation by its own Typer sub-app, and dataset inspection by rfgen.inspection.audit. rfgen.commands is an entry-point group, not a Python module.

Attributes

• name: canonical command name (used by the CLI for dispatch and listings).

• summary: one-line human description.

• params_schema: Pydantic model describing the parameter surface.

run(*, params, registry, log_sink)

• params: validated parameters; an instance of params_schema.

• registry: the plugin registry the command should resolve names through.

• log_sink: structured-log sink, typed as object here. The registry module cannot import BaseLogSink without a circular dependency; the CLI dispatcher tightens the type where that import is already in scope.

Returns a process exit code from the closed set EXIT_CODES_VALID = frozenset({0, 1, 2, 3, 130}) (success, config error, runtime error, validation failure, keyboard interrupt). Any other return value is a contract violation; the contract test in validation-and-audit asserts every shipped command returns a member of EXIT_CODES_VALID.

---

End-to-end: shipping a third-party emitter plugin

This is the canonical path from “I wrote an emitter” to “rfgen finds and loads it.” No registry code in rfgen needs to change to support a new plugin; only the plugin author’s pyproject.toml does.

1. The plugin author writes a class

# rfgen_myradar/emitter.py
from rfgen.waveforms.base import BaseEmitter

class MyRadarEmitter(BaseEmitter):
    """A polyphase pulsed radar emitter shipped as a third-party package."""
    ...

2. They declare an entry point in their pyproject.toml

[project]
name = "rfgen-myradar"
version = "0.2.1"

[project.entry-points."rfgen.emitters"]
my_radar = "rfgen_myradar.emitter:MyRadarEmitter"

The entry-point group rfgen.emitters is one of the twenty-three documented groups. The key (my_radar) is the canonical name the framework will resolve through; the value is the dotted path to the factory.

3. A source-checkout user installs both packages

uv pip install -e . -e /path/to/rfgen-myradar

4. rfgen discovers it via EntryPointRegistry

from rfgen.core.registry import EntryPointRegistry
from rfgen.waveforms.base import BaseEmitter

emitters: EntryPointRegistry[BaseEmitter] = EntryPointRegistry("rfgen.emitters")
emitters.discover()                  # records 'my_radar' in the pending table
assert "my_radar" in emitters.available()

instance = emitters.get("my_radar")  # this is the call that imports rfgen_myradar.emitter
assert isinstance(instance, BaseEmitter)

The discover() call walks the cached entry-point index, filters by group, and records my_radar as pending. Importing rfgen_myradar.emitter does not happen until get("my_radar") is called.

5. End users hit this path through config, not directly

In normal use, a user writes a YAML config that names the emitter as a string:

emitters:
  - name: my_radar
    params: {...}

Hydra parses the YAML, the Pydantic emitter config validates the string at instantiation time, and the framework resolves it through the same EntryPointRegistry instance. The user never calls get directly.

Lazy-import contract

import rfgen.core.registry does not import any plugin module. Even registry.discover() does not. Only get(name) triggers the import for that specific entry. This keeps cold-start cost bounded by the number of plugins the user actually resolves, not the number installed on the system.

---

Optional hook-based discovery via pluggy

pluggy is the optional second primitive, gated behind the rfgen[plugin-hooks] extra. It is imported defensively in rfgen.core.registry:

try:
    import pluggy as _pluggy
    _PLUGGY_AVAILABLE = True
except ImportError:
    _PLUGGY_AVAILABLE = False

pluggy is the layered hook-mechanism behind pytest, tox, and devpi. It complements EntryPointRegistry for discovery patterns that go beyond “one name, one factory”, for example, a hook that lets multiple plugins register schema fragments under the same config key, or one that lets plugins observe pipeline events. pluggy serves the “observe annotation events” hook surface; entry points remain the primary registration channel.

The base rfgen install does not require pluggy. Code that wants the hook surface checks _PLUGGY_AVAILABLE and degrades cleanly when the extra is not installed.

---

Why JsonManifestRegistry is intentionally omitted

JsonManifestRegistry, PrivateIndexRegistry, and CompositeRegistry are not public registry implementations.

importlib.metadata.entry_points plus pluggy already cover wheel-time and layered hook-based discovery. A manifest reader would grow maintenance surface (a JSON schema, a fetcher, a cache, an offline-mode flag, a validation pass on remote records) without adding capability the existing two primitives do not already provide. Private corporate registries are covered by configuring the local Python environment to install from a private index (the standard pip workflow), at which point EntryPointRegistry finds the plugin like any other installed package.

The public registry contract is therefore entry-point discovery plus optional hook registration; no manifest-reader API is exposed.

---

See Also

• Reference / Plugin Metadata: declaration site of PLUGIN: PluginMetadata on plugin packages, plus the PLUGIN_CARD.md convention.

• Reference / Project layout: the package directories the entry-point groups resolve into. The list above is the authority for all twenty-three groups: pyproject.toml declares twenty-one of them and the remaining two are declared in code. It declares built-in entries within those groups, but no group beyond them.

• BackendUnavailableError, PluginNotFoundError, RegistryError: the error surface this module raises.

• Emitter API: the retained emitter contract for an installed plugin.

---

Legacy class names

Legacy: JsonManifestRegistry

Intentionally not shipped. See the “Why JsonManifestRegistry is intentionally omitted” section above for the documented user-confirmed decision. Existing cross-references (e.g., from docs/background/open-questions.md) resolve to this anchor so the docs render; the class itself does not exist in rfgen.core.registry.

---

Load a class or factory without constructing it

EntryPointRegistry.load_class(name: str) -> RegistryFactory[T] returns the registered class or factory for name; it does not call the factory or return an instance. Like get(name), it loads a pending entry point lazily before it returns the registered target. It raises PluginNotFoundError when the name is not registered.

Use it when a caller must inspect a component type before construction. A caller can verify that a selected target implements the relevant emitter or channel ABC before instantiating it. Registry errors preserve the group, selector, and conflicting entry-point values in their error context.

Installed implementation ownership

implementation_identity returns the canonical module-qualified class name. implementation_distribution resolves the single installed distribution that owns that module, refusing ambiguous or absent production ownership. Graph implementation bindings and bounded waveform execution share this primitive so an unrelated installed package cannot be claimed as provenance.