Source Data
The source-data API complements deterministic known-solution generators with
attributed real-data subsets, complete local loaders, and checksum-verified
snapshots. Compact subsets validate integration; complete benchmark claims
still require the official source protocol.
Registry
from silva_networks import available_source_datasets, source_dataset_info
for name in available_source_datasets():
info = source_dataset_info(name)
print(name, info.domain, info.expected_storage)
The registry covers CIFAR-10, MNIST, SVHN, Cora, CiteSeer, PubMed, MPI Sintel,
KITTI Flow, FlyingChairs, the small public motion source, and DarcyFlowSmall.
Dataset citations are numbered
[81]-[86].
Snapshot Verification
from silva_networks import load_source_snapshot
snapshot = load_source_snapshot(
"docs/assets/source-data/cora-induced-96.pt",
verify=True,
)
print(snapshot.receipt.content_sha256)
print(snapshot.tensors.keys())
The serializer accepts an ordered tensor mapping and rejects content that does
not match the receipt:
from silva_networks import save_source_snapshot
save_source_snapshot(
"compact.pt",
tensors={"images": subset.images, "labels": subset.labels},
receipt=subset.receipt,
)
Shape Contracts
| Result object |
Required tensors |
SILVAVisionSourceSubset |
images: (B,C,H,W), labels: (B,) |
SILVAGraphSourceSubset |
graph features and edges, split masks, original node ids |
SILVAFlowSourceSubset |
frame1, frame2, optional (B,2,H,W) flow and valid mask |
SILVAOperatorSourceSubset |
input and target fields with the same sample count |
SILVASourceSnapshot |
named tensors plus a verified SourceDataReceipt |
API
Installed packages retain the same three compact snapshots used by the
executed documentation. Load one without relying on a repository-relative
path:
from silva_networks import load_bundled_source_snapshot
cora = load_bundled_source_snapshot("cora")
print(cora.receipt.content_sha256)
print(cora.tensors["x"].shape)
The names cifar10, cora, and motion are returned by
available_bundled_source_snapshots(). Complete experiments still use the
source-specific loaders and official local data.
silva_networks.source_data
Source-dataset adapters and reproducibility receipts for SILVA experiments.
The compact generators in :mod:silva_networks.structured_data provide known
solutions for mechanism tests. This module complements them with deterministic
subsets of public datasets used by source-scale experiments. The adapters never
claim that a subset reproduces a published benchmark; each result records the
exact data selection and preprocessing needed to distinguish a teaching run
from the complete protocol.
SourceDatasetInfo
dataclass
Stable metadata for a dataset used by a source-scale SILVA recipe.
Source code in src/silva_networks/source_data.py
| @dataclass(frozen=True)
class SourceDatasetInfo:
"""Stable metadata for a dataset used by a source-scale SILVA recipe."""
name: str
domain: str
task: str
source: str
homepage: str
citation_url: str
access: str
expected_storage: str
official_protocol: str
|
SourceDataReceipt
dataclass
Machine-readable record of one deterministic source-data selection.
Source code in src/silva_networks/source_data.py
| @dataclass(frozen=True)
class SourceDataReceipt:
"""Machine-readable record of one deterministic source-data selection."""
dataset: str
split: str
source: str
homepage: str
citation_url: str
access: str
version: str
subset_size: int
seed: int
selected_indices: tuple[int, ...]
content_sha256: str
preprocessing: tuple[str, ...]
adapter: str
adapter_revision: str = "1"
def as_dict(self) -> dict[str, Any]:
"""Return a JSON-serializable receipt."""
return {
"dataset": self.dataset,
"split": self.split,
"source": self.source,
"homepage": self.homepage,
"citation_url": self.citation_url,
"access": self.access,
"version": self.version,
"subset_size": self.subset_size,
"seed": self.seed,
"selected_indices": list(self.selected_indices),
"content_sha256": self.content_sha256,
"preprocessing": list(self.preprocessing),
"adapter": self.adapter,
"adapter_revision": self.adapter_revision,
}
|
as_dict
Return a JSON-serializable receipt.
Source code in src/silva_networks/source_data.py
| def as_dict(self) -> dict[str, Any]:
"""Return a JSON-serializable receipt."""
return {
"dataset": self.dataset,
"split": self.split,
"source": self.source,
"homepage": self.homepage,
"citation_url": self.citation_url,
"access": self.access,
"version": self.version,
"subset_size": self.subset_size,
"seed": self.seed,
"selected_indices": list(self.selected_indices),
"content_sha256": self.content_sha256,
"preprocessing": list(self.preprocessing),
"adapter": self.adapter,
"adapter_revision": self.adapter_revision,
}
|
SILVAVisionSourceSubset
dataclass
Image tensor, labels, and source record for a deterministic subset.
Source code in src/silva_networks/source_data.py
| @dataclass
class SILVAVisionSourceSubset:
"""Image tensor, labels, and source record for a deterministic subset."""
images: Tensor
labels: Tensor
receipt: SourceDataReceipt
def to(self, device: str | torch.device) -> SILVAVisionSourceSubset:
return SILVAVisionSourceSubset(self.images.to(device), self.labels.to(device), self.receipt)
|
SILVAGraphSourceSubset
dataclass
SILVA graph tensors, split masks, original node ids, and source record.
Source code in src/silva_networks/source_data.py
| @dataclass
class SILVAGraphSourceSubset:
"""SILVA graph tensors, split masks, original node ids, and source record."""
graph: GraphTensorBatch
train_mask: Tensor
validation_mask: Tensor
test_mask: Tensor
node_ids: Tensor
receipt: SourceDataReceipt
def to(self, device: str | torch.device) -> SILVAGraphSourceSubset:
return SILVAGraphSourceSubset(
graph=self.graph.to(device),
train_mask=self.train_mask.to(device),
validation_mask=self.validation_mask.to(device),
test_mask=self.test_mask.to(device),
node_ids=self.node_ids.to(device),
receipt=self.receipt,
)
|
SILVAFlowSourceSubset
dataclass
One or more real image pairs with optional flow supervision.
Source code in src/silva_networks/source_data.py
| @dataclass
class SILVAFlowSourceSubset:
"""One or more real image pairs with optional flow supervision."""
frame1: Tensor
frame2: Tensor
flow: Tensor | None
valid: Tensor | None
receipt: SourceDataReceipt
def to(self, device: str | torch.device) -> SILVAFlowSourceSubset:
return SILVAFlowSourceSubset(
frame1=self.frame1.to(device),
frame2=self.frame2.to(device),
flow=None if self.flow is None else self.flow.to(device),
valid=None if self.valid is None else self.valid.to(device),
receipt=self.receipt,
)
|
SILVAOperatorSourceSubset
dataclass
Input/output fields and source record for an operator-learning dataset.
Source code in src/silva_networks/source_data.py
| @dataclass
class SILVAOperatorSourceSubset:
"""Input/output fields and source record for an operator-learning dataset."""
inputs: Tensor
targets: Tensor
receipt: SourceDataReceipt
def to(self, device: str | torch.device) -> SILVAOperatorSourceSubset:
return SILVAOperatorSourceSubset(
self.inputs.to(device), self.targets.to(device), self.receipt
)
|
SILVASourceSnapshot
dataclass
Named tensors and a verified source-data receipt stored with the repository.
Source code in src/silva_networks/source_data.py
| @dataclass
class SILVASourceSnapshot:
"""Named tensors and a verified source-data receipt stored with the repository."""
tensors: dict[str, Tensor]
receipt: SourceDataReceipt
def to(self, device: str | torch.device) -> SILVASourceSnapshot:
return SILVASourceSnapshot(
{name: value.to(device) for name, value in self.tensors.items()},
self.receipt,
)
|
available_source_datasets
available_source_datasets(domain=None)
Return registered source datasets, optionally filtered by domain.
Source code in src/silva_networks/source_data.py
| def available_source_datasets(domain: str | None = None) -> tuple[str, ...]:
"""Return registered source datasets, optionally filtered by domain."""
return tuple(
sorted(
name
for name, info in SOURCE_DATASET_REGISTRY.items()
if domain is None or info.domain == domain
)
)
|
source_dataset_info
source_dataset_info(name)
Return stable source metadata for one dataset.
Source code in src/silva_networks/source_data.py
| def source_dataset_info(name: str) -> SourceDatasetInfo:
"""Return stable source metadata for one dataset."""
try:
return SOURCE_DATASET_REGISTRY[name]
except KeyError as exc:
available = ", ".join(available_source_datasets())
raise KeyError(f"Unknown source dataset {name!r}. Available datasets: {available}") from exc
|
load_vision_source_subset
load_vision_source_subset(name, *, root='data', train=True, samples_per_class=4, seed=0, image_size=None, normalization='unit', download=False, dataset=None)
Load a deterministic class-balanced vision subset.
Passing dataset is useful for private mirrors and offline tests. When it
is omitted, the corresponding TorchVision dataset is opened at root.
normalization='source' applies the conventional dataset statistics;
'unit' keeps values in [0, 1].
Source code in src/silva_networks/source_data.py
| def load_vision_source_subset(
name: Literal["CIFAR10", "MNIST", "SVHN"],
*,
root: str | Path = "data",
train: bool = True,
samples_per_class: int = 4,
seed: int = 0,
image_size: tuple[int, int] | None = None,
normalization: VisionNormalization = "unit",
download: bool = False,
dataset: Any | None = None,
) -> SILVAVisionSourceSubset:
"""Load a deterministic class-balanced vision subset.
Passing ``dataset`` is useful for private mirrors and offline tests. When it
is omitted, the corresponding TorchVision dataset is opened at ``root``.
``normalization='source'`` applies the conventional dataset statistics;
``'unit'`` keeps values in ``[0, 1]``.
"""
if samples_per_class < 1:
raise ValueError("samples_per_class must be positive")
if normalization not in {"unit", "source", "none"}:
raise ValueError("normalization must be 'unit', 'source', or 'none'")
info = source_dataset_info(name)
loaded = (
dataset
if dataset is not None
else load_torchvision_dataset(name, root=root, train=train, download=download)
)
labels = _dataset_labels(loaded)
selected = _stratified_indices(labels, samples_per_class, seed)
images: list[Tensor] = []
selected_labels: list[int] = []
for index in selected:
image, label = loaded[index][:2]
images.append(_image_tensor(image))
selected_labels.append(int(torch.as_tensor(label).item()))
batch = torch.stack(images)
if image_size is not None and tuple(batch.shape[-2:]) != tuple(image_size):
batch = F.interpolate(batch, size=image_size, mode="bilinear", align_corners=False)
batch, normalization_step = _normalize_vision(batch, name, normalization)
target = torch.tensor(selected_labels, dtype=torch.long)
preprocessing = ["deterministic class-balanced selection", normalization_step]
if image_size is not None:
preprocessing.append(f"bilinear resize to {image_size[0]}x{image_size[1]}")
receipt = _receipt(
info,
split="train" if train else "test",
version=_package_version("torchvision"),
seed=seed,
indices=selected,
tensors=(batch, target),
preprocessing=preprocessing,
adapter="load_vision_source_subset",
)
return SILVAVisionSourceSubset(batch, target, receipt)
|
load_planetoid_source_subset
load_planetoid_source_subset(name='Cora', *, root='data/planetoid', subset_nodes=None, seed=0, download=False, dataset=None)
Load a Planetoid graph or a deterministic induced teaching subset.
subset_nodes=None preserves the official full transductive graph and
masks. A compact induced graph is intended for executable tutorials only;
the receipt records its original node ids.
Source code in src/silva_networks/source_data.py
| def load_planetoid_source_subset(
name: Literal["Cora", "CiteSeer", "PubMed"] = "Cora",
*,
root: str | Path = "data/planetoid",
subset_nodes: int | None = None,
seed: int = 0,
download: bool = False,
dataset: Any | None = None,
) -> SILVAGraphSourceSubset:
"""Load a Planetoid graph or a deterministic induced teaching subset.
``subset_nodes=None`` preserves the official full transductive graph and
masks. A compact induced graph is intended for executable tutorials only;
the receipt records its original node ids.
"""
info = source_dataset_info(name)
loaded = dataset
if loaded is None:
try:
from torch_geometric.datasets import Planetoid
except ImportError as exc:
raise ImportError(
"Install the benchmark extra before loading Planetoid data: "
'python -m pip install "silva-networks[benchmarks]"'
) from exc
dataset_root = Path(root) / name
if not download and not _planetoid_present(dataset_root):
raise FileNotFoundError(
f"{name} was not found at {dataset_root}. Download it explicitly "
"or call with download=True."
)
loaded = Planetoid(root=str(root), name=name)
data = loaded if hasattr(loaded, "x") else loaded[0]
graph = pyg_data_to_silva_graph(data)
nodes = graph.num_entities
train_mask = _mask_from_data(data, "train_mask", nodes)
validation_mask = _mask_from_data(data, "val_mask", nodes)
test_mask = _mask_from_data(data, "test_mask", nodes)
node_ids = torch.arange(nodes, dtype=torch.long)
preprocessing = ["source node features", "source edges", "source split masks"]
if subset_nodes is not None:
if subset_nodes < 3:
raise ValueError("subset_nodes must be at least three")
if subset_nodes > nodes:
raise ValueError("subset_nodes cannot exceed the number of graph nodes")
node_ids = _connected_subset_nodes(
graph.edge_index,
(train_mask, validation_mask, test_mask),
subset_nodes,
seed,
)
graph, train_mask, validation_mask, test_mask = _induced_graph(
graph, node_ids, train_mask, validation_mask, test_mask
)
preprocessing.append(
"deterministic connected induced subset; teaching protocol, not source benchmark"
)
selected = tuple(int(value) for value in node_ids.tolist())
tensors = (
graph.x,
graph.edge_index if graph.edge_index is not None else torch.empty(2, 0),
graph.y if graph.y is not None else torch.empty(0),
train_mask,
validation_mask,
test_mask,
)
receipt = _receipt(
info,
split="public Planetoid masks",
version=_package_version("torch-geometric"),
seed=seed,
indices=selected,
tensors=tensors,
preprocessing=preprocessing,
adapter="load_planetoid_source_subset",
)
graph.metadata = {
**(graph.metadata or {}),
"dataset": name,
"node_ids": selected,
"subset_protocol": "full" if subset_nodes is None else "induced teaching subset",
}
return SILVAGraphSourceSubset(graph, train_mask, validation_mask, test_mask, node_ids, receipt)
|
normalized_graph_operator
normalized_graph_operator(edge_index, num_nodes, *, add_self_loops=True, dense=True, dtype=torch.float32, device=None)
Build the symmetric normalized operator D^{-1/2} A D^{-1/2}.
Source code in src/silva_networks/source_data.py
| def normalized_graph_operator(
edge_index: Tensor,
num_nodes: int,
*,
add_self_loops: bool = True,
dense: bool = True,
dtype: torch.dtype = torch.float32,
device: str | torch.device | None = None,
) -> Tensor:
r"""Build the symmetric normalized operator ``D^{-1/2} A D^{-1/2}``."""
if num_nodes < 1:
raise ValueError("num_nodes must be positive")
edges = torch.as_tensor(edge_index, dtype=torch.long, device=device)
if edges.dim() != 2 or edges.shape[0] != 2:
raise ValueError("edge_index must have shape (2, edges)")
if edges.numel() and (int(edges.min()) < 0 or int(edges.max()) >= num_nodes):
raise ValueError("edge_index contains a node outside [0, num_nodes)")
if add_self_loops:
diagonal = torch.arange(num_nodes, device=edges.device)
edges = torch.cat((edges, torch.stack((diagonal, diagonal))), dim=1)
values = torch.ones(edges.shape[1], dtype=dtype, device=edges.device)
adjacency = torch.sparse_coo_tensor(
edges, values, (num_nodes, num_nodes), device=edges.device
).coalesce()
row, col = adjacency.indices()
degree = torch.zeros(num_nodes, dtype=dtype, device=edges.device)
degree.scatter_add_(0, row, adjacency.values())
normalized_values = adjacency.values() * degree[row].clamp_min(1).rsqrt()
normalized_values = normalized_values * degree[col].clamp_min(1).rsqrt()
operator = torch.sparse_coo_tensor(
adjacency.indices(), normalized_values, adjacency.shape, device=edges.device
).coalesce()
return operator.to_dense() if dense else operator
|
load_optical_flow_source_subset
load_optical_flow_source_subset(name, *, root, split='train', index=0, pass_name='clean', image_size=None, dataset=None)
Load one local Sintel, KITTI, or FlyingChairs pair without downloading.
Source code in src/silva_networks/source_data.py
| def load_optical_flow_source_subset(
name: FlowDatasetName,
*,
root: str | Path,
split: str = "train",
index: int = 0,
pass_name: str = "clean",
image_size: tuple[int, int] | None = None,
dataset: Any | None = None,
) -> SILVAFlowSourceSubset:
"""Load one local Sintel, KITTI, or FlyingChairs pair without downloading."""
info = source_dataset_info(name)
loaded = dataset
if loaded is None:
try:
from torchvision import datasets
except ImportError as exc:
raise ImportError(
"Install the vision extra before loading optical-flow data: "
'python -m pip install "silva-networks[vision]"'
) from exc
dataset_type = getattr(datasets, name)
kwargs: dict[str, Any] = {"root": str(root), "split": split}
if name == "Sintel":
kwargs["pass_name"] = pass_name
loaded = dataset_type(**kwargs)
if not 0 <= index < len(loaded):
raise IndexError(f"index {index} is outside a dataset of length {len(loaded)}")
sample = loaded[index]
if len(sample) not in {3, 4}:
raise ValueError("optical-flow samples must contain images, flow, and optional mask")
frame1, frame2, flow = sample[:3]
valid = sample[3] if len(sample) == 4 else None
first = _image_tensor(frame1).unsqueeze(0)
second = _image_tensor(frame2).unsqueeze(0)
flow_tensor = None if flow is None else _flow_tensor(flow).unsqueeze(0)
valid_tensor = None if valid is None else torch.as_tensor(np.asarray(valid)).bool().unsqueeze(0)
original_size = tuple(first.shape[-2:])
if image_size is not None and original_size != tuple(image_size):
first = F.interpolate(first, image_size, mode="bilinear", align_corners=False)
second = F.interpolate(second, image_size, mode="bilinear", align_corners=False)
if flow_tensor is not None:
flow_tensor = F.interpolate(
flow_tensor, image_size, mode="bilinear", align_corners=False
)
flow_tensor[:, 0] *= image_size[1] / original_size[1]
flow_tensor[:, 1] *= image_size[0] / original_size[0]
if valid_tensor is not None:
valid_tensor = (
F.interpolate(valid_tensor.float().unsqueeze(1), image_size, mode="nearest")
.squeeze(1)
.bool()
)
preprocessing = ["convert images to channel-first unit tensors"]
if image_size is not None:
preprocessing.append(
f"resize images and flow to {image_size[0]}x{image_size[1]} with vector rescaling"
)
tensors = [first, second]
if flow_tensor is not None:
tensors.append(flow_tensor)
if valid_tensor is not None:
tensors.append(valid_tensor)
receipt = _receipt(
info,
split=f"{split}:{pass_name}" if name == "Sintel" else split,
version=_package_version("torchvision"),
seed=0,
indices=(index,),
tensors=tensors,
preprocessing=preprocessing,
adapter="load_optical_flow_source_subset",
)
return SILVAFlowSourceSubset(first, second, flow_tensor, valid_tensor, receipt)
|
load_public_motion_subset
load_public_motion_subset(video_path, *, frame_indices=(100, 101), image_size=(128, 192))
Load selected frames from the small real-motion tutorial video.
The pair has no ground-truth flow and is therefore a qualitative mechanism
check. It is deliberately distinguished from Sintel and KITTI evaluation.
Source code in src/silva_networks/source_data.py
| def load_public_motion_subset(
video_path: str | Path,
*,
frame_indices: Sequence[int] = (100, 101),
image_size: tuple[int, int] | None = (128, 192),
) -> SILVAFlowSourceSubset:
"""Load selected frames from the small real-motion tutorial video.
The pair has no ground-truth flow and is therefore a qualitative mechanism
check. It is deliberately distinguished from Sintel and KITTI evaluation.
"""
if len(frame_indices) < 2:
raise ValueError("frame_indices must contain at least two entries")
indices = tuple(int(value) for value in frame_indices)
if min(indices) < 0:
raise ValueError("frame indices must be nonnegative")
path = Path(video_path)
if not path.exists():
raise FileNotFoundError(path)
try:
import av
except ImportError as exc:
raise ImportError("Install PyAV before reading a video source.") from exc
requested = set(indices)
decoded: dict[int, Tensor] = {}
with av.open(str(path)) as container:
stream = container.streams.video[0]
stream.codec_context.thread_count = 1
fps = str(stream.average_rate or "unknown")
for frame_index, frame in enumerate(container.decode(stream)):
if frame_index in requested:
array = frame.to_ndarray(format="rgb24")
decoded[frame_index] = torch.from_numpy(array).permute(2, 0, 1)
if frame_index >= max(indices):
break
missing = sorted(requested.difference(decoded))
if missing:
raise IndexError(f"video ended before requested frames {missing}")
selected = torch.stack([decoded[index] for index in indices]).float() / 255.0
if image_size is not None and tuple(selected.shape[-2:]) != tuple(image_size):
selected = F.interpolate(selected, image_size, mode="bilinear", align_corners=False)
first = selected[:-1]
second = selected[1:]
info = source_dataset_info("PublicBasketballMotion")
preprocessing = ["selected consecutive real-video frames", "scaled uint8 values to [0, 1]"]
if image_size is not None:
preprocessing.append(f"bilinear resize to {image_size[0]}x{image_size[1]}")
receipt = _receipt(
info,
split=f"frames at {fps} fps",
version=_package_version("av"),
seed=0,
indices=indices,
tensors=(first, second),
preprocessing=preprocessing,
adapter="load_public_motion_subset",
)
return SILVAFlowSourceSubset(first, second, None, None, receipt)
|
load_darcy_source_subset
load_darcy_source_subset(path, *, samples=None, seed=0, input_key='x', target_key='y')
Load a deterministic Darcy subset from a local .pt or .npz file.
This format-neutral boundary lets a full experiment use the official
NeuralOperator loader, a source archive, or a private mirror without
changing the SILVA operator. Arrays may use x/y, inputs/targets,
coeff/solution, or caller-supplied keys.
Source code in src/silva_networks/source_data.py
| def load_darcy_source_subset(
path: str | Path,
*,
samples: int | None = None,
seed: int = 0,
input_key: str = "x",
target_key: str = "y",
) -> SILVAOperatorSourceSubset:
"""Load a deterministic Darcy subset from a local ``.pt`` or ``.npz`` file.
This format-neutral boundary lets a full experiment use the official
NeuralOperator loader, a source archive, or a private mirror without
changing the SILVA operator. Arrays may use ``x/y``, ``inputs/targets``,
``coeff/solution``, or caller-supplied keys.
"""
source_path = Path(path)
if not source_path.exists():
raise FileNotFoundError(source_path)
if source_path.suffix == ".pt":
payload = torch.load(source_path, map_location="cpu", weights_only=True)
elif source_path.suffix == ".npz":
with np.load(source_path) as archive:
payload = {key: archive[key] for key in archive.files}
else:
raise ValueError("Darcy source files must use .pt or .npz")
inputs, targets = _operator_arrays(payload, input_key, target_key)
if inputs.shape[0] != targets.shape[0]:
raise ValueError("Darcy inputs and targets must have the same sample count")
count = inputs.shape[0] if samples is None else samples
if count < 1 or count > inputs.shape[0]:
raise ValueError("samples must be between one and the available sample count")
generator = torch.Generator().manual_seed(seed)
indices = torch.randperm(inputs.shape[0], generator=generator)[:count]
selected_inputs = inputs[indices].float()
selected_targets = targets[indices].float()
info = source_dataset_info("DarcyFlowSmall")
receipt = _receipt(
info,
split="local source archive",
version="source file",
seed=seed,
indices=tuple(int(value) for value in indices.tolist()),
tensors=(selected_inputs, selected_targets),
preprocessing=("preserve source field values", "deterministic sample selection"),
adapter="load_darcy_source_subset",
)
return SILVAOperatorSourceSubset(selected_inputs, selected_targets, receipt)
|
save_source_snapshot
save_source_snapshot(path, *, tensors, receipt)
Write a compact source subset while preserving its tensor hash.
Source code in src/silva_networks/source_data.py
| def save_source_snapshot(
path: str | Path,
*,
tensors: dict[str, Tensor],
receipt: SourceDataReceipt,
) -> Path:
"""Write a compact source subset while preserving its tensor hash."""
if not tensors:
raise ValueError("a source snapshot must contain at least one tensor")
values = {
name: torch.as_tensor(value).detach().cpu() for name, value in tensors.items()
}
content_hash = _tensor_sha256(tuple(values.values()))
if content_hash != receipt.content_sha256:
raise ValueError(
"snapshot tensors do not match the source receipt content checksum"
)
target = Path(path)
target.parent.mkdir(parents=True, exist_ok=True)
torch.save(
{
"format_version": 1,
"tensor_keys": list(values),
"tensors": values,
"receipt": receipt.as_dict(),
},
target,
)
return target
|
load_source_snapshot
load_source_snapshot(path, *, verify=True)
Load a compact source subset and verify its receipt checksum.
Source code in src/silva_networks/source_data.py
| def load_source_snapshot(
path: str | Path, *, verify: bool = True
) -> SILVASourceSnapshot:
"""Load a compact source subset and verify its receipt checksum."""
source_path = Path(path)
if not source_path.exists():
raise FileNotFoundError(source_path)
payload = torch.load(source_path, map_location="cpu", weights_only=True)
if payload.get("format_version") != 1:
raise ValueError("unsupported source snapshot format")
keys = payload.get("tensor_keys")
tensors = payload.get("tensors")
receipt_values = payload.get("receipt")
if not isinstance(keys, list) or not isinstance(tensors, dict):
raise TypeError("source snapshot tensor metadata is incomplete")
if not isinstance(receipt_values, dict):
raise TypeError("source snapshot receipt is missing")
ordered = {str(key): torch.as_tensor(tensors[key]) for key in keys}
normalized_receipt = dict(receipt_values)
normalized_receipt["selected_indices"] = tuple(
normalized_receipt["selected_indices"]
)
normalized_receipt["preprocessing"] = tuple(
normalized_receipt["preprocessing"]
)
receipt = SourceDataReceipt(**normalized_receipt)
if verify and _tensor_sha256(tuple(ordered.values())) != receipt.content_sha256:
raise ValueError("source snapshot content checksum does not match its receipt")
return SILVASourceSnapshot(ordered, receipt)
|
available_bundled_source_snapshots
available_bundled_source_snapshots()
Return compact attributed snapshots shipped with the package.
Source code in src/silva_networks/source_data.py
| def available_bundled_source_snapshots() -> tuple[str, ...]:
"""Return compact attributed snapshots shipped with the package."""
return tuple(_BUNDLED_SOURCE_SNAPSHOTS)
|
load_bundled_source_snapshot
load_bundled_source_snapshot(name, *, verify=True)
Load a packaged compact snapshot by registry name.
The returned tensors use the same format and checksum verification as
:func:load_source_snapshot. These compact records validate mechanisms;
they are not replacements for official complete benchmark splits.
Source code in src/silva_networks/source_data.py
| def load_bundled_source_snapshot(
name: str, *, verify: bool = True
) -> SILVASourceSnapshot:
"""Load a packaged compact snapshot by registry name.
The returned tensors use the same format and checksum verification as
:func:`load_source_snapshot`. These compact records validate mechanisms;
they are not replacements for official complete benchmark splits.
"""
try:
filename = _BUNDLED_SOURCE_SNAPSHOTS[name]
except KeyError as exc:
choices = ", ".join(_BUNDLED_SOURCE_SNAPSHOTS)
raise KeyError(f"unknown bundled source snapshot {name!r}; choose from {choices}") from exc
resource = importlib.resources.files("silva_networks").joinpath(
"source_snapshots", filename
)
with importlib.resources.as_file(resource) as path:
return load_source_snapshot(path, verify=verify)
|
Where to Go Next