API reference¶

API documentation for the technoeconomics-app package.

technoeconomics ¶

backend ¶

preset ¶

Presets: curated default plants plus their presentation.

Each preset lives in its own module and subclasses Preset. The registry below discovers them so a route can look one up by name.

IndustrialHeat ¶

Bases: Preset

Industrial process heat: grid power, a heat pump vs an electric boiler, a battery, heat demand.

build ¶

build() -> Plant

Build the default industrial-heat plant.

Source code in src/technoeconomics/backend/preset/industrial_heat.py

def build(self) -> Plant:
    """Build the default industrial-heat plant."""
    electricity = Bus(id="electricity", carrier="electricity")
    heat = Bus(id="heat", carrier="heat")
    return Plant(
        name=self.name,
        snapshots=annual_snapshots(2030),
        buses=[electricity, heat],
        components=[
            GridElectricity(
                id="grid",
                price=Sinusoidal(mean=100, amplitude=50, period=24),
                max_capacity=100,
                capex=1000000,
                bus="electricity",
                fixed=True,
                plot_color=PlotColor.GREY,
            ),
            HeatPump(
                id="heat_pump",
                electricity_bus="electricity",
                heat_bus="heat",
                fixed=True,
                plot_color=PlotColor.RED,
            ),
            ElectricBoiler(
                id="electric_boiler",
                electricity_bus="electricity",
                heat_bus="heat",
                plot_color=PlotColor.PURPLE,
            ),
            Battery(id="battery", bus="electricity", plot_color=PlotColor.GREEN),
            HeatDemand(
                id="heat_demand",
                bus="heat",
                fixed=True,  # disabling the demand leaves a degenerate problem
                load=Sinusoidal(mean=5, amplitude=4.0, period=24.0),
                plot_color=PlotColor.ORANGE,
            ),
        ],
    )

Preset ¶

Bases: ABC

A named preset: a default plant plus its presentation.

The preset completes a model: alongside the default plant and its copy it declares which results to present -- the plots and numbers the frontend should show once the plant is solved.

A preset is a class, not a dataclass, so these are ordinary class attributes: a subclass overrides them with a bare assignment and inherits the declared type from here. The ClassVar on the two sequences is what marks them as shared class state rather than per-instance defaults; the scalars need no restating.

Attributes:

name (str) –

Stable identifier used to look the preset up (e.g. in a URL).
title (str) –

Human-readable title.
description (str) –

Short blurb shown above the model.
schematic (Path | None) –

Path to an SVG drawing of the plant, or None for no drawing.
plots (tuple[Plot, ...]) –

The charts to render from a solve, in display order. A tuple: it is shared by every instance of the preset, so it must not be mutable.
numbers (tuple[Number, ...]) –

The headline numbers to render from a solve, in display order. A tuple, for the same reason as plots.

build `abstractmethod` ¶

build() -> Plant

Return a fresh default plant for this preset.

Source code in src/technoeconomics/backend/preset/base.py

@abstractmethod
def build(self) -> Plant:
    """Return a fresh default plant for this preset."""

schematic_svg ¶

schematic_svg() -> str

Read the schematic SVG so it can be inlined into the page.

Returns:

str –

The SVG markup, or an empty string when the preset declares no schematic (the
str –

client renders the drawing only when there is one).

Source code in src/technoeconomics/backend/preset/base.py

def schematic_svg(self) -> str:
    """Read the schematic SVG so it can be inlined into the page.

    Returns:
        The SVG markup, or an empty string when the preset declares no schematic (the
        client renders the drawing only when there is one).
    """
    return self.schematic.read_text(encoding="utf-8") if self.schematic else ""

get ¶

get(name: str) -> type[Preset]

Look a preset up by name.

Parameters:

name (str) –

The preset's name.

Returns:

type[Preset] –

The preset class.

Raises:

KeyError –

If no preset has that name.

Source code in src/technoeconomics/backend/preset/__init__.py

def get(name: str) -> type[Preset]:
    """Look a preset up by name.

    Args:
        name: The preset's [`name`][technoeconomics.backend.preset.base.Preset].

    Returns:
        The preset class.

    Raises:
        KeyError: If no preset has that name.
    """
    try:
        return _PRESETS[name]
    except KeyError:
        raise KeyError(
            f"unknown preset {name!r}; available: {sorted(_PRESETS)}"
        ) from None

presets ¶

presets() -> dict[str, type[Preset]]

All presets, keyed by name.

Source code in src/technoeconomics/backend/preset/__init__.py

def presets() -> dict[str, type[Preset]]:
    """All presets, keyed by name."""
    return dict(_PRESETS)

base ¶

Preset base: a curated preset pairing a default plant with how to present it.

Preset ¶

Bases: ABC

A named preset: a default plant plus its presentation.

The preset completes a model: alongside the default plant and its copy it declares which results to present -- the plots and numbers the frontend should show once the plant is solved.

A preset is a class, not a dataclass, so these are ordinary class attributes: a subclass overrides them with a bare assignment and inherits the declared type from here. The ClassVar on the two sequences is what marks them as shared class state rather than per-instance defaults; the scalars need no restating.

Attributes:

name (str) –

Stable identifier used to look the preset up (e.g. in a URL).
title (str) –

Human-readable title.
description (str) –

Short blurb shown above the model.
schematic (Path | None) –

Path to an SVG drawing of the plant, or None for no drawing.
plots (tuple[Plot, ...]) –

The charts to render from a solve, in display order. A tuple: it is shared by every instance of the preset, so it must not be mutable.
numbers (tuple[Number, ...]) –

The headline numbers to render from a solve, in display order. A tuple, for the same reason as plots.

build abstractmethod ¶

build() -> Plant

Return a fresh default plant for this preset.

Source code in src/technoeconomics/backend/preset/base.py

@abstractmethod
def build(self) -> Plant:
    """Return a fresh default plant for this preset."""

schematic_svg ¶

schematic_svg() -> str

Read the schematic SVG so it can be inlined into the page.

Returns:

str –

The SVG markup, or an empty string when the preset declares no schematic (the
str –

client renders the drawing only when there is one).

Source code in src/technoeconomics/backend/preset/base.py

def schematic_svg(self) -> str:
    """Read the schematic SVG so it can be inlined into the page.

    Returns:
        The SVG markup, or an empty string when the preset declares no schematic (the
        client renders the drawing only when there is one).
    """
    return self.schematic.read_text(encoding="utf-8") if self.schematic else ""

industrial_heat ¶

The industrial process heat preset.

IndustrialHeat ¶

Bases: Preset

Industrial process heat: grid power, a heat pump vs an electric boiler, a battery, heat demand.

build ¶

build() -> Plant

Build the default industrial-heat plant.

Source code in src/technoeconomics/backend/preset/industrial_heat.py

def build(self) -> Plant:
    """Build the default industrial-heat plant."""
    electricity = Bus(id="electricity", carrier="electricity")
    heat = Bus(id="heat", carrier="heat")
    return Plant(
        name=self.name,
        snapshots=annual_snapshots(2030),
        buses=[electricity, heat],
        components=[
            GridElectricity(
                id="grid",
                price=Sinusoidal(mean=100, amplitude=50, period=24),
                max_capacity=100,
                capex=1000000,
                bus="electricity",
                fixed=True,
                plot_color=PlotColor.GREY,
            ),
            HeatPump(
                id="heat_pump",
                electricity_bus="electricity",
                heat_bus="heat",
                fixed=True,
                plot_color=PlotColor.RED,
            ),
            ElectricBoiler(
                id="electric_boiler",
                electricity_bus="electricity",
                heat_bus="heat",
                plot_color=PlotColor.PURPLE,
            ),
            Battery(id="battery", bus="electricity", plot_color=PlotColor.GREEN),
            HeatDemand(
                id="heat_demand",
                bus="heat",
                fixed=True,  # disabling the demand leaves a degenerate problem
                load=Sinusoidal(mean=5, amplitude=4.0, period=24.0),
                plot_color=PlotColor.ORANGE,
            ),
        ],
    )

results ¶

Compute the results of a solved plant for the frontend.

Number ¶

Bases: StrEnum

A headline number a preset can request.

Plot ¶

Bases: StrEnum

A chart a preset can request.

numbers ¶

numbers(network: Network, requested: list[Number]) -> list[dict]

Compute the requested headline numbers from a solved network.

Parameters:

network (Network) –

A solved, sanitised PyPSA network.
requested (list[Number]) –

The numbers to compute, in display order.

Returns:

list[dict] –

One {"label", "value", "unit"} dict per requested number.

Source code in src/technoeconomics/backend/results.py

def numbers(network: pypsa.Network, requested: list[Number]) -> list[dict]:
    """Compute the requested headline numbers from a solved network.

    Args:
        network: A solved, sanitised PyPSA network.
        requested: The numbers to compute, in display order.

    Returns:
        One ``{"label", "value", "unit"}`` dict per requested number.
    """
    return [_number(network, n) for n in requested]

plots ¶

plots(network: Network, requested: list[Plot]) -> list[dict]

Compute the requested charts as identified ECharts options.

Parameters:

network (Network) –

A solved, sanitised PyPSA network.
requested (list[Plot]) –

The plots to compute, in display order.

Returns:

list[dict] –

{"id", "option"} dicts, where option is ready for
list[dict] –

echarts.init(el).setOption(option) and id is stable across solves so the
list[dict] –

client can update each chart in place. One request may yield several charts -- an
list[dict] –

energy balance gives one per bus carrier.

Raises:

ValueError –

If a requested plot is unknown.

Source code in src/technoeconomics/backend/results.py

def plots(network: pypsa.Network, requested: list[Plot]) -> list[dict]:
    """Compute the requested charts as identified ECharts options.

    Args:
        network: A solved, sanitised PyPSA network.
        requested: The plots to compute, in display order.

    Returns:
        ``{"id", "option"}`` dicts, where ``option`` is ready for
        ``echarts.init(el).setOption(option)`` and ``id`` is stable across solves so the
        client can update each chart in place. One request may yield several charts -- an
        energy balance gives one per bus carrier.

    Raises:
        ValueError: If a requested plot is unknown.
    """
    out: list[dict] = []
    for p in requested:
        match p:
            case Plot.ENERGY_BALANCE:
                out.extend(_energy_balance(network))
            case _:
                raise ValueError(f"unknown plot: {p}")
    return out

solve ¶

Build, optimise, and read out a plant -- the one solve path shared by web and CLI.

SolveError ¶

Bases: RuntimeError

The optimisation ran but did not reach an optimal solution.

Distinct from an internal failure: this is a property of the plant the user submitted (infeasible, unbounded, ...), so its message is written to be shown to them. Anything else escaping solve is a bug and is reported generically.

solve ¶

solve(plant: Plant, preset: Preset) -> dict

Build a plant's network, optimise it, and read out the preset's results.

Reports progress per phase via progress.emit, which the caller (e.g. a web run) may stream to the user; unheard when no sink is bound.

Parameters:

plant (Plant) –

The plant to solve.
preset (Preset) –

The preset whose numbers and plots to compute from the solution.

Returns:

dict –

{"numbers": [...], "plots": [...]} -- ready for the page.

Raises:

SolveError –

If the optimisation does not reach an optimal solution.

Source code in src/technoeconomics/backend/solve.py

