API Reference¶

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

System¶

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

Bases: System

R2X Core System class extending infrasys.System.

This class extends infrasys.System to provide R2X-specific functionality for data model translation and system construction. It maintains compatibility with infrasys while adding convenience methods for component export and system manipulation.

The System serves as the central data store for all components (buses, generators, branches, etc.) and their associated time series data. It provides methods for: - Adding and retrieving components - Managing time series data - Serialization/deserialization (JSON) - Exporting components to various formats (CSV, records, etc.)

Parameters:

name (str) – Unique identifier for the system.
description (str, optional) – Human-readable description of the system.
auto_add_composed_components (bool, default True) – If True, automatically add composed components (e.g., when adding a Generator with a Bus, automatically add the Bus to the system if not already present).
base_power (float | None)
kwargs (Any)

name¶

System identifier.

Type:: str

description¶

System description.

Type:: str

Examples

Create a basic system:

>>> from r2x_core import System
>>> system = System(name="MySystem", description="Test system")

Create a system with auto-add for composed components:

>>> system = System(name="MySystem", auto_add_composed_components=True)

Add components to the system:

>>> from infrasys import Component
>>> # Assuming you have component classes defined
>>> bus = ACBus(name="Bus1", voltage=230.0)
>>> system.add_component(bus)

Serialize and deserialize:

>>> system.to_json("system.json")
>>> loaded_system = System.from_json("system.json")

See also

infrasys.system.System: Parent class providing core system functionality
r2x_core.parser.BaseParser: Parser framework for building systems

Notes

This class maintains backward compatibility with the legacy r2x.api.System while being simplified for r2x-core’s focused scope. The main differences:

Legacy r2x.api.System: Full-featured with CSV export, filtering, version tracking
r2x-core.System: Lightweight wrapper focusing on system construction and serialization

The r2x-core.System delegates most functionality to infrasys.System, adding only R2X-specific enhancements as needed.

Initialize R2X Core System.

Parameters:

system_base_power (float, optional) – System base power in MVA for per-unit calculations. Can be provided as first positional argument or as keyword argument. Default is 100.0 MVA if not provided.
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).
base_power (float | None)

Examples

Various ways to create a system:

>>> System()  # Uses defaults: name=auto, system_base_power=100.0
>>> System(200.0)  # Positional: system_base_power=200.0
>>> System(200.0, name="MySystem")  # Both
>>> System(name="MySystem")  # Name only
>>> System(system_base_power=200.0, name="MySystem")  # Both as keywords
>>> System(name="MySystem", system_base_power=200.0)  # Order doesn't matter

Notes

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.

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(filename=None, overwrite=False, indent=None, data=None)[source]¶

Serialize system to JSON file or stdout.

Parameters:

filename (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.

Examples

>>> system.to_json("output/system.json", overwrite=True, indent=2)
>>> system.to_json()  # Print to stdout

See also

from_json: Load system from JSON file

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

Deserialize system from JSON file.

Parameters:

filename (Path or str) – Input JSON file path.
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.

Examples

>>> system = System.from_json("input/system.json")

With version upgrade handling:

>>> def upgrade_v1_to_v2(data):
...     # Custom upgrade logic
...     return data
>>> system = System.from_json("old_system.json", upgrade_handler=upgrade_v1_to_v2)

See also

to_json: Serialize system to JSON file

serialize_system_attributes()[source]¶

Serialize R2X-specific system attributes.

Returns:: Dictionary containing system_base_power.
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

components_to_records(filter_func=None, fields=None, key_mapping=None)[source]¶

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

This method retrieves components from the system and converts them to dictionary records, with optional filtering, field selection, and key mapping.

Parameters:

filter_func (Callable, optional) – Function to filter components. Should accept a component and return bool. If None, converts all components in the system.
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]]

Examples

Get all components as records:

>>> records = system.components_to_records()

Get only generators:

>>> from my_components import Generator
>>> records = system.components_to_records(
...     filter_func=lambda c: isinstance(c, Generator)
... )

Get specific fields with renamed keys:

>>> records = system.components_to_records(
...     fields=["name", "voltage"],
...     key_mapping={"voltage": "voltage_kv"}
... )

See also

export_components_to_csv: Export components to CSV file
get_components: Retrieve components by type with filtering

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

Export all components or filtered components to CSV file.

This method exports components from the system to a CSV file. You can optionally provide a filter function to select specific components.

Parameters:

file_path (PathLike) – Output CSV file path.
filter_func (Callable, optional) – Function to filter components. Should accept a component and return bool. If None, exports all components in the system.
fields (list, optional) – List of field names to include. If None, exports all fields.
key_mapping (dict, optional) – Dictionary mapping component field names to CSV column names.
**dict_writer_kwargs – Additional arguments passed to csv.DictWriter.

Return type:

None

Examples

Export all components:

>>> system.export_components_to_csv("all_components.csv")

Export only generators using a filter:

>>> from my_components import Generator
>>> system.export_components_to_csv(
...     "generators.csv",
...     filter_func=lambda c: isinstance(c, Generator)
... )

Export buses with custom filter:

>>> from my_components import ACBus
>>> system.export_components_to_csv(
...     "high_voltage_buses.csv",
...     filter_func=lambda c: isinstance(c, ACBus) and c.voltage > 100
... )

Export with field selection and renaming:

>>> system.export_components_to_csv(
...     "buses.csv",
...     filter_func=lambda c: isinstance(c, ACBus),
...     fields=["name", "voltage"],
...     key_mapping={"voltage": "voltage_kv"}
... )

See also

components_to_records: Convert components to dictionary records
get_components: Retrieve components by type with filtering

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

base: str | None = None¶

unit: str¶

