26. Inductive API¶

def make_numpy_rng(seed: int) -> np.random.Generator:
    """Create a numpy Generator seeded with the given seed."""
    return np.random.default_rng(int(seed))

def seed_everything(seed: int, *, deterministic: bool = True) -> None:
    """Best-effort deterministic seeding across random, numpy, and torch (if available)."""
    seed_i = int(seed)
    random.seed(seed_i)
    np.random.seed(seed_i)

    try:
        torch = optional_import("torch", extra="inductive-torch")
    except OptionalDependencyError:
        return

    torch.manual_seed(seed_i)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed_i)

    if deterministic:
        if torch.cuda.is_available():
            os.environ.setdefault("CUBLAS_WORKSPACE_CONFIG", ":4096:8")
        if hasattr(torch, "use_deterministic_algorithms"):
            with suppress(Exception):
                torch.use_deterministic_algorithms(True)
        if hasattr(torch.backends, "cudnn"):
            torch.backends.cudnn.deterministic = True
            torch.backends.cudnn.benchmark = False

def to_numpy_dataset(data: InductiveDatasetLike) -> NumpyDataset:
    """Validate and wrap an inductive dataset backed by numpy arrays."""
    validate_inductive_dataset(data)

    X_l = _require_numpy(data.X_l, name="X_l")
    y_l = _require_numpy(data.y_l, name="y_l")
    X_u = _require_numpy(data.X_u, name="X_u") if data.X_u is not None else None
    X_u_w = _require_numpy(data.X_u_w, name="X_u_w") if data.X_u_w is not None else None
    X_u_s = _require_numpy(data.X_u_s, name="X_u_s") if data.X_u_s is not None else None
    views = _require_numpy_views(data.views)

    return NumpyDataset(
        X_l=X_l,
        y_l=y_l,
        X_u=X_u,
        X_u_w=X_u_w,
        X_u_s=X_u_s,
        views=views,
        meta=data.meta,
    )

def to_torch_dataset(
    data: InductiveDatasetLike,
    *,
    device: DeviceSpec | None = None,
    require_same_device: bool = True,
) -> TorchDataset:
    """Validate and wrap an inductive dataset backed by torch tensors.

    If device.device == "auto", only device consistency is enforced.
    """
    if data is None:
        raise InductiveValidationError("data must not be None.")
    X_l = _require_tensor(data.X_l, name="X_l")
    y_l = _require_tensor(data.y_l, name="y_l")
    _check_2d(X_l, name="X_l")

    def get_len(x):
        if isinstance(x, dict) and "x" in x:
            return x["x"].shape[0]
        return x.shape[0]

    def get_features(x):
        if isinstance(x, dict) and "x" in x:
            return x["x"].shape[1]
        return x.shape[1]

    _check_y(y_l, n=int(get_len(X_l)))

    X_u = _require_tensor(data.X_u, name="X_u") if data.X_u is not None else None
    X_u_w = _require_tensor(data.X_u_w, name="X_u_w") if data.X_u_w is not None else None
    X_u_s = _require_tensor(data.X_u_s, name="X_u_s") if data.X_u_s is not None else None

    n_features = int(get_features(X_l))
    if X_u is not None:
        _check_2d(X_u, name="X_u")
        _check_feature_dim(X_u, n_features=n_features, name="X_u")
    if X_u_w is not None:
        _check_2d(X_u_w, name="X_u_w")
        _check_feature_dim(X_u_w, n_features=n_features, name="X_u_w")
    if X_u_s is not None:
        _check_2d(X_u_s, name="X_u_s")
        _check_feature_dim(X_u_s, n_features=n_features, name="X_u_s")
    if X_u_w is not None and X_u_s is not None and int(get_len(X_u_w)) != int(get_len(X_u_s)):
        raise InductiveValidationError("X_u_w and X_u_s must have the same number of rows")

    views = _require_views(data.views)

    tensors = [X_l, y_l, X_u, X_u_w, X_u_s]
    if views:
        tensors.extend(views.values())

    if require_same_device:
        _check_same_device(tensors)

    if device is not None and device.device != "auto":
        expected = torch_backend.resolve_device(device)
        for t in tensors:
            if t is None:
                continue
            if t.device != expected:
                raise InductiveValidationError(
                    f"Tensor device mismatch: expected {expected}, got {t.device}"
                )

    meta = data.meta
    if isinstance(meta, Mapping):
        meta = dict(meta)
        torch = _torch()

        def _device_of(obj: Any) -> Any | None:
            if obj is None:
                return None
            if isinstance(obj, dict) and "x" in obj:
                return obj["x"].device
            return getattr(obj, "device", None)

        idx_device = _device_of(X_u) or _device_of(X_u_w) or _device_of(X_u_s) or _device_of(X_l)
        if idx_device is not None:
            for key in ("idx_u", "unlabeled_idx", "unlabeled_indices"):
                if key not in meta:
                    continue
                idx_val = meta[key]
                if isinstance(idx_val, torch.Tensor):
                    if idx_val.dtype != torch.int64 or idx_val.device != idx_device:
                        meta[key] = idx_val.to(device=idx_device, dtype=torch.int64)
                else:
                    meta[key] = torch.as_tensor(
                        np.asarray(idx_val), device=idx_device, dtype=torch.int64
                    )

    return TorchDataset(
        X_l=X_l,
        y_l=y_l,
        X_u=X_u,
        X_u_w=X_u_w,
        X_u_s=X_u_s,
        views=views,
        meta=meta,
    )

