Skip to content

Commit

Permalink
harmonize f-strings, fix typos
Browse files Browse the repository at this point in the history
  • Loading branch information
@muelleram
muelleram committed Oct 25, 2024
1 parent 83fd815 commit 687cbfa
Show file tree
Hide file tree
Showing 7 changed files with 182 additions and 129 deletions.
27 changes: 15 additions & 12 deletions bw_timex/dynamic_biosphere_builder.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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__(
Expand All @@ -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
Expand Down Expand Up @@ -104,20 +105,20 @@ 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
absolute temporal distribution (ATD), it looks up the amount from the ATD corresponding
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
----------
Expand All @@ -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():
Expand Down Expand Up @@ -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.
"""
Expand All @@ -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:
Expand All @@ -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()
Expand All @@ -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
----------
Expand Down
47 changes: 29 additions & 18 deletions bw_timex/edge_extractor.py
View file
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
----------
Expand All @@ -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.
"""

Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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)}"
)
16 changes: 7 additions & 9 deletions bw_timex/helper_classes.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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__(
Expand All @@ -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:
Expand Down Expand Up @@ -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]
Expand Down
45 changes: 29 additions & 16 deletions bw_timex/matrix_modifier.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
----------
Expand All @@ -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
----------
Expand All @@ -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
----------
Expand Down Expand Up @@ -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
----------
Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand All @@ -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(
Expand Down
Loading

0 comments on commit 687cbfa

Please sign in to comment.