class r2x_core.units.UnitSystem(*values)[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(folder=None, reader=None)[source]¶

Container for managing data file mappings and loading data.

The DataStore class provides a centralized interface for managing collections of data files, their metadata, and coordinating data loading operations. It maintains a registry of DataFile instances and delegates actual file reading operations to a DataReader instance.

Parameters:

folder (str or Path, optional) – Base directory containing the data files. If None, uses current working directory.
reader (DataReader, optional) – Custom data reader instance for handling file I/O operations. If None, creates a default DataReader instance.

folder¶

Resolved absolute path to the base data directory.

Type:: Path

reader¶

Data reader instance used for file operations.

Type:: DataReader

Examples

Create a basic data store:

>>> store = DataStore(folder="/path/to/data")
>>> data_file = DataFile(name="generators", fpath="gen_data.csv")
>>> store.add_data_file(data_file)
>>> data = store.read_data_file("generators")

Load from JSON configuration:

>>> store = DataStore.from_json("config.json", folder="/path/to/data")
>>> files = store.list_data_files()
>>> print(files)
['generators', 'transmission', 'load']

Batch operations:

>>> files = [DataFile(name="gen", fpath="gen.csv"), DataFile(name="load", fpath="load.csv")]
>>> store.add_data_files(files)
>>> store.remove_data_files(["gen", "load"])

See also

DataFile: Data file metadata and configuration
DataReader: File reading and processing operations

Notes

The DataStore maintains DataFile metadata in memory but delegates actual file reading to the DataReader, which may implement its own caching strategies. The store itself does not cache file contents, only the DataFile configurations.

Initialize the DataStore.

Parameters:

folder (str | Path | None, optional) – Base directory containing the data files. If None, uses current working directory. Default is None.
reader (DataReader | None, optional) – Custom data reader instance for handling file I/O operations. If None, creates a default DataReader instance. Default is None.

Raises:

FileNotFoundError – If the specified folder does not exist.

classmethod from_plugin_config(config, folder)[source]¶

Create a DataStore instance from a PluginConfig.

This is a convenience constructor that automatically discovers and loads the file mapping JSON associated with a plugin configuration class.

Parameters:

config (PluginConfig) – Plugin configuration instance. The file mapping path will be discovered from the config class using get_file_mapping_path().
folder (Path or str) – Base directory containing the data files referenced in the configuration.

Returns:

A new DataStore instance populated with DataFile configurations from the plugin’s file mapping.

Return type:

DataStore

Raises:

FileNotFoundError – If the configuration file does not exist or if data files are missing.
TypeError – If the JSON file does not contain a valid array structure.
ValidationError – If any DataFile configuration in the JSON is invalid.

Examples

Simple usage:

>>> from r2x_reeds.config import ReEDSConfig
>>> config = ReEDSConfig(solve_year=2030, weather_year=2012)
>>> store = DataStore.from_plugin_config(config, folder="/data/reeds")
>>> store.list_data_files()
['generators', 'buses', 'transmission']

See also

from_json: Create from an explicit JSON file path
PluginConfig.get_file_mapping_path: Get the file mapping path

Notes

This method provides a cleaner API than manually calling config.get_file_mapping_path() and then DataStore.from_json(). It’s the recommended way to create a DataStore for plugin-based workflows.

classmethod from_json(fpath, folder)[source]¶

Create a DataStore instance from a JSON configuration file.

Parameters:

fpath (Path or str) – Path to the JSON configuration file containing DataFile specifications.
folder (Path or str) – Base directory containing the data files referenced in the configuration.

Returns:

A new DataStore instance populated with DataFile configurations from the JSON file.

Return type:

DataStore

Raises:

FileNotFoundError – If the configuration file does not exist.
TypeError – If the JSON file does not contain a valid array structure.
ValidationError – If any DataFile configuration in the JSON is invalid (raised by Pydantic during DataFile creation).
KeyError – If any data file names are duplicated (raised during add_data_files).

Examples

Create a JSON configuration file:

>>> config = [
...     {"name": "generators", "fpath": "gen_data.csv", "description": "Generator capacity data"},
...     {"name": "load", "fpath": "load_data.csv", "description": "Load profiles"},
... ]
>>> import json
>>> with open("config.json", "w") as f:
...     json.dump(config, f)

Load the DataStore:

>>> store = DataStore.from_json("config.json", "/path/to/data")
>>> store.list_data_files()
['generators', 'load']

See also

to_json: Save DataStore configuration to JSON
DataFile: Individual data file configuration structure

Notes

The JSON file must contain an array of objects, where each object represents a valid DataFile configuration with at minimum ‘name’ and ‘fpath’ fields.

add_data_file(data_file, overwrite=False)[source]¶

Add a single data file to the store.

Parameters:

data_file (DataFile) – The data file configuration to add to the store.
overwrite (bool, optional) – Whether to overwrite an existing file with the same name. Default is False.

Raises:

KeyError – If a file with the same name already exists and overwrite is False.

Return type:

None

Examples

>>> store = DataStore("/path/to/data")
>>> data_file = DataFile(name="generators", fpath="gen_data.csv")
>>> store.add_data_file(data_file)

>>> # Overwrite existing file
>>> new_data_file = DataFile(name="generators", fpath="new_gen_data.csv")
>>> store.add_data_file(new_data_file, overwrite=True)

See also

add_data_files: Add multiple data files at once
remove_data_file: Remove a data file from the store

add_data_files(data_files, overwrite=False)[source]¶

Add multiple data files to the store.

Parameters:

data_files (Iterable[DataFile]) – Collection of data file configurations to add to the store.
overwrite (bool, optional) – Whether to overwrite existing files with the same names. Default is False.

Raises:

KeyError – If any file with the same name already exists and overwrite is False.

Return type:

None

Examples

>>> store = DataStore("/path/to/data")
>>> files = [
...     DataFile(name="generators", fpath="gen.csv"),
...     DataFile(name="transmission", fpath="trans.csv"),
...     DataFile(name="load", fpath="load.csv")
... ]
>>> store.add_data_files(files)

See also

add_data_file: Add a single data file
remove_data_files: Remove multiple data files

remove_data_file(name)[source]¶

Remove a data file from the store.

Parameters:: name (str) – Name of the data file to remove.
Raises:: KeyError – If the specified file name is not present in the store.
Return type:: None

Examples

>>> store = DataStore("/path/to/data")
>>> data_file = DataFile(name="generators", fpath="gen.csv")
>>> store.add_data_file(data_file)
>>> store.remove_data_file("generators")

See also

remove_data_files: Remove multiple data files at once
add_data_file: Add a data file to the store

remove_data_files(names)[source]¶

Remove multiple data files from the store.

Parameters:: names (Iterable[str]) – Collection of data file names to remove.
Raises:: KeyError – If any specified file name is not present in the store.
Return type:: None

Examples

>>> store = DataStore("/path/to/data")
>>> files = [
...     DataFile(name="gen", fpath="gen.csv"),
...     DataFile(name="load", fpath="load.csv")
... ]
>>> store.add_data_files(files)
>>> store.remove_data_files(["gen", "load"])

See also

remove_data_file: Remove a single data file
add_data_files: Add multiple data files

get_data_file_by_name(name)[source]¶

Retrieve a data file configuration by name.

Parameters:: name (str) – Name of the data file to retrieve.
Returns:: The data file configuration object.
Return type:: DataFile
Raises:: KeyError – If the specified file name is not present in the store.

Examples

>>> store = DataStore("/path/to/data")
>>> data_file = store.get_data_file_by_name("generators")
>>> print(data_file.fpath)
generators.csv

See also

list_data_files: Get all data file names
read_data_file: Load the actual file contents

list_data_files()[source]¶

List all data file names in the store.

Returns:: Sorted list of all data file names present in the store.
Return type:: list[str]

Examples

>>> store = DataStore("/path/to/data")
>>> files = [
...     DataFile(name="generators", fpath="gen.csv"),
...     DataFile(name="load", fpath="load.csv")
... ]
>>> store.add_data_files(files)
>>> store.list_data_files()
['generators', 'load']

See also

get_data_file_by_name: Get a specific data file configuration
__contains__: Check if a specific file exists

read_data_file(*, name, use_cache=True)[source]¶

Load data from a file using the configured reader.

Parameters:

name (str) – Name of the data file to load.
use_cache (bool, optional) – Whether to use cached data if available. Default is True.

Returns:

The loaded data, type depends on file type and reader configuration.

Return type:

Any

Raises:

KeyError – If the specified file name is not present in the store.
FileNotFoundError – If the file does not exist and is not marked as optional.

Examples

>>> store = DataStore("/path/to/data")
>>> data_file = DataFile(name="generators", fpath="gen.csv")
>>> store.add_data_file(data_file)
>>> data = store.read_data_file("generators")

See also

get_data_file_by_name: Get the file configuration
clear_cache: Clear the reader’s cache

clear_cache()[source]¶

Clear both the data reader’s cache and the data store’s file configurations.

This method clears the underlying DataReader’s cache of loaded file contents and also removes all data file configurations from the DataStore.

Examples

>>> store = DataStore("/path/to/data")
>>> # Load some data files...
>>> store.clear_cache()  # Clear cached file contents and configurations

See also

reader: Access the underlying DataReader instance

Return type:: None

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

Save the DataStore configuration to a JSON file.

Parameters:

fpath (str or Path) – Path where the JSON configuration file will be saved.
**model_dump_kwargs (dict[str, Any]) – Additional keyword arguments passed to the DataFile.model_dump method for controlling serialization behavior.

Return type:

None

Examples

>>> store = DataStore("/path/to/data")
>>> files = [DataFile(name="generators", fpath="gen.csv"), DataFile(name="load", fpath="load.csv")]
>>> store.add_data_files(files)
>>> store.to_json("config.json")

>>> # Save with custom serialization options
>>> store.to_json("config.json", exclude_none=True)

See also

from_json: Load DataStore configuration from JSON
DataFile.model_dump: Individual file serialization method

Notes

The resulting JSON file will contain an array of DataFile configurations that can be loaded back using the from_json class method.

property reader: DataReader¶

Get the data reader instance.

Returns:: The configured data reader instance.
Return type:: DataReader

Examples

>>> store = DataStore("/path/to/data")
>>> reader = store.reader
>>> reader.clear_cache()

class r2x_core.DataReader(max_cache_size=100)[source]¶

Reader class for loading data files with caching support.

The DataReader handles the actual file I/O operations and caching strategies, while delegating file-type-specific reading logic to single dispatch methods.

Parameters:: max_cache_size (int, optional) – Maximum number of files to keep in cache. Default is 100.

max_cache_size¶

Maximum cache size limit.

Type:: int

Initialize the data reader with cache configuration.

Parameters:: max_cache_size (int, optional) – Maximum number of files to keep in cache. Default is 100.

read_data_file(folder, data_file, use_cache=True)[source]¶

Read a data file using cache if available.

Parameters:

folder (Path) – Base directory containing the data files.
data_file (DataFile) – Data file configuration with metadata.
use_cache (bool, optional) – Whether to use cached data if available. Default is True.

Returns:

The loaded data, type depends on file type.

Return type:

Any

Raises:

FileNotFoundError – If the file does not exist and is not optional.

clear_cache()[source]¶

Clear cached files.

Parameters:: pattern (str, optional) – If provided, only clear cache entries matching this pattern. If None, clears all cached data.
Return type:: None

get_cache_info()[source]¶

Get information about the current cache state.

Returns:: Cache statistics and information.
Return type:: dict[str, Any]

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.

Return type:

None

Examples

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

pydantic model r2x_core.DataFile[source]¶

DataModel class for data files.

This class defines how individual data files should be read, processed, and filtered within the R2X framework. It uses Pydantic for validation and automatic type conversion.

Parameters:

name (str) – Unique identifier for this file mapping configuration.
fpath (pathlib.Path) – Path to the data file relative to the ReEDS case directory. Must have a supported extension (.csv, .tsv, .h5, .hdf5, .json, .xml).
description (str, optional) – Human-readable description of the data file contents.
is_input (bool, default True) – Whether this file represents input data (True) or output data (False).
is_optional (bool, default False) – Whether the file is optional. If True, missing files will not raise errors.
is_timeseries (bool, default False) – Whether the file contains time series data. Time series files must use formats that support time series (CSV, TSV, HDF5, Parquet). Files marked as time series with unsupported formats will raise a validation error.
units (str, optional) – Physical units for numeric data in the file (e.g., “MW”, “$/MWh”).
reader_function (Callable[[Path], Any], optional) – Custom reader function (callable) to use instead of the default file type reader. The function should accept a Path argument and return the loaded data.
column_mapping (dict[str, str], optional) – Mapping of original column names to desired column names as {old_name: new_name}.
index_columns (list[str], optional) – List of column names to treat as index columns when selecting data.
value_columns (list[str], optional) – List of column names containing the actual data values to retain.
drop_columns (list[str], optional) – List of column names to remove from the data after loading.
column_schema (dict[str, str], optional) – Schema defining column names and types as {column_name: type_string}. Used when the input file lacks headers. Type strings: “string”, “int”, “float”.
filter_by (dict[str, Any], optional) – Row-level filters to apply as {column_name: value_or_list}. Supports special values “solve_year” and “weather_year”.
pivot_on (str, optional) – Column name to pivot the data on (for reshaping operations).
aggregate_function (str, optional) – Function name for aggregating data after pivoting.

file_type¶

Computed property that returns the appropriate FileFormat class based on the file extension. Automatically determined from fpath.suffix.

Type:: FileFormat

Examples

Basic file mapping for a CSV file:

>>> mapping = DataFile(
...     name="generation_data",
...     fpath="outputs/gen_h.csv",
...     description="Hourly generation by technology",
...     units="MWh",
... )
>>> mapping.file_type
<class 'TableFormat'>

File mapping with column operations:

>>> mapping = DataFile(
...     name="capacity_data",
...     fpath="inputs/cap_tech.csv",
...     column_mapping={"old_tech": "technology", "cap_mw": "capacity"},
...     drop_columns=["unused_col"],
...     filter_by={"year": 2030, "region": ["CA", "TX"]},
... )

File mapping with custom reader function:

>>> from plexosdb import PlexosDB
>>> mapping = DataFile(
...     name="plexos_data",
...     fpath="model.xml",
...     reader_function=PlexosDB.from_xml,  # Callable function
... )

Optional file with lambda reader:

>>> mapping = DataFile(
...     name="simple_text",
...     fpath="data.txt",
...     is_optional=True,
...     reader_function=lambda p: p.read_text().strip().split(r"\n"),
...     column_schema={"line": "string"},
... )

Notes

File paths are validated to ensure they have supported extensions
The file_type property is computed automatically and excluded from serialization
Column operations are applied in order: mapping → dropping → schema → filtering

See also

FileFormat: Class for file formats.
DataStore: Container for managing multiple DataFile instances
DataReader: Service class for actually loading and processing the 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.

Fields:

aggregate_function (str | None)
column_mapping (dict[str, str] | None)
column_schema (dict[str, str] | None)
description (str | None)
drop_columns (list[str] | None)
filter_by (dict[str, Any] | None)
fpath (pathlib.Path)
index_columns (list[str] | None)
is_input (bool)
is_optional (bool)
is_timeseries (bool)
key_mapping (dict[str, str] | None)
name (str)
pivot_on (str | None)
reader_function (collections.abc.Callable[[pathlib.Path], Any] | None)
reader_kwargs (dict[str, Any] | None)
units (str | None)
value_columns (list[str] | None)

field aggregate_function: Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Aggregation function')] = None¶: Aggregation function

field column_mapping: Annotated[dict[str, str] | None, FieldInfo(annotation=NoneType, required=True, description='Column name mappings')] = None¶: Column name mappings

field column_schema: Annotated[dict[str, str] | None, FieldInfo(annotation=NoneType, required=True, description='User-defined column names/types (used if input data has no column headers)')] = None¶: User-defined column names/types (used if input data has no column headers)

field description: Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Description of the data file')] = None¶: Description of the data file

field drop_columns: Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=True, description='Columns to drop')] = None¶: Columns to drop

