Tasks

Warning

Pre-implementation. This page describes proposed contracts. Class signatures, parameter types, schema fields, 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 frozen-backbone protocol, readout heads, task registry, and OOD (out-of-distribution) protocols. These are the SUPERB (Speech processing Universal PERformance Benchmark) contracts: one frozen backbone produces representations, lightweight task-specific readout heads consume them, and an OOD protocol slices evaluation into held-out folds. See Signal as a modality for the structural template.

Three roles

Role	Examples
Protocol	FrozenBackbone
ABCs	ReadoutHead, Task, OODProtocol
Concrete classes	AngularRegressionHead, ClassificationHead, BeamPredictionHead, CaptionHead, QAHead, TaskRegistry, LeaveOneEnvironmentOut, LeaveOneUnitOut, LeaveOneBandOut

Class index

Class	Role	Purpose
FrozenBackbone	Protocol	The SUPERB upstream contract; weights frozen
ReadoutHead	ABC	Task-specific head built from a backbone
AngularRegressionHead	Head	Regression head for AoA (angle of arrival)
ClassificationHead	Head	Classification head for detection and identity tasks
BeamPredictionHead	Head	Beam codebook prediction head
CaptionHead	Head	Text generation head for RF (radio-frequency) captioning
QAHead	Head	Question answering head
Task	ABC	Wires a task id to its metric, axis, and head
TaskRegistry	Concrete	Looks up tasks by id and version
OODProtocol	ABC	Fold-construction contract
LeaveOneEnvironmentOut	Protocol	Holds out one channel environment
LeaveOneUnitOut	Protocol	Holds out every capture of one device
LeaveOneBandOut	Protocol	Holds out one frequency band

---

Frozen backbone

class emma.tasks.FrozenBackbone(Protocol)

The SUPERB upstream contract. A pretrained foundation model whose weights are not updated during task evaluation. One backbone is scored across every task; the readout heads carry the task-specific capacity. The protocol is structural (runtime-checkable), so any object with an encode method satisfies it.

class FrozenBackbone(Protocol):
    def encode(self, iq: torch.Tensor) -> torch.Tensor: ...

Methods

encode(iq)

Parameter	Type	Purpose
iq	torch.Tensor	Multi-antenna I/Q (in-phase and quadrature), shape (num_rx, 2, N)

Returns. torch.Tensor, the backbone representation consumed by a ReadoutHead.

Invariants

• Weights are frozen. No gradient flows through the backbone during readout-head training or evaluation.

• The same backbone instance feeds every task, so the score reflects representation quality, not head capacity.

---

Readout heads

class emma.tasks.ReadoutHead(ABC)

The task-specific head contract. Built from a frozen backbone; consumes backbone representations and produces task outputs. Heads are lightweight; the backbone carries the representational load.

class ReadoutHead(ABC):
    @abstractmethod
    def build(self, backbone: FrozenBackbone) -> None: ...

    @abstractmethod
    def forward(self, representations: torch.Tensor) -> torch.Tensor: ...

Abstract methods

Method	Returns	Purpose
build(backbone)	None	Initialize head parameters from the backbone’s output dimension
forward(representations)	torch.Tensor	Map backbone representations to task outputs

Notes

• Subclasses are re-fit per OOD fold because the training scenes differ per holdout.

• A head does not own weights that persist across tasks; each task instantiation builds its own head.

---

class emma.tasks.AngularRegressionHead(ReadoutHead)

Regression head for E_LOC_AOA (angle-of-arrival). Predicts a continuous angle rather than binned classes, because the physically meaningful quantity is an angle and binning hides systematic bias.

class AngularRegressionHead(ReadoutHead):
    def __init__(self, hidden_dim: int = 256) -> None: ...

    def build(self, backbone: FrozenBackbone) -> None: ...
    def forward(self, representations: torch.Tensor) -> torch.Tensor: ...

Parameter	Type	Default	Purpose
hidden_dim	int	256	Hidden layer dimension of the regression MLP

class emma.tasks.ClassificationHead(ReadoutHead)

Classification head for detection, identity, and scene tasks. Produces a class distribution.

class ClassificationHead(ReadoutHead):
    def __init__(self, num_classes: int, hidden_dim: int = 256) -> None: ...

    def build(self, backbone: FrozenBackbone) -> None: ...
    def forward(self, representations: torch.Tensor) -> torch.Tensor: ...

Parameter	Type	Default	Purpose
num_classes	int	required	Number of target classes
hidden_dim	int	256	Hidden layer dimension

class emma.tasks.BeamPredictionHead(ReadoutHead)

Beam codebook prediction head for E_CH_BEAM. Outputs a distribution over the beam codebook.

class BeamPredictionHead(ReadoutHead):
    def __init__(self, codebook_size: int, hidden_dim: int = 256) -> None: ...

    def build(self, backbone: FrozenBackbone) -> None: ...
    def forward(self, representations: torch.Tensor) -> torch.Tensor: ...

Parameter	Type	Default	Purpose
codebook_size	int	required	Number of beams in the codebook
hidden_dim	int	256	Hidden layer dimension

class emma.tasks.CaptionHead(ReadoutHead)

Text generation head for E_S2T_CAP (RF captioning).

class CaptionHead(ReadoutHead):
    def __init__(self, vocab_size: int, hidden_dim: int = 512) -> None: ...

    def build(self, backbone: FrozenBackbone) -> None: ...
    def forward(self, representations: torch.Tensor) -> torch.Tensor: ...

Parameter	Type	Default	Purpose
vocab_size	int	required	Output vocabulary size
hidden_dim	int	512	Hidden layer dimension

