diff --git a/CHANGELOG.md b/CHANGELOG.md index eef465ad9..e129e434c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -60,8 +60,17 @@ If upgrading from v2.x, see the [v3.0.0 release notes](https://github.com/flixOp ### 💥 Breaking Changes ### ♻️ Changed +**Improved repr methods:** +- **Results classes** (`ComponentResults`, `BusResults`, `FlowResults`, `EffectResults`) now show concise header with key metadata followed by xarray Dataset repr +- **Element classes** (`Component`, `Bus`, `Flow`, `Effect`, `Storage`) now show one-line summaries with essential information (connections, sizes, capacities, constraints) + +**Container-based access:** +- **FlowSystem** now provides dict-like access patterns for all elements +- Use `flow_system['element_label']`, `flow_system.keys()`, `flow_system.values()`, and `flow_system.items()` for unified element access +- Specialized containers (`components`, `buses`, `effects`, `flows`) offer type-specific access with helpful error messages ### 🗑️ Deprecated +- **`FlowSystem.all_elements`** property is deprecated in favor of dict-like interface (`flow_system['label']`, `.keys()`, `.values()`, `.items()`). Will be removed in v4.0.0. ### 🔥 Removed @@ -91,21 +100,21 @@ If upgrading from v2.x, see the [v3.0.0 release notes](https://github.com/flixOp **Color management:** - **`setup_colors()` method** for `CalculationResults` and `SegmentedCalculationResults` to configure consistent colors across all plots - - Group components by colorscales: `results.setup_colors({'CHP': 'reds', 'Storage': 'blues', 'Greys': ['Grid', 'Demand']})` - - Automatically propagates to all segments in segmented calculations - - Colors persist across all plot calls unless explicitly overridden + - Group components by colorscales: `results.setup_colors({'CHP': 'reds', 'Storage': 'blues', 'Greys': ['Grid', 'Demand']})` + - Automatically propagates to all segments in segmented calculations + - Colors persist across all plot calls unless explicitly overridden - **Flexible color inputs**: Supports colorscale names (e.g., 'turbo', 'plasma'), color lists, or label-to-color dictionaries - **Cross-backend compatibility**: Seamless color handling for both Plotly and Matplotlib **Plotting customization:** - **Plotting kwargs support**: Pass additional arguments to plotting backends via `px_kwargs`, `plot_kwargs`, and `backend_kwargs` parameters - **New `CONFIG.Plotting` configuration section**: - - `default_show`: Control default plot visibility - - `default_engine`: Choose 'plotly' or 'matplotlib' - - `default_dpi`: Set resolution for saved plots - - `default_facet_cols`: Configure default faceting columns - - `default_sequential_colorscale`: Default for heatmaps (now 'turbo') - - `default_qualitative_colorscale`: Default for categorical plots (now 'plotly') + - `default_show`: Control default plot visibility + - `default_engine`: Choose 'plotly' or 'matplotlib' + - `default_dpi`: Set resolution for saved plots + - `default_facet_cols`: Configure default faceting columns + - `default_sequential_colorscale`: Default for heatmaps (now 'turbo') + - `default_qualitative_colorscale`: Default for categorical plots (now 'plotly') **I/O improvements:** - Centralized JSON/YAML I/O with auto-format detection @@ -272,12 +281,12 @@ This replaces `specific_share_to_other_effects_*` parameters and inverts the dir **API and Behavior Changes:** - **Effect system redesigned** (no deprecation): - - **Terminology changes**: Effect domains renamed for clarity: `operation` → `temporal`, `invest`/`investment` → `periodic` - - **Sharing system**: The old `specific_share_to_other_effects_*` parameters were completely replaced with the new `share_from_temporal` and `share_from_periodic` syntax (see 🔥 Removed section) + - **Terminology changes**: Effect domains renamed for clarity: `operation` → `temporal`, `invest`/`investment` → `periodic` + - **Sharing system**: The old `specific_share_to_other_effects_*` parameters were completely replaced with the new `share_from_temporal` and `share_from_periodic` syntax (see 🔥 Removed section) - **FlowSystem independence**: FlowSystems cannot be shared across multiple Calculations anymore. A copy of the FlowSystem is created instead, making every Calculation independent. Each Subcalculation in `SegmentedCalculation` now has its own distinct `FlowSystem` object - **Bus and Effect object assignment**: Direct assignment of Bus/Effect objects is no longer supported. Use labels (strings) instead: - - `Flow.bus` must receive a string label, not a Bus object - - Effect shares must use effect labels (strings) in dictionaries, not Effect objects + - `Flow.bus` must receive a string label, not a Bus object + - Effect shares must use effect labels (strings) in dictionaries, not Effect objects - **Logging defaults** (from v2.2.0): Console and file logging are now disabled by default. Enable explicitly with `CONFIG.Logging.console = True` and `CONFIG.apply()` **Class and Method Renaming:** @@ -291,14 +300,14 @@ This replaces `specific_share_to_other_effects_*` parameters and inverts the dir - Investment binary variable: `is_invested` → `invested` in `InvestmentModel` - Switch tracking variables in `OnOffModel`: - - `switch_on` → `switch|on` - - `switch_off` → `switch|off` - - `switch_on_nr` → `switch|count` + - `switch_on` → `switch|on` + - `switch_off` → `switch|off` + - `switch_on_nr` → `switch|count` - Effect submodel variables (following terminology changes): - - `Effect(invest)|total` → `Effect(periodic)` - - `Effect(operation)|total` → `Effect(temporal)` - - `Effect(operation)|total_per_timestep` → `Effect(temporal)|per_timestep` - - `Effect|total` → `Effect` + - `Effect(invest)|total` → `Effect(periodic)` + - `Effect(operation)|total` → `Effect(temporal)` + - `Effect(operation)|total_per_timestep` → `Effect(temporal)|per_timestep` + - `Effect|total` → `Effect` **Data Structure Changes:** @@ -519,7 +528,7 @@ This replaces `specific_share_to_other_effects_*` parameters and inverts the dir ### ✨ Added - **Network Visualization**: Added `FlowSystem.start_network_app()` and `FlowSystem.stop_network_app()` to easily visualize the network structure of a flow system in an interactive Dash web app - - *Note: This is still experimental and might change in the future* + - *Note: This is still experimental and might change in the future* ### ♻️ Changed - **Multi-Flow Support**: `Sink`, `Source`, and `SourceAndSink` now accept multiple `flows` as `inputs` and `outputs` instead of just one. This enables modeling more use cases with these classes @@ -561,8 +570,8 @@ This replaces `specific_share_to_other_effects_*` parameters and inverts the dir ### 🐛 Fixed - Storage losses per hour were not calculated correctly, as mentioned by @brokenwings01. This might have led to issues when modeling large losses and long timesteps. - - Old implementation: $c(\text{t}_{i}) \cdot (1-\dot{\text{c}}_\text{rel,loss}(\text{t}_i)) \cdot \Delta \text{t}_{i}$ - - Correct implementation: $c(\text{t}_{i}) \cdot (1-\dot{\text{c}}_\text{rel,loss}(\text{t}_i)) ^{\Delta \text{t}_{i}}$ + - Old implementation: $c(\text{t}_{i}) \cdot (1-\dot{\text{c}}_\text{rel,loss}(\text{t}_i)) \cdot \Delta \text{t}_{i}$ + - Correct implementation: $c(\text{t}_{i}) \cdot (1-\dot{\text{c}}_\text{rel,loss}(\text{t}_i)) ^{\Delta \text{t}_{i}}$ ### 🚧 Known Issues - Just to mention: Plotly >= 6 may raise errors if "nbformat" is not installed. We pinned plotly to <6, but this may be fixed in the future. @@ -587,10 +596,10 @@ This replaces `specific_share_to_other_effects_*` parameters and inverts the dir ### 💥 Breaking Changes - Restructured the modeling of the On/Off state of Flows or Components - - Variable renaming: `...|consecutive_on_hours` → `...|ConsecutiveOn|hours` - - Variable renaming: `...|consecutive_off_hours` → `...|ConsecutiveOff|hours` - - Constraint renaming: `...|consecutive_on_hours_con1` → `...|ConsecutiveOn|con1` - - Similar pattern for all consecutive on/off constraints + - Variable renaming: `...|consecutive_on_hours` → `...|ConsecutiveOn|hours` + - Variable renaming: `...|consecutive_off_hours` → `...|ConsecutiveOff|hours` + - Constraint renaming: `...|consecutive_on_hours_con1` → `...|ConsecutiveOn|con1` + - Similar pattern for all consecutive on/off constraints ### 🐛 Fixed - Fixed the lower bound of `flow_rate` when using optional investments without OnOffParameters @@ -636,10 +645,10 @@ This replaces `specific_share_to_other_effects_*` parameters and inverts the dir **Variable Structure:** - Restructured the modeling of the On/Off state of Flows or Components - - Variable renaming: `...|consecutive_on_hours` → `...|ConsecutiveOn|hours` - - Variable renaming: `...|consecutive_off_hours` → `...|ConsecutiveOff|hours` - - Constraint renaming: `...|consecutive_on_hours_con1` → `...|ConsecutiveOn|con1` - - Similar pattern for all consecutive on/off constraints + - Variable renaming: `...|consecutive_on_hours` → `...|ConsecutiveOn|hours` + - Variable renaming: `...|consecutive_off_hours` → `...|ConsecutiveOff|hours` + - Constraint renaming: `...|consecutive_on_hours_con1` → `...|ConsecutiveOn|con1` + - Similar pattern for all consecutive on/off constraints ### 🔥 Removed - **Pyomo dependency** (replaced by linopy) diff --git a/docs/user-guide/index.md b/docs/user-guide/index.md index df97bf768..b20d15263 100644 --- a/docs/user-guide/index.md +++ b/docs/user-guide/index.md @@ -15,15 +15,22 @@ Every FlixOpt model starts with creating a FlowSystem. It: - Contains and connects [components](#components), [buses](#buses), and [flows](#flows) - Manages the [effects](#effects) (objectives and constraints) +FlowSystem provides two ways to access elements: + +- **Dict-like interface**: Access any element by label: `flow_system['Boiler']`, `'Boiler' in flow_system`, `flow_system.keys()` +- **Direct containers**: Access type-specific containers: `flow_system.components`, `flow_system.buses`, `flow_system.effects`, `flow_system.flows` + +Element labels must be unique across all types. See the [`FlowSystem` API reference][flixopt.flow_system.FlowSystem] for detailed examples and usage patterns. + ### Flows [`Flow`][flixopt.elements.Flow] objects represent the movement of energy or material between a [Bus](#buses) and a [Component](#components) in a predefined direction. -- Have a `size` which, generally speaking, defines how fast energy or material can be moved. Usually measured in MW, kW, m³/h, etc. -- Have a `flow_rate`, which is defines how fast energy or material is transported. Usually measured in MW, kW, m³/h, etc. +- Have a `size` which, generally speaking, defines how much energy or material can be moved. Usually measured in MW, kW, m³/h, etc. +- Have a `flow_rate`, which defines how fast energy or material is transported. Usually measured in MW, kW, m³/h, etc. - Have constraints to limit the flow-rate (min/max, total flow hours, on/off etc.) - Can have fixed profiles (for demands or renewable generation) -- Can have [Effects](#effects) associated by their use (operation, investment, on/off, ...) +- Can have [Effects](#effects) associated by their use (costs, emissions, labour, ...) #### Flow Hours While the **Flow Rate** defines the rate in which energy or material is transported, the **Flow Hours** define the amount of energy or material that is transported. @@ -50,21 +57,21 @@ Examples: [`Component`][flixopt.elements.Component] objects usually represent physical entities in your system that interact with [`Flows`][flixopt.elements.Flow]. The generic component types work across all domains: - [`LinearConverters`][flixopt.components.LinearConverter] - Converts input flows to output flows with (piecewise) linear relationships - - *Energy: boilers, heat pumps, turbines* - - *Manufacturing: assembly lines, processing equipment* - - *Chemistry: reactors, separators* + - *Energy: boilers, heat pumps, turbines* + - *Manufacturing: assembly lines, processing equipment* + - *Chemistry: reactors, separators* - [`Storages`][flixopt.components.Storage] - Stores energy or material over time - - *Energy: batteries, thermal storage, gas storage* - - *Logistics: warehouses, buffer inventory* - - *Water: reservoirs, tanks* + - *Energy: batteries, thermal storage, gas storage* + - *Logistics: warehouses, buffer inventory* + - *Water: reservoirs, tanks* - [`Sources`][flixopt.components.Source] / [`Sinks`][flixopt.components.Sink] / [`SourceAndSinks`][flixopt.components.SourceAndSink] - Produce or consume flows - - *Energy: demands, renewable generation* - - *Manufacturing: raw material supply, product demand* - - *Supply chain: suppliers, customers* + - *Energy: demands, renewable generation* + - *Manufacturing: raw material supply, product demand* + - *Supply chain: suppliers, customers* - [`Transmissions`][flixopt.components.Transmission] - Moves flows between locations with possible losses - - *Energy: pipelines, power lines* - - *Logistics: transport routes* - - *Water: distribution networks* + - *Energy: pipelines, power lines* + - *Logistics: transport routes* + - *Water: distribution networks* **Pre-built specialized components** for energy systems include [`Boilers`][flixopt.linear_converters.Boiler], [`HeatPumps`][flixopt.linear_converters.HeatPump], [`CHPs`][flixopt.linear_converters.CHP], etc. These can serve as blueprints for custom domain-specific components. @@ -105,7 +112,7 @@ FlixOpt offers different calculation modes: The results of a calculation are stored in a [`CalculationResults`][flixopt.results.CalculationResults] object. This object contains the solutions of the optimization as well as all information about the [`Calculation`][flixopt.calculation.Calculation] and the [`FlowSystem`][flixopt.flow_system.FlowSystem] it was created from. -The solutions is stored as an `xarray.Dataset`, but can be accessed through their assotiated Component, Bus or Effect. +The solution is stored as an `xarray.Dataset`, but can be accessed through their assotiated Component, Bus or Effect. This [`CalculationResults`][flixopt.results.CalculationResults] object can be saved to file and reloaded from file, allowing you to analyze the results anytime after the solve. diff --git a/flixopt/calculation.py b/flixopt/calculation.py index f744c5247..5e919dbf5 100644 --- a/flixopt/calculation.py +++ b/flixopt/calculation.py @@ -112,7 +112,7 @@ def main_results(self) -> dict[str, Scalar | dict]: 'periodic': effect.submodel.periodic.total.solution.values, 'total': effect.submodel.total.solution.values, } - for effect in self.flow_system.effects + for effect in sorted(self.flow_system.effects.values(), key=lambda e: e.label_full.upper()) }, 'Invest-Decisions': { 'Invested': { diff --git a/flixopt/components.py b/flixopt/components.py index 09156e1dc..8f89378ae 100644 --- a/flixopt/components.py +++ b/flixopt/components.py @@ -11,6 +11,7 @@ import numpy as np import xarray as xr +from . import io as fx_io from .core import PeriodicDataUser, PlausibilityError, TemporalData, TemporalDataUser from .elements import Component, ComponentModel, Flow from .features import InvestmentModel, PiecewiseModel @@ -528,6 +529,15 @@ def _plausibility_checks(self) -> None: f'{self.discharging.size.minimum_size=}, {self.discharging.size.maximum_size=}.' ) + def __repr__(self) -> str: + """Return string representation.""" + # Use build_repr_from_init directly to exclude charging and discharging + return fx_io.build_repr_from_init( + self, + excluded_params={'self', 'label', 'charging', 'discharging', 'kwargs'}, + skip_default_size=True, + ) + fx_io.format_flow_details(self) + @register_class_for_io class Transmission(Component): diff --git a/flixopt/effects.py b/flixopt/effects.py index 6225734fe..757549223 100644 --- a/flixopt/effects.py +++ b/flixopt/effects.py @@ -16,9 +16,10 @@ import numpy as np import xarray as xr +from . import io as fx_io from .core import PeriodicDataUser, Scalar, TemporalData, TemporalDataUser from .features import ShareAllocationModel -from .structure import Element, ElementModel, FlowSystemModel, Submodel, register_class_for_io +from .structure import Element, ElementContainer, ElementModel, FlowSystemModel, Submodel, register_class_for_io if TYPE_CHECKING: from collections.abc import Iterator @@ -448,13 +449,13 @@ def _do_modeling(self): EffectExpr = dict[str, linopy.LinearExpression] # Used to create Shares -class EffectCollection: +class EffectCollection(ElementContainer[Effect]): """ Handling all Effects """ def __init__(self, *effects: Effect): - self._effects = {} + super().__init__(element_type_name='effects') self._standard_effect: Effect | None = None self._objective_effect: Effect | None = None @@ -474,7 +475,7 @@ def add_effects(self, *effects: Effect) -> None: self.standard_effect = effect if effect.is_objective: self.objective_effect = effect - self._effects[effect.label] = effect + self.add(effect) # Use the inherited add() method from ElementContainer logger.info(f'Registered new Effect: {effect.label}') def create_effect_values_dict( @@ -520,10 +521,13 @@ def _plausibility_checks(self) -> None: # Check circular loops in effects: temporal, periodic = self.calculate_effect_share_factors() - # Validate all referenced sources exist - unknown = {src for src, _ in list(temporal.keys()) + list(periodic.keys()) if src not in self.effects} + # Validate all referenced effects (both sources and targets) exist + edges = list(temporal.keys()) + list(periodic.keys()) + unknown_sources = {src for src, _ in edges if src not in self} + unknown_targets = {tgt for _, tgt in edges if tgt not in self} + unknown = unknown_sources | unknown_targets if unknown: - raise KeyError(f'Unknown effects used in in effect share mappings: {sorted(unknown)}') + raise KeyError(f'Unknown effects used in effect share mappings: {sorted(unknown)}') temporal_cycles = detect_cycles(tuples_to_adjacency_list([key for key in temporal])) periodic_cycles = detect_cycles(tuples_to_adjacency_list([key for key in periodic])) @@ -552,31 +556,23 @@ def __getitem__(self, effect: str | Effect | None) -> Effect: else: raise KeyError(f'Effect {effect} not found!') try: - return self.effects[effect] + return super().__getitem__(effect) # Leverage ContainerMixin suggestions except KeyError as e: - raise KeyError(f'Effect "{effect}" not found! Add it to the FlowSystem first!') from e + # Extract the original message and append context for cleaner output + original_msg = str(e).strip('\'"') + raise KeyError(f'{original_msg} Add the effect to the FlowSystem first.') from None - def __iter__(self) -> Iterator[Effect]: - return iter(self._effects.values()) - - def __len__(self) -> int: - return len(self._effects) + def __iter__(self) -> Iterator[str]: + return iter(self.keys()) # Iterate over keys like a normal dict def __contains__(self, item: str | Effect) -> bool: """Check if the effect exists. Checks for label or object""" if isinstance(item, str): - return item in self.effects # Check if the label exists + return super().__contains__(item) # Check if the label exists elif isinstance(item, Effect): - if item.label_full in self.effects: - return True - if item in self.effects.values(): # Check if the object exists - return True + return item.label_full in self and self[item.label_full] is item return False - @property - def effects(self) -> dict[str, Effect]: - return self._effects - @property def standard_effect(self) -> Effect: if self._standard_effect is None: @@ -611,7 +607,7 @@ def calculate_effect_share_factors( dict[tuple[str, str], xr.DataArray], ]: shares_periodic = {} - for name, effect in self.effects.items(): + for name, effect in self.items(): if effect.share_from_periodic: for source, data in effect.share_from_periodic.items(): if source not in shares_periodic: @@ -620,7 +616,7 @@ def calculate_effect_share_factors( shares_periodic = calculate_all_conversion_paths(shares_periodic) shares_temporal = {} - for name, effect in self.effects.items(): + for name, effect in self.items(): if effect.share_from_temporal: for source, data in effect.share_from_temporal.items(): if source not in shares_temporal: @@ -670,7 +666,7 @@ def add_share_to_penalty(self, name: str, expression: linopy.LinearExpression) - def _do_modeling(self): super()._do_modeling() - for effect in self.effects: + for effect in self.effects.values(): effect.create_model(self._model) self.penalty = self.add_submodels( ShareAllocationModel(self._model, dims=(), label_of_element='Penalty'), @@ -684,7 +680,7 @@ def _do_modeling(self): ) def _add_share_between_effects(self): - for target_effect in self.effects: + for target_effect in self.effects.values(): # 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( diff --git a/flixopt/elements.py b/flixopt/elements.py index a0fd306c0..2a9a2cf4f 100644 --- a/flixopt/elements.py +++ b/flixopt/elements.py @@ -11,6 +11,7 @@ import numpy as np import xarray as xr +from . import io as fx_io from .config import CONFIG from .core import PlausibilityError, Scalar, TemporalData, TemporalDataUser from .features import InvestmentModel, OnOffModel @@ -86,10 +87,12 @@ def __init__( super().__init__(label, meta_data=meta_data) self.inputs: list[Flow] = inputs or [] self.outputs: list[Flow] = outputs or [] - self._check_unique_flow_labels() self.on_off_parameters = on_off_parameters self.prevent_simultaneous_flows: list[Flow] = prevent_simultaneous_flows or [] + self._check_unique_flow_labels() + self._connect_flows() + self.flows: dict[str, Flow] = {flow.label: flow for flow in self.inputs + self.outputs} def create_model(self, model: FlowSystemModel) -> ComponentModel: @@ -115,6 +118,48 @@ def _check_unique_flow_labels(self): def _plausibility_checks(self) -> None: self._check_unique_flow_labels() + def _connect_flows(self): + # Inputs + for flow in self.inputs: + if flow.component not in ('UnknownComponent', self.label_full): + raise ValueError( + f'Flow "{flow.label}" already assigned to component "{flow.component}". ' + f'Cannot attach to "{self.label_full}".' + ) + flow.component = self.label_full + flow.is_input_in_component = True + # Outputs + for flow in self.outputs: + if flow.component not in ('UnknownComponent', self.label_full): + raise ValueError( + f'Flow "{flow.label}" already assigned to component "{flow.component}". ' + f'Cannot attach to "{self.label_full}".' + ) + flow.component = self.label_full + flow.is_input_in_component = False + + # Validate prevent_simultaneous_flows: only allow local flows + if self.prevent_simultaneous_flows: + # Deduplicate while preserving order + seen = set() + self.prevent_simultaneous_flows = [ + f for f in self.prevent_simultaneous_flows if id(f) not in seen and not seen.add(id(f)) + ] + local = set(self.inputs + self.outputs) + foreign = [f for f in self.prevent_simultaneous_flows if f not in local] + if foreign: + names = ', '.join(f.label_full for f in foreign) + raise ValueError( + f'prevent_simultaneous_flows for "{self.label_full}" must reference its own flows. ' + f'Foreign flows detected: {names}' + ) + + def __repr__(self) -> str: + """Return string representation with flow information.""" + return fx_io.build_repr_from_init( + self, excluded_params={'self', 'label', 'inputs', 'outputs', 'kwargs'}, skip_default_size=True + ) + fx_io.format_flow_details(self) + @register_class_for_io class Bus(Element): @@ -216,6 +261,10 @@ def _plausibility_checks(self) -> None: def with_excess(self) -> bool: return False if self.excess_penalty_per_flow_hour is None else True + def __repr__(self) -> str: + """Return string representation.""" + return super().__repr__() + fx_io.format_flow_details(self) + @register_class_for_io class Connection: @@ -493,6 +542,10 @@ 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 + def _format_invest_params(self, params: InvestParameters) -> str: + """Format InvestParameters for display.""" + return f'size: {params.format_for_repr()}' + class FlowModel(ElementModel): element: Flow # Type hint diff --git a/flixopt/flow_system.py b/flixopt/flow_system.py index fd0f6a98d..26364c4b4 100644 --- a/flixopt/flow_system.py +++ b/flixopt/flow_system.py @@ -6,12 +6,14 @@ import logging import warnings +from itertools import chain from typing import TYPE_CHECKING, Any, Literal, Optional import numpy as np import pandas as pd import xarray as xr +from . import io as fx_io from .config import CONFIG from .core import ( ConversionError, @@ -32,7 +34,7 @@ TemporalEffectsUser, ) from .elements import Bus, Component, Flow -from .structure import Element, FlowSystemModel, Interface +from .structure import CompositeContainerMixin, Element, ElementContainer, FlowSystemModel, Interface if TYPE_CHECKING: import pathlib @@ -43,11 +45,13 @@ logger = logging.getLogger('flixopt') -class FlowSystem(Interface): +class FlowSystem(Interface, CompositeContainerMixin[Element]): """ - A FlowSystem organizes the high level Elements (Components, Buses & Effects). + A FlowSystem organizes the high level Elements (Components, Buses, Effects & Flows). - This is the main container class that users work with to build and manage their System. + This is the main container class that users work with to build and manage their energy or material flow system. + FlowSystem provides both direct container access (via .components, .buses, .effects, .flows) and a unified + dict-like interface for accessing any element by label across all container types. Args: timesteps: The timesteps of the model. @@ -69,10 +73,74 @@ class FlowSystem(Interface): - False: All flow rates are optimized separately per scenario - list[str]: Only specified flows (by label_full) are equalized across scenarios + Examples: + Creating a FlowSystem and accessing elements: + + >>> import flixopt as fx + >>> import pandas as pd + >>> timesteps = pd.date_range('2023-01-01', periods=24, freq='h') + >>> flow_system = fx.FlowSystem(timesteps) + >>> + >>> # Add elements to the system + >>> boiler = fx.Component('Boiler', inputs=[heat_flow], on_off_parameters=...) + >>> heat_bus = fx.Bus('Heat', excess_penalty_per_flow_hour=1e4) + >>> costs = fx.Effect('costs', is_objective=True, is_standard=True) + >>> flow_system.add_elements(boiler, heat_bus, costs) + + Unified dict-like access (recommended for most cases): + + >>> # Access any element by label, regardless of type + >>> boiler = flow_system['Boiler'] # Returns Component + >>> heat_bus = flow_system['Heat'] # Returns Bus + >>> costs = flow_system['costs'] # Returns Effect + >>> + >>> # Check if element exists + >>> if 'Boiler' in flow_system: + ... print('Boiler found in system') + >>> + >>> # Iterate over all elements + >>> for label in flow_system.keys(): + ... element = flow_system[label] + ... print(f'{label}: {type(element).__name__}') + >>> + >>> # Get all element labels and objects + >>> all_labels = list(flow_system.keys()) + >>> all_elements = list(flow_system.values()) + >>> for label, element in flow_system.items(): + ... print(f'{label}: {element}') + + Direct container access for type-specific operations: + + >>> # Access specific container when you need type filtering + >>> for component in flow_system.components.values(): + ... print(f'{component.label}: {len(component.inputs)} inputs') + >>> + >>> # Access buses directly + >>> for bus in flow_system.buses.values(): + ... print(f'{bus.label}') + >>> + >>> # Flows are automatically collected from all components + >>> for flow in flow_system.flows.values(): + ... print(f'{flow.label_full}: {flow.size}') + >>> + >>> # Access effects + >>> for effect in flow_system.effects.values(): + ... print(f'{effect.label}') + Notes: + - The dict-like interface (`flow_system['element']`) searches across all containers + (components, buses, effects, flows) to find the element with the matching label. + - Element labels must be unique across all container types. Attempting to add + elements with duplicate labels will raise an error, ensuring each label maps to exactly one element. + - The `.all_elements` property is deprecated. Use the dict-like interface instead: + `flow_system['element']`, `'element' in flow_system`, `flow_system.keys()`, + `flow_system.values()`, or `flow_system.items()`. + - Direct container access (`.components`, `.buses`, `.effects`, `.flows`) is useful + when you need type-specific filtering or operations. + - The `.flows` container is automatically populated from all component inputs and outputs. - Creates an empty registry for components and buses, an empty EffectCollection, and a placeholder for a SystemModel. - The instance starts disconnected (self._connected_and_transformed == False) and will be - connected_and_transformed automatically when trying to solve a calculation. + connected_and_transformed automatically when trying to solve a calculation. """ def __init__( @@ -104,8 +172,8 @@ def __init__( self.hours_per_timestep = self.fit_to_model_coords('hours_per_timestep', hours_per_timestep) # Element collections - self.components: dict[str, Component] = {} - self.buses: dict[str, Bus] = {} + self.components: ElementContainer[Component] = ElementContainer(element_type_name='components') + self.buses: ElementContainer[Bus] = ElementContainer(element_type_name='buses') self.effects: EffectCollection = EffectCollection() self.model: FlowSystemModel | None = None @@ -113,6 +181,7 @@ def __init__( self._used_in_calculation = False self._network_app = None + self._flows_cache: ElementContainer[Flow] | None = None # Use properties to validate and store scenario dimension settings self.scenario_independent_sizes = scenario_independent_sizes @@ -232,7 +301,7 @@ def _create_reference_structure(self) -> tuple[dict, dict[str, xr.DataArray]]: # Extract from effects effects_structure = {} - for effect in self.effects: + for effect in self.effects.values(): effect_structure, effect_arrays = effect._create_reference_structure() all_extracted_arrays.update(effect_arrays) effects_structure[effect.label] = effect_structure @@ -433,7 +502,7 @@ def connect_and_transform(self): self.weights = self.fit_to_model_coords('weights', self.weights, dims=['period', 'scenario']) self._connect_network() - for element in list(self.components.values()) + list(self.effects.effects.values()) + list(self.buses.values()): + for element in chain(self.components.values(), self.effects.values(), self.buses.values()): element.transform_data(self) self._connected_and_transformed = True @@ -581,7 +650,7 @@ def network_infos(self) -> tuple[dict[str, dict[str, str]], dict[str, dict[str, 'class': 'Bus' if isinstance(node, Bus) else 'Component', 'infos': node.__str__(), } - for node in list(self.components.values()) + list(self.buses.values()) + for node in chain(self.components.values(), self.buses.values()) } edges = { @@ -603,10 +672,8 @@ def _check_if_element_is_unique(self, element: Element) -> None: Args: element: new element to check """ - if element in self.all_elements.values(): - raise ValueError(f'Element {element.label_full} already added to FlowSystem!') # check if name is already used: - if element.label_full in self.all_elements: + if element.label_full in self: raise ValueError(f'Label of Element {element.label_full} already used in another element!') def _add_effects(self, *args: Effect) -> None: @@ -616,13 +683,15 @@ def _add_components(self, *components: Component) -> None: for new_component in list(components): logger.info(f'Registered new Component: {new_component.label_full}') self._check_if_element_is_unique(new_component) # check if already exists: - self.components[new_component.label_full] = new_component # Add to existing components + self.components.add(new_component) # Add to existing components + self._flows_cache = None # Invalidate flows cache def _add_buses(self, *buses: Bus): for new_bus in list(buses): logger.info(f'Registered new Bus: {new_bus.label_full}') self._check_if_element_is_unique(new_bus) # check if already exists: - self.buses[new_bus.label_full] = new_bus # Add to existing components + self.buses.add(new_bus) # Add to existing buses + self._flows_cache = None # Invalidate flows cache def _connect_network(self): """Connects the network of components and buses. Can be rerun without changes if no elements were added""" @@ -632,7 +701,7 @@ def _connect_network(self): flow.is_input_in_component = True if flow in component.inputs else False # Add Bus if not already added (deprecated) - if flow._bus_object is not None and flow._bus_object not in self.buses.values(): + if flow._bus_object is not None and flow._bus_object.label_full not in self.buses: warnings.warn( f'The Bus {flow._bus_object.label_full} was added to the FlowSystem from {flow.label_full}.' f'This is deprecated and will be removed in the future. ' @@ -659,62 +728,40 @@ def _connect_network(self): ) def __repr__(self) -> str: - """Compact representation for debugging.""" - status = '✓' if self.connected_and_transformed else '⚠' - - # Build dimension info - dims = f'{len(self.timesteps)} timesteps [{self.timesteps[0].strftime("%Y-%m-%d")} to {self.timesteps[-1].strftime("%Y-%m-%d")}]' - if self.periods is not None: - dims += f', {len(self.periods)} periods' - if self.scenarios is not None: - dims += f', {len(self.scenarios)} scenarios' - - return f'FlowSystem({dims}, {len(self.components)} Components, {len(self.buses)} Buses, {len(self.effects)} Effects, {status})' - - def __str__(self) -> str: - """Structured summary for users.""" - - def format_elements(element_names: list, label: str, alignment: int = 12): - name_list = ', '.join(element_names[:3]) - if len(element_names) > 3: - name_list += f' ... (+{len(element_names) - 3} more)' + """Return a detailed string representation showing all containers.""" + r = fx_io.format_title_with_underline('FlowSystem', '=') - suffix = f' ({name_list})' if element_names else '' - padding = alignment - len(label) - 1 # -1 for the colon - return f'{label}:{"":<{padding}} {len(element_names)}{suffix}' - - time_period = f'Time period: {self.timesteps[0].date()} to {self.timesteps[-1].date()}' + # Timestep info + time_period = f'{self.timesteps[0].date()} to {self.timesteps[-1].date()}' freq_str = str(self.timesteps.freq).replace('<', '').replace('>', '') if self.timesteps.freq else 'irregular' - - lines = [ - f'Timesteps: {len(self.timesteps)} ({freq_str}) [{time_period}]', - ] + r += f'Timesteps: {len(self.timesteps)} ({freq_str}) [{time_period}]\n' # Add periods if present if self.periods is not None: period_names = ', '.join(str(p) for p in self.periods[:3]) if len(self.periods) > 3: period_names += f' ... (+{len(self.periods) - 3} more)' - lines.append(f'Periods: {len(self.periods)} ({period_names})') + r += f'Periods: {len(self.periods)} ({period_names})\n' + else: + r += 'Periods: None\n' # Add scenarios if present if self.scenarios is not None: scenario_names = ', '.join(str(s) for s in self.scenarios[:3]) if len(self.scenarios) > 3: scenario_names += f' ... (+{len(self.scenarios) - 3} more)' - lines.append(f'Scenarios: {len(self.scenarios)} ({scenario_names})') - - lines.extend( - [ - format_elements(list(self.components.keys()), 'Components'), - format_elements(list(self.buses.keys()), 'Buses'), - format_elements(list(self.effects.effects.keys()), 'Effects'), - f'Status: {"Connected & Transformed" if self.connected_and_transformed else "Not connected"}', - ] - ) - lines = ['FlowSystem:', f'{"─" * max(len(line) for line in lines)}'] + lines + r += f'Scenarios: {len(self.scenarios)} ({scenario_names})\n' + else: + r += 'Scenarios: None\n' + + # Add status + status = '✓' if self.connected_and_transformed else '⚠' + r += f'Status: {status}\n' + + # Add grouped container view + r += '\n' + self._format_grouped_containers() - return '\n'.join(lines) + return r def __eq__(self, other: FlowSystem): """Check if two FlowSystems are equal by comparing their dataset representations.""" @@ -734,38 +781,46 @@ def __eq__(self, other: FlowSystem): return True - def __getitem__(self, item) -> Element: - """Get element by exact label with helpful error messages.""" - if item in self.all_elements: - return self.all_elements[item] - - # Provide helpful error with suggestions - from difflib import get_close_matches - - suggestions = get_close_matches(item, self.all_elements.keys(), n=3, cutoff=0.6) - - if suggestions: - suggestion_str = ', '.join(f"'{s}'" for s in suggestions) - raise KeyError(f"Element '{item}' not found. Did you mean: {suggestion_str}?") - else: - raise KeyError(f"Element '{item}' not found in FlowSystem") - - def __contains__(self, item: str) -> bool: - """Check if element exists in the FlowSystem.""" - return item in self.all_elements - - def __iter__(self): - """Iterate over element labels.""" - return iter(self.all_elements.keys()) + def _get_container_groups(self) -> dict[str, ElementContainer]: + """Return ordered container groups for CompositeContainerMixin.""" + return { + 'Components': self.components, + 'Buses': self.buses, + 'Effects': self.effects, + 'Flows': self.flows, + } @property - def flows(self) -> dict[str, Flow]: - set_of_flows = {flow for comp in self.components.values() for flow in comp.inputs + comp.outputs} - return {flow.label_full: flow for flow in set_of_flows} + def flows(self) -> ElementContainer[Flow]: + if self._flows_cache is None: + flows = [f for c in self.components.values() for f in c.inputs + c.outputs] + # Deduplicate by id and sort for reproducibility + flows = sorted({id(f): f for f in flows}.values(), key=lambda f: f.label_full.lower()) + self._flows_cache = ElementContainer(flows, element_type_name='flows') + return self._flows_cache @property def all_elements(self) -> dict[str, Element]: - return {**self.components, **self.effects.effects, **self.flows, **self.buses} + """ + Get all elements as a dictionary. + + .. deprecated:: 3.2.0 + Use dict-like interface instead: `flow_system['element']`, `'element' in flow_system`, + `flow_system.keys()`, `flow_system.values()`, or `flow_system.items()`. + This property will be removed in v4.0.0. + + Returns: + Dictionary mapping element labels to element objects. + """ + warnings.warn( + "The 'all_elements' property is deprecated. Use dict-like interface instead: " + "flow_system['element'], 'element' in flow_system, flow_system.keys(), " + 'flow_system.values(), or flow_system.items(). ' + 'This property will be removed in v4.0.0.', + DeprecationWarning, + stacklevel=2, + ) + return {**self.components, **self.effects, **self.flows, **self.buses} @property def coords(self) -> dict[FlowSystemDimensions, pd.Index]: diff --git a/flixopt/interface.py b/flixopt/interface.py index 8264e2392..eae0a8511 100644 --- a/flixopt/interface.py +++ b/flixopt/interface.py @@ -1051,6 +1051,27 @@ def minimum_or_fixed_size(self) -> PeriodicData: def maximum_or_fixed_size(self) -> PeriodicData: return self.fixed_size if self.fixed_size is not None else self.maximum_size + def format_for_repr(self) -> str: + """Format InvestParameters for display in repr methods. + + Returns: + Formatted string showing size information + """ + from .io import numeric_to_str_for_repr + + if self.fixed_size is not None: + val = numeric_to_str_for_repr(self.fixed_size) + status = 'mandatory' if self.mandatory else 'optional' + return f'{val} ({status})' + + # Show range if available + parts = [] + if self.minimum_size is not None: + parts.append(f'min: {numeric_to_str_for_repr(self.minimum_size)}') + if self.maximum_size is not None: + parts.append(f'max: {numeric_to_str_for_repr(self.maximum_size)}') + return ', '.join(parts) if parts else 'invest' + @staticmethod def compute_linked_periods(first_period: int, last_period: int, periods: pd.Index | list[int]) -> xr.DataArray: return xr.DataArray( diff --git a/flixopt/io.py b/flixopt/io.py index 059670ddd..8df03401c 100644 --- a/flixopt/io.py +++ b/flixopt/io.py @@ -1,5 +1,6 @@ from __future__ import annotations +import inspect import json import logging import pathlib @@ -8,6 +9,7 @@ from typing import TYPE_CHECKING, Any import numpy as np +import pandas as pd import xarray as xr import yaml @@ -547,3 +549,348 @@ def update(self, new_name: str | None = None, new_folder: pathlib.Path | None = raise FileNotFoundError(f'Folder {new_folder} does not exist or is not a directory.') self.folder = new_folder self._update_paths() + + +def numeric_to_str_for_repr( + value: int | float | np.integer | np.floating | np.ndarray | pd.Series | pd.DataFrame | xr.DataArray, + precision: int = 1, + atol: float = 1e-10, +) -> str: + """Format value for display in repr methods. + + For single values or uniform arrays, returns the formatted value. + For arrays with variation, returns a range showing min-max. + + Args: + value: Numeric value or container (DataArray, array, Series, DataFrame) + precision: Number of decimal places (default: 1) + atol: Absolute tolerance for considering values equal (default: 1e-10) + + Returns: + Formatted string representation: + - Single/uniform values: "100.0" + - Nearly uniform values: "~100.0" (values differ slightly but display similarly) + - Varying values: "50.0-150.0" (shows range from min to max) + + Raises: + TypeError: If value cannot be converted to numeric format + """ + # Handle simple scalar types + if isinstance(value, (int, float, np.integer, np.floating)): + return f'{float(value):.{precision}f}' + + # Extract array data for variation checking + arr = None + if isinstance(value, xr.DataArray): + arr = value.values.flatten() + elif isinstance(value, (np.ndarray, pd.Series)): + arr = np.asarray(value).flatten() + elif isinstance(value, pd.DataFrame): + arr = value.values.flatten() + else: + # Fallback for unknown types + try: + return f'{float(value):.{precision}f}' + except (TypeError, ValueError) as e: + raise TypeError(f'Cannot format value of type {type(value).__name__} for repr') from e + + # Normalize dtype and handle empties + arr = arr.astype(float, copy=False) + if arr.size == 0: + return '?' + + # Filter non-finite values + finite = arr[np.isfinite(arr)] + if finite.size == 0: + return 'nan' + + # Check for single value + if finite.size == 1: + return f'{float(finite[0]):.{precision}f}' + + # Check if all values are the same or very close + min_val = float(np.nanmin(finite)) + max_val = float(np.nanmax(finite)) + + # First check: values are essentially identical + if np.allclose(min_val, max_val, atol=atol): + return f'{float(np.mean(finite)):.{precision}f}' + + # Second check: display values are the same but actual values differ slightly + min_str = f'{min_val:.{precision}f}' + max_str = f'{max_val:.{precision}f}' + if min_str == max_str: + return f'~{min_str}' + + # Values vary significantly - show range + return f'{min_str}-{max_str}' + + +def _format_value_for_repr(value) -> str: + """Format a single value for display in repr. + + Args: + value: The value to format + + Returns: + Formatted string representation of the value + """ + # Format numeric types using specialized formatter + if isinstance(value, (int, float, np.integer, np.floating, np.ndarray, pd.Series, pd.DataFrame, xr.DataArray)): + try: + return numeric_to_str_for_repr(value) + except Exception: + value_repr = repr(value) + if len(value_repr) > 50: + value_repr = value_repr[:47] + '...' + return value_repr + + # Format dicts with numeric/array values nicely + elif isinstance(value, dict): + try: + formatted_items = [] + for k, v in value.items(): + if isinstance( + v, (int, float, np.integer, np.floating, np.ndarray, pd.Series, pd.DataFrame, xr.DataArray) + ): + v_str = numeric_to_str_for_repr(v) + else: + v_str = repr(v) + if len(v_str) > 30: + v_str = v_str[:27] + '...' + formatted_items.append(f'{repr(k)}: {v_str}') + value_repr = '{' + ', '.join(formatted_items) + '}' + if len(value_repr) > 50: + value_repr = value_repr[:47] + '...' + return value_repr + except Exception: + value_repr = repr(value) + if len(value_repr) > 50: + value_repr = value_repr[:47] + '...' + return value_repr + + # Default repr with truncation + else: + value_repr = repr(value) + if len(value_repr) > 50: + value_repr = value_repr[:47] + '...' + return value_repr + + +def build_repr_from_init( + obj: object, + excluded_params: set[str] | None = None, + label_as_positional: bool = True, + skip_default_size: bool = False, +) -> str: + """Build a repr string from __init__ signature, showing non-default parameter values. + + This utility function extracts common repr logic used across flixopt classes. + It introspects the __init__ method to build a constructor-style repr showing + only parameters that differ from their defaults. + + Args: + obj: The object to create repr for + excluded_params: Set of parameter names to exclude (e.g., {'self', 'inputs', 'outputs'}) + Default excludes 'self', 'label', and 'kwargs' + label_as_positional: If True and 'label' param exists, show it as first positional arg + skip_default_size: If True, skip 'size' parameter when it equals CONFIG.Modeling.big + + Returns: + Formatted repr string like: ClassName("label", param=value) + """ + if excluded_params is None: + excluded_params = {'self', 'label', 'kwargs'} + else: + # Always exclude 'self' + excluded_params = excluded_params | {'self'} + + try: + # Get the constructor arguments and their current values + init_signature = inspect.signature(obj.__init__) + init_params = init_signature.parameters + + # Check if this has a 'label' parameter - if so, show it first as positional + has_label = 'label' in init_params and label_as_positional + + # Build kwargs for non-default parameters + kwargs_parts = [] + label_value = None + + for param_name, param in init_params.items(): + # Skip *args and **kwargs + if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD): + continue + + # Handle label separately if showing as positional (check BEFORE excluded_params) + if param_name == 'label' and has_label: + label_value = getattr(obj, param_name, None) + continue + + # Now check if parameter should be excluded + if param_name in excluded_params: + continue + + # Get current value + value = getattr(obj, param_name, None) + + # Skip if value matches default + if param.default != inspect.Parameter.empty: + # Special handling for empty containers (even if default was None) + if isinstance(value, (dict, list, tuple, set)) and len(value) == 0: + if param.default is None or ( + isinstance(param.default, (dict, list, tuple, set)) and len(param.default) == 0 + ): + continue + + # Handle array comparisons (xarray, numpy) + elif isinstance(value, (xr.DataArray, np.ndarray)): + try: + if isinstance(param.default, (xr.DataArray, np.ndarray)): + # Compare arrays element-wise + if isinstance(value, xr.DataArray) and isinstance(param.default, xr.DataArray): + if value.equals(param.default): + continue + elif np.array_equal(value, param.default): + continue + elif isinstance(param.default, (int, float, np.integer, np.floating)): + # Compare array to scalar (e.g., after transform_data converts scalar to DataArray) + if isinstance(value, xr.DataArray): + if np.all(value.values == float(param.default)): + continue + elif isinstance(value, np.ndarray): + if np.all(value == float(param.default)): + continue + except Exception: + pass # If comparison fails, include in repr + + # Handle numeric comparisons (deals with 0 vs 0.0, int vs float) + elif isinstance(value, (int, float, np.integer, np.floating)) and isinstance( + param.default, (int, float, np.integer, np.floating) + ): + try: + if float(value) == float(param.default): + continue + except (ValueError, TypeError): + pass + + elif value == param.default: + continue + + # Skip None values if default is None + if value is None and param.default is None: + continue + + # Special case: hide CONFIG.Modeling.big for size parameter + if skip_default_size and param_name == 'size': + from .config import CONFIG + + try: + if isinstance(value, (int, float, np.integer, np.floating)): + if float(value) == CONFIG.Modeling.big: + continue + except Exception: + pass + + # Format value using helper function + value_repr = _format_value_for_repr(value) + kwargs_parts.append(f'{param_name}={value_repr}') + + # Build args string with label first as positional if present + if has_label and label_value is not None: + # Use label_full if available, otherwise label + if hasattr(obj, 'label_full'): + label_repr = repr(obj.label_full) + else: + label_repr = repr(label_value) + + if len(label_repr) > 50: + label_repr = label_repr[:47] + '...' + args_str = label_repr + if kwargs_parts: + args_str += ', ' + ', '.join(kwargs_parts) + else: + args_str = ', '.join(kwargs_parts) + + # Build final repr + class_name = obj.__class__.__name__ + + return f'{class_name}({args_str})' + + except Exception: + # Fallback if introspection fails + return f'{obj.__class__.__name__}()' + + +def format_flow_details(obj, has_inputs: bool = True, has_outputs: bool = True) -> str: + """Format inputs and outputs as indented bullet list. + + Args: + obj: Object with 'inputs' and/or 'outputs' attributes + has_inputs: Whether to check for inputs + has_outputs: Whether to check for outputs + + Returns: + Formatted string with flow details (including leading newline), or empty string if no flows + """ + flow_lines = [] + + if has_inputs and hasattr(obj, 'inputs') and obj.inputs: + flow_lines.append(' inputs:') + for flow in obj.inputs: + flow_lines.append(f' * {repr(flow)}') + + if has_outputs and hasattr(obj, 'outputs') and obj.outputs: + flow_lines.append(' outputs:') + for flow in obj.outputs: + flow_lines.append(f' * {repr(flow)}') + + return '\n' + '\n'.join(flow_lines) if flow_lines else '' + + +def format_title_with_underline(title: str, underline_char: str = '-') -> str: + """Format a title with underline of matching length. + + Args: + title: The title text + underline_char: Character to use for underline (default: '-') + + Returns: + Formatted string: "Title\\n-----\\n" + """ + return f'{title}\n{underline_char * len(title)}\n' + + +def format_sections_with_headers(sections: dict[str, str], underline_char: str = '-') -> list[str]: + """Format sections with underlined headers. + + Args: + sections: Dict mapping section headers to content + underline_char: Character for underlining headers + + Returns: + List of formatted section strings + """ + formatted_sections = [] + for section_header, section_content in sections.items(): + underline = underline_char * len(section_header) + formatted_sections.append(f'{section_header}\n{underline}\n{section_content}') + return formatted_sections + + +def build_metadata_info(parts: list[str], prefix: str = ' | ') -> str: + """Build metadata info string from parts. + + Args: + parts: List of metadata strings (empty strings are filtered out) + prefix: Prefix to add if parts is non-empty + + Returns: + Formatted info string or empty string + """ + # Filter out empty strings + parts = [p for p in parts if p] + if not parts: + return '' + info = ' | '.join(parts) + return prefix + info if prefix else info diff --git a/flixopt/results.py b/flixopt/results.py index 950570df3..9232f0175 100644 --- a/flixopt/results.py +++ b/flixopt/results.py @@ -17,6 +17,7 @@ from .color_processing import process_colors from .config import CONFIG from .flow_system import FlowSystem +from .structure import CompositeContainerMixin, ElementContainer, ResultsContainer if TYPE_CHECKING: import matplotlib.pyplot as plt @@ -53,7 +54,7 @@ class _FlowSystemRestorationError(Exception): pass -class CalculationResults: +class CalculationResults(CompositeContainerMixin['ComponentResults | BusResults | EffectResults | FlowResults']): """Comprehensive container for optimization calculation results and analysis tools. This class provides unified access to all optimization results including flow rates, @@ -238,13 +239,18 @@ def __init__( self.name = name self.model = model self.folder = pathlib.Path(folder) if folder is not None else pathlib.Path.cwd() / 'results' - self.components = { + + # Create ResultsContainers for better access patterns + components_dict = { label: ComponentResults(self, **infos) for label, infos in self.solution.attrs['Components'].items() } + self.components = ResultsContainer(elements=components_dict, element_type_name='component results') - self.buses = {label: BusResults(self, **infos) for label, infos in self.solution.attrs['Buses'].items()} + buses_dict = {label: BusResults(self, **infos) for label, infos in self.solution.attrs['Buses'].items()} + self.buses = ResultsContainer(elements=buses_dict, element_type_name='bus results') - self.effects = {label: EffectResults(self, **infos) for label, infos in self.solution.attrs['Effects'].items()} + effects_dict = {label: EffectResults(self, **infos) for label, infos in self.solution.attrs['Effects'].items()} + self.effects = ResultsContainer(elements=effects_dict, element_type_name='effect results') if 'Flows' not in self.solution.attrs: warnings.warn( @@ -252,11 +258,14 @@ def __init__( 'is not availlable. We recommend to evaluate your results with a version <2.2.0.', stacklevel=2, ) - self.flows = {} + flows_dict = {} + self._has_flow_data = False else: - self.flows = { + flows_dict = { label: FlowResults(self, **infos) for label, infos in self.solution.attrs.get('Flows', {}).items() } + self._has_flow_data = True + self.flows = ResultsContainer(elements=flows_dict, element_type_name='flow results') self.timesteps_extra = self.solution.indexes['time'] self.hours_per_timestep = FlowSystem.calculate_hours_per_timestep(self.timesteps_extra) @@ -273,16 +282,22 @@ def __init__( self.colors: dict[str, str] = {} - def __getitem__(self, key: str) -> ComponentResults | BusResults | EffectResults: - if key in self.components: - return self.components[key] - if key in self.buses: - return self.buses[key] - if key in self.effects: - return self.effects[key] - if key in self.flows: - return self.flows[key] - raise KeyError(f'No element with label {key} found.') + def _get_container_groups(self) -> dict[str, ResultsContainer]: + """Return ordered container groups for CompositeContainerMixin.""" + return { + 'Components': self.components, + 'Buses': self.buses, + 'Effects': self.effects, + 'Flows': self.flows, + } + + def __repr__(self) -> str: + """Return grouped representation of all results.""" + r = fx_io.format_title_with_underline(self.__class__.__name__, '=') + r += f'Name: "{self.name}"\nFolder: {self.folder}\n' + # Add grouped container view + r += '\n' + self._format_grouped_containers() + return r @property def storages(self) -> list[ComponentResults]: @@ -547,6 +562,8 @@ def flow_rates( To recombine filtered dataarrays, use `xr.concat` with dim 'flow': >>>xr.concat([results.flow_rates(start='Fernwärme'), results.flow_rates(end='Fernwärme')], dim='flow') """ + if not self._has_flow_data: + raise ValueError('Flow data is not available in this results object (pre-v2.2.0).') if self._flow_rates is None: self._flow_rates = self._assign_flow_coords( xr.concat( @@ -608,6 +625,8 @@ def sizes( >>>xr.concat([results.sizes(start='Fernwärme'), results.sizes(end='Fernwärme')], dim='flow') """ + if not self._has_flow_data: + raise ValueError('Flow data is not available in this results object (pre-v2.2.0).') if self._sizes is None: self._sizes = self._assign_flow_coords( xr.concat( @@ -620,11 +639,12 @@ def sizes( def _assign_flow_coords(self, da: xr.DataArray): # Add start and end coordinates + flows_list = list(self.flows.values()) da = da.assign_coords( { - 'start': ('flow', [flow.start for flow in self.flows.values()]), - 'end': ('flow', [flow.end for flow in self.flows.values()]), - 'component': ('flow', [flow.component for flow in self.flows.values()]), + 'start': ('flow', [flow.start for flow in flows_list]), + 'end': ('flow', [flow.end for flow in flows_list]), + 'component': ('flow', [flow.component for flow in flows_list]), } ) @@ -743,8 +763,6 @@ def _compute_effect_total( temporal = temporal.sum('time') if periodic.isnull().all(): return temporal.rename(f'{element}->{effect}') - if 'time' in temporal.indexes: - temporal = temporal.sum('time') return periodic + temporal total = xr.DataArray(0) @@ -1106,6 +1124,14 @@ def constraints(self) -> linopy.Constraints: raise ValueError('The linopy model is not available.') return self._calculation_results.model.constraints[self._constraint_names] + def __repr__(self) -> str: + """Return string representation with element info and dataset preview.""" + class_name = self.__class__.__name__ + header = f'{class_name}: "{self.label}"' + sol = self.solution.copy(deep=False) + sol.attrs = {} + return f'{header}\n{"-" * len(header)}\n{repr(sol)}' + def filter_solution( self, variable_dims: Literal['scalar', 'time', 'scenario', 'timeonly', 'scenarioonly'] | None = None, diff --git a/flixopt/structure.py b/flixopt/structure.py index 6ea618454..065769cd2 100644 --- a/flixopt/structure.py +++ b/flixopt/structure.py @@ -7,12 +7,16 @@ import inspect import logging +import re from dataclasses import dataclass +from difflib import get_close_matches from io import StringIO from typing import ( TYPE_CHECKING, Any, + Generic, Literal, + TypeVar, ) import linopy @@ -168,7 +172,7 @@ def solution(self): }, 'Effects': { effect.label_full: effect.submodel.results_structure() - for effect in sorted(self.flow_system.effects, key=lambda effect: effect.label_full.upper()) + for effect in sorted(self.flow_system.effects.values(), key=lambda effect: effect.label_full.upper()) }, 'Flows': { flow.label_full: flow.submodel.results_structure() @@ -242,9 +246,7 @@ def __repr__(self) -> str: } # Format sections with headers and underlines - formatted_sections = [] - for section_header, section_content in sections.items(): - formatted_sections.append(f'{section_header}\n{"-" * len(section_header)}\n{section_content}') + formatted_sections = fx_io.format_sections_with_headers(sections) title = f'FlowSystemModel ({self.type})' all_sections = '\n'.join(formatted_sections) @@ -793,40 +795,7 @@ def to_json(self, path: str | pathlib.Path): def __repr__(self): """Return a detailed string representation for debugging.""" - try: - # Get the constructor arguments and their current values - init_signature = inspect.signature(self.__init__) - init_args = init_signature.parameters - - # Create a dictionary with argument names and their values, with better formatting - args_parts = [] - for name in init_args: - if name == 'self': - continue - value = getattr(self, name, None) - # Truncate long representations - value_repr = repr(value) - if len(value_repr) > 50: - value_repr = value_repr[:47] + '...' - args_parts.append(f'{name}={value_repr}') - - args_str = ', '.join(args_parts) - return f'{self.__class__.__name__}({args_str})' - except Exception: - # Fallback if introspection fails - return f'{self.__class__.__name__}()' - - def __str__(self): - """Return a user-friendly string representation.""" - try: - data = self.get_structure(clean=True, stats=True) - with StringIO() as output_buffer: - console = Console(file=output_buffer, width=1000) # Adjust width as needed - console.print(Pretty(data, expand_all=True, indent_guides=True)) - return output_buffer.getvalue() - except Exception: - # Fallback if structure generation fails - return f'{self.__class__.__name__} instance' + return fx_io.build_repr_from_init(self, excluded_params={'self', 'label', 'kwargs'}) def copy(self) -> Interface: """ @@ -876,6 +845,10 @@ def create_model(self, model: FlowSystemModel) -> ElementModel: def label_full(self) -> str: return self.label + def __repr__(self) -> str: + """Return string representation.""" + return fx_io.build_repr_from_init(self, excluded_params={'self', 'label', 'kwargs'}, skip_default_size=True) + @staticmethod def _valid_label(label: str) -> str: """Checks if the label is valid. If not, it is replaced by the default label. @@ -895,6 +868,329 @@ def _valid_label(label: str) -> str: return label +# Precompiled regex pattern for natural sorting +_NATURAL_SPLIT = re.compile(r'(\d+)') + + +def _natural_sort_key(text): + """Sort key for natural ordering (e.g., bus1, bus2, bus10 instead of bus1, bus10, bus2).""" + return [int(c) if c.isdigit() else c.lower() for c in _NATURAL_SPLIT.split(text)] + + +# Type variable for containers +T = TypeVar('T') + + +class ContainerMixin(dict[str, T]): + """ + Mixin providing shared container functionality with nice repr and error messages. + + Subclasses must implement _get_label() to extract the label from elements. + """ + + def __init__( + self, + elements: list[T] | dict[str, T] | None = None, + element_type_name: str = 'elements', + ): + """ + Args: + elements: Initial elements to add (list or dict) + element_type_name: Name for display (e.g., 'components', 'buses') + """ + super().__init__() + self._element_type_name = element_type_name + + if elements is not None: + if isinstance(elements, dict): + for element in elements.values(): + self.add(element) + else: + for element in elements: + self.add(element) + + def _get_label(self, element: T) -> str: + """ + Extract label from element. Must be implemented by subclasses. + + Args: + element: Element to get label from + + Returns: + Label string + """ + raise NotImplementedError('Subclasses must implement _get_label()') + + def add(self, element: T) -> None: + """Add an element to the container.""" + label = self._get_label(element) + if label in self: + raise ValueError( + f'Element with label "{label}" already exists in {self._element_type_name}. ' + f'Each element must have a unique label.' + ) + self[label] = element + + def __setitem__(self, label: str, element: T) -> None: + """Set element with validation.""" + element_label = self._get_label(element) + if label != element_label: + raise ValueError( + f'Key "{label}" does not match element label "{element_label}". ' + f'Use the correct label as key or use .add() method.' + ) + super().__setitem__(label, element) + + def __getitem__(self, label: str) -> T: + """ + Get element by label with helpful error messages. + + Args: + label: Label of the element to retrieve + + Returns: + The element with the given label + + Raises: + KeyError: If element is not found, with suggestions for similar labels + """ + try: + return super().__getitem__(label) + except KeyError: + # Provide helpful error with close matches suggestions + suggestions = get_close_matches(label, self.keys(), n=3, cutoff=0.6) + error_msg = f'Element "{label}" not found in {self._element_type_name}.' + if suggestions: + error_msg += f' Did you mean: {", ".join(suggestions)}?' + else: + available = list(self.keys()) + if len(available) <= 5: + error_msg += f' Available: {", ".join(available)}' + else: + error_msg += f' Available: {", ".join(available[:5])} ... (+{len(available) - 5} more)' + raise KeyError(error_msg) from None + + def __repr__(self) -> str: + """Return a string representation similar to linopy.model.Variables.""" + count = len(self) + title = f'{self._element_type_name.capitalize()} ({count} item{"s" if count != 1 else ""})' + + if not self: + r = fx_io.format_title_with_underline(title) + r += '\n' + else: + r = fx_io.format_title_with_underline(title) + for name in sorted(self.keys(), key=_natural_sort_key): + r += f' * {name}\n' + + return r + + +class ElementContainer(ContainerMixin[T]): + """ + Container for Element objects (Component, Bus, Flow, Effect). + + Uses element.label_full for keying. + """ + + def _get_label(self, element: T) -> str: + """Extract label_full from Element.""" + return element.label_full + + +class ResultsContainer(ContainerMixin[T]): + """ + Container for Results objects (ComponentResults, BusResults, etc). + + Uses element.label for keying. + """ + + def _get_label(self, element: T) -> str: + """Extract label from Results object.""" + return element.label + + +T_element = TypeVar('T_element') + + +class CompositeContainerMixin(Generic[T_element]): + """ + Mixin providing unified dict-like access across multiple typed containers. + + This mixin enables classes that manage multiple containers (e.g., components, + buses, effects, flows) to provide a unified interface for accessing elements + across all containers, as if they were a single collection. + + Type Parameter: + T_element: The type of elements stored in the containers. Can be a union type + for containers holding multiple types (e.g., 'ComponentResults | BusResults'). + + Key Features: + - Dict-like access: `obj['element_name']` searches all containers + - Iteration: `for label in obj:` iterates over all elements + - Membership: `'element' in obj` checks across all containers + - Standard dict methods: keys(), values(), items() + - Grouped display: Formatted repr showing elements by type + - Type hints: Full IDE and type checker support + + Subclasses must implement: + _get_container_groups() -> dict[str, dict]: + Returns a dictionary mapping group names (e.g., 'Components', 'Buses') + to container dictionaries. Containers are displayed in the order returned. + + Example: + ```python + class MySystem(CompositeContainerMixin[Component | Bus]): + def __init__(self): + self.components = {'Boiler': Component(...), 'CHP': Component(...)} + self.buses = {'Heat': Bus(...), 'Power': Bus(...)} + + def _get_container_groups(self): + return { + 'Components': self.components, + 'Buses': self.buses, + } + + + system = MySystem() + comp = system['Boiler'] # Type: Component | Bus (with proper IDE support) + 'Heat' in system # True + labels = system.keys() # Type: list[str] + elements = system.values() # Type: list[Component | Bus] + ``` + + Integration with ContainerMixin: + This mixin is designed to work alongside ContainerMixin-based containers + (ElementContainer, ResultsContainer) by aggregating them into a unified + interface while preserving their individual functionality. + """ + + def _get_container_groups(self) -> dict[str, ContainerMixin[Any]]: + """ + Return ordered dict of container groups to aggregate. + + Returns: + Dictionary mapping group names to container objects (e.g., ElementContainer, ResultsContainer). + Group names should be capitalized (e.g., 'Components', 'Buses'). + Order determines display order in __repr__. + + Example: + ```python + return { + 'Components': self.components, + 'Buses': self.buses, + 'Effects': self.effects, + } + ``` + """ + raise NotImplementedError('Subclasses must implement _get_container_groups()') + + def __getitem__(self, key: str) -> T_element: + """ + Get element by label, searching all containers. + + Args: + key: Element label to find + + Returns: + The element with the given label + + Raises: + KeyError: If element not found, with helpful suggestions + """ + # Search all containers in order + for container in self._get_container_groups().values(): + if key in container: + return container[key] + + # Element not found - provide helpful error + all_elements = {} + for container in self._get_container_groups().values(): + all_elements.update(container) + + suggestions = get_close_matches(key, all_elements.keys(), n=3, cutoff=0.6) + error_msg = f'Element "{key}" not found.' + + if suggestions: + error_msg += f' Did you mean: {", ".join(suggestions)}?' + else: + available = list(all_elements.keys()) + if len(available) <= 5: + error_msg += f' Available: {", ".join(available)}' + else: + error_msg += f' Available: {", ".join(available[:5])} ... (+{len(available) - 5} more)' + + raise KeyError(error_msg) + + def __iter__(self): + """Iterate over all element labels across all containers.""" + for container in self._get_container_groups().values(): + yield from container.keys() + + def __len__(self) -> int: + """Return total count of elements across all containers.""" + return sum(len(container) for container in self._get_container_groups().values()) + + def __contains__(self, key: str) -> bool: + """Check if element exists in any container.""" + return any(key in container for container in self._get_container_groups().values()) + + def keys(self) -> list[str]: + """Return all element labels across all containers.""" + return list(self) + + def values(self) -> list[T_element]: + """Return all element objects across all containers.""" + vals = [] + for container in self._get_container_groups().values(): + vals.extend(container.values()) + return vals + + def items(self) -> list[tuple[str, T_element]]: + """Return (label, element) pairs for all elements.""" + items = [] + for container in self._get_container_groups().values(): + items.extend(container.items()) + return items + + def _format_grouped_containers(self, title: str | None = None) -> str: + """ + Format containers as grouped string representation using each container's repr. + + Args: + title: Optional title for the representation. If None, no title is shown. + + Returns: + Formatted string with groups and their elements. + Empty groups are automatically hidden. + + Example output: + ``` + Components (1 item) + ------------------- + * Boiler + + Buses (2 items) + --------------- + * Heat + * Power + ``` + """ + parts = [] + + if title: + parts.append(fx_io.format_title_with_underline(title)) + + container_groups = self._get_container_groups() + for container in container_groups.values(): + if container: # Only show non-empty groups + if parts: # Add spacing between sections + parts.append('') + parts.append(repr(container).rstrip('\n')) + + return '\n'.join(parts) + + class Submodel(SubmodelsMixin): """Stores Variables and Constraints. Its a subset of a FlowSystemModel. Variables and constraints are stored in the main FlowSystemModel, and are referenced here. @@ -1056,9 +1352,7 @@ def __repr__(self) -> str: } # Format sections with headers and underlines - formatted_sections = [] - for section_header, section_content in sections.items(): - formatted_sections.append(f'{section_header}\n{"-" * len(section_header)}\n{section_content}') + formatted_sections = fx_io.format_sections_with_headers(sections) model_string = f'Submodel "{self.label_of_model}":' all_sections = '\n'.join(formatted_sections) @@ -1102,7 +1396,7 @@ def __contains__(self, name: str) -> bool: def __repr__(self) -> str: """Simple representation of the submodels collection.""" if not self.data: - return 'flixopt.structure.Submodels:\n----------------------------\n \n' + return fx_io.format_title_with_underline('flixopt.structure.Submodels') + ' \n' total_vars = sum(len(submodel.variables) for submodel in self.data.values()) total_cons = sum(len(submodel.constraints) for submodel in self.data.values()) @@ -1110,18 +1404,15 @@ def __repr__(self) -> str: title = ( f'flixopt.structure.Submodels ({total_vars} vars, {total_cons} constraints, {len(self.data)} submodels):' ) - underline = '-' * len(title) - if not self.data: - return f'{title}\n{underline}\n \n' - sub_models_string = '' + result = fx_io.format_title_with_underline(title) for name, submodel in self.data.items(): type_name = submodel.__class__.__name__ var_count = len(submodel.variables) con_count = len(submodel.constraints) - sub_models_string += f'\n * {name} [{type_name}] ({var_count}v/{con_count}c)' + result += f' * {name} [{type_name}] ({var_count}v/{con_count}c)\n' - return f'{title}\n{underline}{sub_models_string}\n' + return result def items(self) -> ItemsView[str, Submodel]: return self.data.items() diff --git a/tests/test_functional.py b/tests/test_functional.py index 0f9fe02ef..a83bf112f 100644 --- a/tests/test_functional.py +++ b/tests/test_functional.py @@ -152,7 +152,7 @@ def test_fixed_size(solver_fixture, time_steps_fixture): ) solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] + boiler = flow_system['Boiler'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(), @@ -193,7 +193,7 @@ def test_optimize_size(solver_fixture, time_steps_fixture): ) solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] + boiler = flow_system['Boiler'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(), @@ -234,7 +234,7 @@ def test_size_bounds(solver_fixture, time_steps_fixture): ) solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] + boiler = flow_system['Boiler'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(), @@ -289,8 +289,8 @@ def test_optional_invest(solver_fixture, time_steps_fixture): ) solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] - boiler_optional = flow_system.all_elements['Boiler_optional'] + boiler = flow_system['Boiler'] + boiler_optional = flow_system['Boiler_optional'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(), @@ -343,7 +343,7 @@ def test_on(solver_fixture, time_steps_fixture): ) solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] + boiler = flow_system['Boiler'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(), @@ -387,7 +387,7 @@ def test_off(solver_fixture, time_steps_fixture): ) solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] + boiler = flow_system['Boiler'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(), @@ -438,7 +438,7 @@ def test_switch_on_off(solver_fixture, time_steps_fixture): ) solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] + boiler = flow_system['Boiler'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(), @@ -502,7 +502,7 @@ def test_on_total_max(solver_fixture, time_steps_fixture): ) solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] + boiler = flow_system['Boiler'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(), @@ -555,13 +555,13 @@ def test_on_total_bounds(solver_fixture, time_steps_fixture): ), ), ) - flow_system.all_elements['Wärmelast'].inputs[0].fixed_relative_profile = np.array( + flow_system['Wärmelast'].inputs[0].fixed_relative_profile = np.array( [0, 10, 20, 0, 12] ) # Else its non deterministic solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] - boiler_backup = flow_system.all_elements['Boiler_backup'] + boiler = flow_system['Boiler'] + boiler_backup = flow_system['Boiler_backup'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(), @@ -624,12 +624,12 @@ def test_consecutive_on_off(solver_fixture, time_steps_fixture): Q_th=fx.Flow('Q_th', bus='Fernwärme', size=100), ), ) - flow_system.all_elements['Wärmelast'].inputs[0].fixed_relative_profile = np.array([5, 10, 20, 18, 12]) + flow_system['Wärmelast'].inputs[0].fixed_relative_profile = np.array([5, 10, 20, 18, 12]) # Else its non deterministic solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] - boiler_backup = flow_system.all_elements['Boiler_backup'] + boiler = flow_system['Boiler'] + boiler_backup = flow_system['Boiler_backup'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(), @@ -686,13 +686,13 @@ def test_consecutive_off(solver_fixture, time_steps_fixture): ), ), ) - flow_system.all_elements['Wärmelast'].inputs[0].fixed_relative_profile = np.array( + flow_system['Wärmelast'].inputs[0].fixed_relative_profile = np.array( [5, 0, 20, 18, 12] ) # Else its non deterministic solve_and_load(flow_system, solver_fixture) - boiler = flow_system.all_elements['Boiler'] - boiler_backup = flow_system.all_elements['Boiler_backup'] + boiler = flow_system['Boiler'] + boiler_backup = flow_system['Boiler_backup'] costs = flow_system.effects['costs'] assert_allclose( costs.submodel.total.solution.item(),