Core Types¶

Module: rfgen.core_types

The data types every layer of the framework speaks. Defined as frozen dataclasses (frozen=True, slots=True) so they cross process boundaries (Spark, Ray) without surprise mutation.

The module ships ten frozen dataclasses (Signal, SignalMetadata, SceneMetadata, Record, StoredRecord, BBox, ChunkedSignal, GeometryPose, GeometryAssetRef, GeometryProvenance) plus four TypeAlias declarations (IQ, Spectrogram, SampleRate, CenterHz). All ten dataclasses use frozen=True, slots=True.

Type aliases¶

`IQ`, `Spectrogram`, `SampleRate`, `CenterHz`¶

from typing import TypeAlias
import torch

IQ: TypeAlias = torch.Tensor
"""Complex baseband. Single-RX shape `(2, N)` float32 (channel-first real/imag);
multi-RX shape `(num_rx, 2, N)` float32. Channel 0 is in-phase; channel 1 is
quadrature. Stored as float32 rather than complex64 for autograd
compatibility. See [Reference / IQ Layout Policy](../data/iq-layout-policy.md)
for the full in-memory vs storage contract."""

Spectrogram: TypeAlias = torch.Tensor
"""Shape `(F, T)` float32, log-magnitude unless otherwise specified."""

SampleRate: TypeAlias = float
"""Sample rate in Hz."""

CenterHz: TypeAlias = float
"""Carrier or center frequency in Hz, absolute, for HIL alignment."""

These are TypeAlias declarations (not NewType). For the canonical IQ shape rules, storage conversion boundary, and interop shapes (Zarr, WebDataset, HDF5, SigMF), see Reference / IQ Layout Policy.

`class BBox`¶

@dataclass(frozen=True, slots=True)
class BBox:
    start_sample: int       # inclusive start in the parent record's sample frame
    duration_samples: int   # number of samples this bbox spans
    low_freq_hz: float      # Hz, baseband-relative
    high_freq_hz: float
    class_id: int           # integer class identifier
    emitter_index: int      # index into the parent Record's emitters tuple

Time-frequency rectangle for one signal. Used in detection and segmentation labels. The low_freq_hz and high_freq_hz edges are baseband-relative; the parent record’s per-RX rate sets the absolute axis.

BBox round-trips losslessly to and from a normalized YOLO-style 5-tuple (t_norm, f_norm, dt_norm, df_norm, class_id) for inputs in valid ranges via BBox.to_yolo and BBox.from_yolo. For the on-disk normalization (YOLO / COCO conventions) see Label Schema.

`class GeometryPose`¶

@dataclass(frozen=True, slots=True)
class GeometryPose:
    position_m: tuple[float, float, float]
    orientation_rad: tuple[float, float, float]
    velocity_mps: tuple[float, float, float] = (0.0, 0.0, 0.0)
    frame: str = "scene"

Typed 3D pose in the Sionna/Mitsuba scene frame. orientation_rad is Euler angles; frame must be "scene" for v1 (__post_init__ raises ValueError otherwise). velocity_mps feeds SionnaRT’s Doppler wiring: a node constructed from a pose with nonzero velocity produces genuinely time-varying channel taps when RTSolverConfig.cir_num_time_steps > 1. The Pydantic config-layer mirror is GeometryPoseConfig.

`class GeometryAssetRef`¶

@dataclass(frozen=True, slots=True)
class GeometryAssetRef:
    kind: GeometryAssetKind
    uri: str
    content_hash: str
    version: str | None = None
    entrypoint: str | None = None
    media_type: str | None = None
    content_addressed: bool = True
    metadata: dict[str, object] = field(default_factory=dict)

