API Reference¶

Complete API documentation for all r2x-core classes and functions.

System¶

class r2x_core.System(system_base=None, *, name=None, **kwargs)[source]¶

Bases: System

R2X Core System class extending infrasys.System.

Extends infrasys.System to provide R2X-specific functionality for data model translation and system construction. Adds convenience methods for component export and system manipulation.

Parameters:

system_base (float | None, optional) – System base power in MVA for per-unit calculations. Default is None.
name (str | None, optional) – Unique identifier for the system. Default is None.
**kwargs – Additional keyword arguments passed to infrasys.System (e.g., description, auto_add_composed_components).

name¶

System identifier.

Type:: str

description¶

System description.

Type:: str

base_power¶

System base power in MVA.

Type:: float | None

See also

infrasys.system.System: Parent class with core system functionality.
BaseParser: Parser framework for building systems.

Initialize R2X Core System.

This method defines the ‘system_base’ unit in the global Pint registry. If you create multiple System instances, the last one’s system_base will be used for all unit conversions. Existing components will detect the change and issue a warning if they access system_base conversions.

Parameters:

base_power (float, optional (defaults: 100.0)) – System base power in MVA for per-unit calculations. Can be provided as first positional argument or as keyword argument.
name (str, optional) – Name of the system. If not provided, a default name will be assigned.
**kwargs – Additional keyword arguments passed to infrasys.System (e.g., description, auto_add_composed_components).
system_base (float | None)

iter_translated_components()[source]¶

Iterate components tagged as rule-produced translations.

If this system contains any SourceProvenance tags, this yields only components carrying SourceProvenance(preserved=False). Untagged components in a provenance-bearing system are neither translated nor preserved; they may be pre-existing target content.

If no components in this system carry provenance tags, this yields every component (i.e. systems that were not built with preserve_source=True behave the same as iter_all_components()).

Yields:: Component – Every component tagged as a rule-produced translation, or every component when no provenance tags exist.
Return type:: Iterator[Component]

iter_preserved_components()[source]¶

Iterate components that were carried over from source without translation.

Yields:: Component – Every component tagged with SourceProvenance(preserved=True).
Return type:: Iterator[Component]

is_preserved(component)[source]¶

Return True if component was carried over from source untranslated.

Cheap single-component check that hits the SA association table directly. When walking many components, prefer iter_preserved_components() or iter_translated_components() which resolve the whole preserved-UUID set once instead of per-component.

Parameters:: component (Component) – Component to check.
Returns:: True when the component carries SourceProvenance(preserved=True).
Return type:: bool

add_components(*components, **kwargs)[source]¶

Add one or more components to the system and set their _system_base.

Parameters:

*components (Component) – Component(s) to add to the system.
**kwargs (Any) – Additional keyword arguments passed to parent’s add_components.

Return type:

None

Notes

If any component is a HasPerUnit model, this method automatically sets the component’s _system_base attribute for use in system-base per-unit display mode.

Raises:

ValueError – If a component already has a different _system_base set.

Parameters:

components (Component)
kwargs (Any)

Return type:

None

to_json(fname=None, overwrite=False, indent=None, data=None)[source]¶

Serialize system to JSON file or return bytes.

Parameters:

fname (Path or str, optional) – Output JSON file path. If None, prints JSON to stdout. Note: When writing to stdout, time series are serialized to a temporary directory that will be cleaned up automatically.
overwrite (bool, default False) – If True, overwrite existing file. If False, raise error if file exists.
indent (int, optional) – JSON indentation level. If None, uses compact format.
data (optional) – Additional data to include in serialization.

Return type:

None

Raises:

FileExistsError – If file exists and overwrite=False.

See also

from_json(): Load system from JSON file

classmethod from_json(source, /, *, upgrade_handler=None, **kwargs)[source]¶

Deserialize system from JSON file.

Parameters:

source (Path, str, or bytes) – Input JSON source.
upgrade_handler (Callable, optional) – Function to handle data model version upgrades.
**kwargs – Additional keyword arguments passed to infrasys deserialization.

Returns:

Deserialized system instance.

Return type:

System

Raises:

FileNotFoundError – If file does not exist.
ValueError – If JSON format is invalid.

See also

to_json(): Serialize system to JSON file.
upgrade_data(): Phase 1 upgrades for parser workflow.

serialize_system_attributes()[source]¶

Serialize R2X-specific system attributes.

Returns:: Dictionary containing system_base_power, r2x_core_version, and source_provenance_info when set.
Return type:: dict[str, Any]

deserialize_system_attributes(data)[source]¶

Deserialize R2X-specific system attributes.

Parameters:: data (dict[str, Any]) – Dictionary containing serialized system attributes.
Return type:: None

Unit handling for power system models.

This module provides unit-aware field annotations, automatic conversion between natural units and per-unit values, and configurable display formatting.

Notes

Annotate numeric fields with Annotated[float, Unit("kV"|"MVA"|"pu", base="base_field")]
Natural-unit inputs ({"value": 138, "unit": "kV"}) are converted to per-unit when base is set
Internal storage is always float (device-base per-unit for relative quantities)
Global display mode (device base, system base, or natural units) affects __repr__ formatting only

class r2x_core.units.HasPerUnit[source]

Bases: HasUnits

Component class with per-unit conversion capabilities.

This class extends HasUnits with system-base per-unit display support. Use this for components that have both absolute unit fields (base values) and per-unit fields that reference those bases.

_system_base

System-wide base power for system-base per-unit display

Type:: float or None

class r2x_core.units.HasUnits[source]

Bases: object

Mixin providing unit-aware field formatting.

This mixin provides unit-aware field formatting in repr without per-unit conversion capabilities. Suitable for components that only have absolute unit fields (e.g., voltages in kV, power in MW) without base conversions.

Can be combined with any Pydantic BaseModel or Component:: class MyComponent(HasUnits, Component): …

r2x_core.units.Unit(unit, *, base=None)

Create a UnitSpec for field annotation.

Parameters:

unit (str) – Unit string (e.g., “MVA”, “kV”, “pu”)
base (str, optional) – Field name for device base lookup

Returns:

Unit specification instance

Return type:

UnitSpec

class r2x_core.units.UnitSpec(unit, base=None)[source]

Bases: object

Metadata descriptor for unit-aware fields.

Parameters:

unit (str)
base (str | None)

unit

Unit string (e.g., “MVA”, “pu”, “kV”)

Type:: str

base

Field name for device base lookup (for pu units)

Type:: str, optional

unit: str

base: str | None = None

