Batch API

Parallel simulation execution with thread-pool parallelism.

simulate_batch

idfkit.simulation.batch.simulate_batch(jobs, *, energyplus=None, max_workers=None, cache=None, progress=None, fs=None, on_progress=None)

Run multiple EnergyPlus simulations in parallel.

Uses ThreadPoolExecutor to dispatch simulations concurrently. Individual job failures are captured as failed SimulationResult entries -- the batch never raises due to a single job failing.

Parameters:

Name	Type	Description	Default
jobs	Sequence[SimulationJob]	Sequence of simulation jobs to execute.	required
energyplus	EnergyPlusConfig | None	Shared EnergyPlus configuration (auto-discovered if None).	None
max_workers	int | None	Maximum number of concurrent simulations. Defaults to min(len(jobs), os.cpu_count() or 1).	None
cache	SimulationCache | None	Optional simulation cache for content-hash lookups.	None
progress	Callable[..., None] | None	Optional callback invoked after each job completes. Called as progress(completed=N, total=M, label=label, success=bool).	None
fs	FileSystem | None	Optional file system backend passed through to each simulate call.	None
on_progress	Callable[[SimulationProgress], None] | None	Optional callback invoked with SimulationProgress events during each individual simulation. Events include job_index and job_label to identify which batch job they belong to. The "tqdm" shorthand is not supported for batch runners; use tqdm_progress with a custom per-job callback instead.	None

Returns:

Type	Description
BatchResult	A BatchResult with results in the same order as jobs.

Raises:

Type	Description
ValueError	If jobs is empty.

Source code in idfkit/simulation/batch.py

def simulate_batch(
    jobs: Sequence[SimulationJob],
    *,
    energyplus: EnergyPlusConfig | None = None,
    max_workers: int | None = None,
    cache: SimulationCache | None = None,
    progress: Callable[..., None] | None = None,
    fs: FileSystem | None = None,
    on_progress: Callable[[SimulationProgress], None] | None = None,
) -> BatchResult:
    """Run multiple EnergyPlus simulations in parallel.

    Uses [ThreadPoolExecutor][concurrent.futures.ThreadPoolExecutor] to dispatch
    simulations concurrently.  Individual job failures are captured as
    failed [SimulationResult][idfkit.simulation.result.SimulationResult] entries -- the batch never raises
    due to a single job failing.

    Args:
        jobs: Sequence of simulation jobs to execute.
        energyplus: Shared EnergyPlus configuration (auto-discovered if
            ``None``).
        max_workers: Maximum number of concurrent simulations.  Defaults
            to ``min(len(jobs), os.cpu_count() or 1)``.
        cache: Optional simulation cache for content-hash lookups.
        progress: Optional callback invoked after each job completes.
            Called as ``progress(completed=N, total=M, label=label,
            success=bool)``.
        fs: Optional file system backend passed through to each
            [simulate][idfkit.simulation.runner.simulate] call.
        on_progress: Optional callback invoked with
            [SimulationProgress][idfkit.simulation.progress.SimulationProgress] events
            during each individual simulation.  Events include
            ``job_index`` and ``job_label`` to identify which batch job
            they belong to.  The ``"tqdm"`` shorthand is not supported
            for batch runners; use
            [tqdm_progress][idfkit.simulation.progress_bars.tqdm_progress]
            with a custom per-job callback instead.

    Returns:
        A [BatchResult][idfkit.simulation.batch.BatchResult] with results in the same order as *jobs*.

    Raises:
        ValueError: If *jobs* is empty.
    """
    if not jobs:
        msg = "jobs must not be empty"
        raise ValueError(msg)

    if on_progress == "tqdm":
        msg = (
            'on_progress="tqdm" is not supported for batch simulations because a single '
            "progress bar cannot represent multiple concurrent jobs. Use the tqdm_progress() "
            "context manager with a custom callback instead."
        )
        raise ValueError(msg)

    progress_cb, progress_cleanup = resolve_on_progress(on_progress)

    try:
        if max_workers is None:
            max_workers = min(len(jobs), os.cpu_count() or 1)

        logger.info("Starting batch of %d jobs with %d workers", len(jobs), max_workers)

        results: list[SimulationResult | None] = [None] * len(jobs)
        completed_count = 0
        total = len(jobs)

        start = time.monotonic()

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_index = {
                executor.submit(_run_job, idx, job, energyplus, cache, fs, progress_cb): idx
                for idx, job in enumerate(jobs)
            }

            for future in as_completed(future_to_index):
                idx = future_to_index[future]
                job = jobs[idx]
                result = future.result()  # _run_job never raises
                results[idx] = result
                completed_count += 1

                if progress is not None:
                    progress(
                        completed=completed_count,
                        total=total,
                        label=job.label,
                        success=result.success,
                    )
    finally:
        if progress_cleanup is not None:
            progress_cleanup()

    elapsed = time.monotonic() - start

    # All slots filled — assert for type checker
    final: list[SimulationResult] = []
    for r in results:
        assert r is not None  # noqa: S101
        final.append(r)

    batch_result = BatchResult(results=tuple(final), total_runtime_seconds=elapsed)
    logger.info(
        "Batch complete: %d succeeded, %d failed in %.1fs",
        len(batch_result.succeeded),
        len(batch_result.failed),
        elapsed,
    )
    return batch_result