Content-addressed reference to a scene-geometry-related asset. kind is a GeometryAssetKind closed enum (Sionna built-in scene, Mitsuba XML bundle, material database, antenna pattern, and related kinds). uri must start with file://, gs://, s3://, https://, or sionna://builtin/. content_hash must be sha256: followed by 64 hex characters. For GeometryAssetKind.SIONNA_BUILTIN_SCENE, the uri suffix after sionna://builtin/ names the Sionna built-in scene (e.g. munich), resolved via sionna.rt.scene.<name>. For Mitsuba XML bundles, entrypoint is the worker-local filesystem path SionnaRT passes directly to sionna.rt.load_scene. content_addressed is True when content_hash is a true digest of the asset’s bytes (the normal case for a directly-constructed typed ref) and False when it is only a URI-identity stand-in that cannot detect an in-place edit at a fixed URI (the legacy-URI fallback for remote or unreadable assets). A False ref bypasses SionnaRT’s scene cache and is loaded fresh on every solve.

`class GeometryProvenance`¶

@dataclass(frozen=True, slots=True)
class GeometryProvenance:
    asset_refs: tuple[GeometryAssetRef, ...]
    tx_pose: GeometryPose
    rx_pose: GeometryPose
    tx_array_id: str | None
    rx_array_id: str | None
    material_db_hash: str | None
    antenna_pattern_hash: str | None
    solver_backend: str
    sionna_version: str
    mitsuba_version: str
    drjit_version: str

Geometry provenance recorded on RT-generated component metadata. SionnaRT.apply populates this on every real solve: asset_refs is the resolved scene/material/antenna asset refs; tx_pose/rx_pose are the typed poses actually used; solver_backend is "sionna-rt"; and sionna_version/mitsuba_version/drjit_version are the installed library versions (__post_init__ requires all three non-empty when solver_backend == "sionna-rt"). Lives on SignalMetadata.geometry for single-RX RT scenes and is threaded through SceneMetadata.extras for joint multi-RX RT scenes.

`class SignalMetadata`¶

@dataclass(frozen=True, slots=True)
class SignalMetadata:
    family: str
    class_name: str
    class_taxonomy: tuple[str, ...]
    generator_name: str
    device_id: str | None
    sample_rate_hz: float
    bandwidth_hz: float
    realized_carrier_hz: float
    start_sample: int
    duration_samples: int
    snr_db: float
    extras: dict[str, object] = field(default_factory=dict)
    schema_version: int = 1

Per-signal ground truth. Carried by every Signal: one for each emitter, one for each component of a scene. Populated by the producing layer; immutable after.

Fields¶

Field	Type	Set by	Purpose
`family`	str	emitter	Top-level emitter family, e.g. `"comms"`, `"radar"`
`class_name`	str	emitter	Specific class within the family, e.g. `"qpsk"`. Field name matches TorchSig’s `SignalMetadataObject.class_name` verbatim
`class_taxonomy`	tuple[str, …]	emitter	Hierarchical path, e.g. `("comms", "wifi", "11g")`
`generator_name`	str	emitter	Concrete class name of the producing emitter
`device_id`	str \| None	scene composer	Stable virtual-device identifier for fingerprinting
`sample_rate_hz`	float	emitter	Sample rate the IQ was synthesized at
`bandwidth_hz`	float	emitter	Occupied bandwidth
`realized_carrier_hz`	float	emitter	Absolute carrier frequency in Hz. Used by BaseChannel for Doppler scaling and by BaseRXMixer to compute the per-RX downconversion offset
`start_sample`	int	scene composer	Inclusive start in the emitter’s IQ buffer
`duration_samples`	int	emitter	Number of samples this signal occupies
`snr_db`	float	channel	`+inf` at emitter output; finite after the channel layer applies AWGN
`extras`	dict[str, object]	any	Backend-specific extras. The on-disk schema preserves these
`schema_version`	int	constructor	Storage schema version. Default `1`; bumped on breaking field changes

Per-impairment values live in `extras`¶

Per-device fingerprint values (CFO, IQ imbalance, PA, phase noise) are not inline fields. They are threaded through SignalMetadata.extras["fingerprint_params"] as a dict whose keys are documented in rfgen.channels.protocols.FINGERPRINT_PARAM_KEYS:

