feat: generic multi-file simulation combining via BaseSimulation.from_paths

src/gat/interfaces.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from pathlib import Path
 
 import pandas as pd
 
@@ -145,3 +146,32 @@ def get_default_compositions(self) -> list[DatasetComposition]:
         ActivePowerVariable__RenewableDispatch, etc.
         """
         return []
+
+    @classmethod
+    def from_paths(
+        cls, paths: str | Path | list[str | Path], **kwargs
+    ) -> "BaseSimulation":
+        """Construct from one or more partition files (e.g. weekly/monthly
+        chunks of one logical simulation, submitted in parallel) treated
+        as a single combined simulation.
+
+        Default: instantiate ``cls(path, **kwargs)`` once per path and
+        combine their datasets via
+        ``gat.simulations.multi_file.MultiFileSimulation`` — free for any
+        subclass whose constructor takes a single file path, which covers
+        most extensions, without them needing to implement multi-file
+        support themselves.
+
+        Override this when your format combines through a different
+        mechanism than "N single-file objects + pandas concat" — e.g. SQL
+        ATTACH across converted files (see
+        ``PlexosDuckDBSimulation.from_paths``).
+        """
+        path_list = [paths] if isinstance(paths, (str, Path)) else list(paths)
+
+        if len(path_list) == 1:
+            return cls(path_list[0], **kwargs)
+
+        from .simulations.multi_file import MultiFileSimulation
+
+        return MultiFileSimulation(cls, path_list, **kwargs)

src/gat/server/ingest.py

@@ -251,16 +251,15 @@ def _create_parser(
             raise ValueError("sienna handler requires simulation_paths")
 
         system = SiennaSystem(system_path)
-        # SiennaSimulation takes a single path; use first file
-        sim_path = (
-            simulation_paths[0]
-            if isinstance(simulation_paths, list)
-            else simulation_paths
-        )
         # ``model`` becomes ``selected_model`` inside the parser, which
         # decides which group of HDF5 datasets get exposed.
         sienna_model = None if model in (None, "default") else model
-        simulation = SiennaSimulation(sim_path, simulation=sienna_model)
+        # from_paths handles both a single path and multiple partition
+        # files (combined into one logical simulation) -- previously this
+        # silently dropped every file after the first.
+        simulation = SiennaSimulation.from_paths(
+            simulation_paths, simulation=sienna_model
+        )
         return system, simulation
 
     elif handler == "reeds":
@@ -281,14 +280,19 @@ def _create_parser(
         return system, simulation
 
     elif handler == "plexos":
-        from gat.systems.plexos import PlexosSystem
-        from gat.simulations.plexos import PlexosSimulation
+        # gat.systems.plexos / gat.simulations.plexos (the legacy H5
+        # backend's would-be v1 wrappers) never existed as modules -- this
+        # branch has always raised ImportError. Use the duckdb backend,
+        # which already reads native PLEXOS Solution.zip/.duckdb files
+        # directly and needs no separate system file.
+        from gat.systems.plexos_duckdb import PlexosDuckDBSystem
+        from gat.simulations.plexos_duckdb import PlexosDuckDBSimulation
 
         if not simulation_paths:
             raise ValueError("plexos handler requires simulation_paths")
 
-        system = PlexosSystem()  # Plexos may not have a separate system file
-        simulation = PlexosSimulation(simulation_paths)
+        system = PlexosDuckDBSystem(simulation_paths)
+        simulation = PlexosDuckDBSimulation.from_paths(simulation_paths)
         return system, simulation
 
     else:

src/gat/simulations/generic_aggregator.py

@@ -74,6 +74,7 @@ def __init__(
         parser_class: Type[BaseSimulationParser],
         parallel: bool = True,
         max_workers: Optional[int] = None,
+        parser_kwargs: Optional[Dict[str, Any]] = None,
     ):
         """
         Initialize the simulation aggregator.
