Skip to content

LCA improvements #17

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SanderNielen merged 3 commits into from
Aug 13, 2026
Merged

LCA improvements #17

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7e6abd4
Skip already existing background processes when
Aug 10, 2026
2a3f9c0
Fix and simplify LCA tests
Aug 11, 2026
0e0c25f
Remove spaces
Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 35 additions & 36 deletions mysite/dpp/lca.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,9 @@ def ensure_methods(family: str):
method_set = IndicatorSet.objects.get(name=family)
except IndicatorSet.DoesNotExist:
logger.debug("Creating a set of environmental indicators.")
method_set = IndicatorSet.objects.create(name=family, start_date=datetime.date.today())
method_set = IndicatorSet.objects.create(
name=family, start_date=datetime.date.today()
)
methods = [
m for m in bwd.methods
if family in m and m[-2:] not in EXCLUDED_METHODS
Expand Down Expand Up @@ -182,7 +184,9 @@ def make_transport_exchange(transports, product, amount):
# Determine the mass of one unit of product (kg)
input_prod = transport.product
if hasattr(input_prod, 'properties'):
mass = input_prod.properties.weight * CONVERSIONS[input_prod.properties.weight_unit]
mass = input_prod.properties.weight * CONVERSIONS[
input_prod.properties.weight_unit
]
elif input_prod.unit in UNIT_CHOICES['Mass']:
mass = CONVERSIONS[input_prod.unit]
else:
Expand All @@ -203,11 +207,20 @@ def convert_dpp_to_brightway(processes: list, db_name: str):
:param db_name: Bightway database name, to add the activity to.
"""
# Import here to avoid circular imports
from .models import ProductExchange, EnvExchange
from .models import ProductExchange, EnvExchange, BackgroundProcess

biosphere = bwd.Database(DEFAULT_REMOTE_PROJECT)
bw_activities = {}
for dpp_process in processes:
# Skip already known background processes
try:
background_process = dpp_process.backgroundprocess
except BackgroundProcess.DoesNotExist:
background_process = None

if background_process and dpp_process.database in bwd.databases:
continue #NOTE: assuming it exists in the bwd database, e.g. ecoinvent

location = str(dpp_process.facility.country) if dpp_process.facility else 'GLO'
transports = dpp_process.functional_flow.productionline.transport
exchanges = [{
Expand All @@ -224,7 +237,7 @@ def convert_dpp_to_brightway(processes: list, db_name: str):
if exc.product.manufacturing_info not in processes:
continue # Cutoff in case max_depth was used.
sign = 1 if exc.direction == 'in' else -1
try:
try: # Find the source DB of background processes
source_db = exc.product.manufacturing_info.database
db_code = exc.product.manufacturing_info.db_code
except AttributeError:
Expand Down Expand Up @@ -262,36 +275,6 @@ def convert_dpp_to_brightway(processes: list, db_name: str):
bw_activities[(db_name, dpp_process.pk)] = activity
return bw_activities

def convert_bw_to_dpp(bw_activity):
# Import here to avoid circular imports
from .models import BackgroundProcess

raise NotImplementedError()
(db_name, code), act = bw_activity
dpp_activity = BackgroundProcess(name=act.name, amount=1, description=act.comment, functional_flow=act.reference_product, database=db_name, db_code=code)
for exchange in act.get('exchanges', []):
if exchange.get('type') == 'technosphere':
dpp_activity.amount = exchange['amount']
else:
pass #TODO: create an exchange
return dpp_activity

def link_to_background_db(activities, background_db): #FIXME: unused
"""
Link DPP processes to ecoinvent or other background DB
only for processes not in DPP system.
"""
for activity in activities:
for exchange in activity.get('exchanges', []):
if exchange.get('type') == 'technosphere':
# If not in foreground, search background
if not exchange.get('input'):
background_match = background_db.search(
exchange['name'], exchange.get('unit')
)
if background_match:
exchange['input'] = background_match

def select_supply_chain(root_product, max_depth=None):
"""
Traverse DPP links to build minimal Brightway database
Expand All @@ -306,10 +289,11 @@ def traverse(flow, depth=0):
visited.add(flow.id)

# Get the ManufacturingProcess for this flow
assert hasattr(flow, 'manufacturing_info'), f"Product {flow} has no manufacturing process!"
assert hasattr(flow, 'manufacturing_info'), (
f"Product {flow} has no manufacturing process!"
)
process = flow.manufacturing_info
processes_to_include.append(process)
# convert_dpp_to_bw_activity(process, db_name)

# Traverse upstream through exchanges
if hasattr(process, 'prod_exchanges'):
Expand Down Expand Up @@ -427,3 +411,18 @@ def create_supply_chain_lca(product):
)

return evaluation

