From 3939ce3a2bc8dea5916092ee06e34865ed7a463d Mon Sep 17 00:00:00 2001 From: Maxime PERALTA Date: Tue, 17 Feb 2026 08:59:23 +0100 Subject: [PATCH 1/2] (update) Support Piecewise in parameter value. --- apparun/expressions.py | 275 +++++++++++++++--- apparun/impact_model.py | 26 +- apparun/impact_tree.py | 2 +- apparun/parameters.py | 186 +----------- apparun/score.py | 3 + apparun/tree_node.py | 3 + .../nvidia_ai_gpu_chip_piecewise.yaml | 105 +++++++ .../test_parameters_values_loading.py | 8 +- 8 files changed, 373 insertions(+), 235 deletions(-) create mode 100644 tests/data/impact_models/nvidia_ai_gpu_chip_piecewise.yaml diff --git a/apparun/expressions.py b/apparun/expressions.py index 9f248b7..ebb66cb 100644 --- a/apparun/expressions.py +++ b/apparun/expressions.py @@ -7,20 +7,21 @@ import math import re from abc import ABC, abstractmethod -from typing import TYPE_CHECKING, Any, Dict, List, Self, Union - -if TYPE_CHECKING: - from apparun.parameters import ImpactModelParams +from collections import defaultdict +from typing import Any, Dict, List, Self, Union import networkx as nx import numpy import sympy +from bw2parameters.errors import ParameterError from pydantic import BaseModel, ValidationError, field_validator, model_validator from pydantic_core import PydanticCustomError from pydantic_core.core_schema import ValidationInfo from sympy import Expr, sympify from apparun.exceptions import InvalidExpr +from apparun.logger import logger +from apparun.parameters import ImpactModelParams def parse_expr(expr: Any) -> Expr: @@ -41,7 +42,7 @@ def validate_expr(expr: str) -> bool: """ Check if an expression is a valid arithmetic expression that can be used in an impact model. Allowed functions inside an - expression are only functions of the modules math and numpy. + expression are only functions of the modules math, numpy and sympy. :param expr: an expression. :returns: True if the expression is a valid arithmetic expression, else False. @@ -52,7 +53,7 @@ def validate_expr(expr: str) -> bool: "NUMBER": r"\-?\d+(\.\d+(e\-?\d+)?)?", "L_PAREN": r"\(", "R_PAREN": r"\)", - "OP": r"\+|\-|\*{1,2}|/{1,2}|%", + "OP": r"\+|\-|\*{1,2}|/{1,2}|%|/|<=?|>=?|={2}", "COMMA": r",", "WS": r"\s+", } @@ -94,7 +95,7 @@ def validate_expr(expr: str) -> bool: valid = False fun_names = re.findall(tokens_patterns["FUN_ID"], expr) - allowed_funcs = dir(math) + dir(numpy) + allowed_funcs = dir(math) + dir(numpy) + dir(sympy) return valid and nb_paren == 0 and all(fun in allowed_funcs for fun in fun_names) @@ -105,6 +106,7 @@ class ParamsValuesSet(BaseModel): """ expressions: Dict[str, ParamExpr] + parameters: ImpactModelParams def __getitem__(self, item): if item not in self.expressions: @@ -151,7 +153,9 @@ def build( ) if errors: raise ValidationError.from_exception_data("", line_errors=errors) - return ParamsValuesSet(**{"expressions": parsed_expressions}) + return ParamsValuesSet( + **{"expressions": parsed_expressions, "parameters": parameters} + ) @property def dependencies_graph(self) -> nx.DiGraph: @@ -160,13 +164,23 @@ def dependencies_graph(self) -> nx.DiGraph: :returns: an oriented graph representing the dependencies between the expressions. """ - return nx.DiGraph( - [ - (name, dep) - for name, expr in self.expressions.items() - for dep in expr.dependencies - ] - ) + graph_nodes = [ + (name, dep) + for name, expr in self.expressions.items() + for dep in expr.dependencies + ] + graph_nodes = [ + ( + self.parameters.find_corresponding_parameter( + param[0], must_find_one=True + ).name, + self.parameters.find_corresponding_parameter( + param[1], must_find_one=True + ).name, + ) + for param in graph_nodes + ] + return nx.DiGraph(graph_nodes) def dependencies_cycle(self) -> List[str]: """ @@ -204,7 +218,10 @@ def evaluate(self) -> Dict[str, Union[float, int, str]]: } values[name] = self.expressions[name].evaluate(deps_values) - + if self.parameters[name].type == "enum": + oh_values = self.parameters[name].transform(values[name]) + for enum_option, oh_value in oh_values.items(): + values[enum_option] = oh_value return values @@ -352,35 +369,29 @@ def validate_dependencies(self, info: ValidationInfo) -> Self: """ parameters = info.context["parameters"] # Check all the dependencies are parameters of the impact model - invalid_deps = sorted(set(self.dependencies) - set(parameters.names)) - if invalid_deps: - raise PydanticCustomError( - "no_such_param", - "No such parameters: {invalid_parameters}", - {"invalid_parameters": tuple(invalid_deps)}, - ) - # Check all the dependencies are float type parameters - non_float_deps = sorted( - [ - dep - for dep in self.dependencies - if dep not in parameters[dep].type != "float" - ] - ) - if non_float_deps: - raise PydanticCustomError( - "dependencies_type", - "Invalid type for the dependencies {invalid_parameters}, expected type {required_type}", - { - "invalid_parameters": tuple(non_float_deps), - "required_type": "float", - }, - ) + for dep in self.dependencies: + try: + parameters.find_corresponding_parameter(dep, must_find_one=True) + except ValueError: + raise PydanticCustomError( + "no_such_param", + "No such parameter: {invalid_parameters}", + {"invalid_parameter": dep}, + ) + if dep in [param.name for param in parameters if param.type == "enum"]: + raise PydanticCustomError( + "dependencies_type", + "Invalid type for the dependency {invalid_parameter}, expected type {required_type}", + { + "invalid_parameter": dep, + "required_type": "float or dummy", + }, + ) return self @property def dependencies(self) -> List[str]: - return re.findall(r"[a-zA-Z_]+\b(?!\()", self.expr) + return [str(symbol) for symbol in parse_expr(self.expr).free_symbols] @property def is_complex(self) -> bool: @@ -519,3 +530,185 @@ def evaluate( return self.options[dependencies_values[self.param]].evaluate( dependencies_values ) + + +class ImpactModelParamsValues(BaseModel): + """ + A set of values for the parameters of an impact model. + + ATTENTION!! Use the method from dict to build instance of this class. + """ + + values: Dict[str, List[Union[float, int, str]]] + + def __getitem__(self, item): + if item in self.values.keys(): + return self.values[item] + else: + raise KeyError() + + @classmethod + def from_dict( + cls, + parameters: ImpactModelParams, + values: Dict[ + str, Union[float, int, str, dict, List[Union[float, int, str, dict]]] + ], + ) -> ImpactModelParamsValues: + # Values with the default values for the parameters not in the values + all_values = { + **values, + **{ + param.name: param.default + for param in parameters + if param.name not in values + }, + } + # Step 1 - Transform all values into lists + empty_list_values = [ + name + for name, value in values.items() + if isinstance(value, list) and len(value) == 0 + ] + if empty_list_values: + raise ValidationError.from_exception_data( + "", + line_errors=[ + { + "loc": ("values",), + "msg": "", + "type": PydanticCustomError( + "empty_list", + "The value for the parameter {parameter} can't be an empty list", + {"parameter": name}, + ), + } + for name in empty_list_values + ], + ) + + list_values = [value for value in values.values() if isinstance(value, list)] + if any( + len(list_values[0]) != len(list_values[i]) + for i in range(1, len(list_values)) + ): + raise ValidationError.from_exception_data( + "", + line_errors=[ + { + "loc": ("values",), + "msg": "", + "type": PydanticCustomError( + "lists_size_match", "List values must have matching sizes" + ), + } + ], + ) + + size = max(map(len, list_values)) if list_values else 1 + list_values = { + name: value if isinstance(value, list) else [value] * size + for name, value in all_values.items() + } + + # Step 2 - Transform the values to expressions + exprs_sets = [] + for idx in range(size): + exprs_sets.append( + ParamsValuesSet.build( + {name: value[idx] for name, value in list_values.items()}, + parameters, + ) + ) + + # Step 3 - Dependencies cycles detection + for exprs_set in exprs_sets: + try: + cycle = exprs_set.dependencies_cycle() + if cycle: + raise ValidationError.from_exception_data( + "", + line_errors=[ + { + "loc": ("values",), + "msg": "", + "type": PydanticCustomError( + "dependencies_cycle", + "The expressions for the parameters {parameters} are inter-dependent", + {"parameters": tuple(sorted(cycle))}, + ), + } + ], + ) + except nx.NetworkXNoCycle: + pass + + # Step 4 - Expressions' evaluation + final_values = defaultdict(list) + for exprs_set in exprs_sets: + evals = exprs_set.evaluate() + for name, value in evals.items(): + # Remove any dummy, if any + if name in parameters.names: + final_values[name].append(value) + + # Step 5 - Validation of the final values + errors = [] + for name, value in final_values.items(): + for idx, elem in enumerate(value): + parameter = parameters[name] + match parameter.type: + case "float": + if parameter.min is None or parameter.max is None: + logger.warning( + f"Parameter {parameter.name} does not have valid bounds. " + f"Consider calling update_bounds()." + ) + elif elem < parameter.min or elem > parameter.max: + if exprs_sets[idx][name].is_complex: + logger.warning( + "The value %s (got after evaluating the expression %s) for the parameter %s is outside its [min, max] range", + str(elem), + name, + str(exprs_sets[idx][name].raw_version), + ) + else: + logger.warning( + "The value %s for the parameter %s is outside its [min, max] range", + str(elem), + name, + ) + case "enum" if elem not in parameter.options: + if exprs_sets[idx][name].is_complex: + errors.append( + { + "type": PydanticCustomError( + "value_error", + "Invalid value {value}, got after evaluating the expression {expr}, for the parameter {target_parameter}", + { + "value": elem, + "target_parameter": name, + "expr": str( + exprs_sets[idx][name].raw_version + ), + }, + ) + } + ) + else: + errors.append( + { + "type": PydanticCustomError( + "value_error", + "Invalid value {value} for the parameter {target_parameter}", + {"value": elem, "target_parameter": name}, + ) + } + ) + if errors: + raise ValidationError.from_exception_data("", line_errors=errors) + + return ImpactModelParamsValues(**{"values": final_values}) + + def items(self): + return self.values.items() diff --git a/apparun/impact_model.py b/apparun/impact_model.py index b0ac81d..fba9cd4 100644 --- a/apparun/impact_model.py +++ b/apparun/impact_model.py @@ -8,9 +8,10 @@ from SALib.analyze import sobol from yaml import YAMLError +from apparun.expressions import ImpactModelParamsValues from apparun.impact_tree import ImpactTreeNode from apparun.logger import logger -from apparun.parameters import ImpactModelParams, ImpactModelParamsValues +from apparun.parameters import ImpactModelParams from apparun.score import LCIAScores from apparun.tree_node import NodeScores @@ -258,6 +259,29 @@ def get_nodes_scores( logger.info("Nodes scores computed with no error") return scores + def get_node_scores( + self, + node_name: str = None, + direct_impacts: Optional[bool] = False, + **params, + ) -> NodeScores: + """ + Get impact scores of one specific node for each impact method, according to the + parameters. + :param node_name: targeted node's name + :param direct_impacts: if True, direct_impacts will be computed instead of + full impacts (i.e. sum of direct impacts and children direct impacts) + :param params: value, or list of values of the impact model's parameters. + List of values must have the same length. If single values are provided + alongside a list of values, it will be duplicated to the appropriate length. + :return: a list of dict mapping impact names and corresponding score, or list + of scores, for each node/property value. + """ + nodes_scores = self.get_nodes_scores(direct_impacts=direct_impacts, **params) + return [ + node_score for node_score in nodes_scores if node_score.name == node_name + ][0] + def get_uncertainty_nodes_scores(self, n) -> List[NodeScores]: """ """ samples = self.parameters.uniform_draw(n) diff --git a/apparun/impact_tree.py b/apparun/impact_tree.py index 342f221..c4ecbda 100644 --- a/apparun/impact_tree.py +++ b/apparun/impact_tree.py @@ -8,7 +8,7 @@ import numpy as np from pydantic import BaseModel, ValidationError, field_validator from pydantic_core import PydanticCustomError -from sympy import Expr, lambdify +from sympy import Expr, Piecewise, lambdify from apparun.exceptions import InvalidExpr from apparun.expressions import parse_expr diff --git a/apparun/parameters.py b/apparun/parameters.py index 4a9462b..6b88cef 100644 --- a/apparun/parameters.py +++ b/apparun/parameters.py @@ -1,7 +1,6 @@ from __future__ import annotations import re -from collections import defaultdict from typing import Dict, List, Optional, Union import networkx as nx @@ -12,9 +11,6 @@ from SALib.sample import sobol from sympy import Expr -from apparun.expressions import ParamsValuesSet -from apparun.logger import logger - class ImpactModelParam(BaseModel): """ @@ -239,7 +235,7 @@ def draw_to_distrib(self, samples: np.ndarray) -> List[str]: return transformed_samples def corresponds(self, symbol_name: str) -> bool: - return symbol_name in self.dummies_names + return symbol_name in self.dummies_names + [self.name] class ImpactModelParams(BaseModel): @@ -386,183 +382,3 @@ def draw_to_distrib( self.parameters[i].name: self.parameters[i].draw_to_distrib(samples[:, i]) for i in range(len(self.parameters)) } - - -class ImpactModelParamsValues(BaseModel): - """ - A set of values for the parameters of an impact model. - - ATTENTION!! Use the method from dict to build instance of this class. - """ - - values: Dict[str, List[Union[float, int, str]]] - - def __getitem__(self, item): - if item in self.values.keys(): - return self.values[item] - else: - raise KeyError() - - @classmethod - def from_dict( - cls, - parameters: ImpactModelParams, - values: Dict[ - str, Union[float, int, str, dict, List[Union[float, int, str, dict]]] - ], - ) -> ImpactModelParamsValues: - # Values with the default values for the parameters not in the values - all_values = { - **values, - **{ - param.name: param.default - for param in parameters - if param.name not in values - }, - } - # Step 1 - Transform all values into lists - empty_list_values = [ - name - for name, value in values.items() - if isinstance(value, list) and len(value) == 0 - ] - if empty_list_values: - raise ValidationError.from_exception_data( - "", - line_errors=[ - { - "loc": ("values",), - "msg": "", - "type": PydanticCustomError( - "empty_list", - "The value for the parameter {parameter} can't be an empty list", - {"parameter": name}, - ), - } - for name in empty_list_values - ], - ) - - list_values = [value for value in values.values() if isinstance(value, list)] - if any( - len(list_values[0]) != len(list_values[i]) - for i in range(1, len(list_values)) - ): - raise ValidationError.from_exception_data( - "", - line_errors=[ - { - "loc": ("values",), - "msg": "", - "type": PydanticCustomError( - "lists_size_match", "List values must have matching sizes" - ), - } - ], - ) - - size = max(map(len, list_values)) if list_values else 1 - list_values = { - name: value if isinstance(value, list) else [value] * size - for name, value in all_values.items() - } - - # Step 2 - Transform the values to expressions - exprs_sets = [] - for idx in range(size): - exprs_sets.append( - ParamsValuesSet.build( - {name: value[idx] for name, value in list_values.items()}, - parameters, - ) - ) - - # Step 3 - Dependencies cycles detection - for exprs_set in exprs_sets: - try: - cycle = exprs_set.dependencies_cycle() - if cycle: - raise ValidationError.from_exception_data( - "", - line_errors=[ - { - "loc": ("values",), - "msg": "", - "type": PydanticCustomError( - "dependencies_cycle", - "The expressions for the parameters {parameters} are inter-dependent", - {"parameters": tuple(sorted(cycle))}, - ), - } - ], - ) - except nx.NetworkXNoCycle: - pass - - # Step 4 - Expressions' evaluation - final_values = defaultdict(list) - for exprs_set in exprs_sets: - evals = exprs_set.evaluate() - for name, value in evals.items(): - final_values[name].append(value) - - # Step 5 - Validation of the final values - errors = [] - for name, value in final_values.items(): - for idx, elem in enumerate(value): - parameter = parameters[name] - match parameter.type: - case "float": - if parameter.min is None or parameter.max is None: - logger.warning( - f"Parameter {parameter.name} does not have valid bounds. " - f"Consider calling update_bounds()." - ) - elif elem < parameter.min or elem > parameter.max: - if exprs_sets[idx][name].is_complex: - logger.warning( - "The value %s (got after evaluating the expression %s) for the parameter %s is outside its [min, max] range", - str(elem), - name, - str(exprs_sets[idx][name].raw_version), - ) - else: - logger.warning( - "The value %s for the parameter %s is outside its [min, max] range", - str(elem), - name, - ) - case "enum" if elem not in parameter.options: - if exprs_sets[idx][name].is_complex: - errors.append( - { - "type": PydanticCustomError( - "value_error", - "Invalid value {value}, got after evaluating the expression {expr}, for the parameter {target_parameter}", - { - "value": elem, - "target_parameter": name, - "expr": str( - exprs_sets[idx][name].raw_version - ), - }, - ) - } - ) - else: - errors.append( - { - "type": PydanticCustomError( - "value_error", - "Invalid value {value} for the parameter {target_parameter}", - {"value": elem, "target_parameter": name}, - ) - } - ) - if errors: - raise ValidationError.from_exception_data("", line_errors=errors) - - return ImpactModelParamsValues(**{"values": final_values}) - - def items(self): - return self.values.items() diff --git a/apparun/score.py b/apparun/score.py index b8deb84..f2f23d8 100644 --- a/apparun/score.py +++ b/apparun/score.py @@ -25,6 +25,9 @@ def method_names(self) -> Set[str]: """ return set(self.scores.keys()) + def __getitem__(self, method_name: str): + return self.scores[method_name] + def to_unpivoted_df(self) -> pd.DataFrame: if isinstance(list(self.scores.values())[0], float) or isinstance( list(self.scores.values())[0], float diff --git a/apparun/tree_node.py b/apparun/tree_node.py index 82be4e2..311ff56 100644 --- a/apparun/tree_node.py +++ b/apparun/tree_node.py @@ -116,6 +116,9 @@ def full_to_direct_impacts(node_scores: List[NodeScores]) -> NodeScores: direct_impact_scores.append(direct_impact_score) return direct_impact_scores + def __getitem__(self, method_name): + return self.lcia_scores[method_name] + def to_unpivoted_df(self) -> pd.DataFrame: df = self.lcia_scores.to_unpivoted_df() df["name"] = self.name diff --git a/tests/data/impact_models/nvidia_ai_gpu_chip_piecewise.yaml b/tests/data/impact_models/nvidia_ai_gpu_chip_piecewise.yaml new file mode 100644 index 0000000..d81313f --- /dev/null +++ b/tests/data/impact_models/nvidia_ai_gpu_chip_piecewise.yaml @@ -0,0 +1,105 @@ +metadata: + author: + name: Maxime PERALTA + organization: CEA + mail: maxime.peralta@cea.fr + reviewer: + name: Mathias TORCASO + organization: CEA + mail: null + report: + link: https://appalca.github.io/ + description: A mock example of Appa LCA's impact model corresponding to a fictive + AI chip accelerator based on NVIDIA GPU. + date: 07/10/2025 + version: '1' + license: proprietary + appabuild_version: 0.3.6 +parameters: +- name: architecture + default: Maxwell + type: enum + weights: + Maxwell: 1.0 + Pascal: 1.0 +- name: cuda_core + default: + architecture: + Maxwell: 1344 + Pascal: 1280 + type: float + min: 256.0 + max: 4096.0 + distrib: linear + pm: null + pm_perc: null +- name: energy_per_inference + default: + architecture: + Maxwell: 0.0878*cuda_core*1000/(90*3600) + Pascal: 0.0679*cuda_core*1000/(90*3600) + type: float + min: 0.06 + max: 1.11 + distrib: linear + pm: null + pm_perc: 0.2 +- name: inference_per_day + default: 30*3600*8 + type: float + min: 0.0 + max: 86400000.0 + distrib: linear + pm: null + pm_perc: null +- name: lifespan + default: Piecewise((2, cuda_core > 1300)(3, True)) + type: float + min: 2 + max: 2 + distrib: linear + pm: 1.0 + pm_perc: null +- name: usage_location + default: FR + type: enum + weights: + FR: 1.0 + EU: 1.0 +tree: + name: nvidia_ai_gpu_chip + models: + EFV3_CLIMATE_CHANGE: 1.0*energy_per_inference*inference_per_day*lifespan*(7.560675e-5*usage_location_EU + + 1.7532e-5*usage_location_FR) - 12500.0*architecture_Maxwell*(-4.6212599075297227e-9*cuda_core + - 7.37132179656539e-6) + 1.90904456522695e-5*architecture_Maxwell*(0.1889809692866578*cuda_core + + 19.47688243064738)**2/((1 - 0.9980542072708582*exp(-1.889809692866578e-5*cuda_core))**2*(70685.775/(0.1889809692866578*cuda_core + + 19.47688243064738) - 471.2385*sqrt(2)/sqrt(0.1889809692866578*cuda_core + + 19.47688243064738))) - 12500.0*architecture_Pascal*(-4.6891975579761074e-9*cuda_core + - 7.808281424221127e-6) + 0.00055747322513291*architecture_Pascal*(0.13184623155305694*cuda_core + + 21.707425626610416)**2/((1 - 0.98920497620445418*exp(-6.5923115776528471e-5*cuda_core))**2*(70685.775/(0.13184623155305694*cuda_core + + 21.707425626610416) - 471.2385*sqrt(2)/sqrt(0.13184623155305694*cuda_core + + 21.707425626610416))) + children: + - name: ai_use_phase + models: + EFV3_CLIMATE_CHANGE: 1.0*energy_per_inference*inference_per_day*lifespan*(7.560675e-5*usage_location_EU + + 1.7532e-5*usage_location_FR) + children: [] + properties: {"phase": "use"} + amount: '1.0' + - name: nvidia_gpu_chip_manufacturing + models: + EFV3_CLIMATE_CHANGE: -12500.0*architecture_Maxwell*(-4.6212599075297227e-9*cuda_core + - 7.37132179656539e-6) + 1.90904456522695e-5*architecture_Maxwell*(0.1889809692866578*cuda_core + + 19.47688243064738)**2/((1 - 0.9980542072708582*exp(-1.889809692866578e-5*cuda_core))**2*(70685.775/(0.1889809692866578*cuda_core + + 19.47688243064738) - 471.2385*sqrt(2)/sqrt(0.1889809692866578*cuda_core + + 19.47688243064738))) - 12500.0*architecture_Pascal*(-4.6891975579761074e-9*cuda_core + - 7.808281424221127e-6) + 0.00055747322513291*architecture_Pascal*(0.13184623155305694*cuda_core + + 21.707425626610416)**2/((1 - 0.98920497620445418*exp(-6.5923115776528471e-5*cuda_core))**2*(70685.775/(0.13184623155305694*cuda_core + + 21.707425626610416) - 471.2385*sqrt(2)/sqrt(0.13184623155305694*cuda_core + + 21.707425626610416))) + children: [] + properties: {"phase": "manufacturing"} + amount: '1.0' + properties: {} + amount: '1.0' diff --git a/tests/functional/test_parameters_values_loading.py b/tests/functional/test_parameters_values_loading.py index 68a06b0..2cec6cd 100644 --- a/tests/functional/test_parameters_values_loading.py +++ b/tests/functional/test_parameters_values_loading.py @@ -53,8 +53,6 @@ def test_float_expr_no_such_param(impact_model): "cuda_core": "surface / unit_per_mm", } - expected = [("surface",), ("surface", "unit_per_mm")] - with pytest.raises(ValidationError) as exc_info: impact_model.params_values(**parameters) @@ -64,7 +62,6 @@ def test_float_expr_no_such_param(impact_model): for idx, err in enumerate(errors): assert err["type"] == "no_such_param" assert err["ctx"]["target_parameter"] == list(parameters.keys())[idx] - assert err["ctx"]["invalid_parameters"] == expected[idx] assert err["input"] == list(parameters.values())[idx] @@ -78,8 +75,6 @@ def test_float_expr_dependencies_type(impact_model): "cuda_core": "energy_per_inference / architecture", } - expected = [("architecture", "usage_location"), ("architecture",)] - with pytest.raises(ValidationError) as exc_info: impact_model.params_values(**parameters) @@ -89,8 +84,7 @@ def test_float_expr_dependencies_type(impact_model): for idx, err in enumerate(errors): assert err["type"] == "dependencies_type" assert err["ctx"]["target_parameter"] == list(parameters.keys())[idx] - assert err["ctx"]["invalid_parameters"] == expected[idx] - assert err["ctx"]["required_type"] == "float" + assert err["ctx"]["required_type"] == "float or dummy" assert err["input"] == list(parameters.values())[idx] From f876e1cf5f79dd6b8765498b2077c0d654a218bd Mon Sep 17 00:00:00 2001 From: Maxime PERALTA Date: Wed, 18 Feb 2026 15:06:05 +0100 Subject: [PATCH 2/2] (fix) Remove exception defined in appabuild. Add a new test for piecewise parameter usage --- apparun/expressions.py | 1 - .../nvidia_ai_gpu_chip_piecewise.yaml | 36 ++++++++++--------- tests/end_to_end/test_piecewise.py | 34 ++++++++++++++++++ 3 files changed, 54 insertions(+), 17 deletions(-) create mode 100644 tests/end_to_end/test_piecewise.py diff --git a/apparun/expressions.py b/apparun/expressions.py index ebb66cb..03d6ae2 100644 --- a/apparun/expressions.py +++ b/apparun/expressions.py @@ -13,7 +13,6 @@ import networkx as nx import numpy import sympy -from bw2parameters.errors import ParameterError from pydantic import BaseModel, ValidationError, field_validator, model_validator from pydantic_core import PydanticCustomError from pydantic_core.core_schema import ValidationInfo diff --git a/tests/data/impact_models/nvidia_ai_gpu_chip_piecewise.yaml b/tests/data/impact_models/nvidia_ai_gpu_chip_piecewise.yaml index d81313f..6549298 100644 --- a/tests/data/impact_models/nvidia_ai_gpu_chip_piecewise.yaml +++ b/tests/data/impact_models/nvidia_ai_gpu_chip_piecewise.yaml @@ -36,11 +36,11 @@ parameters: - name: energy_per_inference default: architecture: - Maxwell: 0.0878*cuda_core*1000/(90*3600) + Maxwell: Piecewise((0, cuda_core > 1300), (0.1*cuda_core*1000/(90*3600), True)) Pascal: 0.0679*cuda_core*1000/(90*3600) type: float - min: 0.06 - max: 1.11 + min: null + max: null distrib: linear pm: null pm_perc: 0.2 @@ -53,10 +53,10 @@ parameters: pm: null pm_perc: null - name: lifespan - default: Piecewise((2, cuda_core > 1300)(3, True)) + default: Piecewise((2, cuda_core > 1300), (0, architecture_Pascal), (5, True)) type: float - min: 2 - max: 2 + min: null + max: null distrib: linear pm: 1.0 pm_perc: null @@ -67,16 +67,19 @@ parameters: FR: 1.0 EU: 1.0 tree: - name: nvidia_ai_gpu_chip + name: nvidia_ai_gpu_chip_piecewise models: EFV3_CLIMATE_CHANGE: 1.0*energy_per_inference*inference_per_day*lifespan*(7.560675e-5*usage_location_EU + 1.7532e-5*usage_location_FR) - 12500.0*architecture_Maxwell*(-4.6212599075297227e-9*cuda_core - - 7.37132179656539e-6) + 1.90904456522695e-5*architecture_Maxwell*(0.1889809692866578*cuda_core - + 19.47688243064738)**2/((1 - 0.9980542072708582*exp(-1.889809692866578e-5*cuda_core))**2*(70685.775/(0.1889809692866578*cuda_core + - 7.37132179656539e-6)*Piecewise((1, lifespan < 3), (2, lifespan < 4), (0, True)) + + 1.90904456522695e-5*architecture_Maxwell*(0.1889809692866578*cuda_core + 19.47688243064738)**2*Piecewise((1, + lifespan < 3), (2, lifespan < 4), (0, True))/((1 - 0.9980542072708582*exp(-1.889809692866578e-5*cuda_core))**2*(70685.775/(0.1889809692866578*cuda_core + 19.47688243064738) - 471.2385*sqrt(2)/sqrt(0.1889809692866578*cuda_core + 19.47688243064738))) - 12500.0*architecture_Pascal*(-4.6891975579761074e-9*cuda_core - - 7.808281424221127e-6) + 0.00055747322513291*architecture_Pascal*(0.13184623155305694*cuda_core - + 21.707425626610416)**2/((1 - 0.98920497620445418*exp(-6.5923115776528471e-5*cuda_core))**2*(70685.775/(0.13184623155305694*cuda_core + - 7.808281424221127e-6)*Piecewise((1, lifespan < 3), (2, lifespan < 4), (0, + True)) + 0.00055747322513291*architecture_Pascal*(0.13184623155305694*cuda_core + + 21.707425626610416)**2*Piecewise((1, lifespan < 3), (2, lifespan < 4), (0, + True))/((1 - 0.98920497620445418*exp(-6.5923115776528471e-5*cuda_core))**2*(70685.775/(0.13184623155305694*cuda_core + 21.707425626610416) - 471.2385*sqrt(2)/sqrt(0.13184623155305694*cuda_core + 21.707425626610416))) children: @@ -85,11 +88,11 @@ tree: EFV3_CLIMATE_CHANGE: 1.0*energy_per_inference*inference_per_day*lifespan*(7.560675e-5*usage_location_EU + 1.7532e-5*usage_location_FR) children: [] - properties: {"phase": "use"} + properties: {} amount: '1.0' - name: nvidia_gpu_chip_manufacturing models: - EFV3_CLIMATE_CHANGE: -12500.0*architecture_Maxwell*(-4.6212599075297227e-9*cuda_core + EFV3_CLIMATE_CHANGE: 1.0*(-12500.0*architecture_Maxwell*(-4.6212599075297227e-9*cuda_core - 7.37132179656539e-6) + 1.90904456522695e-5*architecture_Maxwell*(0.1889809692866578*cuda_core + 19.47688243064738)**2/((1 - 0.9980542072708582*exp(-1.889809692866578e-5*cuda_core))**2*(70685.775/(0.1889809692866578*cuda_core + 19.47688243064738) - 471.2385*sqrt(2)/sqrt(0.1889809692866578*cuda_core @@ -97,9 +100,10 @@ tree: - 7.808281424221127e-6) + 0.00055747322513291*architecture_Pascal*(0.13184623155305694*cuda_core + 21.707425626610416)**2/((1 - 0.98920497620445418*exp(-6.5923115776528471e-5*cuda_core))**2*(70685.775/(0.13184623155305694*cuda_core + 21.707425626610416) - 471.2385*sqrt(2)/sqrt(0.13184623155305694*cuda_core - + 21.707425626610416))) + + 21.707425626610416))))*Piecewise((1, lifespan < 3), (2, lifespan < 4), (0, + True)) children: [] - properties: {"phase": "manufacturing"} - amount: '1.0' + properties: {} + amount: Piecewise((1, lifespan < 3), (2, lifespan < 4), (0, True)) properties: {} amount: '1.0' diff --git a/tests/end_to_end/test_piecewise.py b/tests/end_to_end/test_piecewise.py new file mode 100644 index 0000000..1543a25 --- /dev/null +++ b/tests/end_to_end/test_piecewise.py @@ -0,0 +1,34 @@ +""" +This module contains the tests related to Piecewise parameters. +""" +import os + +from apparun.impact_model import ImpactModel +from tests import DATA_DIR + + +def test_piecewise_in_parameter_value(): + """ + Assert that a model can be parameterised using Piecewise functions, including with + dummies. + """ + model_path = os.path.join( + DATA_DIR, "impact_models/nvidia_ai_gpu_chip_piecewise.yaml" + ) + model = ImpactModel.from_yaml(model_path) + assert ( + model.get_node_scores( + node_name="nvidia_gpu_chip_manufacturing", + cuda_core="Piecewise((1000, architecture_Pascal), (100, True))", + architecture="Maxwell", + )["EFV3_CLIMATE_CHANGE"][0] + == 0 + ) + assert ( + model.get_node_scores( + node_name="nvidia_gpu_chip_manufacturing", + cuda_core="Piecewise((1000, architecture_Pascal), (100, True))", + architecture="Pascal", + )["EFV3_CLIMATE_CHANGE"][0] + != 0 + )