23. Views API¶

def generate_views(
    dataset: LoadedDataset,
    *,
    plan: ViewsPlan,
    seed: int = 0,
    cache: bool = True,
    fit_indices: np.ndarray | None = None,
) -> ViewsResult:
    """Generate multiple feature views from a dataset.

    Parameters
    ----------
    dataset:
        Input dataset from :mod:`modssc.data_loader` (train/test splits).
    plan:
        ViewsPlan describing how to create each view.
    seed:
        Global seed controlling stochastic view operations (e.g. random feature split).
    cache:
        Passed through to :func:`modssc.preprocess.preprocess` when preprocessing is used.
    fit_indices:
        Indices (relative to the *train* split) to use when fitting preprocessing steps
        (e.g. PCA). Defaults to ``np.arange(len(train))``.

    Returns
    -------
    ViewsResult
        Each view is returned as a `LoadedDataset` where `.train.X` and `.test.X` are view-specific
        feature matrices, while labels / edges / masks are preserved.
    """

    start = perf_counter()
    plan.validate()

    dataset_fp = None
    if hasattr(dataset, "meta") and isinstance(dataset.meta, dict):
        dataset_fp = dataset.meta.get("dataset_fingerprint")
    logger.info(
        "Views start: views=%s seed=%s cache=%s dataset_fp=%s",
        [v.name for v in plan.views],
        seed,
        bool(cache),
        dataset_fp,
    )

    train_y = _as_numpy(dataset.train.y)
    n_train = int(train_y.shape[0])
    if fit_indices is None:
        fit_indices = np.arange(n_train, dtype=np.int64)

    views: dict[str, LoadedDataset] = {}
    columns: dict[str, np.ndarray] = {}
    n_features_map: dict[str, int] = {}

    for view in plan.views:
        view_start = perf_counter()
        # 1) Optional preprocessing (cached, deterministic)
        ds = dataset
        if view.preprocess is not None:
            res = run_preprocess(
                ds, plan=view.preprocess, seed=int(seed), fit_indices=fit_indices, cache=bool(cache)
            )
            ds = res.dataset

        X_train = _as_numpy(ds.train.X)
        X_test = _as_numpy(ds.test.X) if ds.test is not None else None

        if X_train.ndim != 2:
            raise ViewsValidationError(
                f"View {view.name!r}: expected train.X to be 2D, got shape={X_train.shape}"
            )
        if X_test is not None and X_test.ndim != 2:
            raise ViewsValidationError(
                f"View {view.name!r}: expected test.X to be 2D, got shape={X_test.shape}"
            )

        n_features = int(X_train.shape[1])
        cols = _resolve_columns(
            spec=view.columns,
            n_features=n_features,
            seed=int(seed),
            view_name=str(view.name),
            resolved=columns,
            n_features_map=n_features_map,
        )
        n_features_map[str(view.name)] = n_features
        columns[str(view.name)] = cols

        X_train_v = X_train[:, cols]
        X_test_v = X_test[:, cols] if X_test is not None else None

        # 2) Preserve y/edges/masks (do NOT copy large arrays)
        train_split = Split(
            X=X_train_v,
            y=ds.train.y,
            edges=ds.train.edges,
            masks=ds.train.masks,
        )
        test_split = (
            Split(X=X_test_v, y=ds.test.y, edges=ds.test.edges, masks=ds.test.masks)
            if ds.test is not None
            else None
        )

        # 3) Meta
        meta: dict[str, Any] = dict(ds.meta) if isinstance(ds.meta, dict) else {}
        meta.setdefault("views", {})
        meta["views"][str(view.name)] = {
            "columns": cols.tolist(),
            "columns_mode": (view.columns.mode if view.columns is not None else "all"),
            "preprocess": (asdict(view.preprocess) if view.preprocess is not None else None),
        }
        if view.meta:
            # view-level metadata override/additions
            meta.setdefault("view_meta", {})
            meta["view_meta"][str(view.name)] = dict(view.meta)

        views[str(view.name)] = LoadedDataset(train=train_split, test=test_split, meta=meta)
        logger.debug(
            "View built: name=%s train_shape=%s test_shape=%s duration_s=%.3f",
            view.name,
            _shape_of(X_train_v),
            _shape_of(X_test_v),
            perf_counter() - view_start,
        )

    result = ViewsResult(
        views=views,
        columns=columns,
        seed=int(seed),
        plan=plan,
        meta={"n_views": len(views)},
    )
    logger.info(
        "Views done: count=%s duration_s=%.3f",
        len(views),
        perf_counter() - start,
    )
    return result

def two_view_random_feature_split(
    *,
    preprocess: PreprocessPlan | None = None,
    fraction: float = 0.5,
    seed_offset: int = 0,
    name_a: str = "view_a",
    name_b: str = "view_b",
) -> ViewsPlan:
    """Convenience helper for classic 2-view feature split.

    The first view picks a random subset of columns, the second view is its complement.
    """

    a = ViewSpec(
        name=name_a,
        preprocess=preprocess,
        columns=ColumnSelectSpec(
            mode="random", fraction=float(fraction), seed_offset=int(seed_offset)
        ),
        meta={"role": "primary"},
    )
    b = ViewSpec(
        name=name_b,
        preprocess=preprocess,
        columns=ColumnSelectSpec(mode="complement", complement_of=name_a),
        meta={"role": "complement"},
    )
    plan = ViewsPlan(views=(a, b))
    plan.validate()
    return plan