pycsamt.ai.data.splits#

Deterministic, lineage-aware geological-realization dataset splits.

Stations, frequency patches, noise variants, and augmented copies derived from one geological model must never be divided across training and evaluation partitions. This module therefore splits unique realization IDs and can bind related IDs through explicit lineage labels.

Functions

`realization_folds`(realization_ids, *[, ...])	Build deterministic group-safe cross-validation test folds.
`split_realizations`(realization_ids, *[, ...])	Create a deterministic realization- or lineage-level random split.

Classes

RealizationSplit(train, validation, test[, ...])

Immutable train/validation/test assignment of geological realizations.

class pycsamt.ai.data.splits.RealizationSplit(train, validation, test, seed=None, lineage=<factory>, strategy='random')[source]

Bases: object

Immutable train/validation/test assignment of geological realizations.

Parameters:

train (sequence of str) – Unique realization identifiers. The three partitions must be disjoint, and training cannot be empty.
validation (sequence of str) – Unique realization identifiers. The three partitions must be disjoint, and training cannot be empty.
test (sequence of str) – Unique realization identifiers. The three partitions must be disjoint, and training cannot be empty.
seed (int or None, optional) – Random seed used to create the assignment. None records that the seed is unknown or intentionally nondeterministic.
lineage (mapping, optional) – Complete mapping from every realization ID to its parent geological lineage. IDs sharing a lineage must occupy the same partition.
strategy (str, default="random") – Human-readable splitting strategy identifier.

Examples

>>> split = RealizationSplit(("r1", "r2"), ("r3",), ("r4",), seed=7)
>>> split.sizes
{'train': 2, 'validation': 1, 'test': 1}
>>> split.partition("r3")
'validation'

train: tuple[str, ...]

validation: tuple[str, ...]

test: tuple[str, ...]

seed: int | None = None

lineage: Mapping[str, str]

strategy: str = 'random'

property all_ids: tuple[str, ...][source]

Return all realization IDs in partition order.

Returns:: Training IDs followed by validation IDs and test IDs.
Return type:: tuple of str

Examples

>>> RealizationSplit(("a",), ("b",), ("c",)).all_ids
('a', 'b', 'c')

property sizes: dict[str, int][source]

Return the number of realizations in each partition.

Returns:: train, validation, and test counts.
Return type:: dict

Examples

>>> RealizationSplit(("a", "b"), (), ("c",)).sizes["train"]
2

property fractions: dict[str, float][source]

Return realized partition fractions.

Returns:: Counts divided by total realization count. These may differ from requested targets when lineages contain multiple realizations.
Return type:: dict

Examples

>>> RealizationSplit(("a", "b"), ("c",), ("d",)).fractions
{'train': 0.5, 'validation': 0.25, 'test': 0.25}

property split_hash: str[source]

Return a deterministic SHA-256 digest of the assignment.

Returns:: Digest covering partitions, seed, lineage, and strategy.
Return type:: str

Examples

>>> len(RealizationSplit(("a",), (), ()).split_hash)
64

partition(realization_id)[source]

Return the partition containing a realization.

Parameters:: realization_id (str) – Exact realization identifier.
Returns:: Partition name.
Return type:: {“train”, “validation”, “test”}
Raises:: KeyError – If the identifier is unknown.

Examples

>>> RealizationSplit(("a",), ("b",), ()).partition("b")
'validation'

ids_for(partition)[source]

Return IDs assigned to a named partition.

Parameters:: partition ({"train", "validation", "test"}) – Partition to retrieve.
Returns:: Immutable realization IDs.
Return type:: tuple of str
Raises:: ValueError – If partition is unsupported.

Examples

>>> RealizationSplit(("a",), ("b",), ()).ids_for("train")
('a',)

mask(realization_ids, partition, *, unknown='raise')[source]

Build a Boolean sample mask from realization IDs.

Parameters:

realization_ids (sequence of str) – Per-sample realization IDs; duplicates are allowed because many stations or noise variants may belong to one realization.
partition ({"train", "validation", "test"}) – Partition selected as True.
unknown ({"raise", "false"}, default="raise") – Raise for IDs absent from the split or mark them false.

Returns:

One value per supplied sample ID.

Return type:

ndarray of bool

Examples

>>> split = RealizationSplit(("a",), (), ("b",))
>>> split.mask(["a", "b", "a"], "train").tolist()
[True, False, True]

assert_complete(expected_ids)[source]

Assert exact coverage of an expected realization collection.

Parameters:: expected_ids (sequence of str) – Unique IDs expected across all partitions.
Returns:: Successful return proves no expected ID is missing or unexpected.
Return type:: None
Raises:: ValueError – If expected IDs are duplicated or coverage differs.

Examples