def solve(plant: Plant, preset: Preset) -> dict:
    """Build a plant's network, optimise it, and read out the preset's results.

    Reports progress per phase via [`progress.emit`][technoeconomics.progress.emit], which the
    caller (e.g. a web run) may stream to the user; unheard when no sink is bound.

    Args:
        plant: The plant to solve.
        preset: The preset whose `numbers` and `plots` to compute from the solution.

    Returns:
        ``{"numbers": [...], "plots": [...]}`` -- ready for the page.

    Raises:
        SolveError: If the optimisation does not reach an optimal solution.
    """
    emit("Building network…")
    n = plant.build_network()
    emit("Solving…")
    status, condition = n.optimize(
        solver_name="highs",
        solver_options={
            "solver": "pdlp",
            "presolve": "on",
            "parallel": "on",
            "threads": 2,
        },
    )
    if status != "ok":
        raise SolveError(
            f"No optimal solution was found for this model (the solver reported "
            f"{condition!r}). Try adjusting the parameters."
        )
    emit("Solve finished.")
    n.sanitize()  # TODO: check if necessary
    return {
        "numbers": numbers(n, list(preset.numbers)),
        "plots": plots(n, list(preset.plots)),
    }

data ¶

Data access: lazy, serialisable datasets resolved to scalars or series.

Constant `dataclass` ¶

Constant(value: Param[float, Magnitude])

Bases: Dataset[float]

A single fixed value.

Attributes:

value (Param[float, Magnitude]) –

The constant to return. A magnitude: its unit and bounds come from the field holding this dataset.

compute ¶

compute(snapshots: DatetimeIndex) -> float

Return value.

Source code in src/technoeconomics/data/sources.py

def compute(self, snapshots: pd.DatetimeIndex) -> float:
    """Return ``value``."""
    return float(self.value)

Dataset `dataclass` ¶

Dataset()

Bases: ABC

Base for all datasets: a lazy, serialisable handle for one value.

The type parameter T is the resolved value's type -- float for a scalar dataset, pandas.Series for a series. Concrete datasets are frozen dataclasses subclassing Dataset[float] or Dataset[pd.Series] and implementing compute.

(De)serialisation is declared once, on the [ScalarDataset][technoeconomics.data.ScalarDataset] / [SeriesDataset][technoeconomics.data.SeriesDataset] aliases, via the class-name-tagged codec from technoeconomics.serialise; a concrete dataset therefore stays a plain pydantic-free @dataclass.

compute `abstractmethod` ¶

compute(snapshots: DatetimeIndex) -> T

Produce the value, aligned to snapshots when a series.

Source code in src/technoeconomics/data/base.py

@abstractmethod
def compute(self, snapshots: pd.DatetimeIndex) -> T:
    """Produce the value, aligned to ``snapshots`` when a series."""

resolve ¶

resolve(snapshots: DatetimeIndex) -> T

Return this dataset's value over snapshots, memoised when cache is set.

The resolution entry point, wrapping the author's compute: resolve_datasets calls it. When cache is True the value is keyed by (self, snapshots) -- a frozen dataset with hashable fields is its own key -- and computed once even under concurrent resolves; otherwise compute runs every time.

Parameters:

snapshots (DatetimeIndex) –

The horizon the value is aligned to.

Returns:

T –

The computed value. A cached value is shared by reference, so callers must treat
T –

it as read-only (resolution copies it into a component rather than mutating it).

Source code in src/technoeconomics/data/base.py

def resolve(self, snapshots: pd.DatetimeIndex) -> T:
    """Return this dataset's value over ``snapshots``, memoised when ``cache`` is set.

    The resolution entry point, wrapping the author's
    [`compute`][technoeconomics.data.Dataset.compute]:
    [`resolve_datasets`][technoeconomics.data.resolve_datasets] calls it. When ``cache``
    is ``True`` the value is keyed by ``(self, snapshots)`` -- a frozen dataset with
    hashable fields is its own key -- and computed once even under concurrent resolves;
    otherwise ``compute`` runs every time.

    Args:
        snapshots: The horizon the value is aligned to.

    Returns:
        The computed value. A cached value is shared by reference, so callers must treat
        it as read-only (resolution copies it into a component rather than mutating it).
    """
    if not self.cache:
        return self.compute(snapshots)
    return _RESULTS.get_or_make(
        (self, _snapshots_key(snapshots)), lambda: self.compute(snapshots)
    )

Sinusoidal `dataclass` ¶

Sinusoidal(mean: Param[float, Magnitude], amplitude: Param[float, Magnitude], period: Param[float, Gt(0), Unit('h')] = 24.0, phase: Param[float, Unit('h')] = 0.0)

Bases: Dataset[Series]

A sine wave over the snapshots -- a stand-in for daily/seasonal shapes.

The value at time t is mean + amplitude * sin(2*pi * (h - phase) / period), where h is the number of hours since the first snapshot.

Attributes:

mean (Param[float, Magnitude]) –

Baseline the wave oscillates around. A magnitude: its unit and bounds come from the field holding this dataset.
amplitude (Param[float, Magnitude]) –

Peak deviation from mean. A magnitude, like mean.
period (Param[float, Gt(0), Unit('h')]) –

Oscillation period in hours (e.g. 24 daily, 8760 yearly).
phase (Param[float, Unit('h')]) –

Horizontal shift in hours.

compute ¶

compute(snapshots: DatetimeIndex) -> Series

Return the sine wave sampled at snapshots.

Source code in src/technoeconomics/data/sources.py

def compute(self, snapshots: pd.DatetimeIndex) -> pd.Series:
    """Return the sine wave sampled at ``snapshots``."""
    hours = np.asarray(
        (snapshots - snapshots[0]) / pd.Timedelta(hours=1), dtype=float
    )
    values = self.mean + self.amplitude * np.sin(
        2 * np.pi * (hours - self.phase) / self.period
    )
    return pd.Series(values, index=snapshots)

resolve_datasets ¶

resolve_datasets(objs: Iterable[C], snapshots: DatetimeIndex) -> list[C]

Return copies of objs with every Dataset field replaced by its value.

Parameters:

objs (Iterable[C]) –

Dataclass instances (typically components) to resolve.
snapshots (DatetimeIndex) –

The horizon series values are aligned to.

Returns:

list[C] –

New instances with concrete values in place of datasets; instances with no
list[C] –

dataset fields are returned as is.

Source code in src/technoeconomics/data/base.py

def resolve_datasets[C: DataclassInstance](
    objs: Iterable[C], snapshots: pd.DatetimeIndex
) -> list[C]:
    """Return copies of `objs` with every ``Dataset`` field replaced by its value.

    Args:
        objs: Dataclass instances (typically components) to resolve.
        snapshots: The horizon series values are aligned to.

    Returns:
        New instances with concrete values in place of datasets; instances with no
        dataset fields are returned as is.
    """
    resolved = []
    for obj in objs:
        updates = {
            f.name: value.resolve(snapshots)
            for f in fields(obj)
            if isinstance(value := getattr(obj, f.name), Dataset)
        }
        resolved.append(replace(obj, **updates) if updates else obj)
    return resolved

base ¶

Core data-access contract: the Dataset type and dataset resolution.

Dataset `dataclass` ¶

Dataset()

Bases: ABC

Base for all datasets: a lazy, serialisable handle for one value.

The type parameter T is the resolved value's type -- float for a scalar dataset, pandas.Series for a series. Concrete datasets are frozen dataclasses subclassing Dataset[float] or Dataset[pd.Series] and implementing compute.

(De)serialisation is declared once, on the [ScalarDataset][technoeconomics.data.ScalarDataset] / [SeriesDataset][technoeconomics.data.SeriesDataset] aliases, via the class-name-tagged codec from technoeconomics.serialise; a concrete dataset therefore stays a plain pydantic-free @dataclass.

compute `abstractmethod` ¶

compute(snapshots: DatetimeIndex) -> T

Produce the value, aligned to snapshots when a series.

Source code in src/technoeconomics/data/base.py

@abstractmethod
def compute(self, snapshots: pd.DatetimeIndex) -> T:
    """Produce the value, aligned to ``snapshots`` when a series."""

resolve ¶

resolve(snapshots: DatetimeIndex) -> T

Return this dataset's value over snapshots, memoised when cache is set.

The resolution entry point, wrapping the author's compute: resolve_datasets calls it. When cache is True the value is keyed by (self, snapshots) -- a frozen dataset with hashable fields is its own key -- and computed once even under concurrent resolves; otherwise compute runs every time.

Parameters:

snapshots (DatetimeIndex) –

The horizon the value is aligned to.

Returns:

T –

The computed value. A cached value is shared by reference, so callers must treat
T –

it as read-only (resolution copies it into a component rather than mutating it).

Source code in src/technoeconomics/data/base.py

def resolve(self, snapshots: pd.DatetimeIndex) -> T:
    """Return this dataset's value over ``snapshots``, memoised when ``cache`` is set.

    The resolution entry point, wrapping the author's
    [`compute`][technoeconomics.data.Dataset.compute]:
    [`resolve_datasets`][technoeconomics.data.resolve_datasets] calls it. When ``cache``
    is ``True`` the value is keyed by ``(self, snapshots)`` -- a frozen dataset with
    hashable fields is its own key -- and computed once even under concurrent resolves;
    otherwise ``compute`` runs every time.

    Args:
        snapshots: The horizon the value is aligned to.

    Returns:
        The computed value. A cached value is shared by reference, so callers must treat
        it as read-only (resolution copies it into a component rather than mutating it).
    """
    if not self.cache:
        return self.compute(snapshots)
    return _RESULTS.get_or_make(
        (self, _snapshots_key(snapshots)), lambda: self.compute(snapshots)
    )

ResultCache ¶

ResultCache(maxsize: int = _MAXSIZE, ttl: float = _TTL_SECONDS)

Process-wide single-flight memoiser: one factory call per key, shared by racers.

A dataset resolved concurrently across the solve worker threads must compute once, not once per thread. The first caller to miss stores a Future under the key and runs the factory; callers arriving while it is in flight find that Future and block on its result. The lock guards only the TTLCache bookkeeping -- which is not thread-safe, and mutates even on a read here to refresh the idle timer -- while the factory runs unlocked, so a slow compute never blocks a hit or an unrelated key.

Parameters:

maxsize (int, default: _MAXSIZE ) –

Distinct keys retained before the least-recently-used is evicted.
ttl (float, default: _TTL_SECONDS ) –

Seconds a key survives untouched (each hit resets the timer).

Source code in src/technoeconomics/data/base.py

def __init__(self, maxsize: int = _MAXSIZE, ttl: float = _TTL_SECONDS) -> None:
    self._cache: TTLCache[Hashable, Future[Any]] = TTLCache(
        maxsize=maxsize, ttl=ttl
    )
    self._lock = Lock()

get_or_make ¶

get_or_make(key: Hashable, factory: Callable[[], T]) -> T

Return the value for key, computing it via factory on a miss.

Parameters:

key (Hashable) –

A hashable key that fully determines the value.
factory (Callable[[], T]) –

Produces the value on a miss; called at most once per in-flight key.

Returns:

T –

The value for key, freshly computed or replayed from an in-flight or earlier
T –

call. A factory that raises propagates to every waiter and is not cached, so a
T –

later call retries.

Source code in src/technoeconomics/data/base.py

