r2x 0.1.8

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

"""Minimal System model for integration tests."""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any
from uuid import uuid4

from r2x_core.utils.files import get_r2x_cache_path


class System:
    """Stub System class with JSON APIs expected by runtime bridges."""

    DB_FILENAME = "time_series_metadata.db"

    def __init__(self, data: dict[str, Any]) -> None:
        self.data = data

    @classmethod
    def from_dict(cls, data: dict[str, Any], time_series_parent_dir: str | Path) -> "System":
        time_series = data.get("time_series")
        if isinstance(time_series, dict) and time_series.get("directory"):
            sidecar_dir = Path(time_series["directory"])
            if not sidecar_dir.is_absolute():
                sidecar_dir = Path(time_series_parent_dir) / sidecar_dir
            if not sidecar_dir.exists() or not (sidecar_dir / cls.DB_FILENAME).exists():
                raise OSError("unable to open database file")
        return cls(data)

    def to_json(
        self, path: str | Path | None = None, overwrite: bool = False
    ) -> bytes | None:
        data = json.loads(json.dumps(self.data, ensure_ascii=False))
        time_series = data.get("time_series")
        if path is None:
            if isinstance(time_series, dict) and time_series.get("directory"):
                sidecar_dir = get_r2x_cache_path() / f"{uuid4()}_time_series"
                sidecar_dir.mkdir(parents=True, exist_ok=True)
                (sidecar_dir / self.DB_FILENAME).write_text("sidecar", encoding="utf-8")
                time_series["directory"] = str(sidecar_dir)
            return json.dumps(data, ensure_ascii=False).encode("utf-8")

        output = Path(path)
        if output.exists() and not overwrite:
            raise FileExistsError(f"{output} already exists")
        output.parent.mkdir(exist_ok=True)
        if isinstance(time_series, dict) and time_series.get("directory"):
            sidecar_dir = Path(time_series["directory"])
            if not sidecar_dir.is_absolute():
                sidecar_dir = output.parent / sidecar_dir
            sidecar_dir.mkdir(parents=True, exist_ok=True)
            (sidecar_dir / self.DB_FILENAME).write_text("sidecar", encoding="utf-8")
        output.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
        return None

    @classmethod
    def from_json(cls, source: bytes | str | Path) -> "System":
        if isinstance(source, bytes):
            return cls.from_dict(json.loads(source.decode("utf-8")), Path.cwd())
        if isinstance(source, Path):
            data = json.loads(source.read_text(encoding="utf-8"))
            return cls.from_dict(data, source.parent)
        if source.lstrip().startswith(("{", "[", '"')):
            return cls(json.loads(source))
        path = Path(source)
        return cls.from_dict(json.loads(path.read_text(encoding="utf-8")), path.parent)