Feature/v3/feature/305 rename specific share to other effects to specific share from effect (#366)

* Step 1

* Bugfix

* Make fit_effects_to_model_coords() more flexible

* Fix dims

* Update conftest.py

* Typos

* Improve Effect examples

* Add extra validation for Effect Shares

Author: FBumann

CHANGELOG.md

@@ -48,6 +48,18 @@ Several internal improvements were made to the codebase.
 
 ### ✨ Added
 
+**Intuitive effect share syntax:**
+Effects now support an intuitive `share_from_*` syntax for cross-effect relationships:
+```python
+costs = fx.Effect(
+    'costs', '€', 'Total costs',
+    is_standard=True, is_objective=True,
+    share_from_temporal={'CO2': 0.2, 'energy': 0.05},  # Costs receive contributions from other effects
+    share_from_nontemporal={'land': 100}  # €100 per m² land use
+)
+```
+This replaces the less intuitive `specific_share_to_other_effects_*` parameters and makes it clearer where effect contributions are coming from, rather than where they are going to.
+
 **Multi-year investments:**
 A flixopt model might be modeled with a "year" dimension.
 This enables modeling transformation pathways over multiple years with several investment decisions
@@ -117,6 +129,10 @@ The weighted sum of the total objective effect of each scenario is used as the o
   - `minimum_operation_per_hour` → `minimum_per_hour`
   - `maximum_operation_per_hour` → `maximum_per_hour`
 
+### 🔥 Removed
+* **Effect share parameters**: The old `specific_share_to_other_effects_*` parameters were replaced WITHOUT DEPRECATION
+  - `specific_share_to_other_effects_operation` → `share_from_temporal` (with inverted direction)
+  - `specific_share_to_other_effects_invest` → `share_from_nontemporal` (with inverted direction)
 
 ### 🐛 Fixed
 * Enhanced NetCDF I/O with proper attribute preservation for DataArrays

examples/01_Simple/simple_example.py

@@ -29,14 +29,14 @@
         description='Kosten',
         is_standard=True,  # standard effect: no explicit value needed for costs
         is_objective=True,  # Minimizing costs as the optimization objective
+        share_from_temporal={'CO2': 0.2},
     )
 
     # CO2 emissions effect with an associated cost impact
     CO2 = fx.Effect(
         label='CO2',
         unit='kg',
         description='CO2_e-Emissionen',
-        specific_share_to_other_effects_operation={costs.label: 0.2},
         maximum_operation_per_hour=1000,  # Max CO2 emissions per hour
     )
 

examples/02_Complex/complex_example.py

@@ -40,8 +40,8 @@
 
     # --- Define Effects ---
     # Specify effects related to costs, CO2 emissions, and primary energy consumption
-    Costs = fx.Effect('costs', '€', 'Kosten', is_standard=True, is_objective=True)
-    CO2 = fx.Effect('CO2', 'kg', 'CO2_e-Emissionen', specific_share_to_other_effects_operation={Costs.label: 0.2})
+    Costs = fx.Effect('costs', '€', 'Kosten', is_standard=True, is_objective=True, share_from_temporal={'CO2': 0.2})
+    CO2 = fx.Effect('CO2', 'kg', 'CO2_e-Emissionen')
     PE = fx.Effect('PE', 'kWh_PE', 'Primärenergie', maximum_total=3.5e3)
 
     # --- Define Components ---

examples/04_Scenarios/scenario_example.py

@@ -35,14 +35,14 @@
         description='Kosten',
         is_standard=True,  # standard effect: no explicit value needed for costs
         is_objective=True,  # Minimizing costs as the optimization objective
+        share_from_temporal={'CO2': 0.2},
     )
 
     # CO2 emissions effect with an associated cost impact
     CO2 = fx.Effect(
         label='CO2',
         unit='kg',
         description='CO2_e-Emissionen',
-        specific_share_to_other_effects_operation={costs.label: 0.2},
         maximum_operation_per_hour=1000,  # Max CO2 emissions per hour
     )
 

flixopt/effects.py

@@ -50,10 +50,10 @@ class Effect(Element):
             without effect dictionaries. Used for simplified effect specification (and less boilerplate code).
         is_objective: If True, this effect serves as the optimization objective function.
             Only one effect can be marked as objective per optimization.
