Experiment Protocol API

Module: silva_networks.experiment_protocols

The protocol registry gives all 64 canonical families explicit smoke, workstation, and full-scale routes. Resource ranges are planning inputs; report observed values through the evidence API after execution.

Operational Contract

This API surface connects three-tier experiment protocols for every canonical family to the same SILVA experiment contract used by the learning pages and notebooks. Its central relation is

\[ P_f=\{P_{f,\mathrm{smoke}},P_{f,\mathrm{workstation}},P_{f,\mathrm{full}}\} \]

Part	What must remain inspectable
State	the selected family construction together with its tier-specific data and runtime contract.
Condition	source relation, data route, preprocessing, seeds, metrics, resources, and acceptance checks.
Diagnostic	validated protocol fields followed by measured task, solver, runtime, memory, and failure records.
Replacement point	data adapter, task lifecycle hook, model options, runtime placement, or acceptance rule.
Scale axes	sample cap, epochs, seeds, resolution, batch policy, accelerators, storage, and wall time.

The relevant method lineage is recorded in the SILVA contract [1] and PDEBench route [93]. Those references define the source mechanisms; this API exposes them through SILVA objects so a reader can inspect, replace, solve, differentiate, and scale the construction.

Complete Compact Study

Run the complete repository program below from the project root. The page uses the same file that is exercised by the test suite, so the displayed call is not an isolated fragment.

"""Create a repeated-measurement record and inspect a scale protocol."""

from silva_networks import run_silva_evidence, silva_family_experiment_protocol


def compact_trial(seed: int) -> dict[str, object]:
    return {
        "metrics": {"absolute_error": 0.04 + 0.005 * seed},
        "residual": 1e-7 * (seed + 1),
        "evaluations": 7 + seed,
        "converged": True,
    }


def main() -> None:
    report = run_silva_evidence(
        "silva_implicit_spatiotemporal",
        "analytic diffusion",
        compact_trial,
        seeds=(0, 1, 2),
        configuration={"dt": 0.2, "steps": 4},
        data_receipt={"generator": "periodic diffusion", "samples": 16},
        bootstrap_samples=200,
    )
    print("mean error", report.summaries[0].mean)

    protocol = silva_family_experiment_protocol("im_pindiff")
    for tier in protocol.tiers:
        print(tier.tier, tier.dataset.name, tier.resources.accelerator_count)


if __name__ == "__main__":
    main()

python examples/evidence_and_protocols.py

Measured Compact Output

mean error 0.045
smoke analytic ODE/PDE trajectory CPU or 1 accelerator
workstation PDEBench subset 1 accelerator
full PDEBench source task 1-8 accelerators

Interpret the Output

The same family exposes compact, subset, and complete-source routes. Resource ranges describe intended capacity and must be replaced by observed measurements in a completed result record.

For a controlled experiment, retain the compact call as a regression case and change one scale axis at a time. Record the resolved constructor, data source and split, preprocessing, seed, forward and backward solver settings, task metric, normalized residual, iteration count, runtime, peak memory, and any failed convergence case. A larger run becomes evidence only when its own resolved configuration and outputs are archived; the compact output above is evidence for the executable mechanism and its stated invariants.

Named data route and the split/access contract that must be retained.

Source code in src/silva_networks/experiment_protocols.py

@dataclass(frozen=True)
class SILVADatasetRoute:
    """Named data route and the split/access contract that must be retained."""

    name: str
    source_url: str
    split: str
    access: str
    expected_storage: str

Conservative planning range for one protocol tier.

Source code in src/silva_networks/experiment_protocols.py

@dataclass(frozen=True)
class SILVAResourceEstimate:
    """Conservative planning range for one protocol tier."""

    accelerator_count: str
    accelerator_memory: str
    host_memory: str
    storage: str
    wall_time: str
    note: str

One complete scale rung for a family experiment.

Source code in src/silva_networks/experiment_protocols.py

@dataclass(frozen=True)
class SILVAExecutionTier:
    """One complete scale rung for a family experiment."""

    tier: ProtocolTier
    evidence_target: str
    dataset: SILVADatasetRoute
    sample_limit: int | None
    epochs: int
    seeds: tuple[int, ...]
    model_options: dict[str, Any]
    runtime_options: dict[str, Any]
    resources: SILVAResourceEstimate
    metrics: tuple[str, ...]
    acceptance_checks: tuple[str, ...]
    command: str

Compact-to-source-scale execution contract for one SILVA family.

Source code in src/silva_networks/experiment_protocols.py

@dataclass(frozen=True)
class SILVAFamilyExperimentProtocol:
    """Compact-to-source-scale execution contract for one SILVA family."""

    family: str
    profile: str
    source_relation: str
    references: tuple[int, ...]
    repositories: tuple[str, ...]
    data_sources: tuple[str, ...]
    preprocessing: tuple[str, ...]
    required_artifacts: tuple[str, ...]
    tiers: tuple[SILVAExecutionTier, ...]

    def tier(self, name: ProtocolTier) -> SILVAExecutionTier:
        """Return one named execution tier."""

        return next(item for item in self.tiers if item.tier == name)

    def validate(self) -> tuple[str, ...]:
        """Return completeness errors for this protocol."""

        errors: list[str] = []
        if tuple(item.tier for item in self.tiers) != ("smoke", "workstation", "full"):
            errors.append("tiers must be ordered smoke, workstation, full")
        for item in self.tiers:
            if not item.dataset.name or not item.dataset.source_url:
                errors.append(f"{item.tier}: incomplete dataset route")
            if not item.metrics or not item.acceptance_checks or not item.command:
                errors.append(f"{item.tier}: incomplete execution contract")
            if item.tier == "full" and item.sample_limit is not None:
                errors.append("full: sample_limit must be null")
        return tuple(errors)

    def as_dict(self) -> dict[str, Any]:
        """Return a stable JSON-compatible representation."""

        return _jsonable(asdict(self))

    def write_json(self, path: str | Path) -> Path:
        """Write this protocol as indented JSON."""

        destination = Path(path)
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_text(json.dumps(self.as_dict(), indent=2) + "\n", encoding="utf-8")
        return destination

