Data¶

Introduction¶

Sourcing the right data is an important part of technoeconomic analysis. technoeconomics-app provides access to many useful datasets behind a common interface. That interface is simple - a Dataset is something that takes in arbitrary arguments, and returns either a single value, or a timeseries.

Available datasets¶

We provide the following datasets out of the box:

Constant - returns a timeseries with a constant value, useful for testing
Sinusoidal - sinusoidal timeseries, useful for testing

Adding a dataset¶

Creating a new Dataset is very simple. It's simply a Python dataclass with a compute() method that returns either a float or a Pandas Series. As a reference, look at the implementation of the Sinusoidal dataset:

@dataclass(frozen=True)
class Sinusoidal(SeriesDataset):
    """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: Baseline the wave oscillates around.
        amplitude: Peak deviation from ``mean``.
        period: Oscillation period in hours (e.g. ``24`` daily, ``8760`` yearly).
        phase: Horizontal shift in hours.
    """

    mean: float
    amplitude: float
    period: float = 24.0
    phase: float = 0.0

    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)

Four important points to note here:

@dataclass decorator above the dataset class (frozen=True to make it immutable)
A class which inherits from SeriesDataset if it returns a timeseries (pd.Series) or from ScalarDataset if it returns a float
Arguments passed with arg: TYPE on top of the class declaration, as one does for a dataclass
The compute(self, snapshots: pd.DatetimeIndex) function. This is where the heavy lifting happens

compute() can do anything - download data from the Internet, read from disk, calculate things from an equation, run a complex simulation. It accepts a pd.DatetimeIndex, because the final goal is to pass the output of our dataset into a time-varying simulation. This simulation may be hourly over a year, or daily over a week. If you return a timeseries, obviously it has to have as many timesteps as the rest of the model. We enforce this by giving you an index. If you return a timeseries, it should use this index. If you return a single value, you can use the index to e.g. aggregate the data appropriately, but feel free to ignore it if you don't need it.

Some recommendations for writing your own compute():

make it fast if you want to play around with your optimisation by changing values and rerunning the model, this function may get called a lot. If you're downloading data from the Internet, or doing slow computations - implement some sort of cache. If you're calculating things, try to vectorise your operations, or use Numba

Full dataset API¶

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

Data¶

Introduction¶

Available datasets¶

Adding a dataset¶

Full dataset API¶

technoeconomics.data.base ¶

Dataset dataclass ¶

compute abstractmethod ¶

resolve ¶

ResultCache ¶

get_or_make ¶

resolve_datasets ¶

Dataset `dataclass` ¶

compute `abstractmethod` ¶