>>> split = RealizationSplit(("a",), (), ("b",))
>>> split.assert_complete(["b", "a"]) is None
True

assert_no_lineage_leakage(lineage=None)[source]

Assert that every parent lineage occurs in one partition only.

Parameters:: lineage (mapping or None, optional) – Complete external ID-to-lineage mapping. When omitted, use the mapping persisted on the split. If neither exists, each realization is already an independent unit and the check succeeds.
Returns:: Successful return proves no supplied lineage crosses partitions.
Return type:: None
Raises:: ValueError – If mapping coverage is incomplete or a lineage leaks.

Examples

>>> split = RealizationSplit(("a1", "a2"), (), ("b1",))
>>> split.assert_no_lineage_leakage(
...     {"a1": "a", "a2": "a", "b1": "b"}
... ) is None
True

reassign(realization_ids, partition)[source]

Return a copy with complete realizations moved to one partition.

Parameters:

realization_ids (sequence of str) – Known IDs to move. When lineage is recorded, all members of each affected lineage must be supplied together.
partition ({"train", "validation", "test"}) – Destination partition.

Returns:

New validated assignment. Existing relative order is retained; moved IDs are appended in the requested order.

Return type:

RealizationSplit

Raises:

KeyError – If an ID is unknown.
ValueError – If a lineage is moved partially or training would become empty.

Examples

>>> split = RealizationSplit(("a", "b"), (), ("c",))
>>> split.reassign(["b"], "validation").validation
('b',)

to_dict()[source]

Return the complete schema-2 JSON representation.

Returns:: Mutable copy of partitions, seed, lineage, and strategy.
Return type:: dict

Examples

>>> RealizationSplit(("a",), (), ()).to_dict()["schema_version"]
2

classmethod from_dict(data)[source]

Restore schema-1 or schema-2 split state.

Parameters:: data (mapping) – Serialized split dictionary.
Returns:: Validated immutable runtime split.
Return type:: RealizationSplit
Raises:: ValueError – If the schema is unsupported or assignments leak/overlap.

Examples

>>> split = RealizationSplit(("a",), (), ("b",), seed=1)
>>> RealizationSplit.from_dict(split.to_dict()) == split
True

pycsamt.ai.data.splits.split_realizations(realization_ids, *, validation_fraction=0.1, test_fraction=0.1, seed=0, lineage=None)[source]

Create a deterministic realization- or lineage-level random split.

Parameters:

realization_ids (sequence of str) – Unique geological realization IDs. Input order does not affect output.
validation_fraction (float, default=0.1) – Target fractions in [0, 1) whose sum is less than one.
test_fraction (float, default=0.1) – Target fractions in [0, 1) whose sum is less than one.
seed (int or None, default=0) – NumPy random-generator seed.
lineage (mapping, optional) – Complete ID-to-parent mapping. Whole lineages are assigned together, so realized fractions can differ from targets.

Returns:

Immutable leakage-checked assignment.

Return type:

RealizationSplit

Raises:

ValueError – If IDs, fractions, or lineage coverage are invalid or no training group can remain.

Examples

>>> ids = [f"r{i}" for i in range(10)]
>>> first = split_realizations(
...     ids, validation_fraction=0.2, test_fraction=0.2, seed=4
... )
>>> second = split_realizations(
...     list(reversed(ids)),
...     validation_fraction=0.2,
...     test_fraction=0.2,
...     seed=4,
... )
>>> first == second
True

Keep multiple noise variants of one parent in a single partition:

>>> lineage = {"a-clean": "a", "a-noisy": "a", "b": "b", "c": "c"}
>>> split = split_realizations(
...     list(lineage),
...     validation_fraction=0.25,
...     test_fraction=0.25,
...     lineage=lineage,
... )
>>> split.partition("a-clean") == split.partition("a-noisy")
True

pycsamt.ai.data.splits.realization_folds(realization_ids, *, n_splits=5, seed=0, lineage=None)[source]

Build deterministic group-safe cross-validation test folds.

Parameters:

realization_ids (sequence of str) – Unique realization IDs.
n_splits (int, default=5) – Number of folds. It cannot exceed independent lineage count.
seed (int or None, default=0) – Seed controlling shuffled group order.
lineage (mapping, optional) – Complete ID-to-parent mapping. Each lineage appears in exactly one test fold and never crosses train/test within a fold.

Returns:

Splits with empty validation partitions. Across the tuple, every input realization appears in test exactly once.

Return type:

tuple of RealizationSplit

Raises:

ValueError – If fewer than two folds are requested or independent groups are insufficient.

Examples

>>> folds = realization_folds(["a", "b", "c", "d"], n_splits=2, seed=1)
>>> len(folds)
2
>>> sorted(item for fold in folds for item in fold.test)
['a', 'b', 'c', 'd']