Add a custom emitter

Add a new waveform family that the framework can use anywhere a built-in emitter can.

Warning

Pre-implementation. APIs target v0.5+.

The shape of a custom emitter

A custom emitter is a subclass of BaseEmitter that implements generate() and schema(). The framework handles registration discovery, config validation, and contract checks. Scene and channel stages handle placement, transmit (TX) impairments, propagation, receive capture, and receiver hardware effects after the emitter returns clean IQ.

Example: a simple chirp emitter

Note

Class names like LinearChirpEmitter, ChirpParams, MyRadioEmitter, and MyRadioParams in this tutorial are illustrative; they show what your own subclass would look like. Real framework classes (e.g. BaseEmitter, SignalMetadata) link to the API reference.

# my_chirps.py
import torch
from pydantic import BaseModel, Field

from rfgen.emitters.base import BaseEmitter
from rfgen.core.types import Signal, SignalMetadata, IQ, SampleRate
from rfgen.core.registry import register_emitter
from rfgen.enums import EmitterFamily


class ChirpParams(BaseModel):
    """Schema for the `params` block this emitter accepts."""
    carrier_hz: float = Field(default=2.4e9, gt=0)
    chirp_rate_hz_per_s: float = Field(default=1e9, gt=0)
    chirp_duration_s: float = Field(default=10e-6, gt=0)
    pulse_repetition_us: float = Field(default=100.0, gt=0)


@register_emitter(family=EmitterFamily.RADAR)
class LinearChirpEmitter(BaseEmitter):
    family = EmitterFamily.RADAR
    supported_classes = ("up_chirp", "down_chirp")

    def schema(self) -> type[BaseModel]:
        """Return the Pydantic model used to validate this emitter's params."""
        return ChirpParams

    def generate(
        self,
        *,
        class_label: str,
        sample_rate: SampleRate,
        duration_s: float,
        f_offset_hz: float,
        rng: torch.Generator,
        device_id: str | None = None,
        params: dict | None = None,
    ) -> Signal:
        cfg = ChirpParams(**(params or {}))
        sign = +1.0 if class_label == "up_chirp" else -1.0

        n = int(round(sample_rate * duration_s))
        t = torch.arange(n, device="cpu") / sample_rate
        chirp_n = int(round(sample_rate * cfg.chirp_duration_s))
        pri_n = int(round(sample_rate * cfg.pulse_repetition_us * 1e-6))

        # Linear chirp pulse train
        iq = torch.zeros((2, n), dtype=torch.float32)
        for start in range(0, n - chirp_n, pri_n):
            t_pulse = torch.arange(chirp_n, device="cpu") / sample_rate
            phase = 2 * torch.pi * (
                f_offset_hz * t_pulse
                + sign * 0.5 * cfg.chirp_rate_hz_per_s * t_pulse ** 2
            )
            iq[0, start:start + chirp_n] = torch.cos(phase)
            iq[1, start:start + chirp_n] = torch.sin(phase)

        bandwidth = cfg.chirp_rate_hz_per_s * cfg.chirp_duration_s
        meta = SignalMetadata(
            signal_id=0,
            family=self.family,
            class_name=class_label,
            class_taxonomy=("radar", "chirp", class_label),
            generator_name=type(self).__name__,
            generator_version="0.1.0",
            sample_rate_hz=float(sample_rate),
            realized_carrier_hz=cfg.carrier_hz,
            bandwidth_hz=bandwidth,
            start_sample=0,
            duration_samples=n,
            snr_db=float("inf"),       # clean emitter output; channel sets SNR
            cfo_hz=0.0,                # TX-impairment transformations update this
            sfo_ppm=0.0,               # sample-rate offset is not applied here
            iq_imbalance_db=0.0,       # TX IQ imbalance is not applied here
            phase_noise_dbc_hz=None,   # TX phase noise is not applied here
            pa_model=None,             # PA nonlinearity is not applied here
            channel_profile="emitter",
            device_id=device_id,
        )
        return Signal(iq=iq, metadata=meta)