class r2x_core.units.UnitSystem(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: str, Enum

Display modes for formatted representation.

DEVICE_BASE

Display per-unit values relative to device base

Type:: str

SYSTEM_BASE

Display per-unit values relative to system base

Type:: str

NATURAL_UNITS

Display values in their natural units (e.g., kV, MVA)

Type:: str

DEVICE_BASE = 'DEVICE_BASE'

SYSTEM_BASE = 'SYSTEM_BASE'

NATURAL_UNITS = 'NATURAL_UNITS'

r2x_core.units.get_unit_system()[source]

Get the current global display mode.

Returns:: Current display mode
Return type:: UnitSystem

r2x_core.units.set_unit_system(unit_system)[source]

Set the global display mode.

Parameters:: unit_system (UnitSystem) – Display mode to set
Return type:: None

r2x_core.units.unit_spec(unit, *, base=None)[source]

Create a UnitSpec for field annotation.

Parameters:

unit (str) – Unit string (e.g., “MVA”, “kV”, “pu”)
base (str, optional) – Field name for device base lookup

Returns:

Unit specification instance

Return type:

UnitSpec

r2x_core.units.unit_system(mode)[source]

Context manager for temporary display mode changes.

Parameters:: mode (UnitSystem) – Temporary display mode to use within the context
Yields:: None
Return type:: Iterator[None]

Examples

>>> gen = Generator(...)
>>> with unit_system(UnitSystem.NATURAL_UNITS):
...     print(gen)  # Displays in natural units
>>> print(gen)  # Back to previous mode

Data Management¶

class r2x_core.DataStore(path=None, *, reader=None, upgrade_handler=None)[source]

Container for managing data file mappings and loading data.

The DataStore provides a high-level interface for managing a collection of DataFile instances, including loading data, caching, and executing version upgrades. It can be initialized from a folder path, JSON configuration, or PluginConfig.

Parameters:

path (str | Path | None, optional) – Path to the folder containing data files or a file_mapping.json. If None, uses current working directory. Default is None.
reader (DataReader | None, optional) – Custom DataReader instance. If None, a default reader is created. Default is None.
upgrade_handler (Callable[..., Any] | None, optional) – Optional upgrade handler to run when missing files require migration.

folder

The resolved folder path containing the data files.

Type:: Path

reader

The data reader instance used to load data.

Type:: DataReader

from_data_files(data_files, path)[source]

Create a DataStore from a list of DataFile instances.

Parameters:

data_files (list[DataFile])
path (Path | str | None)
upgrade_handler (Callable[[...], Any] | None)

Return type:

DataStore

from_json(json_fpath, path)[source]

Create a DataStore from a JSON configuration file.

Parameters:

json_fpath (Path | str)
path (Path | str | None)
upgrade_handler (Callable[[...], Any] | None)

Return type:

DataStore

from_plugin_config(plugin_config, path)[source]

Create a DataStore from a PluginConfig instance.

Parameters:

plugin_config (PluginConfig)
path (Path | str)
upgrade_handler (Callable[[...], Any] | None)

Return type:

DataStore

add_data(*data_files, overwrite)[source]

Add one or more DataFile instances to the store.

Parameters:

data_files (Sequence[DataFile])
overwrite (bool)

Return type:

None

read_data(name, placeholders)[source]

Load data from a file by name.

Parameters:

name (str)
placeholders (dict[str, Any] | None)

Return type:

Any

list_data()[source]

List all data file names in the store.

Return type:: list[str]

remove_data(*names)[source]

Remove one or more data files from the store.

Parameters:: names (str)
Return type:: None

to_json(fpath, \*\*model_dump_kwargs)[source]

Export store configuration to JSON.

Parameters:

fpath (str | Path)
model_dump_kwargs (dict[str, Any])

Return type:

None

See also

DataFile: Individual data file configuration.
DataReader: Reader for loading data files.
PluginConfig: Plugin configuration source.

Notes

All data file names must be unique within a store. The reader cache is managed separately from the file configuration cache.

Initialize the DataStore.

property folder: Path: Return the resolved folder path containing data files.

property reader: DataReader: Return the DataReader instance.

classmethod from_data_files(data_files, *, path=None, upgrade_handler=None)[source]

Create a DataStore from a list of DataFile instances.

Parameters:

data_files (list[DataFile]) – List of DataFile instances to add to the store.
path (Path | str | None, optional) – Path to the folder containing data files. Default is None.
upgrade_handler (Callable[..., Any] | None, optional) – Optional upgrade handler for missing files during reads.

Returns:

New DataStore instance with provided data files.

Return type:

DataStore

classmethod from_json(json_fpath, *, path=None, upgrade_handler=None)[source]

Create a DataStore from a JSON configuration file.

Parameters:

json_fpath (Path | str) – Path to the JSON file containing data file configurations.
path (Path | str | None) – Path to the folder containing data files. Defaults to the JSON’s parent folder.
upgrade_handler (Callable[..., Any] | None, optional) – Optional upgrade handler for missing files during reads.

Returns:

New DataStore instance with data files from JSON.

Return type:

DataStore

Raises:

FileNotFoundError – If the data folder or json_fpath does not exist.
TypeError – If JSON file is not a JSON array.
ValidationError – If data files in JSON are invalid.

classmethod from_plugin_config(plugin_config, *, path, upgrade_handler=None)[source]

Create a DataStore from a PluginConfig instance.

Parameters:

plugin_config (PluginConfig) – Plugin configuration containing file mappings.
path (Path | str) – Path to the folder containing data files.
upgrade_handler (Callable[..., Any] | None, optional) – Optional upgrade handler to run if file mappings fail to validate. If None, the store attempts to resolve a handler from the plugin package.

Returns:

New DataStore instance with data files from plugin config.

Return type:

DataStore

classmethod load_file(fpath, *, name=None, proc_spec=None)[source]

Load a single data file conveniently without creating a full DataStore.

This is a convenience method for loading a single file with minimal setup. For complex use cases with multiple files, use the full DataStore API.

Parameters:

fpath (str | Path) – Path to the data file to load.
name (str | None, optional) – Name identifier for the file. If None, uses the file stem (name without extension). Default is None.
proc_spec (FileProcessing | None, optional) – Process to apply to the file data. Can be a TabularProcessing or JSONProcessing instance, or a dictionary with transformation parameters. Default is None.

Returns:

Loaded data from the file (type depends on file format).

Return type:

Any

Raises:

FileNotFoundError – If the file does not exist.
Exception – If the file cannot be read.

add_data(data_files, *, overwrite=False)[source]

Add one or more DataFile instances to the store.

Parameters:

data_files (Sequence[DataFile]) – DataFile instances to add.
overwrite (bool, optional) – If True, overwrite existing data files with the same name. Default is False.

Raises:

TypeError – If any item is not a DataFile instance.
KeyError – If a data file with the same name exists and overwrite is False.

Return type:

None

list_data()[source]

List all data file names in the store.

Returns:: Sorted list of all data file names.
Return type:: list[str]

read_data(name, *, placeholders=None)[source]

Load data from a file using the configured DataReader.

Parameters:

name (str) – Name of the data file to load.
placeholders (dict[str, Any] | None, optional) – Placeholder values for template substitution. Default is None.

Returns:

Loaded data from the file.

Return type:

Any

Raises:

KeyError – If the data file name is not in the store.

configure_upgrades(*, plugin_config=None, upgrade_handler=None)[source]

Configure upgrade handling for this store.

Parameters:

plugin_config (PluginConfig | None)
upgrade_handler (Callable[[...], Any] | None)

Return type:

None

remove_data(*names)[source]

Remove one or more data files from the store.

Parameters:: *names (str) – Data file names to remove. Example: remove_data(“file1”, “file2”)
Raises:: KeyError – If any data file name is not in the store.
Return type:: None

to_json(fpath, **model_dump_kwargs)[source]

Save the DataStore configuration to a JSON file.

Parameters:

fpath (str | Path) – Path where JSON file will be written.
**model_dump_kwargs (dict[str, Any]) – Additional keyword arguments passed to DataFile.model_dump().

Return type:

None

Notes

Output JSON is formatted with 2-space indentation and includes Unicode.

class r2x_core.DataReader[source]¶

Reader class for loading data files with automatic processing.

The DataReader handles the complete reading pipeline: file discovery, format detection, reading, and transformations. File-type-specific reading logic is delegated via singledispatch based on file extension.

See also

DataFile: File configuration with specs.
read_data_file(): Main method for reading data with processing.

Initialize the data reader.

No configuration needed; the reader is stateless and uses configuration from DataFile objects and optional placeholders at read time.

read_data_file(data_file, *, folder_path, placeholders=None)[source]¶

Read a data file.

Parameters:

data_file (DataFile) – Data file configuration with metadata.
folder_path (Path) – Base directory containing the data files.
placeholders (dict[str, Any] | None, optional) – Dictionary mapping placeholder variable names to their values. Used to substitute placeholders like {solve_year} in filter_by. Default is None.

Returns:

The loaded data, type depends on file type.

Return type:

Any

Raises:

FileNotFoundError – If a required file does not exist or if a glob pattern matches no files.
ValueError – If glob pattern is malformed (no wildcards) or if placeholders are found in filter_by but no placeholders dict provided.
MultipleFileError – If a glob pattern matches multiple files (subclass of ValueError).

See also

read_file_by_type(): File-type-specific reading.
apply_processing(): Transformation pipeline.
get_file_path(): Path resolution.

get_supported_file_types()[source]¶

Get list of supported file extensions.

Returns:: List of supported file extensions.
Return type:: list[str]

register_custom_transformation(data_types, *, transform_func)[source]¶

Parameters:

data_types (type or tuple of types) – Data type(s) the function can handle.
transform_func (callable) – Function that transforms data given a DataFile configuration. Signature: (data_file: DataFile, data: Any) -> Any

Return type:

None

Examples

>>> def my_transform(data_file: DataFile, data: MyClass) -> MyClass:
...     # Custom logic here
...     return data
>>> reader.register_custom_transformation(MyClass, transform_func=my_transform)

pydantic model r2x_core.DataFile[source]¶

Data file configuration with nested structure.

Defines how to locate, read, and transform a single data file. Supports absolute paths, relative paths, and glob patterns. Processing is applied after reading and is type-specific (tabular or JSON).

Parameters:

name (str) – Unique identifier for this file mapping.
fpath (Path | None, optional) – Absolute path to the data file. Exactly one of fpath, relative_fpath, or glob must be specified. Default is None.
relative_fpath (Path | str | None, optional) – Path relative to DataStore folder. Default is None.
glob (str | None, optional) – Glob pattern to locate files. Default is None.
info (FileInfo | None, optional) – File metadata including role, optionality, time series flag. Default is None.
reader (ReaderConfig | None, optional) – Reader configuration specifying kwargs and custom functions. Default is None.
proc_spec (FileProcessing | None, optional) – Data transformations (TabularProcessing or JSONProcessing). Default is None.

Raises:

ValueError – If path sources are not exactly one of: fpath, relative_fpath, glob.
ValueError – If file type does not support time series and is_timeseries is True.

See also

FileInfo: File metadata.
ReaderConfig: Reader configuration.
TabularProcessing: Transformations for tabular files.
JSONProcessing: Transformations for JSON files.
DataStore: Container that manages DataFile instances.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Fields:

fpath (pathlib.Path | None)
glob (str | None)
info (r2x_core.datafile.FileInfo | None)
name (str)
proc_spec (r2x_core.datafile.TabularProcessing | r2x_core.datafile.JSONProcessing | None)
reader (r2x_core.datafile.ReaderConfig | None)
relative_fpath (pathlib.Path | str | None)

field fpath: Annotated[Path | None, AfterValidator(func=_validate_optional_file_extension), FieldInfo(annotation=NoneType, required=True, description='Absolute file path')] = None¶

Absolute file path

Constraints:

func = <function _validate_optional_file_extension at 0x7f1919f3a200>

field glob: Annotated[str | None, AfterValidator(func=validate_glob_pattern), FieldInfo(annotation=NoneType, required=True, description='Glob pattern')] = None¶

Glob pattern

Constraints:

func = <function validate_glob_pattern at 0x7f191a5faa20>

field info: Annotated[FileInfo | None, FieldInfo(annotation=NoneType, required=True, description='File metadata')] = None¶: File metadata

field name: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Name of the mapping')] [Required]¶: Name of the mapping

field proc_spec: Annotated[TabularProcessing | JSONProcessing | None, FieldInfo(annotation=NoneType, required=True, description='Data transformations')] = None¶: Data transformations

field reader: Annotated[ReaderConfig | None, FieldInfo(annotation=NoneType, required=True, description='Reader configuration')] = None¶: Reader configuration

field relative_fpath: Annotated[Path | str | None, FieldInfo(annotation=NoneType, required=True, description='Relative file path')] = None¶: Relative file path

classmethod from_record(record, *, folder_path)[source]¶

Build a DataFile from a single record dictionary.

Parameters:

record (dict[str, Any])
folder_path (Path)

Return type:

DataFile

classmethod from_records(records, *, folder_path)[source]¶

Construct multiple DataFile instances from JSON records.

Parameters:

records (list[dict[str, Any]])
folder_path (Path)

Return type:

list[DataFile]

property file_type: FileFormat¶: Computed file type based on file extension.

class r2x_core.FileInfo(*, description=None, is_input=True, is_optional=False, is_timeseries=False, units=None)[source]

File metadata and properties.

Contains descriptive information about the data file, including its role in the workflow, whether it contains time series data, and units for single-column files.

Parameters:

description (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Description of the data file')])
is_input (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether this is an input file')])
is_optional (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether this file is optional')])
is_timeseries (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether file contains time series data')])
units (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Units for single-column data')])

description

Human-readable description of the file purpose.

Type:: str | None

is_input

Whether the file is an input source. Default is True.

Type:: bool

is_optional

Whether the file is optional for processing. Default is False.

Type:: bool

is_timeseries

Whether the file contains time series data. Default is False.

Type:: bool

units

Units for single-column numeric data. Default is None.

Type:: str | None

See also

DataFile: Complete data file configuration.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class r2x_core.ReaderConfig(*, kwargs=<factory>, function=None)[source]

Reader configuration for file loading.

Specifies how to read the data file, including keyword arguments for the default reader or a custom function for specialized file formats.

Parameters:

kwargs (Annotated[dict[str, Any], FieldInfo(annotation=NoneType, required=False, default_factory=dict, description='Keyword arguments for reader')])
function (Annotated[Callable[[Path], Any] | None, FieldInfo(annotation=NoneType, required=True, description='Custom reader function')])

function

Custom reader function that takes file path and returns loaded data. If None, uses the default reader for the file type. Default is None.

Type:: Callable[[Path], Any] | None

kwargs

Keyword arguments passed to the reader function. Default is empty.

Type:: dict[str, Any]

See also

DataFile: Complete data file configuration.
DataStore: Container that uses ReaderConfig to load files.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class r2x_core.TabularProcessing(*, select_columns=None, drop_columns=None, column_mapping=None, rename_index=None, column_schema=None, filter_by=None, set_index=None, reset_index=None, pivot_on=None, unpivot_on=None, group_by=None, aggregate_on=None, sort_by=None, distinct_on=None, replace_values=None, fill_null=None)[source]

Data transformations for tabular files (CSV, HDF5, Parquet, etc.).

Defines a sequence of operations to apply to tabular data after loading. Supports column selection, filtering, reshaping, aggregation, and more.

Parameters:

select_columns (Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=True, description='Columns to keep')])
drop_columns (Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=True, description='Columns to remove')])
column_mapping (Annotated[dict[str, str] | None, FieldInfo(annotation=NoneType, required=True, description='Column rename mapping')])
rename_index (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Rename the index')])
column_schema (Annotated[dict[str, str] | None, FieldInfo(annotation=NoneType, required=True, description='Column type definitions')])
filter_by (Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=True, description='Row filters')])
set_index (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Column to set as index')])
reset_index (Annotated[bool | None, FieldInfo(annotation=NoneType, required=True, description='Convert index to column')])
pivot_on (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Column to pivot on')])
unpivot_on (Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=True, description='Columns to unpivot')])
group_by (Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=True, description='Columns to group by')])
aggregate_on (Annotated[dict[str, str] | None, FieldInfo(annotation=NoneType, required=True, description='Aggregation spec')])
sort_by (Annotated[dict[str, str] | None, FieldInfo(annotation=NoneType, required=True, description='Sort specification')])
distinct_on (Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=True, description='Columns for deduplication')])
replace_values (Annotated[dict[Any, Any] | None, FieldInfo(annotation=NoneType, required=True, description='Value replacement map')])
fill_null (Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=True, description='Null fill values')])

