Skip to content

io.loader

Unify the GDX and .mif backends behind one loader with candidate resolution.

RemindLoader

Load IAM output symbols from a GDX file or .mif, resolving name candidates.

Source code in src/iampypsa/io/loader.py 
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
class RemindLoader:
    """Load IAM output symbols from a GDX file or ``.mif``, resolving name candidates."""

    def __init__(self, source: str | PathLike, backend: Backend | None = None) -> None:
        """Bind to a source file; detect the backend from the suffix if unset."""
        self.source: Path = Path(source)
        self.backend: Backend = backend or self.detect_backend(self.source)

    @staticmethod
    def detect_backend(source: Path) -> Backend:
        """Infer the backend (``gdx``/``mif``) from a source path's suffix."""
        suffix = Path(source).suffix.lower()
        if suffix == ".gdx":
            return "gdx"
        if suffix in (".mif", ".csv"):
            return "iamc"
        raise ValueError(f"Cannot infer backend from suffix {suffix!r} ({source})")

    def list_symbols(self) -> list[str]:
        """List the symbols/variables available in the bound source."""
        if self.backend == "gdx":
            return list_gdx_symbols(self.source)
        return list_iamc_variables(self.source)

    def resolve_symbol(self, symbol: SymbolRef) -> str:
        """Pick the first candidate name actually present in the source; raise if none."""
        candidates = [symbol] if isinstance(symbol, str) else list(symbol)
        available = set(self.list_symbols())
        for name in candidates:
            if name in available:
                return name
        raise KeyError(
            f"None of {candidates} found in {self.source.name}. "
            f"First available: {sorted(available)[:10]}"
        )

    def load_symbol(
        self,
        symbol: SymbolRef,
        rename_columns: Mapping[str, str] | None = None,
    ) -> pd.DataFrame:
        """Resolve a symbol reference (name or candidates), then load it."""
        name = self.resolve_symbol(symbol)
        if self.backend == "gdx":
            return read_gdx_symbol(self.source, name, rename_columns)
        df = read_iamc(self.source, [name])
        return df.rename(columns=dict(rename_columns)) if rename_columns else df

    def load_scalar(self, symbol: SymbolRef) -> float | str:
        """Load a scalar/string symbol (e.g. model version, run name)."""
        name = self.resolve_symbol(symbol)
        if self.backend == "gdx":
            return read_gdx_scalar(self.source, name)
        return read_iamc(self.source, [name])["value"].iloc[0]

__init__(source, backend=None)

Bind to a source file; detect the backend from the suffix if unset.

Source code in src/iampypsa/io/loader.py 
25
26
27
28
def __init__(self, source: str | PathLike, backend: Backend | None = None) -> None:
    """Bind to a source file; detect the backend from the suffix if unset."""
    self.source: Path = Path(source)
    self.backend: Backend = backend or self.detect_backend(self.source)

detect_backend(source) staticmethod

Infer the backend (gdx/mif) from a source path's suffix.

Source code in src/iampypsa/io/loader.py 
30
31
32
33
34
35
36
37
38
@staticmethod
def detect_backend(source: Path) -> Backend:
    """Infer the backend (``gdx``/``mif``) from a source path's suffix."""
    suffix = Path(source).suffix.lower()
    if suffix == ".gdx":
        return "gdx"
    if suffix in (".mif", ".csv"):
        return "iamc"
    raise ValueError(f"Cannot infer backend from suffix {suffix!r} ({source})")

list_symbols()

List the symbols/variables available in the bound source.

Source code in src/iampypsa/io/loader.py 
40
41
42
43
44
def list_symbols(self) -> list[str]:
    """List the symbols/variables available in the bound source."""
    if self.backend == "gdx":
        return list_gdx_symbols(self.source)
    return list_iamc_variables(self.source)

load_scalar(symbol)

Load a scalar/string symbol (e.g. model version, run name).

Source code in src/iampypsa/io/loader.py 
70
71
72
73
74
75
def load_scalar(self, symbol: SymbolRef) -> float | str:
    """Load a scalar/string symbol (e.g. model version, run name)."""
    name = self.resolve_symbol(symbol)
    if self.backend == "gdx":
        return read_gdx_scalar(self.source, name)
    return read_iamc(self.source, [name])["value"].iloc[0]

load_symbol(symbol, rename_columns=None)

Resolve a symbol reference (name or candidates), then load it.

Source code in src/iampypsa/io/loader.py 
58
59
60
61
62
63
64
65
66
67
68
def load_symbol(
    self,
    symbol: SymbolRef,
    rename_columns: Mapping[str, str] | None = None,
) -> pd.DataFrame:
    """Resolve a symbol reference (name or candidates), then load it."""
    name = self.resolve_symbol(symbol)
    if self.backend == "gdx":
        return read_gdx_symbol(self.source, name, rename_columns)
    df = read_iamc(self.source, [name])
    return df.rename(columns=dict(rename_columns)) if rename_columns else df

resolve_symbol(symbol)

Pick the first candidate name actually present in the source; raise if none.

Source code in src/iampypsa/io/loader.py 
46
47
48
49
50
51
52
53
54
55
56
def resolve_symbol(self, symbol: SymbolRef) -> str:
    """Pick the first candidate name actually present in the source; raise if none."""
    candidates = [symbol] if isinstance(symbol, str) else list(symbol)
    available = set(self.list_symbols())
    for name in candidates:
        if name in available:
            return name
    raise KeyError(
        f"None of {candidates} found in {self.source.name}. "
        f"First available: {sorted(available)[:10]}"
    )