nirs4all.data.raw_multisource module

Raw source-specific staging for heterogeneous multi-source repetitions (N3).

RawMultiSourceDataset holds heterogeneous per-source spectra before any rectangular materialisation. It is the in-memory staging object that lets nirs4all carry e.g. MIR=2 / RAMAN=3 / NIRS=2 measurements per physical sample without fabricating a fake rectangular matrix or refactoring Features into a ragged container (an explicit non-goal of the first delivery, per the roadmap).

The object owns:

• X_by_source – one rectangular block per source (rows are that source’s raw observations, in their input order);

• headers_by_source – the feature headers per source;

• a validated NormalizedObservationTable resolving sample / source / observation identity once;

• targets_by_sample and metadata_by_level derived from the table;

• a deterministic source/row mapping (so the staging is replayable);

• a content fingerprint().

Crucially, there is no implicit ragged -> rectangular coercion. Converting to an aligned matrix must be requested explicitly via RawMultiSourceDataset.materialize(), naming a replayable RepresentationPlan. N5a implements the non-cartesian representations; N5b adds bounded cartesian representations with lineage, reduction and fit-influence contracts.

class nirs4all.data.raw_multisource.AlignedMaterialization(representation: str, X: ndarray, sample_ids: list[str], headers: list[str], targets: list[Any], representation_plan: RepresentationPlan | None = None, unit_ids: list[str] | None = None, source_ids: list[str | None] | None = None, feature_mask: ndarray | None = None, lineage: list[dict[str, Any]] | None = None, partitions: list[str] | None = None)

Bases: object

Result of an explicit rectangular materialisation.

Variables:

• representation (str) – The representation name that produced X.

• X (numpy.ndarray) – The rectangular feature matrix (n_samples, n_features).

• sample_ids (list[str]) – Physical sample ids, aligned to X rows (sorted order).

• headers (list[str]) – Feature headers, aligned to X columns (source-prefixed).

• targets (list[Any]) – Sample-level targets aligned to sample_ids (None where no target is declared).

X: ndarray

feature_mask: ndarray | None = None

property fingerprint: str

Stable SHA-256 over the materialised arrays and replay plan.

headers: list[str]

lineage: list[dict[str, Any]] | None = None

partitions: list[str] | None = None

representation: str

representation_plan: RepresentationPlan | None = None

sample_ids: list[str]

source_ids: list[str | None] | None = None

targets: list[Any]

to_feature_matrix() → tuple[ndarray, list[str]]

Return the numeric matrix and headers consumed by legacy models.

Masked relational representations keep the original padded values in X plus a boolean feature_mask. Legacy estimators cannot consume a side-channel mask, so the model matrix fills missing values with zero and appends deterministic mask columns.

to_manifest() → dict[str, Any]

Return a JSON-serialisable replay/export manifest without arrays.

to_spectro_dataset(name: str = 'relation_materialized') → SpectroDataset

Convert this explicit materialisation to the legacy rectangular dataset API.

unit_ids: list[str] | None = None

nirs4all.data.raw_multisource.CARTESIAN_REPRESENTATIONS: frozenset[str] = frozenset({'cartesian_augmentation', 'cartesian_full', 'cartesian_mc'})

Cartesian representations executable once their N6/N7/N9 contracts are present.

class nirs4all.data.raw_multisource.CombinationPlan(combo_selection: str = 'deterministic_all', max_combos_per_sample: int | None = None, max_total_combos: int | None = None, max_total_rows: int | None = None, memory_budget: int | str | None = None, random_state: int | None = None, train_only: bool = False, version: int = 1)

Bases: object

Replayable selection and cap contract for cartesian representations.

combo_selection: str = 'deterministic_all'

fingerprint() → str

Stable SHA-256 of the combination contract.

classmethod from_dict(data: Mapping[str, Any]) → CombinationPlan

Build a combination plan from a manifest mapping.

max_combos_per_sample: int | None = None

max_total_combos: int | None = None

max_total_rows: int | None = None

memory_budget: int | str | None = None

random_state: int | None = None

to_dict() → dict[str, Any]

Return a JSON-serialisable combination manifest.

train_only: bool = False

version: int = 1

