Skip to content

Sampled geodesic flow

neureptrace.decoding.geodesic_flow implements dependency-light sampled geodesic-flow features for cross-subject transfer.

The protocol is Category 2 / unlabeled target-adaptive. It fits a PCA basis on source features and a second PCA basis on unlabeled target features, samples orthonormal intermediate bases between them, and represents rows by concatenating projections onto the sampled bases.

No target labels are accepted by the public API.

Typical usage:

from neureptrace.decoding.geodesic_flow import fit_sampled_geodesic_flow_features

result = fit_sampled_geodesic_flow_features(
    source_features=X_source,
    target_adaptation_features=X_target_calib,  # optional; otherwise target_test is transductive
    target_test_features=X_target_test,
    config={"n_components": 16, "n_steps": 5},
)

X_source_gf = result.train_features
X_target_gf = result.test_features

neureptrace.decoding.geodesic_flow

Sampled geodesic-flow features for Category-2 transfer.

GeodesicFlowConfig dataclass

Configuration for sampled geodesic-flow features.

Source code in src/neureptrace/decoding/geodesic_flow.py 
21
22
23
24
25
26
27
28
29
30
31
@dataclass(frozen=True, slots=True)
class GeodesicFlowConfig:
    """Configuration for sampled geodesic-flow features."""

    n_components: int | str | float = DEFAULT_GEODESIC_COMPONENTS
    n_steps: int | str = DEFAULT_GEODESIC_STEPS
    center: bool = True
    scale: bool = False
    include_endpoints: bool = True
    normalize_blocks: bool = True
    epsilon: float = 1e-8

GeodesicFlowBasis dataclass

One sampled orthonormal basis along the source-target path.

Source code in src/neureptrace/decoding/geodesic_flow.py 
34
35
36
37
38
39
@dataclass(frozen=True, slots=True)
class GeodesicFlowBasis:
    """One sampled orthonormal basis along the source-target path."""

    position: float
    basis: np.ndarray

GeodesicFlowResult dataclass

Geodesic-flow train/test features and protocol metadata.

Source code in src/neureptrace/decoding/geodesic_flow.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@dataclass(frozen=True, slots=True)
class GeodesicFlowResult:
    """Geodesic-flow train/test features and protocol metadata."""

    train_features: np.ndarray
    test_features: np.ndarray
    source_components: np.ndarray
    target_components: np.ndarray
    bases: tuple[GeodesicFlowBasis, ...]
    source_mean: np.ndarray
    target_mean: np.ndarray
    source_scale: np.ndarray
    target_scale: np.ndarray
    metadata: dict[str, Any] = field(default_factory=dict)

fit_sampled_geodesic_flow_features(*, source_features, target_test_features, config=None, target_adaptation_features=None)

Fit sampled geodesic-flow features from source and unlabeled target data.

Source code in src/neureptrace/decoding/geodesic_flow.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def fit_sampled_geodesic_flow_features(
    *,
    source_features: Sequence[Sequence[float]] | np.ndarray,
    target_test_features: Sequence[Sequence[float]] | np.ndarray,
    config: GeodesicFlowConfig | Mapping[str, Any] | None = None,
    target_adaptation_features: Sequence[Sequence[float]] | np.ndarray | None = None,
) -> GeodesicFlowResult:
    """Fit sampled geodesic-flow features from source and unlabeled target data."""

    cfg = geodesic_flow_config() if config is None else _coerce_config(config)
    source = _feature_matrix(source_features, name="source_features")
    target_test = _feature_matrix(target_test_features, name="target_test_features")
    if source.shape[1] != target_test.shape[1]:
        raise ValueError(
            "source_features and target_test_features must have the same feature width: "
            f"{source.shape[1]} != {target_test.shape[1]}."
        )
    target_fit = target_test if target_adaptation_features is None else _feature_matrix(target_adaptation_features, name="target_adaptation_features")
    if target_fit.shape[1] != source.shape[1]:
        raise ValueError(
            "target_adaptation_features and source_features must have the same feature width: "
            f"{target_fit.shape[1]} != {source.shape[1]}."
        )

    n_components = _effective_components(cfg.n_components, max_components=_max_components(source, target_fit, center=cfg.center))
    n_steps = _positive_int(cfg.n_steps, name="n_steps")
    source_prepared, source_mean, source_scale = _prepare_domain(source, center=cfg.center, scale=cfg.scale, epsilon=cfg.epsilon)
    target_fit_prepared, target_mean, target_scale = _prepare_domain(target_fit, center=cfg.center, scale=cfg.scale, epsilon=cfg.epsilon)
    target_test_prepared = (target_test - target_mean) / target_scale
    source_components = _pca_components(source_prepared, n_components=n_components)
    target_components = _pca_components(target_fit_prepared, n_components=n_components)
    bases = sample_geodesic_bases(source_components, target_components, n_steps=n_steps, include_endpoints=cfg.include_endpoints)
    train_features = transform_with_geodesic_bases(source_prepared, bases, normalize_blocks=cfg.normalize_blocks)
    test_features = transform_with_geodesic_bases(target_test_prepared, bases, normalize_blocks=cfg.normalize_blocks)

    metadata = _metadata(
        cfg,
        n_source_rows=source.shape[0],
        n_target_fit_rows=target_fit.shape[0],
        n_target_test_rows=target_test.shape[0],
        feature_dim=source.shape[1],
        n_components=n_components,
        n_bases=len(bases),
        target_feature_source=TARGET_FEATURE_SOURCE_TRANSDUCTIVE if target_adaptation_features is None else TARGET_FEATURE_SOURCE_CALIBRATION,
    )
    return GeodesicFlowResult(
        train_features=train_features.astype(np.float32, copy=False),
        test_features=test_features.astype(np.float32, copy=False),
        source_components=source_components.astype(np.float32, copy=False),
        target_components=target_components.astype(np.float32, copy=False),
        bases=bases,
        source_mean=source_mean.astype(float, copy=False),
        target_mean=target_mean.astype(float, copy=False),
        source_scale=source_scale.astype(float, copy=False),
        target_scale=target_scale.astype(float, copy=False),
        metadata=metadata,
    )

