Python API

Public exports from r2x_plexos. Select a class or member in the contents tree.

PLEXOSConfig¶

class r2x_plexos.PLEXOSConfig(*, models=<factory>, config_path_override=None, fpath=None, model_name='default', timeseries_dir=None, horizon_year=None, weather_year=None, output_path=None, template=None, simulation_config=None)[source]¶

Configuration for PLEXOS model parser.

This configuration class defines all parameters needed to parse PLEXOS model data, including model identification, time series handling, and simulation settings. Model-specific defaults and constants should be loaded using the load_defaults() class method and used in parser logic.

Parameters:

fpath (str, optional) – Path to the XML file or its directory. A DataStore entry named xml_file takes precedence.
model_name (str, optional) – Name of the PLEXOS model. Defaults to “default”.
timeseries_dir (DirectoryPath, optional) – Optional subdirectory containing time series files. If passed it must exist.
horizon_year (int, optional) – Horizon year for the model simulation
weather_year (int, optional) – Optional weather-year metadata used by the exporter filename.
output_path (str, optional) – Directory for native XML and exported profile files.
template (str, optional) –
Selects the base XML template used to initialise the PLEXOS database. Accepts either:
- A supported PLEXOS version key. The corresponding master XML is taken from the plexosdb package (plexosdb/config/):
  
  Key
  
  File (from plexosdb)
  
  PLEXOS9.0
  
  master_9.2R6_btu.xml
  
  PLEXOS10.0
  
  master_10.0R2_btu.xml
  
  PLEXOS11.0
  
  master_11.0R4_btu.xml
  
  PLEXOS12.0
  
  master_12.0R3_btu.xml
- An absolute or relative file path to a custom XML template.
When omitted the default template (PLEXOS10.0) is used.
simulation_config (SimulationConfig, optional) – Simulation configuration parameters
models (tuple[str, ...])
config_path_override (Path | None)

Key	File (from plexosdb)
`PLEXOS9.0`	master_9.2R6_btu.xml
`PLEXOS10.0`	master_10.0R2_btu.xml
`PLEXOS11.0`	master_11.0R4_btu.xml
`PLEXOS12.0`	master_12.0R3_btu.xml

Examples

Basic configuration with a checked model/year selection:

>>> from r2x_plexos import PLEXOSConfig
>>> config = PLEXOSConfig(model_name="MyPLEXOSModel", horizon_year=2030)
>>> assert (config.model_name, config.horizon_year) == ("MyPLEXOSModel", 2030)

PLEXOSExporter¶

class r2x_plexos.PLEXOSExporter[source]¶

PLEXOS XML exporter.

Initialize the exporter with minimal state.

Notes

Actual initialization happens after context is set via from_context().

on_export()[source]¶

Initialize the exporter after context is set.

Returns:: Ok(None) on success, Err with error message on failure
Return type:: Result[None, str]

setup_configuration()[source]¶

Set up the simulation configuration in the PlexosDB.

Loads static model and horizon definitions from JSON files, merges them, and builds the simulation configuration for the specified year and resolution. Also applies simulation configuration objects (Performance, Production, etc.) if provided. Returns Ok(None) on success or Err with an error message.

Returns:: Ok(None) if setup is successful, Err(error message) otherwise.
Return type:: Result[None, str]

prepare_export()[source]¶

Add component objects to the database.

This method bulk inserts component objects (generators, nodes, etc.) into the database. It skips simulation configuration objects (Model, Horizon) as those are handled in setup_configuration(). It does NOT add properties or memberships - those are added in postprocess_export().

Return type:: Result[None, str]

postprocess_export()[source]¶

Add properties and memberships to the database.

This method: 1. Adds component properties using bulk insert from system.to_records() 2. Adds component memberships (relationships between components) 3. Exports time series to CSV files

Components without properties (PLEXOSDatafile, PLEXOSMembership) are filtered out.

Return type:: Result[None, str]

export_time_series()[source]¶

Export all time series data from the system to CSV files and update property references.