Two ways to register

In-tree (decorator)

@register_emitter(family=EmitterFamily.RADAR)
class LinearChirpEmitter(BaseEmitter):
    ...

Importing the module is enough. For first-party emitters in src/rfgen/emitters/<family>/, the package’s __init__.py imports them at startup.

Third-party (entry point)

In your own package’s pyproject.toml:

[project.entry-points."rfgen.emitters"]
chirp_radar = "my_pkg.chirps:LinearChirpEmitter"

The framework discovers it via importlib.metadata.entry_points(). No source changes to rfgen needed.

Wire it into a config

The decorator and class attribute use the broad EmitterFamily enum, such as RADAR, for taxonomy. The entry-point key, such as chirp_radar, is the registry name users select in config and CLI commands.

# configs/emitter_zoo/with_my_chirp.yaml
families:
  - family: chirp_radar
    classes: [up_chirp, down_chirp]
    weight: 1.0
    params:
      carrier_hz: 2.4e9
      chirp_rate_hz_per_s: 1.0e9
      chirp_duration_s: 1.0e-5
      pulse_repetition_us: 100

rfgen generate emitter_zoo=with_my_chirp scene=narrowband run.num_samples=10

Verify

# 1. Confirm the emitter is registered
rfgen list-emitters | grep chirp_radar
# chirp_radar    LinearChirpEmitter    up_chirp, down_chirp

# 2. Inspect its schema
rfgen list-emitters --schema chirp_radar
# {
#   "type": "object",
#   "properties": {
#     "chirp_rate_hz_per_s": {"type": "number", "default": 1e9, "exclusiveMinimum": 0},
#     ...
#   }
# }

# 3. Generate one sample and inspect
rfgen generate emitter_zoo=with_my_chirp scene=narrowband \
    run.num_samples=1 storage.path=./out/chirp-test
rfgen inspect ./out/chirp-test sample --first 1

Contract checklist

Before opening a PR, verify your emitter satisfies every clause:

  • Output shape. iq.shape == (2, int(sample_rate * duration_s)).

  • Dtype. float32; the first axis has length 2 and stores real then imaginary IQ components. BaseEmitter._validate_invariants checks this.

  • Baseband output with f_offset_hz applied. The emitter mixes to f_offset_hz, not the caller. (Some emitters, like LoRa CSS and FMCW, sweep frequency intrinsically; mixing them post-hoc breaks structure.)

  • Determinism. All randomness comes from rng (torch.Generator). Same rng state → identical IQ.

  • snr_db = float("inf") at output. The emitter never adds noise.

  • Self-consistent SignalMetadata. Populate the required metadata fields, keep start_sample within [0, int(sample_rate * duration_s)), and keep bandwidth_hz <= sample_rate. Receiver-relative frequency boxes are derived later by the label layer.

  • Pydantic schema for params. Field types, defaults, validators.

  • Stateless. No mutable state on self. Spark workers spin emitters per shard.

Add tests

Drop a contract test under tests/contract/:

# tests/contract/test_chirp_radar_emitter.py
import pytest
import torch
from rfgen.emitters.base import BaseEmitter
from rfgen.core.registry import get_emitter

@pytest.mark.parametrize("class_label", ["up_chirp", "down_chirp"])
def test_chirp_emitter_contract(class_label):
    emitter = get_emitter("chirp_radar")
    rng = torch.Generator().manual_seed(42)
    signal = emitter.generate(
        class_label=class_label,
        sample_rate=10e6,
        duration_s=0.001,
        f_offset_hz=0.0,
        rng=rng,
    )
    assert signal.iq.shape == (2, 10_000)
    assert signal.iq.dtype == torch.float32
    assert signal.metadata.snr_db == float("inf")
    assert signal.metadata.start_sample == 0
    assert signal.metadata.duration_samples == 10_000

The framework also ships a parametrized contract test (tests/contract/test_base_emitter_contract.py) that runs against every registered emitter; your subclass joins the suite automatically once it’s registered.

See also