select_columns

List of column names to keep. Removes all other columns.

Type:: list[str] | None

drop_columns

List of column names to remove.

Type:: list[str] | None

column_mapping

Maps original column names to new names.

Type:: dict[str, str] | None

rename_index

New name for the index.

Type:: str | None

column_schema

Maps column names to data types for type coercion.

Type:: dict[str, str] | None

filter_by

Conditions for filtering rows by column values.

Type:: dict[str, Any] | None

set_index

Column name to use as the index.

Type:: str | None

reset_index

If True, converts index to a regular column.

Type:: bool | None

pivot_on

Column to pivot on for reshaping data.

Type:: str | None

unpivot_on

Columns to unpivot for reshaping data.

Type:: list[str] | None

group_by

Columns to group by for aggregation.

Type:: list[str] | None

aggregate_on

Mapping of columns to aggregation functions.

Type:: dict[str, str] | None

sort_by

Columns and sort directions for ordering.

Type:: dict[str, str] | None

distinct_on

Columns to use for deduplication.

Type:: list[str] | None

replace_values

Maps old values to new values for replacement.

Type:: dict[Any, Any] | None

fill_null

Specifies fill values for null entries by column.

Type:: dict[str, Any] | None

See also

DataFile: Uses TabularProcessing in proc_spec field.
JSONProcessing: Transformations for JSON files.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class r2x_core.JSONProcessing(*, key_mapping=None, rename_index=None, drop_keys=None, filter_by=None, replace_values=None, select_keys=None)[source]

