Create Function-Based Transform Plugins

Function-based plugins are simple, zero-overhead transforms that modify a System using a PluginConfig. They are marked with the @expose_plugin decorator for CLI discovery.

Basic Transform Function

Create a simple transform that modifies a system:

>>> from r2x_core import expose_plugin, PluginConfig, System
>>> from pydantic import Field
>>> from rust_ok import Ok, Result

>>> class RenameConfig(PluginConfig):
...     suffix: str = Field(default="_v2", description="Suffix to append to system name")

>>> @expose_plugin
... def rename_system(system: System, config: RenameConfig) -> Result[System, str]:
...     """Append suffix to system name."""
...     system.name = f"{system.name}{config.suffix}"
...     return Ok(system)

>>> # Call directly in Python:
>>> config = RenameConfig(suffix="_updated")
>>> system = System(name="my_system")
>>> result = rename_system(system, config)
>>> result.is_ok()
True
>>> result.unwrap().name
'my_system_updated'

With Configuration Validation

Use Pydantic field validators to enforce configuration constraints:

>>> from r2x_core import expose_plugin, PluginConfig, System
>>> from pydantic import Field, field_validator
>>> from rust_ok import Ok, Result

>>> class ThresholdConfig(PluginConfig):
...     threshold: int = Field(default=100, ge=0, le=1000)
...     description: str = Field(default="Filter components")
...
...     @field_validator("threshold")
...     @classmethod
...     def validate_threshold(cls, v):
...         if v % 10 != 0:
...             raise ValueError("Threshold must be a multiple of 10")
...         return v

>>> @expose_plugin
... def filter_by_threshold(
...     system: System,
...     config: ThresholdConfig,
... ) -> Result[System, str]:
...     """Filter components by threshold."""
...     # Implementation would filter based on config.threshold
...     return Ok(system)

>>> config = ThresholdConfig(threshold=100)  # Valid: multiple of 10
>>> system = System(name="test")
>>> result = filter_by_threshold(system, config)
>>> result.is_ok()
True

With Complex Configuration

Use nested Pydantic models for complex configurations:

>>> from r2x_core import expose_plugin, PluginConfig, System
>>> from pydantic import BaseModel, Field
>>> from rust_ok import Ok, Result

>>> class FilterCriteria(BaseModel):
...     min_capacity: float = Field(default=0.0)
...     max_capacity: float = Field(default=10000.0)

>>> class AdvancedFilterConfig(PluginConfig):
...     filter_criteria: FilterCriteria = Field(
...         default_factory=FilterCriteria,
...         description="Filtering thresholds"
...     )
...     remove_unmatched: bool = Field(default=False)

>>> @expose_plugin
... def advanced_filter(
...     system: System,
...     config: AdvancedFilterConfig,
... ) -> Result[System, str]:
...     """Advanced filtering with nested config."""
...     # Use config.filter_criteria.min_capacity, etc.
...     return Ok(system)

>>> config = AdvancedFilterConfig(
...     filter_criteria=FilterCriteria(min_capacity=5.0, max_capacity=500.0),
...     remove_unmatched=True,
... )
>>> system = System(name="test")
>>> result = advanced_filter(system, config)
>>> result.is_ok()
True

Error Handling with Result

Return errors when operations fail:

>>> from r2x_core import expose_plugin, PluginConfig, System
>>> from rust_ok import Ok, Err, Result

>>> class ValidateConfig(PluginConfig):
...     min_name_length: int = 3

>>> @expose_plugin
... def validate_system_name(
...     system: System,
...     config: ValidateConfig,
... ) -> Result[System, str]:
...     """Validate system name meets minimum length requirement."""
...     if len(system.name) < config.min_name_length:
...         return Err(f"System name too short (< {config.min_name_length} chars)")
...     return Ok(system)

>>> config = ValidateConfig(min_name_length=10)
>>> system = System(name="short")
>>> result = validate_system_name(system, config)
>>> result.is_err()
True
>>> result.error
'System name too short (< 10 chars)'

Direct Usage in Python

Call exposed functions directly with explicit arguments:

>>> from r2x_core import expose_plugin, PluginConfig, System
>>> from rust_ok import Ok, Result

>>> class SimpleConfig(PluginConfig):
...     action: str = "modify"

>>> @expose_plugin
... def simple_transform(system: System, config: SimpleConfig) -> Result[System, str]:
...     """Simple transform action."""
...     return Ok(system)

>>> # Direct Python usage - explicit and clear:
>>> config = SimpleConfig(action="modify")
>>> system = System(name="original")
>>> result = simple_transform(system, config)

>>> # Get the result:
>>> if result.is_ok():
...     modified_system = result.unwrap()
...     print(f"Success: {modified_system.name}")
... else:
...     print(f"Error: {result.error()}")
Success: original

Advanced: Generic Configuration Types

