pycsamt.ai.data.manifest#

Immutable, versioned provenance manifests for generated EM datasets.

A manifest identifies how a dataset was generated, which geological realizations belong to each split, and which external artifacts belong to the dataset. It contains no training arrays itself. All hashes use SHA-256 and all serialized content uses deterministic canonical JSON.

Functions

`canonical_hash`(value)	Return the SHA-256 digest of deterministic canonical JSON.
`sha256_file`(path, *[, chunk_size])	Hash a file without loading the complete artifact into memory.

Classes

`ArtifactRecord`(sha256[, size_bytes, ...])	Integrity metadata for one external dataset artifact.
`DatasetManifest`(dataset_id, generator, ...)	Identify a generated dataset and its complete reproducibility state.

class pycsamt.ai.data.manifest.ArtifactRecord(sha256, size_bytes=None, media_type=None, role=None)[source]

Bases: object

Integrity metadata for one external dataset artifact.

Parameters:

sha256 (str) – Lowercase 64-character SHA-256 digest of the complete file.
size_bytes (int or None, optional) – Exact file size. When present it is checked before hashing.
media_type (str or None, optional) – MIME type such as "application/x-npz".
role (str or None, optional) – Human-readable role such as "responses" or "models".

Examples

>>> record = ArtifactRecord("0" * 64, size_bytes=1024, role="models")
>>> record.algorithm
'sha256'

sha256: str

size_bytes: int | None = None

media_type: str | None = None

role: str | None = None

property algorithm: str[source]

Return the checksum algorithm identifier.

Returns:: Always "sha256" for the current artifact schema.
Return type:: str

Examples

>>> ArtifactRecord("a" * 64).algorithm
'sha256'

classmethod from_file(path, *, media_type=None, role=None, chunk_size=1048576)[source]

Create an integrity record from an existing file.

Parameters:

path (str or pathlib.Path) – Existing regular file.
media_type (str, optional) – Optional descriptive metadata.
role (str, optional) – Optional descriptive metadata.
chunk_size (int, default=1048576) – Bytes read per hashing iteration.

Returns:

Digest and exact file size captured from disk.

Return type:

ArtifactRecord

Examples

>>> from tempfile import TemporaryDirectory
>>> with TemporaryDirectory() as directory:
...     path = Path(directory) / "models.npz"
...     _ = path.write_bytes(b"model-data")
...     record = ArtifactRecord.from_file(path, role="models")
>>> record.size_bytes
10

to_dict()[source]

Return a JSON-serializable artifact record.

Returns:: Versioned checksum, size, media type, and role fields.
Return type:: dict

Examples

>>> ArtifactRecord("f" * 64, size_bytes=2).to_dict()["size_bytes"]
2

classmethod from_dict(data)[source]

Restore a validated artifact record.

Parameters:: data (mapping) – Versioned state returned by to_dict().
Returns:: Immutable integrity record.
Return type:: ArtifactRecord
Raises:: ValueError – If the schema or checksum algorithm is unsupported.

Examples

>>> state = ArtifactRecord("1" * 64).to_dict()
>>> ArtifactRecord.from_dict(state).sha256 == "1" * 64
True

class pycsamt.ai.data.manifest.DatasetManifest(dataset_id, generator, generator_version, configuration, split, sample_count, created_utc=None, artifacts=<factory>, schema_version=2)[source]

Bases: object

Identify a generated dataset and its complete reproducibility state.

Parameters:

dataset_id (str) – Portable identifier containing letters, digits, dots, underscores, or hyphens. It must start with a letter or digit.
generator (str) – Fully qualified generator name and its version or source revision.
generator_version (str) – Fully qualified generator name and its version or source revision.
configuration (mapping) – Finite JSON-compatible generator configuration. It is recursively copied and frozen.
split (RealizationSplit) – Disjoint realization-level train/validation/test assignment.
sample_count (int) – Number of samples represented by the dataset.
created_utc (str or None, optional) – Timezone-aware ISO-8601 creation time. It is normalized to UTC.
artifacts (mapping, optional) – Normalized relative paths mapped to ArtifactRecord objects or their serialized dictionaries.
schema_version (int, default=2) – Manifest format version. New manifests use version 2.