def list_background_processes() -> dict:
"""
Create a dictionary of available background processes
Structured as: {database_name: {activity_name: activity_code}}
"""
setup_project("L4M-DPP")
background = {}
for db_name in bwd.databases:
if db_name[:4] != "dpp_":
background[db_name] = {
act["name"]: act["code"]
for act in bwd.Database(db_name)
}
return background
98 changes: 29 additions & 69 deletions mysite/dpp/tests/test_lca.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from django.test import TestCase
import datetime
from unittest.mock import MagicMock, patch
from ..models import IndicatorSet
from dpp.lca import (
EXCLUDED_METHODS, setup_project, ensure_methods, select_supply_chain,
convert_dpp_to_brightway, lca_calculations, create_supply_chain_lca,
Expand Down Expand Up @@ -63,7 +64,7 @@ def test_create_new_project(self, mock_bwd, mock_bwi):

mock_bwd.projects.set_current.assert_called_once_with("L4M-test")
mock_bwi.remote.install_project.assert_called_once_with(
"ecoinvent-3.10-biosphere", "L4M-test"
"ecoinvent-3.12-biosphere", "L4M-test"
)

@patch("dpp.lca.bwi")
Expand Down Expand Up @@ -95,40 +96,18 @@ def _bwd_methods(self):
def test_creates_indicator_set_when_missing(self, mock_bwd):
mock_bwd.methods = self._bwd_methods()

indicator_set = MagicMock()
indicator_set.name = "EF v3.1"

with patch("dpp.lca.models.IndicatorSet") as MockIS, \
patch("dpp.lca.models.ImpactIndicator") as MockII, \
patch("dpp.lca.models.ImpactCategory") as MockIC:

MockIS.objects.get.side_effect = MockIS.DoesNotExist
MockIS.DoesNotExist = Exception
MockIS.objects.create.return_value = indicator_set
MockIC.objects.get_or_create.return_value = (MagicMock(), True)
MockII.objects.filter.return_value = [] # no existing indicators
result = ensure_methods("EF v3.1")

assert result is indicator_set
MockIS.objects.create.assert_called_once_with(
name="EF v3.1", start_date=datetime.date.today()
)
result = ensure_methods("EF v3.1")
assert result.name == "EF v3.1"
assert result.start_date == datetime.date.today()

@patch("dpp.lca.bwd")
def test_returns_existing_indicator_set(self, mock_bwd):
mock_bwd.methods = self._bwd_methods()
existing_set = MagicMock()

with patch("dpp.lca.models.IndicatorSet") as MockIS, \
patch("dpp.lca.models.ImpactIndicator") as MockII, \
patch("dpp.lca.models.ImpactCategory") as MockIC:

MockIS.objects.get.return_value = existing_set
# Simulate: existing indicators >= methods, so early return
MockII.objects.filter.return_value = [MagicMock(), MagicMock(), MagicMock()]
result = ensure_methods("EF v3.1")

assert result is existing_set
existing_set = IndicatorSet.objects.create(
name="CML2001", start_date=datetime.date(2020,2,2)
)
result = ensure_methods("CML2001")
assert result == existing_set

@patch("dpp.lca.bwd")
def test_excludes_known_zero_impact_methods(self, mock_bwd):
Expand All @@ -144,9 +123,9 @@ def test_excludes_known_zero_impact_methods(self, mock_bwd):
indicator_set = MagicMock()
created_methods = []

with patch("dpp.lca.models.IndicatorSet") as MockIS, \
patch("dpp.lca.models.ImpactIndicator") as MockII, \
patch("dpp.lca.models.ImpactCategory") as MockIC:
with patch("dpp.models.IndicatorSet") as MockIS, \
patch("dpp.models.ImpactIndicator") as MockII, \
patch("dpp.models.ImpactCategory") as MockIC:

MockIS.objects.get.side_effect = MockIS.DoesNotExist
MockIS.DoesNotExist = Exception
Expand Down Expand Up @@ -233,19 +212,15 @@ class TestConvertDppToBrightway(TestCase):
@patch("dpp.lca.bwd")
def test_output_contains_activity_key(self, mock_bwd):
"""Each process must produce a key of the form (db_name, pk)."""
mock_bwd.Database.return_value = iter([]) # empty biosphere

proc = make_process(pk=42, name="my proc", unit="kg")

with patch("dpp.lca.models.ProductExchange") as MockPE, \
patch("dpp.lca.models.EnvExchange") as MockEE:
with patch("dpp.models.ProductExchange") as MockPE, \
patch("dpp.models.EnvExchange") as MockEE:

MockPE.objects.filter.return_value = []
MockEE.objects.filter.return_value = []

# biosphere db search (used by find_biosphere_flow) not called
mock_bwd.Database.return_value.__iter__ = lambda s: iter([])

result = convert_dpp_to_brightway([proc], "testdb")

assert ("testdb", 42) in result
Expand All @@ -255,8 +230,8 @@ def test_production_exchange_is_present(self, mock_bwd):
"""The activity must always have a production exchange."""
proc = make_process(pk=1, unit="kg", amount=2.0)

with patch("dpp.lca.models.ProductExchange") as MockPE, \
patch("dpp.lca.models.EnvExchange") as MockEE:
with patch("dpp.models.ProductExchange") as MockPE, \
patch("dpp.models.EnvExchange") as MockEE:

MockPE.objects.filter.return_value = []
MockEE.objects.filter.return_value = []
Expand All @@ -273,8 +248,8 @@ def test_stage_raw_material_for_mass_unit(self, mock_bwd):
"""Processes with a mass-unit functional flow should be 'Raw material acquisition'."""
proc = make_process(pk=1, unit="kg")

with patch("dpp.lca.models.ProductExchange") as MockPE, \
patch("dpp.lca.models.EnvExchange") as MockEE:
with patch("dpp.models.ProductExchange") as MockPE, \
patch("dpp.models.EnvExchange") as MockEE:
MockPE.objects.filter.return_value = []
MockEE.objects.filter.return_value = []
result = convert_dpp_to_brightway([proc], "db")
Expand All @@ -286,8 +261,8 @@ def test_stage_manufacturing_for_non_resource_unit(self, mock_bwd):
"""Processes with unit 'pcs' (not a resource unit) should be 'Manufacturing'."""
proc = make_process(pk=1, unit="pcs")

with patch("dpp.lca.models.ProductExchange") as MockPE, \
patch("dpp.lca.models.EnvExchange") as MockEE:
with patch("dpp.models.ProductExchange") as MockPE, \
patch("dpp.models.EnvExchange") as MockEE:
MockPE.objects.filter.return_value = []
MockEE.objects.filter.return_value = []
result = convert_dpp_to_brightway([proc], "db")
Expand All @@ -306,8 +281,8 @@ def test_cutoff_excludes_external_products(self, mock_bwd):
pe.product.model.unit = "kg"
pe.product.manufacturing_info = external_proc # not in [proc]

with patch("dpp.lca.models.ProductExchange") as MockPE, \
patch("dpp.lca.models.EnvExchange") as MockEE:
with patch("dpp.models.ProductExchange") as MockPE, \
patch("dpp.models.EnvExchange") as MockEE:
MockPE.objects.filter.return_value = [pe]
MockEE.objects.filter.return_value = []
result = convert_dpp_to_brightway([proc], "db")
Expand All @@ -327,7 +302,6 @@ def test_returns_results_list(self, mock_bwd):
(family, "acidification", "AP"),
]
mock_bwd.methods = {m: {"unit": "kg CO2 eq"} for m in methods}
mock_bwd.methods.__iter__ = lambda s: iter(methods)