-        specific_share_to_other_effects_operation: Operational cross-effect contributions.
-            Maps this effect's operational values to contributions to other effects
-        specific_share_to_other_effects_invest: Investment cross-effect contributions.
-            Maps this effect's investment values to contributions to other effects.
+        share_from_temporal: Temporal cross-effect contributions.
+            Maps temporal contributions from other effects to this effect
+        share_from_nontemporal: Nontemporal cross-effect contributions.
+            Maps nontemporal contributions from other effects to this effect.
         minimum_temporal: Minimum allowed total contribution across all timesteps.
         maximum_temporal: Maximum allowed total contribution across all timesteps.
         minimum_per_hour: Minimum allowed contribution per hour.
@@ -77,17 +77,21 @@ class Effect(Element):
         Basic cost objective:
 
         ```python
-        cost_effect = Effect(label='system_costs', unit='€', description='Total system costs', is_objective=True)
+        cost_effect = Effect(
+            label='system_costs',
+            unit='€',
+            description='Total system costs',
+            is_objective=True,
+        )
         ```
 
-        CO2 emissions with carbon pricing:
+        CO2 emissions:
 
         ```python
         co2_effect = Effect(
-            label='co2_emissions',
+            label='CO2',
             unit='kg_CO2',
             description='Carbon dioxide emissions',
-            specific_share_to_other_effects_operation={'costs': 50},  # €50/t_CO2
             maximum_total=1_000_000,  # 1000 t CO2 annual limit
         )
         ```
@@ -110,7 +114,21 @@ class Effect(Element):
             label='primary_energy',
             unit='kWh_primary',
             description='Primary energy consumption',