def get_or_make[T](self, key: Hashable, factory: Callable[[], T]) -> T:
    """Return the value for `key`, computing it via `factory` on a miss.

    Args:
        key: A hashable key that fully determines the value.
        factory: Produces the value on a miss; called at most once per in-flight key.

    Returns:
        The value for `key`, freshly computed or replayed from an in-flight or earlier
        call. A `factory` that raises propagates to every waiter and is not cached, so a
        later call retries.
    """
    with self._lock:
        future = self._cache.get(key)
        if future is not None:
            self._cache[key] = future  # re-insert to refresh the idle TTL
            mine = False
        else:
            future = self._cache[key] = Future()
            mine = True
    if mine:
        try:
            future.set_result(factory())
        except Exception as exc:  # noqa: BLE001 -- re-raised to every waiter below
            with self._lock:
                # Drop only if still ours; a refresh/eviction may have replaced it.
                if self._cache.get(key) is future:
                    del self._cache[key]
            future.set_exception(exc)
    return future.result()

resolve_datasets ¶

resolve_datasets(objs: Iterable[C], snapshots: DatetimeIndex) -> list[C]

Return copies of objs with every Dataset field replaced by its value.

Parameters:

objs (Iterable[C]) –

Dataclass instances (typically components) to resolve.
snapshots (DatetimeIndex) –

The horizon series values are aligned to.

Returns:

list[C] –

New instances with concrete values in place of datasets; instances with no
list[C] –

dataset fields are returned as is.

Source code in src/technoeconomics/data/base.py

def resolve_datasets[C: DataclassInstance](
    objs: Iterable[C], snapshots: pd.DatetimeIndex
) -> list[C]:
    """Return copies of `objs` with every ``Dataset`` field replaced by its value.

    Args:
        objs: Dataclass instances (typically components) to resolve.
        snapshots: The horizon series values are aligned to.

    Returns:
        New instances with concrete values in place of datasets; instances with no
        dataset fields are returned as is.
    """
    resolved = []
    for obj in objs:
        updates = {
            f.name: value.resolve(snapshots)
            for f in fields(obj)
            if isinstance(value := getattr(obj, f.name), Dataset)
        }
        resolved.append(replace(obj, **updates) if updates else obj)
    return resolved

sources ¶

Concrete datasets.

Constant `dataclass` ¶

Constant(value: Param[float, Magnitude])

Bases: Dataset[float]

A single fixed value.

Attributes:

value (Param[float, Magnitude]) –

The constant to return. A magnitude: its unit and bounds come from the field holding this dataset.

compute ¶

compute(snapshots: DatetimeIndex) -> float

Return value.

Source code in src/technoeconomics/data/sources.py

def compute(self, snapshots: pd.DatetimeIndex) -> float:
    """Return ``value``."""
    return float(self.value)

Sinusoidal `dataclass` ¶

Sinusoidal(mean: Param[float, Magnitude], amplitude: Param[float, Magnitude], period: Param[float, Gt(0), Unit('h')] = 24.0, phase: Param[float, Unit('h')] = 0.0)

Bases: Dataset[Series]

A sine wave over the snapshots -- a stand-in for daily/seasonal shapes.

The value at time t is mean + amplitude * sin(2*pi * (h - phase) / period), where h is the number of hours since the first snapshot.

Attributes:

mean (Param[float, Magnitude]) –

Baseline the wave oscillates around. A magnitude: its unit and bounds come from the field holding this dataset.
amplitude (Param[float, Magnitude]) –

Peak deviation from mean. A magnitude, like mean.
period (Param[float, Gt(0), Unit('h')]) –

Oscillation period in hours (e.g. 24 daily, 8760 yearly).
phase (Param[float, Unit('h')]) –

Horizontal shift in hours.

compute ¶

compute(snapshots: DatetimeIndex) -> Series

Return the sine wave sampled at snapshots.

Source code in src/technoeconomics/data/sources.py

def compute(self, snapshots: pd.DatetimeIndex) -> pd.Series:
    """Return the sine wave sampled at ``snapshots``."""
    hours = np.asarray(
        (snapshots - snapshots[0]) / pd.Timedelta(hours=1), dtype=float
    )
    values = self.mean + self.amplitude * np.sin(
        2 * np.pi * (hours - self.phase) / self.period
    )
    return pd.Series(values, index=snapshots)

model ¶

component ¶

Component API and component directory for the model.

A model is composed of components connected through carriers.

from technoeconomics.data import Constant, Sinusoidal

heat_pump = HeatPump(
    electricity_bus="electricity", heat_bus="heat", capex=Constant(900)
)
grid = GridElectricity(
    bus="electricity", price=Sinusoidal(mean=120, amplitude=40, period=24)
)

AnyComponent ¶

AnyComponent = Annotated[Component, _component_validate, _component_dump]

A component (de)serialised through the class-name-tagged subclass registry.

Battery `dataclass` ¶

Battery(*, id: str = '', enabled: bool = True, fixed: bool = False, plot_color: PlotColor | None = None, bus: Param[str, Hidden], max_hours: Param[Param[Scalar, Gt(0)] | ScalarDataset, Unit('h')] = 4.0, capex: Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit('EUR/MW'), Advanced] = 12000.0, round_trip_efficiency: Param[Param[Scalar, Gt(0), Lt(1)] | ScalarDataset, Label('Round-trip efficiency'), Advanced] = 0.85)

Bases: Component

Electricity storage with a fixed energy-to-power ratio.

Attributes:

bus (Param[str, Hidden]) –

Electricity bus the battery attaches to.
max_hours (Param[Param[Scalar, Gt(0)] | ScalarDataset, Unit('h')]) –

Storage duration at rated power [h].
capex (Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit('EUR/MW'), Advanced]) –

Annuitised investment cost [EUR/MW].
round_trip_efficiency (Param[Param[Scalar, Gt(0), Lt(1)] | ScalarDataset, Label('Round-trip efficiency'), Advanced]) –

Fraction of stored energy returned over a full charge-discharge cycle. Split evenly across the two directions (sqrt each way) when building the PyPSA StorageUnit. Must be < 1 -- a lossless battery makes simultaneous charge and discharge free, leaving the dispatch split degenerate (non-physical "wash").

add_to_network ¶

add_to_network(n: Network) -> None

Add a StorageUnit on the electricity bus.

Source code in src/technoeconomics/model/component.py

def add_to_network(self, n: pypsa.Network) -> None:
    """Add a `StorageUnit` on the electricity bus."""
    # add_to_network runs on a resolved copy, so the dataset field is a concrete float
    # here even though its declared type still admits a ScalarDataset.
    one_way = self.round_trip_efficiency**0.5  # ty: ignore[unsupported-operator]
    n.add(
        "StorageUnit",
        self.id,
        bus=self.bus,
        carrier=self.id,
        max_hours=self.max_hours,
        capital_cost=self.capex,
        efficiency_store=one_way,
        efficiency_dispatch=one_way,
        cyclic_state_of_charge=True,
        p_nom_extendable=True,
    )

Component `dataclass` ¶

Component(*, id: str = '', enabled: bool = True, fixed: bool = False, plot_color: PlotColor | None = None)

Bases: ABC

Base for all components: shared identity, serialisation, and the build contract.

Currently components are a thin layer on top of PyPSA Network Components.

Components can conceptually be divided into four main types:

Input - inject energy into the system (e.g. grid electricity, gas connection, solar PV)
Converter - turn one form of energy into another (e.g. heat pump, gas boiler)
Storage - buffer energy in time (e.g. battery, thermal inertia of a house)
Output - a demand or a loss, destroys energy (e.g. heat demand, heat loss in a house)

The mapping to PyPSA is:

Component	PyPSA component
`Input`	`Generator`
`Converter`	`Process`
`Storage`	`StorageUnit`
`Output`	`Load`

A component is a dataclass of technoeconomic parameters plus add_to_network() -- a recipe for building a PyPSA component integrated into a model.

Parameter fields are annotated with the vocabulary from technoeconomics.model.params (unit, bounds, visibility), which the form-spec generator reads; a field that is not a numeric scalar or a dataset (e.g. a bus reference) must be marked Hidden.

Bus references are held as bus ids (strings) by name convention: a field named bus or ending in _bus is a reference to a bus in Plant.buses. The plant checks these on construction (every reference must name an existing bus).

Attributes:

id (str) –

Unique name within the plant; used as the PyPSA component name and as its own carrier, so results are attributable per component. Optional -- if left blank, the plant assigns one from the snake-case class name, enumerating (e.g. heat_pump, heat_pump_2) when a type appears more than once.
enabled (bool) –

If False, the component is skipped when the network is built.
fixed (bool) –

If True, this component may not be disabled -- a property of its role in a preset, not of its class (set at preset assembly, e.g. a demand whose removal would leave a degenerate problem). A fixed component gets no enable toggle in the form, and a submission that tries to switch it off is rejected.
plot_color (PlotColor | None) –

Colour for this component's flows in result plots. If None, PyPSA assigns one when the network is sanitised.

add_to_network `abstractmethod` ¶

add_to_network(n: Network) -> None

Expand this component into one or more PyPSA elements on n.

Called on a copy whose dataset fields have already been resolved, so self.<field> yields a concrete number or snapshot-aligned series.

Source code in src/technoeconomics/model/component.py

@abstractmethod
def add_to_network(self, n: pypsa.Network) -> None:
    """Expand this component into one or more PyPSA elements on `n`.

    Called on a copy whose dataset fields have already been resolved, so
    ``self.<field>`` yields a concrete number or snapshot-aligned series.
    """

ElectricBoiler `dataclass` ¶

ElectricBoiler(*, id: str = '', enabled: bool = True, fixed: bool = False, plot_color: PlotColor | None = None, electricity_bus: Param[str, Hidden], heat_bus: Param[str, Hidden], efficiency: Param[Scalar, Gt(0), Le(1)] | Timeseries | SeriesDataset = 0.99, capex: Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit('EUR/MW'), Advanced] = 100.0)

Bases: Component

Electric resistance heating: cheap to install but ~unity efficiency.

Attributes:

electricity_bus (Param[str, Hidden]) –

Bus the boiler draws electricity from (input).
heat_bus (Param[str, Hidden]) –

Bus the boiler delivers heat to (output).
efficiency (Param[Scalar, Gt(0), Le(1)] | Timeseries | SeriesDataset) –

Heat out per unit electricity in.
capex (Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit('EUR/MW'), Advanced]) –

Annuitised investment cost [EUR/MW of electricity input].

add_to_network ¶

add_to_network(n: Network) -> None

Add a Process converting electricity to heat at efficiency.

Source code in src/technoeconomics/model/component.py

def add_to_network(self, n: pypsa.Network) -> None:
    """Add a `Process` converting electricity to heat at `efficiency`."""
    n.add(
        "Process",
        self.id,
        bus0=self.electricity_bus,
        bus1=self.heat_bus,
        carrier=self.id,
        rate1=self.efficiency,
        capital_cost=self.capex,
        p_nom_extendable=True,
    )

GridElectricity `dataclass` ¶

GridElectricity(*, id: str = '', enabled: bool = True, fixed: bool = False, plot_color: PlotColor | None = None, bus: Param[str, Hidden], price: Param[Scalar | Timeseries | SeriesDataset, Unit('EUR/MWh')] = 120.0, max_capacity: Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit('MW'), Advanced] = 1000.0, capex: Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit('EUR/MW'), Advanced] = 0.0)

Bases: Component

Grid connection injecting electricity at a (possibly time-varying) price.

Attributes:

bus (Param[str, Hidden]) –