Return type:: Result[None, str]

classmethod PLEXOSExporter.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

For a complete tested example, use the Python workflow.

PLEXOSExporter.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'

For a complete tested example, use the Python workflow.

PLEXOSObject¶

class r2x_plexos.PLEXOSObject(*, uuid=<factory>, name, category=None, description=None, object_id=None, ext=None)[source]¶

Base class for PLEXOS components with automatic property resolution.

When accessing a field that contains a PropertyValue, automatically calls get_value() to resolve using global scenario priority context.

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:

uuid (UUID)
name (str)
category (str | None)
description (str | None)
object_id (int | None)
ext (dict[str, Any] | None)

model_config: ClassVar = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'protected_namespaces': (), 'str_strip_whitespace': True, 'use_enum_values': False, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True, 'validate_default': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

category: str | None¶

description: str | None¶

object_id: int | None¶

ext: dict[str, Any] | None¶

get_property_value(field_name)[source]¶

Get the raw PropertyValue object for a field without auto-resolution.

This method bypasses the automatic get_value() resolution that occurs when accessing fields directly. Use this to access the full PropertyValue object with all its complex data (filepath, datafile refs, variable refs, etc.).

Parameters:: field_name (str) – Name of the field to retrieve
Returns:: The raw field value (PropertyValue if field contains one, otherwise raw value)
Return type:: PLEXOSPropertyValue | Any

Examples

>>> datafile.filename  # Returns resolved value (e.g., 0.0)
>>> datafile.get_property_value('filename')  # Returns PLEXOSPropertyValue object
>>> datafile.get_property_value('filename').get_filepath()  # Returns filepath string

PLEXOSParser¶

class r2x_plexos.PLEXOSParser[source]¶

Parse PLEXOS XML models into r2x-core system representation.

Implements a three-phase parsing strategy: 1. Component creation with property parsing and time series reference collection 2. Time series attachment from collected references 3. Post-processing to set system metadata

Key Features¶

Scenario-aware property resolution using PLEXOS read order
Multi-band time series support for load/generation profiles
Variable-based property modification (multiply, add, subtract operations)
Collection properties tied to parent-child memberships
File parsing cache to avoid redundant CSV reads
Automatic constant detection for single-value time series
Horizon-based time series trimming

Initialize PLEXOSParser with internal state only.

Notes

Actual initialization happens in on_build() after context is set. Initializes multiple caches (python dictionaries): - _component_cache: object_id -> PLEXOSObject mapping - _parsed_files_cache: file_path -> parsed data mapping - _datafile_cache: deprecated, for backward compatibility - _membership_cache: membership_id -> PLEXOSMembership mapping - _collection_properties_cache: object_id -> property records mapping - _attached_timeseries: (uuid, field_name) -> bool attachment tracker

property config: PLEXOSConfig¶: Return the PLEXOSConfig instance.

on_build()[source]¶

Build the System from PLEXOS XML data.

Returns:: Ok(system) on success, Err() with error message on failure
Return type:: Result[System, str]

validate_inputs()[source]¶

Validate input data before parsing.

Return type:: Result[None, str]

build_system_components()[source]¶

Create PLEXOS components and establish relationships.

Return type:: Result[None, str]

build_time_series()[source]¶

Attach time series data using three-stage processing.

Return type:: Result[None, str]

postprocess_system()[source]¶

Set system metadata and log summary statistics.

Return type:: Result[None, str]

classmethod PLEXOSParser.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

For a complete tested example, use the Python workflow.

PLEXOSParser.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'

For a complete tested example, use the Python workflow.

PLEXOSProperty¶

r2x_plexos.PLEXOSProperty = <r2x_plexos.models.property_specification._PLEXOSPropertyFactory object>¶

Factory that can be used as both a callable and a default instance.

Allows both PLEXOSProperty() and PLEXOSProperty to work.

Parameters:

units (str | None)
allow_bands (bool)
is_enum (bool)

Return type:

PropertySpecification

PLEXOSPropertyValue¶