@@ -83,6 +84,10 @@ def __init__(
             parser_class: Parser class to instantiate (must inherit from BaseSimulationParser)
             parallel: Whether to load files in parallel (default: True)
             max_workers: Maximum number of parallel workers (default: CPU count)
+            parser_kwargs: Extra keyword arguments forwarded to
+                ``parser_class(file_path, **parser_kwargs)`` for every file.
+                For parser classes whose constructor needs more than a bare
+                path (default: none).
 
         Raises:
             ValueError: If no valid files provided or parser_class is invalid
@@ -96,6 +101,7 @@ def __init__(
         self.parser_class = parser_class
         self.parallel = parallel
         self.max_workers = max_workers or mp.cpu_count()
+        self.parser_kwargs = parser_kwargs or {}
 
         # Validate inputs
         if not self.file_paths:
@@ -144,7 +150,7 @@ def _initialize_parsers_sequential(self) -> List[BaseSimulationParser]:
                 logger.debug(
                     f"Loading file {i + 1}/{len(self.file_paths)}: {file_path.name}"
                 )
-                parser = self.parser_class(str(file_path))  # type: ignore
+                parser = self.parser_class(str(file_path), **self.parser_kwargs)  # type: ignore
                 parsers.append(parser)
             except Exception as e:
                 logger.error(f"Failed to load {file_path}: {e}")
@@ -199,7 +205,7 @@ def _load_single_file(self, file_path: str) -> BaseSimulationParser:
         Returns:
             Initialized parser instance
         """
-        return self.parser_class(file_path)  # type: ignore
+        return self.parser_class(file_path, **self.parser_kwargs)  # type: ignore
 
     def _validate_parsers(self):
         """Validate that all parsers have compatible simulation models."""

src/gat/simulations/multi_file.py

@@ -0,0 +1,73 @@
+"""Default multi-file BaseSimulation implementation.
+
+This is what BaseSimulation.from_paths() falls back to when a format
+doesn't override it: one inner-class instance per path, datasets combined
+with gat.simulations.utils.combine_overlapping_frames. It's the "parse
+each file into a Python object, then let GAT combine them" case -- the
+common one, free for any BaseSimulation subclass whose constructor takes
+a single file path.
+
+Formats that combine through a different mechanism (e.g. PlexosDuckDBSimulation,
+which ATTACHes multiple files at the SQL layer) override from_paths
+directly instead of going through this class.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import TYPE_CHECKING, Type
+
+import pandas as pd
+
+from ..interfaces import BaseSimulation
+from .utils import combine_overlapping_frames
+
+if TYPE_CHECKING:
+    from ..categories import CategoryMap
+    from ..datasets import DatasetComposition, DatasetInfo
+
+
+class MultiFileSimulation(BaseSimulation):
+    """Combines N single-file BaseSimulation instances into one.
+
+    Args:
+        inner_cls: A BaseSimulation subclass whose constructor accepts a
+            single path (plus optional kwargs).
+        paths: The partition files to combine.
+        merge_strategy: Passed to combine_overlapping_frames for every
+            dataset -- "earlier_wins"/"later_wins" (or the legacy
+            "right"/"left" spellings). Default "earlier_wins".
+        **inner_kwargs: Forwarded to inner_cls(path, **inner_kwargs) for
+            every instance.
+    """
+
+    def __init__(
+        self,
+        inner_cls: Type[BaseSimulation],
+        paths: list[str | Path],
+        merge_strategy: str = "earlier_wins",
+        **inner_kwargs,
+    ):
+        if len(paths) < 1:
+            raise ValueError("MultiFileSimulation requires at least one path")
+        self._instances = [inner_cls(p, **inner_kwargs) for p in paths]
+        self._merge_strategy = merge_strategy
+
+    def list_datasets(self) -> list["DatasetInfo"]:
+        """Union of every instance's datasets by name -- metadata comes
+        from whichever instance first reported a given name."""
+        by_name: dict[str, "DatasetInfo"] = {}
+        for instance in self._instances:
+            for info in instance.list_datasets():
+                by_name.setdefault(info.name, info)
+        return list(by_name.values())
+
+    def get_dataset(self, name: str) -> pd.DataFrame:
+        frames = [instance.get_dataset(name) for instance in self._instances]
+        return combine_overlapping_frames(frames, merge_strategy=self._merge_strategy)
+
+    def get_default_category_maps(self) -> list["CategoryMap"]:
+        return self._instances[0].get_default_category_maps()
+
+    def get_default_compositions(self) -> list["DatasetComposition"]:
+        return self._instances[0].get_default_compositions()

src/gat/simulations/plexos_duckdb.py

@@ -85,6 +85,15 @@ def __init__(
             len(self._get_raw_tables()),
         )
 
+    @classmethod
+    def from_paths(cls, paths, **kwargs) -> "PlexosDuckDBSimulation":
+        """Override of BaseSimulation.from_paths: PlexosDuckDBSource already
+        combines multiple files by ATTACHing each as its own DuckDB schema
+        (see gat.datahelpers.plexos_duckdb), so there's no need to route
+        through MultiFileSimulation's per-file-object model — just pass
+        the paths straight through."""
+        return cls(paths, **kwargs)
+
     @property
     def source(self) -> PlexosDuckDBSource:
         """Access the underlying PlexosDuckDBSource for advanced use."""

src/gat/simulations/sienna_v1.py

@@ -89,6 +89,48 @@ def __init__(
             len(self._get_raw_datasets()),
         )
 
+    @classmethod
+    def from_paths(
+        cls,
+        paths: str | Path | list[str | Path],
+        simulation: str | None = None,
+        **kwargs,
+    ) -> "BaseSimulation":
+        """Override of BaseSimulation.from_paths: the across-file merge
+        direction comes from the selected model's own
+        ``SiennaModelConfig.merge`` (already exposed as
+        ``SiennaSimulationParser.merge_strategy``) rather than
+        ``MultiFileSimulation``'s generic default -- a UC/decision model
+        and an emulation model can legitimately want different behavior
+        at a partition seam, and this reuses the existing per-model
+        setting instead of picking one global default.
+        """
+        path_list = [paths] if isinstance(paths, (str, Path)) else list(paths)
+
+        if len(path_list) == 1:
+            return cls(path_list[0], simulation=simulation, **kwargs)
+
+        from .sienna import SiennaSimulationParser
+
+        # Every partition of one logical simulation shares the same
+        # model, so the first file's config is representative -- this is
+        # a metadata-only read (SiennaSimulationConfig.from_h5_file opens
+        # the file in a `with` block), not a full data parse.
+        probe = SiennaSimulationParser(str(path_list[0]))
+        if simulation is not None:
+            probe.selected_model = simulation
+        merge_strategy = probe.merge_strategy or "earlier_wins"
+
+        from .multi_file import MultiFileSimulation
+
+        return MultiFileSimulation(
+            cls,
+            path_list,
+            merge_strategy=merge_strategy,
+            simulation=simulation,
+            **kwargs,
+        )
+
     @property
     def parser(self) -> object:
         """Access the underlying SiennaSimulationParser for advanced use."""

src/gat/simulations/utils.py

@@ -31,7 +31,17 @@
     import polars as pl
 
 
-block_combination_strategy = Literal["left", "right"]
+block_combination_strategy = Literal["left", "right", "earlier_wins", "later_wins"]
+
+# "left"/"right" describe which side gets truncated at the overlap, not
+# which side wins -- confusing enough that multiple docstrings in this
+# module apologize for it. "earlier_wins"/"later_wins" are the canonical,
+# self-documenting names going forward; "left"/"right" stay accepted for
+# backward compatibility with existing call sites.
+_MERGE_STRATEGY_ALIASES = {
+    "earlier_wins": "right",
+    "later_wins": "left",
+}
 
 
 def resolve_compositions(
@@ -81,20 +91,20 @@ def dedup_slices(
 
     See SiennaSimulationParser._get_decision_data() for detailed strategy examples.
     """
-    # Create mapping of block start time to (original_index, block)
-    block_mapping = {}
-    for i, block in enumerate(blocks):
-        start_time = block.min()
-        block_mapping[start_time] = (i, block)
-
-    # Sort blocks by start time
-    sorted_start_times = sorted(block_mapping.keys())
+    # (start_time, original_index, block) triples, sorted by start_time.
+    # A plain dict keyed by start_time would silently clobber one block
+    # whenever two blocks share an identical start timestamp; sort() is
+    # stable, so ties instead keep their original input order.
+    indexed_blocks = sorted(
+        ((block.min(), i, block) for i, block in enumerate(blocks)),
+        key=lambda t: t[0],
+    )
+    sorted_start_times = [t[0] for t in indexed_blocks]
 
     # Create result list in original order
     result = [None] * len(blocks)
 
-    for j, start_time in enumerate(sorted_start_times):
-        original_idx, current_block = block_mapping[start_time]
+    for j, (start_time, original_idx, current_block) in enumerate(indexed_blocks):
 
         if ignore_previous:
             # LEFT strategy: Keep data from current time forward, remove overlap with next block
@@ -130,8 +140,7 @@ def dedup_slices(
             # RIGHT strategy: Remove overlap with previous block, keep data forward
             if j > 0:
                 # Not the first block - find overlap with previous block
-                prev_start_time = sorted_start_times[j - 1]
-                prev_original_idx, prev_block = block_mapping[prev_start_time]
+                _, _, prev_block = indexed_blocks[j - 1]
                 prev_end_time = prev_block.max()
 
                 # Find where previous block ends in current block
@@ -186,8 +195,11 @@ def combine_overlapping_frames(
 
     Args:
         frames: DataFrames to combine, each with a sortable time index.
-        merge_strategy: "right" (earlier wins, default) or "left" (later
-            wins) — see the module docstring for the visual explanation.
+        merge_strategy: "right"/"earlier_wins" (default) or "left"/
+            "later_wins" — "earlier_wins"/"later_wins" are the canonical
+            names (self-documenting); "left"/"right" remain accepted for
+            backward compatibility. See the module docstring for the
+            visual explanation of what gets truncated.
 
     Returns:
         A single combined, chronologically-sorted DataFrame.
@@ -197,6 +209,7 @@ def combine_overlapping_frames(
     if len(frames) == 1:
         return frames[0]
 
+    merge_strategy = _MERGE_STRATEGY_ALIASES.get(merge_strategy, merge_strategy)
     sorted_frames = sorted(frames, key=lambda df: df.index.min())
     ignore_previous = merge_strategy == "left"