diff --git a/README.md b/README.md index 16467fd..5212d5c 100644 --- a/README.md +++ b/README.md @@ -349,3 +349,89 @@ A DPP is uniquely identified by its registration number. It can be accessed thro 'update_interval': 'A', } ``` + +# Decision Support System manual +The Sustainability and Cost Module features a Decision Support System (DSS) that can be accessed through an API. It is specifically designed for supporting decisions in laser welding. The DSS has a user interface to guide you through the process of setting decision preferences. + +## API access +To receive decision support for comparing different laser welding parameters, send a post request to [main-domain.com/dss/experiments/]. The request should contain a list of experiments with the following structure: +```JSON +[ + { + "experimentId": 12, + "weldLength": 29, + "weldSpeed": 100, + "country": "DE", + "laserPowerkW": 4, + "weldingStationPowerkW": 12, + "consumables": [ + { + "name": "Nitrogen", + "flowRate": 15.2, + "unit": "m3/h" + }, + { + "name": "Argon", + "flowRate": 6, + "unit": "m3/h" + }, + { + "name": "Aluminium (filler wire)", + "flowRate": 0.02, + "unit": "kg/h" + } + ], + "qualityParameters": [ + { + "name": "Porosity", + "value": 3.2, + "target": "min" + }, + { + "name": "Tensile strength", + "value": 0.18, + "target": "max" + }, + { + "name": "Weld depth", + "value": 3, + "target": 3.5 + } + ] + } +] +``` + + +## Criteria calculations + +Calculations for deriving the KPIs used as decision criteria in the DSS, for the Process Developer. These calculations are implemented in the function calculate_kpis, in script views.py. +- Amount of shielding gas = gas flow rate × processing time +- Amount of filler wire = filler wire use rate × processing time +- Consumables costs = Σ amount × price (for all consumables) +- Labour costs = processing time × 3 × wages (in the specific country) +- Process energy use = processing time × (laser power + welding station power) +- Energy costs = process energy use × electricity price (in the specific country) +- Operating costs = consumables costs + labour costs + energy costs +- Consumables footprint = Σ amount × material carbon footprint (for all consumables) +- Process carbon footprint = consumables footprint + process energy use × electricity footprint (in the specific country) +- The technical quality indicators are directly used without calculations. + +Calculations for deriving KPIs for the Key Account Manager. This is not implemented yet (July 2026). +- Amount of shielding gas = gas flow rate × processing time +- Amount of filler wire = filler wire use rate × processing time +- Consumables costs = Σ amount × price (for all consumables) +- Labour costs = cycle time × 3 × wages (in the specific country) +- Process energy use = processing time × laser power + cycle time × welding station power +- Energy costs = process energy use × electricity price (in the specific country) +- Operating costs = consumables costs + labour costs + energy costs + maintenance costs +- Materials costs = Σ material weight × material price / (1 – scrap rate) (for all parts) +- Service life = 15 years (fixed value) +- Annual production = 240 days/year × 16 hours/day × 3600 seconds/hour / cycle time +- Amortized CAPEX = investment costs / (service life × annual production) +- Total production costs = operating costs + materials costs + amortized CAPEX +- Consumables footprint = Σ amount × material carbon footprint (for all consumables) +- Process carbon footprint = consumables footprint + process energy use × electricity footprint (in the specific country) +- Materials footprint = Σ material weight × material carbon footprint / (1 – scrap rate) (for all parts) +- Product carbon footprint = process carbon footprint + materials footprint +- The remaining KPIs are directly used without calculations: recyclability, process monitoring, process automation level, operator specialization level, production lead time, machine saturation, technical quality indicators. diff --git a/mysite/dpp/migrations/0041_alter_transport_mode.py b/mysite/dpp/migrations/0041_alter_transport_mode.py new file mode 100644 index 0000000..aa3366c --- /dev/null +++ b/mysite/dpp/migrations/0041_alter_transport_mode.py @@ -0,0 +1,18 @@ +# Generated by Django 6.0.5 on 2026-07-08 11:40 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('dpp', '0040_productionline_created_by_user'), + ] + + operations = [ + migrations.AlterField( + model_name='transport', + name='mode', + field=models.CharField(choices=[('train', 'Freight train'), ('ocean ship', 'Sea freight container ship'), ('truck', 'Lorry / truck, average'), ('inland ship', 'Inland waterway ship'), ('airplane', 'Aircraft'), ('delivery van', 'Light commercial vehicle (delivery van)'), ('NA', 'Unspecified')], default='NA', max_length=12, verbose_name='Main mode of transport'), + ), + ] diff --git a/mysite/dss/__init__.py b/mysite/dss/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/mysite/dss/admin.py b/mysite/dss/admin.py new file mode 100644 index 0000000..8c38f3f --- /dev/null +++ b/mysite/dss/admin.py @@ -0,0 +1,3 @@ +from django.contrib import admin + +# Register your models here. diff --git a/mysite/dss/apps.py b/mysite/dss/apps.py new file mode 100644 index 0000000..89d0f90 --- /dev/null +++ b/mysite/dss/apps.py @@ -0,0 +1,5 @@ +from django.apps import AppConfig + + +class DssConfig(AppConfig): + name = 'dss' diff --git a/mysite/dss/forms.py b/mysite/dss/forms.py new file mode 100644 index 0000000..4b2cb3a --- /dev/null +++ b/mysite/dss/forms.py @@ -0,0 +1,447 @@ +from django import forms +from django.forms import inlineformset_factory +from .models import McdaSession, Criterion, CritGroup, GroupOrder, LocalOrder, DecisionMatrix + + +# ------------------------------------------------------------------ +# Custom fields +# ------------------------------------------------------------------ + +class WeightField(forms.Field): + """ + Accepts a single positive float ("0.5") or a range ("0.2, 0.8"). + Returns float | tuple[float, float]. + """ + widget = forms.TextInput(attrs={"placeholder": "e.g. 0.5 or 0.2, 0.8"}) + + def to_python(self, value): + if not value: + return None + value = value.strip() + + if "," in value: + parts = value.split(",") + if len(parts) != 2: + raise forms.ValidationError("Range must be exactly two values.") + try: + lo, hi = float(parts[0].strip()), float(parts[1].strip()) + except ValueError: + raise forms.ValidationError("Range values must be numbers.") + if lo < 0 or hi < 0: + raise forms.ValidationError("Values must be positive.") + if lo > hi: + raise forms.ValidationError("min must be ≤ max.") + return (lo, hi) + + try: + v = float(value) + except ValueError: + raise forms.ValidationError("Must be a positive number or range 'min, max'.") + if v < 0: + raise forms.ValidationError("Value must be positive.") + return v + + +class PositiveFloatField(forms.FloatField): + def validate(self, value): + super().validate(value) + if value is not None and value < 0: + raise forms.ValidationError("Value must be positive.") + + +class OrderingEntryField(forms.Field): + """ + Represents a single ordering constraint: "A > B, 0.8" + Returns tuple[str, str, float]. + Rendered as a set of rows via OrderingWidget (see below). + """ + def to_python(self, value): + if not value: + return None + try: + parts = [p.strip() for p in value.split(",")] + if len(parts) != 3: + raise ValueError + a, b, intensity = parts[0], parts[1], float(parts[2]) + if intensity < 0: + raise forms.ValidationError("Strength must be positive.") + return (a, b, intensity) + except ValueError: + raise forms.ValidationError("Format must be 'A, B, intensity'.") + +# ------------------------------------------------------------------ +# Form 0: criteria selection (always shown) +# ------------------------------------------------------------------ + +class KpiSelectionForm(forms.Form): + def __init__(self, session: McdaSession, *args, **kwargs): + super().__init__(*args, **kwargs) + + self.groups = ( + CritGroup.objects.filter(session=session).prefetch_related("criteria") + ) + + for group in self.groups: + criteria = group.criteria.all() + field_name = f"group_{group.pk}" + + # Group checkbox + self.fields[field_name] = forms.BooleanField( + required=False, + initial=all(c.used for c in criteria), + label=group.name, + ) + group.field = self[field_name] + + # Criterion checkboxes + for criterion in criteria: + field_name = f"criterion_{criterion.pk}" + self.fields[field_name] = ( + forms.BooleanField( + required=False, + initial=criterion.used, + label=criterion.name, + ) + ) + criterion.field = self[field_name] + + # Find KPI values that are missing in the input + self.alternatives = session.alternatives.all() + self.value_rows = [] + + for criterion in session.criteria: + row = {"criterion": criterion, "cells": []} + has_missing = False + + for alternative in self.alternatives: + if criterion.name in alternative.values: + row["cells"].append(None) + else: + has_missing = True + field_name = (f"value_{criterion.pk}_{alternative.pk}") + self.fields[field_name] = forms.FloatField() + row["cells"].append(self[field_name]) + + if has_missing: + self.value_rows.append(row) + + def save(self, *args): + criteria = [] + + for name, value in self.cleaned_data.items(): + if name.startswith("criterion_"): + pk = int(name.split("_")[1]) + criteria.append((pk, value)) + + criterion_map = Criterion.objects.in_bulk(pk for pk, _ in criteria) + for pk, used in criteria: + criterion = criterion_map[pk] + criterion.used = used + + Criterion.objects.bulk_update(criterion_map.values(), ["used"]) + + # Add newly provided KPI values to DecisionMatrix.values + alternatives = {alt.pk: alt for alt in self.session.alternatives.all()} + + for name, value in self.cleaned_data.items(): + if not name.startswith("value_"): + continue + + _, criterion_id, alternative_id = name.split("_") + alternative = alternatives[int(alternative_id)] + criterion = Criterion.objects.get(pk=int(criterion_id)) + alternative.values[criterion.name] = value + + # Save the changes made to all alternative.values + DecisionMatrix.objects.bulk_update(alternatives.values(), ["values"]) + +# ------------------------------------------------------------------ +# Form 1: basic settings (always shown) +# ------------------------------------------------------------------ + +class BaseConfigForm(forms.Form): + """ + Collects the top-level choices that drive which fields appear in Forms 2 and 3. + Saved to: session.scenario, .method, .weight_mode, .veto_type + """ + scenario = forms.ChoiceField( + choices=McdaSession.Scenario.choices, widget=forms.RadioSelect + ) + method = forms.ChoiceField( + choices=McdaSession.Method.choices, widget=forms.RadioSelect + ) + weight_mode = forms.ChoiceField( + choices=McdaSession.WeightMode.choices, widget=forms.RadioSelect + ) + sampling_mode = forms.ChoiceField( + choices=McdaSession.SamplingMode.choices, widget=forms.RadioSelect + ) + veto_type = forms.ChoiceField( + choices=McdaSession.VetoType.choices, widget=forms.RadioSelect + ) + + def clean(self): + cleaned = super().clean() + if cleaned.get("scenario") == "uncertain" and not cleaned.get("sampling_mode"): + self.add_error("sampling_mode", "Required for uncertain analysis.") + return cleaned + + def save(self, session: McdaSession) -> None: + data = self.cleaned_data + session.scenario = data["scenario"] + session.method = data["method"] + session.weight_mode = data["weight_mode"] + session.sampling_mode = data["sampling_mode"] + session.veto_type = data["veto_type"] + session.save() + + +# ------------------------------------------------------------------ +# Form 2: conditional fields, dynamically built from session state +# ------------------------------------------------------------------ + +class WeightsThresholdsForm(forms.Form): + """ + Fields are added dynamically in __init__ based on Form 1 choices. + All fields are optional at the Django level; required-ness is + enforced in clean() based on the session context. + + Saved to: session.group_weights, .local_weights, .thresholds, + .veto_thresholds, .penalty_factor, .sampling_mode + """ + + def __init__(self, session: McdaSession, *args, **kwargs): + super().__init__(*args, **kwargs) + self.session = session + criteria = list(session.criteria.values_list("name", flat=True)) + self.groups = list(session.groups.values_list("name", flat=True)) + + if session.sampling_mode != "random": + # -- group_weights: weight_mode in {group, hierarchical} ----- + if session.weight_mode in ("group", "hierarchical"): + for group in self.groups: + self.fields[f"group_weight_{group}"] = WeightField( + label=f"Weight — {group.title()}", required=False + ) + + # -- local_weights: weight_mode in hierarchical, flat -------- + if session.weight_mode in ("hierarchical", "flat"): + for criterion in criteria: + self.fields[f"local_weight_{criterion}"] = WeightField( + label=f"Criterion weight — {criterion}", required=False + ) + + # -- thresholds: always, shape depends on method -------------- + for criterion in criteria: + if session.method == "promethee": + self.fields[f"threshold_{criterion}"] = WeightField( + label=f"Threshold — {criterion} (indifference, preference)", + required=False, + ) + else: + self.fields[f"threshold_{criterion}"] = PositiveFloatField( + label=f"Threshold — {criterion}", required=False + ) + + # -- veto_thresholds: veto_type != "no" ----------------------- + if session.veto_type != "no": + for criterion in criteria: + self.fields[f"veto_{criterion}"] = PositiveFloatField( + label=f"Veto threshold — {criterion}", required=False + ) + + # -- penalty_factor: veto_type == "soft" ---------------------- + if session.veto_type == "soft": + self.fields["penalty_factor"] = PositiveFloatField( + label="Penalty factor", required=False + ) + + def clean(self): + cleaned = super().clean() + session = self.session + + if session.sampling_mode != "random": + if session.weight_mode in ("group", "hierarchical"): + for group in self.groups: + if not cleaned.get(f"group_weight_{group}"): + self.add_error( + f"group_weight_{group}", + "Required when weight mode is group or hierarchical." + ) + if session.weight_mode in ("hierarchical", "flat"): + for name in session.criteria.values_list("name", flat=True): + if not cleaned.get(f"local_weight_{name}"): + self.add_error( + f"local_weight_{name}", + "Required for hierarchical weighting." + ) + + if session.veto_type != "no": + for name in session.criteria.values_list("name", flat=True): + if cleaned.get(f"veto_{name}") is None: + self.add_error(f"veto_{name}", "Required when veto is active.") + + if session.veto_type == "soft" and cleaned.get("penalty_factor") is None: + self.add_error("penalty_factor", "Required for soft veto.") + + return cleaned + + def save(self, session: McdaSession) -> None: + d = self.cleaned_data + criteria = list(session.criteria.values_list("name", flat=True)) + + if session.weight_mode in ("group", "hierarchical"): + session.group_weights = {g: d[f"group_weight_{g}"] for g in self.groups} + + if session.weight_mode in ("hierarchical", "flat"): + session.local_weights = {c: d[f"local_weight_{c}"] for c in criteria} + + session.thresholds = {c: d[f"threshold_{c}"] for c in criteria} + + if session.veto_type != "no": + session.veto_thresholds = {c: d[f"veto_{c}"] for c in criteria} + + if session.veto_type == "soft": + session.penalty_factor = d["penalty_factor"] + + session.save() + + +# ------------------------------------------------------------------ +# Form 3: sampling & uncertainty parameters +# ------------------------------------------------------------------ + +class SamplingConfigForm(forms.Form): + """ + Only shown when scenario == "uncertain". + Fields are added dynamically based on scenario, weight_mode, and sampling_mode. + + Saved to: session.n_samples, .alpha, .alpha_group, .alpha_local, + .group_order, .local_order + """ + + def __init__(self, session: McdaSession, *args, **kwargs): + super().__init__(*args, **kwargs) + self.session = session + + # -- n_samples: scenario != deterministic + self.fields["n_samples"] = forms.IntegerField( + label="Number of samples", min_value=1, required=False + ) + + self.fields["alpha"] = PositiveFloatField( + label="Alpha (global confidence level)", required=False + ) + if session.weight_mode in ("group", "hierarchical"): + self.fields["alpha_group"] = PositiveFloatField( + label="Alpha for group level", required=False + ) + if session.weight_mode == "hierarchical": + self.fields["alpha_local"] = PositiveFloatField( + label="Alpha for local level", required=False + ) + + def _parse_order_lines(self, raw: str, valid_names: list[str]) -> list[tuple]: + """Parses multi-line ordering constraints into list of (a, b, intensity) tuples.""" + result = [] + for i, line in enumerate(raw.strip().splitlines(), start=1): + line = line.strip() + if not line: + continue + parts = [p.strip() for p in line.split(",")] + if len(parts) != 3: + raise forms.ValidationError(f"Line {i}: expected 'A, B, intensity'.") + a, b = parts[0], parts[1] + for name in (a, b): + if name not in valid_names: + raise forms.ValidationError( + f"Line {i}: '{name}' is not a recognised name." + ) + try: + intensity = float(parts[2]) + except ValueError: + raise forms.ValidationError(f"Line {i}: intensity must be a number.") + if intensity < 0: + raise forms.ValidationError(f"Line {i}: intensity must be positive.") + result.append((a, b, intensity)) + return result + + def clean(self): + cleaned = super().clean() + session = self.session + + if not cleaned.get("n_samples"): + self.add_error("n_samples", "Required for non-deterministic scenarios.") + + if session.scenario == "uncertain": + if cleaned.get("alpha") is None: + self.add_error("alpha", "Required for uncertain scenario.") + if session.weight_mode in ("group", "hierarchical") and cleaned.get("alpha_group") is None: + self.add_error("alpha_group", "Required for uncertain + group/hierarchical weighting.") + if session.weight_mode == "hierarchical" and cleaned.get("alpha_local") is None: + self.add_error("alpha_local", "Required for uncertain + hierarchical weighting.") + + return cleaned + + def save(self, session: McdaSession) -> None: + data = self.cleaned_data + session.n_samples = data.get("n_samples") + session.alpha = data.get("alpha") + session.alpha_group = data.get("alpha_group") + session.alpha_local = data.get("alpha_local") + # session.group_order = data.get("group_order") + # session.local_order = data.get("local_order") + session.save() + + +class GroupOrderForm(forms.ModelForm): + class Meta: + model = GroupOrder + fields = ["group1", "group2", "intensity"] + widgets = { + "group1": forms.Select(), + "group2": forms.Select(), + "intensity": forms.NumberInput(attrs={"min": 0, "step": "0.01"}), + } + + def clean(self): + cleaned = super().clean() + if cleaned.get("group1") == cleaned.get("group2"): + raise forms.ValidationError("Choose two different groups.") + return cleaned + + +class LocalOrderForm(forms.ModelForm): + class Meta: + model = LocalOrder + fields = ["criterion1", "criterion2", "intensity"] + widgets = { + "criterion1": forms.Select(), + "criterion2": forms.Select(), + "intensity": forms.NumberInput(attrs={"min": 0, "step": "0.01"}), + } + + def clean(self): + cleaned = super().clean() + if cleaned.get("criterion1") == cleaned.get("criterion2"): + raise forms.ValidationError("Choose two different criteria.") + return cleaned + + +# Formsets +# extra=1 shows one blank row by default; can_delete=True adds a remove checkbox +GroupOrderFormSet = inlineformset_factory( + parent_model = McdaSession, + model = GroupOrder, + form = GroupOrderForm, + extra = 1, + can_delete = True, +) + +LocalOrderFormSet = inlineformset_factory( + parent_model = McdaSession, + model = LocalOrder, + form = LocalOrderForm, + extra = 1, + can_delete = True, +) diff --git a/mysite/dss/mcda.py b/mysite/dss/mcda.py new file mode 100644 index 0000000..865e318 --- /dev/null +++ b/mysite/dss/mcda.py @@ -0,0 +1,1287 @@ +# Generated from: Promethee_Smaa.ipynb +# Converted at: 2026-06-30 + +from dataclasses import dataclass +import numpy as np +import pandas as pd +from typing import Literal, Optional + +from . import plot + +# Aliases for reused data structures +RangeDict = dict[str, float | tuple[float, float]] +Bounds = dict[str, float] +OrderConstraint = list[tuple[str, str, float]] + +@dataclass(slots=True) +class WeightConstraints: + """ + Class for storing weights and constraints. + Each criterion and group can have a weight. + Weight can be a single value or a range. + The ordering of groups or criteria can also be specified. + """ + group: Optional[RangeDict] = None + group_order: Optional[OrderConstraint] = None + #TODO: local parameters not needed; ungrouped criterion can be used, group can be derived + local: Optional[dict[str, RangeDict]] = None + local_order: Optional[dict[str, OrderConstraint]] = None + criterion: Optional[RangeDict] = None + criterion_order: Optional[OrderConstraint] = None + +@dataclass(slots=True) +class McdaConfig: + # General + df: pd.DataFrame + # decision_matrix: dict[str, RangeDict] + directions: dict[str, Literal["min", "max"] | float] + scenario: str + method: str + criteria: Optional[list[str]] = None # automatically created + + # Weighing and grouping + weight_mode: str = "flat" + groups: Optional[dict[str, list[str]]] = None + _weights: Optional[dict[str, float]] = None + + # PROMETHEE parameters + thresholds: Optional[RangeDict] = None + veto_type: str = "no" + _use_veto: bool = False + veto_thresholds: Optional[Bounds] = None + penalty_factor: float = 0.5 + + # Sampling + sampling_mode: Optional[str] = None + n_samples: int = 10000 + alpha: float = 1.0 + alpha_group: float = 1.0 + alpha_local: float = 1.0 + + # Weight constraints + constraints: Optional[WeightConstraints] = None + + +# ============================================================ +# PERFORMANCE UNCERTAINTY SAMPLING +# ============================================================ + +def has_uncertain_performances(df): + """ + Returns True if at least one cell contains + an interval instead of a deterministic value. + """ + for col in df.columns: + for value in df[col]: + if isinstance(value, (tuple, list)): + return True + return False + +def sample_performance_matrix(df, rng): + """ + This function generates one sampled decision matrix. + If a performance value is deterministic, it is left unchanged. + If a performance value is an interval (lower, upper) + then one value is sampled uniformly inside that interval: + x ~ Uniform(lower, upper) + """ + sampled_df = df.copy() + for col in sampled_df.columns: + for alt in sampled_df.index: + value = sampled_df.loc[alt, col] + if isinstance(value, (tuple, list)): + lower, upper = value + sampled_df.loc[alt, col] = rng.uniform(lower, upper) + return sampled_df + +# ============================================================ +# PROMETHEE HELPER FUNCTIONS +# ============================================================ + +def preference_difference(x_a: float, x_b: float, direction: str) -> float: + """ + Compute the performance difference between alternatives a and b. + The returned value is positive when alternative a is preferred to + alternative b on the considered criterion. + + For benefit criteria: + higher values are better, so difference = x_a - x_b + + For cost criteria: + lower values are better, so difference = x_b - x_a + """ + if isinstance(direction, (int, float)): + return abs(x_b - direction) - abs(x_a - direction) + elif direction == "max": + return x_a - x_b + elif direction == "min": + return x_b - x_a + else: + raise ValueError(f"Unknown direction: {direction}") + + +def promethee_like_preference(d: float, q: float = 0.0) -> float: + """ + PROMETHEE-like binary preference function. + + If the advantage of a over b is greater than the indifference + threshold q, then preference is complete: + P(a,b) = 1 + Otherwise: + P(a,b) = 0 + + With q = 0, this corresponds to a usual criterion (type 1). + With q > 0, this introduces an indifference threshold (type 2). + """ + return 1.0 if d > q else 0.0 + + +def promethee_linear_preference(d: float, q: float, p: float) -> float: + """ + PROMETHEE linear preference function. + + Parameters: + q : indifference threshold + p : preference threshold + + Preference is computed as follows: + + if d <= q: + P(a,b) = 0 + + if q < d < p: + P(a,b) increases linearly from 0 to 1 + + if d >= p: + P(a,b) = 1 + + If q = 0, this corresponds to a V-shape preference function (type 3). + If q > 0, this corresponds to a linear preference function with + indifference area (type 5). + """ + if d <= q: + return 0.0 + if d >= p: + return 1.0 + return (d - q) / (p - q) + + +def veto_is_triggered( + a, b, df_current, directions_dict, veto_thresholds_dict +): + """ + Check whether alternative a is much worse than alternative b + on at least one criterion with a veto threshold. + """ + for c, v in veto_thresholds_dict.items(): + # disadvantage is the inverse of preference + disadvantage = -preference_difference( + df_current.loc[a, c], + df_current.loc[b, c], + directions_dict[c] + ) + if disadvantage > v: + return True + return False + + +def set_fixed_weights(config: McdaConfig): + """ + The dictionary `weights` contains the final weight assigned to + each criterion. + Equal weighting: + each final criterion receives weight 1 / number_of_criteria. + Group-based weighting: + each group has a fixed weight W_g. + The weight of the group is distributed uniformly among + the criteria belonging to that group. + For criterion c belonging to group g: + w_c = W_g / |G_g| + Hierarchical weighting: + each final criterion weight is obtained by multiplying: + group-level weight × normalized local weight + For criterion c belonging to group g: + w_c = W_g × w_{c|g} + where local weights are normalized within each group. + """ + + if config.weight_mode == "flat": + weights = {c: 1 / len(config.criteria) for c in config.criteria} + + elif config.weight_mode == "group": + weights = {} + for g, crits in config.groups.items(): + wg = config.constraints.group[g] + if isinstance(wg, (list, tuple)): + wg = sum(wg) + for c in crits: + weights[c] = wg / len(crits) + + elif config.weight_mode == "hierarchical": + weights = {} + for g, crits in config.groups.items(): + Wg = config.constraints.group[g] + if isinstance(Wg, (list, tuple)): + Wg = sum(Wg) + total_local = sum([ + sum(v)/len(v) if isinstance(v, (list, tuple)) else v + for v in config.constraints.local[g].values() + ]) + for c in crits: + w_local = config.constraints.local[g][c] + if isinstance(w_local, (list, tuple)): + w_local = sum(w_local) + weights[c] = Wg * w_local / total_local + + config._weights = weights + + +# ============================================================ +# WEIGHTING AND RANKING HELPER FUNCTIONS +# ============================================================ +def ranking_to_pairwise_constraints(ranking: list, expected_items: list=None): + """ + Convert a complete or incomplete ordinal ranking into + pairwise constraints. + + Example 1 — complete ranking: + [ + ["C1"], + ["C2", "C3"], + ["C4"] + ] + + Example 2 — incomplete ranking: + [ + ["C1"] + ] + + If expected_items contains C1, C2, C3, and C4, the second + example means that C1 is first, while no order is imposed + among C2, C3, and C4. + + Items belonging to the same rank are not compared. + """ + + if ranking is None: + return [] + + # Copy the ranking to avoid modifying the original input. + effective_ranking = [list(rank_level) for rank_level in ranking] + + ranked_items = [ + item for rank_level in effective_ranking for item in rank_level + ] + ranked_items_set = set(ranked_items) + + if len(ranked_items) != len(ranked_items_set): + raise ValueError("Each item can appear only once in the ranking.") + + if expected_items is not None: + expected_items_set = set(expected_items) + unknown_items = ranked_items_set - expected_items_set + + if unknown_items: + raise ValueError( + f"Unknown items in ranking: {sorted(unknown_items)}" + ) + + # Items omitted from the ranking are interpreted as + # belonging to a final, unordered rank. + unranked_items = list(expected_items_set - ranked_items_set) + + if unranked_items: + effective_ranking.append(unranked_items) + + constraints = [] + + for higher_position in range(len(effective_ranking)): + for lower_position in range(higher_position + 1, len(effective_ranking)): + for higher_item in effective_ranking[higher_position]: + for lower_item in effective_ranking[lower_position]: + constraints.append((higher_item, lower_item, 1.0)) + + return constraints + + +def combine_order_constraints( + ranking=None, pairwise_constraints=None, expected_items=None +): + """ + Combine: + - a complete or incomplete ordinal ranking; + - optional partial pairwise constraints. + + Pairwise constraints can be: + (more_important, less_important) + or: + (more_important, less_important, intensity) + """ + + constraints = ranking_to_pairwise_constraints(ranking, expected_items) + + if pairwise_constraints is not None: + constraints.extend(pairwise_constraints) + + # Remove duplicate constraints. + # If the same comparison has different intensities, + # retain the strongest one. + + merged_constraints = {} + + for constraint in constraints: + intensity = 1.0 if len(constraint) == 2 else constraint[2] + key = constraint[:2] + merged_constraints[key] = max( + intensity, merged_constraints.get(key, 0.0) + ) + + return [ + (*key, intensity) + for key, intensity in merged_constraints.items() + ] + +# ============================================================ +# ACTIVATE CONSTRAINTS ACCORDING TO SAMPLING MODE +# ============================================================ +def get_active_constraints(config: McdaConfig) -> WeightConstraints: + """ + This block translates the selected sampling_mode into the + actual constraints used by the rejection sampler. + + The active constraints depend on sampling_mode: + + ------------------------------------------------------------ + sampling_mode = "random" + - group weights are only constrained by the simplex: + W_g >= 0, sum_g W_g = 1 + - local weights are only constrained by the simplex: + w_{j|g} >= 0, sum_j w_{j|g} = 1 + - no specific lower/upper bounds are used; + - no ordinal constraints are used. + + ------------------------------------------------------------ + sampling_mode = "bounded" + - group lower/upper bounds are activated; + - local lower/upper bounds are activated; + - ordinal constraints are not activated. + + ------------------------------------------------------------ + sampling_mode = "ordered" + - ordinal constraints and preference intensities are activated; + - specific lower/upper bounds are not activated; + - only natural simplex bounds 0 <= w <= 1 are used. + """ + groups = config.groups + active = WeightConstraints() + + if config.sampling_mode == "random": + active.group = {g: (0.0, 1.0) for g in groups} + active.group_order = [] + + active.local = { + g: {c: (0.0, 1.0) for c in crits} + for g, crits in groups.items() + } + active.local_order = {g: [] for g in groups} + + elif config.sampling_mode == "bounded": + active = config.constraints + active.group_order = [] + active.local_order = {g: [] for g in groups} + + elif config.sampling_mode == "ordered": + active.group = {g: (0.0, 1.0) for g in groups} + active.group_order = config.constraints.group_order + + active.local = { + g: {c: (0.0, 1.0) for c in crits} + for g, crits in groups.items() + } + active.local_order = config.constraints.local_order + + elif config.sampling_mode == "bounded_ordered": + active = config.constraints + + else: + raise ValueError( + "sampling_mode must be either " + "'random', 'bounded', 'ordered', or 'bounded_ordered'" + ) + + # Add ungrouped criterion dict + if config.weight_mode == "flat": + active.criterion = {} + active.criterion_order = [] + for group in groups: + active.criterion.update(active.local[group]) + active.criterion_order.append(active.local_order[group]) + + return active + + +def sample_weights_dirichlet_constrained( + bounds_dict: dict, + order_cons: list=[], + max_tries=100000, + batch_size=1000, + alpha=1.0, + rng=None, + stats=None, + stats_key="weights" +): + """ + Sample one feasible weight vector from a constrained Dirichlet + distribution using rejection sampling. + + The function first draws a candidate vector: + w ~ Dirichlet(alpha, ..., alpha) + + and then accepts it only if all active constraints are satisfied. + + The constraints are: + 1) Lower and upper bounds (`bounds_dict`): + lb_i <= w_i <= ub_i + + 2) Optional ordinal constraints (`order_cons`): + w_superior >= w_inferior + + 3) Optional preference intensity constraints (`order_cons`): + w_superior >= intensity × w_inferior + + Notes + ----- + With alpha = 1, the Dirichlet distribution is uniform on the + simplex. Therefore, rejection sampling produces samples uniformly + distributed over the feasible region defined by the active + constraints. + + The optional `stats` dictionary records how many candidate + vectors are generated, accepted and rejected. This is useful to + evaluate the efficiency of rejection sampling. + """ + + if rng is None: + rng = np.random.default_rng() + + if stats is not None and stats_key not in stats: + stats[stats_key] = {"draws": 0, "accepted": 0, "rejected": 0} + + keys = list(bounds_dict.keys()) + idx = {x: i for i, x in enumerate(keys)} + bounds_arr = np.array(list(bounds_dict.values()), dtype=float) + alpha_vec = np.full(len(keys), alpha, dtype=float) + + # Precompute constraint indices/intensities once, outside the loop + hi_idx, lo_idx, intensities = [], [], [] + for constraint in order_cons: + if len(constraint) == 2: + hi, lo = constraint + intensity = 1.0 + elif len(constraint) == 3: + hi, lo, intensity = constraint + else: + raise ValueError( + "Each ordinal constraint must be either " + "(superior, inferior) or " + "(superior, inferior, intensity)." + ) + hi_idx.append(idx[hi]) + lo_idx.append(idx[lo]) + intensities.append(intensity) + + hi_idx = np.array(hi_idx, dtype=int) + lo_idx = np.array(lo_idx, dtype=int) + intensities = np.array(intensities, dtype=float) + + tries_done = 0 + while tries_done < max_tries: + n = min(batch_size, max_tries - tries_done) + + # Draw a whole batch of weights at once (vectorized) + w_batch = rng.dirichlet(alpha_vec, size=n) # shape (n, keys) + tries_done += n + + if stats is not None: + stats[stats_key]["draws"] += n + + # Vectorized bound check across the batch + ok = np.all( + ( w_batch >= bounds_arr[:, 0] - 1e-12) + & (w_batch <= bounds_arr[:, 1] + 1e-12), + axis=1, + ) + + # Vectorized ordinal constraint check across the batch + if ok.any() and len(hi_idx) > 0: + super_w = w_batch[:, hi_idx] + 1e-12 + infer_w = w_batch[:, lo_idx] * intensities + ok &= np.all(super_w >= infer_w, axis=1) + + n_accepted = int(ok.sum()) + if n_accepted > 0: + if stats is not None: + stats[stats_key]["accepted"] += 1 + stats[stats_key]["rejected"] += n - n_accepted + weights = w_batch[np.argmax(ok)] + return {x: float(weights[i]) for x, i in idx.items()} + + elif stats is not None: + stats[stats_key]["rejected"] += n + + raise RuntimeError( + "Could not sample a feasible weight vector. " + "Relax bounds/constraints or increase max_tries." + ) + + +def sample_hierarchical_weights( + groups, + constraints, + + alpha_group=1.0, + alpha_local=1.0, + rng=None, + sampler_stats=None +): + """ + Sample hierarchical criterion weights. + + The hierarchical model is based on two sampling levels. + + Level 1: group weights + ---------------------- + A vector of group weights is sampled: + W = (W_CRM, W_Circularity, W_Environmental, W_Manufacturer) + with: + sum_g W_g = 1 + and with the active group-level constraints. + + Level 2: local criterion weights + --------------------------------- + For each group g, a local weight vector is sampled: + w_{j|g} + with: + sum_{j in G_g} w_{j|g} = 1 + and with the active local constraints for that group. + + Final global weights + -------------------- + The final criterion weight is obtained as: + w_j = W_g × w_{j|g} + where criterion j belongs to group g. + + The optional sampler_stats object stores rejection sampling + diagnostics separately for: + - group weights; + - local weights of each group. + """ + + if rng is None: + rng = np.random.default_rng() + + group_weights_sampled = sample_weights_dirichlet_constrained( + bounds_dict=constraints.group, + order_cons=constraints.group_order, + alpha=alpha_group, + rng=rng, + stats=sampler_stats, + stats_key="group_weights" + ) + + final_weights = {} + + for g, crits in groups.items(): + + local_weights_sampled = sample_weights_dirichlet_constrained( + bounds_dict=constraints.local[g], + order_cons=constraints.local_order[g], + alpha=alpha_local, + rng=rng, + stats=sampler_stats, + stats_key=f"local_weights_{g}" + ) + + for c in crits: + final_weights[c] = group_weights_sampled[g] * local_weights_sampled[c] + + return final_weights + + +def sample_smaa_weights(config: McdaConfig, sampler_stats={}, rng=None): + """ + Sample one criterion-weight vector according to the selected + SMAA weight structure. + + Available weight structures + --------------------------- + + 1) flat + Final criterion weights are sampled directly: + w ~ Dirichlet(alpha, ..., alpha) + + No group structure is used. + + 2) group + Only group weights are sampled: + W_g + + Then each group weight is uniformly distributed among the + criteria in that group: + w_j = W_g / |G_g| + + This mode is useful when the decision maker can express + uncertainty at the group level, but not within groups. + + 3) hierarchical + Group weights and local within-group weights are both sampled: + w_j = W_g × w_{j|g} + + This is the most flexible structured model, because it allows + both group-level and criterion-level uncertainty. + + Returns + ------- + final_weights : dict + Dictionary mapping each final criterion to its sampled weight. + """ + if rng is None: + rng = np.random.default_rng() + + if config.weight_mode == "flat": + active_constr = get_active_constraints(config) + config._weights = sample_weights_dirichlet_constrained( + bounds_dict=active_constr.criterion, + order_cons=[], + alpha=config.alpha, + rng=rng, + stats=sampler_stats, + stats_key="flat_weights" + ) + + elif config.weight_mode == "group": + active_constr = get_active_constraints(config) + group_weights_sampled = sample_weights_dirichlet_constrained( + bounds_dict=active_constr.group, + order_cons=active_constr.group_order, + alpha=config.alpha_group, + rng=rng, + stats=sampler_stats, + stats_key="group_weights" + ) + + final_weights = {} + for g, crits in config.groups.items(): + for c in crits: + final_weights[c] = group_weights_sampled[g] / len(crits) + config._weights = final_weights + + elif config.weight_mode == "hierarchical": + config._weights = sample_hierarchical_weights( + groups=config.groups, + constraints=config.constraints, + alpha_group=config.alpha_group, + alpha_local=config.alpha_local, + rng=rng, + sampler_stats=sampler_stats, + ) + + else: + raise ValueError( + "weight_mode must be either " + "'flat', 'group', or 'hierarchical'" + ) + +# ============================================================ +# PROMETHEE EVALUATION ENGINE +# ============================================================ + +def promethee_nfs(config: McdaConfig, decision_matrix: pd.DataFrame=None): + """ + Compute PROMETHEE net flow scores for a given decision matrix + and a given criterion-weight vector. + + For each ordered pair of alternatives (a,b), the function computes + an aggregated preference score: + S(a,b) = sum_j w_j × P_j(a,b) + + where: + w_j = criterion weight + P_j(a,b) = unicriterion preference of a over b + + Two preference models are supported: + + 1) promethee_like + Binary preference function: + P_j(a,b) = 1 if d_j(a,b) > q_j + = 0 otherwise + + 2) promethee + Linear PROMETHEE preference function with indifference and + preference thresholds: + P_j(a,b) = 0 if d <= q + = (d - q) / (p - q) if q < d < p + = 1 if d >= p + + After computing S(a,b), an optional veto or penalty can be applied. + + The final PROMETHEE flows are: + + phi_plus(a) = sum_b S(a,b) + phi_minus(a) = sum_b S(b,a) + phi(a) = phi_plus(a) - phi_minus(a) + + Returns + ------- + nfs : pandas.Series + Net flow score for each alternative. + """ + # Fall-back to defaults for unspecified parameters + df = decision_matrix if decision_matrix is not None else config.df + method = config.method or "promethee_like" + alts = config.df.index.tolist() + + S = pd.DataFrame(0.0, index=alts, columns=alts) + + if config._use_veto and config.veto_thresholds is None: + raise ValueError("veto_thresholds must be provided when use_veto=True.") + + for a in alts: + for b in alts: + if a == b: + continue + + score = 0.0 + + for c in df.columns: + d = preference_difference( + df.loc[a, c], + df.loc[b, c], + config.directions[c] + ) + + if method == "promethee_like": + q = config.thresholds[c] + pref = promethee_like_preference(d, q) + elif method == "promethee": + q, p = config.thresholds[c] + pref = promethee_linear_preference(d, q, p) + else: + raise ValueError("method must be either 'promethee_like' or 'promethee'.") + + score += config._weights[c] * pref + + # Apply optional veto / penalty after aggregation + if config._use_veto and veto_is_triggered( + a=a, + b=b, + df=df, + directions_dict=config.directions, + veto_thresholds_dict=config.veto_thresholds, + ): + if config.veto_type == "hard": + score = 0.0 + elif config.veto_type == "soft": + score *= config.penalty_factor + else: + raise ValueError("veto_type must be either 'hard' or 'soft'.") + + S.loc[a, b] = score + + phi_plus = S.sum(axis=1) + phi_minus = S.sum(axis=0) + nfs = phi_plus - phi_minus + + return nfs + +def deterministic_promethee(config: McdaConfig): + """ + This function is applicable when: + scenario = "deterministic" + and the decision matrix does not contain interval-valued performances. + + In this case: + - the decision matrix is fixed; + - the weights are fixed; + - one PROMETHEE evaluation is performed; + - the final output is a deterministic ranking. + """ + alts = config.df.index.tolist() + S = pd.DataFrame(0.0, index=alts, columns=alts) + + for a in alts: + for b in alts: + if a == b: + continue + + score = 0.0 + + for c in config.criteria: + d = preference_difference( + config.df.loc[a, c], config.df.loc[b, c], config.directions[c] + ) + + if config.method == "promethee_like": + q = config.thresholds[c] + pref = promethee_like_preference(d, q) + elif config.method == "promethee": + if isinstance(config.thresholds[c], (float, int)): + print(config.method, c, config.thresholds[c]) + q, p = config.thresholds[c] + pref = promethee_linear_preference(d, q, p) + else: + raise ValueError( + "method must be either 'promethee_like' or 'promethee'" + ) + score += config._weights[c] * pref + + if config._use_veto and veto_is_triggered( + a, b, config.df, config.directions, config.veto_thresholds + ): + if config.veto_type == "hard": + score = 0.0 + elif config.veto_type == "soft": + score *= config.penalty_factor + else: + raise ValueError( + "veto_type must be either 'hard' or 'soft'" + ) + + S.loc[a, b] = score + + # ============================================================ + # OUTRANKING FLOWS + # ============================================================ + + phi_plus = S.sum(axis=1) + phi_minus = S.sum(axis=0) + nfs = phi_plus - phi_minus + + results = pd.DataFrame({ + "FOR (phi+)": phi_plus, + "AGAINST (phi-)": phi_minus, + "NFS (phi)": nfs + }).sort_values("NFS (phi)", ascending=False) + return results, S + +def run_performance_uncertainty(config: McdaConfig, rng=None): + """ + Run a Monte Carlo analysis with uncertainty only on alternative + performances. + + In this analysis: + - criterion weights are fixed; + - performance values may be uncertain; + - uncertain performances are sampled from their intervals; + - PROMETHEE is applied to each sampled decision matrix; + - rank acceptability indices are computed from the simulated rankings. + """ + + if rng is None: + rng = np.random.default_rng() + + n_samples = config.n_samples + alts = config.df.index.tolist() + n_alts = len(alts) + + rank_counts = pd.DataFrame( + 0, + index=alts, + columns=[f"rank_{r}" for r in range(1, n_alts + 1)], + dtype=int + ) + outrank_counts = pd.DataFrame( + 0, index=alts, columns=alts, dtype=int + ) + + win_counts = pd.Series(0, index=alts, dtype=int) + + nfs_samples = {a: [] for a in alts} + rank_samples = {a: [] for a in alts} + + for s in range(n_samples): + current_df = sample_performance_matrix(config.df, rng) + nfs = promethee_nfs(config, decision_matrix=current_df) + + order = nfs.sort_values(ascending=False).index.tolist() + rank_map = {a: r for r, a in enumerate(order, start=1)} + + for a in alts: + nfs_samples[a].append(float(nfs[a])) + rank_samples[a].append(rank_map[a]) + + for r, a in enumerate(order, start=1): + rank_counts.loc[a, f"rank_{r}"] += 1 + + win_counts[order[0]] += 1 + + for a in alts: + for b in alts: + if a != b and nfs[a] > nfs[b]: + outrank_counts.loc[a, b] += 1 + + rank_accept = rank_counts / n_samples + win_prob = win_counts / n_samples + outrank_prob = outrank_counts / n_samples + + ranks = np.arange(1, n_alts + 1, dtype=float) + + exp_rank = (rank_accept.values * ranks).sum(axis=1) + exp_rank = pd.Series(exp_rank, index=alts).sort_values() + + mean_nfs = pd.Series( + {a: float(np.mean(nfs_samples[a])) for a in alts} + ).sort_values(ascending=False) + + sim_rows = [] + + for a in alts: + for i in range(n_samples): + sim_rows.append({ + "Alternative": a, + "Simulation": i + 1, + "NFS": nfs_samples[a][i], + "Rank": rank_samples[a][i] + }) + + sim_df = pd.DataFrame(sim_rows) + + return { + "rank_acceptability": rank_accept, + "expected_rank": exp_rank, + "win_probability": win_prob.sort_values(ascending=False), + "outrank_probability": outrank_prob, + "mean_nfs": mean_nfs, + "simulations": sim_df + } + + +# ============================================================ +# SMAA MONTE CARLO SIMULATION +# ============================================================ + +def run_smaa(config: McdaConfig, analysis_type="full_smaa", rng=None): + """ + Run the SMAA Monte Carlo simulation. + + At each simulation, the procedure performs three steps: + + 1) Generate one realization of the decision problem + ------------------------------------------------ + If analysis_type = "smaa_weights": + the decision matrix is fixed. + + If analysis_type = "full_smaa": + uncertain performance values are sampled from their + intervals, generating a new decision matrix. + + 2) Sample one feasible criterion-weight vector + ------------------------------------------- + The sampled weights depend on weight_mode: + flat -> sample final criterion weights directly + group -> sample group weights only + hierarchical -> sample group and local weights + + Active constraints depend on sampling_mode: + random -> no specific constraints + bounded -> lower/upper bounds + ordered -> ordinal and intensity constraints + bounded_ordered -> bounds + constraints + + 3) Evaluate alternatives using PROMETHEE + ------------------------------------- + PROMETHEE net flow scores are computed and converted into + rankings. + + The function then aggregates all simulations to compute: + - rank acceptability indices; + - winning probabilities; + - pairwise outranking probabilities; + - expected ranks; + - mean sampled weights; + - rejection sampling diagnostics. + + Returns + ------- + dict + Dictionary containing SMAA outputs and diagnostic information. + """ + + if rng is None: + rng = np.random.default_rng() + + # The flat model samples final criterion weights directly. + # Since no group or local structure is used, this implementation + # allows flat sampling only in the unconstrained random case. + + if config.weight_mode == "flat" and config.sampling_mode != "random": + raise ValueError( + "With weight_mode='flat', only sampling_mode='random' is supported. " + "Use weight_mode='group' or weight_mode='hierarchical' for bounded or ordered sampling." + ) + + alts = config.df.index.tolist() + n_alts = len(alts) + + rank_counts = pd.DataFrame( + 0, + index=alts, + columns=[f"rank_{r}" for r in range(1, n_alts + 1)], + dtype=int + ) + + win_counts = pd.Series(0, index=alts, dtype=int) + outrank_counts = pd.DataFrame(0, index=alts, columns=alts, dtype=int) + weight_samples = {c: [] for c in config.criteria} + + nfs_samples = {a: [] for a in alts} + rank_samples = {a: [] for a in alts} + accepted = 0 + sampler_stats = {} + + for _ in range(config.n_samples): + # Generate one realization of uncertainty + if analysis_type == "smaa_weights": + current_df = config.df + elif analysis_type == "full_smaa": + current_df = sample_performance_matrix(config.df, rng) + else: + raise ValueError( + "analysis_type must be either " + "'smaa_weights' or 'full_smaa'" + ) + + # Sample one feasible weight vector + sample_smaa_weights(config, sampler_stats, rng) + + for c in config.criteria: + weight_samples[c].append(config._weights[c]) + + # Evaluate alternatives using PROMETHEE + nfs = promethee_nfs(config, decision_matrix=current_df) + # Convert NFS values into a ranking + order = nfs.sort_values(ascending=False).index.tolist() + rank_map = {a: r for r, a in enumerate(order, start=1)} + + # Update SMAA statistics + for a in alts: + nfs_samples[a].append(float(nfs[a])) + rank_samples[a].append(rank_map[a]) + + for r, a in enumerate(order, start=1): + rank_counts.loc[a, f"rank_{r}"] += 1 + + win_counts[order[0]] += 1 + + for a in alts: + for b in alts: + if a == b: + continue + if nfs[a] > nfs[b]: + outrank_counts.loc[a, b] += 1 + + accepted += 1 + + rank_accept = rank_counts / accepted + win_prob = win_counts / accepted + outrank_prob = outrank_counts / accepted + + ranks = np.arange(1, n_alts + 1, dtype=float) + exp_rank = (rank_accept.values * ranks).sum(axis=1) + exp_rank = pd.Series(exp_rank, index=alts).sort_values() + + # Mean PROMETHEE net flow across all simulations + mean_net_flows = pd.Series( + {a: float(np.mean(nfs_samples[a])) for a in alts}, + name="Mean net flow", + ).sort_values(ascending=False) + + w_mean = pd.Series( + {c: float(np.mean(weight_samples[c])) for c in config.criteria} + ).sort_values(ascending=False) + + sim_rows = [] + + for a in alts: + for i in range(accepted): + sim_rows.append({ + "Alternative": a, + "Simulation": i + 1, + "NFS": nfs_samples[a][i], + "Rank": rank_samples[a][i] + }) + + sim_df = pd.DataFrame(sim_rows) + + sampler_stats_df = pd.DataFrame.from_dict( + sampler_stats, orient="index" + ) + + if not sampler_stats_df.empty: + sampler_stats_df["acceptance_rate"] = ( + sampler_stats_df["accepted"] / sampler_stats_df["draws"] + ) + + sampler_stats_df["rejection_rate"] = ( + sampler_stats_df["rejected"] / sampler_stats_df["draws"] + ) + + return { + "rank_acceptability": rank_accept, + "expected_rank": exp_rank, + "win_probability": win_prob.sort_values(ascending=False), + "outrank_probability": outrank_prob, + "mean_weights": w_mean, + "mean_net_flows": mean_net_flows, + "n_samples": accepted, + "simulations": sim_df, + "sampler_stats": sampler_stats_df + } + + +def mcda(config: McdaConfig): + # ============================================================ + # UNCERTAINTY SETTINGS + # ============================================================ + + # Detect whether the decision matrix contains interval-valued performances. + # If at least one cell is a tuple/list, performance uncertainty is activated. + performance_uncertainty = has_uncertain_performances(config.df) + config._use_veto = (config.veto_type!="no") + config.criteria = list(config.directions.keys()) + set_fixed_weights(config) + + if config.method == "promethee_like": # Try to fix if thresholds has tuples + for c, val in config.thresholds.items(): + if not isinstance(val, float): + config.thresholds[c] = sum(val)/len(val) + + # Select the actual analysis type. + if config.scenario == "deterministic": + if performance_uncertainty: + analysis_type = "performance_uncertainty" + else: + analysis_type = "deterministic" + + elif config.scenario == "uncertain": + if performance_uncertainty: + analysis_type = "full_smaa" + else: + analysis_type = "smaa_weights" + + else: + raise ValueError( + "scenario must be either 'deterministic' or 'uncertain'" + ) + + # ============================================================ + # CASE 1: DETERMINISTIC PROMETHEE ANALYSIS + # ============================================================ + + if analysis_type == "deterministic": + results, pairwise = deterministic_promethee(config) + + pd.set_option("display.precision", 6) + print("\n--- DETERMINISTIC SCENARIO ---") + print("\nDecision matrix (numeric):") + print(config.df) + print("\nPairwise preference matrix S(a,b):") + print(pairwise) + print("\nPROMETHEE flows and Net Flow Scores:") + print(results) + print("\nRanking (best to worst):") + print(list(results.index)) + + # plot.deterministic_promethee_figures(results, pairwise) + + # ============================================================ + # CASE 2: PERFORMANCE UNCERTAINTY ANALYSIS + # ============================================================ + """ + This block is executed when: + scenario = "deterministic" + but the decision matrix contains interval-valued performances. + + In this case: + - criterion weights are fixed; + - performance values are sampled through Monte Carlo; + - PROMETHEE is applied to each sampled matrix; + - rank acceptability indices and winning probabilities are + computed from the resulting rankings. + """ + + elif analysis_type == "performance_uncertainty": + rng = np.random.default_rng(42) + results = run_performance_uncertainty(config, rng) + + print("\n--- PERFORMANCE UNCERTAINTY SCENARIO ---") + print(f"Samples used: {config.n_samples}") + + print("\nWinning probabilities (P[rank=1]):") + print(results["win_probability"]) + + print("\nExpected rank (lower is better):") + print(results["expected_rank"]) + + print("\nRank acceptability indices b_{i,r}:") + print(results["rank_acceptability"]) + + print("\nMean NFS:") + print(results["mean_nfs"]) + + # plot.performance_uncertainty_figures(results) + + # ============================================================ + # UNCERTAIN SCENARIO EXECUTION + # ============================================================ + """ + This block is executed when: + scenario = "uncertain" + + Depending on the automatically selected analysis_type, the function + run_smaa performs either: + smaa_weights + uncertainty only on weights; + + full_smaa + uncertainty on both weights and performance values. + + The outputs include: + - winning probabilities; + - expected ranks; + - rank acceptability indices; + - mean sampled weights; + - rejection sampling diagnostics; + - pairwise outranking probabilities. + """ + + if config.scenario == "uncertain": + rng = np.random.default_rng(42) + results = run_smaa(config, analysis_type=analysis_type, rng=rng) + + print("\n--- SMAA SCENARIO ---") + print(f"Samples used: {results['n_samples']}") + + print("\nWinning probabilities (P[rank=1]):") + print(results["win_probability"]) + + mean_net_flow_ranking = list(results["mean_net_flows"].index) + print("\\nRanking based on mean net flows:") + print(mean_net_flow_ranking) + + print("\nExpected rank (lower is better):") + print(results["expected_rank"]) + + print("\nRank acceptability indices b_{i,r}:") + print(results["rank_acceptability"]) + + print("\nMean sampled weights (barycenter):") + print(results["mean_weights"]) + + print("\nRejection sampling diagnostics:") + print(results["sampler_stats"]) + + print("\nPairwise outranking probabilities P(i outranks j) based on NFS:") + print(results["outrank_probability"]) + + # plot.smaa_figures(results) diff --git a/mysite/dss/migrations/0001_initial.py b/mysite/dss/migrations/0001_initial.py new file mode 100644 index 0000000..1046c10 --- /dev/null +++ b/mysite/dss/migrations/0001_initial.py @@ -0,0 +1,95 @@ +# Generated by Django 6.0.5 on 2026-07-08 12:48 + +import django.core.validators +import django.db.models.deletion +import dss.models +from django.db import migrations, models + + +class Migration(migrations.Migration): + + initial = True + dependencies = [] + + operations = [ + migrations.CreateModel( + name='Group', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('name', models.CharField(max_length=40)), + ('weight', models.CharField(max_length=10, null=True, validators=[dss.models.validate_number_or_range])), + ], + ), + migrations.CreateModel( + name='McdaSession', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('status', models.CharField(choices=[('pending', 'Pending'), ('in_progress', 'In Progress'), ('complete', 'Complete')], default='pending', max_length=11)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('scenario', models.CharField(choices=[('deterministic', 'Deterministic'), ('uncertain', 'Uncertain')], max_length=15, null=True)), + ('method', models.CharField(choices=[('promethee_like', 'Promethee Like'), ('promethee', 'Promethee')], max_length=15, null=True)), + ('weight_mode', models.CharField(choices=[('flat', 'Flat'), ('group', 'Group'), ('hierarchical', 'Hierarchical')], max_length=12, null=True)), + ('veto_type', models.CharField(choices=[('no', 'No'), ('hard', 'Hard'), ('soft', 'Soft')], max_length=4, null=True)), + ('group_weights', models.JSONField(null=True)), + ('local_weights', models.JSONField(null=True)), + ('thresholds', models.JSONField(null=True)), + ('veto_thresholds', models.JSONField(null=True)), + ('penalty_factor', models.FloatField(default=0.5)), + ('sampling_mode', models.CharField(choices=[('random', 'Random'), ('bounded', 'Bounded'), ('ordered', 'Ordered'), ('bounded_ordered', 'Bounded Ordered')], max_length=15, null=True)), + ('n_samples', models.IntegerField(null=True)), + ('alpha', models.FloatField(null=True)), + ('alpha_group', models.FloatField(null=True)), + ('alpha_local', models.FloatField(null=True)), + ('group_order', models.JSONField(null=True)), + ('local_order', models.JSONField(null=True)), + ], + ), + migrations.CreateModel( + name='Criterion', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('name', models.CharField(max_length=40)), + ('direction', models.CharField(help_text="Enter 'min', 'max' or a target value", max_length=10, validators=[dss.models.validate_number_or_range])), + ('weight', models.CharField(max_length=10, null=True, validators=[dss.models.validate_number_or_range])), + ('group', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='criteria', to='dss.group')), + ('session', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='criteria', to='dss.mcdasession')), + ], + options={ + 'verbose_name_plural': 'Criteria', + 'unique_together': {('name', 'group')}, + }, + ), + migrations.CreateModel( + name='GroupOrder', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('intensity', models.FloatField(validators=[django.core.validators.MinValueValidator(0)])), + ('group1', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='ordered_higher', to='dss.group')), + ('group2', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='ordered_lower', to='dss.group')), + ], + ), + migrations.CreateModel( + name='LocalOrder', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('intensity', models.FloatField(validators=[django.core.validators.MinValueValidator(0)])), + ('criterion1', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='ordered_higher', to='dss.criterion')), + ('criterion2', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='ordered_lower', to='dss.criterion')), + ('group', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='dss.group')), + ], + ), + migrations.CreateModel( + name='DecisionMatrix', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('alternative', models.CharField(max_length=40)), + ('values', models.JSONField()), + ('session', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='alternatives', to='dss.mcdasession')), + ], + ), + migrations.AddField( + model_name='group', + name='session', + field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='groups', to='dss.mcdasession'), + ), + ] diff --git a/mysite/dss/migrations/0002_rename_group.py b/mysite/dss/migrations/0002_rename_group.py new file mode 100644 index 0000000..91b64e9 --- /dev/null +++ b/mysite/dss/migrations/0002_rename_group.py @@ -0,0 +1,17 @@ +# Generated by Django 6.0.5 on 2026-07-08 13:32 + +from django.db import migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ('dss', '0001_initial'), + ] + + operations = [ + migrations.RenameModel( + old_name='Group', + new_name='CritGroup', + ), + ] \ No newline at end of file diff --git a/mysite/dss/migrations/0003_alter_criterion_fields.py b/mysite/dss/migrations/0003_alter_criterion_fields.py new file mode 100644 index 0000000..1f21da3 --- /dev/null +++ b/mysite/dss/migrations/0003_alter_criterion_fields.py @@ -0,0 +1,39 @@ +# Generated by Django 6.0.5 on 2026-07-08 14:34 + +import dss.models +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('dss', '0002_rename_group'), + ] + + operations = [ + migrations.AlterField( + model_name='criterion', + name='direction', + field=models.CharField(blank=True, help_text="Enter 'min', 'max' or a target value", max_length=10, validators=[dss.models.validate_number_or_range]), + ), + migrations.AlterField( + model_name='criterion', + name='weight', + field=models.CharField(blank=True, default='', max_length=10, validators=[dss.models.validate_number_or_range]), + preserve_default=False, + ), + migrations.RenameField( + model_name='decisionmatrix', + old_name='alternative', + new_name='name', + ), + migrations.AlterField( + model_name='criterion', + name='direction', + field=models.CharField(blank=True, help_text="Enter 'min', 'max' or a target value", max_length=10, validators=[dss.models.validate_direction]), + ), + migrations.AlterUniqueTogether( + name='decisionmatrix', + unique_together={('session', 'name')}, + ), + ] diff --git a/mysite/dss/migrations/0004_alter_localorder_grouporder.py b/mysite/dss/migrations/0004_alter_localorder_grouporder.py new file mode 100644 index 0000000..53574c0 --- /dev/null +++ b/mysite/dss/migrations/0004_alter_localorder_grouporder.py @@ -0,0 +1,38 @@ +# Generated by Django 6.0.5 on 2026-07-14 13:11 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('dss', '0003_alter_criterion_fields'), + ] + + operations = [ + migrations.AddField( + model_name='grouporder', + name='session', + field=models.ForeignKey(default=24, on_delete=django.db.models.deletion.CASCADE, related_name='group_orders', to='dss.mcdasession'), + preserve_default=False, + ), + migrations.RemoveField( + model_name='localorder', + name='group', + ), + migrations.AddField( + model_name='localorder', + name='session', + field=models.ForeignKey(default=24, on_delete=django.db.models.deletion.CASCADE, related_name='local_orders', to='dss.mcdasession'), + preserve_default=False, + ), + migrations.AlterUniqueTogether( + name='localorder', + unique_together={('session', 'criterion1', 'criterion2')}, + ), + migrations.AlterUniqueTogether( + name='grouporder', + unique_together={('session', 'group1', 'group2')}, + ), + ] diff --git a/mysite/dss/migrations/0005_criterion_used_mcdasession_user_type.py b/mysite/dss/migrations/0005_criterion_used_mcdasession_user_type.py new file mode 100644 index 0000000..cd1b886 --- /dev/null +++ b/mysite/dss/migrations/0005_criterion_used_mcdasession_user_type.py @@ -0,0 +1,23 @@ +# Generated by Django 6.0.5 on 2026-08-05 12:30 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('dss', '0004_alter_localorder_grouporder'), + ] + + operations = [ + migrations.AddField( + model_name='criterion', + name='used', + field=models.BooleanField(default=True), + ), + migrations.AddField( + model_name='mcdasession', + name='user_type', + field=models.CharField(choices=[('pd', 'Process developer'), ('kam', 'Key account manager')], max_length=3, null=True), + ), + ] diff --git a/mysite/dss/migrations/__init__.py b/mysite/dss/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/mysite/dss/models.py b/mysite/dss/models.py new file mode 100644 index 0000000..ad21a15 --- /dev/null +++ b/mysite/dss/models.py @@ -0,0 +1,232 @@ +from django.db import models +from .mcda import McdaConfig, WeightConstraints +from django.core.exceptions import ValidationError +from django.core.validators import MinValueValidator + + +def validate_direction(value): + try: + float(value) + except ValueError: + if value not in ["min", "max"]: + raise ValidationError("Must be either 'min', 'max' or target value.") + +def validate_number_or_range(value): + """ + Validates that the value is either a single number or a range (min-max). + Raises ValidationError if the format is invalid. + """ + if not isinstance(value, str): + raise ValidationError("Must be a string in the format 'number' or 'min,max'.") + elif value == "": + return + + try: # Check for single number + float(value) + except ValueError: + try: # Check for range + min_val, max_val = map(float, value.split(',')) + if min_val > max_val: + raise ValidationError("'min' cannot be greater than 'max'.") + return {'min': min_val, 'max': max_val} + except ValueError: + raise ValidationError("Invalid value, use 'min,max' or a number.") + +class McdaSession(models.Model): + """Ties the whole flow together. Tracks where the user is in the decision tree.""" + class Status(models.TextChoices): + PENDING = "pending" # request received, forms not started + IN_PROGRESS = "in_progress" # user is in the form wizard + COMPLETE = "complete" # mcda() has run + + class Scenario(models.TextChoices): + DETERMINISTIC = "deterministic" + UNCERTAIN = "uncertain" + + class Method(models.TextChoices): + PROMETHEE_LIKE = "promethee_like" + PROMETHEE = "promethee" + + class WeightMode(models.TextChoices): + FLAT = "flat" + GROUP = "group" + HIERARCHICAL = "hierarchical" + + class VetoType(models.TextChoices): + NO = "no" + HARD = "hard" + SOFT = "soft" + + class SamplingMode(models.TextChoices): + RANDOM = "random" + BOUNDED = "bounded" + ORDERED = "ordered" + BOUNDED_ORDERED = "bounded_ordered" + + status = models.CharField(max_length=11, choices=Status, default=Status.PENDING) + created_at = models.DateTimeField(auto_now_add=True) + user_type = models.CharField(max_length=3, choices={"pd": "Process developer", "kam": "Key account manager"}, null=True) + + # Form 1 - always present + scenario = models.CharField(max_length=15, choices=Scenario, null=True) + method = models.CharField(max_length=15, choices=Method, null=True) + weight_mode = models.CharField(max_length=12, choices=WeightMode, null=True) + sampling_mode = models.CharField(max_length=15, choices=SamplingMode, null=True) + veto_type = models.CharField(max_length=4, choices=VetoType, null=True) + + # Form 2 - conditional; stored as JSON to handle float|tuple values + group_weights = models.JSONField(null=True) # {"group_name": float | [lo, hi]} + local_weights = models.JSONField(null=True) # {"criterion": float | [lo, hi]} + thresholds = models.JSONField(null=True) # {"criterion": float | [lo, hi]} + veto_thresholds = models.JSONField(null=True) # {"criterion": float} + penalty_factor = models.FloatField(default=0.5) + + # Form 3 - conditional + n_samples = models.IntegerField(null=True) + alpha = models.FloatField(null=True) + alpha_group = models.FloatField(null=True) + alpha_local = models.FloatField(null=True) + group_order = models.JSONField(null=True) # [[group1, group2, float], ...] + local_order = models.JSONField(null=True) # [[criterion1, criterion2, float], ...] + + @property + def criteria(self): + return self.all_criteria.filter(used=True) + + def init_criteria_n_groups(self): + """Create all criteria and groups needed for the MCDA. + Some criteria and all groups are pre-defined. + """ + cost = CritGroup(session=self, name="Costs") + sust = CritGroup(session=self, name="Sustainability") + circ = CritGroup(session=self, name="Circularity", weight=0) + qual = CritGroup(session=self, name="Technical quality") + cost.save() + sust.save() + circ.save() + qual.save() + Criterion(session=self, name="Operating costs", group=cost, direction="min").save() + Criterion(session=self, name="Process energy use", group=sust, direction="min").save() + Criterion(session=self, name="Process carbon footprint", group=sust, direction="min").save() + #TODO: if self.user_type == self.UserType.KAM: + # Add more groups and criteria + + def build_config(self) -> McdaConfig: + import pandas as pd + criteria = list(self.criteria) + criterion_names = [c.name for c in criteria] + criterion_names.sort() + + matrix = { + alt.name: [alt.values[c] for c in criterion_names] + for alt in self.alternatives.all() + } + groups = { + group.name: [c.name for c in group.criteria.all()] + for group in self.groups.all() + } + + constraints = WeightConstraints( + group = self.group_weights, + group_order = self.group_order, + local = self.local_weights, + local_order = self.local_order, + ) + + return McdaConfig( + df = pd.DataFrame.from_dict(matrix, orient='index', columns=criterion_names), + directions = {c.name: c.direction for c in criteria}, + scenario = self.scenario, + method = self.method, + criteria = criterion_names, + weight_mode = self.weight_mode, + groups = groups, + thresholds = self.thresholds, + veto_type = self.veto_type, + veto_thresholds = self.veto_thresholds, + penalty_factor = self.penalty_factor, + sampling_mode = self.sampling_mode, + n_samples = self.n_samples, + alpha = self.alpha, + alpha_group = self.alpha_group, + alpha_local = self.alpha_local, + constraints = constraints + ) + +class DecisionMatrix(models.Model): + """One row with performance values per alternative.""" + session = models.ForeignKey(McdaSession, on_delete=models.CASCADE, + related_name="alternatives") + name = models.CharField(max_length=40) + values = models.JSONField() # {"price": 100, "quality": 0.8} + + class Meta: + unique_together = ['session', 'name'] + +class CritGroup(models.Model): + session = models.ForeignKey(McdaSession, on_delete=models.CASCADE, + related_name="groups") + name = models.CharField(max_length=40) + weight = models.CharField(max_length=10, null=True, validators=[validate_number_or_range]) # filled in during form wizard + + def __str__(self): + return self.name + +class Criterion(models.Model): + """One row per criterion. Partially populated from the request, + completed during the form wizard. + """ + session = models.ForeignKey(McdaSession, on_delete=models.CASCADE, + related_name="all_criteria") + used = models.BooleanField(default=True) + name = models.CharField(max_length=40) # e.g. "price" + group = models.ForeignKey(CritGroup, on_delete=models.SET_NULL, blank=True, null=True, related_name="criteria") + direction = models.CharField(max_length=10, blank=True, validators=[validate_direction], help_text="Enter 'min', 'max' or a target value") + weight = models.CharField(max_length=10, blank=True, validators=[validate_number_or_range]) # filled in during form wizard + + def __str__(self): + return f"{self.name} ({self.group})" + + class Meta: + # To have distinct criteria in the decision matrix, they need to be unique in a session + verbose_name_plural = "Criteria" + unique_together = ['name', 'session'] + + def clean(self): + if self.direction not in ["min", "max"]: + try: + float(self.direction) + except ValueError: + raise ValidationError("The direction must be either 'min', 'max', or a numeric target value.") + + def save(self, *args, **kwargs): + self.full_clean() + super().save(*args, **kwargs) + +class GroupOrder(models.Model): + """Defines the importance order of groups""" + session = models.ForeignKey(McdaSession, on_delete=models.CASCADE, related_name="group_orders") + group1 = models.ForeignKey(CritGroup, on_delete=models.CASCADE, related_name="ordered_higher") + group2 = models.ForeignKey(CritGroup, on_delete=models.CASCADE, related_name="ordered_lower") + intensity = models.FloatField(validators=[MinValueValidator(0)], default=1) + + class Meta: + unique_together = ['session', 'group1', 'group2'] + +class LocalOrder(models.Model): + """Defines the importance order of criteria in a group""" + session = models.ForeignKey(McdaSession, on_delete=models.CASCADE, related_name="local_orders") + criterion1 = models.ForeignKey(Criterion, on_delete=models.CASCADE, related_name="ordered_higher") + criterion2 = models.ForeignKey(Criterion, on_delete=models.CASCADE, related_name="ordered_lower") + intensity = models.FloatField(validators=[MinValueValidator(0)], default=1) + + class Meta: + unique_together = ['session', 'criterion1', 'criterion2'] + + def clean(self): + if self.criterion1.group != self.criterion2.group: + raise ValidationError("Criteria must belong to the same group.") + + def save(self, *args, **kwargs): + self.full_clean() + super().save(*args, **kwargs) diff --git a/mysite/dss/plot.py b/mysite/dss/plot.py new file mode 100644 index 0000000..41a8074 --- /dev/null +++ b/mysite/dss/plot.py @@ -0,0 +1,470 @@ +import numpy as np +import pandas as pd +import matplotlib.pyplot as plt + +# ============================================================ +# FIGURES FOR SHOWING THE RESULTS +# ============================================================ + +def deterministic_promethee_figures(results: pd.DataFrame, S: pd.DataFrame): + """Plot two figures for the deterministic PROMETHEE case + """ + plt.rcParams.update({ + "font.size": 20, + "axes.titlesize": 20, + "axes.labelsize": 20, + "xtick.labelsize": 20, + "ytick.labelsize": 20 + }) + + # ============================================================ + # FIGURE 1: NET FLOW SCORE BAR PLOT + # ============================================================ + + # Sort NFS values according to the final PROMETHEE ranking + sorted_nfs = results["NFS (phi)"].sort_values(ascending=False) + alternatives = sorted_nfs.index + nfs_values = sorted_nfs.values + + # Bar plot of the Net Flow Scores + plt.figure(figsize=(12, 8)) + bars = plt.bar(alternatives, nfs_values) + + # Add a horizontal line at zero to distinguish positive and negative NFS + plt.axhline(0, linewidth=1) + + # Place numerical labels just above the zero line + offset = max(abs(nfs_values)) * 0.03 + + for i, v in enumerate(nfs_values): + plt.text( + i, + 0 + offset, + f"{v:.3f}", + ha="center", + va="bottom", + fontsize=20 + ) + + plt.ylabel("Net Flow Score") + plt.xticks(rotation=45, ha="right") + plt.tight_layout() + + plt.savefig( + "Net_Flow_Score.pdf", format="pdf", bbox_inches="tight" + ) + plt.show() + + # ============================================================ + # FIGURE 2: PAIRWISE PREFERENCE MATRIX WITH FLOWS + # ============================================================ + """ + Create augmented matrix: + - original pairwise preference matrix S(a,b) + - FOR column, i.e. phi_plus + - AGAINST row, i.e. phi_minus + """ + S_aug = S.copy() + S_aug["FOR"] = results["FOR (phi+)"] + + against_row = pd.DataFrame( + [list(results["AGAINST (phi-)"]) + [np.nan]], + columns=S_aug.columns, + index=["AGAINST"] + ) + S_aug = pd.concat([S_aug, against_row]) + + # Convert to numerical array for plotting + data = S_aug.values.astype(float) + + fig, ax = plt.subplots(figsize=(12, 8)) + + # Heatmap of pairwise preference scores and flows + im = ax.imshow( + data, + aspect="auto", + cmap="Oranges", + vmin=0, + vmax=np.nanmax(data) + ) + + # Axis labels + ax.set_xticks(np.arange(len(S_aug.columns))) + ax.set_yticks(np.arange(len(S_aug.index))) + + ax.set_xticklabels(S_aug.columns, rotation=45, ha="right") + ax.set_yticklabels(S_aug.index) + + # Separation lines to visually separate: + # - the FOR column from the pairwise preference matrix; + # - the AGAINST row from the pairwise preference matrix. + n_rows, n_cols = data.shape + ax.axvline(x=n_cols - 1.5, color="black", linewidth=3) + ax.axhline(y=n_rows - 1.5, color="black", linewidth=3) + + # ------------------------------------------------------------ + # Add numerical values inside each cell + # ------------------------------------------------------------ + + max_value = np.nanmax(data) + + for i in range(data.shape[0]): + for j in range(data.shape[1]): + if not np.isnan(data[i, j]): + text_color = "white" if data[i, j] > max_value * 0.55 else "black" + + ax.text( + j, + i, + f"{data[i, j]:.3f}", + ha="center", + va="center", + fontsize=20, + color=text_color + ) + + plt.tight_layout() + # Save as PDF + plt.savefig( + "Pairwise_Preference_Matrix.pdf", format="pdf", bbox_inches="tight" + ) + plt.show() + +def performance_uncertainty_figures(perf_out: dict): + """Create two figures for the performance uncertainty case + """ + FIG_W = 12 + FIG_H = 8 + FONT = 20 + CELL_FONT = 20 + + plt.rcParams.update({ + "font.size": FONT, + "axes.titlesize": FONT, + "axes.labelsize": FONT, + "xtick.labelsize": FONT, + "ytick.labelsize": FONT + }) + + # ============================================================ + # FIGURE 1: RANK ACCEPTABILITY HEATMAP + # ============================================================ + + rank_accept = perf_out["rank_acceptability"].copy() + + # Rename columns for clearer plotting + rank_accept.columns = [ + f"Rank {i}" + for i in range(1, len(rank_accept.columns) + 1) + ] + + # Reorder alternatives according to expected rank + rank_accept = rank_accept.loc[perf_out["expected_rank"].index] + + fig, ax = plt.subplots(figsize=(FIG_W, FIG_H)) + + im = ax.imshow( + rank_accept.values, cmap="Blues", vmin=0, vmax=1 + ) + + ax.set_xticks(np.arange(rank_accept.shape[1])) + ax.set_xticklabels(rank_accept.columns, fontsize=FONT) + + ax.set_yticks(np.arange(rank_accept.shape[0])) + ax.set_yticklabels(rank_accept.index, fontsize=FONT) + + for i in range(rank_accept.shape[0]): + for j in range(rank_accept.shape[1]): + val = rank_accept.iloc[i, j] + ax.text( + j, + i, + f"{val:.2f}", + ha="center", + va="center", + fontsize=CELL_FONT, + color="white" if val > 0.4 else "black" + ) + + cbar = fig.colorbar(im, ax=ax, fraction=0.046, pad=0.04) + cbar.set_label("Probability", fontsize=FONT) + cbar.ax.tick_params(labelsize=FONT) + + for spine in ax.spines.values(): + spine.set_visible(False) + + plt.tight_layout() + plt.savefig( + "Performance_Uncertainty_Rank_Acceptability.pdf", + format="pdf", + bbox_inches="tight" + ) + plt.show() + + # ============================================================ + # FIGURE 2: PAIRWISE OUTRANKING PROBABILITIES + # WITH EXPECTED RANK + # ============================================================ + + outrank_prob = perf_out["outrank_probability"].copy() + expected_rank = perf_out["expected_rank"].copy() + + # Reorder alternatives according to expected rank + order = expected_rank.index + outrank_prob = outrank_prob.loc[order, order] + + # Add expected rank as last column + plot_df = outrank_prob.copy() + plot_df["Expected rank"] = expected_rank.loc[order] + + # Replace diagonal values with NaN + # because an alternative is not compared with itself + for a in order: + plot_df.loc[a, a] = np.nan + + data = plot_df.values.astype(float) + + fig, ax = plt.subplots(figsize=(FIG_W, FIG_H)) + + im = ax.imshow( + data, + aspect="auto", + cmap="Greens", + vmin=0, + vmax=np.nanmax(data) + ) + + ax.set_xticks(np.arange(len(plot_df.columns))) + ax.set_yticks(np.arange(len(plot_df.index))) + ax.set_xticklabels(plot_df.columns, rotation=45, ha="right") + ax.set_yticklabels(plot_df.index) + + # Separation line before Expected rank column + ax.axvline( + x=len(plot_df.columns) - 1.5, color="black", linewidth=3 + ) + + for i in range(data.shape[0]): + for j in range(data.shape[1]): + if not np.isnan(data[i, j]): + text_color = ( + "white" + if data[i, j] > np.nanmax(data) * 0.55 + else "black" + ) + ax.text( + j, + i, + f"{data[i, j]:.3f}", + ha="center", + va="center", + fontsize=CELL_FONT, + color=text_color + ) + + else: + ax.text( + j, i, "--", ha="center", va="center", fontsize=16, color="black" + ) + + plt.tight_layout() + plt.savefig( + "Performance_Uncertainty_Pairwise_Outranking_Expected_Rank.pdf", + format="pdf", + bbox_inches="tight" + ) + plt.show() + +def smaa_figures(smaa_out: dict): + """Create two figures for deterministic or uncertain SMAA results. + """ + # Common figure settings + FIG_W = 12 + FIG_H = 8 + FONT = 20 + CELL_FONT = 20 + + plt.rcParams.update({ + "font.size": FONT, + "axes.titlesize": FONT, + "axes.labelsize": FONT, + "xtick.labelsize": FONT, + "ytick.labelsize": FONT + }) + + # ============================================================ + # FIGURE 1: RANK ACCEPTABILITY HEATMAP + # ============================================================ + + rank_accept = smaa_out["rank_acceptability"].copy() + + # Rename columns for clearer plotting + rank_accept.columns = [ + f"Rank {i}" + for i in range(1, len(rank_accept.columns) + 1) + ] + + # Reorder alternatives according to expected rank + rank_accept = rank_accept.loc[smaa_out["expected_rank"].index] + + fig, ax = plt.subplots(figsize=(FIG_W, FIG_H)) + + im = ax.imshow( + rank_accept.values, cmap="Blues", vmin=0, vmax=1 + ) + + # Axis ticks + ax.set_xticks(np.arange(rank_accept.shape[1])) + ax.set_yticks(np.arange(rank_accept.shape[0])) + ax.set_xticklabels( + rank_accept.columns, + fontsize=FONT, + rotation=25, + ha="right" + ) + ax.set_yticklabels(rank_accept.index, fontsize=FONT) + + # Add values inside cells + for i in range(rank_accept.shape[0]): + for j in range(rank_accept.shape[1]): + val = rank_accept.iloc[i, j] + ax.text( + j, + i, + f"{val:.2f}", + ha="center", + va="center", + fontsize=CELL_FONT, + color="white" if val > 0.4 else "black" + ) + + # Colorbar + cbar = fig.colorbar(im, ax=ax, fraction=0.046, pad=0.04) + cbar.set_label("Probability", fontsize=FONT) + cbar.ax.tick_params(labelsize=FONT) + + # Remove spines for a cleaner look + for spine in ax.spines.values(): + spine.set_visible(False) + + plt.tight_layout() + plt.savefig( + "Rank_Acceptability_Heatmap.pdf", + format="pdf", + bbox_inches="tight" + ) + plt.show() + + # ============================================================ + # FIGURE 2: PAIRWISE OUTRANKING PROBABILITIES + # WITH EXPECTED RANK AND MEAN NET FLOW + # ============================================================ + + outrank_prob = smaa_out["outrank_probability"].copy() + expected_rank = smaa_out["expected_rank"].copy() + mean_net_flows = smaa_out["mean_net_flows"].copy() + + # Order alternatives according to expected rank + order = expected_rank.index + outrank_prob = outrank_prob.loc[order, order] + + # Add expected rank and mean NFS as summary columns + plot_df = outrank_prob.copy() + plot_df["Expected rank"] = expected_rank.loc[order] + plot_df["Mean net flow"] = mean_net_flows.loc[order] + + # Replace diagonal values with NaN + # because an alternative is not compared with itself + for a in order: + plot_df.loc[a, a] = np.nan + + data = plot_df.values.astype(float) + n_alts = len(order) + + # ------------------------------------------------------------ + # COLOUR DATA + # ------------------------------------------------------------ + # + # Only pairwise outranking probabilities are represented + # through the green colour scale. + # + # Expected ranks and mean net flows use a neutral background + # because they are measured on different scales. + + colour_data = np.full_like(data, np.nan, dtype=float) + colour_data[:, :n_alts] = data[:, :n_alts] + + # Use a neutral colour for: + # - diagonal cells; + # - expected-rank column; + # - mean-net-flow column. + + cmap = plt.get_cmap("Greens").copy() + cmap.set_bad(color="#f2f2f2") + + fig, ax = plt.subplots(figsize=(FIG_W + 2, FIG_H)) + im = ax.imshow( + colour_data, + aspect="auto", + cmap=cmap, + vmin=0, + vmax=1, + ) + + # Axis ticks + ax.set_xticks(np.arange(len(plot_df.columns))) + ax.set_yticks(np.arange(len(plot_df.index))) + + ax.set_xticklabels(plot_df.columns, rotation=45, ha="right") + ax.set_yticklabels(plot_df.index) + + # Separation line before Expected rank column + ax.axvline(x=n_alts- 0.5, color="black", linewidth=3) + # Separate expected rank from mean net flow. + ax.axvline(x=n_alts + 0.5, color="black", linewidth=1) + + # ------------------------------------------------------------ + # NUMERICAL VALUES + # ------------------------------------------------------------ + + for i in range(data.shape[0]): + for j in range(data.shape[1]): + value = data[i, j] + if np.isnan(value): + ax.text( + j, + i, + "--", + ha="center", + va="center", + fontsize=16, + color="black" + ) + + else: + # White text is used only for high probabilities. + # Summary columns always use black text. + if j < n_alts and value > 0.55: + text_color = "white" + else: + text_color = "black" + ax.text( + j, + i, + f"{value:.3f}", + ha="center", + va="center", + fontsize=CELL_FONT, + color=text_color + ) + + for spine in ax.spines.values(): + spine.set_visible(False) + + plt.tight_layout() + plt.savefig( + "Pairwise_Outranking_Expected_Rank_Mean_Net_Flow.pdf", + format="pdf", + bbox_inches="tight" + ) + plt.show() diff --git a/mysite/dss/serializers.py b/mysite/dss/serializers.py new file mode 100644 index 0000000..eb772a6 --- /dev/null +++ b/mysite/dss/serializers.py @@ -0,0 +1,218 @@ +from rest_framework import serializers +from .models import Criterion, CritGroup, GroupOrder, LocalOrder + +# Serializers for input data (API requests) +#TODO: configure https://github.com/vbabiy/djangorestframework-camel-case + +class ConsumableSerializer(serializers.Serializer): + name = serializers.CharField() + flowRate = serializers.FloatField() + unit = serializers.CharField() + +class MaterialSerializer(serializers.Serializer): + name = serializers.CharField() + weight = serializers.FloatField() + +class KpiSerializer(serializers.Serializer): + name = serializers.CharField() + value = serializers.FloatField() + target = serializers.CharField() # Can be "min", "max", or a float value + +class ExperimentSerializer(serializers.Serializer): + experimentId = serializers.IntegerField() + weldLength = serializers.FloatField() + weldSpeed = serializers.FloatField() + country = serializers.CharField() + laserPowerkW = serializers.FloatField() + weldingStationPowerkW = serializers.FloatField() + consumables = ConsumableSerializer(many=True) + qualityParameters = KpiSerializer(many=True) + +class ExperimentComparisonSerializer(serializers.ListSerializer): + child = ExperimentSerializer() + + def validate(self, data): + """ + Validate that all experiments in the list have the same set of quality parameter names. + """ + if not data: + return data + + # Collect all quality parameter names from the first experiment + first_experiment = data[0] + first_kpis = { + qp['name'] for qp in first_experiment['qualityParameters'] + } + + # Check all other experiments + for i, experiment in enumerate(data[1:], start=1): + current_kpis = { + qp['name'] for qp in experiment['qualityParameters'] + } + if current_kpis != first_kpis: + raise serializers.ValidationError( + f"Experiment {experiment['experimentId']} has different " + "quality parameters than the first experiment. Expected: " + f"{first_kpis}, Got: {current_kpis}" + ) + + # Check ID uniqueness + ids = set([exp["experimentId"] for exp in data]) + if len(ids) != len(data): + raise serializers.ValidationError("Duplicate experiment IDs found") + + return data + +class WeldStationSerializer(ExperimentSerializer): + maintenanceCosts = serializers.FloatField() + cycleTime = serializers.FloatField() + scrapRate = serializers.FloatField(min_value=0, max_value=0.99) + recyclability = serializers.FloatField(min_value=0, max_value=1) + materials = MaterialSerializer(many=True) + productivity = KpiSerializer(many=True) + +class WeldStationComparisonSerializer(serializers.ListSerializer): + child = WeldStationSerializer() + +class McdaRequestSerializer(serializers.Serializer): + SCENARIO_CHOICES = ["uncertain", "deterministic"] + METHOD_CHOICES = ["promethee", "promethee_like"] + WEIGHT_CHOICES = ["flat", "group", "hierarchical"] + VETO_CHOICES = ["no", "soft", "hard"] + SAMPLING_CHOICES = ["random", "bounded", "ordered", "bounded_ordered"] + + decision_matrix = serializers.DictField() # {"alternative": {"criterion": value|(min, max)},} + directions = serializers.DictField() # {"criterion": "min"|"max"|value} + + scenario = serializers.ChoiceField(choices=SCENARIO_CHOICES) + method = serializers.ChoiceField(choices=METHOD_CHOICES, default=METHOD_CHOICES[1]) + weight_mode = serializers.ChoiceField(choices=WEIGHT_CHOICES, default=WEIGHT_CHOICES[0]) + + groups = serializers.DictField(required=False) # {"group": [criteria]} + group_weights = serializers.DictField(required=False) # {"group": weight} + local_weights = serializers.DictField(required=False) # {"group": {"criterion": weight}} + + thresholds = serializers.DictField(required=False) # Promethee parameters: {"criterion": (q, p)} or {"criterion": q}, where q=indifference, p=preference + veto_type = serializers.ChoiceField(choices=VETO_CHOICES, default="no") + veto_thresholds = serializers.DictField(required=False) # {"criterion": value} + penalty_factor = serializers.FloatField(default=0.5) # used if veto_type=="soft" + + sampling_mode = serializers.ChoiceField(SAMPLING_CHOICES, required=False) # If weight_mode=="flat", this must be "random" + n_samples = serializers.IntegerField(required=False) + alpha = serializers.FloatField(required=False) + alpha_group = serializers.FloatField(required=False) + alpha_local = serializers.FloatField(required=False) + group_lb = serializers.DictField(required=False) # {"group": lower_bound} + group_ub = serializers.DictField(required=False) # {"group": upper_bound} + group_order_constraints = serializers.ListField(required=False) # [("group1", "group2", intensity)] + local_lb = serializers.DictField(required=False) # {"group": {"criterion": lower_bound}} + local_ub = serializers.DictField(required=False) # {"group": {"criterion": lower_bound}} + local_order_constraints = serializers.DictField(required=False) # {"group: [("criterion1", "criterion2", intensity)]} + + def validate(self, data): + # Check that criteria in decision_matrix are in directions + matrix_criteria = set( + criterion + for values in data["decision_matrix"].values() + for criterion in values.keys() + ) + direction_criteria = set(data["directions"].keys()) + missing = matrix_criteria - direction_criteria + if missing: + raise serializers.ValidationError( + "Criteria found in matrix but missing from directions:" + f"{missing}" + ) + # Check direction values + for dir in data["directions"].values(): + if not isinstance(dir, (str, int, float)): + raise serializers.ValidationError(f"Wrong data in directions: {dir}") + elif isinstance(dir, str) and dir not in ["min", "max"]: + raise serializers.ValidationError(f"Expected 'min' or 'max', received: {dir}") + # All criteria must belong to a group + if data["groups"]: + grouped_criteria = set() + for group in data["groups"].values(): + grouped_criteria.update(group) + missing = matrix_criteria - grouped_criteria + if missing: + raise serializers.ValidationError( + f"Criteria missing from groups: {missing}" + ) + # Some weight_modes require group data + if data["weight_mode"] == "group": + if not data["groups"]: + serializers.ValidationError("For grouped weighting, specify 'groups'") + elif data["weight_mode"] == "hierarchical": + if (not data["group_weights"]) or (data["groups"]): + serializers.ValidationError("Hierarchical weighting requires both 'groups' and 'group_weights'") + # Each group must have a weight; they must sum to 1 + if data["group_weights"]: + missing = set(data["groups"].keys()) - set(data["group_weights"].keys()) + if missing: + raise serializers.ValidationError( + f"Criteria missing from grouped_weights: {missing}" + ) + sum = sum(data["group_weights"].values()) + if abs(1 - sum) > 1e-6: + raise serializers.ValidationError( + f"Sum of group weights should be 1, it is {sum}" + ) + # Check that thresholds matches the method + # promethee => tuples; promethee_like => float + if data["method"] == self.METHOD_CHOICES[0] and data["thresholds"]: + for th in data["thresholds"]: + if not isinstance(th, (tuple, list)) or len(th) != 2: + raise serializers.ValidationError( + f"Expected threshold as (q, p), received: {th}" + ) + elif data["method"] == self.METHOD_CHOICES[1] and data["thresholds"]: + for th in data["thresholds"]: + if not isinstance(th, (float, int)): + raise serializers.ValidationError( + f"Expected threshold as number, received: {th}" + ) + # When veto is used, veto_thresholds must be specified + if data["veto_type"] != "no" and not data["veto_thresholds"]: + raise serializers.ValueError( + "veto_thresholds must be provided when veto is used." + ) + + return data + +# Serializers for output data + +class GroupOrderSerializer(serializers.ModelSerializer): + group1_name = serializers.CharField(source='group1.name') + group2_name = serializers.CharField(source='group2.name') + + class Meta: + model = GroupOrder + fields = ['group1_name', 'group2_name', 'intensity'] + + def to_representation(self, instance): + return (instance.group1.name, instance.group2.name, instance.intensity) + +class LocalOrderSerializer(serializers.ModelSerializer): + criterion1_name = serializers.CharField(source='criterion1.name') + criterion2_name = serializers.CharField(source='criterion2.name') + + class Meta: + model = GroupOrder + fields = ['criterion1_name', 'criterion2_name', 'intensity'] + + def to_representation(self, instance): + return (instance.criterion1.name, instance.criterion2.name, instance.intensity) + +class IndicatorScoresSerializer(serializers.Serializer): + indicator = serializers.CharField() + score = serializers.FloatField() + +class ExperimentRankingSerializer(serializers.Serializer): + experiment_id = serializers.IntegerField() + rank = serializers.IntegerField() + score = serializers.IntegerField() + indicators = IndicatorScoresSerializer + +class McdaResultSerializer(serializers.Serializer): + experiment_ranking = serializers.ListField(child=ExperimentRankingSerializer()) diff --git a/mysite/dss/templates/dss/step_0.html b/mysite/dss/templates/dss/step_0.html new file mode 100644 index 0000000..8dc8f49 --- /dev/null +++ b/mysite/dss/templates/dss/step_0.html @@ -0,0 +1,85 @@ +{% extends "base.html" %} +{% block title %}Step 1 — KPI selection{% endblock %} + +{% block content %} +
+ + {# ── Progress bar ── #} +
+
+
+ +
+ + {# ── Header ── #} +
+

Step 1 of 4

+

KPI selection

+

+ Select which KPIs should be used as decision criteria. +

+
+ +
+ {% csrf_token %} + + {% for group in groups %} +
+ + {# {{ group.field }} #} {{ group.name }} + + +
    + {% for criterion in group.criteria.all %} +
  • + {{ criterion.field }} {{ criterion.name }} +
    • + {% endfor %} +
+
+ {% endfor %} + + {% if form.value_rows %} + {% include "partials/section_header.html" with title="Missing values" hint="Please provide any missing values (for selected KPIs)." %} + + + + + + {% for alternative in form.alternatives %} + + {% endfor %} + + + + + {% for row in form.value_rows %} + + + {% for cell in row.cells %} + + {% endfor %} + + {% endfor %} + +
Criterion{{ alternative.name }}
{{ row.criterion.name }}{% if cell %}{{ cell }}{% endif %}
+ {% endif %} + + {# ── Submit ── #} +
+ +
+ +
+
+
+{% endblock %} diff --git a/mysite/dss/templates/dss/step_1.html b/mysite/dss/templates/dss/step_1.html new file mode 100644 index 0000000..9321043 --- /dev/null +++ b/mysite/dss/templates/dss/step_1.html @@ -0,0 +1,71 @@ +{% extends "base.html" %} +{% block title %}Step 2 — Analysis Setup{% endblock %} + +{% block content %} +
+ + {# ── Progress bar ── #} +
+
+
+ +
+ + {# ── Header ── #} +
+

Step 2 of 4

+

Analysis setup

+

+ These four choices drive which options appear in the next step. +

+
+ +
+ {% csrf_token %} + + {# ── Non-field errors ── #} + {% if form.non_field_errors %} +
+ {{ form.non_field_errors }} +
+ {% endif %} + + {# ── Scenario ── #} + {% include "partials/radio_group.html" with field=form.scenario label="Scenario" options=scenario_options %} + + {# ── Method ── #} + {% include "partials/radio_group.html" with field=form.method label="MCDA method" options=method_options %} + + {# ── Weight mode ── #} + {% include "partials/radio_group.html" with field=form.weight_mode label="Weight mode" options=weight_mode_options %} + + {# ── Veto type ── #} + {% include "partials/radio_group.html" with field=form.veto_type label="Veto type" options=veto_type_options %} + + {# ── Navigation ── #} +
+ + + + + Back + + +
+ +
+
+
+{% endblock %} diff --git a/mysite/dss/templates/dss/step_2.html b/mysite/dss/templates/dss/step_2.html new file mode 100644 index 0000000..d629e14 --- /dev/null +++ b/mysite/dss/templates/dss/step_2.html @@ -0,0 +1,276 @@ +{% extends "base.html" %} +{% load dss_tags %} +{% block title %}Step 3 — Weights & Thresholds{% endblock %} + +{% block content %} +
+ + {# ── Progress bar ── #} +
+
+
+ +
+ + {# ── Header ── #} +
+

Step 3 of 4

+

Weights & thresholds

+

+ Weights accept a single value 0.5 + or a range 0.2, 0.8. + All values must be positive. +

+
+ + {# ── Session context pills ── #} +
+ + Decision data: {{ session.get_scenario_display }} + + + Method: {{ session.get_method_display }} + + + Weights: {{ session.get_weight_mode_display }} + + + Veto: {{ session.get_veto_type_display }} + +
+ +
+ {% csrf_token %} + + {% if form.non_field_errors %} +
+ {{ form.non_field_errors }} +
+ {% endif %} + + {% comment %} + SECTION 1 — Sampling mode + Visible when weight_mode == "group" + {% endcomment %} + {% if session.weight_mode == "group" %} + {% include "partials/section_header.html" with title="Sampling mode" hint="Decide how to draw samples from weight and/or KPI values, for Monte Carlo simulation." %} + + {% with field=form.sampling_mode %} +
+ {% for opt in sampling_mode_options %} + + {% endfor %} +
+ {% if field.errors %} +

{{ field.errors|join:", " }}

+ {% endif %} + {% endwith %} + {% endif %} + + {% comment %} + SECTION 2 — Group weights + Visible when weight_mode in {group, hierarchical} + {% endcomment %} + {% if session.weight_mode in "group,hierarchical" %} + {% include "partials/section_header.html" with title="Group weights" hint="Relative importance of each pre-defined group. Enter a value or range." %} + +
+ {% for group in groups %} + {% with fname="group_weight_"|add:group %} + {% include "partials/weight_field.html" with field=form|get_field:fname label=group|capfirst %} + {% endwith %} + {% endfor %} +
+ {% endif %} + + {% comment %} + SECTION 3 — Local weights + Visible when weight_mode == hierarchical + {% endcomment %} + {% if session.weight_mode == "hierarchical" %} + {% include "partials/section_header.html" with title="Local weights" hint="Weight of each criterion within its group." %} + +
+ {% for criterion in criteria %} + {% with fname="local_weight_"|add:criterion %} + {% include "partials/weight_field.html" with field=form|get_field:fname label=criterion %} + {% endwith %} + {% endfor %} +
+ {% endif %} + + {% comment %} + SECTION 4 — Thresholds (always) + Shape differs between promethee (range) and others (float) + {% endcomment %} + {% include "partials/section_header.html" with title="Thresholds" hint=threshold_hint %} + +
+
+ Criterion + + {% if session.method == "promethee" %}Indifference, Preference{% else %}Threshold{% endif %} + + Target +
+ + {% for criterion, target in criteria.items %} + {% with fname="threshold_"|add:criterion %} +
+ +
+ {{ criterion }} +
+ +
+ {% with field=form|get_field:fname %} + + {% if field.errors %} +

{{ field.errors|join:", " }}

+ {% endif %} + {% endwith %} +
+ +
+ {% if target == "max" %} + max + + {% elif target == "min" %} + min + + {% else %} + {{target}} + + {% endif %} +
+ +
+ {% endwith %} + {% endfor %} +
+ + {% comment %} + SECTION 5 — Veto thresholds + Visible when veto_type != "no" + {% endcomment %} + {% if session.veto_type != "no" %} + {% include "partials/section_header.html" with title="Veto thresholds" hint="Maximum tolerable disadvantage per criterion." %} + {% comment %} +
+ {% for criterion in criteria %} + {% with fname="veto_"|add:criterion %} + {% with field=form|get_field:fname %} +
+ +
+ + {% if field.errors %} +

{{ field.errors|join:", " }}

+ {% endif %} +
+
+ {% endwith %} + {% endwith %} + {% endfor %} +
+ {% endcomment %} + + {# ── Penalty factor — only for soft veto ── #} + {% if session.veto_type == "soft" %} +
+ {% with field=form.penalty_factor %} + +

+ Scales the severity of the soft veto. Must be positive. +

+ + {% if field.errors %} +

{{ field.errors|join:", " }}

+ {% endif %} + {% endwith %} +
+ {% endif %} + {% endif %} + + {# ── Navigation ── #} +
+ + + + + Back + + +
+ +
+
+
+{% endblock %} diff --git a/mysite/dss/templates/dss/step_3.html b/mysite/dss/templates/dss/step_3.html new file mode 100644 index 0000000..de036d9 --- /dev/null +++ b/mysite/dss/templates/dss/step_3.html @@ -0,0 +1,202 @@ +{% extends "base.html" %} +{% block title %}Step 4 — Sampling settings{% endblock %} + +{% block content %} +
+ + {# ── Progress bar ── #} +
+
+
+ +
+ + {# ── Header ── #} +
+

Step 4 of 4

+

Sampling settings

+

+ To assess the effect of uncertain parameters, sampling and uncertainty parameters + are configured here. +

+
+ + {# ── Session context pills ── #} +
+ + Decision data: {{ session.get_scenario_display }} + + + Method: {{ session.get_method_display }} + + + Weights: {{ session.get_weight_mode_display }} + + + Veto: {{ session.get_veto_type_display }} + +
+ +
+ {% csrf_token %} + + {# ── Non-field errors ── #} + {% if form.non_field_errors %} +
+ {{ form.non_field_errors }} +
+ {% endif %} + + {% include "partials/section_header.html" with title="Sampling parameters" hint="All values must be positive." %} + + {{ form.as_p }} + + {% include "partials/section_header.html" with title="Ordering constraints" hint=order_hint %} + {% if group_order_formset %} + {{ group_order_formset.management_form }} + + + + + + {% for row in group_order_formset %} + + + + + + + {% endfor %} + +
Superior GroupInferior GroupIntensityRemove
{{ row.group1 }}{{ row.group2 }}{{ row.intensity }}{{ row.DELETE }}
+ + {% endif %} + + {% if local_order_formset %} + {{ local_order_formset.management_form }} + + + + + + {% for row in local_order_formset %} + + + + + + + {% endfor %} + +
Superior CriterionInferior CriterionIntensityRemove
{{ row.criterion1 }}{{ row.criterion2 }}{{ row.intensity }}{{ row.DELETE }}
+ + {% endif %} + + {# ── Navigation ── #} +
+ + + + + Back + + +
+ +
+
+
+ + +{% endblock %} diff --git a/mysite/dss/templates/partials/radio_group.html b/mysite/dss/templates/partials/radio_group.html new file mode 100644 index 0000000..dfded66 --- /dev/null +++ b/mysite/dss/templates/partials/radio_group.html @@ -0,0 +1,53 @@ +{% comment %} + Partial: radio_group.html + Variables: + field — form field (for id, name, errors) + label — display label + options — list of dicts: [{value, label, hint}] + hint is optional; used for a short explanatory line below the label +{% endcomment %} +
+ + {{ label }} + {% if field.errors %} + {{ field.errors|join:", " }} + {% endif %} + + +
+ {% for opt in options %} + + {% endfor %} +
+
diff --git a/mysite/dss/templates/partials/section_header.html b/mysite/dss/templates/partials/section_header.html new file mode 100644 index 0000000..0513384 --- /dev/null +++ b/mysite/dss/templates/partials/section_header.html @@ -0,0 +1,13 @@ +{% comment %} + Partial: section_header.html + Variables: + title — section title + hint — one-line explanation shown below the title +{% endcomment %} +
+

{{ title }}

+ +
+{% if hint %} +

{{ hint }}

+{% endif %} diff --git a/mysite/dss/templates/partials/weight_field.html b/mysite/dss/templates/partials/weight_field.html new file mode 100644 index 0000000..b274061 --- /dev/null +++ b/mysite/dss/templates/partials/weight_field.html @@ -0,0 +1,24 @@ +{% comment %} + Partial: weight_field.html + Variables: + field — form field (WeightField instance) + label — display label string +{% endcomment %} +
+ + + {% if field.errors %} +

{{ field.errors|join:", " }}

+ {% endif %} +
diff --git a/mysite/dss/templatetags/__init__.py b/mysite/dss/templatetags/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/mysite/dss/templatetags/dss_tags.py b/mysite/dss/templatetags/dss_tags.py new file mode 100644 index 0000000..b8d55d8 --- /dev/null +++ b/mysite/dss/templatetags/dss_tags.py @@ -0,0 +1,8 @@ +from django import template + +register = template.Library() + +@register.filter +def get_field(form, field_name: str): + """{{ form|get_field:"group_weight_economic" }} → the BoundField.""" + return form[field_name] diff --git a/mysite/dss/tests.py b/mysite/dss/tests.py new file mode 100644 index 0000000..d72cb19 --- /dev/null +++ b/mysite/dss/tests.py @@ -0,0 +1,707 @@ +import logging +import pytest +from unittest.mock import patch + +import pandas as pd +from django.test import TestCase +from .views import lookup, calculate_experiment_kpis, calculate_kpis +from .models import DecisionMatrix, Criterion +from .mcda import McdaConfig, WeightConstraints, mcda + +logger = logging.getLogger(__name__) +logger.setLevel('DEBUG') +logging.getLogger('django.db.backends').setLevel(logging.WARNING) + + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def material_data_df(): + return pd.DataFrame( + {"price": [10.0, 5.0], "CO2eq": [2.0, 1.0]}, + index=pd.Index(["Argon", "Steel"], name="material"), + ) + +@pytest.fixture(autouse=True) +def patched_material_data(material_data_df): + """Auto-applied so every test sees a controlled material_data table.""" + with patch("dss.views.material_data", material_data_df): + yield material_data_df + +@pytest.fixture +def country_data_df(): + return pd.DataFrame( + { + "wages": [30.0, 0.0, 20.0], + "electricity_price": [0.25, 0.23, 0.15], + "electricity_CO2eq": [0.4, 0.4, 0.6], + }, + index=pd.Index(["NL", "DE", "?"], name="country"), + ) + +@pytest.fixture(autouse=True) +def patched_country_data(country_data_df): + """Auto-applied so every test sees a controlled country_data table.""" + with patch("dss.views.country_data", country_data_df): + yield country_data_df + +@pytest.fixture +def pd_experiment(): + return { + "experimentId": 1, + "country": "NL", + "weldLength": 2.0, + "weldSpeed": 3.0, + "laserPowerkW": 5.0, + "weldingStationPowerkW": 1.0, + "consumables": [ + {"name": "Argon", "unit": "L/min", "flowRate": 10.0}, + ], + "qualityParameters": [ + {"name": "Tensile strength", "value": 450, "target": "max"}, + ], + } + +@pytest.fixture +def kam_experiment(pd_experiment): + exp = pd_experiment + exp.update({ + "cycleTime": 8.0, + "scrapRate": 0.1, + "maintenanceCosts": 1000.0, + "recyclability": 0.8, + "materials": [ + {"name": "Steel", "weight": 4.0}, + ], + "productivity": [ + {"name": "Machine saturation", "value": 0.7, "target": "max"}, + {"name": "Process automation level", "value": "medium", "target": "max"}, + ], + }) + return exp + + +class TestLookup: + + def test_returns_value_for_known_country(self, country_data_df): + assert lookup("wages", "NL") == 30.0 + + def test_falls_back_to_default_for_unknown_country(self, country_data_df): + assert lookup("wages", "XX") == 20.0 + + def test_falls_back_to_default_when_value_is_false(self, country_data_df): + assert lookup("wages", "DE") == 20.0 + +# --------------------------------------------------------------------------- +# calculate_experiment_kpis +# --------------------------------------------------------------------------- + +class TestCalculateExperimentKpisPd: + + def test_pd_kpi_results(self, pd_experiment): + result = calculate_experiment_kpis(pd_experiment, "pd") + + assert set(result.keys()) == { + "Process energy use", + "Operating costs", + "Process carbon footprint", + "Tensile strength", + } + + assert result["Tensile strength"] == 450 + assert result["Operating costs"] > 0 + assert result["Process energy use"] > 0 + assert result["Process carbon footprint"] > 0 + + # PD yield_rate == 1, cycle_time == processing_time, so process energy + # use should equal (processing_time * (laser + welding station kW)) / 3600. + processing_time = pd_experiment["weldLength"] * pd_experiment["weldSpeed"] + expected_energy = ( + pd_experiment["laserPowerkW"] + pd_experiment["weldingStationPowerkW"] + ) * processing_time / 3600 + assert result["Process energy use"] == pytest.approx(expected_energy) + + def test_unrecognised_consumable_unit_raises(self, pd_experiment): + pd_experiment["consumables"][0]["unit"] = "bogus/min" + + with pytest.raises(RuntimeError, match="Cannot process unit 'bogus'"): + calculate_experiment_kpis(pd_experiment, "pd") + + def test_unrecognised_material_raises(self, pd_experiment): + pd_experiment["consumables"][0]["name"] = "Unobtainium" + + with pytest.raises(RuntimeError, match="Material 'Unobtainium' not recognised"): + calculate_experiment_kpis(pd_experiment, "pd") + + +class TestCalculateExperimentKpisKam: + + def test_returns_expected_keys(self, kam_experiment): + result = calculate_experiment_kpis(kam_experiment, "kam") + + expected = { + "Process energy use", "Operating costs", "Process carbon footprint", + "Tensile strength", "Maintenance costs", "Scrap rate", + "Recyclability", "Production cycle time", "Total production costs", + "Product carbon footprint", "Machine saturation", + "Process automation level", + } + assert set(result.keys()) == expected + + assert result["Maintenance costs"] == kam_experiment["maintenanceCosts"] + assert result["Scrap rate"] == kam_experiment["scrapRate"] + assert result["Recyclability"] == kam_experiment["recyclability"] + assert result["Production cycle time"] == kam_experiment["cycleTime"] + assert result["Machine saturation"] == 0.7 + + def test_calculation_result(self, pd_experiment, kam_experiment): + # With scrapRate > 0, yield_rate < 1, so KAM's process energy use + # (divided by yield_rate) should exceed the PD figure computed from + # the same underlying weld parameters. + pd_result = calculate_experiment_kpis(pd_experiment, "pd") + result = calculate_experiment_kpis(kam_experiment, "kam") + + assert result["Process energy use"] > pd_result["Process energy use"] + + assert result["Total production costs"] > result["Operating costs"] + + def test_scrap_rate_of_one_raises_zero_division(self, kam_experiment): + # yield_rate = 1 - scrapRate; a scrapRate of 1 means yield_rate = 0, + # which should surface as a ZeroDivisionError rather than silently + # producing inf/NaN KPIs. + kam_experiment["scrapRate"] = 1.0 + + with pytest.raises(ZeroDivisionError): + calculate_experiment_kpis(kam_experiment, "kam") + + +# --------------------------------------------------------------------------- +# calculate_kpis +# --------------------------------------------------------------------------- + +@pytest.mark.django_db +def test_calculate_kpis_propagates_unrecognised_unit_error(pd_experiment): + pd_experiment["consumables"][0]["unit"] = "something/min" + + with pytest.raises(RuntimeError, match="Cannot process unit 'something'"): + calculate_kpis([pd_experiment], user="pd") + + +@pytest.mark.django_db +class TestCalculateKpis: + + def test_creates_session_and_matrix(self, pd_experiment): + exp2 = dict(pd_experiment, experimentId=2) + valid_data = [pd_experiment, exp2] + + session = calculate_kpis(valid_data, user="pd") + rows = DecisionMatrix.objects.filter(session=session) + assert rows.count() == 2 + assert {row.name for row in rows} == {"Experiment #1", "Experiment #2"} + + def test_pd_decision_matrix(self, pd_experiment): + session = calculate_kpis([pd_experiment], user="pd") + + names = set(Criterion.objects.filter(session=session).values_list("name", flat=True)) + print(names) + assert "Tensile strength" in names + + row = DecisionMatrix.objects.get(session=session, name="Experiment #1") + expected = { + 'Process energy use': 0.01, + 'Operating costs': 0.16250000000000003, + 'Process carbon footprint': 0.006, + 'Tensile strength': 450 + } + assert row.values == expected + + def test_kam_decision_matrix(self, kam_experiment): + exp2 = dict(kam_experiment, experimentId=2, scrapRate=0.2) + session = calculate_kpis([kam_experiment, exp2], user="kam") + + names = set(Criterion.objects.filter(session=session).values_list("name", flat=True)) + assert "Machine saturation" in names + + row = DecisionMatrix.objects.get(session=session, name="Experiment #1") + expected = { + 'Process energy use': 0.011728395061728396, + 'Operating costs': 0.2135108024691358, + 'Process carbon footprint': 0.006913580246913581, + 'Tensile strength': 450, + 'Maintenance costs': 1000.0, + 'Scrap rate': 0.1, + 'Recyclability': 0.8, + 'Production cycle time': 8.0, + 'Total production costs': 22.48946330589849, + 'Product carbon footprint': 4.451358024691358, + 'Machine saturation': 0.7, + 'Process automation level': 'medium' + } + assert row.values == expected + + matrix = DecisionMatrix.objects.filter(session=session) + rows = {row.name: row.values for row in matrix} + assert rows["Experiment #1"]["Scrap rate"] == 0.1 + assert rows["Experiment #2"]["Scrap rate"] == 0.2 + + +class McdaTest(TestCase): + """ + Class for testing an MCDA request, + including decision matrix and weighting preferences. + """ + # ============================================================ + # DECISION MATRIX + # ============================================================ + + def test_mcda(self): + df = pd.DataFrame( + data=[ + [9.3, 0.0, 0.0, 0.0, 0.0, 1, 0.551, 0.19108, 4.95, 0.0282, 0.00500, 0.000000439, 26.9, 1.91, 0.298], + [9.3, 0.0, 0.0, 0.0, 0.0, 2, 0.551, 0.19108, 4.63, 0.0265, 0.00501, 0.000000412, 26.5, 1.94, 0.397], + [0.0, 0.027, 0.2, 0.3, 3.3, 2, 0.396, 0.01072, 15.0, 0.1160, 0.01020, 0.000001270, 52.7, 3.34, 0.592], + [0.0, 0.027, 0.2, 0.3, 3.3, 3, 0.396, 0.01072, 8.43, 0.0807, 0.00756, 0.000000774, 17.8, 9.38, 0.297], + ], + index=["Steel RoW", "Steel RER", "AlLi RoW", "AlLi Canada"], + columns=[ + "Ni concentration (%)", + "Li concentration (%)", + "Mg concentration (%)", + "Ti concentration (%)", + "Cu concentration (%)", + "Operator", + "Recycled input (kg/kg)", + "Waste output (kg/kg)", + "Climate change (GWP100)", + "Acidification (AE)", + "Eutrophication Freshwater (P)", + "Particulate Matter (human health)", + "LandUse (soil quality index)", + "WaterUse (m³ world eq deprived)", + "Ionising radiation (kBq U-235 eq)", + ], + ) + + # ============================================================ + # CRITERION ORIENTATION + # ============================================================ + + directions = { + "Ni concentration (%)": "min", + "Li concentration (%)": "min", + "Mg concentration (%)": "min", + "Ti concentration (%)": "min", + "Cu concentration (%)": "min", + "Operator": "max", + "Recycled input (kg/kg)": "max", + "Waste output (kg/kg)": "min", + "Climate change (GWP100)": "min", + "Acidification (AE)": "min", + "Eutrophication Freshwater (P)": "min", + "Particulate Matter (human health)": "min", + "LandUse (soil quality index)": "min", + "WaterUse (m³ world eq deprived)": "min", + "Ionising radiation (kBq U-235 eq)": "min", + } + + # ============================================================ + # CRITERION GROUPS + # ============================================================ + + groups = { + "CRM": [ + "Ni concentration (%)", + "Li concentration (%)", + "Mg concentration (%)", + "Ti concentration (%)", + "Cu concentration (%)", + ], + "Circularity": [ + "Recycled input (kg/kg)", + "Waste output (kg/kg)", + ], + "Environmental": [ + "Climate change (GWP100)", + "Acidification (AE)", + "Eutrophication Freshwater (P)", + "Particulate Matter (human health)", + "LandUse (soil quality index)", + "WaterUse (m³ world eq deprived)", + "Ionising radiation (kBq U-235 eq)", + ], + "Manufacturer": [ + "Operator", + ] + } + + # ============================================================ + # CRITERION WEIGHTS + # ============================================================ + + group_weights = { + "CRM": 0.25, + "Circularity": 0.25, + "Environmental": 0.40, + "Manufacturer": 0.10 + } + + local_weights = { + "CRM":{ + "Ni concentration (%)": 0.30, + "Li concentration (%)": 0.20, + "Mg concentration (%)": 0.15, + "Ti concentration (%)": 0.15, + "Cu concentration (%)": 0.20 + }, + "Circularity":{ + "Recycled input (kg/kg)": 0.70, + "Waste output (kg/kg)": 0.30 + }, + "Environmental":{ + "Climate change (GWP100)": 0.30, + "Acidification (AE)": 0.10, + "Eutrophication Freshwater (P)": 0.10, + "Particulate Matter (human health)": 0.10, + "LandUse (soil quality index)": 0.10, + "WaterUse (m³ world eq deprived)": 0.20, + "Ionising radiation (kBq U-235 eq)": 0.10 + }, + "Manufacturer":{ + "Operator": 1.0 + } + } + + # ============================================================ + # PROMETHEE SETTINGS + # ============================================================ + + thresholds = { + "Ni concentration (%)": (0.0, 2.0), + "Li concentration (%)": (0.0, 0.01), + "Mg concentration (%)": (0.0, 0.05), + "Ti concentration (%)": (0.0, 0.05), + "Cu concentration (%)": (0.0, 1.0), + "Operator": (0.0, 1.0), + "Recycled input (kg/kg)": (0.0, 0.10), + "Waste output (kg/kg)": (0.0, 0.05), + "Climate change (GWP100)": (0.0, 5.0), + "Acidification (AE)": (0.0, 0.05), + "Eutrophication Freshwater (P)": (0.0, 0.003), + "Particulate Matter (human health)": (0.0, 0.0000005), + "LandUse (soil quality index)": (0.0, 20.0), + "WaterUse (m³ world eq deprived)": (0.0, 3.0), + "Ionising radiation (kBq U-235 eq)": (0.0, 0.2), + } + + # Veto thresholds: if alternative a is worse than b by more than v, + # then S(a,b) is penalized. + veto_thresholds = { + "Climate change (GWP100)": 8.0, + "WaterUse (m³ world eq deprived)": 5.0, + "Ni concentration (%)": 5.0, + "Cu concentration (%)": 2.0, + } + + # ============================================================ + # GROUP-LEVEL WEIGHT UNCERTAINTY + # ============================================================ + """ + Group-level weights represent the relative importance of the + main dimensions of the decision problem: + - CRM + - Circularity + - Environmental + - Manufacturer + + These weights are denoted as: + W_g + and must satisfy: + W_g >= 0 + sum_g W_g = 1 + + In the "bounded" sampling mode, lower and upper bounds are + imposed on each group weight. + In the "ordered" sampling mode, ordinal constraints are imposed. + These can also include preference intensities. + """ + # ============================================================ + + group_bounds = { + "CRM": (0.10,0.40), + "Circularity": (0.10,0.40), + "Environmental": (0.20,0.60), + "Manufacturer": (0.05,0.25), + } + + group_ranking = None + group_pairwise_constraints = [] + group_order_constraints = [ + ("Environmental", "Manufacturer", 1.5), + ("CRM", "Manufacturer", 1.2), + ("Circularity", "Manufacturer", 1.0), + ] + + # ============================================================ + # LOCAL CRITERION-LEVEL WEIGHT UNCERTAINTY + # ============================================================ + """ + Local weights represent the importance of criteria within + each group. + + For a criterion j belonging to group g, the local weight is: + w_{j|g} + and must satisfy, within each group: + w_{j|g} >= 0 + sum_{j in G_g} w_{j|g} = 1 + + In the hierarchical model, the final global criterion weight is: + w_j = W_g × w_{j|g} + + where: + W_g = sampled group weight + w_{j|g} = sampled local weight of criterion j within group g + + In the "bounded" sampling mode, local lower and upper bounds + are imposed. + In the "ordered" sampling mode, local ordinal constraints and + preference intensities are imposed. + + A local ordinal constraint can be specified as: + ("A", "B") + meaning: + w_A >= w_B + or as: + ("A", "B", intensity) + meaning: + w_A >= intensity × w_B + """ + + local_bounds = { + "CRM": { + "Ni concentration (%)": (0.10,0.40), + "Li concentration (%)": (0.05,0.30), + "Mg concentration (%)": (0.05,0.25), + "Ti concentration (%)": (0.05,0.25), + "Cu concentration (%)": (0.10,0.40), + }, + "Circularity": { + "Recycled input (kg/kg)": (0.40,0.80), + "Waste output (kg/kg)": (0.20,0.60), + }, + "Environmental": { + "Climate change (GWP100)": (0.15,0.35), + "Acidification (AE)": (0.05,0.20), + "Eutrophication Freshwater (P)": (0.05,0.20), + "Particulate Matter (human health)": (0.05,0.20), + "LandUse (soil quality index)": (0.05,0.20), + "WaterUse (m³ world eq deprived)": (0.10,0.30), + "Ionising radiation (kBq U-235 eq)": (0.05,0.20), + }, + "Manufacturer": { + "Operator": (1.00,1.00), + } + } + + local_order_constraints = { + "CRM": [ + ("Ni concentration (%)", "Li concentration (%)"), + ("Ni concentration (%)", "Mg concentration (%)"), + ("Cu concentration (%)", "Mg concentration (%)"), + ("Cu concentration (%)", "Ti concentration (%)"), + ], + "Circularity": [ + ("Recycled input (kg/kg)", "Waste output (kg/kg)"), + ], + "Environmental": [ + ("Climate change (GWP100)", "Acidification (AE)"), + ("Climate change (GWP100)", "Eutrophication Freshwater (P)"), + ("Climate change (GWP100)", "Particulate Matter (human health)"), + ("Climate change (GWP100)", "LandUse (soil quality index)"), + ("Climate change (GWP100)", "Ionising radiation (kBq U-235 eq)"), + ("WaterUse (m³ world eq deprived)", "Ionising radiation (kBq U-235 eq)"), + ], + "Manufacturer": [] + } + + constraints = WeightConstraints( + group_bounds, group_order_constraints, + local_bounds, local_order_constraints, + ) + + # ============================================================ + # COMPOSING ALMOST ALL COMBINATIONS OF PARAMETERS + # ============================================================ + + # "scenario", "method", "weight", "veto", "sampling" + test_cases = [ + ["deterministic", "promethee", "flat", "no", "random"], + ["deterministic", "promethee", "flat", "soft", "random"], + ["deterministic", "promethee", "flat", "hard", "random"], + ["deterministic", "promethee", "group", "no", "random"], + ["deterministic", "promethee", "hierarchical", "no", "random"], + ["uncertain", "promethee", "flat", "no", "random"], + ["uncertain", "promethee", "group", "no", "random"], + ["uncertain", "promethee", "hierarchical", "no", "random"], + ["uncertain", "promethee", "group", "no", "bounded"], + ["uncertain", "promethee", "hierarchical", "no", "bounded"], + ["uncertain", "promethee", "group", "no", "ordered"], + ["uncertain", "promethee", "hierarchical", "no", "ordered"], + ["deterministic", "promethee_like", "group", "no", "ordered"], + ["deterministic", "promethee_like", "hierarchical", "no", "bounded_ordered"], + ] + for settings in test_cases: + logger.debug(f"Testing case: {settings}") + config = McdaConfig( + df=df, + directions=directions, + scenario=settings[0], + method=settings[1], + weight_mode=settings[2], + groups=groups, + thresholds=thresholds, + veto_type=settings[3], + veto_thresholds=veto_thresholds, + sampling_mode=settings[4], + n_samples=10, + constraints=constraints, + ) + result = mcda(config) + return + +class Explanations(): + # ============================================================ + # ANALYSIS SCENARIO SELECTION + # ============================================================ + """ + The variable `scenario` controls whether the analysis uses: + 1) Fixed weights + scenario = "deterministic" + 2) Uncertain weights sampled through SMAA + scenario = "uncertain" + + The uncertainty on the performances is not selected manually. + It is detected automatically by checking whether the decision + matrix contains interval values, represented as tuples or lists: + (lower_bound, upper_bound) + + Therefore, the final analysis type is determined by combining: + - scenario + - presence/absence of uncertain performances in df + + This generates four analysis_type cases: + 1. deterministic: Fixed weights + deterministic performance values + 2. performance_uncertainty: Fixed weights + uncertain performance values + 3. smaa_weights: Uncertain weights + deterministic performance values + 4. full_smaa: Uncertain weights + uncertain performance values + """ + + scenario = "uncertain" + + # ============================================================ + # DETERMINISTIC WEIGHT MODE + # ============================================================ + """ + This parameter is used only when the analysis relies on fixed + criterion weights, namely in: + - deterministic + - performance_uncertainty + + Available options: + "flat": All final criteria receive the same weight. + "group": Group weights are fixed, and each group weight is distributed + equally among the criteria belonging to that group. + "hierarchical": Final criterion weights are obtained as: + final weight = group weight × local criterion weight + This allows criteria within a group to have different relative importance. + """ + + weight_mode = "flat" + + # ============================================================ + # METHOD SELECTION + # ============================================================ + """ + method = "promethee_like" + + If q = 0: Type I / usual criterion + If q > 0: Type II / U-shape criterion + + In both cases, preference is binary: + P(a,b) = 1 if d(a,b) > q + = 0 otherwise + + method = "promethee" + + If q = 0 and p > 0: Type III / V-shape criterion + If q > 0 and p > q: Type V / linear preference with indifference area + """ + + method = "promethee_like" + + # ============================================================ + # VETO SETTINGS + # ============================================================ + + use_veto = False + + # Type of veto: + # "hard" => set S(a,b) to 0 + # "soft" => multiply S(a,b) by a penalty factor + veto_type = "soft" + + penalty_factor = 0.5 + + + # ============================================================ + # WEIGHT UNCERTAINTY MODEL + # ============================================================ + """ + This section defines how criterion weights are generated in + the SMAA analysis. + + The model supports three alternative weighting structures: + + 1) flat + Final criterion weights are sampled directly. + No group structure is imposed. + w = (w_1, ..., w_m) + sum_j w_j = 1 + + This mode is currently allowed only with: + sampling_mode = "random" + + 2) group + Only group-level weights are sampled. + Each sampled group weight is then uniformly distributed + among the criteria belonging to that group. + w_j = W_g / |G_g| + where criterion j belongs to group g. + + 3) hierarchical + Both group-level weights and local within-group weights + are sampled. + w_j = W_g × w_{j|g} + This allows criteria in the same group to have different + local importance. + + ------------------------------------------------------------ + The variable `sampling_mode` defines which constraints are + activated during weight sampling: + + random: + no bounds and no ordinal constraints are imposed. + bounded: + lower and upper bounds are imposed, but no ordinal + constraints are imposed. + ordered: + ordinal constraints and preference intensities are imposed, + but specific lower/upper bounds are not imposed. + bounded_ordered: + lower/upper bounds and ordinal/intensity constraints are + imposed simultaneously. + """ + + weight_mode = "flat" + sampling_mode = "random" diff --git a/mysite/dss/urls.py b/mysite/dss/urls.py new file mode 100644 index 0000000..9a19230 --- /dev/null +++ b/mysite/dss/urls.py @@ -0,0 +1,12 @@ +from django.urls import path +from .views import ( + ExperimentComparisonInitView, KamComparisonInitView, + McdaWizardView, ExperimentResultsView, +) + +urlpatterns = [ + path("experiments/", ExperimentComparisonInitView.as_view(), name="pd-init"), + path("welding-stations/", KamComparisonInitView.as_view(), name="kam-init"), + path("/step//", McdaWizardView.as_view(), name="wizard"), + path("results//", ExperimentResultsView.as_view(), name="results"), +] diff --git a/mysite/dss/views.py b/mysite/dss/views.py new file mode 100644 index 0000000..84e4115 --- /dev/null +++ b/mysite/dss/views.py @@ -0,0 +1,499 @@ +import pandas as pd + +from django.shortcuts import get_object_or_404, redirect, render, reverse +from django.views.generic import DetailView +from rest_framework.views import APIView, View +from rest_framework.response import Response +from rest_framework import status + +from .serializers import ExperimentComparisonSerializer, WeldStationComparisonSerializer, McdaRequestSerializer, McdaResultSerializer +from .mcda import mcda, McdaConfig, WeightConstraints +from .models import McdaSession, DecisionMatrix, Criterion, CritGroup +from .forms import KpiSelectionForm, BaseConfigForm, WeightsThresholdsForm, SamplingConfigForm, GroupOrderFormSet, LocalOrderFormSet + +# wages, electricity carbon intensity, electricity price +country_data = pd.read_csv("init_data/country_data.csv", index_col="country") +# price, carbon footprint of consumables +material_data = pd.read_csv("init_data/material_data.csv", index_col="material") +MULTIPLIERS = {"h": 1/3600, "min": 1/60, "s": 1, "kg": 1, "g": 1/1000, "mg": 10**-6, "m3": 1, "L": 1/1000} + +# Helper functions + +def lookup(info: str, country: str): + """Find the info for country in country_data + """ + if country in country_data.index: + value = country_data.loc[country, info] + if value: + return value + return country_data.loc["?", info] + +def create_criteria(session: McdaSession, user_type: str="pd"): + """Create default groups and criteria""" + cost = CritGroup.objects.create(session=session, name="Economic") + sust = CritGroup.objects.create(session=session, name="Sustainability") + qual = CritGroup.objects.create(session=session, name="Technical quality") + Criterion.objects.create(session=session, name="Operating costs", group=cost, direction="min") + Criterion.objects.create(session=session, name="Process energy use", group=sust, direction="min") + Criterion.objects.create(session=session, name="Process carbon footprint", group=sust, direction="min") + if user_type == "kam": + circ = CritGroup.objects.create(session=session, name="Circularity", weight=0) + oper = CritGroup.objects.create(session=session, name="Productivity") + Criterion.objects.create(session=session, name="Total production costs", group=cost, direction="min") + Criterion.objects.create(session=session, name="Maintenance costs", group=cost, direction="min") + Criterion.objects.create(session=session, name="Production cycle time", group=oper, direction="min") + Criterion.objects.create(session=session, name="Product carbon footprint", group=sust, direction="min") + Criterion.objects.create(session=session, name="Scrap rate", group=circ, direction="min") + Criterion.objects.create(session=session, name="Recyclability", group=circ, direction="max") + return qual, oper + return qual, None + +def calculate_experiment_kpis(exp: dict, user_type: str) -> dict: + """ + Compute the KPI values for a single experiment or alternative. + + Converts consumable flow rates to absolute usage, then derives cost, + energy-use, and carbon-footprint KPIs shared by all users. For the + KAM user, additionally accounts for scrap-adjusted yield, maintenance + and material costs, amortized CAPEX, and operational-efficiency KPIs. + + Parameters: + exp: A single experiment's raw data (one entry of `valid_data`), + containing weld parameters, consumables, quality parameters, + and for KAM: materials, operational efficiency data. + user_type: Either "pd" or "kam". Determines the KPI set. + + Returns: + dict: KPI name -> value, ready to store as a DecisionMatrix row. + + Raises: + RuntimeError: If a consumable's quantity/time unit or material + name is not recognised. + """ + processing_time = exp["weldLength"] * exp["weldSpeed"] #TODO: check units + consumables_use = {} + for cons in exp["consumables"]: + qnt_unit, time_unit = cons["unit"].split("/") + if qnt_unit not in MULTIPLIERS: + raise RuntimeError(f"Cannot process unit '{qnt_unit}'") + elif time_unit not in MULTIPLIERS: + raise RuntimeError(f"Cannot process unit '{time_unit}'") + elif cons["name"] not in material_data.index: + raise RuntimeError(f"Material '{cons["name"]}' not recognised.") + amount = MULTIPLIERS[qnt_unit] * cons["flowRate"] * \ + processing_time * MULTIPLIERS[time_unit] + consumables_use[cons["name"]] = amount + + if user_type == "kam": + cycle_time = exp["cycleTime"] + yield_rate = 1 - exp["scrapRate"] + else: # Simplifications for the PD user + cycle_time = processing_time + yield_rate = 1 + consumables_costs = sum([ + amount * material_data.loc[cons, "price"] + for cons, amount in consumables_use.items() + ]) + labour_costs = cycle_time/3600 * lookup("wages", exp["country"]) * 3 + process_energy_use = ( + processing_time * exp["laserPowerkW"] + + cycle_time * exp["weldingStationPowerkW"] + ) / 3600 / yield_rate + operating_costs = consumables_costs + labour_costs + ( + process_energy_use * lookup("electricity_price", exp["country"]) + ) + consumables_footprint = sum([ + amount * material_data.loc[cons, "CO2eq"] + for cons, amount in consumables_use.items() + ]) / yield_rate + proc_carbon_footprint = consumables_footprint + ( + process_energy_use * lookup("electricity_CO2eq", exp["country"]) + ) + if user_type == "kam": + annual_production = 240 * 16 * 3600 / cycle_time + operating_costs += exp["maintenanceCosts"] / annual_production + material_costs = sum([ + mat["weight"] * material_data.loc[mat["name"], "price"] + for mat in exp["materials"] + ]) + material_footprint = sum([ + mat["weight"] * material_data.loc[mat["name"], "CO2eq"] + for mat in exp["materials"] + ]) + service_life = 15 + CAPEX = 700000 + amortized_capex = CAPEX / (service_life * annual_production) + total_production_costs = ( + operating_costs + material_costs + amortized_capex + ) / yield_rate + prod_carbon_footprint = ( + proc_carbon_footprint + material_footprint / yield_rate + ) + + # Create dict of KPIs + kpi_dict = { + "Process energy use": process_energy_use, + "Operating costs": operating_costs, + "Process carbon footprint": proc_carbon_footprint, + } + for kpi in exp["qualityParameters"]: + kpi_dict[kpi["name"]] = kpi["value"] + if user_type == "kam": + kam_kpis = { + "Maintenance costs": exp["maintenanceCosts"], + "Scrap rate": exp["scrapRate"], + "Recyclability": exp["recyclability"], + "Production cycle time": cycle_time, + "Total production costs": total_production_costs, + "Product carbon footprint": prod_carbon_footprint, + } + kpi_dict.update(kam_kpis) + for kpi in exp["productivity"]: + kpi_dict[kpi["name"]] = kpi["value"] + + return kpi_dict + +def calculate_kpis(valid_data: list, user: str="pd") -> McdaSession: + """ + Calculate the KPIs (criteria) of interest for user (pd or kam) + and attach these to a new McdaSession. + + Creates the session's criteria groups (via `create_criteria`), + registers the dynamic quality parameters + and for KAM, operational efficiency criteria. + Then convert each experiment in `valid_data` to a DecisionMatrix row. + + Args: + valid_data: List of validated experiment dicts, as produced by + the PD or KAM serializer. All entries are assumed to share + the same set of quality parameters / operational-efficiency + criteria as `valid_data[0]`. + user: Either "pd" or "kam". Determines which criteria are + created and which KPIs are calculated per experiment. + + Returns: + McdaSession: The newly created session, with criteria, groups, + and one DecisionMatrix row per experiment attached. + """ + # Start a session and initiate groups and criteria + session = McdaSession.objects.create() + quality_group, oper_group = create_criteria(session, user_type=user) + for kpi in valid_data[0]["qualityParameters"]: + Criterion.objects.create( + session=session, name=kpi["name"], + group=quality_group, direction=kpi["target"], + ) + if oper_group: + for kpi in valid_data[0]["productivity"]: + Criterion.objects.create( + session=session, name=kpi["name"], + group=oper_group, direction=kpi["target"], + ) + + # Extract data from all experiments, compute and store KPIs + for experiment in valid_data: + kpi_dict = calculate_experiment_kpis(experiment, user) + name = "Experiment #" + str(experiment["experimentId"]) + DecisionMatrix.objects.create( + session=session, name=name, values=kpi_dict + ) + return session + + +def step1_context() -> dict: + """ + Static context for step 1 form. + Each option dict: {value, label, hint}. + Hints help users who aren't MCDA experts. + """ + return { + "scenario_options": [ + {"value": "deterministic", "label": "Deterministic", + "hint": "Criteria have a fixed importance."}, + {"value": "uncertain", "label": "Uncertain", + "hint": "Criteria weights are given as ranges."}, + # {"value": "stochastic", "label": "Stochastic", + # "hint": "Weights drawn from distributions."}, + ], + "method_options": [ + {"value": "promethee", "label": "PROMETHEE", + "hint": "Define indifference and preference thresholds."}, + {"value": "promethee_like", "label": "PROMETHEE-like", + "hint": "Define the preference of criteria."}, + ], + "weight_mode_options": [ + {"value": "flat", "label": "Flat", + "hint": "One weight per criterion."}, + {"value": "group", "label": "Group", + "hint": "Criteria are grouped; weights are equal within a group."}, + {"value": "hierarchical", "label": "Hierarchical", + "hint": "Group weights × local criterion weights."}, + ], + "veto_type_options": [ + {"value": "no", "label": "None", "hint": "No veto applied"}, + {"value": "hard", "label": "Hard", "hint": "Disqualify alternatives"}, + {"value": "soft", "label": "Soft", "hint": "Penalise alternatives"}, + ], + "sampling_mode_options": [ + {"value": "random", "label": "Random", + "hint": "No bounds or constraints are imposed."}, + {"value": "bounded", "label": "Bounded", + "hint": "Lower and upper bounds are imposed."}, + {"value": "ordered", "label": "Ordered", + "hint": "A preference order of the importance of groups is imposed."}, + {"value": "bounded_ordered", "label": "Bounded ordered", + "hint": "Both lower and upper bounds, and a preference order imposed."}, + ] + } + + +def step2_context(session) -> dict: + """ + Dynamic context for step 2 — depends on session choices. + """ + directions = {crit.name: crit.direction for crit in session.criteria} + group_names = list(session.groups.values_list("name", flat=True)) + + threshold_hint = ( + "PROMETHEE requires an indifference and preference threshold per criterion " + "(enter as a range: q, p)." + if session.method == "promethee" + else "Enter the threshold value for each criterion." + ) + + return { + "criteria": directions, + "groups": group_names, + "threshold_hint": threshold_hint, + } + +class ExperimentComparisonInitView(APIView): + def post(self, request): + serializer = ExperimentComparisonSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + + session = calculate_kpis(serializer.validated_data) + + return redirect( + reverse("wizard", kwargs={"session_id": session.id, "step": 0}) + ) + return Response({"session_id": session.id}, status=status.HTTP_201_CREATED) + +class KamComparisonInitView(APIView): + def post(self, request): + serializer = WeldStationComparisonSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + + session = calculate_kpis(serializer.validated_data, user="kam") + + return redirect( + reverse("wizard", kwargs={"session_id": session.id, "step": 0}) + ) + return Response({"session_id": session.id}, status=status.HTTP_201_CREATED) + +class McdaWizardView(View): + def _next_step(self, current_step: int, session: McdaSession) -> int | None: + """Decision tree: select the next form""" + if current_step < 2: # Form 0 and 1 are always shown + return current_step + 1 + if current_step == 2: + if session.scenario != "deterministic": + return 3 + return None # deterministic: skip Form 3 + return None # step 3 is always the last + + def _ask_for_orders(self, session: McdaSession) -> bool: + return ( + session.scenario == "uncertain" and + session.sampling_mode in ["ordered", "bounded_ordered"] + ) + + def _get_order_formsets(self, session, data=None): + """Instantiate both formsets with querysets scoped to this session.""" + session_groups = CritGroup.objects.filter(session=session) + session_criteria = Criterion.objects.filter(session=session) + + group_formset = GroupOrderFormSet(data, instance=session, prefix="gfs") + for form in group_formset.forms: + form.fields["group1"].queryset = session_groups + form.fields["group2"].queryset = session_groups + + local_formset = LocalOrderFormSet(data, instance=session, prefix="lfs") + for form in local_formset.forms: + form.fields["criterion1"].queryset = session_criteria + form.fields["criterion2"].queryset = session_criteria + + return group_formset, local_formset + + def get_form(self, step: int, session: McdaSession, data=None): + """Returns the right form for the current step.""" + if step == 0: + return KpiSelectionForm(session, data) + elif step == 1: + return BaseConfigForm(data) + elif step == 2: + return WeightsThresholdsForm(session, data) + elif step == 3: + return SamplingConfigForm(session, data) + + def get_context(self, step: int, session: McdaSession, **kwargs): + if step == 0: + context = {"groups": kwargs["form"].groups} + elif step == 1: + context = step1_context() + elif step == 2: + context = step2_context(session) + elif step == 3: + if self._ask_for_orders(session): + context = dict(zip( + ["group_order_formset", "local_order_formset"], + self._get_order_formsets(session), + )) + for formset in ["group_order_formset", "local_order_formset"]: + if formset in kwargs: + context[formset] = kwargs[formset] + context["order_hint"] = "Ordering the importance of groups or criteria relative to each other." + if session.weight_mode=="hierarchical": + context["order_hint"] += " Only specify the order of criteria belonging to the same group." + else: + context = {} + context.update(kwargs) + context["session"] = session + return context + + def get(self, request, session_id: int, step: int): + session = get_object_or_404(McdaSession, pk=session_id) + form = self.get_form(step, session) + context = self.get_context(step, session=session, form=form) + return render(request, f"dss/step_{step}.html", context) + + def post(self, request, session_id: int, step: int): + session = get_object_or_404(McdaSession, pk=session_id) + form = self.get_form(step, session, data=request.POST) + + if step == 3 and self._ask_for_orders(session): + group_order, local_order = self._get_order_formsets( + session, data=request.POST + ) + else: + group_order = local_order = None + + formsets_valid = ( + (group_order is None or group_order.is_valid()) and + (local_order is None or local_order.is_valid()) + ) + if not form.is_valid() or not formsets_valid: + context = self.get_context( + step, session, form=form, group_order_formset=group_order, local_order_formset=local_order + ) + return render(request, f"dss/step_{step}.html", context) + + # Save the form and GroupOrder + LocalOrder formsets to tables + form.save(session) + if group_order: + group_order.save() + if local_order: + local_order.save() + + next_step = self._next_step(step, session) + if next_step: + return redirect("wizard", session_id=session_id, step=next_step) + + # All steps done. Run the MCDA calculations + config = session.build_config() + result = mcda(config) + response_data = McdaResultSerializer(result).data + return redirect("results", session_id=session_id) + return Response(response_data, status=status.HTTP_200_OK) + +class ExperimentResultsView(DetailView): + model = "ExperimentResults" + +def parse_request(valid_data: dict) -> McdaSession: + """ + Converts validated serializer output → ORM models. + Called once when the initial API request comes in. + """ + session = McdaSession.objects.create() + + if valid_data["groups"] is not None: + for criterion_name, direction in valid_data["directions"].items(): + Criterion(session, name=criterion_name, direction=direction).save() + for group_name, criteria in valid_data["groups"].items(): + group = CritGroup(session, name=group_name) + group.save() + Criterion.objects.filter(name__in=criteria).update(group=group) + + quality_grp = create_criteria(session) + + for alt_name, values in valid_data["decision_matrix"].items(): + DecisionMatrix(session=session, name=alt_name, values=values).save() + + for criterion_name, direction in valid_data["directions"].items(): + if not Criterion.objects.filter(name=criterion_name).exists(): + Criterion( + session, name=criterion_name, group=quality_grp, direction=direction + ).save() + + return session + +class McdaInitView(APIView): + def post(self, request): + serializer = McdaRequestSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + + session = parse_request(serializer.validated_data) + + return Response({"session_id": session.id}, status=status.HTTP_201_CREATED) + + +#TODO: remove this View +class McdaCalculationView(APIView): + def post(self, request): + request_serializer = McdaRequestSerializer(data=request.data) + request_serializer.is_valid(raise_exception=True) + data = request_serializer.validated_data + + # Convert some raw data to read-to-use data types + try: + decision_matrix = pd.DataFrame(data["decision_matrix"]) + except BaseException as e: + return Response({"Issue with decision matrix": str(e)}, status=status.HTTP_400_BAD_REQUEST) + + try: + constraints = WeightConstraints( + group=data.get("group_weights"), + group_order=data.get("group_order_constraints"), + local=data.get("local_weights"), + local_order=data.get("local_order_constraints"), + ) + config = McdaConfig( + df=decision_matrix, + directions=data["directions"], + scenario=data["scenario"], + method=data["method"], + weight_mode=data["weight_mode"], + groups=data.get("groups"), + thresholds=data.get("thresholds"), + + # Veto settings + veto_type=data["veto_type"], + veto_thresholds=data.get("veto_thresholds"), + penalty_factor=data.get("penalty_factor"), + + # Sampling + sampling_mode=data.get("sampling_mode"), + n_samples=data.get("n_samples"), + alpha=data.get("alpha"), + alpha_group=data.get("alpha_group"), + alpha_local=data.get("alpha_local"), + + # Weight structure settings (group-level and local) + constraints=constraints, + ) + result = mcda(config) + except ValueError as e: + return Response({"error": str(e)}, status=status.HTTP_400_BAD_REQUEST) + + results = McdaResultSerializer(result) + return Response(results.data, status=status.HTTP_200_OK) diff --git a/mysite/init_data/country_data.csv b/mysite/init_data/country_data.csv new file mode 100644 index 0000000..1f9a6ac --- /dev/null +++ b/mysite/init_data/country_data.csv @@ -0,0 +1,3 @@ +country,wages,electricity_price,electricity_CO2eq +DE,60,0.25,21 +?,70,0.214,23 diff --git a/mysite/init_data/material_data.csv b/mysite/init_data/material_data.csv new file mode 100644 index 0000000..df775ca --- /dev/null +++ b/mysite/init_data/material_data.csv @@ -0,0 +1,5 @@ +material,price,CO2eq,unit +Argon,13,9.47,m3 +Nitrogen,7.05,7.11,m3 +Aluminium (filler wire),100,88,kg +Steel,56,42,kg diff --git a/mysite/mysite/settings.py b/mysite/mysite/settings.py index 00efae9..4d4cd75 100644 --- a/mysite/mysite/settings.py +++ b/mysite/mysite/settings.py @@ -63,6 +63,7 @@ 'api', 'accounts', 'dpp.apps.DppConfig', + 'dss', 'rest_framework', ] diff --git a/mysite/mysite/urls.py b/mysite/mysite/urls.py index 06be71e..08f0b25 100644 --- a/mysite/mysite/urls.py +++ b/mysite/mysite/urls.py @@ -19,8 +19,9 @@ urlpatterns = [ path('admin/', admin.site.urls), - path('api/', include("api.urls")), # API endpoints - path('dpp/', include("dpp.urls")), # Frontend application + path('api/', include("api.urls")), # API endpoints for DPP models + path('dpp/', include("dpp.urls")), # Frontend for DPP creation + path('dss/', include("dss.urls")), # API for decision support path("accounts/", include("accounts.urls")), # User registration path("accounts/", include("django.contrib.auth.urls")), # User login ]