field filter_by: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=True, description='Column filters as {column_name: value}')] = None¶: Column filters as {column_name: value}

field fpath: Annotated[Path, PathType(path_type=file), AfterValidator(func=validate_file_extension), FieldInfo(annotation=NoneType, required=True, description='File path (must exist)')] [Required]¶

File path (must exist)

Constraints:

path_type = file
func = <function validate_file_extension at 0x7f07757b76a0>

field index_columns: Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=True, description='Index column names')] = None¶: Index column names

field is_input: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether this is an input file')] = True¶: Whether this is an input file

field is_optional: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether this file is optional')] = False¶: Whether this file is optional

field is_timeseries: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether this file contains time series data. Time series files must use supported formats (CSV, HDF5, Parquet).')] = False¶: Whether this file contains time series data. Time series files must use supported formats (CSV, HDF5, Parquet).

field key_mapping: Annotated[dict[str, str] | None, FieldInfo(annotation=NoneType, required=True, description='Keys name mappings (applicable for JSON files).')] = None¶: Keys name mappings (applicable for JSON files).

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

field pivot_on: Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Column to pivot on')] = None¶: Column to pivot on

field reader_function: Annotated[Callable[[Path], Any] | None, FieldInfo(annotation=NoneType, required=True, description='Custom reader function (callable) that takes a Path and returns data')] = None¶: Custom reader function (callable) that takes a Path and returns data

field reader_kwargs: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=True, description='Key-Word arguments passed to the reader function.')] = None¶: Key-Word arguments passed to the reader function.

field units: Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Units for the data')] = None¶: Units for the data

field value_columns: Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=True, description='Value column names')] = None¶: Value column names

property file_type: FileFormat¶

Computed file type based on file extension.

Returns:: FileFormat instance determined from file extension
Return type:: FileFormat
Raises:: ValueError – If the file extension is not supported or if marked as time series but the file type doesn’t support time series data.

Parser Framework¶

class r2x_core.BaseParser(config, data_store, *, name=None, auto_add_composed_components=True, skip_validation=False)[source]¶

Bases: ABC

Abstract base class for building infrasys.System objects from model data.

The BaseParser provides a standardized framework for creating model-specific parsers. It orchestrates data loading, validation, transformation, and system construction, while delegating model-specific logic to subclasses through abstract methods.

Applications create parsers by: 1. Inheriting from BaseParser 2. Implementing required abstract methods 3. Optionally overriding hook methods for custom behavior 4. Defining a model-specific PluginConfig subclass

The typical workflow is: 1. Initialize parser with configuration and data store 2. Call build_system() which orchestrates the full build process 3. Receive populated infrasys.System object

Parameters:

config (PluginConfig) – Model-specific configuration instance containing model parameters, default values, and file mappings.
data_store (DataStore) – Initialized DataStore instance with file mappings loaded. The parser uses this to read and cache data files.
name (str, optional) – Name for the system being built. If None, a default name will be used.
auto_add_composed_components (bool, default=True) – Whether to automatically add composed components to the system. Passed directly to infrasys.System constructor.
skip_validation (bool, default=False) – Skip Pydantic validation when creating component instances. Useful for performance optimization or when handling incomplete/legacy data. Use with caution as it bypasses type and constraint checking.

config¶

The model configuration instance.

Type:: PluginConfig

data_store¶

Data store for file management and reading.

Type:: DataStore

system¶

The infrasys.System instance being built. None until build_system() is called.

Type:: System or None

name¶

Name of the system being built.

Type:: str

auto_add_composed_components¶

Whether to auto-add composed components.

Type:: bool

skip_validation¶

Whether to skip component validation.

Type:: bool

build_system()[source]¶

Main entry point: Build and return the complete infrasys.System.

Return type:: Any

get_data(key)[source]¶

Retrieve parsed data from the data store by key.

Parameters:: key (str)
Return type:: Any

