V3.0.0/invest #294
V3.0.0/invest #294
Conversation
* Change fit_to_model_coords to work with a Collection of dims * Improve fit_to_model_coords
This will only be an issue if the used effect is part of the objective. And it might be confusing when analyzing the effect results and using ffill on the years. Unfortunately I cannot see this resolved without changing the effects structure. |
# Conflicts: # flixopt/flow_system.py # flixopt/interface.py # flixopt/structure.py
|
Ich weiß nicht, ob ich es ganz verstehe, was du meinst, aber in diesem Zuge müsste man ja entsprechend Investitionszeitpunkt dann auch den Restwert am Ende des Betrachtungszeitraums mitdenken in der Zielfunktion. Lass uns das gerne bei Gelegenheit nochmal vertiefen. |
|
Ich schreib das Konzept mal zusammen die nächsten Tage. Auch für die Leute von @oemof. |
# Conflicts: # .github/ISSUE_TEMPLATE/bug_report.yml # .github/ISSUE_TEMPLATE/config.yml # .github/ISSUE_TEMPLATE/feature_request.yml # .github/pull_request_template.md # CHANGELOG.md # README.md # docs/SUMMARY.md # docs/contribute.md # flixopt/calculation.py # flixopt/effects.py # flixopt/elements.py # flixopt/features.py # flixopt/flow_system.py # flixopt/interface.py # flixopt/modeling.py # flixopt/structure.py # pyproject.toml
WalkthroughAdds a year-aware investment feature: new InvestTimingParameters and InvestmentTimingModel, year_of_investment dimension, per-year durations, and duration-aware primitives. Broad updates wire timing-based investments through Flow, FlowSystem, and features. Multiple transform_data signatures gain an optional name_prefix. Tests added for investment timing and annualized effects. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
actor U as User
participant F as Flow
participant FS as FlowSystem
participant FM as FlowModel
participant ITM as InvestmentTimingModel
participant MP as ModelingPrimitives
participant SLV as Solver
U->>F: new Flow(size=InvestTimingParameters)
F->>FS: transform_data(flow_system, name_prefix)
FS-->>F: fitted data (incl. year_of_investment, years_per_year)
F->>FM: _create_investment_model(...)
FM->>ITM: construct with parameters
ITM->>MP: consecutive_duration_tracking(duration_dim='time', duration_per_step)
ITM->>MP: link_changes_to_level_with_binaries(coord='year')
ITM-->>FM: variables + constraints
FM-->>SLV: model assembled
SLV-->>U: results (size, investment_occurs, effects)
note over ITM,MP: New/changed duration- and year-aware interactions
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✨ Finishing touches
🧪 Generate unit tests
Comment |
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (13)
flixopt/effects.py (2)
195-203: Fix type aliases: Non-temporal invest shares are no longer scalars.
NonTemporalEffectsUsercorrectly switched toNonTemporalDataUser, butNonTemporalEffectsand the return type ofcreate_effect_values_dictstill assume scalars. This is inconsistent and risks wrong typing/usages downstream.Apply:
- from .core import NonTemporalDataUser, Scalar, TemporalData, TemporalDataUser + from .core import NonTemporalData, NonTemporalDataUser, Scalar, TemporalData, TemporalDataUser-NonTemporalEffects = Dict[str, Scalar] -""" This datatype is used internally to handle scalar shares to an effect. """ +NonTemporalEffects = Dict[str, NonTemporalData] +""" This datatype is used internally to handle non-temporal (year/scenario) shares to an effect. """- ) -> Optional[Dict[str, Union[Scalar, TemporalDataUser]]]: + ) -> Optional[Dict[str, Union[NonTemporalDataUser, TemporalDataUser]]]:Also applies to: 16-16
428-430: Objective double-counts investments when years aggregate multiple modeled years.Multiplying
total = operation + investby year-weights inflates one-off invest effects across represented years — exactly the major issue called out in the PR.Separate weighting: use per-(year,scenario) weights for operation, and scenario-only weights for invest:
- self._model.add_objective( - (self.effects.objective_effect.submodel.total * self._model.weights).sum() + self.penalty.total.sum() - ) + op_term = (self.effects.objective_effect.submodel.operation.total * self._model.weights).sum() + inv_weights = self._model.weights.sum('year') # scenario-only weighting + inv_term = (self.effects.objective_effect.submodel.invest.total * inv_weights).sum() + self._model.add_objective(op_term + inv_term + self.penalty.total.sum())If you prefer centralized accessors, add
weights_operationandweights_investon FlowSystemModel and use them here. I can wire that up.flixopt/flow_system.py (1)
66-81: Persist and restore years_of_last_year.
years_of_last_yearinfluencesyears_per_year, but it’s neither stored on the instance nor restored from datasets.Apply:
@@ - years_of_last_year: Optional[int] = None, + years_of_last_year: Optional[int] = None, @@ - if years is None: + self.years_of_last_year = years_of_last_year + if years is None: self.years, self.years_per_year, self.years_of_investment = None, None, None else: self.years = self._validate_years(years) - self.years_per_year = self.calculate_years_per_year(self.years, years_of_last_year) + self.years_per_year = self.calculate_years_per_year(self.years, self.years_of_last_year) self.years_of_investment = self.years.rename('year_of_investment')@@ def from_dataset(cls, ds: xr.Dataset) -> 'FlowSystem': - flow_system = cls( + flow_system = cls( timesteps=ds.indexes['time'], years=ds.indexes.get('year'), scenarios=ds.indexes.get('scenario'), + years_of_last_year=reference_structure.get('years_of_last_year'), weights=cls._resolve_dataarray_reference(reference_structure['weights'], arrays_dict) if 'weights' in reference_structure else None,Also applies to: 88-94, 282-291
flixopt/structure.py (1)
232-243: Interface.transform_data signature mismatch — fix Effects.transform_dataflixopt/effects.py:94 defines
def transform_data(self, flow_system)— change todef transform_data(self, flow_system: 'FlowSystem', name_prefix: str = '') -> Noneand ensure the method handles/propagatesname_prefix; otherwise callers that pass a prefix will raise TypeError.flixopt/__init__.py (1)
9-38: Add alias YearAwareInvestParameters to match tests and PR wordingTests in this PR use YearAwareInvestParameters. Provide a stable alias to avoid breakage and reduce churn.
Apply this diff:
from .commons import ( @@ - InvestParameters, - InvestTimingParameters, + InvestParameters, + InvestTimingParameters, @@ ) CONFIG.load_config() + +# Backward-compat/test-friendly alias +YearAwareInvestParameters = InvestTimingParametersflixopt/calculation.py (1)
486-491: SegmentedCalculation: warn for both investment modelsThe warning currently checks only InvestmentModel. Include InvestmentTimingModel to avoid silent misuse.
Apply this diff:
- if isinstance(model, InvestmentModel) + if isinstance(model, (InvestmentModel, InvestmentTimingModel))flixopt/features.py (1)
1-10: Forward ref in type hints: prevent runtime NameErrorInvestmentModel annotates piecewise_effects: Optional[PiecewiseEffectsModel] before the class is defined. Without postponed evaluation, this can fail at import time on older Python versions.
Apply this diff at file top:
+from __future__ import annotationsAlternatively, quote the annotation: Optional["PiecewiseEffectsModel"].
flixopt/modeling.py (1)
225-235: consecutive_duration_tracking: require duration_per_step and support scalarsduration_per_step may be None or scalar; current .sum() call would error. Enforce presence and coerce scalars to a DataArray aligned to state_variable.
Apply this diff:
def consecutive_duration_tracking( @@ - duration_dim: str = 'time', - duration_per_step: Union[Scalar, xr.DataArray] = None, + duration_dim: str = 'time', + duration_per_step: Union[Scalar, xr.DataArray] = None, @@ - mega = duration_per_step.sum(duration_dim) + previous_duration # Big-M value + if duration_per_step is None: + raise ValueError("duration_per_step must be provided") + if not isinstance(duration_per_step, xr.DataArray): + duration_per_step = xr.full_like(state_variable, fill_value=float(duration_per_step)) + mega = duration_per_step.sum(duration_dim) + previous_duration # Big-M valueflixopt/elements.py (5)
73-79: name_prefix is accepted but not used; propagate it and pass to child flows.
Currently ignored, which defeats the purpose of hierarchical labeling.Apply:
- def transform_data(self, flow_system: 'FlowSystem', name_prefix: str = '') -> None: - if self.on_off_parameters is not None: - self.on_off_parameters.transform_data(flow_system, self.label_full) - - for flow in self.inputs + self.outputs: - flow.transform_data(flow_system) + def transform_data(self, flow_system: 'FlowSystem', name_prefix: str = '') -> None: + label_base = f'{name_prefix}|{self.label_full}' if name_prefix else self.label_full + if self.on_off_parameters is not None: + self.on_off_parameters.transform_data(flow_system, label_base) + for flow in self.inputs + self.outputs: + flow.transform_data(flow_system, name_prefix=label_base)
121-125: name_prefix is accepted but not used in Bus.transform_data.Apply:
- def transform_data(self, flow_system: 'FlowSystem', name_prefix: str = '') -> None: - self.excess_penalty_per_flow_hour = flow_system.fit_to_model_coords( - f'{self.label_full}|excess_penalty_per_flow_hour', self.excess_penalty_per_flow_hour - ) + def transform_data(self, flow_system: 'FlowSystem', name_prefix: str = '') -> None: + prefix_label = f'{name_prefix}|{self.label_full}' if name_prefix else self.label_full + self.excess_penalty_per_flow_hour = flow_system.fit_to_model_coords( + f'{prefix_label}|excess_penalty_per_flow_hour', self.excess_penalty_per_flow_hour + )
241-266: name_prefix is unused in Flow.transform_data; compute and use a combined prefix.
Without this, upstream prefixes have no effect.Apply:
- def transform_data(self, flow_system: 'FlowSystem', name_prefix: str = '') -> None: - self.relative_minimum = flow_system.fit_to_model_coords( - f'{self.label_full}|relative_minimum', self.relative_minimum - ) + def transform_data(self, flow_system: 'FlowSystem', name_prefix: str = '') -> None: + prefix_label = f'{name_prefix}|{self.label_full}' if name_prefix else self.label_full + self.relative_minimum = flow_system.fit_to_model_coords( + f'{prefix_label}|relative_minimum', self.relative_minimum + ) self.relative_maximum = flow_system.fit_to_model_coords( - f'{self.label_full}|relative_maximum', self.relative_maximum + f'{prefix_label}|relative_maximum', self.relative_maximum ) self.fixed_relative_profile = flow_system.fit_to_model_coords( - f'{self.label_full}|fixed_relative_profile', self.fixed_relative_profile + f'{prefix_label}|fixed_relative_profile', self.fixed_relative_profile ) self.effects_per_flow_hour = flow_system.fit_effects_to_model_coords( - self.label_full, self.effects_per_flow_hour, 'per_flow_hour' + prefix_label, self.effects_per_flow_hour, 'per_flow_hour' ) self.flow_hours_total_max = flow_system.fit_to_model_coords( - f'{self.label_full}|flow_hours_total_max', self.flow_hours_total_max, dims=['year', 'scenario'] + f'{prefix_label}|flow_hours_total_max', self.flow_hours_total_max, dims=['year', 'scenario'] ) self.flow_hours_total_min = flow_system.fit_to_model_coords( - f'{self.label_full}|flow_hours_total_min', self.flow_hours_total_min, dims=['year', 'scenario'] + f'{prefix_label}|flow_hours_total_min', self.flow_hours_total_min, dims=['year', 'scenario'] ) self.load_factor_max = flow_system.fit_to_model_coords( - f'{self.label_full}|load_factor_max', self.load_factor_max, dims=['year', 'scenario'] + f'{prefix_label}|load_factor_max', self.load_factor_max, dims=['year', 'scenario'] ) self.load_factor_min = flow_system.fit_to_model_coords( - f'{self.label_full}|load_factor_min', self.load_factor_min, dims=['year', 'scenario'] + f'{prefix_label}|load_factor_min', self.load_factor_min, dims=['year', 'scenario'] )
267-273: Propagate prefix to nested transforms and scalar size.Apply:
- if self.on_off_parameters is not None: - self.on_off_parameters.transform_data(flow_system, self.label_full) + if self.on_off_parameters is not None: + self.on_off_parameters.transform_data(flow_system, prefix_label) if isinstance(self.size, (InvestParameters, InvestTimingParameters)): - self.size.transform_data(flow_system, self.label_full) + self.size.transform_data(flow_system, prefix_label) else: - self.size = flow_system.fit_to_model_coords(f'{self.label_full}|size', self.size, dims=['year', 'scenario']) + self.size = flow_system.fit_to_model_coords( + f'{prefix_label}|size', self.size, dims=['year', 'scenario'] + )
318-326: size_is_fixed and invest_is_optional ignore InvestTimingParameters.
This will return misleading values with timing-based investments.Apply:
def size_is_fixed(self) -> bool: - # Wenn kein InvestParameters existiert --> True; Wenn Investparameter, den Wert davon nehmen - return False if (isinstance(self.size, InvestParameters) and self.size.fixed_size is None) else True + # True for scalar sizes. For investments, True only if a fixed_size is provided. + if isinstance(self.size, (InvestParameters, InvestTimingParameters)): + return getattr(self.size, 'fixed_size', None) is not None + return True def invest_is_optional(self) -> bool: - # Wenn kein InvestParameters existiert: # Investment ist nicht optional -> Keine Variable --> False - return False if (isinstance(self.size, InvestParameters) and not self.size.optional) else True + # No investment object => not optional. + if isinstance(self.size, (InvestParameters, InvestTimingParameters)): + return bool(getattr(self.size, 'optional', False)) + return False
♻️ Duplicate comments (4)
tests/test_models.py (4)
133-141: Same rename here- params = fx.YearAwareInvestParameters(fixed_size=50) + params = fx.InvestTimingParameters(fixed_size=50)
142-154: Same rename here- params = fx.YearAwareInvestParameters( + params = fx.InvestTimingParameters( fixed_year_of_investment=2, minimum_duration=3, maximum_duration=5, earliest_year_of_decommissioning=4, )
155-170: Same rename here- params = fx.YearAwareInvestParameters( + params = fx.InvestTimingParameters( effects_of_investment={'costs': 1000}, effects_of_investment_per_size={'costs': 100}, allow_divestment=True, effects_of_divestment={'costs': 500}, effects_of_divestment_per_size={'costs': 50}, )
171-184: Same rename here- params_fixed = fx.YearAwareInvestParameters(fixed_size=50) + params_fixed = fx.InvestTimingParameters(fixed_size=50) @@ - params_range = fx.YearAwareInvestParameters(minimum_size=10, maximum_size=100) + params_range = fx.InvestTimingParameters(minimum_size=10, maximum_size=100)
🧹 Nitpick comments (22)
flixopt/effects.py (2)
450-519: Path explosion risk in BFS conversion accumulation.Current BFS accumulates all indirect paths and sums factors; dense graphs can grow exponentially. If shares are meant to be unique per target, consider shortest-path (or first valid path) or detect/limit multiple paths per (origin,target).
94-133: Effect.transform_data signature not aligned with Interface.Base
Interface.transform_datanow acceptsname_prefix;Effect.transform_dataignores it. Unify signatures to avoid surprises for future callers.- def transform_data(self, flow_system: 'FlowSystem'): + def transform_data(self, flow_system: 'FlowSystem', name_prefix: str = '') -> None:flixopt/flow_system.py (2)
187-197: Docstring nit: years, not timesteps.“Calculate duration of each timestep” → “each year.” No functional change.
366-381: with_year_of_investment: clarify and guard.Good addition. Consider guarding against accidental use when
self.years is Noneto raise a clear error instead of broadcasting an empty coord.+ if with_year_of_investment and 'year' not in coords: + raise ConversionError('with_year_of_investment=True requires "year" dimension in coords.')Also applies to: 388-395
flixopt/components.py (3)
97-104: Forward name_prefix consistently.
LinearConverter.transform_dataacceptsname_prefixbut callssuper().transform_data(flow_system)without forwarding. Align for consistency (even if unused today).- super().transform_data(flow_system) + super().transform_data(flow_system, name_prefix)
210-251: Same: pass name_prefix in Storage.transform_data.- super().transform_data(flow_system) + super().transform_data(flow_system, name_prefix)
396-404: Same: pass name_prefix in Transmission.transform_data.- super().transform_data(flow_system) + super().transform_data(flow_system, name_prefix)flixopt/interface.py (2)
410-489: No-broadcast guard for investment-year effects is good; error message could hintflow_system.years_of_investment.Optional string tweak; logic is fine.
568-596: Consistent label argument names.Use explicit keywords in
fit_effects_to_model_coordscalls (label_prefix,label_suffix) for readability and future-proofing; no behavior change.flixopt/commons.py (1)
59-60: Add YearAwareInvestParameters alias to allKeeps public API aligned with tests and PR terminology.
Apply this diff:
__all__ = [ @@ - 'InvestTimingParameters', + 'InvestTimingParameters', + 'YearAwareInvestParameters', ] + +# Alias for compatibility with tests/terminology +YearAwareInvestParameters = InvestTimingParameterstests/test_models.py (2)
199-214: Test setup nit: drop no-op specific_effects and prefer OSS solver for CI
- Multiplying by -0 is unnecessary; pass 0 or omit specific_effects.
- Gurobi may be unavailable in CI. Prefer an OSS solver (e.g., HiGHS) if supported by flixopt.solvers.
- specific_effects=xr.DataArray( - [25, 30, 35, 40, 45, 50, 55, 60], - coords=(flow_system.years,), - ) - * -0, + # No specific effects here + # specific_effects=0, @@ - calculation.solve(fx.solvers.GurobiSolver(0, 60)) + # Prefer OSS solver in tests if available + # calculation.solve(fx.solvers.HighsSolver()) + calculation.solve(fx.solvers.GurobiSolver(0, 60))If HiGHS/CBC are available in flixopt.solvers, switch to them to avoid vendor lock-in for tests.
220-231: Replace prints with minimal assertionsEnsure the test fails meaningfully instead of relying on prints.
Example assertions you can add after solve:
- assert ds['is_invested'].sum('year') <= 1
- assert (calculation.results.solution['costs(invest)|total'] <= 0).all()
Do you want me to propose concrete assertions based on expected values?flixopt/features.py (3)
14-18: Interface attribute naming: confirm parameter field namesInvestmentTimingModel uses parameters.fix_effects and parameters.specific_effects. Tests and PR text mention effects_of_investment, effects_of_investment_per_size, etc. Ensure InvestTimingParameters exposes both (aliases), or adapt here.
I can add a thin compatibility layer in InvestTimingParameters (properties mapping new→old names) if desired.
212-234: Lifetime computation: check minimum-lifetime handling at horizon endCurrent lb constraint doesn’t enforce minimum lifetime if the asset is still invested in the last modeled year. If that’s intentional, add a comment; else enforce a lower bound when decommissioning occurs or investment persists to horizon end.
I can propose a conditional bound using last-period state if you want it enforced in both cases.
343-357: Docstrings: size_increase/size_decrease aren’t binaryThese are continuous nonnegative variables. Update docstrings to avoid confusion.
- def size_decrease(self) -> linopy.Variable: - """Binary decrease decision variable""" + def size_decrease(self) -> linopy.Variable: + """Nonnegative decrease variable""" @@ - def size_increase(self) -> linopy.Variable: - """Binary increase decision variable""" + def size_increase(self) -> linopy.Variable: + """Nonnegative increase variable"""flixopt/modeling.py (1)
597-606: Typo in error message class nameFunction lives under ModelingPrimitives; message mentions BoundingPatterns.
- raise ValueError('BoundingPatterns.continuous_transition_bounds() can only be used with a Submodel') + raise ValueError('ModelingPrimitives.continuous_transition_bounds() can only be used with a Submodel')flixopt/elements.py (6)
161-166: Update docstring to reflect InvestTimingParameters.
The argument docs still only mention InvestParameters.Apply:
- size: size of the flow. If InvestmentParameters is used, size is optimized. + size: size of the flow. If InvestParameters or InvestTimingParameters are used, size is optimized.
379-402: Investment model selection: message improvement and note on effects once-per-investment.
- Error message mentions “InvestParameters” only.
- Ensure InvestmentTimingModel registers investment effects once per actual investment year, not per represented years (addresses the PR’s double-counting risk).
Apply:
- else: - raise ValueError(f'Invalid InvestParameters type: {type(self.element.size)}') + else: + raise ValueError(f'Unsupported size type for investment: {type(self.element.size)}')
434-441: Use non-deprecated alias for scaling_variable.Apply:
- scaling_variable=self._investment.size, + scaling_variable=self.investment.size,
175-186: Docstring typos and clarity.
Minor fixes improve readability and accuracy.Apply:
- label: The label of the FLow. Used to identify it in the FlowSystem. Its `full_label` consists of the label of the Component and the label of the Flow. - bus: blabel of the bus the flow is connected to. - size: size of the flow. If InvestmentParameters is used, size is optimized. + label: The label of the Flow. Used to identify it in the FlowSystem. Its `full_label` consists of the label of the Component and the label of the Flow. + bus: label of the bus the flow is connected to. + size: size of the flow. If InvestParameters or InvestTimingParameters are used, size is optimized.
309-311: Fix typo in error message.Apply:
- f'Different values in different years or scenarios are not yetsupported.' + f'Different values in different years or scenarios are not yet supported.'
548-553: Return type and docstring for investment property.
It may be InvestmentModel or InvestmentTimingModel; docstring is misleading.Apply:
- def investment(self) -> Optional[InvestmentModel]: - """OnOff feature""" + def investment(self) -> Optional[Union[InvestmentModel, InvestmentTimingModel]]: + """Investment feature"""
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (12)
flixopt/__init__.py(1 hunks)flixopt/calculation.py(2 hunks)flixopt/commons.py(2 hunks)flixopt/components.py(3 hunks)flixopt/effects.py(2 hunks)flixopt/elements.py(10 hunks)flixopt/features.py(7 hunks)flixopt/flow_system.py(8 hunks)flixopt/interface.py(8 hunks)flixopt/modeling.py(4 hunks)flixopt/structure.py(2 hunks)tests/test_models.py(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (4)
- GitHub Check: test (3.12)
- GitHub Check: test (3.13)
- GitHub Check: test (3.10)
- GitHub Check: test (3.11)
🔇 Additional comments (18)
flixopt/effects.py (1)
432-448: Broadcast safety for invest shares.factor may be a non-temporal DataArray over ['year','scenario']; explicitly align/reindex it to origin_effect.submodel.invest.total coords (e.g., xarray.align / reindex_like or explicit broadcast) before calling add_share to avoid silent broadcasting mismatches.
File: flixopt/effects.py lines 432–448
Automated repo search returned no results — manual verification of other call-sites or that add_share enforces coord alignment is required.
flixopt/components.py (1)
439-447: Equation form is correct; comment matches.The rearranged constraint with
lhs += absolute_lossesstill matches the documented equation.flixopt/interface.py (1)
74-77: Piece/Piecewise/PiecewiseConversion: transform_data propagation looks correct.Signatures and prefixing are consistent.
Also applies to: 107-110, 138-142
flixopt/__init__.py (1)
19-20: Public API: OK to expose InvestTimingParametersRe-export looks good and matches the new feature.
flixopt/commons.py (1)
21-29: Expose InvestTimingParameters: LGTMThe import block and exposure are consistent.
flixopt/calculation.py (2)
31-33: Include InvestmentTimingModel: LGTMType inclusion in imports is correct.
116-125: Main results: correctly include timing-based investmentsExtending isinstance filter to InvestmentTimingModel is correct.
flixopt/features.py (5)
78-106: InvestmentModel effects logic: OKEffects hookup is correct and consistent with existing API.
133-141: Great: single invest/decommission windowModeling matches the intended contiguous on-period between invest and decommission.
161-181: Bounds with state for size: OKSize controlled by is_invested with [min,max] is correct.
195-211: At most one invest and one decommission: OKSum ≤ 1 constraints are correct for a single asset cycle.
235-260: size/Δsize linkage: correct pattern choiceLinking increases/decreases to binaries with exclusivity is solid.
flixopt/modeling.py (2)
664-675: Nice addition: link_changes_to_level_with_binariesAPI is useful and well-composed.
458-466: OnOff: duration tracking: LGTMPassing duration_per_step and duration_dim is correct and future-proof.
Also applies to: 476-479
flixopt/elements.py (4)
16-18: Imports look correct and consistent with new timing model.
279-287: Good exclusion of InvestTimingParameters from the default-size warning.
451-452: with_investment correctly recognizes both investment parameter types.
524-530: Resolved — InvestTimingParameters (year-aware invest class) exposes minimum_or_fixed_size and maximum_or_fixed_size. Properties are implemented in flixopt/interface.py and covered by tests; no runtime error expected.
| if self.parameters.fixed_effects_by_investment_year: | ||
| # Effects depending on when the investment is made | ||
| remapped_variable = self.investment_occurs.rename({'year': 'year_of_investment'}) | ||
|
|
||
| self._model.effects.add_share_to_effects( | ||
| name=self.label_of_element, | ||
| expressions={ | ||
| effect: -self.is_invested * factor + factor | ||
| for effect, factor in self.parameters.divest_effects.items() | ||
| effect: (remapped_variable * factor).sum('year_of_investment') | ||
| for effect, factor in self.parameters.fixed_effects_by_investment_year.items() | ||
| }, | ||
| target='invest', | ||
| ) | ||
|
|
||
| if self.parameters.specific_effects: | ||
| if self.parameters.specific_effects_by_investment_year: | ||
| # Annual effects proportional to investment size | ||
| remapped_variable = self.size_increase.rename({'year': 'year_of_investment'}) | ||
|
|
||
| self._model.effects.add_share_to_effects( | ||
| name=self.label_of_element, | ||
| expressions={effect: self.size * factor for effect, factor in self.parameters.specific_effects.items()}, | ||
| expressions={ | ||
| effect: (remapped_variable * factor).sum('year_of_investment') | ||
| for effect, factor in self.parameters.specific_effects_by_investment_year.items() | ||
| }, | ||
| target='invest', | ||
| ) | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Avoid double counting one-off effects when representative years have weights
For fixed_effects_by_investment_year, summing only over 'year_of_investment' leaves a 'year' dim if the factor carries it, which can trigger replication when totals get weighted by years_per_year. Sum over both dims present in the factor to ensure one-shot effects remain one-shot.
Apply this diff:
- self._model.effects.add_share_to_effects(
- name=self.label_of_element,
- expressions={
- effect: (remapped_variable * factor).sum('year_of_investment')
- for effect, factor in self.parameters.fixed_effects_by_investment_year.items()
- },
- target='invest',
- )
+ exprs = {}
+ for effect, factor in self.parameters.fixed_effects_by_investment_year.items():
+ expr = remapped_variable * factor
+ # Sum across any investment-year dim and drop plain 'year' if present to avoid replication
+ if hasattr(factor, "dims"):
+ for dim in ('year_of_investment', 'year'):
+ if dim in getattr(factor, "dims", ()):
+ expr = expr.sum(dim)
+ else:
+ expr = expr.sum('year_of_investment')
+ exprs[effect] = expr
+ self._model.effects.add_share_to_effects(
+ name=self.label_of_element, expressions=exprs, target='invest'
+ )Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
flixopt/features.py around lines 281 to 306: the current
.sum('year_of_investment') can leave a remaining 'year' dimension when the
factor carries it, causing double-counting with representative-year weights;
change the summation to collapse both dims present by summing over a tuple of
dims that exist (e.g. create dims_to_sum = tuple(d for d in
('year_of_investment','year') if d in factor.dims) and call .sum(dims_to_sum)
for each expression) so one-off effects are summed out correctly and not
replicated by year weights.
| def duration(self) -> linopy.Variable: | ||
| """Investment duration variable""" | ||
| return self._variables['duration'] | ||
|
|
There was a problem hiding this comment.
Property ‘duration’ references a non-existent variable
No variable named 'duration' is created; property will KeyError. Remove it or create and populate 'duration'.
Quick fix: remove the property until implemented.
- @property
- def duration(self) -> linopy.Variable:
- """Investment duration variable"""
- return self._variables['duration']If you want me to implement a duration variable tied to lifetime/years, I can propose it.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def duration(self) -> linopy.Variable: | |
| """Investment duration variable""" | |
| return self._variables['duration'] |
🤖 Prompt for AI Agents
In flixopt/features.py around lines 379-382, the duration property returns
self._variables['duration'] but no 'duration' key is ever created, causing a
KeyError; either remove this property until a duration variable is implemented,
or add code where self._variables is populated to create and assign a
linopy.Variable named 'duration' (and ensure it is kept in sync with the model's
lifetime/years semantics), i.e., create the variable with the correct shape/type
and insert it into self._variables['duration'] where other variables are
initialized.
| ds = self.to_dataset() | ||
|
|
||
| # Build indexers dict from non-None parameters | ||
| indexers = {} | ||
| if time is not None: | ||
| indexers['time'] = time | ||
| if year is not None: | ||
| indexers['year'] = year | ||
| if 'year_of_investment' in ds.dims: | ||
| indexers['year_of_investment'] = year | ||
| if scenario is not None: | ||
| indexers['scenario'] = scenario | ||
|
|
||
| if not indexers: | ||
| return self.copy() # Return a copy when no selection | ||
|
|
||
| selected_dataset = self.to_dataset().sel(**indexers) | ||
| selected_dataset = ds.sel(**indexers) | ||
| if 'year_of_investment' in selected_dataset.coords and selected_dataset.coords['year_of_investment'].size == 1: | ||
| logger.critical( | ||
| 'Selected a single year while using InvestmentTiming. This is not supported and will lead to Errors ' | ||
| 'when trying to create a Calculation from this FlowSystem. Please select multiple years instead, ' | ||
| 'or remove the InvestmentTimingParameters.' | ||
| ) | ||
| return self.__class__.from_dataset(selected_dataset) | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Selecting a single year under InvestmentTiming: escalate to exception.
A logger.critical without raising still returns a FlowSystem that will fail later. Raise early to simplify debugging.
- if 'year_of_investment' in selected_dataset.coords and selected_dataset.coords['year_of_investment'].size == 1:
- logger.critical(
+ if 'year_of_investment' in selected_dataset.coords and selected_dataset.coords['year_of_investment'].size == 1:
+ raise ValueError(
'Selected a single year while using InvestmentTiming. This is not supported and will lead to Errors '
'when trying to create a Calculation from this FlowSystem. Please select multiple years instead, '
'or remove the InvestmentTimingParameters.'
)Also applies to: 812-835
🤖 Prompt for AI Agents
In flixopt/flow_system.py around lines 767-791 and 812-835, the code currently
logs a critical error when a selection results in a single 'year_of_investment'
but continues and returns a FlowSystem that will fail later; change this to
raise an exception immediately (e.g., raise ValueError with the same descriptive
message) instead of only logging, or log then raise, so the caller fails fast
and debugging is easier.
| def _plausibility_checks(self, flow_system): | ||
| """Validate parameter consistency.""" | ||
| if flow_system.years is None: | ||
| raise ValueError("InvestTimingParameters requires the flow_system to have a 'years' dimension.") | ||
|
|
||
| if (self.force_investment.sum('year') > 1).any(): | ||
| raise ValueError('force_investment can only be True for a single year.') | ||
| if (self.force_decommissioning.sum('year') > 1).any(): | ||
| raise ValueError('force_decommissioning can only be True for a single year.') | ||
|
|
||
| if (self.force_investment.sum('year') == 1).any() and (self.force_decommissioning.sum('year') == 1).any(): | ||
| year_of_forced_investment = ( | ||
| self.force_investment.where(self.force_investment) * self.force_investment.year | ||
| ).sum('year') | ||
| year_of_forced_decommissioning = ( | ||
| self.force_decommissioning.where(self.force_decommissioning) * self.force_decommissioning.year | ||
| ).sum('year') | ||
| if not (year_of_forced_investment < year_of_forced_decommissioning).all(): | ||
| raise ValueError( | ||
| f'force_investment needs to be before force_decommissioning. Got:\n' | ||
| f'{self.force_investment}\nand\n{self.force_decommissioning}' | ||
| ) | ||
|
|
||
| if self.previous_lifetime is not None: | ||
| if self.fixed_size is None: | ||
| # TODO: Might be only a warning | ||
| raise ValueError('previous_lifetime can only be used if fixed_size is defined.') | ||
| if self.force_investment is False: | ||
| # TODO: Might be only a warning | ||
| raise ValueError('previous_lifetime can only be used if force_investment is True.') | ||
|
|
||
| if self.minimum_or_fixed_lifetime is not None and self.maximum_or_fixed_lifetime is not None: |
There was a problem hiding this comment.
Incorrect checks for forced investment/decommissioning.
self.force_investment is Falsecompares identity to a DataArray; always false.- “exactly one forced year” logic uses
> 1where== 1(or.any()) is intended.
Apply:
- if (self.force_investment.sum('year') > 1).any():
+ if (self.force_investment.sum('year') > 1).any():
raise ValueError('force_investment can only be True for a single year.')
- if (self.force_decommissioning.sum('year') > 1).any():
+ if (self.force_decommissioning.sum('year') > 1).any():
raise ValueError('force_decommissioning can only be True for a single year.')
@@
- if self.previous_lifetime is not None:
+ if self.previous_lifetime is not None:
if self.fixed_size is None:
# TODO: Might be only a warning
raise ValueError('previous_lifetime can only be used if fixed_size is defined.')
- if self.force_investment is False:
+ if not bool(self.force_investment.any()):
# TODO: Might be only a warning
raise ValueError('previous_lifetime can only be used if force_investment is True.')
@@
- specify_timing = (
- (self.fixed_lifetime is not None)
- + bool((self.force_investment.sum('year') > 1).any())
- + bool((self.force_decommissioning.sum('year') > 1).any())
- )
+ specify_timing = (
+ (self.fixed_lifetime is not None)
+ + bool(self.force_investment.any())
+ + bool(self.force_decommissioning.any())
+ )📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def _plausibility_checks(self, flow_system): | |
| """Validate parameter consistency.""" | |
| if flow_system.years is None: | |
| raise ValueError("InvestTimingParameters requires the flow_system to have a 'years' dimension.") | |
| if (self.force_investment.sum('year') > 1).any(): | |
| raise ValueError('force_investment can only be True for a single year.') | |
| if (self.force_decommissioning.sum('year') > 1).any(): | |
| raise ValueError('force_decommissioning can only be True for a single year.') | |
| if (self.force_investment.sum('year') == 1).any() and (self.force_decommissioning.sum('year') == 1).any(): | |
| year_of_forced_investment = ( | |
| self.force_investment.where(self.force_investment) * self.force_investment.year | |
| ).sum('year') | |
| year_of_forced_decommissioning = ( | |
| self.force_decommissioning.where(self.force_decommissioning) * self.force_decommissioning.year | |
| ).sum('year') | |
| if not (year_of_forced_investment < year_of_forced_decommissioning).all(): | |
| raise ValueError( | |
| f'force_investment needs to be before force_decommissioning. Got:\n' | |
| f'{self.force_investment}\nand\n{self.force_decommissioning}' | |
| ) | |
| if self.previous_lifetime is not None: | |
| if self.fixed_size is None: | |
| # TODO: Might be only a warning | |
| raise ValueError('previous_lifetime can only be used if fixed_size is defined.') | |
| if self.force_investment is False: | |
| # TODO: Might be only a warning | |
| raise ValueError('previous_lifetime can only be used if force_investment is True.') | |
| if self.minimum_or_fixed_lifetime is not None and self.maximum_or_fixed_lifetime is not None: | |
| def _plausibility_checks(self, flow_system): | |
| """Validate parameter consistency.""" | |
| if flow_system.years is None: | |
| raise ValueError("InvestTimingParameters requires the flow_system to have a 'years' dimension.") | |
| if (self.force_investment.sum('year') > 1).any(): | |
| raise ValueError('force_investment can only be True for a single year.') | |
| if (self.force_decommissioning.sum('year') > 1).any(): | |
| raise ValueError('force_decommissioning can only be True for a single year.') | |
| if (self.force_investment.sum('year') == 1).any() and (self.force_decommissioning.sum('year') == 1).any(): | |
| year_of_forced_investment = ( | |
| self.force_investment.where(self.force_investment) * self.force_investment.year | |
| ).sum('year') | |
| year_of_forced_decommissioning = ( | |
| self.force_decommissioning.where(self.force_decommissioning) * self.force_decommissioning.year | |
| ).sum('year') | |
| if not (year_of_forced_investment < year_of_forced_decommissioning).all(): | |
| raise ValueError( | |
| f'force_investment needs to be before force_decommissioning. Got:\n' | |
| f'{self.force_investment}\nand\n{self.force_decommissioning}' | |
| ) | |
| if self.previous_lifetime is not None: | |
| if self.fixed_size is None: | |
| # TODO: Might be only a warning | |
| raise ValueError('previous_lifetime can only be used if fixed_size is defined.') | |
| if not bool(self.force_investment.any()): | |
| # TODO: Might be only a warning | |
| raise ValueError('previous_lifetime can only be used if force_investment is True.') | |
| if self.minimum_or_fixed_lifetime is not None and self.maximum_or_fixed_lifetime is not None: | |
| specify_timing = ( | |
| (self.fixed_lifetime is not None) | |
| bool(self.force_investment.any()) | |
| bool(self.force_decommissioning.any()) | |
| ) |
🤖 Prompt for AI Agents
In flixopt/interface.py around lines 338-369, fix two checks: replace the
identity comparison "if self.force_investment is False" with a proper
DataArray-based test (e.g. check whether there is any True year: use
self.force_investment.any(...) or verify no forced year via
(self.force_investment.sum('year') == 0).any()), and ensure the "exactly one
forced year" logic uses equality checks where intended (use
(self.force_investment.sum('year') == 1).any() and similarly for
force_decommissioning) rather than identity or incorrect comparisons; update the
conditional expressions accordingly so all branches correctly test DataArray
contents across the 'year' dimension.
| weights = self.flow_system.fit_to_model_coords( | ||
| 'weights', | ||
| 1 if self.flow_system.years is None else self.flow_system.years_per_year, | ||
| dims=['year', 'scenario'], | ||
| ) | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Split weights: operation vs investment.
Using years_per_year for all effects causes invest double-counting. Keep current property for operation; add a scenario-only weights_invest and use it in effects objective.
Proposed additions:
@property
def weights(self) -> Union[int, xr.DataArray]:
"""Returns the scenario weights of the FlowSystem. If None, return weights that are normalized to 1 (one)"""
if self.flow_system.weights is None:
weights = self.flow_system.fit_to_model_coords(
'weights',
1 if self.flow_system.years is None else self.flow_system.years_per_year,
dims=['year', 'scenario'],
)
return weights / weights.sum()
return self.flow_system.weights
+
+ @property
+ def weights_invest(self) -> xr.DataArray:
+ """Scenario-only weights (no year scaling) for investment terms."""
+ if self.flow_system.weights is None:
+ w = self.flow_system.fit_to_model_coords('weights_invest', 1, dims=['year', 'scenario'])
+ # normalize over scenarios only
+ scen_sum = w.sum('year')
+ scen_sum = scen_sum / scen_sum.sum()
+ return xr.DataArray(
+ np.broadcast_to(scen_sum.values, w.shape),
+ dims=w.dims,
+ coords=w.coords,
+ name='weights_invest',
+ )
+ # user provided weights: drop year scaling by summing over years
+ return (self.flow_system.weights / self.flow_system.weights.sum()).sum('year')Then use weights_invest in the effects objective (see effects.py comment).
🤖 Prompt for AI Agents
In flixopt/structure.py around lines 181-186, the current weights calculation
uses years_per_year for all effects which double-counts investments; keep the
existing weights property for operational effects (using years_per_year) but add
a new scenario-only weights_invest property that does not multiply by years.
Create self.weights_invest = self.flow_system.fit_to_model_coords('weights', 1,
dims=['scenario']) (i.e. scalar 1 per scenario) and ensure the effects objective
(effects.py) is changed to use self.weights_invest for investment-related terms
while leaving operational terms using the existing self.weights.
| def calculate_annual_payment(total_cost: float, remaining_years: int, discount_rate: float) -> float: | ||
| """Calculate annualized payment for given remaining years. | ||
| Args: | ||
| total_cost: Total cost to be annualized. | ||
| remaining_years: Number of remaining years. | ||
| discount_rate: Discount rate for annualization. | ||
| Returns: | ||
| Annual payment amount. | ||
| """ | ||
| if remaining_years == 1: | ||
| return total_cost | ||
|
|
||
| return ( | ||
| total_cost | ||
| * (discount_rate * (1 + discount_rate) ** remaining_years) | ||
| / ((1 + discount_rate) ** remaining_years - 1) | ||
| ) | ||
|
|
There was a problem hiding this comment.
Handle zero discount rate in annuity helper
Avoid division by zero when discount_rate == 0.
Apply this diff:
def calculate_annual_payment(total_cost: float, remaining_years: int, discount_rate: float) -> float:
@@
- if remaining_years == 1:
+ if remaining_years == 1:
return total_cost
-
- return (
+ if discount_rate == 0:
+ return total_cost / remaining_years
+ return (
total_cost
* (discount_rate * (1 + discount_rate) ** remaining_years)
/ ((1 + discount_rate) ** remaining_years - 1)
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def calculate_annual_payment(total_cost: float, remaining_years: int, discount_rate: float) -> float: | |
| """Calculate annualized payment for given remaining years. | |
| Args: | |
| total_cost: Total cost to be annualized. | |
| remaining_years: Number of remaining years. | |
| discount_rate: Discount rate for annualization. | |
| Returns: | |
| Annual payment amount. | |
| """ | |
| if remaining_years == 1: | |
| return total_cost | |
| return ( | |
| total_cost | |
| * (discount_rate * (1 + discount_rate) ** remaining_years) | |
| / ((1 + discount_rate) ** remaining_years - 1) | |
| ) | |
| def calculate_annual_payment(total_cost: float, remaining_years: int, discount_rate: float) -> float: | |
| """Calculate annualized payment for given remaining years. | |
| Args: | |
| total_cost: Total cost to be annualized. | |
| remaining_years: Number of remaining years. | |
| discount_rate: Discount rate for annualization. | |
| Returns: | |
| Annual payment amount. | |
| """ | |
| if remaining_years == 1: | |
| return total_cost | |
| if discount_rate == 0: | |
| return total_cost / remaining_years | |
| return ( | |
| total_cost | |
| * (discount_rate * (1 + discount_rate) ** remaining_years) | |
| / ((1 + discount_rate) ** remaining_years - 1) | |
| ) |
🤖 Prompt for AI Agents
In tests/test_models.py around lines 24 to 43, the annuity helper divides by
((1+discount_rate)**remaining_years - 1) which will raise a ZeroDivisionError
when discount_rate == 0; add an early branch handling discount_rate == 0 (after
the existing remaining_years == 1 check) that returns total_cost /
remaining_years, otherwise keep the existing formula.
| def test_basic_initialization(self): | ||
| """Test basic parameter initialization.""" | ||
| params = fx.YearAwareInvestParameters( | ||
| minimum_size=10, | ||
| maximum_size=100, | ||
| ) | ||
|
|
||
| assert params.minimum_size == 10 | ||
| assert params.maximum_size == 100 | ||
| assert params.fixed_size is None | ||
| assert not params.allow_divestment | ||
| assert params.fixed_year_of_investment is None | ||
| assert params.fixed_year_of_decommissioning is None | ||
| assert params.fixed_duration is None | ||
|
|
There was a problem hiding this comment.
Class name mismatch: use InvestTimingParameters (or add alias)
The codebase exposes InvestTimingParameters; YearAwareInvestParameters isn’t exported. Either rely on the alias added in init.py/commons.py or switch tests to InvestTimingParameters.
Apply this diff if you prefer to update tests:
- params = fx.YearAwareInvestParameters(
+ params = fx.InvestTimingParameters(
minimum_size=10,
maximum_size=100,
)Repeat the same rename for all occurrences in this file.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def test_basic_initialization(self): | |
| """Test basic parameter initialization.""" | |
| params = fx.YearAwareInvestParameters( | |
| minimum_size=10, | |
| maximum_size=100, | |
| ) | |
| assert params.minimum_size == 10 | |
| assert params.maximum_size == 100 | |
| assert params.fixed_size is None | |
| assert not params.allow_divestment | |
| assert params.fixed_year_of_investment is None | |
| assert params.fixed_year_of_decommissioning is None | |
| assert params.fixed_duration is None | |
| def test_basic_initialization(self): | |
| """Test basic parameter initialization.""" | |
| params = fx.InvestTimingParameters( | |
| minimum_size=10, | |
| maximum_size=100, | |
| ) | |
| assert params.minimum_size == 10 | |
| assert params.maximum_size == 100 | |
| assert params.fixed_size is None | |
| assert not params.allow_divestment | |
| assert params.fixed_year_of_investment is None | |
| assert params.fixed_year_of_decommissioning is None | |
| assert params.fixed_duration is None |
🤖 Prompt for AI Agents
In tests/test_models.py around lines 118 to 132, the test uses
YearAwareInvestParameters which is not exported; replace all occurrences of
YearAwareInvestParameters in this file with the exported class name
InvestTimingParameters (or add an explicit alias import if you prefer), and
update any imports at the top of the file accordingly so the tests reference the
exported InvestTimingParameters class consistently.
FBumann
added
the
Not in Focus
|
Closed by #348 |
Description
Adding InvestmentTiming to flixOpt.
This is not yet finished and needs some work. Including proper testing.
One identified major issue is:
Type of Change
Related Issues
Closes #(issue number)
Testing
Checklist
Summary by CodeRabbit