Schemas

Type-safe dataclass schemas with built-in validation for all shared data structures in XPCS Viewer.

Overview

The schemas module provides frozen dataclasses with validation at construction time, ensuring data integrity at I/O boundaries. Each schema validates shape, dtype, NaN presence, and physical constraints.

Quick Start

import numpy as np
from xpcsviewer.schemas import QMapSchema, GeometryMetadata

# Create a validated Q-map
shape = (512, 512)
qmap = QMapSchema(
    sqmap=np.random.rand(*shape).astype(np.float64),
    dqmap=np.zeros(shape, dtype=np.float64),
    phis=np.random.rand(*shape).astype(np.float64),
    sqmap_unit="nm^-1",
    dqmap_unit="nm^-1",
    phis_unit="rad",
)

# Convert from legacy dictionary
qmap = QMapSchema.from_dict(legacy_dict)

# Convert back for backward compatibility
d = qmap.to_dict()

Schema Classes

QMapSchema

class xpcsviewer.schemas.QMapSchema(sqmap, dqmap, phis, sqmap_unit, dqmap_unit, phis_unit, mask=None, partition_map=None)[source]

Bases: object

Q-map data structure with validation.

This schema represents the momentum transfer (Q) maps used for XPCS analysis.

Parameters:
sqmap

Static Q-map, shape=(H, W), dtype=float64, units specified in sqmap_unit

Type:

np.ndarray

dqmap

Dynamic Q-map, shape=(H, W), dtype=float64, units specified in dqmap_unit

Type:

np.ndarray

phis

Azimuthal angle map, shape=(H, W), dtype=float64, units specified in phis_unit

Type:

np.ndarray

sqmap_unit

Units for static Q-map, must be ‘nm^-1’ or ‘A^-1’

Type:

str

dqmap_unit

Units for dynamic Q-map, must be ‘nm^-1’ or ‘A^-1’

Type:

str

phis_unit

Units for azimuthal angle, must be ‘rad’ or ‘deg’

Type:

str

mask

Mask array, shape=(H, W), dtype=int32, values 0 (masked) or 1 (valid)

Type:

np.ndarray, optional

partition_map

Q-bin partition map, shape=(H, W), dtype=int32

Type:

np.ndarray, optional

Validation Rules:

  • All arrays must be 2D with identical shapes

  • Arrays must be float64 dtype

  • No NaN values allowed in Q-map arrays

  • Unit strings must be 'nm^-1' or 'A^-1' for Q, 'rad' or 'deg' for phi

  • Mask values (if provided) must be 0 or 1

sqmap: ndarray
dqmap: ndarray
phis: ndarray
sqmap_unit: Literal['nm^-1', 'A^-1']
dqmap_unit: Literal['nm^-1', 'A^-1']
phis_unit: Literal['rad', 'deg']
mask: ndarray | None = None
partition_map: ndarray | None = None
__post_init__()[source]

Validate Q-map schema after initialization.

classmethod from_dict(data)[source]

Create QMapSchema from legacy dictionary.

Supports both schema naming (sqmap, dqmap, phis) and compute_qmap naming (q, phi). Falls back to compute_qmap keys when schema keys are not present.

Parameters:

data (dict) – Dictionary with keys: sqmap/q, dqmap, phis/phi, sqmap_unit, dqmap_unit, phis_unit, and optionally mask, partition_map

Returns:

Validated Q-map schema instance

Return type:

QMapSchema

classmethod from_compute_qmap(qmap_dict, units_dict)[source]

Create QMapSchema from compute_qmap output.

Maps compute_qmap keys (q, phi) to schema fields (sqmap, phis). The dqmap field is set to zeros since compute_qmap does not produce it.

Parameters:

  • qmap_dict (dict[str, np.ndarray]) – Dictionary from compute_qmap with keys: q, phi, TTH, qx, qy, x, y

  • units_dict (dict[str, str]) – Dictionary from compute_qmap with unit strings for each key

Returns:

Validated Q-map schema instance

Return type:

QMapSchema

to_dict()[source]

Convert to dictionary for backward compatibility.

Returns:

Dictionary representation of Q-map

Return type:

dict

__init__(sqmap, dqmap, phis, sqmap_unit, dqmap_unit, phis_unit, mask=None, partition_map=None)
Parameters:
Return type:

None

GeometryMetadata

class xpcsviewer.schemas.GeometryMetadata(bcx, bcy, det_dist, lambda_, pix_dim, shape, det_rotation=None, incident_angle=None)[source]

Bases: object

Detector geometry metadata with validation.

Parameters:
bcx

Beam center X coordinate (column), in pixels, 0-indexed

Type:

float

bcy

Beam center Y coordinate (row), in pixels, 0-indexed

Type:

float

det_dist

Detector-to-sample distance in millimeters, must be positive

Type:

float

lambda_

X-ray wavelength in Angstroms, must be positive

Type:

float

pix_dim

Pixel size in millimeters, must be positive

Type:

float

shape

Detector shape as (height, width) in pixels

Type:

tuple[int, int]

det_rotation

Detector rotation angle in degrees

Type:

float, optional

incident_angle

Grazing incidence angle in degrees

Type:

float, optional

Validation Rules:

  • det_dist, lambda_, pix_dim must be positive

  • shape must be a 2-tuple of positive integers

  • Beam center is warned (not rejected) if outside detector bounds

bcx: float
bcy: float
det_dist: float
lambda_: float
pix_dim: float
shape: tuple[int, int]
det_rotation: float | None = None
incident_angle: float | None = None
__post_init__()[source]

Validate geometry metadata after initialization.

classmethod from_dict(data)[source]

Create GeometryMetadata from legacy dictionary.

Parameters:

data (dict) – Dictionary with keys: bcx, bcy, det_dist, lambda_, pix_dim, shape, and optionally det_rotation, incident_angle

Returns:

Validated geometry metadata instance

Return type:

GeometryMetadata

Raises:

ValueError – If any critical numeric field is NaN or infinite (BUG-048).

to_dict()[source]

Convert to dictionary for backward compatibility.

Returns:

Dictionary representation of geometry metadata

Return type:

dict

__init__(bcx, bcy, det_dist, lambda_, pix_dim, shape, det_rotation=None, incident_angle=None)
Parameters:
Return type:

None

G2Data

class xpcsviewer.schemas.G2Data(g2, g2_err, delay_times, q_values)[source]

Bases: object

G2 correlation data structure with validation.

Parameters:
g2

G2 correlation values, shape=(n_delay, n_q), dtype=float64. Axis 0 is time (delay), axis 1 is momentum transfer (Q).

Type:

np.ndarray

g2_err

G2 correlation errors, shape=(n_delay, n_q), dtype=float64

Type:

np.ndarray

delay_times

Delay times in seconds, shape=(n_delay,), dtype=float64

Type:

np.ndarray

q_values

Q values for each bin, length=n_q

Type:

list[float]

Validation Rules:

  • g2 and g2_err must have identical 2D shapes (n_delay, n_q)

  • All arrays must be float64 dtype

  • delay_times must be strictly monotonically increasing

  • g2_err values must be non-negative

  • q_values length must match number of Q-bins

g2: ndarray
g2_err: ndarray
delay_times: ndarray
q_values: list[float]
__post_init__()[source]

Validate G2 data after initialization.

classmethod from_dict(data)[source]

Create G2Data from legacy dictionary.

Parameters:

data (dict) – Dictionary with keys: g2, g2_err, delay_times, q_values

Returns:

Validated G2 data instance

Return type:

G2Data

to_dict()[source]

Convert to dictionary for backward compatibility.

Returns:

Dictionary representation of G2 data

Return type:

dict

__init__(g2, g2_err, delay_times, q_values)
Parameters:
Return type:

None

PartitionSchema

class xpcsviewer.schemas.PartitionSchema(partition_map, num_pts, val_list, num_list, metadata, version='1.0.0', mask=None, method=None)[source]

Bases: object

Q-bin partition data structure with validation.

Parameters:
partition_map

Partition map with Q-bin indices, shape=(H, W), dtype=int32

Type:

np.ndarray

num_pts

Number of Q-bins, must be positive

Type:

int

val_list

Q-bin center values in nm^-1 or A^-1, length=num_pts

Type:

list[float]

num_list

Number of pixels per Q-bin, length=num_pts

Type:

list[int]

metadata

Geometry metadata used for partition computation

Type:

GeometryMetadata

version

Schema version string (e.g., “1.0.0”)