class r2x_plexos.PLEXOSPropertyValue(entries=<factory>, units=None, action=None, _by_scenario=<factory>, _by_band=<factory>, _by_timeslice=<factory>, _by_date=<factory>, _by_variable=<factory>, _by_text=<factory>)[source]¶

Optimized property value class for PLEXOS components.

Uses a hash-based dictionary for O(1) lookups with pre-built indexes for filtering by dimension (scenario, band, timeslice, dates). Designed to handle millions of property values efficiently.

Serialization¶

Supports Pydantic serialization for compatibility with infrasys System.to_json(). When used as a field in Pydantic models, serializes to a list of records compatible with plexosdb_from_records:

>>> from pydantic import TypeAdapter
>>> from r2x_plexos import PLEXOSPropertyValue
>>> adapter = TypeAdapter(PLEXOSPropertyValue)
>>> value = PLEXOSPropertyValue.from_dict({"value": 10.0})
>>> records = adapter.dump_python(value, mode="json")
>>> assert isinstance(records, list) and len(records) == 1
>>> assert records[0]["value"] == 10.0
>>> assert PLEXOSPropertyValue.from_records(records).get_value() == 10.0

Round-trip serialization is supported via Pydantic’s model_dump(mode=’json’) when used as a field type, and reconstruction from serialized format.

entries: dict[PLEXOSPropertyKey, PLEXOSRow]¶

units: str | None¶

action: str | None¶

classmethod from_dict(data)[source]¶

Create property from a dictionary specification.

Parameters:: data (Any)
Return type:: PLEXOSPropertyValue

classmethod from_db_results(results)[source]¶

Create a property from database results.

Parameters:: results (list[PLEXOSRow])
Return type:: PLEXOSPropertyValue

classmethod from_records(records, units=None)[source]¶

Create property from a list of record dictionaries.

Parameters:

records (list[dict[str, Any]])
units (str | None)

Return type:

PLEXOSPropertyValue

add_entry(value, scenario=None, band=1, timeslice=None, date_from=None, date_to=None, period_type_id=None, datafile_name=None, datafile_id=None, column_name=None, variable_name=None, variable_id=None, action=None, units=None, text=None, text_class_name=None, memo=None)[source]¶

Add a property value entry with full metadata.

Parameters:

value (Any)
scenario (str | None)
band (int)
timeslice (str | None)
date_from (str | None)
date_to (str | None)
period_type_id (int | None)
datafile_name (str | None)
datafile_id (int | None)
column_name (str | None)
variable_name (str | None)
variable_id (int | None)
action (str | None)
units (str | None)
text (str | None)
text_class_name (str | None)
memo (str | None)

Return type:

None

add_from_db_rows(rows)[source]¶

Add multiple database results - stores PLEXOSRow directly.

Parameters:: rows (PLEXOSRow | list[PLEXOSRow])
Return type:: None

get_value()[source]¶

Get property value with automatic scenario priority and horizon resolution.

Resolution order: 1. Filter by horizon (date range) if set 2. If priority context is set, use priority-based resolution 3. Pure default entry (no scenario/timeslice) takes precedence 4. Non-scenario timeslices preferred over scenarios 5. Non-scenario bands preferred over scenarios 6. Return scenario/timeslice/band dicts or simple values as appropriate

Return type:: Any

get_value_for(scenario=None, band=1, timeslice=None, date_from=None, date_to=None)[source]¶

Get value for specific dimensions with fallback logic.

Parameters:

scenario (str | None)
band (int)
timeslice (str | None)
date_from (str | None)
date_to (str | None)

Return type:

Any

get_bands()[source]¶

Get all unique bands.

Return type:: list[int]

get_dates()[source]¶

Get all unique date ranges.

Return type:: list[tuple[str | None, str | None]]

get_timeslices()[source]¶

Get all unique timeslices.

Return type:: list[str]

get_scenarios()[source]¶

Get all unique scenarios.

Return type:: list[str]

get_text()[source]¶