SimulationJob

idfkit.simulation.batch.SimulationJob dataclass

Specification for a single simulation within a batch.

Attributes:

Name	Type	Description
model	object	The EnergyPlus model to simulate.
weather	str | Path	Path to the weather file.
label	str	Human-readable label for progress reporting.
output_dir	str | Path | None	Directory for output files (default: auto temp dir).
expand_objects	bool	Run ExpandObjects before simulation.
annual	bool	Run annual simulation.
design_day	bool	Run design-day-only simulation.
output_prefix	str	Prefix for output files.
output_suffix	Literal['C', 'L', 'D']	Output file naming suffix ("C", "L", or "D").
readvars	bool	Run ReadVarsESO after simulation.
timeout	float	Maximum runtime in seconds for the main EnergyPlus subprocess.
preprocessor_timeout	float | None	Per-subprocess timeout (seconds) applied individually to ExpandObjects, Slab, and Basement when they run automatically. None consults the IDFKIT_PREPROCESSOR_TIMEOUT environment variable, falling back to 120 s.
extra_args	tuple[str, ...] | None	Additional command-line arguments.

Source code in idfkit/simulation/batch.py

@dataclass(frozen=True, slots=True)
class SimulationJob:
    """Specification for a single simulation within a batch.

    Attributes:
        model: The EnergyPlus model to simulate.
        weather: Path to the weather file.
        label: Human-readable label for progress reporting.
        output_dir: Directory for output files (default: auto temp dir).
        expand_objects: Run ExpandObjects before simulation.
        annual: Run annual simulation.
        design_day: Run design-day-only simulation.
        output_prefix: Prefix for output files.
        output_suffix: Output file naming suffix (``"C"``, ``"L"``, or ``"D"``).
        readvars: Run ReadVarsESO after simulation.
        timeout: Maximum runtime in seconds for the main EnergyPlus
            subprocess.
        preprocessor_timeout: Per-subprocess timeout (seconds) applied
            individually to ExpandObjects, Slab, and Basement when they
            run automatically.  ``None`` consults the
            ``IDFKIT_PREPROCESSOR_TIMEOUT`` environment variable, falling
            back to 120 s.
        extra_args: Additional command-line arguments.
    """

    model: object  # IDFDocument — use object to avoid import at class level
    weather: str | Path
    label: str = ""
    output_dir: str | Path | None = None
    expand_objects: bool = True
    annual: bool = False
    design_day: bool = False
    output_prefix: str = "eplus"
    output_suffix: Literal["C", "L", "D"] = "C"
    readvars: bool = False
    timeout: float = 3600.0
    preprocessor_timeout: float | None = None
    extra_args: tuple[str, ...] | None = None

