r2x 0.1.10

A framework plugin manager for the r2x power systems modeling ecosystem.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

"""Stub ReEDS parser used in integration tests."""

from __future__ import annotations

import json
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from r2x_core import PluginContext


class ReEDSConfig:
    """Minimal config placeholder matching runtime expectations."""

    def __init__(
        self,
        weather_year: int | None = None,
        solve_year: int | None = None,
        **kwargs: Any,
    ) -> None:
        self.weather_year = weather_year
        self.solve_year = solve_year
        self.extra = kwargs


class ReEDSParser:
    """Parser that returns canned JSON for tests."""

    def __init__(
        self, config: ReEDSConfig | None = None, data_store: Any | None = None, **_: Any
    ) -> None:
        self.config = config
        self.data_store = data_store

    @classmethod
    def from_context(cls, ctx: PluginContext) -> ReEDSParser:
        """Create parser instance from a PluginContext."""
        return cls(config=ctx.config, data_store=ctx.store)

    def build_system(self) -> str:
        return '{"system": "reeds", "status": "ok"}'


def no_stdin_function() -> dict[str, str]:
    """Function plugin intentionally omits stdin/system params."""
    return {"system": "reeds", "status": "function-no-stdin"}


def with_system_function(system) -> dict[str, str]:
    """Function plugin that requires system/stdin deserialization."""
    data = getattr(system, "data", {})
    if isinstance(data, str):
        try:
            data = json.loads(data)
        except json.JSONDecodeError:
            data = {}
    source = data.get("system", "unknown")
    return {"system": source, "status": "function-with-system"}


def system_with_sidecar():
    """Return a System with a time-series sidecar."""
    from r2x_core.system import System

    return System(
        {
            "system": "reeds",
            "status": "with-sidecar",
            "time_series": {"directory": "system_time_series"},
        }
    )


def mark_streamed_system(system):
    """Verify and annotate a System received through a direct stream."""
    system.data["status"] = "streamed"
    return system


def noisy_json():
    """Emit a plugin diagnostic while returning machine-readable output."""
    print("plugin diagnostic")
    return {"status": "ok"}


def test_exporter(system):
    """Consume a System as a terminal exporter sink."""
    if system.data.get("system") != "reeds":
        raise ValueError("expected a ReEDS System")
    return None


class ContextExporter:
    """Context-based exporter that reads System from PluginContext."""

    def __init__(self, context):
        self.context = context

    @classmethod
    def from_context(cls, ctx: PluginContext) -> "ContextExporter":
        """Create exporter instance from a PluginContext."""
        return cls(ctx)

    def run(self) -> None:
        """Export using only PluginContext.system and return no output."""
        system = self.context.system
        if system is None or getattr(system, "data", {}).get("system") != "reeds":
            raise ValueError("expected a ReEDS System in PluginContext.system")
        return None