Source feature masking

neureptrace.decoding.source_masking implements strict source-only feature masking augmentation for M/EEG feature matrices.

The protocol is Category 1 / strict source-only. It uses source features, source labels, and optional source-domain identifiers for provenance. It does not accept held-out target data.

Supported masking modes:

• feature: randomly selected feature columns
• block: one contiguous feature block

Supported fill modes:

• feature_mean: source-only column means
• row_mean: the sampled row mean
• zero: zeros

Typical usage:

from neureptrace.decoding.source_masking import augment_source_with_feature_masking

result = augment_source_with_feature_masking(
    X_source,
    y_source,
    source_domains=subject_ids,
    config={
        "synthetic_per_class": 8,
        "mask_fraction": 0.15,
        "mask_mode": "feature",
        "fill_mode": "feature_mean",
    },
)

X_aug = result.features
y_aug = result.labels

neureptrace.decoding.source_masking

Strict source-only feature masking augmentation.

The utilities in this module create masked copies of labeled source feature rows. They are intended as a simple domain-generalization baseline for M/EEG feature matrices: rows keep their source labels while a random feature subset or a contiguous feature block is replaced by source-only fill statistics.

This is a Protocol-1 helper. The public API uses source features and source labels only, plus optional source-domain ids for provenance; held-out target data are not accepted.

SourceFeatureMaskingConfig dataclass

Configuration for strict source-only feature masking.

Source code in src/neureptrace/decoding/source_masking.py

@dataclass(frozen=True, slots=True)
class SourceFeatureMaskingConfig:
    """Configuration for strict source-only feature masking."""

    synthetic_per_class: int = 0
    mask_fraction: float = DEFAULT_MASK_FRACTION
    mask_mode: str = "feature"
    block_size: int | None = None
    fill_mode: str = "feature_mean"
    noise_std: float = 0.0
    preserve_original: bool = True
    random_state: int | None = 13

    @property
    def enabled(self) -> bool:
        """Whether synthetic rows should be generated."""

        return self.synthetic_per_class > 0 and self.mask_fraction > 0.0

enabled property

Whether synthetic rows should be generated.

SourceFeatureMaskingResult dataclass

Augmented source rows and provenance.

Source code in src/neureptrace/decoding/source_masking.py

@dataclass(frozen=True, slots=True)
class SourceFeatureMaskingResult:
    """Augmented source rows and provenance."""

    features: np.ndarray
    labels: np.ndarray
    synthetic_mask: np.ndarray
    content_indices: np.ndarray
    masked_feature_indices: tuple[np.ndarray, ...]
    metadata: dict[str, Any] = field(default_factory=dict)

    @property
    def n_synthetic(self) -> int:
        """Number of generated rows in the returned matrix."""

        return int(np.sum(self.synthetic_mask))

n_synthetic property

Number of generated rows in the returned matrix.

augment_source_with_feature_masking(source_features, source_labels, *, source_domains=None, config=None)

Append masked source-row copies.

Parameters:

Name	Type	Description	Default
source_features	Sequence[Sequence[float]] | ndarray	Source feature matrix with rows as trials/windows and columns as features.	required
source_labels	Sequence[Any] | ndarray	One source label per row. Synthetic rows inherit the sampled row label.	required
source_domains	Sequence[Hashable] | ndarray | None	Optional source-domain ids recorded only for provenance.	None
config	SourceFeatureMaskingConfig | Mapping[str, Any] | None	Masking options. Mappings are normalized through :func:source_feature_masking_config.	None

Returns:

Type	Description
SourceFeatureMaskingResult	Augmented feature rows, labels, synthetic-row mask, sampled content row ids, per-synthetic-row masked feature indices, and protocol metadata.

Source code in src/neureptrace/decoding/source_masking.py