annual = False class-attribute instance-attribute

design_day = False class-attribute instance-attribute

expand_objects = True class-attribute instance-attribute

extra_args = None class-attribute instance-attribute

label = '' class-attribute instance-attribute

model instance-attribute

output_dir = None class-attribute instance-attribute

output_prefix = 'eplus' class-attribute instance-attribute

output_suffix = 'C' class-attribute instance-attribute

preprocessor_timeout = None class-attribute instance-attribute

readvars = False class-attribute instance-attribute

timeout = 3600.0 class-attribute instance-attribute

weather instance-attribute

BatchResult

idfkit.simulation.batch.BatchResult dataclass

Bases: Sequence[SimulationResult]

Aggregated results from a batch simulation run.

A real Sequence of the results, in job order, so for result in batch, enumerate(batch), batch[0], len(batch), reversed(batch) and result in batch all work and all type-check.

It behaved that way at runtime long before it type-checked. __len__ and __getitem__ alone make an object iterable through the legacy sequence protocol, so for result in batch ran fine and every static checker rejected it, which pushed seven documented examples into writing batch.results instead. Documenting a workaround for a call that already worked is worse than either fixing it or removing it, so the base class is declared.

Attributes:

Name	Type	Description
results	tuple[SimulationResult, ...]	Simulation results in the same order as the input jobs.
total_runtime_seconds	float	Wall-clock time for the entire batch.

Source code in idfkit/simulation/batch.py

@dataclass(frozen=True, slots=True)
class BatchResult(Sequence[SimulationResult]):
    """Aggregated results from a batch simulation run.

    A real [Sequence][collections.abc.Sequence] of the results, in job order, so
    ``for result in batch``, ``enumerate(batch)``, ``batch[0]``, ``len(batch)``,
    ``reversed(batch)`` and ``result in batch`` all work and all type-check.

    It behaved that way at runtime long before it type-checked. ``__len__`` and
    ``__getitem__`` alone make an object iterable through the legacy sequence
    protocol, so ``for result in batch`` ran fine and every static checker rejected
    it, which pushed seven documented examples into writing ``batch.results``
    instead. Documenting a workaround for a call that already worked is worse than
    either fixing it or removing it, so the base class is declared.

    Attributes:
        results: Simulation results in the same order as the input jobs.
        total_runtime_seconds: Wall-clock time for the entire batch.
    """

    results: tuple[SimulationResult, ...]
    total_runtime_seconds: float

    @property
    def succeeded(self) -> tuple[SimulationResult, ...]:
        """Results that completed successfully."""
        return tuple(r for r in self.results if r.success)

    @property
    def failed(self) -> tuple[SimulationResult, ...]:
        """Results that failed."""
        return tuple(r for r in self.results if not r.success)

    @property
    def all_succeeded(self) -> bool:
        """Whether every job in the batch succeeded."""
        return all(r.success for r in self.results)

    def __len__(self) -> int:
        return len(self.results)

    @overload
    def __getitem__(self, index: int) -> SimulationResult: ...

    @overload
    def __getitem__(self, index: slice) -> tuple[SimulationResult, ...]: ...

    def __getitem__(self, index: int | slice) -> SimulationResult | tuple[SimulationResult, ...]:
        return self.results[index]

    def __iter__(self) -> Iterator[SimulationResult]:
        # Sequence would synthesise this from __getitem__, one index at a time. Iterating the
        # underlying tuple directly is the same answer without the index arithmetic, and it is
        # the path every `for result in batch` actually takes.
        return iter(self.results)

all_succeeded property

Whether every job in the batch succeeded.

failed property

Results that failed.

results instance-attribute

succeeded property

Results that completed successfully.

total_runtime_seconds instance-attribute