Electricity bus to inject into.
price (Param[Scalar | Timeseries | SeriesDataset, Unit('EUR/MWh')]) –

Marginal cost of electricity [EUR/MWh].
max_capacity (Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit('MW'), Advanced]) –

Maximum injection power [MW].
capex (Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit('EUR/MW'), Advanced]) –

Annuitised investment cost [EUR/MW].

add_to_network ¶

add_to_network(n: Network) -> None

Add a Generator injecting electricity at price.

Source code in src/technoeconomics/model/component.py

def add_to_network(self, n: pypsa.Network) -> None:
    """Add a `Generator` injecting electricity at `price`."""
    n.add(
        "Generator",
        self.id,
        bus=self.bus,
        carrier=self.id,
        marginal_cost=self.price,
        capital_cost=self.capex,
        p_nom_max=self.max_capacity,
        p_nom_extendable=True,
    )

HeatDemand `dataclass` ¶

HeatDemand(*, id: str = '', enabled: bool = True, fixed: bool = False, plot_color: PlotColor | None = None, bus: Param[str, Hidden], load: Param[Param[Scalar, Ge(0)] | Timeseries | SeriesDataset, Unit('MW')] = 10.0)

Bases: Component

A heat demand (load) on a heat bus.

Attributes:

bus (Param[str, Hidden]) –

Heat bus the demand is drawn from.
load (Param[Param[Scalar, Ge(0)] | Timeseries | SeriesDataset, Unit('MW')]) –

Heat demand [MW].

add_to_network ¶

add_to_network(n: Network) -> None

Add a Load representing the heat demand.

Source code in src/technoeconomics/model/component.py

def add_to_network(self, n: pypsa.Network) -> None:
    """Add a `Load` representing the heat demand."""
    n.add(
        "Load",
        self.id,
        bus=self.bus,
        carrier=self.id,
        p_set=self.load,
    )

HeatPump `dataclass` ¶

HeatPump(*, id: str = '', enabled: bool = True, fixed: bool = False, plot_color: PlotColor | None = None, electricity_bus: Param[str, Hidden], heat_bus: Param[str, Hidden], cop: Param[Param[Scalar, Gt(0)] | Timeseries | SeriesDataset, Label('COP')] = 3.0, capex: Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit('EUR/MW'), Advanced] = 900000.0)

Bases: Component

Electricity-to-heat conversion (a PyPSA Process) with a constant or varying COP.

Attributes:

electricity_bus (Param[str, Hidden]) –

Bus the heat pump draws electricity from (input).
heat_bus (Param[str, Hidden]) –

Bus the heat pump delivers heat to (output).
cop (Param[Param[Scalar, Gt(0)] | Timeseries | SeriesDataset, Label('COP')]) –

Coefficient of performance.
capex (Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit('EUR/MW'), Advanced]) –

Annuitised investment cost [EUR/MW of electricity input].

add_to_network ¶

add_to_network(n: Network) -> None

Add a Process converting electricity (rate0=-1) to heat (rate1=cop).

Source code in src/technoeconomics/model/component.py

def add_to_network(self, n: pypsa.Network) -> None:
    """Add a `Process` converting electricity (`rate0=-1`) to heat (`rate1=cop`)."""
    n.add(
        "Process",
        self.id,
        bus0=self.electricity_bus,
        bus1=self.heat_bus,
        carrier=self.id,
        rate1=self.cop,
        capital_cost=self.capex,
        p_nom_extendable=True,
    )

PlotColor ¶

Bases: StrEnum

A curated palette for component colours in result plots.

params ¶

The preset-author vocabulary: everything a parameter annotation needs, one import.

Preset authors describe a component or dataset field with a single Param annotation carrying markers from this module -- unit, bounds, visibility -- and never import typing, pydantic, or annotated_types themselves:

from technoeconomics.model.params import Advanced, Ge, Hidden, Param, Unit

period: Param[float, Gt(0), Unit("h")] = 24.0
capex: Param[Param[Scalar, Ge(0)] | ScalarDataset, Unit("EUR/MW"), Advanced] = 400.0
bus: Param[str, Hidden]

Param is typing.Annotated under a friendlier name, so type checkers see the plain value type; the bounds (Ge/Gt/Le/Lt) are annotated_types re-exports, so pydantic enforces them natively with per-field error locations.

Bounds constrain user-supplied scalars only -- their job is helpful input validation, and pydantic cannot guarantee what a dataset will compute. On a field that can also hold a dataset, the bound therefore goes on the scalar branch of the union (Param[Scalar, Ge(0)] | ScalarDataset), never on the union itself: pydantic applies an outer bound to whatever the union yields and raises a bare TypeError when that is a dataset value. The form-spec generator (technoeconomics.web.spec) reads the scalar-branch bound as the field's effective bound, and propagates it (with the unit) to the Magnitude leaves of a dataset in that field -- those leaves are user-supplied scalars too, checked by the solve endpoint's spec-loop since they sit outside pydantic's reach.

A propagated bound constrains each leaf it lands on, not the value the dataset computes. load: Param[Param[Scalar, Ge(0)] | SeriesDataset, Unit("MW")] holding a Sinusoidal bounds mean >= 0 and amplitude >= 0 -- it does not stop mean=1, amplitude=100 from resolving to a series that dips to -99 MW. Bounds are input hygiene; a dataset that has a resolved-value invariant has to enforce it itself.

Advanced `module-attribute` ¶

Advanced = _Flag('Advanced')

Render this field under the form's "Advanced" section rather than inline.

Hidden `module-attribute` ¶

Hidden = _Flag('Hidden')

Exclude this field from the form entirely (bus references, non-input plumbing).

A hidden field gets no spec path, so it is not addressable by a submission at all: it keeps whatever the preset gave it.

Magnitude `module-attribute` ¶

Magnitude = _Flag('Magnitude')

Mark a dataset field as carrying the dataset's magnitude.

A magnitude leaf takes its unit and bounds from the field holding the dataset (the use), not from the dataset class -- mean is EUR/MWh in a price field and MW in a demand field.

Param `module-attribute` ¶

Param = Annotated

typing.Annotated under an author-friendly name: Param[float, Ge(0), Unit("h")].

Label `dataclass` ¶

Label(text: str)

The human-readable name a field is shown under (e.g. Label("COP")).

Without one the form derives a label from the field's path (round_trip_efficiency, price.mean), which cannot produce an initialism or a properly cased phrase.

On a component field holding a dataset the label is a prefix: the form renders each leaf as "<field label> <leaf label>" (so Label("Heat demand") over a Sinusoidal gives "Heat demand mean", "Heat demand amplitude", ...), where the leaf label is the dataset field's own Label if it has one, else its name.

Attributes:

text (str) –

The display text, e.g. "Round-trip efficiency".

Unit `dataclass` ¶

Unit(text: str)

The unit a numeric field is displayed and edited in (e.g. Unit("EUR/MWh")).

On a component's dataset field the unit is per-use: it propagates to the leaves the dataset marks Magnitude, so one dataset class can serve as a price in one field and a demand in another.

Attributes:

text (str) –

The display text, e.g. "EUR/MWh".

plant ¶

Plant: the intermediate representation of a model.

Plant `dataclass` ¶

Plant(name: str, snapshots: Snapshots, buses: list[Bus], components: list[AnyComponent])

Editable description of a model -- the system of record.

Attributes:

name (str) –

Human-readable model name.
snapshots (Snapshots) –

Optimisation horizon (use [annual_snapshots][] or any index).
buses (list[Bus]) –

Named nodes; several may share a carrier (e.g. multiple heat buses).
components (list[AnyComponent]) –

Building blocks referencing buses by id. Components may omit id; blanks are filled from the snake-case class name and enumerated (heat_pump, heat_pump_2, ...) so every component ends up uniquely named.

build_network ¶

build_network() -> Network

Compile this plant into a disposable PyPSA network ready to optimise.

Registers a carrier for each bus (the energy carriers) and a distinct carrier per enabled component (named after its id), so per-component flows stay attributable. Adds the buses, then resolves the enabled components' dataset fields to concrete values.

Source code in src/technoeconomics/model/plant.py

def build_network(self) -> pypsa.Network:
    """Compile this plant into a disposable PyPSA network ready to optimise.

    Registers a carrier for each bus (the energy carriers) and a distinct
    carrier per enabled component (named after its id), so per-component flows
    stay attributable. Adds the buses, then resolves the enabled components'
    dataset fields to concrete values.
    """
    import pypsa

    n = pypsa.Network()
    n.set_snapshots(self.snapshots)  # ty: ignore[invalid-argument-type]

    enabled = [c for c in self.components if c.enabled]

    for carrier in {b.carrier for b in self.buses}:
        n.add("Carrier", carrier)
    for c in enabled:
        n.add("Carrier", c.id, **({"color": c.plot_color} if c.plot_color else {}))

    for b in self.buses:
        n.add("Bus", b.id, carrier=b.carrier)

    for c in resolve_datasets(enabled, self.snapshots):
        c.add_to_network(n)

    return n

from_dict `classmethod` ¶

from_dict(d: dict) -> Plant

Reconstruct a plant from to_dict output.

The pydantic schema validates the structure and every field, and __post_init__ checks referential integrity, so an invalid dict is rejected here rather than failing later at solve time.

Parameters:

d (dict) –

A dict produced by to_dict.

Returns:

Plant –

The reconstructed plant.

Raises:

ValueError –