("cfo_hz", "iq_imbalance_db", "iq_imbalance_rad",
 "pa_p", "pa_a", "phase_noise_db_per_rt_hz")

The Layer 3 device-fingerprint module produces these values from FingerprintParams (Pydantic v2). tx-impairments and rx-frontend transformations consume them only via these documented keys, so a device-fingerprint field rename is caught by a contract test rather than silently changing the shape scene-composer threads through.

Notes¶

Field names mirror TorchSig where applicable. Hz-suffixed and sample-suffixed variants are used here for unambiguous units; both forms exist in the on-disk schema. See TorchSig Interop.
Carrier is absolute. realized_carrier_hz is the absolute RF carrier (e.g. 2.412e9 for Wi-Fi channel 1). There is no scene-level frequency anchor; per-emitter and per-RX frames meet at the BaseRXMixer step in Group.RX_CAPTURE.

`class SceneMetadata`¶

@dataclass(frozen=True, slots=True)
class SceneMetadata:
    scene_id: str
    duration_s: float
    num_emitters: int
    num_rx: int
    rx_index: int | None
    rx_pose: GeometryPose | None
    rx_antenna_id: str | None
    scene_geometry_hash: str | None
    material_db_hash: str | None
    antenna_pattern_hash: str | None
    geometry_asset_refs: tuple[GeometryAssetRef, ...]
    channel_realization_seed: int | None
    realized_emitter_count: int
    realized_snr_db_stats: dict[str, float]
    realized_class_histogram: dict[str, int]
    realized_cochannel_overlap_rate: float
    realized_spectral_occupancy: float
    extras: dict[str, object] = field(default_factory=dict)
    schema_version: int = 1

Scene-level ground truth, with per-RX context. One per Record. SceneMetadata deliberately does not expose center_freq_hz or bandwidth_hz at the scene level (the no-shared-scene-RF-anchor decision); RF context lives on per-signal SignalMetadata, on the typed rx_pose, and on the resolved geometry-asset refs preserved for RT and Mitsuba-backed scenes.

Fields¶

Field	Type	Purpose
`scene_id`	str	Stable identifier shared across records produced from one scene
`duration_s`	float	Scene duration in seconds
`num_emitters`	int	Realized emitter count
`num_rx`	int	Number of receivers in the parent scene
`rx_index`	int \| None	Index of this RX within the parent scene; `None` for RecordAxis.JOINT
`rx_pose`	`GeometryPose \| None`	Typed RX pose in scene-frame coordinates; populated for single-RX records whenever the resolved receiver carries a pose, `None` for joint records or pose-less receivers
`rx_antenna_id`	str \| None	Antenna pattern identifier within the antenna-pattern blob
`scene_geometry_hash`	str \| None	SHA-256 of the active scene-geometry asset, typically a Mitsuba XML bundle or a compatibility hash for a typed built-in scene ref; `None` when the scene carries no geometry asset provenance
`material_db_hash`	str \| None	SHA-256 of the material database blob
`antenna_pattern_hash`	str \| None	SHA-256 of the antenna pattern blob
`geometry_asset_refs`	`tuple[GeometryAssetRef, ...]`	Resolved typed asset refs active for this scene. Preserved even for URI-only configs via compatibility refs synthesized by the composer.
`channel_realization_seed`	int \| None	Seed reproducing this exact channel realization, given the geometry and material hashes
`realized_emitter_count`	int	Audit field: realized emitter count
`realized_snr_db_stats`	dict[str, float]	Audit field: SNR statistics; required keys `{"min", "max", "mean", "p10", "p50", "p90"}`. Geometry-backed scenes without stable scalar SNR labels store `nan` for each value and mark the status in `extras`.
`realized_class_histogram`	dict[str, int]	Audit field: `class_name -> count` mapping
`realized_cochannel_overlap_rate`	float	Audit field: realized overlap rate in `[0, 1]`
`realized_spectral_occupancy`	float	Audit field: realized spectral occupancy in `[0, 1]`
`extras`	dict[str, object]	Backend-specific extras
`schema_version`	int	Storage schema version. Default `1`; bumped on breaking field changes