Get all unique text values.

Return type:: list[str]

get_variables()[source]¶

Get all unique variables.

Return type:: list[str]

get_filepath()[source]¶

Get filepath if this property references a Data File via text field.

Returns the filepath string from the text field when text_class_name=’Data File’. Useful for properties like DataFile.Filename.

Returns:: Filepath string or None if no filepath reference exists
Return type:: str | None

get_variable_reference()[source]¶

Get variable reference if this property uses a Variable.

Returns dictionary with variable metadata including name, id, and action.

Returns:: Dictionary with keys: ‘name’, ‘id’, ‘action’ or None if no variable reference
Return type:: dict | None

get_datafile_reference()[source]¶

Get datafile reference if this property uses another DataFile object.

Returns dictionary with datafile metadata including name and id. Different from get_filepath() which returns a direct filepath string.

Returns:: Dictionary with keys: ‘name’, ‘id’ or None if no datafile reference
Return type:: dict | None

get_text_value()[source]¶

Get single text value if property has exactly one text entry.

Convenience method for properties with a single text value. For multiple texts, use get_text() instead.

Returns:: Text string if exactly one text exists, None otherwise
Return type:: str | None

has_bands()[source]¶

Check if this property has multiple bands.

Return type:: bool

has_date_from()[source]¶

Check if this property has date_from constraints.

Return type:: bool

has_date_to()[source]¶

Check if this property has date_to constraints.

Return type:: bool

has_scenarios()[source]¶

Check if this property has scenario-specific values.

Return type:: bool

has_timeslices()[source]¶

Check if this property has timesliced data.

Return type:: bool

has_datafile()[source]¶

Check if this property references a datafile.

Return type:: bool

has_variable()[source]¶

Check if this property references a variable.

Return type:: bool

has_complex_data()[source]¶

Check if property has complex data beyond a simple value.

Complex data includes: - Datafile or variable references - Multiple scenarios - Multiple entries (bands, timeslices, dates)

Returns:: True if property has complex data structure
Return type:: bool

has_text()[source]¶

Check if this property has text values.

Return type:: bool

get_entry()[source]¶

Get property entry with automatic scenario priority resolution.

Similar to get_value() but returns the full PLEXOSRow entry instead of just the value. This is useful when you need to access other fields from the entry like variable_name, variable_id, action, etc.

Returns:: The highest-priority entry or None if no entries exist
Return type:: PLEXOSRow | None

get_text_with_priority()[source]¶

Get text field using scenario priority resolution.

When multiple entries have the same priority (e.g., all with no scenario), returns the first non-None text value found.

Returns:: Text from highest priority entry, or None if no text exists
Return type:: str | None

get_variable_with_priority()[source]¶

Get variable reference using scenario priority resolution.

Returns:: Dictionary with ‘name’, ‘id’, ‘action’ from highest priority entry
Return type:: dict | None

Parameters:

entries (dict[PLEXOSPropertyKey, PLEXOSRow])
units (str | None)
action (str | None)
_by_scenario (dict[str, set[PLEXOSPropertyKey]])
_by_band (dict[int, set[PLEXOSPropertyKey]])
_by_timeslice (dict[str, set[PLEXOSPropertyKey]])
_by_date (dict[tuple[str | None, str | None], set[PLEXOSPropertyKey]])
_by_variable (dict[str, set[PLEXOSPropertyKey]])
_by_text (dict[str, set[PLEXOSPropertyKey]])

PLEXOSRow¶

class r2x_plexos.PLEXOSRow(value=None, period_type=None, period_name=None, timeslice_name=None, timeslice_id=None, band=1, units=None, action='=', scenario_name=None, datafile_name=None, datafile_id=None, column_name=None, variable_name=None, variable_id=None, date_from=None, date_to=None, text=None, text_class_name=None, memo=None)[source]¶

Represents a result row from a PLEXOS database query.

Parameters:

