3. Build a Multi-Output Data Pipeline with Label Masking

Why a mask is part of the data contract, not the loss

A restraint response dataset is rarely fully labeled. One sample may carry a measured belt_force and lap_force but no shoulder_force; another may have shoulder_force only. If the loader naively fills missing entries with 0.0, the downstream loss will treat that zero as a real measurement and push the corresponding head toward a physically wrong target. The fix is to make validity an explicit, first-class part of every batch: each sample emits features, a multi-output target tensor, and a per-output validity mask of identical shape to the target. The mask is produced by the data pipeline, travels with the batch, and is consumed by the loss — never inferred later from target values, because a legitimate measured value can itself be zero.

Target schema and mask semantics

Assume the modeling contract fixes four restraint response channels, each a scalar regression output in newtons (N):

Two rules keep the pipeline honest. First, the target tensor always has shape (batch, 4) and the mask always has shape (batch, 4); missing entries in the target are set to 0.0 but are neutralized by a mask value of 0. Second, a mask value of 1 means "this number is a trustworthy regression target," so any sample whose label is out of physical range (for example a negative force magnitude) must be masked to 0 rather than silently clipped. This makes the mask a data-quality gate as well as a presence flag.

Pipeline architecture

The diagram below shows how a raw record flows through validation, target assembly, and masking, and how the three tensors stay aligned by channel index all the way to the loss.

Implementing the loader

The loader below is self-contained: it defines the channel order, validates each record, builds the target and mask arrays, and returns a batch dictionary. It uses only the standard library plus NumPy so it can be dropped into any training loop.

import numpy as np

CHANNELS = ["belt_force", "lap_force", "shoulder_force", "torso_force"]
N_OUT = len(CHANNELS)
FEATURE_DIM = 10

def build_sample(record):
    """record: dict with 'features' (list[float]) and optional channel keys."""
    x = np.asarray(record["features"], dtype=np.float32)
    if x.shape != (FEATURE_DIM,):
        raise ValueError(f"expected {FEATURE_DIM} features, got {x.shape}")

    y = np.zeros(N_OUT, dtype=np.float32)
    m = np.zeros(N_OUT, dtype=np.float32)

    for i, ch in enumerate(CHANNELS):
        val = record.get(ch, None)
        if val is None:
            continue                      # missing -> y=0.0, m=0
        val = float(val)
        if not np.isfinite(val) or val < 0.0:
            continue                      # invalid -> masked out
        y[i] = val
        m[i] = 1.0
    return x, y, m

def collate(records):
    xs, ys, ms = zip(*(build_sample(r) for r in records))
    return {
        "features": np.stack(xs),         # (B, 10)
        "targets":  np.stack(ys),         # (B, 4)
        "mask":     np.stack(ms),         # (B, 4)
    }

class MaskedRestraintLoader:
    def __init__(self, records, batch_size=8, shuffle=True, seed=0):
        self.records = list(records)
        self.batch_size = batch_size
        self.shuffle = shuffle
        self.rng = np.random.default_rng(seed)

    def __len__(self):
        return (len(self.records) + self.batch_size - 1) // self.batch_size

    def __iter__(self):
        idx = np.arange(len(self.records))
        if self.shuffle:
            self.rng.shuffle(idx)
        for start in range(0, len(idx), self.batch_size):
            chunk = idx[start:start + self.batch_size]
            yield collate([self.records[i] for i in chunk])

Two design choices matter here. The mask is built from the same loop that fills the target, so the two arrays can never drift out of alignment. And invalid values are dropped rather than clipped: a negative force reading is a data error, and masking it prevents a corrupted label from contributing to the loss while still letting the sample contribute its valid channels.

Verifying the batch contract

Run the loader on a small synthetic set that deliberately mixes complete and partial records, then assert the invariants before training ever starts:

records = [
    {"features": [0.1]*10, "belt_force": 412.0, "lap_force": 205.0,
     "shoulder_force": 88.5, "torso_force": 150.0},
    {"features": [0.2]*10, "belt_force": 390.0, "shoulder_force": 91.0},
    {"features": [0.3]*10, "lap_force": 210.0, "torso_force": -5.0},  # invalid torso
]

batch = next(iter(MaskedRestraintLoader(records, batch_size=3, shuffle=False)))

assert batch["features"].shape == (3, 10)
assert batch["targets"].shape  == (3, 4)
assert batch["mask"].shape     == (3, 4)
assert np.all((batch["mask"] == 0) | (batch["mask"] == 1))
# masked entries must be zero in the target
assert np.all(batch["targets"][batch["mask"] == 0] == 0.0)
# the invalid torso label must be masked, not clipped
assert batch["mask"][2, 3] == 0.0
print("mask coverage per channel:", batch["mask"].mean(axis=0))

Expected observable output for this input: shapes (3,10), (3,4), (3,4); the mask is strictly binary; every masked target entry is exactly 0.0; and the per-channel coverage prints [0.667 0.667 0.667 0.333], reflecting that torso_force is valid in only one of three samples. That coverage vector is the quickest sanity check that masking is behaving as intended.

Failure symptoms and fixes

• Loss decreases then plateaus at a biased value: missing labels were filled with 0.0 and no mask was applied. Check that the loss consumes batch["mask"].
• Shape mismatch at the loss: the mask was built per-sample instead of per-batch, or a channel was dropped. Assert targets.shape == mask.shape in the collate function.
• One head never learns: its channel coverage is near zero in the sampled batches. Inspect the per-channel coverage vector and consider stratified sampling or a coverage-aware split.
• Negative or NaN targets leak through: validation was skipped. The np.isfinite and range checks in build_sample are the guard; keep them before masking.