If d is not a dict, carries an unsupported schema version, or is not a valid plant (pydantic's ValidationError is itself a ValueError).

Source code in src/technoeconomics/model/plant.py

@classmethod
def from_dict(cls, d: dict) -> Plant:
    """Reconstruct a plant from [`to_dict`][technoeconomics.model.plant.Plant.to_dict] output.

    The pydantic schema validates the structure and every field, and `__post_init__`
    checks referential integrity, so an invalid dict is rejected here rather than
    failing later at solve time.

    Args:
        d: A dict produced by [`to_dict`][technoeconomics.model.plant.Plant.to_dict].

    Returns:
        The reconstructed plant.

    Raises:
        ValueError: If `d` is not a dict, carries an unsupported schema version, or is
            not a valid plant (pydantic's `ValidationError` is itself a `ValueError`).
    """
    if not isinstance(d, dict):
        raise ValueError("plant must be an object")
    payload = dict(d)
    if payload.pop("v", None) != 1:
        raise ValueError("unsupported plant version")
    return type_adapter(cls).validate_python(payload)

to_dict ¶

to_dict() -> dict

Serialise to a JSON-able dict; round-trips through from_dict.

The whole plant is one declared pydantic schema: buses become plain dicts, bus references stay bus ids, snapshots compress to {start, periods, freq}, and each component/dataset is a class-name-tagged dict. The output is stamped with a schema version so a dict written against an older layout is rejected by from_dict rather than silently mis-decoded.

Source code in src/technoeconomics/model/plant.py

def to_dict(self) -> dict:
    """Serialise to a JSON-able dict; round-trips through [`from_dict`][technoeconomics.model.plant.Plant.from_dict].

    The whole plant is one declared pydantic schema: buses become plain dicts, bus
    references stay bus ids, snapshots compress to ``{start, periods, freq}``, and each
    component/dataset is a class-name-tagged dict. The output is stamped with a schema
    version so a dict written against an older layout is rejected by
    [`from_dict`][technoeconomics.model.plant.Plant.from_dict] rather than silently
    mis-decoded.
    """
    return {"v": 1, **type_adapter(Plant).dump_python(self, mode="json")}

annual_snapshots ¶

annual_snapshots(year: int = 2013, freq: str = 'h') -> DatetimeIndex

Snapshots spanning one calendar year (hourly by default).

Convenience for the common case; Plant.snapshots accepts any pandas.DatetimeIndex.

Source code in src/technoeconomics/model/plant.py

def annual_snapshots(year: int = 2013, freq: str = "h") -> pd.DatetimeIndex:
    """Snapshots spanning one calendar year (hourly by default).

    Convenience for the common case; `Plant.snapshots` accepts any
    `pandas.DatetimeIndex`.
    """
    return pd.date_range(
        f"{year}-01-01", f"{year + 1}-01-01", freq=freq, inclusive="left"
    )

structure ¶

Structural primitives of a model: the balancing nodes components attach to.

Bus `dataclass` ¶

Bus(id: str, carrier: str)

A named balancing node, tagged with one energy carrier.

Attributes:

id (str) –

Unique bus name. Used as the PyPSA bus name when the plant is built.
carrier (str) –

Energy carrier of the node (e.g. "electricity", "heat").

progress ¶

A tiny progress channel: components emit lines, the caller decides where they go.

Neutral and low-level -- it imports nothing from web/backend/model, so any layer (the solver here, and later dataset retrieval deep in the model) can emit a progress line without knowing who, if anyone, is listening. A listener is a sink bound around some work with sink; a web run binds one that streams to the page for the duration of a solve. With no sink bound -- the CLI, tests -- emit is a no-op.

The sink lives in a ContextVar, so emit anywhere in the dynamic scope of the sink(...) block reaches it -- including deep inside a synchronous call tree like build_network.

emit ¶

emit(text: str) -> None

Report a progress line to the bound sink, if any.

A no-op when no sink is bound, so callers can emit unconditionally.

Parameters:

text (str) –

The human-readable progress line (e.g. "Building network…").

Source code in src/technoeconomics/progress.py

def emit(text: str) -> None:
    """Report a progress line to the bound sink, if any.

    A no-op when no sink is bound, so callers can emit unconditionally.

    Args:
        text: The human-readable progress line (e.g. ``"Building network…"``).
    """
    sink = _sink.get()
    if sink is not None:
        sink(text)

sink ¶

sink(fn: Callable[[str], None]) -> Iterator[None]

Bind fn as the progress sink for the duration of the with block.

Parameters:

fn (Callable[[str], None]) –

Receives each emitted progress line.

Yields:

None –

Nothing; the sink is bound until the block exits.

Source code in src/technoeconomics/progress.py

@contextmanager
def sink(fn: Callable[[str], None]) -> Iterator[None]:
    """Bind `fn` as the progress sink for the duration of the ``with`` block.

    Args:
        fn: Receives each emitted progress line.

    Yields:
        Nothing; the sink is bound until the block exits.
    """
    token = _sink.set(fn)
    try:
        yield
    finally:
        _sink.reset(token)

serialise ¶

Framework-internal (de)serialisation machinery, collected in one place.

Snapshots ¶

Snapshots = Annotated[InstanceOf[DatetimeIndex], PlainSerializer(_snapshots_to_dict), BeforeValidator(_snapshots_from_dict_or_passthrough)]

A DatetimeIndex serialised as a compact {start, periods, freq} (or explicit values).

concrete_subclasses ¶

concrete_subclasses(base: type) -> dict[str, type]

Map class name -> every (recursive) subclass of base (for type-tag lookup).

Source code in src/technoeconomics/serialise.py

def concrete_subclasses(base: type) -> dict[str, type]:
    """Map class name -> every (recursive) subclass of ``base`` (for type-tag lookup)."""
    found: dict[str, type] = {}
    for cls in base.__subclasses__():
        found[cls.__name__] = cls
        found.update(concrete_subclasses(cls))
    return found

is_timeseries ¶

is_timeseries(value: object) -> bool

True for a value the Timeseries alias admits, so callers need not respell the union.

The alias's arms are not introspectable at runtime (it is a PEP 695 alias over an Annotated), so the membership test lives here, next to the definition it mirrors.

Parameters:

value (object) –

The candidate value.

Returns:

bool –

True for an ndarray, a pandas.Series, or a plain sequence of numbers (a str
bool –

is a sequence but never a timeseries).

Source code in src/technoeconomics/serialise.py

def is_timeseries(value: object) -> bool:
    """True for a value the `Timeseries` alias admits, so callers need not respell the union.

    The alias's arms are not introspectable at runtime (it is a PEP 695 alias over an
    `Annotated`), so the membership test lives here, next to the definition it mirrors.

    Args:
        value: The candidate value.

    Returns:
        True for an ndarray, a `pandas.Series`, or a plain sequence of numbers (a `str`
        is a sequence but never a timeseries).
    """
    if isinstance(value, (np.ndarray, pd.Series)):
        return True
    return isinstance(value, Sequence) and not isinstance(value, (str, bytes))

tagged_codec ¶

tagged_codec(base: type) -> tuple[WrapValidator, PlainSerializer]

Symmetric class-name-tagged (de)serialisation for an open subclass registry.

The returned pair is attached to an Annotated alias of base: dumping writes a {"type": <class name>, ...fields} dict, and loading reads that tag to pick the concrete subclass and validate its fields. This keeps polymorphic model types (datasets, components) as plain dataclasses that gain declarative (de)serialisation from the alias.

Parameters:

base (type) –

The registry root whose concrete subclasses are the dispatch targets.

Returns:

tuple[WrapValidator, PlainSerializer] –

A (WrapValidator, PlainSerializer) pair to place in an Annotated[base, ...].

Source code in src/technoeconomics/serialise.py

def tagged_codec(base: type) -> tuple[WrapValidator, PlainSerializer]:
    """Symmetric class-name-tagged (de)serialisation for an open subclass registry.

    The returned pair is attached to an `Annotated` alias of `base`: dumping writes a
    ``{"type": <class name>, ...fields}`` dict, and loading reads that tag to pick the
    concrete subclass and validate its fields. This keeps polymorphic model types (datasets,
    components) as plain dataclasses that gain declarative (de)serialisation from the alias.

    Args:
        base: The registry root whose concrete subclasses are the dispatch targets.

    Returns:
        A ``(WrapValidator, PlainSerializer)`` pair to place in an `Annotated[base, ...]`.
    """

    def dump(v: object) -> dict:
        cls = type(v)
        assert "type" not in {f.name for f in fields(cls)}, (
            f"{cls.__name__} defines a field named 'type', clashing with the type tag"
        )
        return {"type": cls.__name__, **type_adapter(cls).dump_python(v, mode="json")}

    def load(v: object, handler: ValidatorFunctionWrapHandler) -> object:
        if isinstance(v, base):  # in-memory construction path: pass the object through
            return v
        if not isinstance(v, dict):
            raise ValueError(f"{base.__name__.lower()} must be an object")
        payload = dict(v)  # never mutate the caller's dict
        cls = concrete_subclasses(base).get(payload.pop("type", None))
        if cls is None:
            raise ValueError(f"unknown {base.__name__.lower()} type")
        return type_adapter(cls).validate_python(payload)

    return WrapValidator(load), PlainSerializer(dump)

type_adapter `cached` ¶

type_adapter(cls: type) -> TypeAdapter[Any]

A cached pydantic validator/serialiser for one model class.

Built from the class's dataclass fields, so model classes (datasets, components, buses) stay plain pydantic-free @dataclasses while gaining declarative field validation and JSON (de)serialisation. Cached because building a TypeAdapter compiles a validator, and there are few classes.

Parameters:

cls (type) –

The concrete dataclass to (de)serialise.

Returns:

TypeAdapter[Any] –

A TypeAdapter for cls.

Source code in src/technoeconomics/serialise.py

@cache
def type_adapter(cls: type) -> TypeAdapter[Any]:
    """A cached pydantic validator/serialiser for one model class.

    Built from the class's dataclass fields, so model classes (datasets, components, buses)
    stay plain pydantic-free ``@dataclass``es while gaining declarative field validation and
    JSON (de)serialisation. Cached because building a `TypeAdapter` compiles a validator, and
    there are few classes.

    Args:
        cls: The concrete dataclass to (de)serialise.

    Returns:
        A `TypeAdapter` for `cls`.
    """
    return TypeAdapter(cls)

web ¶

api ¶

The JSON API: registry-driven endpoints wiring the spec generator to the run store.

Overlay-solve throughout: the client posts {preset, overlay, enabled}; the server rebuilds from the trusted preset default, validates and applies the whitelisted edits, and launches a run. Results are a GET-able resource on the run; progress streams over SSE (progress-only, one terminal poke, then the server closes the stream). No preset is hardcoded -- everything is keyed off the preset registry, so a new preset is picked up with no changes here.

Throttling is day-one API shape:

Single-flight on a client-generated UUID (X-Client-Id): a new solve supersedes that client's own in-flight run -- a common path (two tabs share one UUID), handled politely, not a hardening measure. A request without the header is treated as its own one-shot client, so omitting it can neither share a single-flight slot nor hide from the caps below.
Caps spoofing can't bypass: a per-IP concurrency cap (N > 1, tolerating campus NAT) and a global live-run cap, both 429 + Retry-After.

In-flight bookkeeping is pruned lazily against the run store on each solve, so it needs no completion callback and cannot leak entries.

create_share ¶

create_share(req: Envelope) -> ShareToken

Encode a {preset, overlay, enabled} envelope to a share token.

Source code in src/technoeconomics/web/api.py

@router.post("/share")
def create_share(req: Envelope) -> ShareToken:
    """Encode a ``{preset, overlay, enabled}`` envelope to a share token."""
    _preset(req.preset, code=422)  # only a known preset can be shared
    return ShareToken(token=share_codec.encode(req))

get_preset ¶

get_preset(name: str) -> PresetDetail

A preset's presentation, default plant, and form spec for the client to render.

Source code in src/technoeconomics/web/api.py

@router.get("/presets/{name}")
def get_preset(name: str) -> PresetDetail:
    """A preset's presentation, default plant, and form spec for the client to render."""
    preset = _preset(name)
    plant = preset.build()
    return PresetDetail(
        title=preset.title,
        description=preset.description,
        schematic_svg=preset.schematic_svg(),
        plant=plant.to_dict(),
        form=plant_to_spec(plant),
    )

get_run ¶

get_run(run_id: str) -> RunSnapshot

A run's status and, once done, its results ({numbers, charts}).

Source code in src/technoeconomics/web/api.py

@router.get("/runs/{run_id}")
def get_run(run_id: str) -> RunSnapshot:
    """A run's status and, once done, its results (``{numbers, charts}``)."""
    run = manager.get(run_id)
    if run is None:
        raise HTTPException(404, "unknown or expired run")
    return run.snapshot()

list_presets ¶

list_presets() -> list[PresetSummary]

List every preset for the picker.

Source code in src/technoeconomics/web/api.py

@router.get("/presets")
def list_presets() -> list[PresetSummary]:
    """List every preset for the picker."""
    return [
        PresetSummary(name=p.name, title=p.title, description=p.description)
        for p in registry.presets().values()
    ]

read_share ¶

read_share(token: str) -> Envelope

Decode a share token back to its {preset, overlay, enabled} envelope.

Source code in src/technoeconomics/web/api.py

@router.get("/share/{token}")
def read_share(token: str) -> Envelope:
    """Decode a share token back to its ``{preset, overlay, enabled}`` envelope."""
    try:
        return share_codec.decode(token)
    except ValueError as e:
        raise HTTPException(422, str(e)) from e

run_events `async` ¶

run_events(run_id: str, request: Request) -> AsyncIterator[ServerSentEvent]

Stream a run's progress as SSE, resuming from Last-Event-ID on reconnect.

Progress-only: the stream carries progress lines and one terminal done/failed/cancelled poke, then ends (the client GETs the result). An unknown/expired run yields one failed event rather than a 404, so the browser's EventSource does not reconnect into a loop.

Source code in src/technoeconomics/web/api.py

@router.get(
    "/runs/{run_id}/events",
    response_class=EventSourceResponse,
    response_model=None,
)
async def run_events(run_id: str, request: Request) -> AsyncIterator[ServerSentEvent]:
    """Stream a run's progress as SSE, resuming from ``Last-Event-ID`` on reconnect.

    Progress-only: the stream carries progress lines and one terminal
    ``done``/``failed``/``cancelled`` poke, then ends (the client GETs the result). An
    unknown/expired run yields one ``failed`` event
    rather than a 404, so the browser's `EventSource` does not reconnect into a loop.
    """
    run = manager.get(run_id)
    if run is None:
        yield ServerSentEvent(event="failed", raw_data="run not found or expired")
        return
    header = request.headers.get("Last-Event-ID")
    last_id = int(header) if header is not None and header.isdigit() else None
    async for event in run.events(last_id):
        yield event

solve `async` ¶

solve(req: Envelope, request: Request, x_client_id: Annotated[str | None, Header()] = None) -> RunAccepted | JSONResponse

Validate and apply an overlay onto the preset default, then launch a run.

Returns 202 {run_id} on success; 422 {errors} for an invalid overlay path or an out-of-bounds value; 429 if a cap is hit. A client's own in-flight run is superseded rather than rejected.

Source code in src/technoeconomics/web/api.py

@router.post("/solve", status_code=202, response_model=RunAccepted)
async def solve(
    req: Envelope,
    request: Request,
    x_client_id: Annotated[str | None, Header()] = None,
) -> RunAccepted | JSONResponse:
    """Validate and apply an overlay onto the preset default, then launch a run.

    Returns 202 ``{run_id}`` on success; 422 ``{errors}`` for an invalid overlay path or an
    out-of-bounds value; 429 if a cap is hit. A client's own in-flight run is superseded
    rather than rejected.
    """
    preset = _preset(req.preset, code=422)
    base = preset.build()
    spec = plant_to_spec(base)
    errors = validate_edits(spec, req.overlay, req.enabled)
    if errors:
        return JSONResponse(status_code=422, content={"errors": errors})
    # An absent header becomes a fresh id rather than a shared empty string: see `_admit`.
    client_id = x_client_id or uuid4().hex
    ip = request.client.host if request.client else ""
    _admit(client_id, ip)
    plant = apply_edits(base, req.overlay, req.enabled)
    run = manager.launch(plant, preset)
    _inflight[run.id] = _InFlight(client_id=client_id, ip=ip)
    return RunAccepted(run_id=run.id)

envelope ¶

The uniform client envelope: a preset plus the user's overlay edits and enable toggles.

One type for the whole client contract -- the solve and share request bodies and the share-link codec all carry the same {preset, overlay, enabled}. A plain frozen dataclass validated at the boundary by pydantic (FastAPI for request bodies, TypeAdapter for the share codec), consistent with the model layer's plain-dataclass style.

Only the envelope's shape is checked here (preset is a string, overlay values are numbers, enable toggles are booleans, no unknown keys). Its meaning -- that overlay paths are real spec paths and values are in bounds -- is enforced at solve against the preset spec by validate_edits, not here.

Envelope `dataclass` ¶

Envelope(preset: str, overlay: dict[str, float] = dict(), enabled: dict[str, bool] = dict())

A preset plus the user's overlay edits and enable toggles.

Attributes:

preset (str) –

The preset name the overlay applies over.
overlay (dict[str, float]) –

{spec_path: value} scalar edits.
enabled (dict[str, bool]) –

{component_id: bool} enable toggles.

main ¶

FastAPI application: the JSON API, plus the built Svelte frontend when served from here.

No gzip middleware: it buffers text/event-stream, which would break the progress SSE. Compression of the static assets and JSON is the reverse proxy's job (Caddy), which must also be configured not to buffer the SSE endpoint.

The frontend is a separate build artifact, not part of this Python package: the wheel contains no static files and this module never reaches outside its own distribution to find any. In production the container builds frontend/dist in its own stage and points FRONTEND_DIST at it. In development the Vite dev server serves the frontend and proxies /api here (see frontend/vite.config.ts), so the variable is unset and nothing is mounted. If it is set and does not name a directory, StaticFiles raises at startup -- a mis-built image should fail loudly, not boot and serve 404s.

lifespan `async` ¶

lifespan(app: FastAPI) -> AsyncIterator[None]

Set up logging on startup; drain in-flight runs on shutdown.

Source code in src/technoeconomics/web/main.py

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Set up logging on startup; drain in-flight runs on shutdown."""
    logging.getLogger("technoeconomics").setLevel(logging.INFO)
    yield
    await api.manager.aclose_all()

runs ¶

Runs: launch a solve, stream its progress, store its result -- addressed by run_id.

A Run is addressable by an opaque run_id; there are no user sessions and no cross-run history. Results are a GET-able resource stored on the run, and the SSE stream is progress-only: it carries progress lines and one terminal poke (done, failed or cancelled), after which the server closes the stream and the client GETs the result (no reconnect-into-404 loop).

The lifecycle invariants that keep this correct, each small but load-bearing:

Task pinning. The event loop holds a running task only weakly and the completed-run cache is evictable, so a running solve's task is pinned in a manager-level set until it finishes -- otherwise an eviction mid-solve could silently drop it.
Result before terminal event. The result is stored on the run before the terminal SSE event is emitted, so a client's done -> GET can never race an empty run.
TTL from completion. A running run is pinned (never evicted); on completion it moves into a bounded TTLCache whose clock starts at completion, so a long run cannot expire from under its own reconnect, and completed-run memory (which holds full chart payloads) stays bounded.
Reconnect replay. Progress events carry monotonic ids; on reconnect the browser sends Last-Event-ID and the run replays only newer events, so the console is not duplicated.
Watchdog. A wedged solve is marked failed after a deadline so its accounting frees up. The worker thread and its CapacityLimiter slot are unrecoverable (threads can't be interrupted) but the user is not locked out until restart.

Run ¶

Run(id: str, manager: RunManager, preset_name: str)

One solve, addressed by id: its status, result, progress tail, and live subscribers.

Parameters:

id (str) –

The opaque run id.
manager (RunManager) –

The owning RunManager, told when this run reaches a terminal state so it can move the run from the live pin into the completed-run TTL cache.
preset_name (str) –

The preset being solved, retained for completion telemetry.

Source code in src/technoeconomics/web/runs.py

def __init__(self, id: str, manager: RunManager, preset_name: str) -> None:
    self.id = id
    self.preset_name = preset_name
    self.started = time.monotonic()
    self._manager = manager
    self.status: Status = "running"
    self.result: RunResults | None = None  # set once done
    self.error: str | None = None
    self._events: list[ServerSentEvent] = []  # progress tail, monotonically id'd
    self._next_event_id = 1
    self._subscribers: set[MemoryObjectSendStream[ServerSentEvent]] = set()
    self._task: asyncio.Task[None] | None = (
        None  # the run coroutine (pinned by manager)
    )

aclose `async` ¶

aclose() -> None

Cancel the run's task and close all subscriber streams (at application shutdown).

Source code in src/technoeconomics/web/runs.py

async def aclose(self) -> None:
    """Cancel the run's task and close all subscriber streams (at application shutdown)."""
    if self._task is not None and not self._task.done():
        self._task.cancel()
        try:
            await self._task
        except asyncio.CancelledError:
            pass
        except Exception:  # noqa: BLE001 -- a failing task must not block teardown
            log.exception("Run %s task failed during close", self.id)
    for send in list(self._subscribers):
        send.close()

cancel ¶

cancel(reason: str) -> None

Terminate the run as superseded, poking any tab still streaming it; idempotent.

The worker thread and its CapacityLimiter slot are unrecoverable (threads can't be killed), so this frees only the run's accounting -- releasing the client's single-flight slot so a newer solve is not locked out by an abandoned one.

Source code in src/technoeconomics/web/runs.py

def cancel(self, reason: str) -> None:
    """Terminate the run as superseded, poking any tab still streaming it; idempotent.

    The worker thread and its `CapacityLimiter` slot are unrecoverable (threads can't be
    killed), so this frees only the run's *accounting* -- releasing the client's
    single-flight slot so a newer solve is not locked out by an abandoned one.
    """
    self._finish("cancelled", error=reason)

events `async` ¶

events(last_id: int | None) -> AsyncIterator[ServerSentEvent]

Yield this run's events for one SSE connection: replay the tail, then live, then close.

Registering the subscriber and snapshotting the tail/status happen in one synchronous block (no await between), so -- since emission runs on the same single-threaded loop -- an event cannot slip in unseen or be delivered twice. On reconnect the browser's Last-Event-ID becomes last_id and only newer tail events replay. The stream ends after the terminal event: either it was in the replayed tail (run already finished) or it arrives live.

Parameters:

last_id (int | None) –

The last event id the client already has (from Last-Event-ID), or None.

Yields:

AsyncIterator[ServerSentEvent] –

Each ServerSentEvent, replayed then live, up to and including the terminal one.

Source code in src/technoeconomics/web/runs.py

async def events(self, last_id: int | None) -> AsyncIterator[ServerSentEvent]:
    """Yield this run's events for one SSE connection: replay the tail, then live, then close.

    Registering the subscriber and snapshotting the tail/status happen in one synchronous
    block (no ``await`` between), so -- since emission runs on the same single-threaded loop
    -- an event cannot slip in unseen or be delivered twice. On reconnect the browser's
    ``Last-Event-ID`` becomes `last_id` and only newer tail events replay. The stream ends
    after the terminal event: either it was in the replayed tail (run already finished) or it
    arrives live.

    Args:
        last_id: The last event id the client already has (from ``Last-Event-ID``), or None.

    Yields:
        Each `ServerSentEvent`, replayed then live, up to and including the terminal one.
    """
    send, receive = create_memory_object_stream[ServerSentEvent](_BUFFER)
    self._subscribers.add(send)
    tail = list(self._events)
    already_terminal = self.status != "running"
    try:
        for event in tail:
            if (
                last_id is not None
                and event.id is not None
                and int(event.id) <= last_id
            ):
                continue
            yield event
        if already_terminal:
            return
        async for event in receive:
            yield event
            if event.event in _TERMINAL:
                return
    finally:
        self._subscribers.discard(send)
        send.close()

snapshot ¶

snapshot() -> RunSnapshot

The run's current state, as the DTO GET /api/runs/{id} returns.

_finish is the only writer of all three fields and keeps them consistent (a done run has results and no error; a failed or cancelled one the reverse), so this is a straight projection.

Source code in src/technoeconomics/web/runs.py

def snapshot(self) -> RunSnapshot:
    """The run's current state, as the DTO ``GET /api/runs/{id}`` returns.

    `_finish` is the only writer of all three fields and keeps them consistent (a done
    run has results and no error; a failed or cancelled one the reverse), so this is a
    straight projection.
    """
    return RunSnapshot(status=self.status, error=self.error, results=self.result)

RunManager ¶

RunManager(*, maxsize: int = 256, ttl: float = 15 * 60, deadline: float = 300, on_complete: Callable[[Run], None] | None = None)

Holds live and recently-completed runs; launches solves and reaps them.

A running run is pinned in _live and never evicted; on completion it moves to _done, a bounded TTLCache whose clock starts at completion. Tasks are pinned in _tasks so a running solve is never garbage-collected out from under the loop.

Parameters:

maxsize (int, default: 256 ) –

Maximum completed runs retained before the least-recently-inserted is evicted.
ttl (float, default: 15 * 60 ) –

Seconds a completed run survives (covers reconnect and reload-restore).
deadline (float, default: 300 ) –

Seconds a solve may run before the watchdog marks it failed.
on_complete (Callable[[Run], None] | None, default: None ) –

Called with each run as it reaches a terminal state (loop-side) -- the telemetry hook. The outcome does not exist at 202-time, so this fires at completion rather than at launch.

Source code in src/technoeconomics/web/runs.py

def __init__(
    self,
    *,
    maxsize: int = 256,
    ttl: float = 15 * 60,
    deadline: float = 300,
    on_complete: Callable[[Run], None] | None = None,
) -> None:
    self._live: dict[str, Run] = {}
    self._done: TTLCache[str, Run] = TTLCache(maxsize=maxsize, ttl=ttl)
    self._tasks: set[asyncio.Task[None]] = set()
    self._deadline = deadline
    self._on_complete = on_complete

live_count `property` ¶

live_count: int

Number of runs currently solving (for the global live-run cap).

aclose_all `async` ¶

aclose_all() -> None

Cancel every live run and dispose it; call once at application shutdown.

Source code in src/technoeconomics/web/runs.py

async def aclose_all(self) -> None:
    """Cancel every live run and dispose it; call once at application shutdown."""
    for run in list(self._live.values()):
        await run.aclose()
    self._live.clear()

get ¶

get(run_id: str) -> Run | None

Look up a run by id, whether it is still live or recently completed.

Source code in src/technoeconomics/web/runs.py

def get(self, run_id: str) -> Run | None:
    """Look up a run by id, whether it is still live or recently completed."""
    return self._live.get(run_id) or self._done.get(run_id)

launch ¶

launch(plant: Plant, preset: Preset) -> Run

Start solving plant in the background and return its Run at once.

Parameters:

plant (Plant) –

The plant to solve (already validated and rebuilt from the preset).
preset (Preset) –

The preset supplying which numbers and plots to compute.

Returns:

Run –

The new run, pinned live until it reaches a terminal state.

Source code in src/technoeconomics/web/runs.py

def launch(self, plant: Plant, preset: Preset) -> Run:
    """Start solving `plant` in the background and return its `Run` at once.

    Args:
        plant: The plant to solve (already validated and rebuilt from the preset).
        preset: The preset supplying which numbers and plots to compute.

    Returns:
        The new run, pinned live until it reaches a terminal state.
    """
    run = Run(uuid4().hex, self, preset.name)
    self._live[run.id] = run
    task = asyncio.create_task(run._run(plant, preset, self._deadline))
    run._task = task
    self._tasks.add(task)
    task.add_done_callback(self._tasks.discard)
    return run

retire ¶

retire(run: Run) -> None

Move a just-terminal run from the live pin into the completed-run TTL cache (loop-side).

Fires the completion telemetry hook once, after the run is stored so a hook that reads the run sees its final state.

Source code in src/technoeconomics/web/runs.py

def retire(self, run: Run) -> None:
    """Move a just-terminal run from the live pin into the completed-run TTL cache (loop-side).

    Fires the completion telemetry hook once, after the run is stored so a hook that reads
    the run sees its final state.
    """
    if self._live.pop(run.id, None) is not None:
        self._done[run.id] = run
        if self._on_complete is not None:
            try:
                self._on_complete(run)
            except Exception:  # noqa: BLE001 -- telemetry must never break a solve
                log.exception("Run %s completion hook failed", run.id)

schemas ¶

Response DTOs for the JSON API.

Plain dataclasses declared as the endpoints' return types so FastAPI emits a precise OpenAPI schema -- the single source the frontend's TypeScript types are generated from (openapi-typescript). Request bodies reuse Envelope; the form spec reuses the ComponentSpec dataclasses, so those shapes are declared once and flow through to the client unchanged.

Status `module-attribute` ¶

Status = Literal['running', 'done', 'failed', 'cancelled']

A run's lifecycle state. Declared here, with the DTO that carries it to the client.

FieldError `dataclass` ¶

FieldError(path: str, message: str)

One validation error, addressed to a spec path (or a component id).

PresetDetail `dataclass` ¶

PresetDetail(title: str, description: str, schematic_svg: str, plant: dict[str, Any], form: list[ComponentSpec])

A preset's presentation, default plant, and form spec (GET /api/presets/{name}).

PresetSummary `dataclass` ¶

PresetSummary(name: str, title: str, description: str)

A preset in the picker list.

RunAccepted `dataclass` ¶

RunAccepted(run_id: str)

The 202 body of a launched solve.

RunResults `dataclass` ¶

RunResults(numbers: list[Any], charts: list[Any])

A solved run's readout: curated headline numbers and chart configs (opaque to the API).

RunSnapshot `dataclass` ¶

RunSnapshot(status: Status, error: str | None = None, results: RunResults | None = None)

A run's current state (GET /api/runs/{id}): status plus error or results when terminal.

ShareToken `dataclass` ¶

ShareToken(token: str)

The body of POST /api/share.

SolveErrors `dataclass` ¶

SolveErrors(errors: list[FieldError])

The 422 body of a rejected solve: the field-addressed validation errors.

share ¶

Encode/decode an Envelope as a URL-safe token.

A token is base64url(gzip(json(envelope))) -- enough to reconstruct a shared model from a link with no server-side storage. The envelope is the same {preset, overlay, enabled} the client edits and solves, so loading a share is just an overlay over the named preset's default.

decode is deliberately defensive: a ?p= value is untrusted input, so every malformed, stale, or oversized token raises ValueError rather than crashing the page or expanding without bound (a gzip bomb). TypeAdapter(Envelope) validates the envelope's shape; that the overlay paths exist and its values are in bounds is enforced at solve against the preset's spec.

decode ¶

decode(token: str) -> Envelope

Reconstruct an envelope from a token.

Parameters:

token (str) –

A token produced by encode.

Returns:

Envelope –

The reconstructed envelope (shape-validated, not solve-validated).

Raises:

ValueError –

If the token is malformed, oversized, or not a well-formed envelope.

Source code in src/technoeconomics/web/share.py

def decode(token: str) -> Envelope:
    """Reconstruct an envelope from a token.

    Args:
        token: A token produced by [`encode`][technoeconomics.web.share.encode].

    Returns:
        The reconstructed envelope (shape-validated, not solve-validated).

    Raises:
        ValueError: If the token is malformed, oversized, or not a well-formed envelope.
    """
    if len(token) > _MAX_TOKEN:
        raise ValueError("share token too large")
    try:
        compressed = base64.urlsafe_b64decode(token)
    except (binascii.Error, ValueError) as e:
        raise ValueError("invalid share token") from e
    raw = _gunzip(compressed, _MAX_JSON)
    try:
        return _envelope.validate_json(raw)
    except ValidationError as e:
        raise ValueError("invalid share payload") from e

encode ¶

encode(envelope: Envelope) -> str

Encode an envelope as a URL-safe token.

Parameters:

envelope (Envelope) –

The envelope to encode.

Returns:

str –

A base64url(gzip(json(...))) token suitable for a ?p= query value.

Source code in src/technoeconomics/web/share.py

def encode(envelope: Envelope) -> str:
    """Encode an envelope as a URL-safe token.

    Args:
        envelope: The envelope to encode.

    Returns:
        A ``base64url(gzip(json(...)))`` token suitable for a ``?p=`` query value.
    """
    raw = _envelope.dump_json(envelope)
    return base64.urlsafe_b64encode(gzip.compress(raw)).decode("ascii")

spec ¶

The form spec and the submission boundary: describe editable params, apply user edits.

Two directions across the same spec:

plant_to_spec turns a plant (the preset default) into a value-free description of its editable parameters. The client renders the form from this spec and binds each input to its path in the plant dict -- the spec says what is editable and how (label, unit, bounds), the values travel separately.
validate_edits / apply_edits are the overlay-solve boundary. The client never posts a plant; it posts {preset, overlay: {path: value}, enabled: {id: bool}}. The server rebuilds from the trusted preset default and applies only whitelisted edits: overlay values at spec paths (any non-spec path is rejected -- structure is never client-controlled) and enable toggles on non-fixed components.

The spec generator reads the parameter vocabulary from technoeconomics.model.params off each component's field annotations, recursing into dataset parameters so a Sinusoidal price exposes price.mean, price.amplitude, ... as separate numeric inputs. It is total-or-error: every author-added field must be projectable (a numeric scalar, a timeseries, or a dataset recursed into) or explicitly marked Hidden -- anything else raises ProjectionError naming the field, rather than being silently dropped.

Per-use unit and bound propagation: the unit and (scalar-branch) bounds declared on a component's dataset field propagate to the dataset leaves marked Magnitude -- so one Sinusoidal class serves as a price (EUR/MWh) in one field and a demand (MW) in another. Intrinsic dataset leaves (period, phase) keep their own class-level unit and bounds. A propagated bound binds each leaf it lands on, not the value the dataset ultimately computes (see technoeconomics.model.params).

Division of labour on the submission path, so no rule is checked twice:

pydantic, at the request boundary. Envelope declares overlay: dict[str, float] and enabled: dict[str, bool], so by the time validate_edits runs the value types are already guaranteed. It does not re-check them.
validate_edits, against the spec. The path whitelist (which is what keeps structure server-controlled), the numeric bounds -- propagated ones especially, which sit outside pydantic's reach entirely -- finiteness, and the fixed-component rule.
the Plant validator, on the rebuilt plant. Class-level types and bounds, as a backstop. It is unreachable by construction once validate_edits passes, so apply_edits does not soften anything for it: a spec path that fails to resolve raises rather than being silently dropped, because that can only mean the spec and the serialised plant have drifted apart.

Bound `dataclass` ¶

Bound(value: float, exclusive: bool)

One numeric bound: its value and whether it is exclusive (Gt/Lt).

Attributes:

value (float) –

The bounding number.
exclusive (bool) –

True for a strict bound (Gt/Lt), False for inclusive (Ge/Le).

ComponentSpec `dataclass` ¶

ComponentSpec(id: str, title: str, color: str | None, enabled: bool, fixed: bool, params: list[FieldSpec])

A component rendered as a fieldset: identity, an optional enable toggle, its params.

Attributes:

id (str) –

The component's id (the first segment of its params' paths).
title (str) –

Human-readable heading derived from the id.
color (str | None) –

The component's plot colour (a hex string), or None.
enabled (bool) –

Whether the component is currently enabled.
fixed (bool) –

If True the component may not be disabled -- the client shows no enable toggle and a submission cannot switch it off.
params (list[FieldSpec]) –

The editable numeric leaves, in declaration order.

FieldSpec `dataclass` ¶

FieldSpec(path: str, kind: Literal['number', 'series'], label: str, unit: str | None, min: Bound | None, max: Bound | None, step: float | None, advanced: bool)

One leaf of a component (possibly nested inside a dataset) surfaced in the form.

Value-free: the current value lives in the plant dict, reached by path.

Attributes:

path (str) –

Id-rooted dotted path into the plant dict, e.g. "grid.price.mean".
kind (Literal['number', 'series']) –

The input kind. "number" is an editable numeric input; "series" is a timeseries, shown read-only -- the whitelist in validate_edits admits an overlay value only for a "number", so a series is surfaced but not yet writable. Further kinds are additive.
label (str) –

Display name -- from Label when the author gave one, else derived from the path ("price.mean").
unit (str | None) –

Display/edit unit, e.g. "EUR/MWh", or None if dimensionless. For a "series" this is the unit of each point.
min (Bound | None) –

Lower bound, or None if unbounded below.
max (Bound | None) –

Upper bound, or None if unbounded above.
step (float | None) –

Suggested input step, or None. (No source in v1; reserved for a future marker.)
advanced (bool) –

Render under the form's "Advanced" section rather than inline.

ProjectionError ¶

Bases: TypeError

An author-added field is neither projectable (scalar/dataset) nor marked Hidden.

apply_edits ¶

apply_edits(plant: Plant, overlay: Mapping[str, float], enabled: Mapping[str, bool]) -> Plant

Rebuild the preset default plant with a validated submission's edits applied.

Overlays each value at its spec path and sets each component's enabled flag, operating on the serialised dict (datasets are frozen), then reparses through Plant.from_dict as a class-level backstop.

Assumes the submission passed validate_edits, and does not soften anything on that assumption: a path or component id that fails to resolve raises, because after validation it can only mean the spec and the serialised plant have drifted apart.

Parameters:

plant (Plant) –

The trusted preset default plant.
overlay (Mapping[str, float]) –

Validated {path: value} edits.
enabled (Mapping[str, bool]) –

Validated {component_id: bool} toggles.

Returns:

Plant –

A new plant with the edits applied.

Raises:

KeyError –

If a validated path or component id does not resolve (a spec/plant drift bug, not a bad submission).

Source code in src/technoeconomics/web/spec.py

def apply_edits(
    plant: Plant, overlay: Mapping[str, float], enabled: Mapping[str, bool]
) -> Plant:
    """Rebuild the preset default `plant` with a validated submission's edits applied.

    Overlays each value at its spec path and sets each component's ``enabled`` flag, operating
    on the serialised dict (datasets are frozen), then reparses through `Plant.from_dict` as a
    class-level backstop.

    Assumes the submission passed
    [`validate_edits`][technoeconomics.web.spec.validate_edits], and does not soften anything
    on that assumption: a path or component id that fails to resolve raises, because after
    validation it can only mean the spec and the serialised plant have drifted apart.

    Args:
        plant: The trusted preset default plant.
        overlay: Validated ``{path: value}`` edits.
        enabled: Validated ``{component_id: bool}`` toggles.

    Returns:
        A new plant with the edits applied.

    Raises:
        KeyError: If a validated path or component id does not resolve (a spec/plant drift
            bug, not a bad submission).
    """
    d = plant.to_dict()
    for path, value in overlay.items():
        _set(d, path, float(value))
    for cid, want in enabled.items():
        component = _component(d, cid)
        if component is None:
            raise KeyError(f"{cid}: spec component is absent from the serialised plant")
        component["enabled"] = want
    return Plant.from_dict(d)

plant_to_spec ¶

plant_to_spec(plant: Plant) -> list[ComponentSpec]

Describe a plant's editable parameters, one ComponentSpec per component.

Propagates ProjectionError from _leaves when an author-added field is neither projectable nor Hidden -- the totality rule.

Parameters:

plant (Plant) –

The plant supplying the structure (which datasets are present, which components are fixed).

Returns:

list[ComponentSpec] –

The form spec: one entry per component, each carrying its editable numeric leaves.

Source code in src/technoeconomics/web/spec.py

def plant_to_spec(plant: Plant) -> list[ComponentSpec]:
    """Describe a plant's editable parameters, one `ComponentSpec` per component.

    Propagates [`ProjectionError`][technoeconomics.web.spec.ProjectionError] from `_leaves`
    when an author-added field is neither projectable nor ``Hidden`` -- the totality rule.

    Args:
        plant: The plant supplying the structure (which datasets are present, which
            components are fixed).

    Returns:
        The form spec: one entry per component, each carrying its editable numeric leaves.
    """
    specs: list[ComponentSpec] = []
    for component in plant.components:
        hints = typing.get_type_hints(type(component), include_extras=True)
        params: list[FieldSpec] = []
        for f in fields(component):
            if f.name in _FRAMEWORK_FIELDS:
                continue
            markers = _markers(hints[f.name])
            if markers.hidden:
                continue
            params.extend(
                _leaves(f"{component.id}.{f.name}", getattr(component, f.name), markers)
            )
        specs.append(
            ComponentSpec(
                id=component.id,
                title=component.id.replace("_", " ").capitalize(),
                color=str(component.plot_color) if component.plot_color else None,
                enabled=component.enabled,
                fixed=component.fixed,
                params=params,
            )
        )
    return specs

validate_edits ¶

validate_edits(spec: Iterable[ComponentSpec], overlay: Mapping[str, float], enabled: Mapping[str, bool]) -> list[dict[str, str]]

Validate a submission's edits against spec; return all path-addressed errors.

An overlay entry must name an editable (kind="number") spec path and carry a finite, in-bounds value; an enabled entry must name a component and may switch off only a non-fixed one. Rejecting non-spec overlay paths is what keeps structure server-controlled: the client can edit values and toggle optional components, nothing else.

The value types are guaranteed by Envelope at the request boundary, so they are not re-checked here.

Parameters:

spec (Iterable[ComponentSpec]) –

The named preset's form spec.
overlay (Mapping[str, float]) –

Submitted {path: value} edits (shape-validated, meaning untrusted).
enabled (Mapping[str, bool]) –

Submitted {component_id: bool} toggles (likewise).

Returns:

list[dict[str, str]] –

One {"path", "message"} entry per problem, empty if the submission is clean.

Source code in src/technoeconomics/web/spec.py

def validate_edits(
    spec: Iterable[ComponentSpec],
    overlay: Mapping[str, float],
    enabled: Mapping[str, bool],
) -> list[dict[str, str]]:
    """Validate a submission's edits against `spec`; return all path-addressed errors.

    An overlay entry must name an *editable* (``kind="number"``) spec path and carry a finite,
    in-bounds value; an enabled entry must name a component and may switch off only a non-fixed
    one. Rejecting non-spec overlay paths is what keeps structure server-controlled: the client
    can edit values and toggle optional components, nothing else.

    The value *types* are guaranteed by [`Envelope`][technoeconomics.web.envelope.Envelope] at
    the request boundary, so they are not re-checked here.

    Args:
        spec: The named preset's form spec.
        overlay: Submitted ``{path: value}`` edits (shape-validated, meaning untrusted).
        enabled: Submitted ``{component_id: bool}`` toggles (likewise).

    Returns:
        One ``{"path", "message"}`` entry per problem, empty if the submission is clean.
    """
    spec = list(spec)
    fields_by_path = {f.path: f for c in spec for f in c.params}
    components_by_id = {c.id: c for c in spec}
    errors: list[dict[str, str]] = []
    for path, value in overlay.items():
        field = fields_by_path.get(path)
        if field is None or field.kind != "number":
            errors.append({"path": path, "message": "not an editable parameter"})
        elif not _is_number(value):
            errors.append({"path": path, "message": "must be a finite number"})
        else:
            errors.extend(_bound_errors(field, float(value)))
    for cid, want in enabled.items():
        component = components_by_id.get(cid)
        if component is None:
            errors.append({"path": cid, "message": "unknown component"})
        elif component.fixed and not want:
            errors.append({"path": cid, "message": "component cannot be disabled"})
    return errors

API reference¶

technoeconomics ¶

backend ¶

preset ¶

IndustrialHeat ¶

build ¶

Preset ¶

build abstractmethod ¶

schematic_svg ¶

get ¶

presets ¶

base ¶

Preset ¶

industrial_heat ¶

IndustrialHeat ¶

results ¶

Number ¶

Plot ¶

numbers ¶

plots ¶

solve ¶

SolveError ¶

solve ¶

data ¶

Constant dataclass ¶

compute ¶

Dataset dataclass ¶

compute abstractmethod ¶

resolve ¶

Sinusoidal dataclass ¶

compute ¶

resolve_datasets ¶

base ¶

Dataset dataclass ¶

compute abstractmethod ¶

resolve ¶

ResultCache ¶

get_or_make ¶

resolve_datasets ¶

sources ¶

Constant dataclass ¶

compute ¶

Sinusoidal dataclass ¶

compute ¶

model ¶

component ¶

AnyComponent ¶

Battery dataclass ¶

add_to_network ¶

Component dataclass ¶

add_to_network abstractmethod ¶

ElectricBoiler dataclass ¶

add_to_network ¶

GridElectricity dataclass ¶

add_to_network ¶

HeatDemand dataclass ¶

add_to_network ¶

HeatPump dataclass ¶

add_to_network ¶

PlotColor ¶

params ¶

Advanced module-attribute ¶

Hidden module-attribute ¶

Magnitude module-attribute ¶

Param module-attribute ¶

Label dataclass ¶

Unit dataclass ¶

plant ¶

Plant dataclass ¶

build_network ¶

from_dict classmethod ¶

to_dict ¶

annual_snapshots ¶

structure ¶

Bus dataclass ¶

progress ¶

emit ¶

sink ¶

serialise ¶

Snapshots ¶

build `abstractmethod` ¶

Constant `dataclass` ¶

Dataset `dataclass` ¶

compute `abstractmethod` ¶

Sinusoidal `dataclass` ¶

Dataset `dataclass` ¶

compute `abstractmethod` ¶

Constant `dataclass` ¶

Sinusoidal `dataclass` ¶

Battery `dataclass` ¶

Component `dataclass` ¶

add_to_network `abstractmethod` ¶

ElectricBoiler `dataclass` ¶

GridElectricity `dataclass` ¶

HeatDemand `dataclass` ¶

HeatPump `dataclass` ¶

Advanced `module-attribute` ¶

Hidden `module-attribute` ¶

Magnitude `module-attribute` ¶

Param `module-attribute` ¶

Label `dataclass` ¶

Unit `dataclass` ¶

Plant `dataclass` ¶

from_dict `classmethod` ¶

Bus `dataclass` ¶

type_adapter `cached` ¶

run_events `async` ¶

solve `async` ¶

Envelope `dataclass` ¶

lifespan `async` ¶

aclose `async` ¶

events `async` ¶

live_count `property` ¶

aclose_all `async` ¶

Status `module-attribute` ¶

FieldError `dataclass` ¶

PresetDetail `dataclass` ¶

PresetSummary `dataclass` ¶

RunAccepted `dataclass` ¶

RunResults `dataclass` ¶

RunSnapshot `dataclass` ¶

ShareToken `dataclass` ¶

SolveErrors `dataclass` ¶

Bound `dataclass` ¶

ComponentSpec `dataclass` ¶

FieldSpec `dataclass` ¶