value (Any)
period_type (str | None)
period_name (str | None)
timeslice_name (str | None)
timeslice_id (int | None)
band (int)
units (str | None)
action (str | None)
scenario_name (str | None)
datafile_name (str | None)
datafile_id (int | None)
column_name (str | None)
variable_name (str | None)
variable_id (int | None)
date_from (str | None)
date_to (str | None)
text (str | None)
text_class_name (str | None)
memo (str | None)

value: Any = None¶

period_type: str | None = None¶

period_name: str | None = None¶

timeslice_name: str | None = None¶

timeslice_id: int | None = None¶

band: int = 1¶

units: str | None = None¶

action: str | None = '='¶

scenario_name: str | None = None¶

datafile_name: str | None = None¶

datafile_id: int | None = None¶

column_name: str | None = None¶

variable_name: str | None = None¶

variable_id: int | None = None¶

date_from: str | None = None¶

date_to: str | None = None¶

text: str | None = None¶

text_class_name: str | None = None¶

memo: str | None = None¶

get_horizon¶

r2x_plexos.get_horizon()[source]¶

Get the current global horizon (date range).

Returns:: Current horizon as (date_from, date_to), or None if no horizon is set.
Return type:: tuple[str, str] or None

get_scenario_priority¶

r2x_plexos.get_scenario_priority()[source]¶

Get the current global scenario priority.

Returns:: Current scenario priority mapping (scenario name -> priority value), or None if no priority is set.
Return type:: dict[str, int] or None

horizon¶

r2x_plexos.horizon(date_from, date_to)[source]¶

Context manager for temporary horizon (date range) changes.

This filters property values to only include entries that overlap with the specified date range.

Parameters:

date_from (str) – Start date of the horizon (e.g., “2024-01-01”)
date_to (str) – End date of the horizon (e.g., “2024-12-31”)

Yields:

None

Return type:

Iterator[None]

Examples

>>> from r2x_plexos import get_horizon, horizon, set_horizon
>>> set_horizon(None)
>>> with horizon("2024-01-01", "2024-06-30"):
...     assert get_horizon() == ("2024-01-01", "2024-06-30")
>>> assert get_horizon() is None

scenario_and_horizon¶

r2x_plexos.scenario_and_horizon(priority, date_from, date_to)[source]¶

Context manager for both scenario priority and horizon.

Combines scenario priority and horizon filtering for complete temporal and scenario-based resolution.

Parameters:

priority (dict[str, int] or None) – Scenario priority mapping (lower number = higher priority)
date_from (str) – Start date of the horizon
date_to (str) – End date of the horizon

Yields:

None

Return type:

Iterator[None]

Examples

>>> from r2x_plexos import get_horizon, get_scenario_priority, scenario_and_horizon
>>> with scenario_and_horizon({"Test": 1}, "2024-01-01", "2024-12-31"):
...     assert get_scenario_priority() == {"Test": 1}
...     assert get_horizon() == ("2024-01-01", "2024-12-31")

scenario_priority¶

r2x_plexos.scenario_priority(priority)[source]¶

Context manager for temporary scenario priority changes.

Parameters:: priority (dict[str, int] or None) – Temporary scenario priority to use within the context (lower number = higher priority)
Yields:: None
Return type:: Iterator[None]

Examples

>>> from r2x_plexos import get_scenario_priority, scenario_priority, set_scenario_priority
>>> set_scenario_priority(None)
>>> with scenario_priority({"Test": 1, "Base": 2}):
...     assert get_scenario_priority() == {"Test": 1, "Base": 2}
>>> assert get_scenario_priority() is None

set_horizon¶

r2x_plexos.set_horizon(horizon)[source]¶

Set the global horizon (date range).

Parameters:: horizon (tuple[str, str] or None) – Horizon to set as (date_from, date_to), or None to clear horizon.
Return type:: None

set_scenario_priority¶

r2x_plexos.set_scenario_priority(priority)[source]¶

Set the global scenario priority.

Parameters:: priority (dict[str, int] or None) – Scenario priority mapping to set (lower number = higher priority), or None to clear priority.
Return type:: None