read_data_file(name, \*\*kwargs)[source]¶

Read a data file through the data store with optional parameters.

Parameters:

name (str)
kwargs (Any)

Return type:

Any

create_component(component_class, \*\*field_values)[source]¶

Factory method to create and validate component instances.

Parameters:

component_class (type[Any])
field_values (Any)

Return type:

Any

add_component(component)[source]¶

Add a component to the system with logging.

Parameters:: component (Any)
Return type:: None

add_time_series(component, time_series, \*\*kwargs)[source]¶

Attach time series data to a component.

Parameters:

component (Any)
time_series (Any)
kwargs (Any)

Return type:

None

validate_inputs()[source]¶

Hook for pre-build validation (override in subclasses).

Return type:: None

post_process_system()[source]¶

Hook for post-build processing (override in subclasses).

Return type:: None

Abstract Methods (must implement in subclass)

----------------------------------------------

build_system_components()[source]¶

Create all system components (buses, generators, etc.).

Return type:: None

build_time_series()[source]¶

Attach time series data to components.

Return type:: None

Raises:

ParserError – If there are issues during parsing or system construction.
ValidationError – If configuration or data validation fails.
ComponentCreationError – If component instantiation fails.

Parameters:

config (PluginConfig)
data_store (DataStore)
name (str | None)
auto_add_composed_components (bool)
skip_validation (bool)

Examples

Create a simple parser:

>>> from r2x_core.parser import BaseParser
>>> from r2x_core.plugin_config import PluginConfig
>>> from r2x_core.store import DataStore
>>>
>>> class MyModelConfig(PluginConfig):
...     model_year: int
>>>
>>> class MyModelParser(BaseParser):
...     def __init__(self, config, data_store, **kwargs):
...         super().__init__(config, data_store, **kwargs)
...         self.model_year = config.model_year
...
...     def validate_inputs(self):
...         if self.model_year < 2020:
...             raise ValidationError("Invalid year")
...
...     def build_system_components(self):
...         bus_data = self.read_data_file("buses")
...         for row in bus_data.iter_rows(named=True):
...             bus = self.create_component(ACBus, name=row["name"])
...             self.add_component(bus)
...
...     def build_time_series(self):
...         load_data = self.read_data_file("load_profiles")
...         # Attach time series...
>>>
>>> config = MyModelConfig(model_year=2030)
>>> store = DataStore.from_json("mappings.json")
>>> parser = MyModelParser(config, store)
>>> system = parser.build_system()

With custom post-processing:

>>> class AdvancedParser(BaseParser):
...     def post_process_system(self):
...         '''Add metadata and validate connectivity.'''
...         logger.info(f"Built {len(self.system.get_components(Bus))} buses")
...         self._validate_connectivity()
...         self.system.metadata = {"year": self.config.model_year}

For plugin integration (calling methods individually):

>>> parser = MyModelParser(config, store)
>>> parser.validate_inputs()
>>> parser.build_system_components()  # Public method
>>> # Apply custom modifications...
>>> parser.build_time_series()  # Public method
>>> parser.post_process_system()
>>> system = parser.system

See also

r2x_core.plugin_config.PluginConfig: Base configuration class
DataStore: Data file management
DataReader: File reading operations
infrasys.system.System: Target system class

Notes

The parser follows the Template Method pattern where build_system() defines the overall algorithm flow, and subclasses fill in the specific steps through abstract methods.

The separation between parser and data store provides: - Independent testing of data loading vs. system building - Reuse of data stores across multiple parsers - Clear separation of I/O concerns from domain logic - Flexible caching strategies

Key design patterns: - Template Method: build_system() defines the workflow skeleton - build_system_components() and build_time_series() are the abstract templates - Hook methods: validate_inputs(), post_process_system() (optional overrides)

Initialize the parser with configuration and data store.

Parameters:

config (PluginConfig) – Model-specific configuration instance.
data_store (DataStore) – Initialized DataStore instance with file mappings.
name (str, optional) – Name for the system being built.
auto_add_composed_components (bool, default=True) – Whether to automatically add composed components.
skip_validation (bool, default=False) – Skip Pydantic validation when creating components.

PLUGIN_TYPE: ClassVar[str] = 'parser'¶

build_system()[source]¶

Build and return the complete infrasys.System.

This is the main entry point for the parser. It orchestrates the complete system building workflow by calling validation, component creation, time series attachment, and post-processing in sequence.

The workflow is: 1. Validate inputs (validate_inputs) 2. Create infrasys.System instance 3. Build system components (build_system_components) 4. Build time series (build_time_series) 5. Post-process system (post_process_system) 6. Return completed system

Returns:

Fully constructed infrasys.System instance with all components and time series attached.

Return type:

System

Raises:

ParserError – If system construction fails.
ValidationError – If validation fails during any step.
ComponentCreationError – If component creation fails.

Examples

Basic usage:

>>> parser = MyModelParser(config, data_store)
>>> system = parser.build_system()
>>> print(f"Built system with {len(system.get_components(Bus))} buses")

With error handling:

>>> try:
...     system = parser.build_system()
... except ValidationError as e:
...     logger.error(f"Validation failed: {e}")
... except ParserError as e:
...     logger.error(f"Parser error: {e}")

See also

validate_inputs: Pre-build validation hook
build_system_components: Component creation
build_time_series: Time series attachment
post_process_system: Post-build processing hook

Notes

This method creates the System instance during execution, not in __init__. This allows multiple systems to be built from the same parser configuration if needed, and keeps the parser lightweight until actually used.

For plugin systems that need finer control, call the individual public methods (build_system_components, build_time_series) directly instead.

get_data(key)[source]¶

Retrieve parsed data from the data store by key.

Provides convenient access to data that has been loaded into the data store. This is a thin wrapper around the data store’s internal data access.

Parameters:: key (str) – The data file identifier/name as registered in the data store.
Returns:: The loaded data (typically polars.LazyFrame, polars.DataFrame, dict, or other format depending on the file type).
Return type:: Any
Raises:: KeyError – If the specified key is not found in the data store.

Examples

>>> bus_data = parser.get_data("buses")
>>> gen_data = parser.get_data("generators")

With error handling:

>>> try:
...     data = parser.get_data("optional_file")
... except KeyError:
...     logger.warning("Optional file not found, using defaults")
...     data = default_data

See also

read_data_file: Read data file through the data store
DataStore.get_data_file_by_name: Underlying data store method

Notes

This method assumes the data has already been loaded into the data store. For on-demand reading, use read_data_file() instead.

read_data_file(name, **kwargs)[source]¶

Read a data file through the data store.

Loads a data file using the data store’s reader, applying any configured transformations (filters, column mapping, etc.) and optionally using cache.

Parameters:

name (str) – The data file identifier as registered in the data store.
**kwargs – Additional keyword arguments passed to the data store’s read_data_file method. Common arguments include: - use_cache : bool, whether to use cached data

Returns:

The loaded and transformed data (format depends on file type).

Return type:

Any

Raises:

KeyError – If the data file name is not found in the data store.
ParserError – If file reading fails.

Examples

Basic usage:

>>> bus_data = parser.read_data_file("buses")

With caching disabled:

>>> fresh_data = parser.read_data_file("generators", use_cache=False)

See also

get_data: Access already-loaded data
DataStore.read_data_file: Underlying data store method
DataFile: File configuration including transformations

Notes

This method applies transformations configured in the DataFile definition. If the file has filter, rename, or cast operations defined, they are applied automatically during reading.

For performance-critical code paths, consider using use_cache=True (default) to avoid repeated file I/O.

create_component(component_class, **field_values)[source]¶

Create and validate a component instance.

Factory method for creating infrasys Component instances with optional validation skipping. Handles field filtering to only pass valid fields for the component class, and provides consistent error handling.

Parameters:

component_class (type) – The Component class to instantiate (e.g., ACBus, Generator, etc.).
**field_values – Field names and values to pass to the component constructor. Invalid fields (not in component’s model_fields) are filtered out. None values are also filtered out.

Returns:

Validated (or constructed) instance of the specified component class.

Return type:

Component

Raises:

ComponentCreationError – If component creation fails due to invalid field values or other errors.

Examples

Basic usage:

>>> bus = parser.create_component(
...     ACBus,
...     name="Bus1",
...     voltage=230.0,
...     bus_type=ACBusTypes.PV
... )
>>> parser.add_component(bus)

With extra fields (filtered out automatically):

>>> gen = parser.create_component(
...     Generator,
...     name="Gen1",
...     capacity=100.0,
...     extra_field="ignored",  # Not in Generator.model_fields
...     bus=None,  # None values are filtered out
... )