Type:

str

mask

Mask used during partitioning, shape=(H, W), dtype=int32

Type:

np.ndarray, optional

method

Binning method, must be ‘linear’ or ‘log’

Type:

str, optional

Validation Rules:

  • partition_map must be 2D integer array

  • num_pts must be positive

  • val_list and num_list lengths must equal num_pts

  • method (if provided) must be 'linear' or 'log'

partition_map: ndarray
num_pts: int
val_list: list[float]
num_list: list[int]
metadata: GeometryMetadata
version: str = '1.0.0'
mask: ndarray | None = None
method: Literal['linear', 'log'] | None = None
__post_init__()[source]

Validate partition schema after initialization.

classmethod from_dict(data)[source]

Create PartitionSchema from legacy dictionary.

Parameters:

data (dict) – Dictionary with keys: partition_map, num_pts, val_list, num_list, metadata, and optionally version, mask, method

Returns:

Validated partition schema instance

Return type:

PartitionSchema

to_dict()[source]

Convert to dictionary for backward compatibility.

Returns:

Dictionary representation of partition schema

Return type:

dict

__init__(partition_map, num_pts, val_list, num_list, metadata, version='1.0.0', mask=None, method=None)
Parameters:
Return type:

None

MaskSchema

class xpcsviewer.schemas.MaskSchema(mask, metadata, version='1.0.0', description=None)[source]

Bases: object

Mask data structure with validation.

Parameters:

  • mask (ndarray)

  • metadata (GeometryMetadata)

  • version (str)

  • description (str | None)

mask

Mask array, shape=(H, W), dtype=int32, values 0 (masked) or 1 (valid)

Type:

np.ndarray

metadata

Geometry metadata associated with the mask

Type:

GeometryMetadata

version

Schema version string (e.g., “1.0.0”)

Type:

str

description

Human-readable description of the mask

Type:

str, optional

Validation Rules:

  • mask must be 2D integer array with values 0 or 1

  • Shape is warned (not rejected) if different from metadata shape

mask: ndarray
metadata: GeometryMetadata
version: str = '1.0.0'
description: str | None = None
__post_init__()[source]

Validate mask schema after initialization.

classmethod from_dict(data)[source]

Create MaskSchema from legacy dictionary.

Parameters:

data (dict) – Dictionary with keys: mask, metadata, and optionally version, description

Returns:

Validated mask schema instance

Return type:

MaskSchema

to_dict()[source]

Convert to dictionary for backward compatibility.

Returns:

Dictionary representation of mask schema

Return type:

dict

__init__(mask, metadata, version='1.0.0', description=None)
Parameters:

  • mask (ndarray)

  • metadata (GeometryMetadata)

  • version (str)

  • description (str | None)

Return type:

None

Immutability

All schemas are frozen dataclasses. Array fields are defensively copied at construction to prevent external mutation. List fields (e.g., q_values) are converted to tuples internally.

# This raises dataclasses.FrozenInstanceError:
qmap.sqmap = new_array

# Use to_dict() / from_dict() to create modified copies:
d = qmap.to_dict()
d["sqmap"] = modified_array
new_qmap = QMapSchema.from_dict(d)

Validators

Validation helper functions used by schema __post_init__ methods.

Data structure validators using dataclasses with validation.

This module provides type-safe, validated data structures for all shared data contracts in the XPCS Viewer codebase.

Each schema includes: - Type annotations for all fields - Validation in __post_init__ - Factory methods for converting from legacy dicts - Docstrings documenting units and ranges

class xpcsviewer.schemas.validators.QMapSchema(sqmap, dqmap, phis, sqmap_unit, dqmap_unit, phis_unit, mask=None, partition_map=None)[source]

Bases: object

Q-map data structure with validation.

This schema represents the momentum transfer (Q) maps used for XPCS analysis.

Parameters:
sqmap

Static Q-map, shape=(H, W), dtype=float64, units specified in sqmap_unit

Type:

np.ndarray

dqmap

Dynamic Q-map, shape=(H, W), dtype=float64, units specified in dqmap_unit

Type:

np.ndarray

phis

Azimuthal angle map, shape=(H, W), dtype=float64, units specified in phis_unit

Type:

np.ndarray

sqmap_unit

Units for static Q-map, must be ‘nm^-1’ or ‘A^-1’