lca_obj = MagicMock()
lca_obj.score = 3.14
Expand All @@ -339,20 +313,6 @@ def test_returns_results_list(self, mock_bwd):
assert len(results) == 2
assert all(len(r) == 3 for r in results) # (method, score, unit)

@patch("dpp.lca.bwd")
def test_returns_none_when_no_methods(self, mock_bwd, capsys):
"""When no matching methods are found the function prints a warning and returns None."""
mock_bwd.methods = {}
mock_bwd.methods.__iter__ = lambda s: iter([])

activity = MagicMock()
activity.__getitem__ = lambda self, k: "x"

result = lca_calculations(activity, "NonExistentFamily")
assert result is None
captured = capsys.readouterr()
assert "No" in captured.out

@patch("dpp.lca.bwd")
def test_switch_method_called_for_subsequent_methods(self, mock_bwd):
"""dpp.lca.switch_method should be called for every method after the first."""
Expand Down Expand Up @@ -411,9 +371,9 @@ def test_evaluation_created_for_new_product(self, mock_bwi, mock_bwd, _wrap):
patch("dpp.lca.lca_calculations", return_value=[
(("EF v3.1", "climate change", "GWP100"), 2.0, "kg CO2 eq")
]), \
patch("dpp.lca.models.SustainabilityEvaluation") as MockEval, \
patch("dpp.lca.models.SustainabilityScore") as MockScore, \
patch("dpp.lca.models.ImpactIndicator") as MockII:
patch("dpp.models.SustainabilityEvaluation") as MockEval, \
patch("dpp.models.SustainabilityScore") as MockScore, \
patch("dpp.models.ImpactIndicator") as MockII:

mock_em.return_value = MagicMock(name="method_set")
MockEval.objects.get_or_create.return_value = (MagicMock(), True)
Expand Down Expand Up @@ -441,9 +401,9 @@ def test_scores_updated_when_evaluation_exists(self, mock_bwi, mock_bwd):
patch("dpp.lca.lca_calculations", return_value=[
(("EF v3.1", "climate change", "GWP100"), 2.0, "kg CO2 eq")
]), \
patch("dpp.lca.models.SustainabilityEvaluation") as MockEval, \
patch("dpp.lca.models.SustainabilityScore") as MockScore, \
patch("dpp.lca.models.ImpactIndicator") as MockII:
patch("dpp.models.SustainabilityEvaluation") as MockEval, \
patch("dpp.models.SustainabilityScore") as MockScore, \
patch("dpp.models.ImpactIndicator") as MockII:

mock_em.return_value = MagicMock(name="method_set")
# created=False -> evaluation already existed
Expand Down