Skip validation for performance:

>>> parser.skip_validation = True
>>> gen = parser.create_component(ThermalGen, **row_data)

See also

add_component: Add component to system
infrasys.component.Component: Base component class

Notes

This method implements the create_model_instance pattern from the old r2x codebase, providing the skip_validation functionality for handling incomplete or legacy data.

When skip_validation=True: - Uses model_construct() instead of model_validate() - Bypasses Pydantic validation (faster but less safe) - Falls back to validation if construction fails

Field filtering removes: - Fields not in component_class.model_fields - Fields with None values

Subclasses can override this method to add model-specific defaults or transformations before component creation.

add_component(component)[source]¶

Add a component to the system with logging.

Convenience method that adds a component to the system and logs the action. Provides consistent logging across all parsers.

Parameters:: component (Component) – The infrasys Component instance to add to the system.
Raises:: ParserError – If system has not been created yet (build_system not called).
Return type:: None

Examples

>>> bus = parser.create_component(ACBus, name="Bus1")
>>> parser.add_component(bus)

In a loop:

>>> for row in bus_data.iter_rows(named=True):
...     bus = parser.create_component(ACBus, **row)
...     parser.add_component(bus)

See also

create_component: Create component instances
infrasys.system.System.add_component: Underlying system method

Notes

This method requires that self.system is not None. Call this within build_system_components(), after build_system() completes, or in plugin workflows after manually creating a System instance.

The logging uses DEBUG level to avoid cluttering output when adding many components, but provides traceability for debugging.

add_time_series(component, time_series, **kwargs)[source]¶

Attach time series data to a component.

Convenience method for adding time series to components with consistent logging and error handling.

Parameters:

component (Component) – The component to attach the time series to.
time_series (TimeSeriesData) – The time series data instance (e.g., SingleTimeSeries).
**kwargs – Additional keyword arguments passed to system.add_time_series.

Raises:

ParserError – If system has not been created yet.

Return type:

None

Examples

>>> from infrasys.time_series_models import SingleTimeSeries
>>>
>>> bus = parser.system.get_component(ACBus, "Bus1")
>>> ts = SingleTimeSeries(
...     data=load_profile.to_numpy(),
...     variable_name="max_active_power"
... )
>>> parser.add_time_series(bus, ts)

See also

build_time_series: Method that typically calls this
infrasys.system.System.add_time_series: Underlying system method

Notes

This is typically called within build_time_series(), but can be used in any workflow where self.system has been initialized.

validate_inputs()[source]¶

Validate configuration and data before building system.

Hook method that subclasses can override to perform custom validation before system construction begins. Called at the start of build_system().

Default implementation does nothing. Override in subclasses to add validation logic.

Raises:: ValidationError – If validation fails. Subclasses should raise this exception with a descriptive message.
Return type:: None

Examples

Basic validation:

>>> class MyParser(BaseParser):
...     def validate_inputs(self):
...         if self.config.model_year < 2020:
...             raise ValidationError("Year must be >= 2020")

Checking data availability:

>>> def validate_inputs(self):
...     required = ["buses", "generators", "branches"]
...     for name in required:
...         if name not in self.data_store._data_files:
...             raise ValidationError(f"Required file '{name}' missing")

Validating against data:

>>> def validate_inputs(self):
...     years_data = self.get_data("available_years")
...     available = years_data["year"].to_list()
...     if self.config.model_year not in available:
...         raise ValidationError(
...             f"Year {self.config.model_year} not in {available}"
...         )

See also

build_system: Calls this method first
ValidationError: Exception to raise on failure

Notes

This is a hook method in the Template Method pattern. The base class calls it at the appropriate time, but subclasses provide the implementation.

Best practices: - Validate configuration parameters - Check required data files are present - Verify cross-field constraints - Fail fast with clear error messages

abstractmethod build_system_components()[source]¶

Create all system components (abstract method for subclass implementation).

Subclasses must implement this method to create and add all components needed for their specific model. This typically includes buses, generators, loads, branches, and other network elements.

The implementation should: 1. Read component data using read_data_file() 2. Iterate over the data 3. Create components using create_component() 4. Add components using add_component()

Raises:

ParserError – If component creation fails.
ComponentCreationError – If specific component instantiation fails.

Return type:

None

Examples

Normal usage (via build_system):

>>> parser = MyModelParser(config, data_store)
>>> system = parser.build_system()  # Calls build_system_components internally

Direct usage (for plugin systems):

>>> parser = MyModelParser(config, data_store)
>>> parser.validate_inputs()
>>> parser.build_system_components()  # Call directly
>>> # Apply custom modifications...
>>> parser.build_time_series()

Typical implementation:

>>> def build_system_components(self):
...     # Create buses
...     bus_data = self.read_data_file("buses")
...     for row in bus_data.iter_rows(named=True):
...         bus = self.create_component(
...             ACBus,
...             name=row["bus_name"],
...             voltage=row["voltage_kv"]
...         )
...         self.add_component(bus)
...
...     # Create generators
...     gen_data = self.read_data_file("generators")
...     for row in gen_data.iter_rows(named=True):
...         gen = self.create_component(
...             ThermalGen,
...             name=row["gen_name"],
...             bus=self.system.get_component(ACBus, row["bus_name"]),
...             active_power=row["capacity_mw"]
...         )
...         self.add_component(gen)

With error handling:

>>> def build_system_components(self):
...     try:
...         bus_data = self.read_data_file("buses")
...     except KeyError:
...         raise ParserError("Required file 'buses' not found")
...
...     for idx, row in enumerate(bus_data.iter_rows(named=True)):
...         try:
...             bus = self.create_component(ACBus, **row)
...             self.add_component(bus)
...         except Exception as e:
...             raise ComponentCreationError(
...                 f"Failed to create bus at row {idx}: {e}"
...             ) from e

See also

build_system: Main workflow that calls this method
create_component: Factory for creating components
add_component: Add component to system
read_data_file: Read data files
build_time_series: Companion method for time series

Notes

This is an abstract method that must be implemented by subclasses. It is called by build_system() as part of the template method workflow.

Common patterns: - Create topology first (buses, areas, zones) - Create devices (generators, loads, storage) - Create connections (branches, interfaces) - Establish relationships (generator.bus = bus_instance)

abstractmethod build_time_series()[source]¶

Attach time series data to components (abstract method for subclass implementation).

Subclasses must implement this method to read and attach time series data to the appropriate components. This typically includes load profiles, generation profiles, and other time-varying data.

The implementation should: 1. Read time series data using read_data_file() 2. Process/transform the data as needed 3. Create time series objects (e.g., SingleTimeSeries) 4. Attach to components using add_time_series()

Raises:: ParserError – If time series processing fails.
Return type:: None

Examples

Normal usage (via build_system):

>>> parser = MyModelParser(config, data_store)
>>> system = parser.build_system()  # Calls build_time_series internally

Direct usage (for plugin systems):

>>> parser = MyModelParser(config, data_store)
>>> parser.build_system_components()
>>> # Apply modifications...
>>> parser.build_time_series()  # Call directly

Typical implementation:

>>> def build_time_series(self):
...     from infrasys.time_series_models import SingleTimeSeries
...
...     # Load profiles for buses
...     load_data = self.read_data_file("load_profiles")
...
...     for bus_name in load_data.columns:
...         bus = self.system.get_component(ACBus, bus_name)
...         ts = SingleTimeSeries(
...             data=load_data[bus_name].to_numpy(),
...             variable_name="max_active_power"
...         )
...         self.add_time_series(bus, ts)
...
...     # Capacity factors for renewables
...     cf_data = self.read_data_file("capacity_factors")
...     for gen_name in cf_data.columns:
...         gen = self.system.get_component(RenewableGen, gen_name)
...         ts = SingleTimeSeries(
...             data=cf_data[gen_name].to_numpy(),
...             variable_name="max_active_power"
...         )
...         self.add_time_series(gen, ts)

See also

build_system: Main workflow that calls this method
add_time_series: Attach time series to component
read_data_file: Read time series data files
build_system_components: Companion method for components

Notes

This is an abstract method that must be implemented by subclasses. It is called by build_system() after build_system_components(), ensuring all components exist before attaching time series.

Common time series types: - Load profiles (demand over time) - Renewable capacity factors (generation availability) - Price curves (cost over time) - Reserve requirements (operating reserve levels)

post_process_system()[source]¶

Perform post-processing after system construction.

Hook method that subclasses can override to perform custom processing after all components and time series have been added. Called at the end of build_system().

Default implementation does nothing. Override in subclasses to add post-processing logic.