Data transformations for JSON files.

Defines operations for processing JSON-structured data, including key selection, filtering, and value transformations.

Parameters:

key_mapping (Annotated[dict[str, str] | None, FieldInfo(annotation=NoneType, required=True, description='Key rename mapping (JSON-specific)')])
rename_index (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Rename dict keys')])
drop_keys (Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=True, description='Keys to drop from nested dicts')])
filter_by (Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=True, description='Filter conditions')])
replace_values (Annotated[dict[Any, Any] | None, FieldInfo(annotation=NoneType, required=True, description='Value replacement map')])
select_keys (Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=True, description='Select certain keys.')])

key_mapping

Maps original JSON keys to new key names.

Type:: dict[str, str] | None

rename_index

New name for the dictionary keys.

Type:: str | None

drop_columns

Keys to remove from nested dictionaries.

Type:: list[str] | None

filter_by

Conditions for filtering JSON objects by key values.

Type:: dict[str, Any] | None

replace_values

Maps old values to new values for replacement.

Type:: dict[Any, Any] | None

select_keys

Select specific keys to keep.

Type:: list[str] | None

See also

DataFile: Uses JSONProcessing in proc_spec field.
TabularProcessing: Transformations for tabular files.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Plugin Configuration¶

pydantic model r2x_core.PluginConfig[source]

Pure Pydantic base configuration class for plugins.

This class provides a base configuration schema for plugins, managing paths to configuration files and component modules. It supports both default and custom configuration paths, with validation and override capabilities.

CONFIG_DIR

Class variable specifying the default configuration directory name. Default is “config”.

Type:: str

models

Module path(s) for component classes. Examples: ‘r2x_sienna.models’, ‘my_package.components’. If omitted, defaults to an empty tuple.

Type:: tuple[str, …]

config_path_override

Optional override for the configuration path. When provided, this path is used instead of the default package configuration directory.

Type:: Path | None

Examples

Basic usage with default configuration:

>>> config = PluginConfig(models='my_package.models')
>>> config.config_path
PosixPath('.../config')
>>> config.defaults_path
PosixPath('.../config/defaults.json')

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Fields:

config_path_override (pathlib.Path | None)
models (tuple[str, ...])

field config_path_override: Path | None = None: Override for the configuration path. This is used if you want to point to a different place than the CONFIG_DIR

field models: tuple[str, ...] [Optional]: Module path(s) for component classes, e.g. ‘r2x_sienna.models’. If omitted, rules will use an empty module list.