as_dict

as_dict()

Return a stable JSON-compatible representation.

Source code in src/silva_networks/experiment_protocols.py

def as_dict(self) -> dict[str, Any]:
    """Return a stable JSON-compatible representation."""

    return _jsonable(asdict(self))

tier

tier(name)

Return one named execution tier.

Source code in src/silva_networks/experiment_protocols.py

def tier(self, name: ProtocolTier) -> SILVAExecutionTier:
    """Return one named execution tier."""

    return next(item for item in self.tiers if item.tier == name)

validate

validate()

Return completeness errors for this protocol.

Source code in src/silva_networks/experiment_protocols.py

def validate(self) -> tuple[str, ...]:
    """Return completeness errors for this protocol."""

    errors: list[str] = []
    if tuple(item.tier for item in self.tiers) != ("smoke", "workstation", "full"):
        errors.append("tiers must be ordered smoke, workstation, full")
    for item in self.tiers:
        if not item.dataset.name or not item.dataset.source_url:
            errors.append(f"{item.tier}: incomplete dataset route")
        if not item.metrics or not item.acceptance_checks or not item.command:
            errors.append(f"{item.tier}: incomplete execution contract")
        if item.tier == "full" and item.sample_limit is not None:
            errors.append("full: sample_limit must be null")
    return tuple(errors)

write_json

write_json(path)

Write this protocol as indented JSON.

Source code in src/silva_networks/experiment_protocols.py

def write_json(self, path: str | Path) -> Path:
    """Write this protocol as indented JSON."""

    destination = Path(path)
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_text(json.dumps(self.as_dict(), indent=2) + "\n", encoding="utf-8")
    return destination

Build the complete three-tier protocol for a canonical family or alias.

Source code in src/silva_networks/experiment_protocols.py

def silva_family_experiment_protocol(family: str) -> SILVAFamilyExperimentProtocol:
    """Build the complete three-tier protocol for a canonical family or alias."""

    key = canonical_silva_family(family)
    dossier = silva_experiment_dossier(key)
    profile = _FAMILY_PROFILE[key]
    routes = _FAMILY_ROUTES.get(key, _PROFILE_ROUTES[profile])
    resources = _RESOURCE_TABLE[profile]
    tiers: list[SILVAExecutionTier] = []
    specifications = (
        ("smoke", "compact-verified", 64, 2, (0,)),
        ("workstation", "subset-verified", 4096, 20, (0, 1, 2)),
        ("full", "source-scale-reproduced", None, 100, (0, 1, 2, 3, 4)),
    )
    for index, (tier, evidence, samples, epochs, seeds) in enumerate(specifications):
        model, runtime = _tier_options(key, tier)
        command = (
            "python experiments/reproduction/run_family_protocol.py "
            f"--family {key} --tier {tier} --work-dir runs/{key}/{tier}"
        )
        tiers.append(
            SILVAExecutionTier(
                tier=tier,
                evidence_target=evidence,
                dataset=routes[index],
                sample_limit=samples,
                epochs=epochs,
                seeds=seeds,
                model_options=model,
                runtime_options=runtime,
                resources=resources[index],
                metrics=dossier.metrics,
                acceptance_checks=dossier.stages[min(index + 3, 5)].acceptance_checks,
                command=command,
            )
        )
    return SILVAFamilyExperimentProtocol(
        family=key,
        profile=profile,
        source_relation=dossier.source_relation,
        references=dossier.paper_refs,
        repositories=dossier.repositories,
        data_sources=dossier.data_sources,
        preprocessing=dossier.preprocessing,
        required_artifacts=dossier.required_artifacts,
        tiers=tuple(tiers),
    )

Return three-tier protocols in canonical family order.

Source code in src/silva_networks/experiment_protocols.py

def all_silva_family_experiment_protocols() -> tuple[SILVAFamilyExperimentProtocol, ...]:
    """Return three-tier protocols in canonical family order."""

    return tuple(silva_family_experiment_protocol(name) for name in available_silva_families())

Return registry and protocol completeness errors.

Source code in src/silva_networks/experiment_protocols.py

def audit_silva_family_experiment_protocols() -> tuple[str, ...]:
    """Return registry and protocol completeness errors."""

    errors: list[str] = []
    expected = set(available_silva_families())
    if set(_FAMILY_PROFILE) != expected:
        for family in sorted(expected - set(_FAMILY_PROFILE)):
            errors.append(f"missing family protocol profile: {family}")
        for family in sorted(set(_FAMILY_PROFILE) - expected):
            errors.append(f"unknown family protocol profile: {family}")
    for protocol in all_silva_family_experiment_protocols():
        errors.extend(f"{protocol.family}: {error}" for error in protocol.validate())
    return tuple(errors)

Write one additive execution-protocol JSON file per family.

Source code in src/silva_networks/experiment_protocols.py

def write_silva_family_experiment_protocols(directory: str | Path) -> tuple[Path, ...]:
    """Write one additive execution-protocol JSON file per family."""

    root = Path(directory)
    return tuple(
        protocol.write_json(root / f"{protocol.family}.json")
        for protocol in all_silva_family_experiment_protocols()
    )

Where to Go Next

Question	Page
What does each evidence target establish?	Evidence and Source-Scale Experiments
Where can I inspect all family dossiers?	Family Reproduction Dossiers
How do I execute a materialized protocol?	Run Everything
Which objects record the measured result?	Evidence API