Examples

>>> split = RealizationSplit(("r1", "r2"), ("r3",), ("r4",), seed=7)
>>> manifest = DatasetManifest(
...     dataset_id="willy-2d-v1",
...     generator="pycsamt.ai.geology.correlated2d",
...     generator_version="0.1.0",
...     configuration={"seed": 7, "correlation_m": [1000, 100]},
...     split=split,
...     sample_count=4,
... )
>>> len(manifest.configuration_hash)
64

dataset_id: str

generator: str

generator_version: str

configuration: Mapping[str, Any]

split: RealizationSplit

sample_count: int

created_utc: str | None = None

artifacts: Mapping[str, ArtifactRecord | Mapping[str, Any] | str]

schema_version: int = 2

property configuration_hash: str[source]

Return the canonical configuration digest.

Returns:: SHA-256 digest of generator configuration only.
Return type:: str

Examples

>>> split = RealizationSplit(("r1",), (), ())
>>> m = DatasetManifest("d", "g", "1", {"seed": 0}, split, 1)
>>> m.configuration_hash == canonical_hash({"seed": 0})
True

property manifest_hash: str[source]

Return a digest of the complete serialized manifest.

Returns:: SHA-256 digest covering configuration, split, timestamps, and all artifact records.
Return type:: str

Examples

>>> split = RealizationSplit(("r1",), (), ())
>>> m = DatasetManifest("d", "g", "1", {}, split, 1)
>>> len(m.manifest_hash)
64

property realization_count: int[source]

Return the total number of split realizations.

Returns:: Length of the combined train, validation, and test ID sets.
Return type:: int

Examples

>>> split = RealizationSplit(("a", "b"), ("c",), ())
>>> DatasetManifest("d", "g", "1", {}, split, 3).realization_count
3

with_artifact(path, record)[source]

Return a copy containing or replacing one artifact record.

Parameters:

path (str) – Portable relative artifact path.
record (ArtifactRecord, mapping, or str) – Integrity record, serialized record, or SHA-256 digest.

Returns:

New immutable manifest; the original is unchanged.

Return type:

DatasetManifest

Examples

>>> split = RealizationSplit(("r1",), (), ())
>>> m = DatasetManifest("d", "g", "1", {}, split, 1)
>>> updated = m.with_artifact("data/models.npz", "a" * 64)
>>> list(updated.artifacts)
['data/models.npz']

verify_artifacts(root, *, paths=None, raise_on_error=False)[source]

Verify recorded artifact sizes and SHA-256 digests on disk.

Parameters:

root (str or pathlib.Path) – Directory against which relative artifact paths are resolved.
paths (sequence of str, optional) – Subset of recorded paths. By default all artifacts are checked.
raise_on_error (bool, default=False) – Raise on the first missing, size-mismatched, or hash-mismatched artifact instead of returning False for it.

Returns:

Normalized artifact paths mapped to verification results.

Return type:

dict

Raises:

KeyError – If a requested path is not recorded.
ValueError – If root is not a directory or verification fails while raise_on_error is true.

Examples

>>> from tempfile import TemporaryDirectory
>>> split = RealizationSplit(("r1",), (), ())
>>> with TemporaryDirectory() as directory:
...     root = Path(directory)
...     file = root / "data.bin"
...     _ = file.write_bytes(b"data")
...     record = ArtifactRecord.from_file(file)
...     manifest = DatasetManifest(
...         "d", "g", "1", {}, split, 1, artifacts={"data.bin": record}
...     )
...     result = manifest.verify_artifacts(root)
>>> result
{'data.bin': True}

to_dict()[source]

Return the complete schema-2 JSON representation.

Returns:: Mutable JSON-compatible copy including the configuration digest.
Return type:: dict