classmethod load_config(*, config_path=None, overrides=None)[source]

Load plugin configuration assets with optional overrides.

Loads all configuration assets (defaults, file mappings, rules) from the specified config directory and optionally merges them with override values.

Parameters:

config_path (Path | str | None, optional) – Optional override for the config directory to load assets from. If None, uses the default package configuration path.
overrides (dict[str, Any], optional) – Values to merge with loaded assets. For list values, items are appended and deduplicated. For scalar values, they replace defaults.

Returns:

Merged assets keyed by asset stem (defaults, file_mapping, etc.).

Return type:

dict[str, Any]

Raises:

FileNotFoundError – If any expected configuration asset file is not found in config_path.

CONFIG_DIR: ClassVar[str] = 'config'

property config_path: Path

Get the resolved configuration directory path.

Returns the configuration path from the override if set, otherwise computes the path relative to the package module. Logs a warning if the path does not exist.

Returns:: The configuration directory path. May not exist on the package.
Return type:: Path

Notes

A warning is logged if the returned path does not exist on the filesystem.

property defaults_path: Path

Get path to defaults configuration file.

Returns:: Path to defaults.json in config directory
Return type:: Path

property exporter_rules_path: Path

Get path to exporter rules configuration file.

Returns:: Path to exporter_rules.json in config directory
Return type:: Path

property fmap_path: Path

Get path to file mapping configuration file.

Returns:: Path to file_mapping.json in config directory
Return type:: Path

property parser_rules_path: Path

Get path to parser rules configuration file.

Returns:: Path to parser_rules.json in config directory
Return type:: Path

property translation_rules_path: Path

Get path to translation rules configuration file.

Returns:: Path to translation_rules.json in config directory
Return type:: Path

Plugin System¶

class r2x_core.Plugin[source]

Bases: ABC, Generic[ConfigT]

Base class for all plugins with capability-based hooks.

Plugins implement only the hooks they need. The run() method executes hooks in a fixed order, skipping any that aren’t implemented.

Plugin capabilities are determined by which hooks are implemented: - on_build() -> Plugin that creates systems from data - on_transform() -> Plugin that modifies existing systems - on_translate() -> Plugin that converts source → target systems - on_export() -> Plugin that writes systems to files

Required context fields are indicated by property return types: - Non-Optional return type (e.g., -> System) = field is required - Optional return type (e.g., -> System | None) = field is optional

Lifecycle (fixed order, all optional)¶

on_validate() -> Validate inputs/config
on_prepare() -> Load data, setup
on_upgrade() -> Upgrade system to target version
on_build() -> Create system from scratch
on_transform() -> Modify existing system
on_translate() -> Source -> Target system
on_export() -> Write system to files
on_cleanup() -> Cleanup resources

Examples

A simple build plugin:

>>> from r2x_core import Plugin, PluginContext, PluginConfig, System
>>> from rust_ok import Ok

>>> class MyConfig(PluginConfig):
...     name: str

>>> class MyPlugin(Plugin[MyConfig]):
...     def on_build(self):
...         system = System(name=self.config.name)
...         return Ok(system)

>>> ctx = PluginContext(config=MyConfig(name="test"))
>>> plugin = MyPlugin.from_context(ctx)
>>> result_ctx = plugin.run()
>>> result_ctx.system.name
'test'

Config type can be extracted via introspection for discovery:

>>> from typing import get_args, get_origin
>>> config_type = MyPlugin.get_config_type()
>>> config_type.__name__
'MyConfig'

Parameterless init for simple instantiation.

property ctx: PluginContext[ConfigT]

Access the current plugin context.

Returns:: The context passed to from_context() or run().
Return type:: PluginContext[ConfigT]

property config: ConfigT

Plugin configuration. Always required.

Returns:: The plugin configuration from the context.
Return type:: ConfigT

property store: DataStore

Data store for file I/O. Required if return type is non-Optional.

Returns:: The data store from the context.
Return type:: DataStore
Raises:: PluginError – If store is not provided in context.

property system: System

Current system. Required for transform/export plugins.

Returns:: The system from the context.
Return type:: System
Raises:: PluginError – If system is not provided in context.

property source_system: System

Source system for translation. Required for translate plugins.

Returns:: The source system from the context.
Return type:: System
Raises:: PluginError – If source_system is not provided in context.

property target_system: System

Target system after translation.

Returns:: The target system from the context.
Return type:: System
Raises:: PluginError – If target_system is not available in context.

property metadata: dict[str, Any]

Metadata dict. Always available.

Returns:: The metadata dictionary from the context.
Return type:: dict[str, Any]

classmethod from_context(ctx)[source]

Create plugin instance from context.

This is the recommended way to instantiate plugins.

Parameters:: ctx (PluginContext[ConfigT]) – The context to use for plugin execution.
Returns:: New plugin instance with context set.
Return type:: Plugin[ConfigT]

Examples

>>> from r2x_core import Plugin, PluginContext, PluginConfig
>>> class Config(PluginConfig):
...     pass
>>> ctx = PluginContext(config=Config())
>>> plugin = Plugin.from_context(ctx)  # Would instantiate subclass in practice

classmethod get_config_type()[source]

Extract the config type from generic parameters.

Returns:: The PluginConfig subclass used for this plugin.
Return type:: type[Any]

Examples

>>> from r2x_core import Plugin, PluginConfig
>>> class MyConfig(PluginConfig):
...     value: int
>>> class MyPlugin(Plugin[MyConfig]):
...     pass
>>> MyPlugin.get_config_type().__name__
'MyConfig'

classmethod get_implemented_hooks()[source]

Get the set of implemented hook methods.

Returns:: Names of implemented hooks (e.g., {‘on_validate’, ‘on_build’}).
Return type:: set[str]

Examples

>>> from r2x_core import Plugin, PluginConfig
>>> from rust_ok import Ok
>>> class MyConfig(PluginConfig):
...     value: int
>>> class MyPlugin(Plugin[MyConfig]):
...     def on_validate(self):
...         return Ok(None)
...     def on_build(self):
...         return Ok(None)
>>> sorted(MyPlugin.get_implemented_hooks())
['on_build', 'on_validate']

run(*, ctx=None)[source]

Execute plugin lifecycle.

Hooks are called in fixed order. Unimplemented hooks are skipped. If any hook returns Err, execution stops and PluginError is raised. Context is updated in-place for memory efficiency.

Parameters:: ctx (PluginContext[ConfigT] | None) – Context to use. If None, uses context from from_context().
Returns:: The context (updated in-place).
Return type:: PluginContext[ConfigT]
Raises:: PluginError – If any hook fails.

Examples

>>> from r2x_core import Plugin, PluginContext, PluginConfig, System
>>> from rust_ok import Ok

>>> class Config(PluginConfig):
...     pass

