bw_temporalis package#

Subpackages#

bw_temporalis.lcia package

Submodules#

bw_temporalis.convolution module#

bw_temporalis.convolution.consolidate(*, indices, amounts)#

Sum all values in amount which have the same index in indices

Parameters:

indices (ndarray[Any, dtype[int64]]) –
amounts (ndarray[Any, dtype[float64]]) –

Return type:

tuple[numpy.ndarray[Any, numpy.dtype[numpy.int64]], numpy.ndarray[Any, numpy.dtype[numpy.float64]]]

bw_temporalis.convolution.convolve(*, first_date, first_amount, second_date, second_amount, return_dtype)#

Parameters:

first_date (ndarray[Any, dtype[ScalarType]]) –
first_amount (ndarray[Any, dtype[float64]]) –
second_date (ndarray[Any, dtype[ScalarType]]) –
second_amount (ndarray[Any, dtype[float64]]) –
return_dtype (dtype[Any] | None | Type[Any] | _SupportsDType[dtype[Any]] | str | Tuple[Any, int] | Tuple[Any, SupportsIndex | Sequence[SupportsIndex]] | List[Any] | _DTypeDict | Tuple[Any, Any]) –

Return type:

tuple[numpy.ndarray[Any, numpy.dtype[+ScalarType]], numpy.ndarray[Any, numpy.dtype[numpy.float64]]]

bw_temporalis.convolution.temporal_convolution_datetime_timedelta(*, first_date, first_amount, second_date, second_amount)#

Parameters:

first_date (ndarray[Any, dtype[dtype('<M8[s]')]]) –
first_amount (ndarray[Any, dtype[float64]]) –
second_date (ndarray[Any, dtype[dtype('<m8[s]')]]) –
second_amount (ndarray[Any, dtype[float64]]) –

Return type:

tuple[numpy.ndarray[Any, numpy.dtype[dtype(‘<M8[s]’)]], numpy.ndarray[Any, numpy.dtype[numpy.float64]]]

bw_temporalis.convolution.temporal_convolution_timedelta_timedelta(*, first_date, first_amount, second_date, second_amount)#

Parameters:

first_date (ndarray[Any, dtype[dtype('<m8[s]')]]) –
first_amount (ndarray[Any, dtype[float64]]) –
second_date (ndarray[Any, dtype[dtype('<m8[s]')]]) –
second_amount (ndarray[Any, dtype[float64]]) –

Return type:

tuple[numpy.ndarray[Any, numpy.dtype[dtype(‘<m8[s]’)]], numpy.ndarray[Any, numpy.dtype[numpy.float64]]]

bw_temporalis.lca module#

exception bw_temporalis.lca.MultipleTechnosphereExchanges#: Bases: Exception

class bw_temporalis.lca.TemporalisLCA(lca_object, starting_datetime='now', cutoff=0.0005, biosphere_cutoff=1e-06, max_calc=2000.0, static_activity_indices={}, skip_coproducts=False, functional_unit_unique_id=-1, graph_traversal=<class 'bw_graph_tools.graph_traversal.NewNodeEachVisitGraphTraversal'>)#

Bases: object

Calculate an LCA using graph traversal, with edges using temporal distributions.

Edges with temporal distributions should store this information using “temporal_distributions”:

```python

exchange[“temporal_distribution”] = bw_temporalis.TemporalDistribution(: times=numpy.array([-2, -1, 0, 1, 2], dtype=”timedelta64[s]”), values=numpy.ones(5)

)

```

Temporal distribution times must always have the data type timedelta64[s]. Not all edges need to have temporal distributions.

Temporal distributions are not density functions - their values should sum to the exchange amount.

As graph traversal is much slower than matrix calculations, we can limit which nodes get traversed in several ways:

All activities in a database marked as static
Any activity ids passed in static_activity_indices
Any activities whose cumulative impact is below the cutoff score

The output of a Temporalis LCA calculation is a bw_temporalis.Timeline, which can be characterized.

Parameters:

lca_object (bw2calc.LCA) – The already instantiated and calculated LCA class (i.e. .lci() and .lcia() have already been done)
starting_datetime (datetime.datetime | str) – When the functional unit happens. Must be a point in time. Normally something like “now” or “2023-01-01”.
cutoff (float) – The fraction of the total score below which graph traversal should stop. In range (0, 1).
biosphere_cutoff (float) – The fraction of the total score below which we don’t include separate biosphere nodes to be characterized in the Timeline. In range (0, 1).
max_calc (int) – Total number of LCA inventory calculations to perform during graph traversal
static_activity_indices (set[int]) – Activity id values where graph traversal will stop
skip_coproducts (bool) – Should we also traverse edges for the other products in multioutput activities?
functional_unit_unique_id (int) – The unique id of the functional unit. Strongly recommended to leave as default.
graph_traversal (bw_graph_tools.NewNodeEachVisitGraphTraversal) – Optional subclass of NewNodeEachVisitGraphTraversal for advanced usage

build_timeline()#

Return type:: Timeline

