fix: allow expand() on clustered FlowSystem without a solution

flixopt/transform_accessor.py

@@ -801,19 +801,20 @@ def expand_flow_system(self) -> FlowSystem:
 
         # 2. Expand solution (with segment total correction for segmented systems)
         reduced_solution = self._fs.solution
-        sol_coord_cache = {k: v for k, v in reduced_solution.coords.items()}
-        sol_coord_names = set(sol_coord_cache)
-        expanded_sol_vars = {}
-        for name in reduced_solution.variables:
-            if name in sol_coord_names:
-                continue
-            da = self._fast_get_da(reduced_solution, name, sol_coord_cache)
-            expanded_sol_vars[name] = self.expand_dataarray(da, name, is_solution=True)
-        expanded_fs._solution = xr.Dataset(expanded_sol_vars, attrs=reduced_solution.attrs)
-        expanded_fs._solution = expanded_fs._solution.reindex(time=self._original_timesteps_extra)
-
-        # 3. Combine charge_state with SOC_boundary for intercluster storages
-        self._combine_intercluster_charge_states(expanded_fs, reduced_solution)
+        if reduced_solution is not None:
+            sol_coord_cache = {k: v for k, v in reduced_solution.coords.items()}
+            sol_coord_names = set(sol_coord_cache)
+            expanded_sol_vars = {}
+            for name in reduced_solution.variables:
+                if name in sol_coord_names:
+                    continue
+                da = self._fast_get_da(reduced_solution, name, sol_coord_cache)
+                expanded_sol_vars[name] = self.expand_dataarray(da, name, is_solution=True)
+            expanded_fs._solution = xr.Dataset(expanded_sol_vars, attrs=reduced_solution.attrs)
+            expanded_fs._solution = expanded_fs._solution.reindex(time=self._original_timesteps_extra)
+
+            # 3. Combine charge_state with SOC_boundary for intercluster storages
+            self._combine_intercluster_charge_states(expanded_fs, reduced_solution)
 
         # Log expansion info
         has_periods = self._fs.periods is not None
@@ -1899,15 +1900,13 @@ def _validate_for_expansion(self) -> Clustering:
             The Clustering object.
 
         Raises:
-            ValueError: If FlowSystem wasn't created with cluster() or has no solution.
+            ValueError: If FlowSystem wasn't created with cluster().
         """
 
         if self._fs.clustering is None:
             raise ValueError(
                 'expand() requires a FlowSystem created with cluster(). This FlowSystem has no aggregation info.'
             )
-        if self._fs.solution is None:
-            raise ValueError('FlowSystem has no solution. Run optimize() or solve() first.')
 
         return self._fs.clustering
 
@@ -1932,7 +1931,6 @@ def expand(self) -> FlowSystem:
 
         Raises:
             ValueError: If the FlowSystem was not created with ``cluster()``.
-            ValueError: If the FlowSystem has no solution.
 
         Examples:
             Two-stage optimization with expansion:

tests/test_clustering/test_cluster_reduce_expand.py

@@ -211,8 +211,8 @@ def test_expand_withoutclustering_raises(solver_fixture, timesteps_2_days):
         fs.transform.expand()
 
 
-def test_expand_without_solution_raises(timesteps_8_days):
-    """Test that expand raises error if no solution."""
+def test_expand_without_solution(timesteps_8_days):
+    """Test that expand works without a solution (e.g. for inspecting clustering_data)."""
     fs = create_simple_system(timesteps_8_days)
 
     fs_reduced = fs.transform.cluster(
@@ -221,8 +221,9 @@ def test_expand_without_solution_raises(timesteps_8_days):
     )
     # Don't optimize - no solution
 
-    with pytest.raises(ValueError, match='no solution'):
-        fs_reduced.transform.expand()
+    fs_expanded = fs_reduced.transform.expand()
+    assert fs_expanded.solution is None
+    assert len(fs_expanded.timesteps) == len(timesteps_8_days)
 
 
 # ==================== Multi-dimensional Tests ====================