Type:

str

dqmap_unit

Units for dynamic Q-map, must be ‘nm^-1’ or ‘A^-1’

Type:

str

phis_unit

Units for azimuthal angle, must be ‘rad’ or ‘deg’

Type:

str

mask

Mask array, shape=(H, W), dtype=int32, values 0 (masked) or 1 (valid)

Type:

np.ndarray, optional

partition_map

Q-bin partition map, shape=(H, W), dtype=int32

Type:

np.ndarray, optional

sqmap: ndarray
dqmap: ndarray
phis: ndarray
sqmap_unit: Literal['nm^-1', 'A^-1']
dqmap_unit: Literal['nm^-1', 'A^-1']
phis_unit: Literal['rad', 'deg']
mask: ndarray | None = None
partition_map: ndarray | None = None
__post_init__()[source]

Validate Q-map schema after initialization.

classmethod from_dict(data)[source]

Create QMapSchema from legacy dictionary.

Supports both schema naming (sqmap, dqmap, phis) and compute_qmap naming (q, phi). Falls back to compute_qmap keys when schema keys are not present.

Parameters:

data (dict) – Dictionary with keys: sqmap/q, dqmap, phis/phi, sqmap_unit, dqmap_unit, phis_unit, and optionally mask, partition_map

Returns:

Validated Q-map schema instance

Return type:

QMapSchema

classmethod from_compute_qmap(qmap_dict, units_dict)[source]

Create QMapSchema from compute_qmap output.

Maps compute_qmap keys (q, phi) to schema fields (sqmap, phis). The dqmap field is set to zeros since compute_qmap does not produce it.

Parameters:

  • qmap_dict (dict[str, np.ndarray]) – Dictionary from compute_qmap with keys: q, phi, TTH, qx, qy, x, y

  • units_dict (dict[str, str]) – Dictionary from compute_qmap with unit strings for each key

Returns:

Validated Q-map schema instance

Return type:

QMapSchema

to_dict()[source]

Convert to dictionary for backward compatibility.

Returns:

Dictionary representation of Q-map

Return type:

dict

__init__(sqmap, dqmap, phis, sqmap_unit, dqmap_unit, phis_unit, mask=None, partition_map=None)
Parameters:
Return type:

None

class xpcsviewer.schemas.validators.GeometryMetadata(bcx, bcy, det_dist, lambda_, pix_dim, shape, det_rotation=None, incident_angle=None)[source]

Bases: object

Detector geometry metadata with validation.

Parameters:
bcx

Beam center X coordinate (column), in pixels, 0-indexed

Type:

float

bcy

Beam center Y coordinate (row), in pixels, 0-indexed

Type:

float

det_dist

Detector-to-sample distance in millimeters, must be positive

Type:

float

lambda_

X-ray wavelength in Angstroms, must be positive

Type:

float

pix_dim

Pixel size in millimeters, must be positive

Type:

float

shape

Detector shape as (height, width) in pixels

Type:

tuple[int, int]

det_rotation

Detector rotation angle in degrees

Type:

float, optional

incident_angle

Grazing incidence angle in degrees

Type:

float, optional

bcx: float
bcy: float
det_dist: float
lambda_: float
pix_dim: float
shape: tuple[int, int]
det_rotation: float | None = None
incident_angle: float | None = None
__post_init__()[source]

Validate geometry metadata after initialization.

classmethod from_dict(data)[source]

Create GeometryMetadata from legacy dictionary.

Parameters:

data (dict) – Dictionary with keys: bcx, bcy, det_dist, lambda_, pix_dim, shape, and optionally det_rotation, incident_angle

Returns:

Validated geometry metadata instance

Return type:

GeometryMetadata

Raises:

ValueError – If any critical numeric field is NaN or infinite (BUG-048).

to_dict()[source]

Convert to dictionary for backward compatibility.

Returns:

Dictionary representation of geometry metadata

Return type:

dict

__init__(bcx, bcy, det_dist, lambda_, pix_dim, shape, det_rotation=None, incident_angle=None)
Parameters:
Return type:

None

class xpcsviewer.schemas.validators.G2Data(g2, g2_err, delay_times, q_values)[source]

Bases: object

G2 correlation data structure with validation.

Parameters:
g2