Examples

Add summary logging:

>>> def post_process_system(self):
...     logger.info(f"System '{self.system.name}' built successfully")
...     logger.info(f"  Buses: {len(self.system.get_components(ACBus))}")
...     logger.info(f"  Generators: {len(self.system.get_components(Generator))}")

Add metadata:

>>> def post_process_system(self):
...     self.system.metadata = {
...         "model_year": self.config.model_year,
...         "scenario": self.config.scenario,
...         "created": datetime.now().isoformat(),
...         "parser_version": __version__
...     }

Validate system integrity:

>>> def post_process_system(self):
...     # Check all generators have valid buses
...     buses = set(self.system.get_components(ACBus))
...     for gen in self.system.get_components(Generator):
...         if gen.bus not in buses:
...             raise ValidationError(
...                 f"Generator {gen.name} has invalid bus"
...             )

See also

build_system: Calls this method last
validate_inputs: Pre-build validation hook

Notes

This is a hook method in the Template Method pattern. The base class calls it at the appropriate time, but subclasses provide the implementation.

Common uses: - Logging summary statistics - Adding metadata - Validating system integrity - Computing derived quantities - Setting up cross-component relationships

Return type:: None

pydantic model r2x_core.PluginConfig[source]¶

Base configuration class for plugin inputs and model parameters.

Applications should inherit from this class to define model-specific configuration parameters for parsers, exporters, and system modifiers. Subclasses define their own fields for model-specific parameters.

Examples

Create a model-specific configuration:

>>> class ReEDSConfig(PluginConfig):
...     '''Configuration for ReEDS parser.'''
...     model_year: int
...     weather_year: int
...     scenario: str = "base"
...
>>> config = ReEDSConfig(
...     model_year=2030,
...     weather_year=2012,
...     scenario="high_re"
... )

With validation:

>>> from pydantic import field_validator
>>>
>>> class ValidatedConfig(PluginConfig):
...     model_year: int
...
...     @field_validator("model_year")
...     @classmethod
...     def validate_year(cls, v):
...         if v < 2020 or v > 2050:
...             raise ValueError("Year must be between 2020 and 2050")
...         return v

See also

r2x_core.parser.BaseParser: Uses this configuration class
r2x_core.exporter.BaseExporter: Uses this configuration class
pydantic.BaseModel: Parent class providing validation

Notes

The PluginConfig uses Pydantic for: - Automatic type checking and validation - JSON serialization/deserialization - Field validation and transformation - Default value management

Subclasses can add: - Model-specific years (solve_year, weather_year, horizon_year, etc.) - Scenario identifiers - Feature flags - File path overrides - Custom validation logic

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.

classmethod get_cli_schema()[source]¶

Get JSON schema for CLI argument generation.

This method generates a CLI-friendly schema from the configuration class, adding metadata useful for building command-line interfaces. It’s designed to help tools like r2x-cli dynamically generate argument parsers from configuration classes.

Returns:: A JSON schema dictionary enhanced with CLI metadata. Each property includes: - cli_flag: The command-line flag (e.g., “–model-year”) - required: Whether the argument is required - All standard Pydantic schema fields (type, description, default, etc.)
Return type:: dict[str, Any]

Examples

Generate CLI schema for a configuration class:

>>> from r2x_core.plugin_config import PluginConfig
>>>
>>> class MyConfig(PluginConfig):
...     '''My model configuration.'''
...     model_year: int
...     scenario: str = "base"
...
>>> schema = MyConfig.get_cli_schema()
>>> print(schema["properties"]["model_year"]["cli_flag"])
--model-year
>>> print(schema["properties"]["model_year"]["required"])
True
>>> print(schema["properties"]["scenario"]["cli_flag"])
--scenario
>>> print(schema["properties"]["scenario"]["required"])
False

Use in CLI generation:

>>> import argparse
>>> parser = argparse.ArgumentParser()
>>> schema = MyConfig.get_cli_schema()
>>> for field_name, field_info in schema["properties"].items():
...     flag = field_info["cli_flag"]
...     required = field_info["required"]
...     help_text = field_info.get("description", "")
...     parser.add_argument(flag, required=required, help=help_text)

See also

load_defaults: Load default constants from JSON file
r2x_core.parser.BaseParser.get_file_mapping_path: Get file mapping path
pydantic.BaseModel.model_json_schema: Underlying schema generation

Notes

The CLI flag naming convention converts underscores to hyphens: - model_year -> –model-year - weather_year -> –weather-year - solve_year -> –solve-year

This follows common CLI conventions (e.g., argparse, click).

The schema includes all Pydantic field information, so CLI tools can: - Determine field types for proper parsing - Extract descriptions for help text - Identify default values - Validate constraints

classmethod get_file_mapping_path()[source]¶

Get the path to this plugin’s file mapping JSON.

This method uses inspect.getfile() to locate the plugin module file, then constructs the path to the file mapping JSON in the config directory. By convention, plugins should store their file_mapping.json in a config/ subdirectory next to the config module.

The filename can be customized by overriding the FILE_MAPPING_NAME class variable.

Returns:: Absolute path to the file_mapping.json file. Note that this path may not exist if the plugin hasn’t created the file yet.
Return type:: Path

Examples

Get file mapping path for a config:

>>> from r2x_reeds.config import ReEDSConfig
>>> mapping_path = ReEDSConfig.get_file_mapping_path()
>>> print(mapping_path)
/path/to/r2x_reeds/config/file_mapping.json

Override the filename in a custom config:

>>> class CustomConfig(PluginConfig):
...     FILE_MAPPING_NAME = "custom_mapping.json"
...
>>> path = CustomConfig.get_file_mapping_path()
>>> print(path.name)
custom_mapping.json

Use with DataStore:

>>> from r2x_core import DataStore
>>> mapping_path = MyModelConfig.get_file_mapping_path()
>>> store = DataStore.from_json(mapping_path, folder="/data/mymodel")

See also

load_defaults: Similar pattern for loading constants
DataStore.from_plugin_config: Direct DataStore creation from config

Notes

This method uses inspect.getfile() to locate the module file, then navigates to the config directory. This works for both installed packages and editable installs.

classmethod load_defaults(defaults_file=None)[source]¶

Load default constants from JSON file.

Provides a standardized way to load model-specific constants, mappings, and default values from JSON files. If no file path is provided, automatically looks for the file specified by DEFAULTS_FILE_NAME in the config directory.

Parameters:: defaults_file (Path, str, or None, optional) – Path to defaults JSON file. If None, looks for the file specified by DEFAULTS_FILE_NAME (default: ‘defaults.json’) in the CONFIG_DIR subdirectory relative to the config module.
Returns:: Dictionary of default constants to use in your parser/exporter logic. Returns empty dict if file doesn’t exist.
Return type:: dict[str, Any]

Examples

Load defaults automatically:

>>> from r2x_reeds.config import ReEDSConfig
>>> defaults = ReEDSConfig.load_defaults()
>>> config = ReEDSConfig(
...     solve_years=2030,
...     weather_years=2012,
... )
>>> # Use defaults dict in your parser/exporter logic
>>> excluded_techs = defaults.get("excluded_techs", [])

Load from custom path:

>>> defaults = ReEDSConfig.load_defaults("/path/to/custom_defaults.json")

See also

PluginConfig: Base configuration class
get_file_mapping_path: Related file discovery method

CONFIG_DIR: ClassVar[str] = 'config'¶

DEFAULTS_FILE_NAME: ClassVar[str] = 'defaults.json'¶

FILE_MAPPING_NAME: ClassVar[str] = 'file_mapping.json'¶

Exporter Framework¶

class r2x_core.BaseExporter(config, system, data_store, **kwargs)[source]¶

Bases: ABC

Abstract base class for exporting infrasys.System objects to model formats.

This class provides the foundational structure for model-specific exporters. Subclasses must implement two abstract methods: - export(): Define the overall export workflow - export_time_series(): Export time series data from system

The BaseExporter provides access to: - System with components and time series (self.system) - DataStore with file paths and configuration (self.data_store) - Export configuration (self.config)

Parameters:

config (BaseModel) – Configuration object containing export parameters. Applications should define their own config class inheriting from pydantic.BaseModel.
system (System) – R2X System object containing components and time series to export.
data_store (DataStore) – DataStore containing output file paths and configuration.
**kwargs (dict, optional) – Additional keyword arguments for subclass customization.

config¶

Export configuration parameters.

Type:: BaseModel

system¶

System object with components and time series.

Type:: System

data_store¶

DataStore with output file configuration.

Type:: DataStore