get_biosphere_exchanges(flow_id, activity_id)#

Parameters:

flow_id (int) –
activity_id (int) –

Return type:

Iterable[ExchangeDataset]

get_technosphere_exchange(input_id, output_id)#

Parameters:

input_id (int) –
output_id (int) –

Return type:

ExchangeDataset

bw_temporalis.temporal_distribution module#

class bw_temporalis.temporal_distribution.TemporalDistribution(date, amount)#

Bases: object

A container for a series of amount spread over time. :param * date: 1D array containg temporal info of amount with type timedelta64 or datetime64 . :type * date: ndarray :param * amount: 1D array containg amount with type float :type * amount: ndarray :param Times and amount must have same length and element of amount must correspond to the element of date: :param with the same index.:

Parameters:

date (ndarray[Any, dtype[datetime64 | timedelta64]]) –
amount (ndarray[Any, dtype[ScalarType]]) –

graph(style='fivethirtyeight')#

Graph the temporal distribution.

This isn’t too difficult, if you need more customization write your own :)

Parameters:: style (str | None) –

nonzero()#

simplify(threshhold=100, num_clusters=None, iterations=30)#

Use clustering to simplify a TemporalDistribution with more than threshhold number of points.

Uses the kmeans2 implementation of KNN from [scipy.cluster.vq](https://docs.scipy.org/doc/scipy/reference/generated/scipy.cluster.vq.kmeans2.html). Subclass and override this method to get a custom clustering algorithm.

This isn’t perfect, kmeans2 produces “lumpy” distributions - to see this graph the simplification of a uniform distribution with many points.

Parameters:

threshhold (int, optional) – The number of date points above which simplification is triggered
iterations (int, optional) – iter parameters to feed to kmeans2
num_clusters (int | None) –

Return type:

Either self (if no simplification) or a new instance of TemporalDistribution.

property total: float#

bw_temporalis.timeline module#

exception bw_temporalis.timeline.EmptyTimeline#

Bases: Exception

Operation on empty timeline

class bw_temporalis.timeline.FlowTD(distribution, flow, activity)#

Bases: object

Class for storing a temporal distribution associated with a flow and activity.

Parameters:

distribution (TemporalDistribution) –
flow (int) –
activity (int) –

distribution#

Type:: TemporalDistribution

flow#

Type:: int

activity#

Type:: int

bw_temporalis.utils module#

exception bw_temporalis.utils.IncongruentDistribution#

Bases: Exception

The sum of TemporalDistribution values is different than the exchange

bw_temporalis.utils.check_database_exchanges(database_label)#

Check the sum of an exchange TemporalDistribution.amount is close to its amount value.

Raises:: IncongruentDistribution – If the two values are more than 1 percent different
Parameters:: database_label (str) – Name of database to check
Return type:: None

bw_temporalis.utils.easy_datetime_distribution(start, end, steps=50, kind='uniform', param=None)#

Generate a datetime TemporalDistribution with a few input parameters.

Can generate distributions whose amount values are uniformly, triangularly, or normally distributed. Please build more complicated distributions manually.

Only the amount values are distributed, the resulting distribution date values are uniformly spaced from start to end.

For triangular distributions, param is the mode (optional), and should be given in the same reference system as start and stop. The param value should be in the same format as start and end, e.g. “2023-01-01”.

For lognormal distributions, param is the standard deviation in relation to a standardized distribution with mu = 0. param is not used for the uniform distribution.

Raises:

ValueError – If the input parameters prevent construction of valide TemporalDistribution.

Parameters:

start (str) – Datetime marking the start (inclusive) of the distribution, e.g. “now”, “2023-02-01”, “2023-03-02T12:34:56”
end (str) – Datetime marking the end (inclusive) of the distribution, e.g. “now”, “2023-02-01”, “2023-03-02T12:34:56”
steps (int, optional) – Number of values in discrete distribution. Normally not more than 50 or 100.
kind (str, optional) – Distribution type. Must be one of “uniform”, “triangular”, or “normal”
param (float, optional) – Input parameter to define triangular or normal distribution

Return type:

A TemporalDistribution instance.

bw_temporalis.utils.easy_timedelta_distribution(start, end, resolution, steps=50, kind='uniform', param=None)#

Generate a timedelta TemporalDistribution with a few input parameters.

Can generate distributions whose amount values are uniformly, triangularly, or normally distributed. Please build more complicated distributions manually.

Only the amount values are distributed, the resulting distribution date values are uniformly spaced from start to end.

For triangular distributions, param is the mode (optional). For lognormal distributions, param is the standard deviation (required). param is not used for the uniform distribution.

Raises:

ValueError – If the input parameters prevent construction of valide TemporalDistribution.

Parameters:

start (int) – Start (inclusive) of the distribution in resolution units
end (int) – End (inclusive) of the distribution in resolution units
resolution (str) – Resolution of the created timedelta64 array. One of Y (year), M (month), D (day), h (hour), m (minute), s (second)
steps (int, optional) – Number of values in discrete distribution. Normally not more than 50 or 100.
kind (str, optional) – Distribution type. Must be one of “uniform”, “triangular”, or “normal”
param (float, optional) – Input parameter to define triangular or normal distribution

Return type:

A TemporalDistribution instance.

bw_temporalis.utils.get_version_tuple()#

Return type:: tuple

bw_temporalis.utils.normalized_data_array(steps, kind, param)#

Parameters:

steps (int) –
kind (str) –
param (float | None) –

Return type:

ndarray[Any, dtype[int]]

bw_temporalis.utils.supplement_dataframe(df, database_label, columns=['name', 'location', 'unit', 'categories'])#

Parameters:

df (DataFrame) –
database_label (str) –
columns (list[str] | None) –

Return type:

DataFrame

Module contents#

exception bw_temporalis.IncongruentDistribution#

Bases: Exception

The sum of TemporalDistribution values is different than the exchange

class bw_temporalis.TemporalDistribution(date, amount)#

Bases: object

Parameters:

date (ndarray[Any, dtype[datetime64 | timedelta64]]) –
amount (ndarray[Any, dtype[ScalarType]]) –

graph(style='fivethirtyeight')#

Graph the temporal distribution.

This isn’t too difficult, if you need more customization write your own :)