G2 correlation values, shape=(n_delay, n_q), dtype=float64. Axis 0 is time (delay), axis 1 is momentum transfer (Q).

Type:

np.ndarray

g2_err

G2 correlation errors, shape=(n_delay, n_q), dtype=float64

Type:

np.ndarray

delay_times

Delay times in seconds, shape=(n_delay,), dtype=float64

Type:

np.ndarray

q_values

Q values for each bin, length=n_q

Type:

list[float]

g2: ndarray
g2_err: ndarray
delay_times: ndarray
q_values: list[float]
__post_init__()[source]

Validate G2 data after initialization.

classmethod from_dict(data)[source]

Create G2Data from legacy dictionary.

Parameters:

data (dict) – Dictionary with keys: g2, g2_err, delay_times, q_values

Returns:

Validated G2 data instance

Return type:

G2Data

to_dict()[source]

Convert to dictionary for backward compatibility.

Returns:

Dictionary representation of G2 data

Return type:

dict

__init__(g2, g2_err, delay_times, q_values)
Parameters:
Return type:

None

class xpcsviewer.schemas.validators.PartitionSchema(partition_map, num_pts, val_list, num_list, metadata, version='1.0.0', mask=None, method=None)[source]

Bases: object

Q-bin partition data structure with validation.

Parameters:
partition_map

Partition map with Q-bin indices, shape=(H, W), dtype=int32

Type:

np.ndarray

num_pts

Number of Q-bins, must be positive

Type:

int

val_list

Q-bin center values in nm^-1 or A^-1, length=num_pts

Type:

list[float]

num_list

Number of pixels per Q-bin, length=num_pts

Type:

list[int]

metadata

Geometry metadata used for partition computation

Type:

GeometryMetadata

version

Schema version string (e.g., “1.0.0”)

Type:

str

mask

Mask used during partitioning, shape=(H, W), dtype=int32

Type:

np.ndarray, optional

method

Binning method, must be ‘linear’ or ‘log’

Type:

str, optional

partition_map: ndarray
num_pts: int
val_list: list[float]
num_list: list[int]
metadata: GeometryMetadata
version: str = '1.0.0'
mask: ndarray | None = None
method: Literal['linear', 'log'] | None = None
__post_init__()[source]

Validate partition schema after initialization.

classmethod from_dict(data)[source]

Create PartitionSchema from legacy dictionary.

Parameters:

data (dict) – Dictionary with keys: partition_map, num_pts, val_list, num_list, metadata, and optionally version, mask, method

Returns:

Validated partition schema instance

Return type:

PartitionSchema

to_dict()[source]

Convert to dictionary for backward compatibility.

Returns:

Dictionary representation of partition schema

Return type:

dict

__init__(partition_map, num_pts, val_list, num_list, metadata, version='1.0.0', mask=None, method=None)
Parameters:
Return type:

None

class xpcsviewer.schemas.validators.MaskSchema(mask, metadata, version='1.0.0', description=None)[source]

Bases: object

Mask data structure with validation.

Parameters:

  • mask (ndarray)

  • metadata (GeometryMetadata)

  • version (str)

  • description (str | None)

mask

Mask array, shape=(H, W), dtype=int32, values 0 (masked) or 1 (valid)

Type:

np.ndarray

metadata

Geometry metadata associated with the mask

Type:

GeometryMetadata

version

Schema version string (e.g., “1.0.0”)

Type:

str

description

Human-readable description of the mask

Type:

str, optional

mask: ndarray
metadata: GeometryMetadata
version: str = '1.0.0'
description: str | None = None
__post_init__()[source]

Validate mask schema after initialization.

classmethod from_dict(data)[source]

Create MaskSchema from legacy dictionary.

Parameters:

data (dict) – Dictionary with keys: mask, metadata, and optionally version, description

Returns:

Validated mask schema instance

Return type:

MaskSchema

to_dict()[source]

Convert to dictionary for backward compatibility.

Returns:

Dictionary representation of mask schema

Return type:

dict

__init__(mask, metadata, version='1.0.0', description=None)
Parameters:

  • mask (ndarray)

  • metadata (GeometryMetadata)

  • version (str)

  • description (str | None)

Return type:

None

See Also

  • I/O Facade - HDF5 facade that returns validated schema objects

  • Backends - Backend array conversion at I/O boundaries