Examples

Create a simple exporter:

>>> from pydantic import BaseModel
>>> from r2x_core import BaseExporter, DataStore, System, DataFile
>>>
>>> class MyConfig(BaseModel):
...     year: int
>>>
>>> class MyExporter(BaseExporter):
...     def export(self) -> None:
...         logger.info(f"Exporting for year {self.config.year}")
...         # Export all components to CSV
...         output_file = self.data_store.data_files["components"]
...         self.system.export_components_to_csv(output_file.file_path)
...         # Export time series
...         self.export_time_series()
...
...     def export_time_series(self) -> None:
...         # Export time series data
...         ts_file = self.data_store.data_files["timeseries"]
...         # ... implementation
>>>
>>> config = MyConfig(year=2030)
>>> system = System(name="MySystem")
>>> data_store = DataStore(
...     data_files={"components": DataFile(name="components", file_path="output.csv")},
...     folder="/output"
... )
>>> exporter = MyExporter(config, system, data_store)
>>> exporter.export()

Export with filtering:

>>> class FilteredExporter(BaseExporter):
...     def export(self) -> None:
...         # Export only specific component types
...         gen_file = self.data_store.data_files["generators"]
...         self.system.export_components_to_csv(
...             file_path=gen_file.file_path,
...             filter_func=lambda c: c.__class__.__name__ == "Generator"
...         )
...         self.export_time_series()
...
...     def export_time_series(self) -> None:
...         # Export time series for generators only
...         pass

Export with field selection and renaming:

>>> class MappedExporter(BaseExporter):
...     def export(self) -> None:
...         gen_file = self.data_store.data_files["generators"]
...         self.system.export_components_to_csv(
...             file_path=gen_file.file_path,
...             filter_func=lambda c: isinstance(c, Generator),
...             fields=["name", "max_active_power"],
...             key_mapping={"max_active_power": "capacity_mw"}
...         )
...         self.export_time_series()
...
...     def export_time_series(self) -> None:
...         # Export time series data
...         pass

See also

r2x_core.system.System.export_components_to_csv: Export components to CSV
r2x_core.system.System.components_to_records: Get components as dict records
r2x_core.store.DataStore: File path management
infrasys.system.System.get_time_series: Get time series from components

Notes

The BaseExporter provides a minimal interface with two abstract methods: - export(): Orchestrate the complete export workflow - export_time_series(): Handle time series data export

This gives applications maximum flexibility to: - Define their own export workflow in export() - Use any combination of System export methods - Implement custom time series export logic - Handle transformations and formatting as needed

Common export patterns: 1. Component export: Use system.export_components_to_csv() 2. Custom transformations: Use system.components_to_records() then transform 3. Time series export: Implement in export_time_series() using system.get_time_series() 4. Multi-file export: Iterate over data_store.data_files

Initialize the exporter.

Parameters:

config (BaseModel) – Export configuration parameters.
system (System) – System object to export.
data_store (DataStore) – DataStore with output file paths.
**kwargs (dict) – Additional arguments for subclass customization.

abstractmethod export()[source]¶

Export the system to the target model format.

This method must be implemented by subclasses to define the complete export workflow for their specific model format.

The implementation should:

Export component data using system.export_components_to_csv() or system.components_to_records()
Export time series data using export_time_series()
Apply any model-specific transformations
Write files to paths configured in data_store

Raises:: ExporterError – If export fails for any reason.
Return type:: None

Examples

Simple export implementation:

>>> def export(self) -> None:
...     logger.info("Starting export")
...     # Export components
...     self._export_components()
...     # Export time series
...     self.export_time_series()
...     logger.info("Export complete")

Export with error handling:

>>> def export(self) -> None:
...     try:
...         self._export_components()
...         self.export_time_series()
...     except Exception as e:
...         raise ExporterError(f"Export failed: {e}") from e

See also

export_time_series: Export time series data from system
r2x_core.system.System.export_components_to_csv: Export components
r2x_core.system.System.components_to_records: Get component records
r2x_core.exceptions.ExporterError: Export error exception

abstractmethod export_time_series()[source]¶

Export time series data from the system.

This method must be implemented by subclasses to export time series data from components to files. The implementation should: 1. Identify which components have time series data 2. Retrieve time series using system.get_time_series() 3. Convert to appropriate format (DataFrame, arrays, etc.) 4. Write to files based on data_store configuration

Subclasses can filter time series files using: >>> ts_files = [ … df for df in self.data_store.data_files.values() … if df.is_timeseries … ]

Raises:: ExporterError – If time series export fails.
Return type:: None

Examples

Export time series to CSV:

>>> def export_time_series(self) -> None:
...     import polars as pl
...     for datafile in self.data_store.data_files.values():
...         if datafile.is_timeseries:
...             # Collect time series data
...             ts_data = self._collect_time_series(datafile.name)
...             # Write to file
...             ts_data.write_csv(datafile.file_path)

Export to HDF5:

>>> def export_time_series(self) -> None:
...     import h5py
...     ts_file = self.data_store.data_files["timeseries"]
...     with h5py.File(ts_file.file_path, "w") as f:
...         for component in self.system.get_components():
...             ts_data = self.system.get_time_series(component)
...             if ts_data is not None:
...                 f.create_dataset(component.name, data=ts_data)

Export with file type matching:

>>> def export_time_series(self) -> None:
...     from r2x_core.file_types import TableFormat, H5Format, ParquetFormat
...     for datafile in self.data_store.data_files.values():
...         if not datafile.is_timeseries:
...             continue
...         ts_data = self._collect_time_series(datafile.name)
...         match datafile.file_type:
...             case TableFormat():
...                 ts_data.write_csv(datafile.file_path)
...             case H5Format():
...                 self._write_h5(datafile, ts_data)
...             case ParquetFormat():
...                 ts_data.write_parquet(datafile.file_path)

See also

infrasys.system.System.get_time_series: Retrieve time series from components
r2x_core.datafile.DataFile.is_timeseries: Check if file is for time series
r2x_core.file_types: File type classes for format handling

Plugin System¶

class r2x_core.PluginManager[source]¶

Singleton registry for parsers, exporters, modifiers, and filters.

PluginManager maintains class-level registries for all plugin types and provides discovery via entry points. It uses the singleton pattern to ensure a single source of truth for all registered plugins.

Class Attributes¶

_instancePluginManager | None: Singleton instance
_initializedbool: Whether entry points have been loaded
_registrydict[str, PluginComponent]: Model plugin registry (name -> PluginComponent)
_modifier_registrydict[str, SystemModifier]: System modifier registry (name -> function)
_filter_registrydict[str, FilterFunction]: Filter function registry (name -> function)

Properties¶

registered_parsersdict[str, type]: All registered parser classes
registered_exportersdict[str, type]: All registered exporter classes
registered_modifiersdict[str, SystemModifier]: All registered system modifiers
registered_filtersdict[str, FilterFunction]: All registered filter functions

register_model_plugin(name, config, parser=None, exporter=None)[source]¶

Parameters:

name (str)
config (type[PluginConfig])
parser (type | None)
exporter (type | None)

Return type:

None

register_system_modifier(name)[source]¶

Decorator to register a system modifier function

Parameters:: name (str | SystemModifier | None)
Return type:: Callable[[SystemModifier], SystemModifier] | SystemModifier

register_filter(name)[source]¶

Decorator to register a filter function

Parameters:: name (str | FilterFunction | None)
Return type:: Callable[[FilterFunction], FilterFunction] | FilterFunction

load_parser(name)[source]¶

Load a parser class by name

Parameters:: name (str)
Return type:: type[BaseParser] | None

load_exporter(name)[source]¶

Load an exporter class by name

Parameters:: name (str)
Return type:: type | None

load_config_class(name)[source]¶

Load config class for a plugin

Parameters:: name (str)
Return type:: type[PluginConfig] | None

Examples

>>> manager = PluginManager()
>>> PluginManager.register_model_plugin("switch", SwitchConfig, SwitchParser, SwitchExporter)
>>> parser_class = manager.load_parser("switch")
>>> print(list(manager.registered_parsers.keys()))
['switch']

Use as decorator:

>>> @PluginManager.register_system_modifier("add_storage")
... def add_storage(system, **kwargs):
...     return system

See also

PluginComponent: Data structure for model plugins
SystemModifier: Protocol for modifier functions
FilterFunction: Protocol for filter functions

Ensure singleton instance.

classmethod register_model_plugin(name, config, parser=None, exporter=None)[source]¶

Registers a model plugin with its configuration and optionally parser and/or exporter classes. At least one of parser or exporter should be provided, though both None is allowed (with a warning).

