Datasets
Dataset helpers download files into a local data/ directory and return
SILVA-ready tensors. Dataset files are not committed to the repository.
The engine itself is independent of any specific dataset. A dataset becomes a
SILVA problem when it is converted into the tensor contract below.
Tensor Contract
| Tensor |
Shape |
Used by |
Meaning |
x |
(entities, features) or categorical (entities,) |
all graph/set models |
Entity, node, sample, atom, pixel, channel features, or categorical atom IDs |
edge_index |
(2, edges) |
local graph branches |
Source row then destination row |
edge_attr |
(edges, edge_features) or (edges,) |
edge-aware branches |
Bond, relation, distance, or edge category features |
batch |
(entities,) |
global branches and graph pooling |
Graph/set id for each entity |
y |
task-specific |
training code |
Node, graph, image, or regression targets |
GraphTensorBatch stores these fields and exposes model_kwargs() for direct
model calls.
from silva_networks import GraphTensorBatch, SILVAGraphNetwork
packed = GraphTensorBatch(x=x, edge_index=edge_index, batch=batch, y=y)
logits = model(**packed.model_kwargs())
Standardization
For a raw feature matrix \(X\in\mathbb R^{n\times d}\), preprocessing uses
column statistics:
\[
\mu_j=\frac1n\sum_{i=1}^n X_{ij},
\qquad
\sigma_j=
\sqrt{\frac1n\sum_{i=1}^n(X_{ij}-\mu_j)^2}.
\]
The standardized feature is
\[
\tilde X_{ij}=\frac{X_{ij}-\mu_j}{\max(\sigma_j,\varepsilon)}.
\]
Missing numeric features are mean-imputed before standardization. For a real
experiment, fit statistics on the training split only and reuse them:
from silva_networks import fit_feature_standardization
stats = fit_feature_standardization(x_train)
x_train = stats.transform(x_train)
x_val = stats.transform(x_val)
x_test = stats.transform(x_test)
Use fit_tensor_standardization for PyTorch tensors. The convenience functions
standardize_features and standardize_tensor fit and transform the same
input, so they are appropriate before splitting or for exploratory checks, not
for leakage-free held-out evaluation.
Download and Load
from silva_networks import available_datasets, load_tabular_dataset
print(available_datasets())
dataset = load_tabular_dataset("iris", root="data", download=True)
x, y = dataset.tensors(device="cpu")
Raw features are the default. Split first, then fit
fit_feature_standardization on the training rows. normalize=True remains a
convenience for exploratory whole-table analysis.
The registry includes compact public tabular cases such as Iris, Wine, WDBC,
Seeds, Abalone, Yeast, Airfoil Self-Noise, Wine Quality, Glass, Banknote
Authentication, Forest Fires, and Cleveland Heart Disease.
Vision datasets are available through the optional vision extra:
from silva_networks import available_torchvision_datasets, load_torchvision_dataset
print(available_torchvision_datasets())
cifar = load_torchvision_dataset("CIFAR10", root="data", download=True)
The image adapters accept channel_last=True or False when an NHWC/NCHW
layout is ambiguous, such as a 3- or 4-pixel spatial dimension.
The supported TorchVision names are MNIST, FashionMNIST, KMNIST,
EMNIST, CIFAR10, CIFAR100, and SVHN.
Equation-Checked Teaching Problems
The recent equilibrium families also include deterministic generated problems
that can be used without a download:
from silva_networks import (
make_affine_homotopy_dataset,
make_graph_transport_dataset,
make_periodic_elliptic_dataset,
make_variable_measure_dataset,
)
field = make_periodic_elliptic_dataset(samples=8, height=16, width=16, seed=7)
graph = make_graph_transport_dataset(graphs=4, nodes_per_graph=12, seed=7)
homotopy = make_affine_homotopy_dataset(samples=16, features=4, seed=7)
measure = make_variable_measure_dataset(samples=8, max_points=20, seed=7)
Each return type carries the tensors required by its SILVA family and an exact
equation or empirical-moment check. These are compact teaching datasets, not
substitutes for published benchmark protocols. Their derivations, tensor
contracts, training examples, citations, and benchmark handoffs are in
Dataset-Backed Equilibrium Labs; the typed
builders are documented in the Recent Equilibrium Dataset API.
From Table to Interaction Graph
A tabular dataset can be converted into a sample graph by connecting each
sample to its nearest neighbors. For a destination entity \(i\), define
\[
\mathcal N_k(i)=
\operatorname{arg\,topk}_{j\ne i}
\left(-d(\tilde x_i,\tilde x_j)\right).
\]
With Euclidean distance,
\[
d(\tilde x_i,\tilde x_j)
=
\left\|\tilde x_i-\tilde x_j\right\|_2.
\]
The adapter returns edges as source -> destination, so every selected neighbor
\(j\in\mathcal N_k(i)\) contributes an edge \((j,i)\):
from silva_networks import load_tabular_dataset, tabular_to_silva_graph
dataset = load_tabular_dataset("wine", root="data", download=True, normalize=True)
graph = tabular_to_silva_graph(dataset, k=8, normalize=True, undirected=True)
logits = model(
graph.x,
edge_index=graph.edge_index,
batch=graph.batch,
)
The same adapter accepts a private NumPy array or tensor:
graph = tabular_to_silva_graph(
my_features,
y=my_labels,
k=12,
normalize=True,
metric="cosine",
)
Images
Vector SILVA image models consume one row per image:
from silva_networks import images_to_silva_vectors
batch = images_to_silva_vectors(images, y=labels)
logits = vector_model(batch.x)
Graph-style image models can instead use one entity per pixel. The helper builds
four-neighbor or eight-neighbor grid edges:
from silva_networks import images_to_silva_pixel_graph
pixels = images_to_silva_pixel_graph(images, y=labels, include_diagonals=True)
graph_logits = graph_model(
pixels.x,
edge_index=pixels.edge_index,
batch=pixels.batch,
)
For CIFAR-style images, the image tensor has shape
\[
x\in\mathbb R^{B\times 3\times 32\times 32}.
\]
The vector route flattens each image:
\[
\operatorname{vec}(x_b)\in\mathbb R^{3072},
\]
while the cortex preset keeps the image layout, applies the convolutional
retina, then solves linked SILVA equilibrium points.
Molecules and Relational Graphs
Molecular and relational datasets already have entities and edges. The adapter
packs them and validates compatible shapes:
from silva_networks import molecular_to_silva_graph
molecules = molecular_to_silva_graph(
x=atom_ids_or_features,
edge_index=bond_index,
edge_attr=bond_ids_or_features,
batch=molecule_ids,
y=targets,
)
prediction = molecular_model(**molecules.model_kwargs())
Categorical atom and bond tensors can be passed directly to
SILVAMolecularRegressor. Continuous atom and bond features use
atom_feature_dim and bond_feature_dim in the model constructor.
Call GraphTensorBatch.validate() before a solve, then report feature
normalization, graph construction, entity and edge counts, split policy, and
dataset version with the model's residual diagnostics. The full preparation
and reporting path is in
Datasets and Preprocessing, with
dataset sources under
Citation Rules for Reports.
DatasetInfo
dataclass
Metadata for a downloadable public dataset.
Source code in src/silva_networks/datasets.py
| @dataclass(frozen=True)
class DatasetInfo:
"""Metadata for a downloadable public dataset."""
name: str
domain: str
task: TaskKind
url: str
file_name: str
description: str
source: str
delimiter: str | None = ","
has_header: bool = False
target_column: int = -1
feature_columns: tuple[int, ...] | None = None
drop_columns: tuple[int, ...] = ()
|
FeatureStandardization
dataclass
Training-split statistics for leakage-free NumPy preprocessing.
Source code in src/silva_networks/datasets.py
| @dataclass(frozen=True)
class FeatureStandardization:
"""Training-split statistics for leakage-free NumPy preprocessing."""
mean: np.ndarray
scale: np.ndarray
def transform(self, x: np.ndarray) -> np.ndarray:
"""Impute and standardize features using these fitted statistics."""
if x.ndim != 2 or x.shape[1] != self.mean.shape[1]:
raise ValueError("x must be a 2D array with the fitted feature width")
values = x.astype(np.float32, copy=True)
values = np.where(np.isfinite(values), values, self.mean)
return (values - self.mean) / self.scale
|
Impute and standardize features using these fitted statistics.
Source code in src/silva_networks/datasets.py
| def transform(self, x: np.ndarray) -> np.ndarray:
"""Impute and standardize features using these fitted statistics."""
if x.ndim != 2 or x.shape[1] != self.mean.shape[1]:
raise ValueError("x must be a 2D array with the fitted feature width")
values = x.astype(np.float32, copy=True)
values = np.where(np.isfinite(values), values, self.mean)
return (values - self.mean) / self.scale
|
GraphTensorBatch
dataclass
SILVA-ready tensor container.
The core SILVA graph and set APIs consume tensors with this structure. A
dataset adapter may create these tensors from public datasets, private files,
simulations, image grids, molecular tables, or user-defined objects.
Attributes:
| Name |
Type |
Description |
x |
Tensor
|
Entity features with shape (entities, features).
|
edge_index |
Tensor | None
|
Optional source/destination edge tensor with shape
(2, edges).
|
y |
Tensor | None
|
Optional labels or regression targets.
|
batch |
Tensor | None
|
Optional graph/set id for each entity with shape (entities,).
|
edge_attr |
Tensor | None
|
Optional edge features with shape (edges, edge_features).
|
metadata |
dict[str, Any] | None
|
Optional descriptive fields such as feature names or source.
|
Source code in src/silva_networks/datasets.py
| @dataclass
class GraphTensorBatch:
"""SILVA-ready tensor container.
The core SILVA graph and set APIs consume tensors with this structure. A
dataset adapter may create these tensors from public datasets, private files,
simulations, image grids, molecular tables, or user-defined objects.
Attributes:
x: Entity features with shape `(entities, features)`.
edge_index: Optional source/destination edge tensor with shape
`(2, edges)`.
y: Optional labels or regression targets.
batch: Optional graph/set id for each entity with shape `(entities,)`.
edge_attr: Optional edge features with shape `(edges, edge_features)`.
metadata: Optional descriptive fields such as feature names or source.
"""
x: torch.Tensor
edge_index: torch.Tensor | None = None
y: torch.Tensor | None = None
batch: torch.Tensor | None = None
edge_attr: torch.Tensor | None = None
metadata: dict[str, Any] | None = None
@property
def num_entities(self) -> int:
"""Number of entity rows in `x`."""
return int(self.x.shape[0])
@property
def num_edges(self) -> int:
"""Number of edges in `edge_index`, or zero when no edges are present."""
return 0 if self.edge_index is None else int(self.edge_index.shape[1])
@property
def num_graphs(self) -> int:
"""Number of graph/set ids represented by `batch`."""
if self.batch is None or self.batch.numel() == 0:
return 1
return int(self.batch.max().item()) + 1
def to(self, device: str | torch.device) -> GraphTensorBatch:
"""Move all tensor fields to a PyTorch device."""
return GraphTensorBatch(
x=self.x.to(device),
edge_index=None if self.edge_index is None else self.edge_index.to(device),
y=None if self.y is None else self.y.to(device),
batch=None if self.batch is None else self.batch.to(device),
edge_attr=None if self.edge_attr is None else self.edge_attr.to(device),
metadata=self.metadata,
)
def model_kwargs(self) -> dict[str, torch.Tensor]:
"""Return keyword tensors accepted by SILVA graph-style models."""
kwargs: dict[str, torch.Tensor] = {"x": self.x}
if self.edge_index is not None:
kwargs["edge_index"] = self.edge_index
if self.edge_attr is not None:
kwargs["edge_attr"] = self.edge_attr
if self.batch is not None:
kwargs["batch"] = self.batch
return kwargs
def validate(self, raise_on_error: bool = True) -> bool:
"""Validate that the tensor fields satisfy the SILVA graph contract."""
return validate_graph_tensor_batch(self, raise_on_error=raise_on_error)
|
num_edges
property
Number of edges in edge_index, or zero when no edges are present.
num_entities
property
Number of entity rows in x.
num_graphs
property
Number of graph/set ids represented by batch.
model_kwargs
Return keyword tensors accepted by SILVA graph-style models.
Source code in src/silva_networks/datasets.py
| def model_kwargs(self) -> dict[str, torch.Tensor]:
"""Return keyword tensors accepted by SILVA graph-style models."""
kwargs: dict[str, torch.Tensor] = {"x": self.x}
if self.edge_index is not None:
kwargs["edge_index"] = self.edge_index
if self.edge_attr is not None:
kwargs["edge_attr"] = self.edge_attr
if self.batch is not None:
kwargs["batch"] = self.batch
return kwargs
|
to
Move all tensor fields to a PyTorch device.
Source code in src/silva_networks/datasets.py
| def to(self, device: str | torch.device) -> GraphTensorBatch:
"""Move all tensor fields to a PyTorch device."""
return GraphTensorBatch(
x=self.x.to(device),
edge_index=None if self.edge_index is None else self.edge_index.to(device),
y=None if self.y is None else self.y.to(device),
batch=None if self.batch is None else self.batch.to(device),
edge_attr=None if self.edge_attr is None else self.edge_attr.to(device),
metadata=self.metadata,
)
|
validate
validate(raise_on_error=True)
Validate that the tensor fields satisfy the SILVA graph contract.
Source code in src/silva_networks/datasets.py
| def validate(self, raise_on_error: bool = True) -> bool:
"""Validate that the tensor fields satisfy the SILVA graph contract."""
return validate_graph_tensor_batch(self, raise_on_error=raise_on_error)
|
TabularDataset
dataclass
In-memory tabular dataset returned by load_tabular_dataset.
Source code in src/silva_networks/datasets.py
| @dataclass
class TabularDataset:
"""In-memory tabular dataset returned by ``load_tabular_dataset``."""
name: str
x: np.ndarray
y: np.ndarray
task: TaskKind
feature_names: list[str]
target_names: list[str]
path: Path
info: DatasetInfo
def tensors(
self, device: str | torch.device | None = None
) -> tuple[torch.Tensor, torch.Tensor]:
"""Return ``(x, y)`` as PyTorch tensors."""
x = torch.as_tensor(self.x, dtype=torch.float32, device=device)
if self.task == "classification":
y = torch.as_tensor(self.y, dtype=torch.long, device=device)
else:
y = torch.as_tensor(self.y, dtype=torch.float32, device=device)
return x, y
|
tensors
Return (x, y) as PyTorch tensors.
Source code in src/silva_networks/datasets.py
| def tensors(
self, device: str | torch.device | None = None
) -> tuple[torch.Tensor, torch.Tensor]:
"""Return ``(x, y)`` as PyTorch tensors."""
x = torch.as_tensor(self.x, dtype=torch.float32, device=device)
if self.task == "classification":
y = torch.as_tensor(self.y, dtype=torch.long, device=device)
else:
y = torch.as_tensor(self.y, dtype=torch.float32, device=device)
return x, y
|
TensorStandardization
dataclass
Training-split statistics for leakage-free tensor preprocessing.
Source code in src/silva_networks/datasets.py
| @dataclass(frozen=True)
class TensorStandardization:
"""Training-split statistics for leakage-free tensor preprocessing."""
mean: torch.Tensor
scale: torch.Tensor
def to(self, device: str | torch.device) -> TensorStandardization:
return TensorStandardization(self.mean.to(device), self.scale.to(device))
def transform(self, x: torch.Tensor) -> torch.Tensor:
"""Impute and standardize features using these fitted statistics."""
if x.dim() != 2 or x.shape[1] != self.mean.shape[1]:
raise ValueError("x must be a 2D tensor with the fitted feature width")
values = x.to(device=self.mean.device, dtype=self.mean.dtype)
values = torch.where(torch.isfinite(values), values, self.mean)
return (values - self.mean) / self.scale
|
Impute and standardize features using these fitted statistics.
Source code in src/silva_networks/datasets.py
| def transform(self, x: torch.Tensor) -> torch.Tensor:
"""Impute and standardize features using these fitted statistics."""
if x.dim() != 2 or x.shape[1] != self.mean.shape[1]:
raise ValueError("x must be a 2D tensor with the fitted feature width")
values = x.to(device=self.mean.device, dtype=self.mean.dtype)
values = torch.where(torch.isfinite(values), values, self.mean)
return (values - self.mean) / self.scale
|
available_datasets
available_datasets(domain=None)
List available package-managed public datasets.
Source code in src/silva_networks/datasets.py
| def available_datasets(domain: str | None = None) -> list[str]:
"""List available package-managed public datasets."""
names = [
name for name, info in DATASET_REGISTRY.items() if domain is None or info.domain == domain
]
return sorted(names)
|
available_torchvision_datasets
available_torchvision_datasets()
Return torchvision dataset names accepted by load_torchvision_dataset.
The vision datasets are loaded through the optional vision extra because
they depend on TorchVision's dataset classes and download mirrors. The
returned names include CIFAR10, CIFAR100, MNIST, FashionMNIST, KMNIST,
EMNIST, and SVHN.
Source code in src/silva_networks/datasets.py
| def available_torchvision_datasets() -> tuple[str, ...]:
"""Return torchvision dataset names accepted by `load_torchvision_dataset`.
The vision datasets are loaded through the optional ``vision`` extra because
they depend on TorchVision's dataset classes and download mirrors. The
returned names include CIFAR10, CIFAR100, MNIST, FashionMNIST, KMNIST,
EMNIST, and SVHN.
"""
return TORCHVISION_DATASETS
|
dataset_info
Return metadata for one registered dataset.
Source code in src/silva_networks/datasets.py
| def dataset_info(name: str) -> DatasetInfo:
"""Return metadata for one registered dataset."""
try:
return DATASET_REGISTRY[name]
except KeyError as exc:
available = ", ".join(available_datasets())
raise KeyError(f"Unknown dataset {name!r}. Available datasets: {available}") from exc
|
dataset_path
dataset_path(name, root='data')
Return the expected local path for a registered dataset file.
Source code in src/silva_networks/datasets.py
| def dataset_path(name: str, root: str | Path = "data") -> Path:
"""Return the expected local path for a registered dataset file."""
info = dataset_info(name)
return Path(root) / info.name / info.file_name
|
download_dataset
download_dataset(name, root='data', force=False)
Download one registered dataset into root/name/file.
Source code in src/silva_networks/datasets.py
| def download_dataset(name: str, root: str | Path = "data", force: bool = False) -> Path:
"""Download one registered dataset into ``root/name/file``."""
info = dataset_info(name)
path = dataset_path(name, root)
if path.exists() and not force:
return path
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(_download_bytes(info.url))
return path
|
download_many
download_many(names=None, root='data', force=False)
Download several datasets and return their local paths.
Source code in src/silva_networks/datasets.py
| def download_many(
names: list[str] | tuple[str, ...] | None = None,
root: str | Path = "data",
force: bool = False,
) -> dict[str, Path]:
"""Download several datasets and return their local paths."""
selected = list(names) if names is not None else available_datasets()
return {name: download_dataset(name, root=root, force=force) for name in selected}
|
fit_feature_standardization
fit_feature_standardization(x, eps=1e-08)
Fit imputation and scaling statistics on a training feature matrix.
Source code in src/silva_networks/datasets.py
| def fit_feature_standardization(
x: np.ndarray,
eps: float = 1e-8,
) -> FeatureStandardization:
"""Fit imputation and scaling statistics on a training feature matrix."""
if eps <= 0:
raise ValueError("eps must be positive")
if x.ndim != 2 or x.shape[0] == 0 or x.shape[1] == 0:
raise ValueError("x must be a nonempty 2D feature array")
values = x.astype(np.float32, copy=True)
finite = np.isfinite(values)
counts = finite.sum(axis=0, keepdims=True)
means = np.divide(
np.where(finite, values, 0.0).sum(axis=0, keepdims=True),
np.maximum(counts, 1),
).astype(np.float32)
imputed = np.where(finite, values, means)
scale = imputed.std(axis=0, keepdims=True)
return FeatureStandardization(means, np.maximum(scale, eps).astype(np.float32))
|
fit_tensor_standardization
fit_tensor_standardization(x, eps=1e-08)
Fit imputation and scaling statistics on a training feature tensor.
Source code in src/silva_networks/datasets.py
| def fit_tensor_standardization(
x: torch.Tensor,
eps: float = 1e-8,
) -> TensorStandardization:
"""Fit imputation and scaling statistics on a training feature tensor."""
if eps <= 0:
raise ValueError("eps must be positive")
if x.dim() != 2 or x.shape[0] == 0 or x.shape[1] == 0:
raise ValueError("x must be a nonempty 2D feature tensor")
values = x.float().clone()
finite = torch.isfinite(values)
safe_values = torch.where(finite, values, torch.zeros_like(values))
counts = finite.sum(dim=0, keepdim=True).clamp_min(1)
means = safe_values.sum(dim=0, keepdim=True) / counts
values = torch.where(finite, values, means)
centered = values - means
std = centered.square().mean(dim=0, keepdim=True).sqrt()
return TensorStandardization(means, std.clamp_min(eps))
|
image_grid_edge_index
image_grid_edge_index(height, width, *, batch_size=1, include_diagonals=False, device=None)
Create pixel-grid edges and a batch vector for image-as-graph models.
Source code in src/silva_networks/datasets.py
| def image_grid_edge_index(
height: int,
width: int,
*,
batch_size: int = 1,
include_diagonals: bool = False,
device: str | torch.device | None = None,
) -> tuple[torch.Tensor, torch.Tensor]:
"""Create pixel-grid edges and a batch vector for image-as-graph models."""
if height < 1 or width < 1 or batch_size < 1:
raise ValueError("height, width, and batch_size must be positive")
offsets = [(-1, 0), (1, 0), (0, -1), (0, 1)]
if include_diagonals:
offsets.extend([(-1, -1), (-1, 1), (1, -1), (1, 1)])
src: list[int] = []
dst: list[int] = []
nodes_per_image = height * width
for batch_index in range(batch_size):
base = batch_index * nodes_per_image
for row in range(height):
for col in range(width):
receiver = base + row * width + col
for drow, dcol in offsets:
nrow = row + drow
ncol = col + dcol
if 0 <= nrow < height and 0 <= ncol < width:
source = base + nrow * width + ncol
src.append(source)
dst.append(receiver)
out_device = torch.device(device) if device is not None else None
edge_index = torch.tensor([src, dst], dtype=torch.long, device=out_device)
batch = torch.arange(batch_size, dtype=torch.long, device=out_device).repeat_interleave(
nodes_per_image
)
return edge_index, batch
|
images_to_silva_pixel_graph
images_to_silva_pixel_graph(images, *, y=None, include_diagonals=False, scale_uint8=True, channel_last=None, device=None)
Convert images into pixel entities with grid-local edges.
The resulting x has one row per pixel. For grayscale images the feature
width is 1; for color images it is the channel count.
Source code in src/silva_networks/datasets.py
| def images_to_silva_pixel_graph(
images: np.ndarray | torch.Tensor,
*,
y: np.ndarray | torch.Tensor | None = None,
include_diagonals: bool = False,
scale_uint8: bool = True,
channel_last: bool | None = None,
device: str | torch.device | None = None,
) -> GraphTensorBatch:
"""Convert images into pixel entities with grid-local edges.
The resulting `x` has one row per pixel. For grayscale images the feature
width is `1`; for color images it is the channel count.
"""
tensor = torch.as_tensor(images, device=device)
if tensor.dtype in {torch.uint8, torch.int8, torch.int16, torch.int32, torch.int64, torch.long}:
tensor = tensor.float()
if scale_uint8:
tensor = tensor / 255.0
else:
tensor = tensor.float()
if tensor.dim() == 3:
tensor = tensor.unsqueeze(1)
if tensor.dim() != 4:
raise ValueError("images_to_silva_pixel_graph expects 3D or 4D image input")
tensor = _channels_first_images(tensor, channel_last)
batch_size, channels, height, width = tensor.shape
x = tensor.permute(0, 2, 3, 1).reshape(batch_size * height * width, channels)
edge_index, batch = image_grid_edge_index(
height,
width,
batch_size=batch_size,
include_diagonals=include_diagonals,
device=x.device,
)
y_tensor = _as_target_tensor(y, None, device=x.device) if y is not None else None
packed = GraphTensorBatch(
x=x,
edge_index=edge_index,
y=y_tensor,
batch=batch,
metadata={"adapter": "images_to_silva_pixel_graph", "height": height, "width": width},
)
packed.validate()
return packed
|
images_to_silva_vectors
images_to_silva_vectors(images, *, y=None, scale_uint8=True, channel_last=None, device=None)
Flatten image batches into vector features for vector SILVA models.
Parameters:
| Name |
Type |
Description |
Default |
images
|
ndarray | Tensor
|
Image tensor/array with shape (batch, channels, height, width)
or (batch, height, width, channels).
|
required
|
y
|
ndarray | Tensor | None
|
|
None
|
scale_uint8
|
bool
|
Whether integer image arrays are divided by 255.
|
True
|
channel_last
|
bool | None
|
Explicitly interpret 4D input as NHWC (True) or NCHW
(False). None uses a conservative shape heuristic.
|
None
|
device
|
str | device | None
|
|
None
|
Returns:
| Type |
Description |
GraphTensorBatch
|
GraphTensorBatch whose x has shape (batch, pixels_or_features).
|
Source code in src/silva_networks/datasets.py
| def images_to_silva_vectors(
images: np.ndarray | torch.Tensor,
*,
y: np.ndarray | torch.Tensor | None = None,
scale_uint8: bool = True,
channel_last: bool | None = None,
device: str | torch.device | None = None,
) -> GraphTensorBatch:
"""Flatten image batches into vector features for vector SILVA models.
Args:
images: Image tensor/array with shape `(batch, channels, height, width)`
or `(batch, height, width, channels)`.
y: Optional labels.
scale_uint8: Whether integer image arrays are divided by 255.
channel_last: Explicitly interpret 4D input as NHWC (`True`) or NCHW
(`False`). `None` uses a conservative shape heuristic.
device: Optional output device.
Returns:
`GraphTensorBatch` whose `x` has shape `(batch, pixels_or_features)`.
"""
tensor = torch.as_tensor(images, device=device)
if tensor.dtype in {torch.uint8, torch.int8, torch.int16, torch.int32, torch.int64, torch.long}:
tensor = tensor.float()
if scale_uint8:
tensor = tensor / 255.0
else:
tensor = tensor.float()
if tensor.dim() < 2:
raise ValueError("images_to_silva_vectors expects a batch dimension")
tensor = _channels_first_images(tensor, channel_last)
x = tensor.flatten(1)
y_tensor = _as_target_tensor(y, None, device=x.device) if y is not None else None
return GraphTensorBatch(x=x, y=y_tensor, metadata={"adapter": "images_to_silva_vectors"})
|
load_tabular_dataset
load_tabular_dataset(name, root='data', download=True, normalize=False)
Download, parse, and optionally standardize a tabular dataset.
The default returns raw features so train/validation/test splits can be
created before fitting FeatureStandardization. Set normalize=True only
when fitting statistics on the complete table is intentional.
Source code in src/silva_networks/datasets.py
| def load_tabular_dataset(
name: str,
root: str | Path = "data",
download: bool = True,
normalize: bool = False,
) -> TabularDataset:
"""Download, parse, and optionally standardize a tabular dataset.
The default returns raw features so train/validation/test splits can be
created before fitting `FeatureStandardization`. Set `normalize=True` only
when fitting statistics on the complete table is intentional.
"""
info = dataset_info(name)
path = download_dataset(name, root=root) if download else dataset_path(name, root)
if not path.exists():
raise FileNotFoundError(f"{path} does not exist. Use download=True or download_dataset().")
rows, header = _read_rows(path, info)
x, y, feature_names, target_names = _rows_to_arrays(rows, header, info)
if normalize:
x = standardize_features(x)
return TabularDataset(
name=info.name,
x=x,
y=y,
task=info.task,
feature_names=feature_names,
target_names=target_names,
path=path,
info=info,
)
|
load_torchvision_dataset
load_torchvision_dataset(name, root='data', train=True, download=True, transform=None, **kwargs)
Load a torchvision dataset by name when the optional vision extra is installed.
Source code in src/silva_networks/datasets.py
| def load_torchvision_dataset(
name: str,
root: str | Path = "data",
train: bool = True,
download: bool = True,
transform: Any | None = None,
**kwargs,
) -> Any:
"""Load a torchvision dataset by name when the optional vision extra is installed."""
try:
from torchvision import datasets, transforms
except ImportError as exc:
raise ImportError(
"Install the vision extra before loading torchvision datasets: "
'python -m pip install "silva-networks[vision]"'
) from exc
if name not in TORCHVISION_DATASETS:
available = ", ".join(TORCHVISION_DATASETS)
raise KeyError(f"Unknown torchvision dataset {name!r}. Available datasets: {available}")
dataset_cls = getattr(datasets, name)
transform = transform or transforms.ToTensor()
if name == "EMNIST":
split = kwargs.pop("split", "balanced")
return _instantiate_torchvision_dataset(
dataset_cls,
root=str(root),
split=split,
train=train,
download=download,
transform=transform,
**kwargs,
)
if name == "SVHN":
split = kwargs.pop("split", "train" if train else "test")
return _instantiate_torchvision_dataset(
dataset_cls,
root=str(root),
split=split,
download=download,
transform=transform,
**kwargs,
)
return _instantiate_torchvision_dataset(
dataset_cls,
root=str(root),
train=train,
download=download,
transform=transform,
**kwargs,
)
|
make_knn_edge_index
make_knn_edge_index(x, k, *, batch=None, metric='euclidean', include_self=False, undirected=False, device=None)
Build a k-nearest-neighbor edge_index tensor.
Edges are returned as source -> destination. For each destination entity
\(i\), the selected sources are its nearest neighbors \(j\), so local message
passing computes incoming neighbor information.
Parameters:
| Name |
Type |
Description |
Default |
x
|
ndarray | Tensor
|
Entity features with shape (entities, features).
|
required
|
k
|
int
|
Number of neighbors per entity.
|
required
|
batch
|
ndarray | Tensor | None
|
Optional graph id for each entity. Neighbor search is performed
independently inside each graph id.
|
None
|
metric
|
Literal['euclidean', 'cosine']
|
Distance geometry, either euclidean or cosine.
|
'euclidean'
|
include_self
|
bool
|
If true, self-neighbors may be selected.
|
False
|
undirected
|
bool
|
If true, append the reverse of every edge and remove
duplicate columns.
|
False
|
device
|
str | device | None
|
Optional output device. Defaults to the feature tensor device.
|
None
|
Returns:
| Type |
Description |
Tensor
|
Long tensor with shape (2, edges).
|
Source code in src/silva_networks/datasets.py
| def make_knn_edge_index(
x: np.ndarray | torch.Tensor,
k: int,
*,
batch: np.ndarray | torch.Tensor | None = None,
metric: Literal["euclidean", "cosine"] = "euclidean",
include_self: bool = False,
undirected: bool = False,
device: str | torch.device | None = None,
) -> torch.Tensor:
r"""Build a k-nearest-neighbor `edge_index` tensor.
Edges are returned as `source -> destination`. For each destination entity
\(i\), the selected sources are its nearest neighbors \(j\), so local message
passing computes incoming neighbor information.
Args:
x: Entity features with shape `(entities, features)`.
k: Number of neighbors per entity.
batch: Optional graph id for each entity. Neighbor search is performed
independently inside each graph id.
metric: Distance geometry, either `euclidean` or `cosine`.
include_self: If true, self-neighbors may be selected.
undirected: If true, append the reverse of every edge and remove
duplicate columns.
device: Optional output device. Defaults to the feature tensor device.
Returns:
Long tensor with shape `(2, edges)`.
"""
if k < 0:
raise ValueError("k must be nonnegative")
features = _as_float_tensor(x, device=device)
if features.dim() != 2:
raise ValueError("make_knn_edge_index expects x with shape (entities, features)")
out_device = features.device
if batch is None:
batch_tensor = torch.zeros(features.shape[0], dtype=torch.long, device=out_device)
else:
batch_tensor = torch.as_tensor(batch, dtype=torch.long, device=out_device)
if batch_tensor.shape != (features.shape[0],):
raise ValueError("batch must have shape (entities,)")
if k == 0 or features.shape[0] <= 1:
return torch.empty(2, 0, dtype=torch.long, device=out_device)
sources: list[torch.Tensor] = []
destinations: list[torch.Tensor] = []
for graph_id in torch.unique(batch_tensor, sorted=True):
members = torch.nonzero(batch_tensor == graph_id, as_tuple=False).flatten()
if members.numel() == 0:
continue
local_x = features[members]
if metric == "euclidean":
distances = torch.cdist(local_x, local_x)
elif metric == "cosine":
normalized = torch.nn.functional.normalize(local_x, dim=-1)
distances = 1.0 - normalized @ normalized.T
else:
raise ValueError("metric must be 'euclidean' or 'cosine'")
candidates = members.numel() if include_self else members.numel() - 1
if candidates <= 0:
continue
k_eff = min(k, int(candidates))
if not include_self:
eye = torch.eye(members.numel(), dtype=torch.bool, device=out_device)
distances = distances.masked_fill(eye, float("inf"))
neighbor_idx = distances.topk(k_eff, largest=False).indices
destination = members.repeat_interleave(k_eff)
source = members[neighbor_idx.reshape(-1)]
sources.append(source)
destinations.append(destination)
if not sources:
return torch.empty(2, 0, dtype=torch.long, device=out_device)
edge_index = torch.stack([torch.cat(sources), torch.cat(destinations)], dim=0)
if undirected:
reverse = edge_index.flip(0)
edge_index = torch.unique(torch.cat([edge_index, reverse], dim=1), dim=1)
return edge_index
|
molecular_to_silva_graph
molecular_to_silva_graph(*, x, edge_index, edge_attr=None, batch=None, y=None, device=None)
Pack atom, bond, and molecule-index tensors for molecular SILVA models.
Source code in src/silva_networks/datasets.py
| def molecular_to_silva_graph(
*,
x: np.ndarray | torch.Tensor,
edge_index: np.ndarray | torch.Tensor,
edge_attr: np.ndarray | torch.Tensor | None = None,
batch: np.ndarray | torch.Tensor | None = None,
y: np.ndarray | torch.Tensor | None = None,
device: str | torch.device | None = None,
) -> GraphTensorBatch:
"""Pack atom, bond, and molecule-index tensors for molecular SILVA models."""
x_tensor = torch.as_tensor(x, device=device)
edge_index_tensor = torch.as_tensor(edge_index, dtype=torch.long, device=device)
if edge_index_tensor.shape[0] != 2:
raise ValueError("edge_index must have shape (2, edges)")
edge_attr_tensor = None if edge_attr is None else torch.as_tensor(edge_attr, device=device)
if edge_attr_tensor is not None and edge_attr_tensor.shape[0] != edge_index_tensor.shape[1]:
raise ValueError("edge_attr must have one row/value per edge")
batch_tensor = (
torch.zeros(x_tensor.shape[0], dtype=torch.long, device=x_tensor.device)
if batch is None
else torch.as_tensor(batch, dtype=torch.long, device=x_tensor.device)
)
y_tensor = _as_target_tensor(y, None, device=x_tensor.device) if y is not None else None
packed = GraphTensorBatch(
x=x_tensor,
edge_index=edge_index_tensor,
y=y_tensor,
batch=batch_tensor,
edge_attr=edge_attr_tensor,
metadata={"adapter": "molecular_to_silva_graph"},
)
packed.validate()
return packed
|
pyg_data_to_silva_graph
pyg_data_to_silva_graph(data, *, device=None)
Convert a PyG-like data object into GraphTensorBatch.
The function does not require PyTorch Geometric as a runtime dependency. It
reads the conventional attributes x, edge_index, edge_attr, batch,
and y when they are present.
Source code in src/silva_networks/datasets.py
| def pyg_data_to_silva_graph(
data: Any,
*,
device: str | torch.device | None = None,
) -> GraphTensorBatch:
"""Convert a PyG-like data object into `GraphTensorBatch`.
The function does not require PyTorch Geometric as a runtime dependency. It
reads the conventional attributes `x`, `edge_index`, `edge_attr`, `batch`,
and `y` when they are present.
"""
if not hasattr(data, "x"):
raise ValueError("PyG-like data must have an x attribute")
x = data.x
edge_index = getattr(data, "edge_index", None)
edge_attr = getattr(data, "edge_attr", None)
batch = getattr(data, "batch", None)
y = getattr(data, "y", None)
packed = GraphTensorBatch(
x=torch.as_tensor(x, device=device),
edge_index=None
if edge_index is None
else torch.as_tensor(edge_index, dtype=torch.long, device=device),
edge_attr=None if edge_attr is None else torch.as_tensor(edge_attr, device=device),
batch=None if batch is None else torch.as_tensor(batch, dtype=torch.long, device=device),
y=None if y is None else torch.as_tensor(y, device=device),
metadata={"adapter": "pyg_data_to_silva_graph"},
)
packed.validate()
return packed
|
standardize_features
standardize_features(x, eps=1e-08)
Column-standardize a feature matrix after mean-imputing missing values.
Source code in src/silva_networks/datasets.py
| def standardize_features(x: np.ndarray, eps: float = 1e-8) -> np.ndarray:
"""Column-standardize a feature matrix after mean-imputing missing values."""
return fit_feature_standardization(x, eps=eps).transform(x)
|
standardize_tensor
standardize_tensor(x, eps=1e-08)
Column-standardize a tensor after mean-imputing nonfinite entries.
Parameters:
| Name |
Type |
Description |
Default |
x
|
Tensor
|
Feature tensor with shape (samples, features).
|
required
|
eps
|
float
|
Minimum divisor used to avoid zero-variance division.
|
1e-08
|
Returns:
| Type |
Description |
Tensor
|
Standardized floating tensor with the same shape as x.
|
Source code in src/silva_networks/datasets.py
| def standardize_tensor(x: torch.Tensor, eps: float = 1e-8) -> torch.Tensor:
"""Column-standardize a tensor after mean-imputing nonfinite entries.
Args:
x: Feature tensor with shape `(samples, features)`.
eps: Minimum divisor used to avoid zero-variance division.
Returns:
Standardized floating tensor with the same shape as `x`.
"""
return fit_tensor_standardization(x, eps=eps).transform(x)
|
tabular_to_silva_graph
tabular_to_silva_graph(data, *, y=None, k=8, batch=None, normalize=False, max_samples=None, metric='euclidean', undirected=False, device=None)
Convert tabular features into a SILVA-ready sample graph.
Parameters:
| Name |
Type |
Description |
Default |
data
|
TabularDataset | ndarray | Tensor
|
A TabularDataset, NumPy feature matrix, or tensor feature matrix.
|
required
|
y
|
ndarray | Tensor | None
|
Optional target array when data is not a TabularDataset.
|
None
|
k
|
int
|
Number of neighbors used for the sample interaction graph.
|
8
|
batch
|
ndarray | Tensor | None
|
Optional graph id for each row; by default all rows form one graph.
|
None
|
normalize
|
bool
|
Whether to standardize features inside this adapter.
|
False
|
max_samples
|
int | None
|
Optional prefix length for compact experiments.
|
None
|
metric
|
Literal['euclidean', 'cosine']
|
Geometry used by kNN graph construction.
|
'euclidean'
|
undirected
|
bool
|
Whether to include reverse edges.
|
False
|
device
|
str | device | None
|
|
None
|
Returns:
| Type |
Description |
GraphTensorBatch
|
GraphTensorBatch with x, edge_index, optional y, and batch.
|
Source code in src/silva_networks/datasets.py
| def tabular_to_silva_graph(
data: TabularDataset | np.ndarray | torch.Tensor,
*,
y: np.ndarray | torch.Tensor | None = None,
k: int = 8,
batch: np.ndarray | torch.Tensor | None = None,
normalize: bool = False,
max_samples: int | None = None,
metric: Literal["euclidean", "cosine"] = "euclidean",
undirected: bool = False,
device: str | torch.device | None = None,
) -> GraphTensorBatch:
"""Convert tabular features into a SILVA-ready sample graph.
Args:
data: A `TabularDataset`, NumPy feature matrix, or tensor feature matrix.
y: Optional target array when `data` is not a `TabularDataset`.
k: Number of neighbors used for the sample interaction graph.
batch: Optional graph id for each row; by default all rows form one graph.
normalize: Whether to standardize features inside this adapter.
max_samples: Optional prefix length for compact experiments.
metric: Geometry used by kNN graph construction.
undirected: Whether to include reverse edges.
device: Optional output device.
Returns:
`GraphTensorBatch` with `x`, `edge_index`, optional `y`, and `batch`.
"""
metadata: dict[str, Any] = {}
if max_samples is not None and max_samples < 1:
raise ValueError("max_samples must be positive")
task: TaskKind | None = None
if isinstance(data, TabularDataset):
x_raw: np.ndarray | torch.Tensor = data.x
y_raw: np.ndarray | torch.Tensor | None = data.y
task = data.task
metadata = {
"name": data.name,
"source": data.info.source,
"feature_names": data.feature_names,
"target_names": data.target_names,
"task": data.task,
}
else:
x_raw = data
y_raw = y
if max_samples is not None:
x_raw = x_raw[:max_samples]
if y_raw is not None:
y_raw = y_raw[:max_samples]
if batch is not None:
batch = batch[:max_samples]
if normalize:
if isinstance(x_raw, torch.Tensor):
x_tensor = standardize_tensor(x_raw).to(device=device)
else:
x_tensor = torch.as_tensor(
standardize_features(np.asarray(x_raw)),
dtype=torch.float32,
device=device,
)
else:
x_tensor = _as_float_tensor(x_raw, device=device)
batch_tensor = (
torch.zeros(x_tensor.shape[0], dtype=torch.long, device=x_tensor.device)
if batch is None
else torch.as_tensor(batch, dtype=torch.long, device=x_tensor.device)
)
edge_index = make_knn_edge_index(
x_tensor,
k,
batch=batch_tensor,
metric=metric,
undirected=undirected,
)
y_tensor = _as_target_tensor(y_raw, task, device=x_tensor.device) if y_raw is not None else None
packed = GraphTensorBatch(
x=x_tensor,
edge_index=edge_index,
y=y_tensor,
batch=batch_tensor,
metadata=metadata,
)
packed.validate()
return packed
|
validate_graph_tensor_batch
validate_graph_tensor_batch(data, raise_on_error=True)
Check the tensor contract expected by SILVA graph-style models.
Parameters:
| Name |
Type |
Description |
Default |
data
|
GraphTensorBatch
|
|
required
|
raise_on_error
|
bool
|
If true, raise ValueError on the first invalid field.
|
True
|
Returns:
| Type |
Description |
bool
|
True when the batch is valid; False when invalid and
|
bool
|
|
Source code in src/silva_networks/datasets.py
| def validate_graph_tensor_batch(data: GraphTensorBatch, raise_on_error: bool = True) -> bool:
"""Check the tensor contract expected by SILVA graph-style models.
Args:
data: Packed tensor batch.
raise_on_error: If true, raise `ValueError` on the first invalid field.
Returns:
`True` when the batch is valid; `False` when invalid and
`raise_on_error=False`.
"""
if not isinstance(data.x, torch.Tensor):
return _validation_error("x must be a torch.Tensor", raise_on_error)
if data.x.dim() not in {1, 2}:
return _validation_error(
"x must have shape (entities,) for categorical ids or (entities, features)",
raise_on_error,
)
if data.edge_index is not None:
if not isinstance(data.edge_index, torch.Tensor):
return _validation_error("edge_index must be a torch.Tensor", raise_on_error)
if data.edge_index.dtype != torch.long:
return _validation_error("edge_index must have dtype torch.long", raise_on_error)
if data.edge_index.dim() != 2 or data.edge_index.shape[0] != 2:
return _validation_error("edge_index must have shape (2, edges)", raise_on_error)
if data.edge_index.device != data.x.device:
return _validation_error("edge_index must be on the x device", raise_on_error)
if data.edge_index.numel() > 0:
if int(data.edge_index.min().item()) < 0:
return _validation_error(
"edge_index contains a negative node index", raise_on_error
)
if int(data.edge_index.max().item()) >= data.x.shape[0]:
return _validation_error("edge_index contains an index outside x", raise_on_error)
if data.edge_attr is not None:
if not isinstance(data.edge_attr, torch.Tensor) or data.edge_attr.dim() == 0:
return _validation_error("edge_attr must be a non-scalar tensor", raise_on_error)
if data.edge_index is None:
return _validation_error("edge_attr requires edge_index", raise_on_error)
if data.edge_attr.device != data.x.device:
return _validation_error("edge_attr must be on the x device", raise_on_error)
if data.edge_attr.shape[0] != data.edge_index.shape[1]:
return _validation_error("edge_attr must have one row/value per edge", raise_on_error)
if data.batch is not None:
if not isinstance(data.batch, torch.Tensor):
return _validation_error("batch must be a torch.Tensor", raise_on_error)
if data.batch.dtype != torch.long:
return _validation_error("batch must have dtype torch.long", raise_on_error)
if data.batch.shape != (data.x.shape[0],):
return _validation_error("batch must have shape (entities,)", raise_on_error)
if data.batch.device != data.x.device:
return _validation_error("batch must be on the x device", raise_on_error)
if data.batch.numel() > 0:
if int(data.batch.min().item()) < 0:
return _validation_error("batch contains a negative graph id", raise_on_error)
ids = torch.unique(data.batch, sorted=True)
expected = torch.arange(ids.numel(), device=ids.device, dtype=ids.dtype)
if not torch.equal(ids, expected):
return _validation_error(
"batch graph ids must be contiguous and start at zero",
raise_on_error,
)
return True
|
Where to Go Next