>>> class MyPlugin(Plugin[Config]):
...     def on_build(self):
...         return Ok(System(name="test"))

>>> ctx = PluginContext(config=Config())
>>> result = MyPlugin.from_context(ctx).run()
>>> result.system.name
'test'

class r2x_core.PluginContext(config, *, store=None, system=None, source_system=None, target_system=None, rules=(), metadata=None, skip_validation=False, auto_add_composed_components=True, current_version=None, target_version=None, version_strategy=None, preserve_source=False)[source]

Bases: Generic[ConfigT]

Unified context for all plugin operations.

A simple, memory-efficient context that plugins use to access configuration, data stores, systems, and metadata. Uses __slots__ to minimize memory overhead.

config

Plugin configuration (always required). The generic type ConfigT allows type-safe access to plugin-specific config fields.

Type:: ConfigT

store

Optional data store for file I/O operations. Used by plugins that read/write files.

Type:: DataStore | None

system

Optional system object. Created by build plugins, consumed by transform/export plugins, or provided from previous step via stdin.

Type:: System | None

source_system

Optional source system for translation. Used by translate plugins that convert from one format to another.

Type:: System | None

target_system

Optional target system (output of translation). Set after a translate plugin runs.

Type:: System | None

rules

Transformation rules for translation. Empty tuple by default.

Type:: tuple[Rule, …]

metadata

Arbitrary metadata key-value pairs that plugins can use.

Type:: dict[str, Any]

skip_validation

Skip Pydantic validation when creating components (faster but less safe). Parser option. Default is False.

Type:: bool

auto_add_composed_components

Whether to automatically add composed components to the system. Parser option. Default is True.

Type:: bool

current_version

Current version of the system being upgraded. Used by on_upgrade() hooks.

Type:: str | None

target_version

Target version after upgrade. Used by on_upgrade() hooks.

Type:: str | None

version_strategy

Strategy for handling version upgrades. Used by on_upgrade() hooks.

Type:: VersionStrategy | None

Examples

Create a context with configuration:

>>> from r2x_core.plugin_context import PluginContext
>>> from r2x_core.plugin_config import PluginConfig

>>> class MyConfig(PluginConfig):
...     name: str

>>> ctx = PluginContext(config=MyConfig(name="test"))
>>> ctx.config.name
'test'

Update context fields directly:

>>> from r2x_core.system import System
>>> system = System(name="my_system")
>>> ctx.system = system
>>> ctx.system.name
'my_system'

Initialize plugin context.

Parameters:

config (ConfigT) – Plugin configuration (required).
store (DataStore | None) – Optional data store for file I/O.
system (System | None) – Optional system object.
source_system (System | None) – Optional source system for translation.
target_system (System | None) – Optional target system.
rules (tuple[Rule, ...]) – Transformation rules. Default is empty tuple.
metadata (dict[str, Any] | None) – Arbitrary metadata. Default is empty dict.
skip_validation (bool) – Skip Pydantic validation. Default is False.
auto_add_composed_components (bool) – Auto-add composed components. Default is True.
current_version (str | None) – Current version for upgrades. Default is None.
target_version (str | None) – Target version for upgrades. Default is None.
version_strategy (VersionStrategy | None) – Version strategy for upgrades. Default is None.
preserve_source (bool) – When True, translation captures source-side provenance so a reverse translation can reproduce the original source. Every translated target component is tagged with a SourceProvenance SA recording its source UUID, and every source component (plus its SAs) with no matching rule is copied into the target as a preserved carry-over. Default is False.

evolve(**kwargs)[source]

Create a new context with updated fields.

Memory-efficient method to create a new context with only the specified fields changed. All other fields are copied from the current context.

Parameters:: **kwargs (Any) – Fields to update (config, store, system, source_system, target_system, rules, metadata, skip_validation, auto_add_composed_components, current_version, target_version, version_strategy, preserve_source).
Returns:: A new context with the specified updates.
Return type:: PluginContext[ConfigT]

Examples

Create a new context with updated system:

>>> from r2x_core.plugin_context import PluginContext
>>> from r2x_core.plugin_config import PluginConfig
>>> from r2x_core.system import System

>>> class MyConfig(PluginConfig):
...     pass

>>> ctx = PluginContext(config=MyConfig())
>>> system = System(name="my_system")
>>> ctx2 = ctx.evolve(system=system)
>>> ctx2.system.name
'my_system'

Update metadata:

>>> ctx3 = ctx2.evolve(metadata={"key": "value"})
>>> ctx3.metadata
{'key': 'value'}

list_rules()[source]

List all available transformation rules.

Returns:: List of all rules in this context.
Return type:: list[Rule]

get_rule(source_type, *, target_type, version=None)[source]

Retrieve a transformation rule by source type, target type, and version.

Parameters:

source_type (str) – Source component type name.
target_type (str) – Target component type name.
version (int | None) – Rule version. If None, uses active_versions from config if available, otherwise defaults to version 1.

Returns:

The matching rule.

Return type:

Rule

Raises:

KeyError – If no rule matches the specified source/target types and version.

list_available_conversions()[source]

List available conversions by source type.

Returns:: Mapping of source types to list of ConversionOption objects, sorted by target type and version.
Return type:: dict[str, list[ConversionOption]]

get_rules_for_source(source_type)[source]

Get all rules for a specific source type.

Parameters:: source_type (str) – Source component type name.
Returns:: List of all rules matching the source type, sorted by target type and version.
Return type:: list[Rule]

get_rules_for_conversion(source_type, *, target_type)[source]

Get all versions of a conversion between two types.

Parameters:

source_type (str) – Source component type name.
target_type (str) – Target component type name.

Returns:

List of all rules for this conversion, sorted by version.

Return type:

list[Rule]

File Types¶

class r2x_core.FileFormat[source]

Bases: object

Lightweight base class for file format types.

This is a minimal sentinel class designed to work with singledispatch. Subclasses act as type markers for dispatch without storing instance data.

supports_timeseries

Whether this file format can store time series data. Default is False.

Type:: bool

supports_timeseries: ClassVar[bool] = False

class r2x_core.H5Format[source]

Bases: FileFormat

HDF5 data format.

Supports time series data storage with hierarchical organization. Use reader_kwargs to configure how the file is read.

supports_timeseries: ClassVar[bool] = True

Rules and Translation¶

class r2x_core.Rule(source_type, target_type, version, field_map=<factory>, getters=<factory>, defaults=<factory>, filter=None, system='source', name=None, depends_on=None, supplemental_attributes=<factory>)[source]¶

Declarative rule for converting one component type to another.

Parameters:

source_type (str | list[str])
target_type (str | list[str])
version (int)
field_map (dict[str, str | list[str]])
getters (dict[str, Callable[[...], Result[Any, ValueError]] | str])
defaults (dict[str, Any])
filter (RuleFilter | None)
system (Literal['source', 'target'])
name (str | None)
depends_on (list[str] | None)
supplemental_attributes (list[SupplementalAttributeRule | dict[str, Any]])