class nirs4all.data.raw_multisource.ComboSelection(value)

Bases: StrEnum

How cartesian combinations are selected.

DETERMINISTIC_ALL = 'deterministic_all'

MATCH_BY = 'match_by'

NONE = 'none'

RANDOM_SEEDED = 'random_seeded'

STRATIFIED = 'stratified'

ZIP = 'zip'

nirs4all.data.raw_multisource.DECLARED_REPRESENTATIONS: frozenset[str] = frozenset({})

Representations that are part of the design but not executable in this phase.

nirs4all.data.raw_multisource.EXECUTABLE_REPRESENTATIONS: frozenset[str] = frozenset({'cartesian_augmentation', 'cartesian_full', 'cartesian_mc', 'per_source_aggregate', 'per_source_observation', 'sample_aggregate', 'stack_fixed', 'stack_padded_masked'})

Representations that RawMultiSourceDataset.materialize() can execute now.

class nirs4all.data.raw_multisource.RawMultiSourceDataset(relation_table: NormalizedObservationTable, X_by_source: Mapping[str, ndarray], *, headers_by_source: Mapping[str, Sequence[str]] | None = None, spec: RepetitionSpec | None = None)

Bases: object

Heterogeneous per-source staging dataset (roadmap N3).

See the module docstring. Construct via from_sources() (the normal path) or directly from an already-built relation table.

X_by_source: dict[str, ndarray]

property aggregate: str | None

Legacy-compatible aggregation setting before explicit materialisation.

aligned_row_order() → list[tuple[str, str, int]]

Deterministic (physical_sample_id, source_id, source_row) order.

The order is canonical – sorted by sample, then source, then repetition – so it does not depend on the input row order or source order. This is the replayable source/row mapping the staging guarantees.

cardinalities() → dict[tuple[str, str], int]

Observation counts keyed by (physical_sample_id, source_id).

content_hash(source_index: int | None = None) → str

Legacy-compatible content hash for pipeline bookkeeping.

feature_dims() → dict[str, int]

Feature width per source.

features_sources() → int

Legacy-compatible number of feature sources.

fingerprint() → str

Deterministic SHA-256 over the table, spec and per-source contents.

The digest binds each feature vector to its identity (physical_sample_id, source_id) rather than its raw row position, so source order and input row order never change it. Repetition order is honoured per RepetitionSpec.rep_order: exchangeable reps are hashed as an order-independent multiset (shuffled reps -> same digest); ordered reps are hashed in rep_id order.

classmethod from_sources(spec: RepetitionSpec, X_by_source: Mapping[str, ndarray], keys_by_source: Mapping[str, Sequence[Any]], *, headers_by_source: Mapping[str, Sequence[str]] | None = None, rep_by_source: Mapping[str, Sequence[Any]] | None = None, targets_by_source: Mapping[str, Sequence[Any]] | None = None, metadata_by_source: Mapping[str, Mapping[str, Sequence[Any]]] | None = None, partition: Partition | str = Partition.TRAIN) → RawMultiSourceDataset

Build a staging dataset by joining per-source observations on a key.

Parameters:

• spec – The source-aware repetition specification.

• X_by_source – Per-source raw feature blocks (input row order).

• keys_by_source – Per-source link_by key values (one per row).

• headers_by_source – Optional feature headers per source.

• rep_by_source – Optional explicit repetition indices per source.

• targets_by_source – Optional per-row targets per source (sample-level; may be declared on a single source and shared via the join).

• metadata_by_source – Optional source_id -> {col -> per-row values}.

• partition – Partition label applied to every produced row.

Returns:

A validated RawMultiSourceDataset.

Raises:

RelationValidationError – If the join or the array shapes are invalid.

headers_by_source: dict[str, list[str]]

materialize(representation: str | Mapping[str, Any] | RepresentationPlan | None = None) → AlignedMaterialization

Materialise an explicit rectangular representation.

There is no default: a representation must be named, by design, so a heterogeneous staging dataset is never silently coerced into a (possibly meaningless) rectangular matrix. Executable representations include per-source aggregates/observations, fixed or padded stacks, and bounded cartesian combo materialisations.

Parameters:

representation – The representation to build. None is rejected.

Returns:

An AlignedMaterialization.

Raises:

RelationValidationError – If no representation is named, if a declared representation is not yet executable, or if the name is unknown.

metadata_by_level: dict[str, dict[str, dict[str, Any]]]

property n_observations: int

Total number of raw observations across all sources.

property n_samples: int

Number of physical samples.

property name: str

Dataset identifier used by legacy pipeline logging.

property num_features: int

Legacy-compatible rectangular width for the default materialized block.

property num_samples: int

Legacy-compatible sample count for pipeline shape logging.

property physical_sample_ids: list[str]

Sorted unique physical sample ids (deterministic).

short_preprocessings_str() → str

Display label used by chain persistence before materialisation.

property source_ids: list[str]

Sorted unique source ids (deterministic).

targets_by_sample() → dict[str, Any]

Sample-level target value per physical sample.

to_manifest() → dict[str, Any]

Serialisable manifest (no feature arrays) for future workspace work.

This carries enough to describe the staging – spec, source ids, sample ids, cardinalities, headers, feature dims, targets and fingerprints – without the heavy arrays. It is intentionally JSON-serialisable; the array persistence (workspace / .n4a) is a later phase (N9).

class nirs4all.data.raw_multisource.RepresentationPlan(representation: str, unit_level: str | None = None, stage: str | None = None, lineage: Sequence[str] | None = None, combo_selection: str = 'none', missing_source_policy: str = 'error', missing_repetition_policy: str = 'error', max_combos_per_sample: int | None = None, max_total_combos: int | None = None, max_total_rows: int | None = None, memory_budget: int | str | None = None, random_state: int | None = None, combination_plan: CombinationPlan | None = None, version: int = 1)

Bases: object

Replayable plan for raw multi-source representation materialisation.

combination_plan: CombinationPlan | None = None

combo_selection: str = 'none'

fingerprint() → str

Stable SHA-256 of the replay contract.

classmethod from_dict(data: Mapping[str, Any]) → RepresentationPlan

Build a plan from a representation manifest.

classmethod from_step_value(value: str | Mapping[str, Any] | RepresentationPlan) → RepresentationPlan

Parse compact pipeline/operator syntax into a plan.

property is_cartesian: bool

Whether the plan is a cartesian representation declared for N5b+.

property is_executable: bool

Whether the plan can be materialised in the current implementation.

lineage: Sequence[str] | None = None

max_combos_per_sample: int | None = None

max_total_combos: int | None = None

max_total_rows: int | None = None

memory_budget: int | str | None = None

missing_repetition_policy: str = 'error'

missing_source_policy: str = 'error'

random_state: int | None = None

representation: str

stage: str | None = None

to_dict(*, include_fingerprint: bool = True) → dict[str, Any]

Return a JSON-serialisable representation manifest.

unit_level: str | None = None

version: int = 1

class nirs4all.data.raw_multisource.RepresentationStage(value)

Bases: StrEnum

Execution stage of a representation plan.

COMBO = 'combo'

RAW_MULTISOURCE = 'raw_multisource'

SAMPLE_AGGREGATE = 'sample_aggregate'

SOURCE_AGGREGATE = 'source_aggregate'

SOURCE_OBSERVATION = 'source_observation'

STACK = 'stack'

class nirs4all.data.raw_multisource.RepresentationUnitLevel(value)

Bases: StrEnum

Unit level produced by a materialised representation.

COMBO = 'combo'

SAMPLE = 'sample'

SOURCE_AGGREGATE = 'source_aggregate'

SOURCE_OBSERVATION = 'source_observation'

STACK = 'stack'

nirs4all.data.raw_multisource.replay_materialization(dataset: RawMultiSourceDataset, manifest: Mapping[str, Any], *, validate_fingerprint: bool = False) → AlignedMaterialization

Replay a materialisation from a minimal N9 representation manifest.

By default this reuses only the replayable RepresentationPlan, so prediction data may have different row/cardinality counts. The feature-space contract (number and names of columns) is still validated because a bundle model cannot safely consume a different rectangular representation. Set validate_fingerprint=True when replaying the exact same dataset and expecting byte-identical materialisation output.