:py:mod:`bw_timex.utils` ======================== .. py:module:: bw_timex.utils Module Contents --------------- Functions ~~~~~~~~~ .. autoapisummary:: bw_timex.utils.extract_date_as_integer bw_timex.utils.extract_date_as_string bw_timex.utils.convert_date_string_to_datetime bw_timex.utils.round_datetime bw_timex.utils.round_datetime_series_to_year bw_timex.utils.add_flows_to_characterization_functions bw_timex.utils.resolve_temporalized_node_name bw_timex.utils.plot_characterized_inventory_as_waterfall bw_timex.utils.get_exchange bw_timex.utils.add_temporal_distribution_to_exchange bw_timex.utils.add_temporal_evolution_to_exchange bw_timex.utils.interactive_td_widget bw_timex.utils.get_temporal_evolution_factor Attributes ~~~~~~~~~~ .. autoapisummary:: bw_timex.utils.time_res_mapping_strftime .. py:data:: time_res_mapping_strftime .. py:function:: extract_date_as_integer(dt_obj: datetime.datetime, time_res: Optional[str] = 'year') -> int Converts a datetime object to an integer for a given temporal resolution `time_res` :param dt_obj: Datetime object to be converted to an integer. :type dt_obj: Datetime object. :param time_res: time resolution to be returned: year=YYYY, month=YYYYMM, day=YYYYMMDD, hour=YYYYMMDDHH :type time_res: str, optional :returns: **date_as_integer** -- Datetime object converted to an integer in the format of time_res :rtype: int .. py:function:: extract_date_as_string(timestamp: datetime.datetime, temporal_grouping: str) -> str 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'. :param timestamp: Datetime object to be converted to a string. :type timestamp: Datetime object :param temporal_grouping: Temporal grouping for the date string. Options are: 'year', 'month', 'day', 'hour' :type temporal_grouping: str :returns: **date_as_string** -- Date as a string in the format of the chosen temporal grouping. :rtype: str .. py:function:: convert_date_string_to_datetime(temporal_grouping, date_string) -> datetime.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 :param temporal_grouping: Temporal grouping for the date string. Options are: 'year', 'month', 'day', 'hour' :type temporal_grouping: str :param date_string: Date as a string :type date_string: str :returns: Datetime object of the date string at the chosen temporal resolution. :rtype: datetime .. py:function:: round_datetime(date: datetime.datetime, resolution: str) -> datetime.datetime Round a datetime object based on a given resolution :param date: datetime object to be rounded :type date: datetime :param resolution: Temporal resolution to round the datetime object to. Options are: 'year', 'month', 'day' and 'hour'. :type resolution: str :returns: rounded datetime object :rtype: datetime .. py:function:: round_datetime_series_to_year(dates: pandas.Series) -> pandas.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``. .. py:function:: 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. :param flows: Flow or list of flows to be added to the characterization function dictionary. :type flows: Union[str, List[str]] :param func: Dynamic characterization function for flow. :type func: Callable :param characterization_functions: Dictionary of flows and their corresponding characterization functions. Default is an empty dictionary. :type characterization_functions: dict, optional :returns: Updated characterization function dictionary with the new flow(s) and function(s). :rtype: dict .. py:function:: 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. :param code: Code of the node to resolve. :type code: str :returns: Name of the node. :rtype: str .. py:function:: 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. :param lca_obj: LCA object with characterized inventory data. :type lca_obj: TimexLCA :param static_scores: Dictionary of static scores. Default is None. :type static_scores: dict, optional :param prospective_scores: Dictionary of prospective scores. Default is None. :type prospective_scores: dict, optional :param order_stacked_activities: List of activities to order the stacked bars in the waterfall plot. Default is None. :type order_stacked_activities: list, optional :returns: plots the waterfall chart. :rtype: None .. py:function:: get_exchange(**kwargs) -> bw2data.backends.proxies.Exchange Get an exchange from the database. :param \*\*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. :rtype: Exchange :raises MultipleResults: If multiple exchanges match the criteria. :raises UnknownObject: If no exchange matches the criteria. .. py:function:: add_temporal_distribution_to_exchange(temporal_distribution: bw_temporalis.TemporalDistribution, **kwargs) Adds a temporal distribution to an exchange specified by kwargs. :param temporal_distribution: TemporalDistribution to be added to the exchange. :type temporal_distribution: TemporalDistribution :param \*\*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. :rtype: None .. py:function:: add_temporal_evolution_to_exchange(temporal_evolution_factors: dict = None, temporal_evolution_amounts: dict = None, **kwargs) Add temporal evolution data to an exchange specified by kwargs. :param temporal_evolution_factors: Dictionary mapping datetime keys to scaling factors. :type temporal_evolution_factors: dict, optional :param temporal_evolution_amounts: Dictionary mapping datetime keys to absolute amounts. :type temporal_evolution_amounts: dict, optional :param \*\*kwargs: Arguments to specify an exchange (same as get_exchange). :returns: The exchange is saved with the temporal evolution data. :rtype: None .. py:function:: interactive_td_widget() 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. :rtype: ipywidgets.VBox .. py:function:: get_temporal_evolution_factor(temporal_evolution: dict, target_date: datetime.datetime) -> float Linearly interpolate a scaling factor for a given date from a temporal evolution dict. :param temporal_evolution: Dictionary mapping datetime keys to float scaling factors. If None or empty, returns 1.0 (no scaling). :type temporal_evolution: dict or None :param target_date: The calendar date to look up the factor for. :type target_date: datetime :returns: The interpolated scaling factor. Clamped to the nearest boundary value for dates outside the specified range. :rtype: float