has_multiple_sources()[source]¶

Check if rule applies to multiple source types.

Return type:: bool

has_multiple_targets()[source]¶

Check if rule creates multiple target types.

Return type:: bool

get_source_types()[source]¶

Return source types as list.

Return type:: list[str]

get_target_types()[source]¶

Return target types as list.

Return type:: list[str]

classmethod from_records(records)[source]¶

Create rules from json objects.

Parameters:: records (list[dict[str, Any]])
Return type:: list[Rule]

class r2x_core.RuleFilter(*, field=None, getter=None, op=None, values=None, prefixes=None, any_of=None, all_of=None, casefold=True, on_missing='exclude')[source]¶

Declarative predicate for selecting source components.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

field (str | None)
getter (Callable[[...], Result[Any, ValueError]] | str | None)
op (Literal['eq', 'neq', 'in', 'not_in', 'geq', 'startswith', 'not_startswith', 'endswith'] | None)
values (list[Any] | None)
prefixes (list[str] | None)
any_of (list[RuleFilter] | None)
all_of (list[RuleFilter] | None)
casefold (bool)
on_missing (Literal['include', 'exclude'])

matches(component, *, context=None)[source]¶

Evaluate this filter against a component instance.

Parameters:

component (Any)
context (PluginContext | None)

Return type:

bool

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)¶

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

self (BaseModel) – The BaseModel instance.
context (Any) – The context.

Return type:

None

Getters¶

r2x_core.getter(func=None, *, name=None)[source]¶

Decorate a callable getter by name.

Can be used in three ways: 1. @getter - Uses function name as registry key 2. @getter() - Uses function name as registry key (parentheses optional) 3. @getter(name=”custom_name”) - Uses custom name as registry key (name kwarg required when parentheses used)

When using parentheses, the name parameter must be provided as a keyword argument.

Parameters:

func (F | None)
name (str | None)

Return type:

F | Callable[[F], F]

>>> from r2x_core import getter, resolve_getter
>>>
>>> @getter(name="api_resolve_getter")
... def api_resolve_getter(component, *, context):
...     _ = component
...     _ = context
...     return "resolved"
>>>
>>> resolved = resolve_getter("api_resolve_getter")
>>> resolved.is_ok()
True
>>> resolved.unwrap() is api_resolve_getter
True

r2x_core.resolve_getter(getter)[source]¶

Resolve a getter reference to a callable.

Parameters:: getter (GetterFunc | str)
Return type:: Result[GetterFunc, TypeError]

r2x_core.replace_single_time_series(system, owner, time_series, **features)[source]¶

Replace exactly one time series association without removing shared data.

The metadata change is rolled back if storing the replacement fails. This preserves the existing association and any backing data shared with another system.

Parameters:

system (System)
owner (Component)
time_series (SingleTimeSeries)
features (Any)

Return type:

None

r2x_core.transfer_time_series_metadata(context)[source]¶

Transfer time series metadata for target system.

Parameters:: context (PluginContext) – Plugin context containing source and target systems.
Return type:: TimeSeriesTransferResult

Notes

Preserved components (those carried over from source when preserve_source=True) live in the target’s component map under their original source UUIDs, so their time series transfer through the normal UUID-matched path with no special handling.

class r2x_core.time_series.TimeSeriesTransferResult(transferred, updated, children_remapped)[source]

Bases: object

Transfer status of time series metadata.

Parameters:

transferred (int)
updated (int)
children_remapped (int)

r2x_core.utils.filter_valid_kwargs(func, *, kwargs)[source]¶

Filter function kwargs, maintaining backward compatibility.

Parameters:

func (Callable[[...], Any])
kwargs (dict[str, Any])

Return type:

dict[str, Any]

r2x_core.utils.validate_file_extension(path, *, info)[source]¶

Validate that the file path has a supported extension.

This is a Pydantic validator that checks if the file extension from the provided path exists as a key in the module-level EXTENSION_MAPPING.

Parameters:

value (str) – The file path string to validate, provided by Pydantic.
info (pydantic.ValidationInfo) – Pydantic’s validation context. Required by the validator signature but not used in this function.
path (Path)

Returns:

The original file path string if its extension is valid.

Return type:

Path

Raises:

AssertionError – If the input value is not a string.
ValueError – If the file path has no extension.
KeyError – If the file’s extension is not found in EXTENSION_MAPPING.

Notes

This function is intended for use as a Pydantic model validator (e.g., with @field_validator or AfterValidator) and should not be called directly.

r2x_core.utils.export_components_to_csv(system, *, file_path, filter_func=None, fields=None, key_mapping=None, **dict_writer_kwargs)[source]¶

Export all components or filtered components to a CSV file.

Parameters:

system (System) – The system to export components from.
file_path (Path | str) – Output CSV file path.
filter_func (Callable, optional) – Function to filter components by. Exports all if None.
fields (list, optional) – List of field names to include. Exports all if None.
key_mapping (dict, optional) – Mapping of component field names to CSV column names.
**dict_writer_kwargs – Additional arguments passed to csv.DictWriter.

Return type:

None

r2x_core.utils.components_to_records(system, *, filter_func=None, fields=None, key_mapping=None)[source]¶

Convert system components to a list of dictionaries (records).

Parameters:

system (System) – The system to extract components from.
filter_func (Callable, optional) – Function to filter components. Should accept a component and return bool.
fields (list, optional) – List of field names to include. If None, includes all fields.
key_mapping (dict, optional) – Dictionary mapping component field names to record keys.

Returns:

List of component records as dictionaries.

Return type:

list[dict[str, Any]]

r2x_core.utils.create_component(component_class, *, skip_none=True, skip_validation=False, **field_values)[source]¶

Create and validate a component instance with optional validation skipping.

This utility function returns a Result to allow recovery from validation errors (e.g., skipping invalid components, logging warnings, etc.).

Parameters:

component_class (type[T]) – The component class to instantiate.
skip_none (bool, default True) – Whether to skip fields with None values when creating the component.
skip_validation (bool, default False) – Skip Pydantic validation when creating components (faster but less safe).
**field_values (Any) – Field values to pass to the component constructor.

Returns:

Ok(component) if validation succeeds, or Err(validation_error) if validation fails. Returns the specific component type T.

Return type:

Result[T, PydanticValidationError]

Examples

>>> from infrasys import Generator
>>> result = create_component(Generator, name="Gen1", capacity=100.0)
>>> if result.is_ok():
...     gen = result.value
...     print(gen.name)
>>> else:
...     print(f"Error: {result.error}")

r2x_core.utils.run_upgrade_step(data, *, step, upgrader_context=None)[source]¶

Execute a single upgrade transformation on data.

