bw_timex.utils#
Module Contents#
Functions#
|
Converts a datetime object to an integer for a given temporal resolution time_res |
|
Extracts the grouping date as a string from a datetime object, based on the chosen temporal |
|
Converts the string of a date to datetime object. |
|
Round a datetime object based on a given resolution |
|
Vectorized equivalent of |
Add a new flow or a list of flows to the available characterization functions. |
|
Getting the name of a node based on the code only. |
|
|
Plot a stacked waterfall chart of characterized inventory data. As comparison, |
|
Get an exchange from the database. |
Adds a temporal distribution to an exchange specified by kwargs. |
|
Add temporal evolution data to an exchange specified by kwargs. |
|
Create an interactive ipywidget for drafting and visualizing temporal distributions and copying |
|
|
Linearly interpolate a scaling factor for a given date from a temporal evolution dict. |
Attributes#
- bw_timex.utils.extract_date_as_integer(dt_obj: datetime.datetime, time_res: str | None = 'year') int[source]#
Converts a datetime object to an integer for a given temporal resolution time_res
- Parameters:
dt_obj (Datetime object.) – Datetime object to be converted to an integer.
time_res (str, optional) – time resolution to be returned: year=YYYY, month=YYYYMM, day=YYYYMMDD, hour=YYYYMMDDHH
- Returns:
date_as_integer – Datetime object converted to an integer in the format of time_res
- Return type:
int
- bw_timex.utils.extract_date_as_string(timestamp: datetime.datetime, temporal_grouping: str) str[source]#
Extracts the grouping date as a string from a datetime object, based on the chosen temporal grouping. E.g. for temporal_grouping = ‘month’, and timestamp = 2023-03-29T01:00:00, it extracts the string ‘202303’.
- Parameters:
timestamp (Datetime object) – Datetime object to be converted to a string.
temporal_grouping (str) – Temporal grouping for the date string. Options are: ‘year’, ‘month’, ‘day’, ‘hour’
- Returns:
date_as_string – Date as a string in the format of the chosen temporal grouping.
- Return type:
str
- bw_timex.utils.convert_date_string_to_datetime(temporal_grouping, date_string) datetime.datetime[source]#
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 object of the date string at the chosen temporal resolution.
- Return type:
datetime
- bw_timex.utils.round_datetime(date: datetime.datetime, resolution: str) datetime.datetime[source]#
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:
rounded datetime object
- Return type:
datetime
- bw_timex.utils.round_datetime_series_to_year(dates: pandas.Series) pandas.Series[source]#
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_datetimeexactly but avoids the per-row Pythonapply.
- bw_timex.utils.add_flows_to_characterization_functions(flows: str | List[str], func: Callable, characterization_functions: dict | None = None) dict[source]#
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:
Updated characterization function dictionary with the new flow(s) and function(s).
- Return type:
dict
- bw_timex.utils.resolve_temporalized_node_name(code: str) str[source]#
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:
Name of the node.
- Return type:
str
- bw_timex.utils.plot_characterized_inventory_as_waterfall(lca_obj, static_scores=None, prospective_scores=None, order_stacked_activities=None)[source]#
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:
plots the waterfall chart.
- Return type:
None
- bw_timex.utils.get_exchange(**kwargs) bw2data.backends.proxies.Exchange[source]#
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:
The exchange object matching the criteria.
- Return type:
Exchange
- Raises:
MultipleResults – If multiple exchanges match the criteria.
UnknownObject – If no exchange matches the criteria.
- bw_timex.utils.add_temporal_distribution_to_exchange(temporal_distribution: bw_temporalis.TemporalDistribution, **kwargs)[source]#
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:
The exchange is saved with the temporal distribution.
- Return type:
None
- bw_timex.utils.add_temporal_evolution_to_exchange(temporal_evolution_factors: dict = None, temporal_evolution_amounts: dict = None, **kwargs)[source]#
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.
**kwargs – Arguments to specify an exchange (same as get_exchange).
- Returns:
The exchange is saved with the temporal evolution data.
- Return type:
None
- bw_timex.utils.interactive_td_widget()[source]#
Create an interactive ipywidget for drafting and visualizing temporal distributions and copying them to the clipboard.
For use in jupyter notebooks.
- Returns:
Interactive widget for drafting temporal distributions.
- Return type:
ipywidgets.VBox
- bw_timex.utils.get_temporal_evolution_factor(temporal_evolution: dict, target_date: datetime.datetime) float[source]#
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:
The interpolated scaling factor. Clamped to the nearest boundary value for dates outside the specified range.
- Return type:
float