Use a bounded type variable for generic plugin functions across all supported Python versions:

>>> from r2x_core import expose_plugin, PluginConfig, System
>>> from rust_ok import Ok, Result
>>> from pydantic import Field

>>> # Define config types that implement a specific interface
>>> class BaseTransformConfig(PluginConfig):
...     """Base config for transform operations."""
...     pass

>>> class ScaleConfig(BaseTransformConfig):
...     """Config for scaling operations."""
...     factor: float = Field(default=1.0, ge=0.1)

>>> class RotateConfig(BaseTransformConfig):
...     """Config for rotation operations."""
...     angle: float = Field(default=0.0)

>>> from typing import TypeVar
>>> C = TypeVar("C", bound=BaseTransformConfig)
>>> @expose_plugin
... def transform_system(
...     system: System,
...     config: C,
... ) -> Result[System, str]:
...     """Generic transform accepting any BaseTransformConfig subclass."""
...     # Config type is guaranteed to be BaseTransformConfig or subclass
...     system.name = f"{system.name}_transformed"
...     return Ok(system)

>>> # Usage - type parameter is inferred from config type
>>> system = System(name="grid")
>>> scale_config = ScaleConfig(factor=2.5)
>>> result = transform_system(system, scale_config)
>>> result.unwrap().name
'grid_transformed'

>>> # Type checker ensures config matches the constraint
>>> rotate_config = RotateConfig(angle=45.0)
>>> result2 = transform_system(system, rotate_config)
>>> result2.is_ok()
True

Generic Configuration Benefits:

The bound on C lets type checkers require a BaseTransformConfig subclass while inferring the specific config type from each argument. Pydantic validates values when constructing the configuration model; the type bound does not add runtime validation.

When to Use:

Use a bounded type variable for plugins accepting multiple config types with a common interface. Use config: SpecificConfig for a single config type. Projects requiring Python 3.12 or later can instead declare the bound inline with def transform_system[C: BaseTransformConfig](...).

CLI Registration via Entry Points

To make your function discoverable by the Rust CLI, register it as an entry point in your package’s pyproject.toml:

[project.entry-points."r2x.transforms"]
rename_system = "my_package.transforms:rename_system"
filter_components = "my_package.transforms:filter_by_threshold"
advanced_filter = "my_package.transforms:advanced_filter"

The Rust CLI will:

Discover exposed functions via AST-grep scanning for @expose_plugin decorator
Introspect the PluginConfig class to generate CLI arguments
Invoke the function with parsed arguments and system data

Best Practices

✅ Do

Use PluginConfig subclasses for type safety and validation
Return Result[System, str] for consistent error handling
Document config fields with descriptions in Field()
Use Pydantic validators for complex validation logic
Keep functions pure - no side effects beyond modifying the system
Name functions clearly - indicate what they do (rename_system, filter_components)

❌ Don’t

Don’t accept arbitrary keyword arguments in the signature
Don’t modify the system in-place without returning it in a Result
Don’t use global state or external dependencies without injecting them
Don’t raise exceptions - use Err() instead
Don’t use overly complex config structures - break into nested models if needed

Full Example: Break Generators Plugin

Here’s a complete example of a real-world transform plugin:

>>> from r2x_core import expose_plugin, PluginConfig, System
>>> from pathlib import Path
>>> from pydantic import Field, field_validator
>>> from rust_ok import Ok, Err, Result

>>> class BreakGensConfig(PluginConfig):
...     """Configuration for breaking generators into sub-units."""
...
...     drop_capacity_threshold: int = Field(
...         default=5,
...         ge=0,
...         description="Generators below this capacity (MW) are dropped"
...     )
...     skip_categories: list[str] = Field(
...         default_factory=list,
...         description="Generator categories to skip"
...     )
...     break_category: str = Field(
...         default="category",
...         description="Field name to break generators on"
...     )

>>> @expose_plugin
... def break_generators(
...     system: System,
...     config: BreakGensConfig,
... ) -> Result[System, str]:
...     """Break generators into sub-units based on reference data.
...
...     This transform splits large generators into smaller units,
...     allowing more granular modeling of generation resources.
...
...     Parameters
...     ----------
...     system : System
...         The system containing generators to break
...     config : BreakGensConfig
...         Configuration for breaking logic
...
...     Returns
...     -------
...     Result[System, str]
...         Modified system with broken generators or error message
...     """
...     if config.drop_capacity_threshold < 0:
...         return Err("drop_capacity_threshold must be non-negative")
...
...     # Implementation would iterate through generators and break them
...     # based on the configuration
...
...     return Ok(system)

>>> # Example usage:
>>> config = BreakGensConfig(
...     drop_capacity_threshold=10,
...     skip_categories=["nuclear"],
...     break_category="fuel_type"
... )
>>> system = System(name="western_grid")
>>> result = break_generators(system, config)
>>> result.is_ok()
True