tetration 0.1.9

Chunked tensor .tet format, tet CLI, and optional C ABI (tetration-ffi)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

"""Committed bench workload contract and post-generate verification."""

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from benchmark.case import BenchCase
from generate.constants import CHUNK_ELEMS
from generate.util import iter_slabs

SPEC_PATH = Path(__file__).resolve().parent / "spec.json"


@dataclass(frozen=True)
class CaseSpec:
    element_count: int
    mean: float
    mean_tolerance: float


def load_spec() -> dict[str, Any]:
    return json.loads(SPEC_PATH.read_text(encoding="utf-8"))


def spec_sha256() -> str:
    raw = SPEC_PATH.read_bytes()
    return hashlib.sha256(raw).hexdigest()


def case_key(case: BenchCase) -> str:
    return f"{case.format}/{case.tier}"


def case_spec(case: BenchCase) -> CaseSpec:
    data = load_spec()["cases"][case_key(case)]
    return CaseSpec(
        element_count=int(data["element_count"]),
        mean=float(data["mean"]),
        mean_tolerance=float(data["mean_tolerance"]),
    )


def fingerprint_case(case: BenchCase) -> tuple[int, float]:
    """Stream `data` once: element count + mean (format-specific reader)."""
    match case.format:
        case "h5":
            return _fingerprint_h5(case)
        case "netcdf":
            return _fingerprint_netcdf(case)
        case "zarr":
            return _fingerprint_zarr(case)
    raise ValueError(case.format)


def verify_case(case: BenchCase) -> None:
    spec = case_spec(case)
    count, mean = fingerprint_case(case)
    if count != spec.element_count:
        raise RuntimeError(
            f"{case_key(case)}: element count {count} != spec {spec.element_count}"
        )
    if abs(mean - spec.mean) > spec.mean_tolerance:
        raise RuntimeError(
            f"{case_key(case)}: mean {mean:.6g} outside "
            f"{spec.mean} ± {spec.mean_tolerance}"
        )


def _fingerprint_h5(case: BenchCase) -> tuple[int, float]:
    import h5py

    with h5py.File(case.src, "r") as f:
        d = f["data"]
        length = int(d.shape[0])
        total = 0.0
        for start, end in iter_slabs(length):
            total += float(d[start:end].sum(dtype=float))
        return length, total / length if length else 0.0


def _fingerprint_netcdf(case: BenchCase) -> tuple[int, float]:
    import netCDF4 as nc

    with nc.Dataset(case.src, "r") as ds:
        var = ds.variables["data"]
        length = int(var.shape[0])
        total = 0.0
        for start, end in iter_slabs(length):
            total += float(var[start:end].sum(dtype=float))
        return length, total / length if length else 0.0


def _fingerprint_zarr(case: BenchCase) -> tuple[int, float]:
    import zarr

    arr = zarr.open_group(str(case.src), mode="r")["data"]
    length = int(arr.shape[0])
    chunk_len = int(arr.chunks[0])
    total = 0.0
    for start in range(0, length, chunk_len):
        end = min(start + chunk_len, length)
        total += float(arr[start:end].sum(dtype=float))
    return length, total / length if length else 0.0


def spec_summary() -> dict[str, Any]:
    spec = load_spec()
    return {
        "schema_version": spec["schema_version"],
        "spec_sha256": spec_sha256(),
        "chunk_elems": CHUNK_ELEMS,
    }