geodesic_flow_config(*, n_components=DEFAULT_GEODESIC_COMPONENTS, n_steps=DEFAULT_GEODESIC_STEPS, center=True, scale=False, include_endpoints=True, normalize_blocks=True, epsilon=1e-08)

Normalize public sampled-geodesic options.

Source code in src/neureptrace/decoding/geodesic_flow.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def geodesic_flow_config(
    *,
    n_components: int | str | float = DEFAULT_GEODESIC_COMPONENTS,
    n_steps: int | str = DEFAULT_GEODESIC_STEPS,
    center: bool | str | int = True,
    scale: bool | str | int = False,
    include_endpoints: bool | str | int = True,
    normalize_blocks: bool | str | int = True,
    epsilon: float | str = 1e-8,
) -> GeodesicFlowConfig:
    """Normalize public sampled-geodesic options."""

    return GeodesicFlowConfig(
        n_components=n_components,
        n_steps=_positive_int(n_steps, name="n_steps"),
        center=_boolean_option(center, name="center"),
        scale=_boolean_option(scale, name="scale"),
        include_endpoints=_boolean_option(include_endpoints, name="include_endpoints"),
        normalize_blocks=_boolean_option(normalize_blocks, name="normalize_blocks"),
        epsilon=_positive_float(epsilon, name="epsilon"),
    )

sample_geodesic_bases(source_components, target_components, *, n_steps=DEFAULT_GEODESIC_STEPS, include_endpoints=True)

Sample orthonormal bases between source and target PCA bases.

Source code in src/neureptrace/decoding/geodesic_flow.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def sample_geodesic_bases(
    source_components: Sequence[Sequence[float]] | np.ndarray,
    target_components: Sequence[Sequence[float]] | np.ndarray,
    *,
    n_steps: int | str = DEFAULT_GEODESIC_STEPS,
    include_endpoints: bool | str | int = True,
) -> tuple[GeodesicFlowBasis, ...]:
    """Sample orthonormal bases between source and target PCA bases."""

    source = _component_matrix(source_components, name="source_components")
    target = _component_matrix(target_components, name="target_components")
    if source.shape != target.shape:
        raise ValueError(f"source_components and target_components must have the same shape: {source.shape} != {target.shape}.")
    steps = _positive_int(n_steps, name="n_steps")
    endpoints = _boolean_option(include_endpoints, name="include_endpoints")
    positions = np.linspace(0.0, 1.0, steps if endpoints else steps + 2)
    if not endpoints:
        positions = positions[1:-1]
    bases = []
    for position in positions:
        interpolated = (1.0 - float(position)) * source.T + float(position) * target.T
        q, _r = np.linalg.qr(interpolated)
        basis = _canonicalize_component_signs(q[:, : source.shape[0]].T)
        bases.append(GeodesicFlowBasis(position=float(position), basis=basis.astype(np.float32, copy=False)))
    return tuple(bases)

transform_with_geodesic_bases(features, bases, *, normalize_blocks=True)

Concatenate projections onto sampled geodesic bases.

Source code in src/neureptrace/decoding/geodesic_flow.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def transform_with_geodesic_bases(
    features: Sequence[Sequence[float]] | np.ndarray,
    bases: Sequence[GeodesicFlowBasis],
    *,
    normalize_blocks: bool | str | int = True,
) -> np.ndarray:
    """Concatenate projections onto sampled geodesic bases."""

    matrix = _feature_matrix(features, name="features")
    basis_tuple = tuple(bases)
    if not basis_tuple:
        raise ValueError("At least one geodesic basis is required.")
    normalize = _boolean_option(normalize_blocks, name="normalize_blocks")
    blocks = []
    for basis in basis_tuple:
        basis_matrix = _component_matrix(basis.basis, name="basis")
        if basis_matrix.shape[1] != matrix.shape[1]:
            raise ValueError(f"basis width {basis_matrix.shape[1]} does not match feature width {matrix.shape[1]}.")
        block = matrix @ basis_matrix.T
        if normalize:
            block = block / np.sqrt(len(basis_tuple))
        blocks.append(block)
    return np.hstack(blocks).astype(np.float32, copy=False)