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 register_method(
    method_id: str,
    import_path: str,
    *,
    status: Literal["implemented", "planned"] = "implemented",
) -> None:
    """Register a method by id and a lazy import string."""
    if not method_id or not isinstance(method_id, str):
        raise ValueError("method_id must be a non-empty string")
    if ":" not in import_path:
        raise ValueError("import_path must be of the form 'pkg.module:ClassName'")
    existing = _REGISTRY.get(method_id)
    if existing is not None and existing.import_path != import_path:
        raise ValueError(
            f"method_id {method_id!r} already registered with import_path={existing.import_path!r}"
        )
    if status not in {"implemented", "planned"}:
        raise ValueError("status must be 'implemented' or 'planned'")
    _REGISTRY[method_id] = MethodRef(method_id=method_id, import_path=import_path, status=status)

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 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")
    _check_y(y_l, n=int(X_l.shape[0]))

    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(X_l.shape[1])
    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(X_u_w.shape[0]) != int(X_u_s.shape[0]):
        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}"
                )

    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=data.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