class emma.tasks.QAHead(ReadoutHead)

Question answering head for E_S2T_QA. Takes a representation plus a question embedding and produces an answer span or text.

class QAHead(ReadoutHead):
    def __init__(self, vocab_size: int, hidden_dim: int = 512) -> None: ...

    def build(self, backbone: FrozenBackbone) -> None: ...
    def forward(self, representations: torch.Tensor) -> torch.Tensor: ...

Parameter	Type	Default	Purpose
vocab_size	int	required	Output vocabulary size
hidden_dim	int	512	Hidden layer dimension

---

Task abstraction

class emma.tasks.Task(ABC)

Wires a task to its metric, OOD axis, readout head, and domain. One Task subclass exists per TaskID member, instantiated by TaskRegistry.

class Task(ABC):
    id: ClassVar[TaskID]
    pillar: ClassVar[Pillar]
    metric: ClassVar[type[Metric]]
    ood_axis: ClassVar[OODAxis]
    readout_protocol: ClassVar[type[ReadoutHead]]

    @abstractmethod
    def build_head(self, backbone: FrozenBackbone) -> ReadoutHead: ...

    @abstractmethod
    def score(
        self,
        predictions: torch.Tensor,
        references: torch.Tensor,
    ) -> float: ...

Class attributes

Attribute	Type	Purpose
id	TaskID	Canonical task identifier
pillar	Pillar	Top-level task family
metric	type[{ref}Metric ]	Metric class the task is scored with
ood_axis	OODAxis	Primary transfer dimension
readout_protocol	type[{ref}ReadoutHead ]	Readout head ABC the task uses

Abstract methods

build_head(backbone)

Parameter	Type	Purpose
backbone	FrozenBackbone	The frozen backbone to build the head from

Returns. A ReadoutHead instance.

score(predictions, references)

Parameter	Type	Purpose
predictions	torch.Tensor	Model outputs
references	torch.Tensor	Ground-truth labels

Returns. float, the raw metric value.

---

class emma.tasks.TaskRegistry

Looks up tasks by id and version. The registry is the single source of truth for which tasks are available in a release; a task not in the registry cannot be scored.

class TaskRegistry:
    @staticmethod
    def get(task_id: TaskID, version: ReleaseVersion) -> Task: ...

    @staticmethod
    def list_tasks(version: ReleaseVersion = ReleaseVersion.V0_1) -> list[TaskID]: ...

Methods

get(task_id, version)

Parameter	Type	Purpose
task_id	TaskID	Task to look up
version	ReleaseVersion	Release to resolve in

Returns. A Task instance. Raises TaskNotReleased if the task does not ship in the requested version.

list_tasks(version)

Parameter	Type	Default	Purpose
version	ReleaseVersion	V0_1	Release to list tasks for

Returns. list[{ref}TaskID ], every task available in the release.

---

OOD protocols

The fold-construction contracts. Each protocol holds out one value of an OOD axis at a time, scores the model on the held-out value, and averages. See OOD protocol for the evaluation mechanics.

class emma.tasks.OODProtocol(ABC)

The fold-construction interface. Three concrete protocols implement it, one per primary axis the v0.1 scored tasks use. The E_ID_AMC continuity column uses the SNR (signal-to-noise ratio) regime axis but has no dedicated protocol at v0.1 because it is excluded from the aggregate.

class OODProtocol(ABC):
    axis: ClassVar[OODAxis]

    @abstractmethod
    def split(
        self,
        dataset: EMMADataset,
        held_out_value: str,
    ) -> tuple[EMMADataset, EMMADataset]: ...

Class attributes

Attribute	Type	Purpose
axis	OODAxis	The transfer dimension this protocol operates on

Abstract methods

split(dataset, held_out_value)

Parameter	Type	Purpose
dataset	EMMADataset	The full dataset to partition
held_out_value	str	The axis value to hold out (environment id, device id, or band id)

Returns. A tuple (train, eval) of EMMADataset instances.

---

class emma.tasks.LeaveOneEnvironmentOut(OODProtocol)

Holds out one channel environment. Training spans a set of Environment members; evaluation scores the model on the held-out environment. Used by E_LOC_AOA, E_LOC_LOS, and E_ID_DRONE.

class LeaveOneEnvironmentOut(OODProtocol):
    axis: ClassVar[OODAxis] = OODAxis.ENVIRONMENT

    def split(self, dataset: EMMADataset, held_out_value: str) -> tuple[EMMADataset, EMMADataset]: ...

---

class emma.tasks.LeaveOneUnitOut(OODProtocol)

Holds out every capture of one device unit. Isolates device identity from capture-session and receiver-front-end confounds. Used by E_ID_FP.

class LeaveOneUnitOut(OODProtocol):
    axis: ClassVar[OODAxis] = OODAxis.DEVICE

    def split(self, dataset: EMMADataset, held_out_value: str) -> tuple[EMMADataset, EMMADataset]: ...

---

class emma.tasks.LeaveOneBandOut(OODProtocol)

Holds out one frequency band. Layers an unseen-band stress on top of the environment axis. Used as a secondary axis for E_LOC_LOS.

class LeaveOneBandOut(OODProtocol):
    axis: ClassVar[OODAxis] = OODAxis.FREQUENCY_BAND

    def split(self, dataset: EMMADataset, held_out_value: str) -> tuple[EMMADataset, EMMADataset]: ...

See Also

• Task reference: the per-task metric, axis, and head wiring.

• OOD protocol: the leave-one-X-out evaluation mechanics.

• Signal as a modality: the one-frozen-backbone, many-readouts template.

• Metrics: the metric classes that score task predictions.