def augment_source_with_feature_masking(
    source_features: Sequence[Sequence[float]] | np.ndarray,
    source_labels: Sequence[Any] | np.ndarray,
    *,
    source_domains: Sequence[Hashable] | np.ndarray | None = None,
    config: SourceFeatureMaskingConfig | Mapping[str, Any] | None = None,
) -> SourceFeatureMaskingResult:
    """Append masked source-row copies.

    Parameters
    ----------
    source_features:
        Source feature matrix with rows as trials/windows and columns as features.
    source_labels:
        One source label per row.  Synthetic rows inherit the sampled row label.
    source_domains:
        Optional source-domain ids recorded only for provenance.
    config:
        Masking options.  Mappings are normalized through
        :func:`source_feature_masking_config`.

    Returns
    -------
    SourceFeatureMaskingResult
        Augmented feature rows, labels, synthetic-row mask, sampled content row ids,
        per-synthetic-row masked feature indices, and protocol metadata.
    """

    cfg = source_feature_masking_config() if config is None else _coerce_config(config)
    features = _feature_matrix(source_features, name="source_features")
    labels = _label_vector(source_labels, expected_length=features.shape[0], name="source_labels")
    domains = _domain_vector(source_domains, expected_length=features.shape[0])
    classes, _class_counts = label_counts(labels)
    domain_ids, _domain_counts = label_counts(domains)
    feature_fill = _feature_fill_values(features, mode=cfg.fill_mode)

    if not cfg.enabled:
        metadata = _metadata(
            cfg,
            n_source_rows=features.shape[0],
            n_synthetic_rows=0,
            n_classes=classes.shape[0],
            n_source_domains=domain_ids.shape[0],
            feature_dim=features.shape[1],
        )
        return SourceFeatureMaskingResult(
            features=features.astype(np.float32, copy=False),
            labels=labels.copy(),
            synthetic_mask=np.zeros(features.shape[0], dtype=bool),
            content_indices=np.empty(0, dtype=int),
            masked_feature_indices=(),
            metadata=metadata,
        )

    rng = np.random.default_rng(cfg.random_state)
    synthetic_rows: list[np.ndarray] = []
    synthetic_labels: list[Any] = []
    content_indices: list[int] = []
    masked_indices: list[np.ndarray] = []
    for class_label in classes.tolist():
        class_indices = np.flatnonzero(label_equal_mask(labels, class_label))
        if class_indices.size == 0:
            continue
        for _ in range(cfg.synthetic_per_class):
            content_index = int(rng.choice(class_indices))
            mask = feature_mask_indices(features.shape[1], mask_fraction=cfg.mask_fraction, mask_mode=cfg.mask_mode, block_size=cfg.block_size, rng=rng)
            row = features[content_index].copy()
            if cfg.fill_mode == "row_mean":
                row_fill = np.full(mask.shape[0], float(np.mean(features[content_index])), dtype=float)
            else:
                row_fill = feature_fill[mask]
            row[mask] = row_fill
            if cfg.noise_std > 0.0:
                row = row + rng.normal(0.0, cfg.noise_std, size=row.shape[0])
            synthetic_rows.append(row)
            synthetic_labels.append(class_label)
            content_indices.append(content_index)
            masked_indices.append(mask.astype(int, copy=False))

    synthetic_features = np.vstack(synthetic_rows).astype(np.float32, copy=False) if synthetic_rows else np.empty((0, features.shape[1]), dtype=np.float32)
    synthetic_labels_array = _object_value_vector(synthetic_labels)
    if cfg.preserve_original:
        output_features = np.vstack([features, synthetic_features]).astype(np.float32, copy=False)
        output_labels = np.concatenate([labels, synthetic_labels_array])
        synthetic_mask = np.concatenate([np.zeros(features.shape[0], dtype=bool), np.ones(synthetic_features.shape[0], dtype=bool)])
    else:
        output_features = synthetic_features
        output_labels = synthetic_labels_array
        synthetic_mask = np.ones(synthetic_features.shape[0], dtype=bool)

    metadata = _metadata(
        cfg,
        n_source_rows=features.shape[0],
        n_synthetic_rows=synthetic_features.shape[0],
        n_classes=classes.shape[0],
        n_source_domains=domain_ids.shape[0],
        feature_dim=features.shape[1],
    )
    return SourceFeatureMaskingResult(
        features=output_features,
        labels=output_labels,
        synthetic_mask=synthetic_mask,
        content_indices=np.asarray(content_indices, dtype=int),
        masked_feature_indices=tuple(masked_indices),
        metadata=metadata,
    )

