TorchSig Interop¶

Warning

Pre-implementation. This page describes proposed contracts. Class signatures, parameter types, config field names, and behavior are subject to change before code lands. Once implementation exists, content here will be regenerated from docstrings or sourced from running tests.

The framework wraps TorchSig v2.1.x rather than forking it. This page specifies the converter functions that round-trip between our internal types (Record, SignalMetadata, BBox) and TorchSig’s Signal / SignalMetadataObject.

A pure-TorchSig consumer reads our HDF5 output as if it were a TorchSig WidebandSig53 file; our additional fields (taxonomy, source_label_raw, fingerprint, multi-RX) are readable but optional.

The interop path lives behind rfgen[torchsig,hdf5]. torchsig supplies the wideband HDF5 helpers and hdf5 supplies h5py; if either extra is missing, HDF5Store fails lazily with BackendUnavailableError on first use.

Field-level mapping¶

We adopt TorchSig v2.1.1’s signal_types.SignalMetadataObject field names verbatim where they overlap:

TorchSig field	Our field	Type	Notes
`start_in_samples`	`BBox.abs.start_sample`	int64	Inclusive
`stop_in_samples`	`BBox.abs.stop_sample`	int64	Exclusive (numpy/COCO half-open)
`duration_in_samples`	derived (`stop − start`)	int64
`start` (normalized 0–1, signal-local)	not stored (TorchSig-local)	n/a	We use scene-normalized YOLO `cx, w` instead
`stop` (normalized 0–1, signal-local)	not stored (TorchSig-local)	n/a	Same
`lower_freq` (Hz)	`BBox.abs.low_freq_hz`	float64
`upper_freq` (Hz)	`BBox.abs.high_freq_hz`	float64
`center_freq` (Hz)	derived (`(low + high) / 2`)	float64
`bandwidth` (Hz)	derived (`high − low`)	float64
`sample_rate` (Hz)	`SignalMetadata.sample_rate_hz`	float64
`class_name` (str)	`SignalMetadata.class_name`	str	Field name preserved verbatim

Reference: TorchSig signal_types source.

Converters¶

Two helpers under rfgen.labels.torchsig_interop:

def from_torchsig_signal(
    ts_signal: torchsig.Signal,
    scene_metadata: SceneMeta,
) -> Record:
    """Construct our Record from a TorchSig wideband Signal.

    Round-trip contract: byte-exact for the fields TorchSig exposes.
    Fields we add (taxonomy_path, source_label_raw, ...) are populated
    from ts_signal.extras when available, else default-initialized.
    """


def to_torchsig_signal(
    record: Record,
) -> torchsig.Signal:
    """Construct a TorchSig wideband Signal from our Record.

    Our additional fields (taxonomy, source, fingerprint) survive in
    ts_signal.extras as JSON-serialized values.
    """

`from_torchsig_signal`: implementation spec¶