Examples

>>> split = RealizationSplit(("r1",), (), ())
>>> m = DatasetManifest("d", "g", "1", {}, split, 1)
>>> m.to_dict()["schema_version"]
2

write_json(path, *, overwrite=True)[source]

Write a deterministic, human-readable manifest file.

Parameters:

path (str or pathlib.Path) – Destination JSON file.
overwrite (bool, default=True) – Permit replacement of an existing file.

Returns:

Destination path.

Return type:

pathlib.Path

Raises:

FileExistsError – If the destination exists and overwrite is false.

Examples

>>> from tempfile import TemporaryDirectory
>>> split = RealizationSplit(("r1",), (), ())
>>> m = DatasetManifest("d", "g", "1", {}, split, 1)
>>> with TemporaryDirectory() as directory:
...     path = m.write_json(Path(directory) / "manifest.json")
...     loaded = DatasetManifest.read_json(path)
>>> loaded.manifest_hash == m.manifest_hash
True

classmethod from_dict(data)[source]

Restore a manifest and verify its recorded configuration digest.

Parameters:: data (mapping) – Schema-1 or schema-2 serialized manifest.
Returns:: Validated schema-2 runtime object.
Return type:: DatasetManifest
Raises:: ValueError – If the schema is unsupported, required content is invalid, or the recorded configuration hash does not match its configuration.

Examples

>>> split = RealizationSplit(("r1",), (), ())
>>> original = DatasetManifest("d", "g", "1", {"seed": 2}, split, 1)
>>> restored = DatasetManifest.from_dict(original.to_dict())
>>> restored.configuration_hash == original.configuration_hash
True

classmethod read_json(path)[source]

Read and validate a UTF-8 JSON manifest.

Parameters:

path (str or pathlib.Path) – Existing manifest file.

Returns:

Validated immutable manifest.

Return type:

DatasetManifest

Raises:

OSError – If the file cannot be read.
json.JSONDecodeError – If its content is not valid JSON.
ValueError – If decoded content violates the manifest contract.

Examples

>>> from tempfile import TemporaryDirectory
>>> split = RealizationSplit(("r1",), (), ())
>>> source = DatasetManifest("d", "g", "1", {}, split, 1)
>>> with TemporaryDirectory() as directory:
...     path = source.write_json(Path(directory) / "manifest.json")
...     loaded = DatasetManifest.read_json(path)
>>> loaded.dataset_id
'd'

pycsamt.ai.data.manifest.canonical_hash(value)[source]

Return the SHA-256 digest of deterministic canonical JSON.

Parameters:: value (Any) – Finite JSON-serializable value. Mapping keys are converted to strings, mappings are sorted recursively, and insignificant whitespace is removed before hashing.
Returns:: Lowercase 64-character hexadecimal SHA-256 digest.
Return type:: str
Raises:: ValueError – If value contains NaN, infinity, bytes, arrays, or another object that cannot be represented safely as JSON.

Examples

Mapping insertion order does not affect the digest:

>>> canonical_hash({"a": 1, "b": 2}) == canonical_hash({"b": 2, "a": 1})
True
>>> len(canonical_hash({"frequencies_hz": [100.0, 10.0]}))
64

pycsamt.ai.data.manifest.sha256_file(path, *, chunk_size=1048576)[source]

Hash a file without loading the complete artifact into memory.

Parameters:

path (str or pathlib.Path) – Existing regular file to hash.
chunk_size (int, default=1048576) – Positive number of bytes read per iteration.

Returns:

Lowercase hexadecimal SHA-256 digest.

Return type:

str

Raises:

ValueError – If chunk_size is not a positive integer.
OSError – If the file cannot be opened or read.

Examples

>>> from tempfile import TemporaryDirectory
>>> with TemporaryDirectory() as directory:
...     path = Path(directory) / "artifact.bin"
...     _ = path.write_bytes(b"pycsamt")
...     digest = sha256_file(path)
>>> len(digest)
64