Automatically detects whether the step function accepts an upgrader_context parameter via introspection.

Parameters:

data (Any)
step (UpgradeStep)
upgrader_context (Any | None)

Return type:

Result[Any, str]

Results¶

class r2x_core.Result[source]

Bases: Generic[T_co, E_co]

Base type for Ok/Err results.

unwrap()[source]

Return the contained value if successful, else raise in subclass.

Return type:: T_co

unwrap_err()[source]

Return the contained error if Err, else raise in subclass.

Return type:: E_co

unwrap_or(default)[source]

Return the contained value if Ok, otherwise return the default.

Parameters:: default (T_co | Any)
Return type:: T_co

unwrap_or_else(func)[source]

Return the contained value if Ok, otherwise compute a default.

Parameters:: func (Callable[[Any], T_co])
Return type:: T_co

expect(msg)[source]

Return the contained value if Ok, otherwise raise with custom message.

Parameters:: msg (str)
Return type:: T_co

is_ok()[source]

Return True if this is Ok.

Return type:: bool

is_err()[source]

Return True if this is Err.

Return type:: bool

map(func)[source]

Apply func to the contained value if Ok, returning a new Result.

Parameters:: func (Callable[[Any], U])
Return type:: Result[U, E_co]

map_err(func)[source]

Apply func to the error if Err, returning a new Result.

Parameters:: func (Callable[[Any], F])
Return type:: Result[T_co, F]

and_then(func)[source]

Chain another computation on the contained value if Ok.

Parameters:: func (Callable[[Any], Result[U, E_co]])
Return type:: Result[U, E_co]

or_else(func)[source]

Handle the error by calling func if Err, returning a new Result.

Parameters:: func (Callable[[Any], Result[T_co, F]])
Return type:: Result[T_co, F]

ok()[source]

Return the success value if Ok, otherwise None.

Return type:: T_co | None

err()[source]

Return the error value if Err, otherwise raise in subclass.

Return type:: E_co

property error: E_co | None: Return the error value if Err, otherwise None.

format_error()[source]

Return a formatted string of the error, including traceback if available.

Return type:: str

unwrap_or_raise(exc_type=<class 'Exception'>, context=None)[source]

Return the Ok value or raise exc_type.

Parameters:

exc_type (type[BaseException])
context (str | None)

Return type:

T_co

class r2x_core.Ok(value: T_co)[source]

class r2x_core.Ok

class r2x_core.Ok(value: Literal[None])

Bases: Result[T_co, E_co]

Success result containing a value.

Parameters:: value (T_co)

unwrap()[source]

Return the contained value if successful, else raise in subclass.

Return type:: T_co

unwrap_err()[source]

Return the contained error if Err, else raise in subclass.

Return type:: E_co

unwrap_or(default)[source]

Return the contained value if Ok, otherwise return the default.

Parameters:: default (T_co | Any)
Return type:: T_co

unwrap_or_else(func)[source]

Return value; func is never called in Ok.

Parameters:: func (Callable[[Any], T_co])
Return type:: T_co

expect(msg)[source]

Return the contained value if Ok, otherwise raise with custom message.

Parameters:: msg (str)
Return type:: T_co

is_ok()[source]

Return True if this is Ok.

Return type:: bool

is_err()[source]

Return True if this is Err.

Return type:: bool

map(func)[source]

Apply func to the contained value if Ok, returning a new Result.

Parameters:: func (Callable[[Any], U])
Return type:: Result[U, E_co]

map_err(func)[source]

Phantom type: func is never called in Ok.

Parameters:: func (Callable[[Any], F])
Return type:: Result[T_co, F]

and_then(func)[source]

Chain another computation on the contained value if Ok.

Parameters:: func (Callable[[Any], Result[U, E_co]])
Return type:: Result[U, E_co]

or_else(func)[source]

Phantom type: func is never called in Ok.

Parameters:: func (Callable[[Any], Result[T_co, F]])
Return type:: Result[T_co, F]

ok()[source]

Return the success value if Ok, otherwise None.

Return type:: T_co

err()[source]

Return the error value if Err, otherwise raise in subclass.

Return type:: E_co

format_error()[source]

Return a formatted string of the error, including traceback if available.

Return type:: str

unwrap_or_raise(exc_type=<class 'Exception'>, context=None)[source]

Return the Ok value or raise exc_type.

Parameters:

exc_type (type[BaseException])
context (str | None)

Return type:

T_co

class r2x_core.Err(error: E_co)[source]

class r2x_core.Err(error: Any)

Bases: Result[T_co, E_co]

Error result containing an error value.

Parameters:: error (E_co | Any)

unwrap()[source]

Return the contained value if successful, else raise in subclass.

Return type:: T_co

unwrap_err()[source]

Return the contained error if Err, else raise in subclass.

Return type:: E_co

unwrap_or(default)[source]

Return the contained value if Ok, otherwise return the default.

Parameters:: default (T_co | Any)
Return type:: T_co

unwrap_or_else(func)[source]

Return the contained value if Ok, otherwise compute a default.

Parameters:: func (Callable[[Any], T_co])
Return type:: T_co

expect(msg)[source]

Return the contained value if Ok, otherwise raise with custom message.

Parameters:: msg (str)
Return type:: T_co

is_ok()[source]

Return True if this is Ok.

Return type:: bool

is_err()[source]

Return True if this is Err.

Return type:: bool

map(func)[source]

Return Err; func is never called.

Parameters:: func (Callable[[Any], U])
Return type:: Result[U, E_co]

map_err(func)[source]

Apply func to the error if Err, returning a new Result.

Parameters:: func (Callable[[Any], F])
Return type:: Result[T_co, F]

and_then(func)[source]

Return Err; func is never called.

Parameters:: func (Callable[[Any], Result[U, E_co]])
Return type:: Result[U, E_co]

or_else(func)[source]

Handle the error by calling func if Err, returning a new Result.

Parameters:: func (Callable[[Any], Result[T_co, F]])
Return type:: Result[T_co, F]

ok()[source]

Return the success value if Ok, otherwise None.

Return type:: T_co | None

err()[source]

Return the error value if Err, otherwise raise in subclass.

Return type:: E_co

format_error()[source]

Return a formatted string of the error, including traceback if available.

Return type:: str

unwrap_or_raise(exc_type=<class 'Exception'>, context=None)[source]

Return the Ok value or raise exc_type.

Parameters:

exc_type (type[BaseException])
context (str | None)

Return type:

T_co

r2x_core.is_ok(result)[source]¶

Return True if the result is Ok.

Parameters:: result (Result[T, E])
Return type:: TypeGuard[Ok[T, E]]

r2x_core.is_err(result)[source]¶

Return True if the result is Err.

Parameters:: result (Result[T, E])
Return type:: TypeGuard[Err[T, E]]

Data Processors¶

See Data Processors for complete processor documentation.