def from_torchsig_signal(ts_signal, scene_metadata):
    bboxes = []
    emitters = []
    for i, ts_meta in enumerate(ts_signal.metadata):
        # 1. Direct fields
        em = SignalMetadata(
            signal_id=i,
            class_name=ts_meta.class_name,
            family=_infer_family(ts_meta.class_name),    # "comms" | "radar" | ... best-effort
            class_taxonomy=tuple(_taxonomy_for(ts_meta.class_name)),
            generator_name="torchsig",
            generator_version=getattr(ts_meta, "torchsig_version", "unknown"),
            sample_rate_hz=ts_meta.sample_rate,
            emitter_offset_hz=ts_meta.center_freq - scene_metadata.center_freq_hz,  # scene-relative
            bandwidth_hz=ts_meta.bandwidth,
            start_sample=ts_meta.start_in_samples,
            duration_samples=ts_meta.duration_in_samples,
            low_freq_hz=ts_meta.lower_freq - scene_metadata.center_freq_hz,      # scene-relative
            high_freq_hz=ts_meta.upper_freq - scene_metadata.center_freq_hz,
            snr_db=getattr(ts_meta, "snr_db", float("nan")),
            cfo_hz=0.0,                  # TorchSig has no fingerprint concept
            sfo_ppm=0.0,
            iq_imbalance_db=0.0,
            phase_noise_dbc_hz=None,
            pa_model=None,
            channel_profile=getattr(ts_meta, "channel_profile", "unknown"),
            device_id=None,
            extras=dict(getattr(ts_meta, "extras", {})),
        )
        # 2. Optional fields from ts_meta.extras (if our previous round-trip wrote them)
        if "rfgen_taxonomy_path" in em.extras:
            em = replace(em, class_taxonomy=tuple(em.extras.pop("rfgen_taxonomy_path")))
        if "rfgen_source_label_raw" in em.extras:
            ...
        if "rfgen_fingerprint" in em.extras:
            ...
        emitters.append(em)

        # 3. Bbox derivation per Label Schema § Bbox derivation algorithm
        bboxes.append(_derive_bbox(em, scene_metadata, emitter_idx=i))

    return Record(
        iq=_to_torch(ts_signal.iq, dtype=torch.float32),
        spectrogram=None,
        bboxes=tuple(bboxes),
        seg_mask=None,                   # TorchSig doesn't ship segmentation
        emitters=tuple(emitters),
        scene=scene_metadata,
        text=None,
    )


def _infer_family(class_name: str) -> str:
    """Best-effort family inference from a TorchSig class_name."""
    if class_name in TORCHSIG_RADAR_NAMES:
        return "radar"
    if class_name.startswith(("ofdm", "lfm", "chirp")):
        return "comms"
    if class_name in TORCHSIG_COMMS_NAMES:
        return "comms"
    return "unknown"

`to_torchsig_signal`: implementation spec¶

def to_torchsig_signal(record):
    import torchsig
    from torchsig.signals.signal_types import Signal as TsSignal, SignalMetadataObject

    ts_metas = []
    for em, bbox in zip(record.emitters, record.bboxes):
        # 1. Mandatory TorchSig fields
        ts_meta = SignalMetadataObject(
            class_name=em.class_name,
            sample_rate=em.sample_rate_hz,
            center_freq=(bbox.abs.low_freq_hz + bbox.abs.high_freq_hz) / 2,
            bandwidth=bbox.abs.high_freq_hz - bbox.abs.low_freq_hz,
            lower_freq=bbox.abs.low_freq_hz,
            upper_freq=bbox.abs.high_freq_hz,
            start_in_samples=bbox.abs.start_sample,
            stop_in_samples=bbox.abs.stop_sample,
            duration_in_samples=bbox.abs.stop_sample - bbox.abs.start_sample,
            start=bbox.abs.start_sample / record.scene.duration_samples,
            stop=bbox.abs.stop_sample / record.scene.duration_samples,
        )

        # 2. Snr in TorchSig's extras
        ts_meta.extras = {
            "snr_db": em.snr_db,
            "channel_profile": em.channel_profile,
        }

        # 3. Our additional fields, prefixed to mark provenance
        ts_meta.extras.update({
            "rfgen_taxonomy_path": list(em.class_taxonomy),
            "rfgen_source_label_raw": em.extras.get("source_label_raw"),
            "rfgen_source_dataset": em.extras.get("source_dataset"),
            "rfgen_source": em.extras.get("source", "synthetic"),
            "rfgen_device_fingerprint_id": em.device_id,
            "rfgen_fingerprint": {
                "cfo_hz": em.cfo_hz,
                "sfo_ppm": em.sfo_ppm,
                "iq_imbalance_db": em.iq_imbalance_db,
                "phase_noise_dbc_hz": em.phase_noise_dbc_hz,
                "pa_model": em.pa_model,
            },
            "rfgen_aoa_deg": em.extras.get("aoa_deg"),
            "rfgen_rx_snr_db": em.extras.get("rx_snr_db"),
            "rfgen_extras": em.extras,
        })
        ts_metas.append(ts_meta)

    # IQ: convert (2, N) float32 → complex64
    iq_complex = record.iq[0] + 1j * record.iq[1]
    return TsSignal(iq=iq_complex.numpy().astype(np.complex64), metadata=ts_metas)

