Dataset¶

technoeconomics.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)

Dataset¶

technoeconomics.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 ¶

Constant `dataclass` ¶

Dataset `dataclass` ¶

compute `abstractmethod` ¶

Sinusoidal `dataclass` ¶

Dataset `dataclass` ¶

compute `abstractmethod` ¶

Constant `dataclass` ¶

Sinusoidal `dataclass` ¶