The hash fields are content-addressed references into the dataset’s sibling assets/ directory; see Concepts / Records, Receivers, and Assets for the storage layout and reconstruction contract.

`extras["receivers"]` receiver-catalog contract¶

When num_rx > 1, SceneMetadata.extras may carry a receiver catalog under "receivers". The shipped scene composer writes one entry per configured receiver, in the same order as the resolved receiver list:

extras["receivers"] = [
    {
        "rx_id": str,
        "position_m": tuple[float, float, float],
        "orientation": tuple[float, float, float, float],
        "antenna_id": str | None,
    },
    ...
]

Contract notes:

The list index is the stable receiver index. Entry extras["receivers"][i] describes receiver i.
For a per-RX record, recover the originating configured receiver with extras["receivers"][scene.rx_index].
For a joint multi-RX record, scene.rx_index is None; the catalog is still present but no single entry is the “active” receiver.
Single-RX records do not need this catalog; the current shipped composer omits it when num_rx == 1.

The realized_* audit fields are read by name by validation-and-audit; any rename or type change is a schema_version bump.

`class Signal`¶

@dataclass(frozen=True, slots=True)
class Signal:
    iq: torch.Tensor
    metadata: SignalMetadata | SceneMetadata
    component_signals: tuple["Signal", ...] = ()

The universal currency of the channel pipeline. An IQ tensor plus its metadata, optionally with component_signals for nested multi-emitter scenes.

Invariants¶

iq has dtype torch.float32, shape (2, N) for one receiver or (num_rx, 2, N) for joint multi-RX records.
A per-signal Signal carries SignalMetadata; a scene-level Signal carries SceneMetadata.
Mutating any field after construction raises dataclasses.FrozenInstanceError.
component_signals is a tuple[Signal, ...] (recursive); the type annotation matches the actual runtime type.

`class Record`¶

@dataclass(frozen=True, slots=True)
class Record:
    iq: torch.Tensor
    scene: SceneMetadata
    emitters: tuple[SignalMetadata, ...]
    bboxes: tuple[BBox, ...]
    seg_mask: torch.Tensor | None = None
    text: dict[str, object] | None = None
    schema_version: int = 1

One sample produced by the pipeline. Record is what the storage layer persists.

Fields¶

Field	Type	Purpose
`iq`	torch.Tensor	Scene IQ; shape `(2, N)` single-RX, `(num_rx, 2, N)` multi-RX
`scene`	SceneMetadata	Scene-level ground truth
`emitters`	tuple[SignalMetadata, …]	Per-emitter metadata, parallel to `bboxes`. Note: this is `tuple[SignalMetadata, ...]`, NOT `tuple[Signal, ...]`
`bboxes`	tuple[BBox, …]	Time-frequency bounding boxes, one per emitter component
`seg_mask`	torch.Tensor \| None	`(F, T)` int16 segmentation mask; `None` if disabled
`text`	dict \| None	Inference-grounded annotations (caption / qa / reasoning / scene_report); `None` until Phase 2 fills it
`schema_version`	int	Storage schema version. Default `1`; bumped on breaking field changes

Record does not carry a pre-computed spectrogram field. When a labeler needs a spectrogram, it computes one from iq at label time. The Record shape in rfgen.core_types is (iq, scene, emitters, bboxes, seg_mask, text, schema_version).

Notes¶

Two-phase generation. A freshly produced Record from Phase 1 has text=None. The annotator (Phase 2) reads emitter and scene metadata, generates text, and produces a StoredRecord-shaped output via StoredRecord.from_record.
Record IDs are content-hashed by default. See Storage Layout.

`class StoredRecord`¶

@dataclass(frozen=True, slots=True)
class StoredRecord:
    scene: SceneMetadata
    emitters: tuple[SignalMetadata, ...]
    bboxes: tuple[BBox, ...]
    seg_mask: torch.Tensor | None = None
    text: dict[str, object] | None = None
    schema_version: int = 1

    @classmethod
    def from_record(cls, record: Record) -> "StoredRecord": ...