Round-trip contract¶

# Byte-exact for IQ:
record = ...
ts = to_torchsig_signal(record)
record_back = from_torchsig_signal(ts, record.scene)
assert torch.equal(record.iq, record_back.iq)            # bit-identical

# Bbox absolute fields round-trip exactly:
for bb_a, bb_b in zip(record.bboxes, record_back.bboxes):
    assert bb_a.abs == bb_b.abs                          # exact equality on int64 / float64

# Per-emitter floats round-trip to within complex64 precision:
for em_a, em_b in zip(record.emitters, record_back.emitters):
    assert math.isclose(em_a.snr_db, em_b.snr_db, abs_tol=1e-6)
    assert em_a.class_taxonomy == em_b.class_taxonomy      # via extras

# Segmentation mask is NOT round-tripped (TorchSig has no segmentation)
assert record_back.seg_mask is None

What survives the round-trip¶

Layer	Round-trip status
IQ tensor	Byte-exact (`complex64`)
Bbox absolute coordinates	Exact (`int64` samples, `float64` Hz)
Class name	Exact string
YOLO normalized coordinates	Re-derived on each direction; no drift since they’re computed from absolute
Per-emitter SNR	Within `complex64` float precision
Channel profile name	Exact string
Hierarchical taxonomy_path	Survives in `extras` (our addition)
Source dataset / source label raw	Survives in `extras`
Fingerprint params	Survives in `extras` (full nested dict)
AoA, multi-RX SNR	Survives in `extras`
Segmentation mask	LOST: TorchSig writes only bbox
Scene-level metadata	LOST except via our HDF5 attrs sidecar
Text annotations (Phase 2)	LOST: TorchSig is single-modal

What TorchSig does that we don’t reuse¶

BBoxLabeler (TorchSig’s wideband bbox writer). Same math; we re-implement because we want our STFT params, our taxonomy mapping, and our consistency checks to be local and testable. The TorchSig version is a fine reference; ours is canonical.
ChannelPipeline (TorchSig’s wideband composer). Replaced by our DefaultSceneComposer per Scene Composition Algorithm. Limitations covered in Concepts / Scenes § “Why we don’t use TorchSig’s wideband composer”.

Integration test¶

# tests/integration/test_torchsig_roundtrip.py
def test_hdf5_roundtrip_byte_exact(tmp_path):
    """Generate via rfgen → read with TorchSig loader → round-trip back."""
    from torchsig.utils.dataset import StaticTorchSigDataset
    from rfgen.storage.hdf5_store import HDF5Store

    rfgen.generate("+preset=wideband_baseline_xs", output=tmp_path)

    ts_ds = StaticTorchSigDataset(root=str(tmp_path / "hdf5"))
    rfgen_ds = HDF5Store().open(str(tmp_path / "hdf5"), mode="r")

    for ts_sample, rfgen_record in zip(ts_ds, rfgen_ds):
        # IQ bit-identical
        assert np.array_equal(ts_sample.iq, rfgen_record.iq.numpy())
        # Bbox fields TorchSig understands round-trip
        for ts_bb, rg_bb in zip(ts_sample.metadata, rfgen_record.bboxes):
            assert ts_bb.lower_freq == rg_bb.abs.low_freq_hz
            assert ts_bb.start_in_samples == rg_bb.abs.start_sample
            assert ts_bb.class_name == rg_bb.class_name

This test runs in the golden marker bucket. See Test Execution.

Versioning¶

The converter functions are pinned to TorchSig v2.1.x. v2.2 (when released) and v3.x are likely to break:

v2.0 → v2.1 (Feb 2026): DatasetMetadata removed; replaced by HierarchicalMetadataObject. SignalBuilder renamed to BaseSignalGenerator. Our converters target the v2.1 form.
v3.x (future): expected to break further. We’ll bump our converter major version when v3 stabilizes.

The CI matrix in Build and CI tests against TorchSig v2.1.1 only. Adding v2.1.2 / v2.1.3 to the matrix is automatic; v2.2 requires a converter audit.