From 687cbfaa4389ec6f2328bc1b229d54555eebaef2 Mon Sep 17 00:00:00 2001 From: muelleram Date: Fri, 25 Oct 2024 15:51:03 +0200 Subject: [PATCH] harmonize f-strings, fix typos --- bw_timex/dynamic_biosphere_builder.py | 27 ++++--- bw_timex/edge_extractor.py | 47 +++++++----- bw_timex/helper_classes.py | 16 ++-- bw_timex/matrix_modifier.py | 45 ++++++++---- bw_timex/timeline_builder.py | 50 ++++++++----- bw_timex/timex_lca.py | 102 ++++++++++++++++---------- bw_timex/utils.py | 24 +++--- 7 files changed, 182 insertions(+), 129 deletions(-) diff --git a/bw_timex/dynamic_biosphere_builder.py b/bw_timex/dynamic_biosphere_builder.py index 2434d1c..9e5699d 100644 --- a/bw_timex/dynamic_biosphere_builder.py +++ b/bw_timex/dynamic_biosphere_builder.py @@ -11,7 +11,8 @@ class DynamicBiosphereBuilder: """ - Class for building a dynamic biosphere matrix with dimensions (biosphere flow at a specific point in time) x (processes) + Class for building a dynamic biosphere matrix with dimensions + (biosphere flow at a specific point in time) x (processes) """ def __init__( @@ -33,7 +34,7 @@ def __init__( Parameters ---------- - lca_obj : LCA objecct + lca_obj : LCA object instance of the bw2calc LCA class, e.g. TimexLCA.lca activity_time_mapping_dict : dict A dictionary mapping activity to their respective timing in the format @@ -104,12 +105,12 @@ def build_dynamic_biosphere_matrix( from_timeline: bool = False, ): """ - This function creates a separate biosphere matrix, with the dimenions - (bio_flows at a specific timestep) x (processes). + This function creates a separate biosphere matrix, with the dimensions + (bio_flows at a specific time step) x (processes). Every temporally resolved biosphere flow has its own row in the matrix, making it highly sparse. The timing of the emitting process and potential additional temporal information of - the bioshpere flow (e.g. delay of emission compared to the timing of the process) are considered. + the biosphere flow (e.g. delay of emission compared to the timing of the process) are considered. Absolute Temporal Distributions for biosphere exchanges are dealt with as a look up function: If an activity happens at timestamp X and the biosphere exchange has an @@ -117,7 +118,7 @@ def build_dynamic_biosphere_matrix( to timestamp X. E.g.: X = 2024, TD=(data=[2020,2021,2022,2023,2024,.....,2120], amount=[3,4,4,5,6,......,3]), it will look up the value 6 corresponding 2024. If timestamp X does not exist, it finds the nearest timestamp available (if two timestamps are equally close, - it will take the first in order of appearance (see numpy.argmin() for this behaviour). + it will take the first in order of appearance (see numpy.argmin() for this behavior). Parameters ---------- @@ -128,7 +129,7 @@ def build_dynamic_biosphere_matrix( Returns ------- dynamic_biomatrix : scipy.sparse.csr_matrix - A sparse matrix with the dimensions (bio_flows at a specific timestep) x (processes). + A sparse matrix with the dimensions (bio_flows at a specific time step) x (processes). """ for row in self.timeline.itertuples(): @@ -271,7 +272,8 @@ def demand_from_timeline(self, row, original_db): Returns ------- demand: dict - A demand-dictionary with as keys the ids of the time-mapped activities and as values the share. + A demand-dictionary with as keys the ids of the time-mapped activities + and as values the share. """ @@ -289,7 +291,7 @@ def demand_from_timeline(self, row, original_db): def demand_from_technosphere(self, idx, process_col_index): """ - Returns a demand dict of background processes based on the technosphere colummn. + Returns a demand dict of background processes based on the technosphere column. Foreground exchanges are skipped as these are added separately. Parameters: @@ -302,7 +304,8 @@ def demand_from_technosphere(self, idx, process_col_index): Returns ------- demand: dict - A demand-dictionary with as keys the brightway ids of the consumed background activities and as values their consumed amount. + A demand-dictionary with as keys the brightway ids of the consumed background + activities and as values their consumed amount. """ technosphere_column = ( self.technosphere_matrix[:, process_col_index].toarray().ravel() @@ -327,8 +330,8 @@ def demand_from_technosphere(self, idx, process_col_index): def add_matrix_entry_for_biosphere_flows(self, row, col, amount): """ Adds an entry to a list of row, col and values, which are then used to construct the - dynamic biosphere matrix. Only unqiue entries are added, i.e. if the same row and col index - already exists, the value is not added again. + dynamic biosphere matrix. Only unique entries are added, i.e. if the same row and + col index already exists, the value is not added again. Parameters ---------- diff --git a/bw_timex/edge_extractor.py b/bw_timex/edge_extractor.py index 7df6db0..2ccd71a 100644 --- a/bw_timex/edge_extractor.py +++ b/bw_timex/edge_extractor.py @@ -1,7 +1,7 @@ from dataclasses import dataclass from heapq import heappop, heappush from numbers import Number -from typing import Callable, List +from typing import Callable import numpy as np from bw_temporalis import TemporalDistribution, TemporalisLCA @@ -34,25 +34,30 @@ class Edge: class EdgeExtractor(TemporalisLCA): """ - Child class of TemporalisLCA that traverses the supply chain just as the parent class but can create a timeline of edges, in addition timeline of flows or nodes. - The edge timeline is then used to match the timestamp of edges to that of background databases and to replace these edges with edges from these background databases using Brightway Datapackages. + Child class of TemporalisLCA that traverses the supply chain just as the parent + class but can create a timeline of edges, in addition timeline of flows or nodes. + The edge timeline is then used to match the timestamp of edges to that of background + databases and to replace these edges with edges from these background databases + using Brightway Datapackages. """ def __init__(self, *args, edge_filter_function: Callable = None, **kwargs) -> None: """ - Initialize the EdgeExtractor class and traverses the supply chain using functions of the parent class TemporalisLCA. + Initialize the EdgeExtractor class and traverses the supply chain using + functions of the parent class TemporalisLCA. Parameters ---------- *args : Variable length argument list edge_filter_function : Callable, optional - A callable that filters edges. If not provided, a function that always returns False is used. + A callable that filters edges. If not provided, a function that always + returns False is used. **kwargs : Arbitrary keyword arguments Returns ------- - None but stores the output of the TemporalisLCA graph traversal (incl. relation of edges (edge_mapping) - and nodes (node_mapping) in the instance of the class. + None but stores the output of the TemporalisLCA graph traversal (incl. relation + of edges (edge_mapping) and nodes (node_mapping) in the instance of the class. """ super().__init__(*args, **kwargs) # use __init__ of TemporalisLCA @@ -64,9 +69,12 @@ def __init__(self, *args, edge_filter_function: Callable = None, **kwargs) -> No def build_edge_timeline(self) -> list: """ - Creates a timeline of the edges from the output of the graph traversal. Starting from the edges of the functional unit node, it goes through each node using a heap, selecting the node with the highest impact first. - It, then, propagates the TemporalDistributions of the edges from node to node through time using convolution-operators. - It stops in case the current edge is known to have no temporal distribution (=leaf) (e.g. part of background database). + Creates a timeline of the edges from the output of the graph traversal. + Starting from the edges of the functional unit node, it goes through + each node using a heap, selecting the node with the highest impact first. + It, then, propagates the TemporalDistributions of the edges from node to + node through time using convolution-operators. It stops in case the current edge + is known to have no temporal distribution (=leaf) (e.g. part of background database). Parameters ---------- @@ -75,7 +83,8 @@ def build_edge_timeline(self) -> list: Returns ------- list - A list of Edge instances with timestamps and amounts, and ids of its producing and consuming node. + A list of Edge instances with timestamps and amounts, and ids of its producing + and consuming node. """ @@ -182,9 +191,11 @@ def join_datetime_and_timedelta_distributions( self, td_producer: TemporalDistribution, td_consumer: TemporalDistribution ) -> TemporalDistribution: """ - Joins a datetime TemporalDistribution (td_producer) with a timedelta TemporalDistribution (td_consumer) to create a new TemporalDistribution. + Joins a datetime TemporalDistribution (td_producer) with a timedelta + TemporalDistribution (td_consumer) to create a new TemporalDistribution. - If the producer does not have a TemporalDistribution, the consumer's TemporalDistribution is returned to continue the timeline. + If the producer does not have a TemporalDistribution, the consumer's + TemporalDistribution is returned to continue the timeline. If both the producer and consumer have TemporalDistributions, they are joined together. Parameters @@ -197,12 +208,14 @@ def join_datetime_and_timedelta_distributions( Returns ------- TemporalDistribution - A new TemporalDistribution that is the result of joining the producer and consumer TemporalDistributions. + A new TemporalDistribution that is the result of joining the producer + and consumer TemporalDistributions. Raises ------ ValueError - If the dtype of `td_consumer.date` is not `datetime64[s]` or the dtype of `td_producer.date` is not `timedelta64[s]`. + If the dtype of `td_consumer.date` is not `datetime64[s]` or the dtype + of `td_producer.date` is not `timedelta64[s]`. """ # if an edge does not have a TD, then return the consumer_td so that the timeline continues @@ -230,7 +243,5 @@ def join_datetime_and_timedelta_distributions( return TemporalDistribution(date, amount) else: raise ValueError( - "Can't join TemporalDistribution and something else: Trying with {} and {}".format( - type(td_consumer.date), type(td_producer.date) - ) + f"Can't join TemporalDistribution and something else: Trying with {type(td_consumer.date)} and {type(td_producer.date)}" ) diff --git a/bw_timex/helper_classes.py b/bw_timex/helper_classes.py index 4eb87b5..9733a09 100644 --- a/bw_timex/helper_classes.py +++ b/bw_timex/helper_classes.py @@ -39,9 +39,7 @@ def add( ] if len(checklist_items) != 0: warnings.warn( - "tried to add {} to the SetList\n, but {} already exist in the SetList in:\n {}. \n Skipping {}".format( - new_set, checklist_items, checklist_sets, new_set - ) + f"tried to add {new_set} to the SetList\n, but {checklist_items} already exist in the SetList in:\n {checklist_sets}. \n Skipping {new_set}" ) pass else: @@ -63,14 +61,12 @@ def __getitem__(self, key: any) -> set: sets = [matching_set for matching_set in self.list if key in matching_set] if len(sets) > 1: warnings.warn( - "Key found in multiple sets!!!! {} Please check! Returning only the first set".format( - sets - ) + f"Key found in multiple sets! Please check {sets} ! Returning only the first set" ) if len(sets) > 0: return sets[0] else: - warnings.warn("Key {} not found in SetList".format(key)) + warnings.warn(f"Key {key} not found in SetList") return None def __len__( @@ -84,7 +80,8 @@ def __repr__(self): class TimeMappingDict(dict): """ - This class is used to create a dictionary that maps a tuple of (flow and timestamp) to an unique integer id. + This class is used to create a dictionary that maps a tuple of (flow and timestamp) + to an unique integer id. """ def __init__(self, start_id=2, *args, **kwargs) -> None: @@ -123,7 +120,8 @@ def add(self, process_time_tuple, unique_id=None): Returns ------- int - An unique id for the process_time_tuple, and adds it to the dictionary, if not already present. + An unique id for the process_time_tuple, and adds it to the dictionary, + if not already present. """ if process_time_tuple in self: return self[process_time_tuple] diff --git a/bw_timex/matrix_modifier.py b/bw_timex/matrix_modifier.py index 8aec067..06c8c9e 100644 --- a/bw_timex/matrix_modifier.py +++ b/bw_timex/matrix_modifier.py @@ -25,7 +25,8 @@ def __init__( name: Optional[str] = None, ) -> None: """ - Initializes the MatrixModifier object and creates empty sets to collect the ids of temporalized processes and temporal markets. + Initializes the MatrixModifier object and creates empty sets to collect + the ids of temporalized processes and temporal markets. Parameters ---------- @@ -49,7 +50,8 @@ def __init__( def create_datapackage(self) -> None: """ - Creates a list of datapackages for the technosphere and biosphere matrices, by calling the respective functions. + Creates a list of datapackages for the technosphere and biosphere matrices, + by calling the respective functions. Parameters ---------- @@ -67,13 +69,17 @@ def create_datapackage(self) -> None: def create_technosphere_datapackage(self) -> bwp.Datapackage: """ - Creates the modifications to the technosphere matrix in form of a datapackage. Datapackages add or overwrite datapoints in the LCA matrices before LCA calculations. - The technosphere datapackage adds the temporalized processes from the timeline to the technosphere matrix. + Creates the modifications to the technosphere matrix in form of a datapackage. + Datapackages add or overwrite data points in the LCA matrices before LCA calculations. + The technosphere datapackage adds the temporalized processes from the timeline + to the technosphere matrix. The heavy lifting of this method happens in the method `add_row_to_datapackage()`. - Here, each node with a temporal distribution is "exploded", which means each occurrence of this node (e.g. steel production on 2020-01-01 - and steel production on 2015-01-01) becomes a separate, time-explicit new node, by adding the new elements to the technosphere matrix. - For processes at the intersection with background databases, the timing of the exchanges determines which background database to link to in so called "Temporal Markets". + Here, each node with a temporal distribution is "exploded", which means each occurrence + of this node (e.g. steel production on 2020-01-01 and steel production on 2015-01-01) + becomes a separate, time-explicit new node, by adding the new elements to the technosphere matrix. + For processes at the intersection with background databases, the timing of the exchanges + determines which background database to link to in so called "Temporal Markets". Parameters ---------- @@ -110,13 +116,15 @@ def create_technosphere_datapackage(self) -> bwp.Datapackage: def create_biosphere_datapackage(self) -> bwp.Datapackage: """ - Creates the modifications to the biosphere matrix in form of a datapackage. Datapackages add or overwrite datapoints in the LCA matrices before LCA calculations. + Creates the modifications to the biosphere matrix in form of a datapackage. + Datapackages add or overwrite data points in the LCA matrices before LCA calculations. It adds the biosphere flows to the exploded technosphere processes. - This function iterates over each unique producer, and for each biosphere exchange of the original activity, - it creates a new biosphere exchange for the new "temporalized" node. + This function iterates over each unique producer, and for each biosphere exchange + of the original activity, it creates a new biosphere exchange for the new "temporalized" node. - Temporal markets have no biosphere exchanges, as they only divide the amount of a technosphere exchange between the different databases. + Temporal markets have no biosphere exchanges, as they only divide the amount of + a technosphere exchange between the different databases. Parameters ---------- @@ -172,10 +180,14 @@ def add_row_to_datapackage( new_nodes: set, ) -> None: """ - This adds the modifications to the technosphere matrix for each time-dependent exchange as datapackage elements to a given `bwp.Datapackage`. + This adds the modifications to the technosphere matrix for each time-dependent exchange + as datapackage elements to a given `bwp.Datapackage`. Modifications include: - 1) Exploded processes: new matrix elements for time-explicit consumer and time-explicit producer, representing the temporal edge between them. - 2) Temporal markets: new matrix entries for "temporal markets" and links to the producers in temporally matching background databases. Processes in the background databases are matched on name, reference product and location. + 1) Exploded processes: new matrix elements for time-explicit consumer and time-explicit + producer, representing the temporal edge between them. + 2) Temporal markets: new matrix entries for "temporal markets" and links to the producers + in temporally matching background databases. Processes in the background databases are + matched on name, reference product and location. 3) Diagonal entries: ones on the diagonal for new nodes. This function also collects the ids of new nodes, temporalized nodes and temporal markets. @@ -191,7 +203,8 @@ def add_row_to_datapackage( Returns ------- - None but adds elements for this edge to the bwp.Datapackage and stores the ids of new nodes, temporalized nodes and temporal markets. + None but adds elements for this edge to the bwp.Datapackage and stores the ids of new nodes, + temporalized nodes and temporal markets. """ if row.consumer == -1: # functional unit @@ -209,7 +222,7 @@ def add_row_to_datapackage( previous_producer_id = row.producer previous_producer_node = bd.get_node( id=previous_producer_id - ) # in future versions, insead of getting node, just provide list of producer ids + ) # in future versions, instead of getting node, just provide list of producer ids # Add entry between exploded consumer and exploded producer (not in background database) datapackage.add_persistent_vector( diff --git a/bw_timex/timeline_builder.py b/bw_timex/timeline_builder.py index 6e70012..72dd228 100644 --- a/bw_timex/timeline_builder.py +++ b/bw_timex/timeline_builder.py @@ -107,11 +107,17 @@ def __init__( def build_timeline(self) -> pd.DataFrame: """ - Create a dataframe with grouped, time-explicit edges and, for each grouped edge, interpolate to the database with the closest time of representativeness. + Create a DataFrame with grouped, time-explicit edges and, for each grouped edge, + interpolate to the database with the closest time of representativeness. - It uses the edge_timeline, an output from the graph traversal in EdgeExtractor. Edges from same producer to same consumer that occur at different times within the same time window (temporal_grouping) are grouped together. Possible temporal groupings are "year", "month", "day" and "hour". + It uses the edge_timeline, an output from the graph traversal in EdgeExtractor. + Edges from same producer to same consumer that occur at different times within + the same time window (temporal_grouping) are grouped together. + Possible temporal groupings are "year", "month", "day" and "hour". - For edges between forground and background system, the column "interpolation weights" assigns the ratio [0-1] of the edge's amount to be taken from the database with the closest time of representativeness. If a process is in the foreground system only, the interpolation weight is set to None. + For edges between foreground and background system, the column "interpolation weights" + assigns the ratio [0-1] of the edge's amount to be taken from the database with the closest + time of representativeness. If a process is in the foreground system only, the interpolation weight is set to None. Available interpolation types are: - "linear": linear interpolation between the two closest databases, based on temporal distance. - "closest": closest database is assigned 1 @@ -195,7 +201,7 @@ def build_timeline(self) -> pd.DataFrame: lambda x: convert_date_string_to_datetime(self.temporal_grouping, x) ) - # add dates as integers as hashes to the dataframe + # add dates as integers as hashes to the DataFrame grouped_edges["hash_producer"] = grouped_edges["date_producer"].apply( lambda x: extract_date_as_integer(x, time_res=self.temporal_grouping) ) @@ -213,7 +219,7 @@ def build_timeline(self) -> pd.DataFrame: ) ) - # store the ids from the time_mapping_dict in dataframe + # store the ids from the time_mapping_dict in DataFrame grouped_edges["time_mapped_producer"] = grouped_edges.apply( lambda row: self.get_time_mapping_key(row.producer, row.hash_producer), axis=1, @@ -228,7 +234,7 @@ def build_timeline(self) -> pd.DataFrame: axis=1, ) - # Add interpolation weights to background databases to the dataframe + # Add interpolation weights to background databases to the DataFrame grouped_edges = self.add_column_interpolation_weights_to_timeline( grouped_edges, interpolation_type=self.interpolation_type, @@ -268,13 +274,13 @@ def build_timeline(self) -> pd.DataFrame: def check_database_names(self) -> None: """ - Check that the strings of the databases exist in the databases of the brightway project. + Check that the strings of the databases exist in the databases of the Brightway project. """ for db in self.database_date_dict_static_only.keys(): assert ( db in bd.databases - ), f"{db} is not in your brightway project databases." + ), f"{db} is not in your Brightway project databases." def extract_edge_data(self, edge: Edge) -> dict: """ @@ -309,7 +315,8 @@ def extract_edge_data(self, edge: Edge) -> dict: def adjust_sign_of_amount_based_on_edge_type(self, edge_type): """ - It checks if the an exchange is of type substitution or a technosphere exchange, based on bw2data labelling convention, and adjusts the amount accordingly. + It checks if the an exchange is of type substitution or a technosphere exchange, + based on bw2data labelling convention, and adjusts the amount accordingly. Flips the sign of the amount value in the timeline for substitution (positive technosphere) exchanges. Parameters @@ -326,7 +333,7 @@ def adjust_sign_of_amount_based_on_edge_type(self, edge_type): if ( edge_type in labels.technosphere_negative_edge_types - ): # variants of technosphere lables + ): # variants of technosphere labels multiplicator = 1 elif ( edge_type in labels.technosphere_positive_edge_types @@ -374,19 +381,22 @@ def add_column_interpolation_weights_to_timeline( interpolation_type: str = "linear", ) -> pd.DataFrame: """ - Add a column to a timeline with the weights for an interpolation between the two nearest dates, from the list of dates of the available databases. + Add a column to a timeline with the weights for an interpolation between + the two nearest dates, from the list of dates of the available databases. Parameters ---------- tl_df: pd.DataFrame - Timeline as a dataframe. + Timeline as a DataFrame. interpolation_type: str, optional - Type of interpolation between the nearest lower and higher dates. Available options: "linear" and "nearest", defaulting to "linear". + Type of interpolation between the nearest lower and higher dates. + Available options: "linear" and "nearest", defaulting to "linear". Returns ------- pd.DataFrame - Timeline as a dataframe with a column 'interpolation_weights' added, this column looks like {database_name: weight, database_name: weight}. + Timeline as a DataFrame with a column 'interpolation_weights' added, + this column looks like {database_name: weight, database_name: weight}. """ if not self.database_date_dict_static_only: tl_df["interpolation_weights"] = None @@ -470,7 +480,8 @@ def get_weights_for_interpolation_between_nearest_years( interpolation_type: str = "linear", ) -> dict: """ - Find the nearest dates (lower and higher) for a given date from a list of dates and calculate the interpolation weights based on temporal proximity. + Find the nearest dates (lower and higher) for a given date from a list of dates + and calculate the interpolation weights based on temporal proximity. Parameters ---------- @@ -540,16 +551,17 @@ def add_interpolation_weights_at_intersection_to_background( self, row ) -> Union[dict, None]: """ - returns the interpolation weights to background databases only for those exchanges, where the producing process - actually comes from a background database (temporal markets). + returns the interpolation weights to background databases only for those exchanges, + where the producing process actually comes from a background database (temporal markets). Only these processes are receiving inputs from the background databases. - All other process in the timeline are not directly linked to the background, so the interpolation weight info is not needed and set to None + All other process in the timeline are not directly linked to the background, + so the interpolation weight info is not needed and set to None Parameters ---------- row : pd.Series - Row of the timeline dataframe + Row of the timeline DataFrame Returns ------- dict diff --git a/bw_timex/timex_lca.py b/bw_timex/timex_lca.py index 0cc6dd1..65ef180 100644 --- a/bw_timex/timex_lca.py +++ b/bw_timex/timex_lca.py @@ -44,7 +44,7 @@ class TimexLCA: """ Class to perform time-explicit LCA calculations. - A TimexLCA contains the LCI of processes occuring at explicit points in time. It tracks the timing of processes, + A TimexLCA contains the LCI of processes occurring at explicit points in time. It tracks the timing of processes, relinks their technosphere and biosphere exchanges to match the technology landscape at that point in time, and also keeps track of the timing of the resulting emissions. As such, it combines prospective and dynamic LCA approaches. @@ -54,7 +54,7 @@ class TimexLCA: the timing of the processes, bw_timex matches the processes at the intersection between foreground and background to the best available background databases. This temporal relinking is achieved by using datapackages to add new time-specific processes. The new processes and their - exchanges to other technosphere processes or biosphere flows extent the technopshere and + exchanges to other technosphere processes or biosphere flows extent the technosphere and biosphere matrices. Temporal information of both processes and biosphere flows is retained, allowing for dynamic @@ -65,7 +65,7 @@ class TimexLCA: 2) a static time-explicit LCA score (`TimexLCA.static_score`), which links LCIs to the respective background databases, but without dynamic characterization of the time-explicit inventory 3) a dynamic time-explicit LCA score (`TimexLCA.dynamic_score`), with dynamic inventory and - dynamic charaterization. These are provided for radiative forcing and GWP but can also be + dynamic characterization. These are provided for radiative forcing and GWP but can also be user-defined. Example @@ -132,7 +132,7 @@ def __init__( } # Create some collections of nodes that will be useful down the line, e.g. all nodes from - # the background databases that link to foregroud nodes. + # the background databases that link to foreground nodes. self.create_node_id_collection_dict() # Calculate static LCA results using a custom prepare_lca_inputs function that includes all @@ -167,8 +167,9 @@ def build_timeline( Parameters ---------- starting_datetime: datetime | str, optional - Point in time when the demand occurs. This is the initial starting point of the graph traversal and the - timeline. Something like `"now"` or `"2023-01-01"`. Default is `"now"`. + Point in time when the demand occurs. This is the initial starting point of the + graph traversal and the timeline. Something like `"now"` or `"2023-01-01"`. + Default is `"now"`. temporal_grouping : str, optional Time resolution for grouping exchanges over time in the timeline. Default is 'year', other options are 'month', 'day', 'hour'. @@ -273,14 +274,17 @@ def lci( Calculates the time-explicit LCI. There are two ways to generate time-explicit LCIs: - If `expand_technosphere' is True, the biosphere and technosphere matrices are expanded by inserting time-specific - processes via the `MatrixModifier` class by calling `TimexLCA.build_datapackage(). Otherwise ('expand_technosphere' is False), it - generates a dynamic inventory directly from the timeline without technosphere matrix calculations. - - Next to the choice above concerning how to retrieve the time-explicit inventory, users can also decide if - they want to retain all temporal information at the biosphere level (build_dynamic_biosphere = True). - Set `build_dynamic_biosphere` to False if you only want to get a new overall score of the time-explicit inventory and don't care about - the timing of the emissions. This saves time and memory. + If `expand_technosphere' is True, the biosphere and technosphere matrices are expanded by inserting + time-specific processes via the `MatrixModifier` class by calling `TimexLCA.build_datapackage(). + Otherwise ('expand_technosphere' is False), it generates a dynamic inventory directly from the + timeline without technosphere matrix calculations. + + Next to the choice above concerning how to retrieve the time-explicit inventory, users + can also decide if they want to retain all temporal information at the biosphere level + (build_dynamic_biosphere = True). + Set `build_dynamic_biosphere` to False if you only want to get a new overall score of + the time-explicit inventory and don't care about the timing of the emissions. + This saves time and memory. Parameters ---------- @@ -288,7 +292,8 @@ def lci( if True, build the dynamic biosphere matrix and calculate the dynamic LCI. Default is True. expand_technosphere: bool - if True, creates an expanded time-explicit technosphere and biosphere matrix and calculates the LCI from it. + if True, creates an expanded time-explicit technosphere and biosphere matrix and + calculates the LCI from it. if False, creates no new technosphere, but calculates the dynamic inventory directly from the timeline. Building from the timeline currently only works if `build_dynamic_biosphere` is also True. @@ -300,14 +305,16 @@ def lci( See also -------- - build_datapackage: Method to create the datapackages that contain the modifications to the technopshere and biosphere matrix using the `MatrixModifier` class. + build_datapackage: Method to create the datapackages that contain the modifications + to the technosphere and biosphere matrix using the `MatrixModifier` class. calculate_dynamic_inventory: Method to calculate the dynamic inventory if `build_dynamic_biosphere` is True. """ if not expand_technosphere and not build_dynamic_biosphere: raise ValueError( - "Currently, it is not possible to skip the construction of the dynamic biosphere when building the inventories from the timeline.\ - Please either set build_dynamic_biosphere=True or expand_technosphere=True" + "Currently, it is not possible to skip the construction of the dynamic \ + biosphere when building the inventories from the timeline.\ + Please either set build_dynamic_biosphere=True or expand_technosphere=True" ) if not hasattr(self, "timeline"): @@ -428,11 +435,12 @@ def dynamic_lcia( Returns ------- pandas.DataFrame - A Dataframe with the characterized inventory for the chosen metric and parameters. + A DataFrame with the characterized inventory for the chosen metric and parameters. See also -------- - dynamic_charaterization: Package handling the dynamic characterization: https://dynamic-characterization.readthedocs.io/en/latest/ + dynamic_characterization: Package handling the dynamic characterization: + https://dynamic-characterization.readthedocs.io/en/latest/ """ if not hasattr(self, "dynamic_inventory"): @@ -528,7 +536,7 @@ def dynamic_score(self) -> float: def build_datapackage(self) -> list: """ - Creates the datapackages that contain the modifications to the technopshere and biosphere + Creates the datapackages that contain the modifications to the technosphere and biosphere matrix using the `MatrixModifier` class. Parameters @@ -579,7 +587,8 @@ def calculate_dynamic_inventory( See also -------- - bw_timex.dynamic_biosphere_builder.DynamicBiosphereBuilder: Class for creating the dynamic biosphere matrix and inventory. + bw_timex.dynamic_biosphere_builder.DynamicBiosphereBuilder: Class for creating + the dynamic biosphere matrix and inventory. """ if not hasattr(self, "lca"): @@ -613,7 +622,7 @@ def calculate_dynamic_inventory( # Build the dynamic inventory count = len(self.dynamic_biosphere_builder.dynamic_supply_array) - # diagnolization of supply array keeps the dimension of the process, which we want to pass + # diagonalization of supply array keeps the dimension of the process, which we want to pass # as additional information to the dynamic inventory dict diagonal_supply_array = sparse.spdiags( [self.dynamic_biosphere_builder.dynamic_supply_array], [0], count, count @@ -750,7 +759,8 @@ def prepare_base_lca_inputs( See also -------- - bw2data.compat.prepare_lca_inputs: Original code this function is adapted from (https://github.com/brightway-lca/brightway2-data/blob/main/bw2data/compat.py). + bw2data.compat.prepare_lca_inputs: Original code this function is adapted from + (https://github.com/brightway-lca/brightway2-data/blob/main/bw2data/compat.py). """ if not projects.dataset.data.get("25"): @@ -863,7 +873,8 @@ def prepare_bw_timex_inputs( See also -------- - bw2data.compat.prepare_lca_inputs: Original code this function is adapted from (https://github.com/brightway-lca/brightway2-data/blob/main/bw2data/compat.py). + bw2data.compat.prepare_lca_inputs: Original code this function is adapted + from (https://github.com/brightway-lca/brightway2-data/blob/main/bw2data/compat.py). """ if not projects.dataset.data.get("25"): @@ -981,13 +992,21 @@ def create_node_id_collection_dict(self) -> None: Available collections are: - ``demand_database_names``: set of database names of the demand processes - - ``demand_dependent_database_names``: set of database names of all processes that depend on the demand processes - - ``demand_dependent_background_database_names``: set of database names of all processes that depend on the demand processes and are in the background databases - - ``demand_dependent_background_node_ids``: set of node ids of all processes that depend on the demand processes and are in the background databases - - ``foreground_node_ids``: set of node ids of all processes that are not in the background databases - - ``first_level_background_node_ids_static``: set of node ids of all processes that are in the background databases and are directly linked to the demand processes - - ``first_level_background_node_ids_interpolated``: like first_level_background_node_ids_static, but includes first level background processes from the other time explicit databases that are used (is filled after timeline is built) - - ``first_level_background_node_id_dbs``: dictionary with the first_level_background_node_ids_static as keys returning their database + - ``demand_dependent_database_names``: set of database names of all processes + that depend on the demand processes + - ``demand_dependent_background_database_names``: set of database names of all + processes that depend on the demand processes and are in the background databases + - ``demand_dependent_background_node_ids``: set of node ids of all processes + that depend on the demand processes and are in the background databases + - ``foreground_node_ids``: set of node ids of all processes that are + not in the background databases + - ``first_level_background_node_ids_static``: set of node ids of all + processes that are in the background databases and are directly linked to the demand processes + - ``first_level_background_node_ids_interpolated``: like first_level_background_node_ids_static, + but includes first level background processes from the other time explicit databases that are + used (is filled after timeline is built) + - ``first_level_background_node_id_dbs``: dictionary with the first_level_background_node_ids_static + as keys returning their database It also initiates an instance of SetList which contains all mappings of equivalent activities across time-specific databases: @@ -1068,7 +1087,7 @@ def add_interdatabase_activity_mapping_from_timeline(self) -> None: with only those activities and background databases that are actually mapped in the timeline. Also adds the ids to the node_id_collection_dict["first_level_background_node_ids_interpolated"]. This avoids - unneccessary peewee calls. + unnecessary peewee calls. Parameters ---------- @@ -1077,7 +1096,9 @@ def add_interdatabase_activity_mapping_from_timeline(self) -> None: Returns ------- - None, but adds the ids of producers in other background databases (only those interpolated to in the timeline) to the `interdatabase_activity_mapping` and `node_id_collection["first_level_background_node_ids_interpolated"]`. + None, but adds the ids of producers in other background databases + (only those interpolated to in the timeline) to the `interdatabase_activity_mapping` + and `node_id_collection["first_level_background_node_ids_interpolated"]`. """ if not hasattr(self, "timeline"): warnings.warn( @@ -1157,7 +1178,7 @@ def collect_temporalized_processes_from_timeline(self) -> None: Returns ------- None, but adds "temporal_markets" and "temporalized_processes" to the - node_id_colletion_dict based on the timeline. + node_id_collection_dict based on the timeline. """ unique_producers = ( @@ -1188,10 +1209,11 @@ def add_static_activities_to_time_mapping_dict(self) -> None: Adds all activities from the static LCA to `activity_time_mapping_dict`, an instance of `TimeMappingDict`. This gives a unique mapping in the form of (('database', 'code'), datetime_as_integer): time_mapping_id) that is later used to uniquely - identify time-resolved processes. Here, the activity_time_mapping_dict is the pre-population with - the static activities. The time-explicit activities (from other temporalized background - databases) are added later on by the TimelineBuilder. Activities in the foreground database are - mapped with (('database', 'code'), "dynamic"): time_mapping_id)" as their timing is not yet known. + identify time-resolved processes. Here, the activity_time_mapping_dict is the + pre-population with the static activities. The time-explicit activities (from other + temporalized background databases) are added later on by the TimelineBuilder. + Activities in the foreground database are mapped with + (('database', 'code'), "dynamic"): time_mapping_id)" as their timing is not yet known. Parameters ---------- @@ -1413,7 +1435,7 @@ def plot_dynamic_characterized_inventory( ) -> None: """ Plot the characterized inventory of the dynamic LCI in a very simple plot. - Legend and title are selcted automatically based on the chosen metric. + Legend and title are selected automatically based on the chosen metric. Parameters ---------- diff --git a/bw_timex/utils.py b/bw_timex/utils.py index 4e92e10..2745a1c 100644 --- a/bw_timex/utils.py +++ b/bw_timex/utils.py @@ -40,9 +40,7 @@ def extract_date_as_integer(dt_obj: datetime, time_res: Optional[str] = "year") """ if time_res not in time_res_to_int_dict.keys(): warnings.warn( - 'time_res: {} is not a valid option. Please choose from: {} defaulting to "year"'.format( - time_res, time_res_to_int_dict.keys() - ), + f'time_res: {time_res} is not a valid option. Please choose from: {list(time_res_to_int_dict.keys())} defaulting to "year"', category=Warning, ) formatted_date = dt_obj.strftime(time_res_to_int_dict[time_res]) @@ -73,24 +71,22 @@ def extract_date_as_string(timestamp: datetime, temporal_grouping: str) -> str: if temporal_grouping not in time_res_to_int_dict.keys(): warnings.warn( - 'temporal_grouping: {} is not a valid option. Please choose from: {} defaulting to "year"'.format( - temporal_grouping, time_res_to_int_dict.keys() - ), + f'temporal_grouping: {temporal_grouping} is not a valid option. Please choose from: {list(time_res_to_int_dict.keys())} defaulting to "year"', category=Warning, ) return timestamp.strftime(time_res_to_int_dict[temporal_grouping]) -def convert_date_string_to_datetime(temporal_grouping, datestring) -> datetime: +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 `datestring` = '202303', it extracts 2023-03-01 + 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' - datestring : str + date_string : str Date as a string Returns @@ -107,12 +103,10 @@ def convert_date_string_to_datetime(temporal_grouping, datestring) -> datetime: if temporal_grouping not in time_res_dict.keys(): warnings.warn( - 'temporal grouping: {} is not a valid option. Please choose from: {} defaulting to "year"'.format( - temporal_grouping, time_res_dict.keys() - ), + f'temporal grouping: {temporal_grouping} is not a valid option. Please choose from: {list(time_res_dict.keys())} defaulting to "year"', category=Warning, ) - return datetime.strptime(datestring, time_res_dict[temporal_grouping]) + return datetime.strptime(date_string, time_res_dict[temporal_grouping]) def round_datetime(date: datetime, resolution: str) -> datetime: @@ -257,8 +251,8 @@ def plot_characterized_inventory_as_waterfall( 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. + 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 ----------