A Record minus IQ. StoredRecord.from_record(record) is the core-type constructor that drops IQ. The annotator pipeline then applies whitelist_filter(record, sample_id=...) to build this same StoredRecord shape with prompt-safe metadata only, including the validated sample id in scene.extras["sample_id"].

StoredRecord lives in rfgen.core_types (rather than in annotators) so Layer 8 (annotators) and later validation or audit consumers can import it from below without a backward-layer typedef edge.

`class ChunkedSignal`¶

@dataclass(frozen=True, slots=True)
class ChunkedSignal:
    num_rx: int
    duration_samples: int
    chunk_samples: int

    def __iter__(self) -> Iterator[tuple[int, torch.Tensor]]: ...
    def materialize(self) -> torch.Tensor: ...

Streaming wrapper for scenes too large to hold in RAM. Produced by the scene composer when the materialized scene IQ would exceed the configured chunk threshold. Iterating yields (chunk_offset, chunk_iq) pairs; each chunk_iq is a torch.float32 tensor of shape (num_rx, 2, chunk_samples) (the Signal.iq layout).

Memory bound¶

The RAM footprint per yielded chunk is chunk_iq.element_size() * chunk_iq.numel() == 4 * num_rx * 2 * chunk_samples bytes. The contract test asserts the tracemalloc.get_traced_memory() peak during one full iteration is bounded above by 2 * (4 * num_rx * 2 * chunk_samples) bytes (one chunk in flight plus the iterator’s local references), regardless of duration_samples.

ChunkedSignal is a frozen dataclass with slots=True; subclassing to override __iter__ is awkward and discouraged. The intended extension pattern is composition (decorator or adapter): a concrete producer wraps a ChunkedSignal instance and delegates field access while replacing the chunk-yielding loop. The default __iter__ yields torch.zeros chunks as a placeholder for the materialization-from-storage iterator that ships with Layer 6 storage.

`iter()`¶

Yields (chunk_offset, chunk_iq) pairs of length chunk_samples (the last chunk may be shorter). Channels consume the iterator and apply impairments per chunk.

`materialize()`¶

Concatenates all chunks into a single torch.Tensor of shape (num_rx, 2, duration_samples). Only safe when the full scene fits in memory; intended for tests and small-scene inspection rather than the streaming path.

Schema versioning¶

SignalMetadata, SceneMetadata, and Record each carry an integer attribute schema_version: int = 1 (literal default 1 at module ship time). StoredRecord carries the same attribute.

Adding a new optional field to any of these dataclasses keeps schema_version at its current value.
Removing a field, changing a field’s type, or renaming a field requires bumping schema_version by +1 and is a breaking change for storage round-trip tests.

Core Types¶

Type aliases¶

IQ, Spectrogram, SampleRate, CenterHz¶

class BBox¶

class GeometryPose¶

class GeometryAssetRef¶

class GeometryProvenance¶

class SignalMetadata¶

Fields¶

Per-impairment values live in extras¶

Notes¶

class SceneMetadata¶

Fields¶

extras["receivers"] receiver-catalog contract¶

class Signal¶

Invariants¶

class Record¶

Fields¶

Notes¶

class StoredRecord¶

class ChunkedSignal¶

Memory bound¶

__iter__()¶

materialize()¶

Schema versioning¶

See Also¶

`IQ`, `Spectrogram`, `SampleRate`, `CenterHz`¶

`class BBox`¶

`class GeometryPose`¶

`class GeometryAssetRef`¶

`class GeometryProvenance`¶

`class SignalMetadata`¶

Per-impairment values live in `extras`¶

`class SceneMetadata`¶

`extras["receivers"]` receiver-catalog contract¶

`class Signal`¶

`class Record`¶

`class StoredRecord`¶

`class ChunkedSignal`¶

`iter()`¶

`materialize()`¶