def validate_inductive_dataset(data: InductiveDatasetLike) -> None:
    """Validate the minimal invariants needed by inductive algorithms."""
    if data is None:
        raise InductiveValidationError("data must not be None")

    X_l = _require_2d(data.X_l, name="X_l")
    _require_y(data.y_l, n=int(X_l.shape[0]))

    n_features = int(X_l.shape[1])

    if data.X_u is not None:
        X_u = _require_2d(data.X_u, name="X_u")
        if int(X_u.shape[1]) != n_features:
            raise InductiveValidationError("X_u must have the same feature dimension as X_l")

    if data.X_u_w is not None:
        X_u_w = _require_2d(data.X_u_w, name="X_u_w")
        if int(X_u_w.shape[1]) != n_features:
            raise InductiveValidationError("X_u_w must have the same feature dimension as X_l")

    if data.X_u_s is not None:
        X_u_s = _require_2d(data.X_u_s, name="X_u_s")
        if int(X_u_s.shape[1]) != n_features:
            raise InductiveValidationError("X_u_s must have the same feature dimension as X_l")

    if data.X_u_w is not None and data.X_u_s is not None:
        X_u_w = _as_numpy(data.X_u_w)
        X_u_s = _as_numpy(data.X_u_s)
        if X_u_w.shape[0] != X_u_s.shape[0]:
            raise InductiveValidationError("X_u_w and X_u_s must have the same number of rows")

    _validate_views(data.views)
    _validate_meta(data.meta)

def validate_torch_model_bundle(bundle: TorchModelBundle) -> TorchModelBundle:
    """Validate a torch model bundle (model + optimizer)."""
    torch = optional_import("torch", extra="inductive-torch")
    if not isinstance(bundle, TorchModelBundle):
        raise InductiveValidationError("model_bundle must be a TorchModelBundle.")
    if not isinstance(bundle.model, torch.nn.Module):
        raise InductiveValidationError("model_bundle.model must be a torch.nn.Module.")
    if not isinstance(bundle.optimizer, torch.optim.Optimizer):
        raise InductiveValidationError("model_bundle.optimizer must be a torch.optim.Optimizer.")

    params = [p for p in bundle.model.parameters() if p.requires_grad]
    if not params:
        raise InductiveValidationError("model_bundle.model must have trainable parameters.")

    model_ids = {id(p) for p in bundle.model.parameters()}
    for group in bundle.optimizer.param_groups:
        for p in group.get("params", []):
            if id(p) not in model_ids:
                raise InductiveValidationError(
                    "model_bundle.optimizer params must come from model parameters."
                )

    if bundle.ema_model is not None and not isinstance(bundle.ema_model, torch.nn.Module):
        raise InductiveValidationError("model_bundle.ema_model must be a torch.nn.Module.")

    return bundle