import functools
import json
from bisect import bisect_left
from datetime import datetime, timedelta
from typing import Callable, List, Optional, Union
import bw2data as bd
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from bw2data.backends import ActivityDataset as AD
from bw2data.backends.proxies import Exchange
from bw2data.backends.schema import ExchangeDataset
from bw2data.errors import MultipleResults, UnknownObject
from bw_temporalis import TemporalDistribution, easy_timedelta_distribution
from IPython.display import Javascript, display
from ipywidgets import (
Button,
Dropdown,
FloatSlider,
HBox,
IntSlider,
IntText,
Label,
Layout,
Output,
Textarea,
ToggleButtons,
VBox,
)
from loguru import logger
[docs]
time_res_mapping_strftime = {
"year": "%Y",
"month": "%Y%m",
"day": "%Y%m%d",
"hour": "%Y%m%d%H",
}
[docs]
def linear_interpolation_weights(
date: datetime, sorted_dates: tuple
) -> dict:
"""Return ``{date: weight}`` linear-interpolation weights for ``date``.
``sorted_dates`` is an ascending tuple of the available (background-database)
dates. The weights are split between the two nearest bracketing dates,
proportional to temporal proximity. If ``date`` falls on or outside the
range of ``sorted_dates``, the single nearest in-range date gets weight 1.
Mirrors ``TimelineBuilder.get_weights_for_interpolation_between_nearest_years``
and is shared by both the timeline builder and the graph-traversal extractors.
"""
if not sorted_dates:
return {}
position = bisect_left(sorted_dates, date)
if position < len(sorted_dates) and sorted_dates[position] == date:
return {sorted_dates[position]: 1}
if position == 0:
return {sorted_dates[0]: 1}
if position == len(sorted_dates):
return {sorted_dates[-1]: 1}
closest_lower = sorted_dates[position - 1]
closest_higher = sorted_dates[position]
weight = int((date - closest_lower).total_seconds()) / int(
(closest_higher - closest_lower).total_seconds()
)
return {closest_lower: round(1 - weight, 3), closest_higher: round(weight, 3)}
[docs]
def nearest_date_weight(date: datetime, sorted_dates: tuple) -> dict:
"""Return ``{date: 1}`` for the single date in ``sorted_dates`` closest to
``date``.
``sorted_dates`` is an ascending tuple of available dates. Mirrors
``TimelineBuilder.find_closest_date`` and is shared by the timeline builder
and the graph-traversal extractors.
"""
if not sorted_dates:
return None
position = bisect_left(sorted_dates, date)
if position == 0:
closest = sorted_dates[0]
elif position == len(sorted_dates):
closest = sorted_dates[-1]
else:
lower_date = sorted_dates[position - 1]
higher_date = sorted_dates[position]
if abs(date - lower_date) <= abs(higher_date - date):
closest = lower_date
else:
closest = higher_date
return {closest: 1}
[docs]
def get_reference_product_production_amount(
node, *, reference_product=None, lca=None
) -> float:
"""Return a node's production amount in a paradigm-agnostic way.
Supports chimaera activities via ``rp_exchange`` and explicit process/product
setups via production exchanges.
"""
if isinstance(node, (int, np.integer)):
node = bd.get_activity(id=int(node))
if hasattr(node, "rp_exchange"):
rp_exc = node.rp_exchange()
if rp_exc is not None:
return rp_exc.amount
productions = list(node.production())
if not productions:
raise ValueError(
f"Could not determine production amount for node `{node}`: no production exchanges found."
)
if reference_product is not None:
ref_id = (
reference_product.id
if hasattr(reference_product, "id")
else int(reference_product)
)
for exc in productions:
if exc.input.id == ref_id:
return exc.amount
raise ValueError(
f"Could not find production exchange from `{node}` to reference product id `{ref_id}`."
)
if len(productions) == 1:
return productions[0].amount
raise ValueError(
f"Node `{node}` has multiple production exchanges; pass `reference_product` to disambiguate."
)
@functools.lru_cache(maxsize=4096)
[docs]
def convert_date_string_to_datetime(temporal_grouping, date_string) -> datetime:
"""
Converts the string of a date to datetime object.
e.g. for `temporal_grouping` = 'month', and `date_string` = '202303', it extracts 2023-03-01
Parameters
----------
temporal_grouping : str
Temporal grouping for the date string. Options are: 'year', 'month', 'day', 'hour'
date_string : str
Date as a string
Returns
-------
datetime
Datetime object of the date string at the chosen temporal resolution.
"""
time_res_dict = {
"year": "%Y",
"month": "%Y%m",
"day": "%Y%m%d",
"hour": "%Y%m%d%H",
}
if temporal_grouping not in time_res_dict.keys():
raise ValueError(
f'temporal grouping: {temporal_grouping} is not a valid option. Please \
choose from: {list(time_res_dict.keys())}, defaulting to "year"',
)
return datetime.strptime(date_string, time_res_dict[temporal_grouping])
[docs]
def round_datetime(date: datetime, resolution: str) -> datetime:
"""
Round a datetime object based on a given resolution
Parameters
----------
date : datetime
datetime object to be rounded
resolution: str
Temporal resolution to round the datetime object to. Options are: 'year', 'month', 'day' and
'hour'.
Returns
-------
datetime
rounded datetime object
"""
if resolution == "year":
mid_year = pd.Timestamp(f"{date.year}-07-01")
return (
pd.Timestamp(f"{date.year+1}-01-01")
if date >= mid_year
else pd.Timestamp(f"{date.year}-01-01")
)
if resolution == "month":
start_of_month = pd.Timestamp(f"{date.year}-{date.month}-01")
next_month = start_of_month + pd.DateOffset(months=1)
mid_month = start_of_month + (next_month - start_of_month) / 2
return next_month if date >= mid_month else start_of_month
if resolution == "day":
start_of_day = datetime(date.year, date.month, date.day)
mid_day = start_of_day + timedelta(hours=12)
return start_of_day + timedelta(days=1) if date >= mid_day else start_of_day
if resolution == "hour":
start_of_hour = datetime(date.year, date.month, date.day, date.hour)
mid_hour = start_of_hour + timedelta(minutes=30)
return start_of_hour + timedelta(hours=1) if date >= mid_hour else start_of_hour
raise ValueError("Resolution must be one of 'year', 'month', 'day', or 'hour'.")
[docs]
def round_datetime_series_to_year(dates: pd.Series) -> pd.Series:
"""
Vectorized equivalent of ``round_datetime(..., resolution="year")`` for a Series.
Dates on/after July 1st round up to January 1st of the next year, otherwise
down to January 1st of the same year. Matches ``round_datetime`` exactly but
avoids the per-row Python ``apply``.
"""
dt = pd.to_datetime(dates)
years = dt.dt.year.to_numpy()
mid_year = pd.to_datetime(
{"year": years, "month": 7, "day": 1}
).to_numpy()
round_up = dt.to_numpy() >= mid_year
rounded_years = years + round_up.astype(int)
return pd.to_datetime(
{"year": rounded_years, "month": 1, "day": 1}
).set_axis(dates.index)
[docs]
def add_flows_to_characterization_functions(
flows: Union[str, List[str]],
func: Callable,
characterization_functions: Optional[dict] = None,
) -> dict:
"""
Add a new flow or a list of flows to the available characterization functions.
Parameters
----------
flows : Union[str, List[str]]
Flow or list of flows to be added to the characterization function dictionary.
func : Callable
Dynamic characterization function for flow.
characterization_functions : dict, optional
Dictionary of flows and their corresponding characterization functions. Default is an empty
dictionary.
Returns
-------
dict
Updated characterization function dictionary with the new flow(s) and function(s).
"""
if characterization_functions is None:
characterization_functions = {}
# Check if the input is a single flow (str) or a list of flows (List[str])
if isinstance(flows, str):
# It's a single flow, add it directly
characterization_functions[flows] = func
elif isinstance(flows, list):
# It's a list of flows, iterate and add each one
for flow in flows:
characterization_functions[flow] = func
return characterization_functions
[docs]
def resolve_temporalized_node_name(code: str) -> str:
"""
Getting the name of a node based on the code only.
Works for non-unique codes if the name is the same across all databases.
Parameters
----------
code: str
Code of the node to resolve.
Returns
-------
str
Name of the node.
"""
qs = AD.select().where(AD.code == code)
names = set([obj.name for obj in qs])
if len(qs) > 1:
if len(names) > 1:
raise ValueError(
"Found multiple names for the given code: {}".format(names)
)
elif not qs:
raise UnknownObject
return names.pop()
[docs]
def plot_characterized_inventory_as_waterfall(
lca_obj,
static_scores=None,
prospective_scores=None,
order_stacked_activities=None,
):
"""
Plot a stacked waterfall chart of characterized inventory data. As comparison,
static and prospective scores can be added. Only works for metric GWP at the moment.
Parameters
----------
lca_obj : TimexLCA
LCA object with characterized inventory data.
static_scores : dict, optional
Dictionary of static scores. Default is None.
prospective_scores : dict, optional
Dictionary of prospective scores. Default is None.
order_stacked_activities : list, optional
List of activities to order the stacked bars in the waterfall plot. Default is None.
Returns
-------
None
plots the waterfall chart.
"""
if not hasattr(lca_obj, "characterized_inventory"):
raise ValueError("LCA object does not have characterized inventory data.")
if not hasattr(lca_obj, "activity_time_mapping"):
raise ValueError("Make sure to pass an instance of a TimexLCA.")
time_res_dict = {
"year": "%Y",
"month": "%Y-%m",
"day": "%Y-%m-%d",
"hour": "%Y-%m-%d %H",
}
plot_data = lca_obj.characterized_inventory.copy()
plot_data["year"] = plot_data["date"].dt.strftime(
time_res_dict[lca_obj.temporal_grouping]
) # TODO make temporal resolution flexible
# Optimized activity label fetching using the TimexLCA's built-in method
unique_activities = plot_data["activity"].unique()
activity_labels = {
idx: lca_obj.get_activity_name_from_time_mapped_id(idx)
for idx in unique_activities
}
plot_data["activity_label"] = plot_data["activity"].map(activity_labels)
plot_data = plot_data.groupby(["year", "activity_label"], as_index=False)[
"amount"
].sum()
pivoted_data = plot_data.pivot(
index="year", columns="activity_label", values="amount"
)
combined_data = []
# Adding exchange_scores as a static column
if static_scores:
static_data = pd.DataFrame(
static_scores.items(), columns=["activity_label", "amount"]
)
static_data["year"] = "static"
pivoted_static_data = static_data.pivot(
index="year", columns="activity_label", values="amount"
)
combined_data.append(pivoted_static_data)
combined_data.append(pivoted_data) # making sure the order is correct
# Adding exchange_scores as a prospective column
if prospective_scores:
prospective_data = pd.DataFrame(
prospective_scores.items(), columns=["activity_label", "amount"]
)
prospective_data["year"] = "prospective"
pivoted_prospective_data = prospective_data.pivot(
index="year", columns="activity_label", values="amount"
)
combined_data.append(pivoted_prospective_data)
combined_df = pd.concat(combined_data, axis=0)
if order_stacked_activities:
combined_df = combined_df[
order_stacked_activities
] # change order of activities in the stacked bars of the waterfall
# Calculate the bottom for only the dynamic data
dynamic_bottom = pivoted_data.sum(axis=1).cumsum().shift(1).fillna(0)
if static_scores and prospective_scores:
bottom = pd.concat([pd.Series([0]), dynamic_bottom, pd.Series([0])])
elif static_scores:
bottom = pd.concat([pd.Series([0]), dynamic_bottom])
elif prospective_scores:
bottom = pd.concat([dynamic_bottom, pd.Series([0])])
else:
bottom = dynamic_bottom
# Plotting
ax = combined_df.plot(
kind="bar",
stacked=True,
bottom=bottom,
figsize=(14, 6),
edgecolor="black",
linewidth=0.5,
)
ax.set_ylabel("GWP [kg CO2-eq]")
ax.set_xlabel("")
plt.xticks(rotation=45, ha="right")
if static_scores:
ax.axvline(x=0.5, color="black", linestyle="--", lw=1)
if prospective_scores:
ax.axvline(x=len(combined_df) - 1.5, color="black", linestyle="--", lw=1)
handles, labels = ax.get_legend_handles_labels()
ax.legend(
handles[::-1],
labels[::-1],
loc="center left",
bbox_to_anchor=(1.02, 0.5), # x=1.02 moves it outside, y=0.5 centers vertically
fontsize="small",
)
ax.set_axisbelow(True)
plt.grid(True)
plt.show()
[docs]
def get_exchange(**kwargs) -> Exchange:
"""
Get an exchange from the database.
Parameters
----------
**kwargs :
Arguments to specify an exchange.
- input_node: Input node object
- input_code: Input node code
- input_database: Input node database
- output_node: Output node object
- output_code: Output node code
- output_database: Output node database
Returns
-------
Exchange
The exchange object matching the criteria.
Raises
------
MultipleResults
If multiple exchanges match the criteria.
UnknownObject
If no exchange matches the criteria.
"""
# Process input_node if present
input_node = kwargs.pop("input_node", None)
if input_node:
kwargs["input_code"] = input_node["code"]
kwargs["input_database"] = input_node["database"]
# Process output_node if present
output_node = kwargs.pop("output_node", None)
if output_node:
kwargs["output_code"] = output_node["code"]
kwargs["output_database"] = output_node["database"]
# Map kwargs to database fields
mapping = {
"input_code": ExchangeDataset.input_code,
"input_database": ExchangeDataset.input_database,
"output_code": ExchangeDataset.output_code,
"output_database": ExchangeDataset.output_database,
}
# Build query filters
filters = []
for key, value in kwargs.items():
field = mapping.get(key)
if field is not None:
filters.append(field == value)
# Execute query with filters
qs = ExchangeDataset.select().where(*filters)
candidates = [Exchange(obj) for obj in qs]
num_candidates = len(candidates)
if num_candidates > 1:
raise MultipleResults(
f"Found {num_candidates} results for the given search. "
"Please be more specific or double-check your system model for duplicates."
)
if num_candidates == 0:
raise UnknownObject("No exchange found matching the criteria.")
return candidates[0]
[docs]
def add_temporal_distribution_to_exchange(
temporal_distribution: TemporalDistribution, **kwargs
):
"""
Adds a temporal distribution to an exchange specified by kwargs.
Parameters
----------
temporal_distribution : TemporalDistribution
TemporalDistribution to be added to the exchange.
**kwargs :
Arguments to specify an exchange.
- input_node: Input node object
- input_id: Input node database ID
- input_code: Input node code
- input_database: Input node database
- output_node: Output node object
- output_id: Output node database ID
- output_code: Output node code
- output_database: Output node database
Returns
-------
None
The exchange is saved with the temporal distribution.
"""
from .validation import TemporalDistributionExchangeInputs
TemporalDistributionExchangeInputs(temporal_distribution=temporal_distribution)
exchange = get_exchange(**kwargs)
exchange["temporal_distribution"] = temporal_distribution
exchange.save()
logger.info(f"Added temporal distribution to exchange {exchange}.")
[docs]
def add_temporal_evolution_to_exchange(
temporal_evolution_factors: dict = None,
temporal_evolution_amounts: dict = None,
temporal_evolution_reference: str = "producer",
**kwargs,
):
"""Add temporal evolution data to an exchange specified by kwargs.
Parameters
----------
temporal_evolution_factors : dict, optional
Dictionary mapping datetime keys to scaling factors.
temporal_evolution_amounts : dict, optional
Dictionary mapping datetime keys to absolute amounts.
temporal_evolution_reference : {"producer", "consumer"}, optional
Whether temporal evolution is evaluated at producer or consumer timestamps.
**kwargs :
Arguments to specify an exchange (same as get_exchange).
Returns
-------
None
The exchange is saved with the temporal evolution data.
"""
from .validation import TemporalEvolutionExchangeInputs
TemporalEvolutionExchangeInputs(
temporal_evolution_factors=temporal_evolution_factors,
temporal_evolution_amounts=temporal_evolution_amounts,
temporal_evolution_reference=temporal_evolution_reference,
)
exchange = get_exchange(**kwargs)
if temporal_evolution_factors is not None:
exchange["temporal_evolution_factors"] = temporal_evolution_factors
if temporal_evolution_amounts is not None:
exchange["temporal_evolution_amounts"] = temporal_evolution_amounts
exchange["temporal_evolution_reference"] = temporal_evolution_reference
exchange.save()
logger.info(f"Added temporal evolution to exchange {exchange}.")
[docs]
def get_temporal_evolution_factor(
temporal_evolution: dict,
target_date: datetime,
) -> float:
"""Linearly interpolate a scaling factor for a given date from a temporal evolution dict.
Parameters
----------
temporal_evolution : dict or None
Dictionary mapping datetime keys to float scaling factors.
If None or empty, returns 1.0 (no scaling).
target_date : datetime
The calendar date to look up the factor for.
Returns
-------
float
The interpolated scaling factor. Clamped to the nearest boundary
value for dates outside the specified range.
"""
if not temporal_evolution:
return 1.0
sorted_dates = sorted(temporal_evolution.keys())
if len(sorted_dates) == 1:
return temporal_evolution[sorted_dates[0]]
# Clamp: below minimum
if target_date <= sorted_dates[0]:
return temporal_evolution[sorted_dates[0]]
# Clamp: above maximum
if target_date >= sorted_dates[-1]:
return temporal_evolution[sorted_dates[-1]]
# Find surrounding dates and interpolate
for i in range(len(sorted_dates) - 1):
lower = sorted_dates[i]
upper = sorted_dates[i + 1]
if lower <= target_date <= upper:
weight = (target_date - lower).total_seconds() / (
upper - lower
).total_seconds()
return (
temporal_evolution[lower] * (1 - weight)
+ temporal_evolution[upper] * weight
)
return 1.0 # fallback