-            specific_share_to_other_effects_operation={'costs': 0.08},  # €0.08/kWh
+        )
+        ```
+
+       Cost objective with carbon and primary energy pricing:
+
+        ```python
+        cost_effect = Effect(
+            label='system_costs',
+            unit='€',
+            description='Total system costs',
+            is_objective=True,
+            share_from_temporal={
+                'primary_energy': 0.08,  # 0.08 €/kWh_primary
+                'CO2': 0.2,  # Carbon pricing: 0.2 €/kg_CO2 into costs if used on a cost effect
+            },
         )
         ```
 
@@ -137,8 +155,7 @@ class Effect(Element):
         across all contributions to each effect manually.
 
         Effects are accumulated as:
-        - Total = Σ(operational contributions) + Σ(investment contributions)
-        - Cross-effects add to target effects based on specific_share ratios
+        - Total = Σ(temporal contributions) + Σ(nontemporal contributions)
 
     """
 
@@ -150,8 +167,8 @@ def __init__(
         meta_data: dict | None = None,
         is_standard: bool = False,
         is_objective: bool = False,
-        specific_share_to_other_effects_operation: TemporalEffectsUser | None = None,
-        specific_share_to_other_effects_invest: NonTemporalEffectsUser | None = None,
+        share_from_temporal: TemporalEffectsUser | None = None,
+        share_from_nontemporal: NonTemporalEffectsUser | None = None,
         minimum_temporal: NonTemporalEffectsUser | None = None,
         maximum_temporal: NonTemporalEffectsUser | None = None,
         minimum_nontemporal: NonTemporalEffectsUser | None = None,
@@ -167,11 +184,9 @@ def __init__(
         self.description = description
         self.is_standard = is_standard
         self.is_objective = is_objective
-        self.specific_share_to_other_effects_operation: TemporalEffectsUser = (
-            specific_share_to_other_effects_operation if specific_share_to_other_effects_operation is not None else {}
-        )
-        self.specific_share_to_other_effects_invest: NonTemporalEffectsUser = (
-            specific_share_to_other_effects_invest if specific_share_to_other_effects_invest is not None else {}
+        self.share_from_temporal: TemporalEffectsUser = share_from_temporal if share_from_temporal is not None else {}
+        self.share_from_nontemporal: NonTemporalEffectsUser = (
+            share_from_nontemporal if share_from_nontemporal is not None else {}
         )
 
         # Handle backwards compatibility for deprecated parameters
@@ -393,8 +408,17 @@ def transform_data(self, flow_system: FlowSystem, name_prefix: str = '') -> None
 
         self.maximum_per_hour = flow_system.fit_to_model_coords(f'{prefix}|maximum_per_hour', self.maximum_per_hour)
 
-        self.specific_share_to_other_effects_operation = flow_system.fit_effects_to_model_coords(
-            f'{prefix}|operation->', self.specific_share_to_other_effects_operation, 'temporal'
+        self.share_from_temporal = flow_system.fit_effects_to_model_coords(
+            label_prefix=None,
+            effect_values=self.share_from_temporal,
+            label_suffix=f'(temporal)->{prefix}(temporal)',
+            dims=['time', 'year', 'scenario'],
+        )
+        self.share_from_nontemporal = flow_system.fit_effects_to_model_coords(
+            label_prefix=None,
+            effect_values=self.share_from_nontemporal,
+            label_suffix=f'(nontemporal)->{prefix}(nontemporal)',
+            dims=['year', 'scenario'],
         )
 
         self.minimum_temporal = flow_system.fit_to_model_coords(
@@ -417,12 +441,6 @@ def transform_data(self, flow_system: FlowSystem, name_prefix: str = '') -> None
         self.maximum_total = flow_system.fit_to_model_coords(
             f'{prefix}|maximum_total', self.maximum_total, dims=['year', 'scenario']
         )
-        self.specific_share_to_other_effects_invest = flow_system.fit_effects_to_model_coords(
-            f'{prefix}|operation->',
-            self.specific_share_to_other_effects_invest,
-            'operation',
-            dims=['year', 'scenario'],
-        )
 
     def create_model(self, model: FlowSystemModel) -> EffectModel:
         self._plausibility_checks()
@@ -569,6 +587,11 @@ def _plausibility_checks(self) -> None:
         # Check circular loops in effects:
         temporal, nontemporal = self.calculate_effect_share_factors()
 
+        # Validate all referenced sources exist
+        unknown = {src for src, _ in list(temporal.keys()) + list(nontemporal.keys()) if src not in self.effects}
+        if unknown:
+            raise KeyError(f'Unknown effects used in in effect share mappings: {sorted(unknown)}')
+
         temporal_cycles = detect_cycles(tuples_to_adjacency_list([key for key in temporal]))
         nontemporal_cycles = detect_cycles(tuples_to_adjacency_list([key for key in nontemporal]))
 
@@ -656,18 +679,20 @@ def calculate_effect_share_factors(
     ]:
         shares_nontemporal = {}
         for name, effect in self.effects.items():
-            if effect.specific_share_to_other_effects_invest:
-                shares_nontemporal[name] = {
-                    target: data for target, data in effect.specific_share_to_other_effects_invest.items()
-                }
+            if effect.share_from_nontemporal:
+                for source, data in effect.share_from_nontemporal.items():
+                    if source not in shares_nontemporal:
+                        shares_nontemporal[source] = {}
+                    shares_nontemporal[source][name] = data
         shares_nontemporal = calculate_all_conversion_paths(shares_nontemporal)
 
         shares_temporal = {}
         for name, effect in self.effects.items():
-            if effect.specific_share_to_other_effects_operation:
-                shares_temporal[name] = {
-                    target: data for target, data in effect.specific_share_to_other_effects_operation.items()
-                }
+            if effect.share_from_temporal:
+                for source, data in effect.share_from_temporal.items():
+                    if source not in shares_temporal:
+                        shares_temporal[source] = {}
+                    shares_temporal[source][name] = data
         shares_temporal = calculate_all_conversion_paths(shares_temporal)
 
         return shares_temporal, shares_nontemporal
@@ -726,19 +751,19 @@ def _do_modeling(self):
         )
 
     def _add_share_between_effects(self):
-        for origin_effect in self.effects:
-            # 1. temporal: -> hier sind es Zeitreihen (share_TS)
-            for target_effect, time_series in origin_effect.specific_share_to_other_effects_operation.items():
-                self.effects[target_effect].submodel.temporal.add_share(
-                    origin_effect.submodel.temporal.label_full,
-                    origin_effect.submodel.temporal.total_per_timestep * time_series,
+        for target_effect in self.effects:
+            # 1. temporal: <- receiving temporal shares from other effects
+            for source_effect, time_series in target_effect.share_from_temporal.items():
+                target_effect.submodel.temporal.add_share(
+                    self.effects[source_effect].submodel.temporal.label_full,
+                    self.effects[source_effect].submodel.temporal.total_per_timestep * time_series,
                     dims=('time', 'year', 'scenario'),
                 )
-            # 2. nontemporal:    -> hier ist es Scalar (share)
-            for target_effect, factor in origin_effect.specific_share_to_other_effects_invest.items():
-                self.effects[target_effect].submodel.nontemporal.add_share(
-                    origin_effect.submodel.nontemporal.label_full,
-                    origin_effect.submodel.nontemporal.total * factor,
+            # 2. nontemporal: <- receiving nontemporal shares from other effects
+            for source_effect, factor in target_effect.share_from_nontemporal.items():
+                target_effect.submodel.nontemporal.add_share(
+                    self.effects[source_effect].submodel.nontemporal.label_full,
+                    self.effects[source_effect].submodel.nontemporal.total * factor,
                     dims=('year', 'scenario'),
                 )
 

flixopt/flow_system.py

@@ -407,6 +407,7 @@ def fit_effects_to_model_coords(
         effect_values: TemporalEffectsUser | NonTemporalEffectsUser | None,
         label_suffix: str | None = None,
         dims: Collection[FlowSystemDimensions] | None = None,
+        delimiter: str = '|',
     ) -> TemporalEffects | NonTemporalEffects | None:
         """
         Transform EffectValues from the user to Internal Datatypes aligned with model coordinates.
@@ -418,7 +419,7 @@ def fit_effects_to_model_coords(
 
         return {
             effect: self.fit_to_model_coords(
-                '|'.join(filter(None, [label_prefix, effect, label_suffix])),
+                str(delimiter).join(filter(None, [label_prefix, effect, label_suffix])),
                 value,
                 dims=dims,
             )

tests/conftest.py

@@ -100,17 +100,12 @@ def costs():
         return fx.Effect('costs', '€', 'Kosten', is_standard=True, is_objective=True)
 
     @staticmethod
-    def co2():
-        return fx.Effect('CO2', 'kg', 'CO2_e-Emissionen')
+    def costs_with_co2_share():
+        return fx.Effect('costs', '€', 'Kosten', is_standard=True, is_objective=True, share_from_temporal={'CO2': 0.2})
 
     @staticmethod
-    def co2_with_costs_share():
-        return fx.Effect(
-            'CO2',
-            'kg',
-            'CO2_e-Emissionen',
-            specific_share_to_other_effects_operation={'costs': 0.2},
-        )
+    def co2():
+        return fx.Effect('CO2', 'kg', 'CO2_e-Emissionen')
 
     @staticmethod
     def primary_energy():
@@ -388,8 +383,8 @@ def simple_flow_system() -> fx.FlowSystem:
     base_timesteps = pd.date_range('2020-01-01', periods=9, freq='h', name='time')
 
     # Define effects
-    costs = Effects.costs()
-    co2 = Effects.co2_with_costs_share()
+    costs = Effects.costs_with_co2_share()
+    co2 = Effects.co2()
     co2.maximum_operation_per_hour = 1000
 
     # Create components
@@ -418,8 +413,8 @@ def simple_flow_system_scenarios() -> fx.FlowSystem:
     base_timesteps = pd.date_range('2020-01-01', periods=9, freq='h', name='time')
 
     # Define effects
-    costs = Effects.costs()
-    co2 = Effects.co2_with_costs_share()
+    costs = Effects.costs_with_co2_share()
+    co2 = Effects.co2()
     co2.maximum_operation_per_hour = 1000
 
     # Create components
@@ -471,7 +466,7 @@ def flow_system_complex() -> fx.FlowSystem:
     # Define the components and flow_system
     costs = Effects.costs()
     co2 = Effects.co2()
-    co2.specific_share_to_other_effects_operation = {'costs': 0.2}
+    costs.share_from_temporal = {'CO2': 0.2}
     pe = Effects.primary_energy()
     pe.maximum_total = 3.5e3