Parameters:

name (str) – Plugin name (e.g., “switch”, “plexos”)
config (type[PluginConfig]) – Pydantic configuration class
parser (type | None, optional) – Parser class (BaseParser subclass)
exporter (type | None, optional) – Exporter class (BaseExporter subclass)

Return type:

None

Warning

Logs a warning if both parser and exporter are None.

Examples

>>> PluginManager.register_model_plugin(
...     name="switch",
...     config=SwitchConfig,
...     parser=SwitchParser,
...     exporter=SwitchExporter,
... )

Parser-only plugin:

>>> PluginManager.register_model_plugin(
...     name="reeds",
...     config=ReEDSConfig,
...     parser=ReEDSParser,
... )

classmethod register_system_modifier(name=None)[source]¶

System modifiers transform a System object and return the modified system. They can accept additional context via **kwargs.

Can be used with or without a name argument: - @register_system_modifier - uses function name - @register_system_modifier(“custom_name”) - uses explicit name

Parameters:: name (str | SystemModifier | None) – Modifier name, or the function itself if used without parentheses
Returns:: Decorator function or decorated function
Return type:: Callable | SystemModifier

Examples

>>> @PluginManager.register_system_modifier
... def add_storage(system: System, capacity_mw: float = 100.0, **kwargs) -> System:
...     # Add storage components
...     return system

>>> @PluginManager.register_system_modifier("custom_name")
... def add_storage(system: System, **kwargs) -> System:
...     return system

classmethod register_filter(name=None)[source]¶

Filter functions process data (typically polars DataFrames) and return processed data.

Can be used with or without a name argument: - @register_filter - uses function name - @register_filter(“custom_name”) - uses explicit name

Parameters:: name (str | FilterFunction | None) – Filter name, or the function itself if used without parentheses
Returns:: Decorator function or decorated function
Return type:: Callable | FilterFunction

Examples

>>> @PluginManager.register_filter
... def rename_columns(data: pl.LazyFrame, mapping: dict[str, str]) -> pl.LazyFrame:
...     return data.rename(mapping)

>>> @PluginManager.register_filter("custom_name")
... def process_data(data: pl.LazyFrame) -> pl.LazyFrame:
...     return data

property registered_parsers: dict[str, type]¶

Get all registered parser classes.

Returns:: Mapping of plugin name to parser class
Return type:: dict[str, type]

property registered_exporters: dict[str, type]¶

Get all registered exporter classes.

Returns:: Mapping of plugin name to exporter class
Return type:: dict[str, type]

property registered_modifiers: dict[str, SystemModifier]¶

Get all registered system modifiers.

Returns:: Mapping of modifier name to function
Return type:: dict[str, SystemModifier]

property registered_filters: dict[str, FilterFunction]¶

Get all registered filter functions.

Returns:: Mapping of filter name to function
Return type:: dict[str, FilterFunction]

load_parser(name)[source]¶

Load a parser class by name.

Parameters:: name (str) – Plugin name
Returns:: Parser class or None if not found
Return type:: type[BaseParser] | None

Examples

>>> manager = PluginManager()
>>> parser_class = manager.load_parser("switch")
>>> if parser_class:
...     parser = parser_class(config=config, data_store=store)

load_exporter(name)[source]¶

Load an exporter class by name.

Parameters:: name (str) – Plugin name
Returns:: Exporter class or None if not found
Return type:: type | None

Examples

>>> manager = PluginManager()
>>> exporter_class = manager.load_exporter("plexos")
>>> if exporter_class:
...     exporter = exporter_class(config=config, system=system, data_store=store)

load_config_class(name)[source]¶

Load configuration class for a plugin.

Parameters:: name (str) – Plugin name
Returns:: Configuration class or None if not found
Return type:: type[PluginConfig] | None

Examples

>>> manager = PluginManager()
>>> config_class = manager.load_config_class("switch")
>>> if config_class:
...     config = config_class(folder="./data", year=2030)

get_file_mapping_path(plugin_name)[source]¶

Get the file mapping path for a registered plugin.

This is a convenience method that loads the plugin’s config class and delegates to its get_file_mapping_path() classmethod. This allows getting the file mapping path without directly importing the config class.

Parameters:: plugin_name (str) – Name of the registered plugin
Returns:: Absolute path to the plugin’s file_mapping.json, or None if the plugin is not registered.
Return type:: Path | None

Examples

Get file mapping path for a registered plugin:

>>> from r2x_core import PluginManager
>>> manager = PluginManager()
>>> mapping_path = manager.get_file_mapping_path("reeds")
>>> if mapping_path:
...     print(f"ReEDS mapping: {mapping_path}")
...     if mapping_path.exists():
...         import json
...         with open(mapping_path) as f:
...             mappings = json.load(f)

Use in CLI tools:

>>> import sys
>>> plugin = sys.argv[1]  # e.g., "switch"
>>> manager = PluginManager()
>>> path = manager.get_file_mapping_path(plugin)
>>> if path and path.exists():
...     # Load and process mappings
...     pass
... else:
...     print(f"No file mapping found for {plugin}")

See also

PluginConfig.get_file_mapping_path: Config classmethod this delegates to
load_config_class: Load the config class directly

Notes

The file may not exist even if a path is returned - the method only constructs the expected path based on the config module location.

Return type:: PluginManager

class r2x_core.PluginComponent(config, parser=None, exporter=None)[source]¶

Model plugin registration data.

Holds the parser, exporter, and config classes for a model plugin. At least one of parser or exporter must be provided.

Parameters:

config (type[PluginConfig]) – Pydantic config class for the model
parser (type | None) – Parser class (BaseParser subclass)
exporter (type | None) – Exporter class (BaseExporter subclass)

config¶

Configuration class

Type:: type[PluginConfig]

parser¶

Parser class or None

Type:: type | None

exporter¶

Exporter class or None

Type:: type | None

config: type[PluginConfig]¶

parser: type | None = None¶

exporter: type | None = None¶

class r2x_core.SystemModifier(*args, **kwargs)[source]¶

Bases: Protocol

Protocol for system modifier functions.

System modifiers transform a System object, optionally using additional context like configuration or parser data. They must return a System object.

Parameters:

system (System) – The system to modify
**kwargs (dict) – Optional context (config, parser, etc.)

Returns:

Modified system object

Return type:

System

Examples

>>> def add_storage(system: System, capacity_mw: float = 100.0, **kwargs) -> System:
...     # Add storage components
...     return system

class r2x_core.FilterFunction(*args, **kwargs)[source]¶

Bases: Protocol

Protocol for filter functions.

Filter functions process data (typically polars DataFrames) and return processed data. They can accept additional parameters for configuration.

Parameters:

data (Any) – Data to filter/process (typically pl.LazyFrame)
**kwargs (Any) – Filter-specific parameters

Returns:

Processed data

Return type:

Any

Examples

>>> def rename_columns(data: pl.LazyFrame, mapping: dict[str, str]) -> pl.LazyFrame:
...     return data.rename(mapping)

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¶

Exceptions¶

class r2x_core.ParserError[source]¶

Bases: R2XCoreError

Exception raised for parser-related errors.

This exception is raised when there are issues during the parsing and system building process, such as invalid data, missing required files, or configuration errors.

Parameters:: message (str) – Description of the error.

Examples

>>> raise ParserError("Required file 'buses.csv' not found in data store")

class r2x_core.ValidationError[source]¶

Bases: R2XCoreError

Exception raised for validation errors.

This exception is raised when data or configuration validation fails, such as invalid years, missing required fields, or constraint violations.

Parameters:: message (str) – Description of the validation error.

Examples

>>> raise ValidationError("Model year 2019 not found in available years: [2020, 2025, 2030]")

class r2x_core.ComponentCreationError[source]¶

Bases: R2XCoreError

Exception raised when component creation fails.

This exception is raised when there are issues creating component instances, such as invalid field values or type mismatches.

Parameters:: message (str) – Description of the component creation error.

Examples

>>> raise ComponentCreationError("Failed to create Bus: missing required field 'voltage'")

class r2x_core.ExporterError[source]¶

Bases: R2XCoreError

Exception raised for exporter-related errors.

This exception is raised when there are issues during the export process, such as missing required components, invalid output formats, or file writing errors.

Parameters:: message (str) – Description of the error.

Examples

>>> raise ExporterError("No Generator components found in system")
>>> raise ExporterError("Output directory does not exist: /path/to/output")

Utilities¶

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

Filter kwargs to only include valid parameters for the given function.

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.

Data Processors¶

See Data Processors for complete processor documentation.