feature_mask_indices(n_features, *, mask_fraction=DEFAULT_MASK_FRACTION, mask_mode='feature', block_size=None, rng=None)

Return sorted feature indices to replace for one synthetic row.

Source code in src/neureptrace/decoding/source_masking.py

def feature_mask_indices(
    n_features: int | str,
    *,
    mask_fraction: float | str = DEFAULT_MASK_FRACTION,
    mask_mode: str = "feature",
    block_size: int | str | None = None,
    rng: np.random.Generator | None = None,
) -> np.ndarray:
    """Return sorted feature indices to replace for one synthetic row."""

    width = _positive_int(n_features, name="n_features")
    fraction = _unit_interval_float(mask_fraction, name="mask_fraction")
    mode = normalize_mask_mode(mask_mode)
    generator = np.random.default_rng() if rng is None else rng
    n_mask = max(1, int(round(width * fraction))) if fraction > 0.0 else 0
    n_mask = min(n_mask, width)
    if n_mask == 0:
        return np.empty(0, dtype=int)
    if mode == "feature":
        return np.sort(generator.choice(width, size=n_mask, replace=False)).astype(int, copy=False)
    size = n_mask if block_size is None else min(_positive_int(block_size, name="block_size"), width)
    start = int(generator.integers(0, width - size + 1))
    return np.arange(start, start + size, dtype=int)

source_feature_masking_config(*, synthetic_per_class=0, mask_fraction=DEFAULT_MASK_FRACTION, mask_mode='feature', block_size=None, fill_mode='feature_mean', noise_std=0.0, preserve_original=True, random_state=13)

Normalize public feature-masking options.

Source code in src/neureptrace/decoding/source_masking.py

def source_feature_masking_config(
    *,
    synthetic_per_class: int | str = 0,
    mask_fraction: float | str = DEFAULT_MASK_FRACTION,
    mask_mode: str | None = "feature",
    block_size: int | str | None = None,
    fill_mode: str | None = "feature_mean",
    noise_std: float | str = 0.0,
    preserve_original: bool | int | str = True,
    random_state: int | str | None = 13,
) -> SourceFeatureMaskingConfig:
    """Normalize public feature-masking options."""

    return SourceFeatureMaskingConfig(
        synthetic_per_class=_nonnegative_int(synthetic_per_class, name="synthetic_per_class"),
        mask_fraction=_unit_interval_float(mask_fraction, name="mask_fraction"),
        mask_mode=normalize_mask_mode(mask_mode),
        block_size=None if block_size in {None, "", "none", "None"} else _positive_int(block_size, name="block_size"),
        fill_mode=normalize_fill_mode(fill_mode),
        noise_std=_nonnegative_float(noise_std, name="noise_std"),
        preserve_original=_bool_value(preserve_original, name="preserve_original"),
        random_state=None if random_state in {None, "", "none", "None"} else _nonnegative_int(random_state, name="random_state"),
    )

normalize_mask_mode(value)

Normalize mask-mode aliases.

Source code in src/neureptrace/decoding/source_masking.py

def normalize_mask_mode(value: str | None) -> str:
    """Normalize mask-mode aliases."""

    normalized = "feature" if value is None else str(value).strip().lower().replace("-", "_")
    normalized = {"random": "feature", "random_features": "feature", "contiguous": "block", "window": "block"}.get(normalized, normalized)
    if normalized not in MASK_MODES:
        raise ValueError(f"Unknown mask_mode {value!r}. Available modes: {', '.join(MASK_MODES)}.")
    return normalized

normalize_fill_mode(value)

Normalize fill-mode aliases.

Source code in src/neureptrace/decoding/source_masking.py

def normalize_fill_mode(value: str | None) -> str:
    """Normalize fill-mode aliases."""

    normalized = "feature_mean" if value is None else str(value).strip().lower().replace("-", "_")
    normalized = {"mean": "feature_mean", "column_mean": "feature_mean", "trial_mean": "row_mean", "sample_mean": "row_mean"}.get(normalized, normalized)
    if normalized not in FILL_MODES:
        raise ValueError(f"Unknown fill_mode {value!r}. Available modes: {', '.join(FILL_MODES)}.")
    return normalized