Parameters:: style (str | None) –

nonzero()#

simplify(threshhold=100, num_clusters=None, iterations=30)#

Use clustering to simplify a TemporalDistribution with more than threshhold number of points.

This isn’t perfect, kmeans2 produces “lumpy” distributions - to see this graph the simplification of a uniform distribution with many points.

Parameters:

threshhold (int, optional) – The number of date points above which simplification is triggered
iterations (int, optional) – iter parameters to feed to kmeans2
num_clusters (int | None) –

Return type:

Either self (if no simplification) or a new instance of TemporalDistribution.

property total: float#

class bw_temporalis.TemporalisLCA(lca_object, starting_datetime='now', cutoff=0.0005, biosphere_cutoff=1e-06, max_calc=2000.0, static_activity_indices={}, skip_coproducts=False, functional_unit_unique_id=-1, graph_traversal=<class 'bw_graph_tools.graph_traversal.NewNodeEachVisitGraphTraversal'>)#

Bases: object

Calculate an LCA using graph traversal, with edges using temporal distributions.

Edges with temporal distributions should store this information using “temporal_distributions”:

```python

exchange[“temporal_distribution”] = bw_temporalis.TemporalDistribution(: times=numpy.array([-2, -1, 0, 1, 2], dtype=”timedelta64[s]”), values=numpy.ones(5)

)

```

Temporal distribution times must always have the data type timedelta64[s]. Not all edges need to have temporal distributions.

Temporal distributions are not density functions - their values should sum to the exchange amount.

As graph traversal is much slower than matrix calculations, we can limit which nodes get traversed in several ways:

All activities in a database marked as static
Any activity ids passed in static_activity_indices
Any activities whose cumulative impact is below the cutoff score

The output of a Temporalis LCA calculation is a bw_temporalis.Timeline, which can be characterized.

Parameters:

lca_object (bw2calc.LCA) – The already instantiated and calculated LCA class (i.e. .lci() and .lcia() have already been done)
starting_datetime (datetime.datetime | str) – When the functional unit happens. Must be a point in time. Normally something like “now” or “2023-01-01”.
cutoff (float) – The fraction of the total score below which graph traversal should stop. In range (0, 1).
biosphere_cutoff (float) – The fraction of the total score below which we don’t include separate biosphere nodes to be characterized in the Timeline. In range (0, 1).
max_calc (int) – Total number of LCA inventory calculations to perform during graph traversal
static_activity_indices (set[int]) – Activity id values where graph traversal will stop
skip_coproducts (bool) – Should we also traverse edges for the other products in multioutput activities?
functional_unit_unique_id (int) – The unique id of the functional unit. Strongly recommended to leave as default.
graph_traversal (bw_graph_tools.NewNodeEachVisitGraphTraversal) – Optional subclass of NewNodeEachVisitGraphTraversal for advanced usage

build_timeline()#

Return type:: Timeline

get_biosphere_exchanges(flow_id, activity_id)#

Parameters:

flow_id (int) –
activity_id (int) –

Return type:

Iterable[ExchangeDataset]

get_technosphere_exchange(input_id, output_id)#

Parameters:

input_id (int) –
output_id (int) –

Return type:

ExchangeDataset

class bw_temporalis.Timeline(data=None)#

Bases: object

Sum and group elements over time. Timeline calculations produce a list of [(datetime, amount)] tuples.

Parameters:: data (list[bw_temporalis.timeline.FlowTD] | None) –

self.data#

Type:: list[FlowTD]

add_flow_temporal_distribution(td, flow, activity)#

Append a TemporalDistribution object to the Timeline.data object.

Parameters:

td (TemporalDistribution) – Temporal distribution to add.
flow (int) – Associated flow.
activity (int) – Associated activity.

Return type:

None