Cache API

Content-addressed simulation result caching.

SimulationCache

idfkit.simulation.cache.SimulationCache

Content-addressed simulation result cache.

Each entry is a directory named by the cache key containing a full copy of the simulation run directory plus a _cache_meta.json manifest.

Source code in src/idfkit/simulation/cache.py

class SimulationCache:
    """Content-addressed simulation result cache.

    Each entry is a directory named by the cache key containing a full copy
    of the simulation run directory plus a ``_cache_meta.json`` manifest.
    """

    __slots__ = ("_cache_dir",)

    _META_FILE = "_cache_meta.json"

    def __init__(self, cache_dir: str | Path | None = None) -> None:
        self._cache_dir = Path(cache_dir) if cache_dir is not None else default_simulation_cache_dir()

    @property
    def cache_dir(self) -> Path:
        """Root directory for cached simulation entries."""
        return self._cache_dir

    def compute_key(
        self,
        model: IDFDocument,
        weather: str | Path,
        *,
        expand_objects: bool = True,
        annual: bool = False,
        design_day: bool = False,
        output_suffix: Literal["C", "L", "D"] = "C",
        readvars: bool = False,
        extra_args: list[str] | tuple[str, ...] | None = None,
    ) -> CacheKey:
        """Compute a deterministic cache key for a simulation invocation.

        The model is copied and normalised (``Output:SQLite`` is ensured) so
        that models differing only in the presence of that object produce the
        same key.

        Args:
            model: The EnergyPlus model.
            weather: Path to the weather file.
            expand_objects: Whether ExpandObjects will run.
            annual: Whether annual simulation is used.
            design_day: Whether design-day-only simulation is used.
            output_suffix: Output file naming suffix (``"C"``, ``"L"``, or ``"D"``).
            readvars: Whether ReadVarsESO post-processing will run.
            extra_args: Additional command-line arguments.

        Returns:
            A [CacheKey][idfkit.simulation.cache.CacheKey] for use with [get][idfkit.simulation.cache.SimulationCache.get] / [put][idfkit.simulation.cache.SimulationCache.put].
        """
        from ..writers import write_idf

        normalised = model.copy()
        if "Output:SQLite" not in normalised:
            normalised.add("Output:SQLite", "", data={"option_type": "SimpleAndTabular"})
        idf_text: str = write_idf(normalised) or ""

        weather_path = Path(weather).resolve()
        weather_bytes = weather_path.read_bytes()

        flags = json.dumps(
            {
                "expand_objects": expand_objects,
                "annual": annual,
                "design_day": design_day,
                "output_suffix": output_suffix,
                "readvars": readvars,
                "extra_args": list(extra_args) if extra_args else [],
            },
            sort_keys=True,
        )

        h = hashlib.sha256()
        h.update(idf_text.encode("utf-8"))
        h.update(weather_bytes)
        h.update(flags.encode("utf-8"))
        key = CacheKey(hex_digest=h.hexdigest())
        logger.debug("Computed cache key %s", key.hex_digest[:12])
        return key

    def get(self, key: CacheKey) -> SimulationResult | None:
        """Retrieve a cached simulation result.

        Args:
            key: Cache key from [compute_key][idfkit.simulation.cache.SimulationCache.compute_key].

        Returns:
            A [SimulationResult][idfkit.simulation.result.SimulationResult] if a cache hit exists, otherwise
            ``None``.
        """
        entry_dir = self._cache_dir / key.hex_digest
        meta_path = entry_dir / self._META_FILE
        if not meta_path.is_file():
            logger.debug("Cache miss for %s", key.hex_digest[:12])
            return None

        try:
            meta = json.loads(meta_path.read_text(encoding="utf-8"))

            from .result import SimulationResult

            logger.debug("Cache hit for %s", key.hex_digest[:12])
            return SimulationResult(
                run_dir=entry_dir,
                success=meta["success"],
                exit_code=meta["exit_code"],
                stdout="",
                stderr="",
                runtime_seconds=meta["runtime_seconds"],
                output_prefix=meta["output_prefix"],
            )
        except (json.JSONDecodeError, KeyError, OSError):
            # Corrupted or incomplete cache entry — remove it so that
            # a subsequent put() can write a fresh copy, then treat as miss.
            logger.debug("Removing corrupted cache entry %s", key.hex_digest[:12])
            shutil.rmtree(entry_dir, ignore_errors=True)
            return None

    def put(self, key: CacheKey, result: SimulationResult) -> None:
        """Store a successful simulation result in the cache.

        Only results with ``success=True`` are cached.  The entire run
        directory is copied into the cache atomically.

        Args:
            key: Cache key from [compute_key][idfkit.simulation.cache.SimulationCache.compute_key].
            result: Successful simulation result to cache.
        """
        if not result.success:
            return

        target_dir = self._cache_dir / key.hex_digest
        if target_dir.is_dir():
            return  # already cached

        self._cache_dir.mkdir(parents=True, exist_ok=True)

        tmp_dir = Path(tempfile.mkdtemp(dir=self._cache_dir, prefix=".tmp_"))
        try:
            # Copy output files (ignore_dangling_symlinks for robustness)
            shutil.copytree(result.run_dir, tmp_dir, dirs_exist_ok=True)

            # Write metadata
            meta = {
                "success": result.success,
                "exit_code": result.exit_code,
                "runtime_seconds": result.runtime_seconds,
                "output_prefix": result.output_prefix,
            }
            meta_path = tmp_dir / self._META_FILE
            meta_path.write_text(json.dumps(meta), encoding="utf-8")

            # Atomic rename — os.rename fails if target_dir already exists
            # (another process beat us), unlike shutil.move which would nest
            # tmp_dir inside the existing target as a subdirectory.
            os.rename(str(tmp_dir), str(target_dir))
            logger.debug("Cached result for %s", key.hex_digest[:12])
        except OSError:
            # Another thread/process beat us, or a real filesystem error
            # — clean up the temporary directory.
            shutil.rmtree(tmp_dir, ignore_errors=True)

    def contains(self, key: CacheKey) -> bool:
        """Check whether a cache entry exists for *key*."""
        return (self._cache_dir / key.hex_digest / self._META_FILE).is_file()

    def clear(self) -> None:
        """Remove all cached entries."""
        if self._cache_dir.is_dir():
            shutil.rmtree(self._cache_dir)
            logger.debug("Cleared simulation cache at %s", self._cache_dir)

cache_dir property

Root directory for cached simulation entries.

compute_key(model, weather, *, expand_objects=True, annual=False, design_day=False, output_suffix='C', readvars=False, extra_args=None)

Compute a deterministic cache key for a simulation invocation.

The model is copied and normalised (Output:SQLite is ensured) so that models differing only in the presence of that object produce the same key.

Parameters:

Name	Type	Description	Default
model	IDFDocument	The EnergyPlus model.	required
weather	str | Path	Path to the weather file.	required
expand_objects	bool	Whether ExpandObjects will run.	True
annual	bool	Whether annual simulation is used.	False
design_day	bool	Whether design-day-only simulation is used.	False
output_suffix	Literal['C', 'L', 'D']	Output file naming suffix ("C", "L", or "D").	'C'
readvars	bool	Whether ReadVarsESO post-processing will run.	False
extra_args	list[str] | tuple[str, ...] | None	Additional command-line arguments.	None

Returns:

Type	Description
CacheKey	A CacheKey for use with get / put.

Source code in src/idfkit/simulation/cache.py

def compute_key(
    self,
    model: IDFDocument,
    weather: str | Path,
    *,
    expand_objects: bool = True,
    annual: bool = False,
    design_day: bool = False,
    output_suffix: Literal["C", "L", "D"] = "C",
    readvars: bool = False,
    extra_args: list[str] | tuple[str, ...] | None = None,
) -> CacheKey:
    """Compute a deterministic cache key for a simulation invocation.

    The model is copied and normalised (``Output:SQLite`` is ensured) so
    that models differing only in the presence of that object produce the
    same key.

    Args:
        model: The EnergyPlus model.
        weather: Path to the weather file.
        expand_objects: Whether ExpandObjects will run.
        annual: Whether annual simulation is used.
        design_day: Whether design-day-only simulation is used.
        output_suffix: Output file naming suffix (``"C"``, ``"L"``, or ``"D"``).
        readvars: Whether ReadVarsESO post-processing will run.
        extra_args: Additional command-line arguments.

    Returns:
        A [CacheKey][idfkit.simulation.cache.CacheKey] for use with [get][idfkit.simulation.cache.SimulationCache.get] / [put][idfkit.simulation.cache.SimulationCache.put].
    """
    from ..writers import write_idf

    normalised = model.copy()
    if "Output:SQLite" not in normalised:
        normalised.add("Output:SQLite", "", data={"option_type": "SimpleAndTabular"})
    idf_text: str = write_idf(normalised) or ""

    weather_path = Path(weather).resolve()
    weather_bytes = weather_path.read_bytes()

    flags = json.dumps(
        {
            "expand_objects": expand_objects,
            "annual": annual,
            "design_day": design_day,
            "output_suffix": output_suffix,
            "readvars": readvars,
            "extra_args": list(extra_args) if extra_args else [],
        },
        sort_keys=True,
    )

    h = hashlib.sha256()
    h.update(idf_text.encode("utf-8"))
    h.update(weather_bytes)
    h.update(flags.encode("utf-8"))
    key = CacheKey(hex_digest=h.hexdigest())
    logger.debug("Computed cache key %s", key.hex_digest[:12])
    return key

get(key)

Retrieve a cached simulation result.

Parameters:

Name	Type	Description	Default
key	CacheKey	Cache key from compute_key.	required

Returns:

Type	Description
SimulationResult | None	A SimulationResult if a cache hit exists, otherwise
SimulationResult | None	None.

Source code in src/idfkit/simulation/cache.py

def get(self, key: CacheKey) -> SimulationResult | None:
    """Retrieve a cached simulation result.

    Args:
        key: Cache key from [compute_key][idfkit.simulation.cache.SimulationCache.compute_key].

    Returns:
        A [SimulationResult][idfkit.simulation.result.SimulationResult] if a cache hit exists, otherwise
        ``None``.
    """
    entry_dir = self._cache_dir / key.hex_digest
    meta_path = entry_dir / self._META_FILE
    if not meta_path.is_file():
        logger.debug("Cache miss for %s", key.hex_digest[:12])
        return None

    try:
        meta = json.loads(meta_path.read_text(encoding="utf-8"))

        from .result import SimulationResult

        logger.debug("Cache hit for %s", key.hex_digest[:12])
        return SimulationResult(
            run_dir=entry_dir,
            success=meta["success"],
            exit_code=meta["exit_code"],
            stdout="",
            stderr="",
            runtime_seconds=meta["runtime_seconds"],
            output_prefix=meta["output_prefix"],
        )
    except (json.JSONDecodeError, KeyError, OSError):
        # Corrupted or incomplete cache entry — remove it so that
        # a subsequent put() can write a fresh copy, then treat as miss.
        logger.debug("Removing corrupted cache entry %s", key.hex_digest[:12])
        shutil.rmtree(entry_dir, ignore_errors=True)
        return None

put(key, result)

Store a successful simulation result in the cache.

Only results with success=True are cached. The entire run directory is copied into the cache atomically.

Parameters:

Name	Type	Description	Default
key	CacheKey	Cache key from compute_key.	required
result	SimulationResult	Successful simulation result to cache.	required

Source code in src/idfkit/simulation/cache.py

def put(self, key: CacheKey, result: SimulationResult) -> None:
    """Store a successful simulation result in the cache.

    Only results with ``success=True`` are cached.  The entire run
    directory is copied into the cache atomically.

    Args:
        key: Cache key from [compute_key][idfkit.simulation.cache.SimulationCache.compute_key].
        result: Successful simulation result to cache.
    """
    if not result.success:
        return

    target_dir = self._cache_dir / key.hex_digest
    if target_dir.is_dir():
        return  # already cached

    self._cache_dir.mkdir(parents=True, exist_ok=True)

    tmp_dir = Path(tempfile.mkdtemp(dir=self._cache_dir, prefix=".tmp_"))
    try:
        # Copy output files (ignore_dangling_symlinks for robustness)
        shutil.copytree(result.run_dir, tmp_dir, dirs_exist_ok=True)

        # Write metadata
        meta = {
            "success": result.success,
            "exit_code": result.exit_code,
            "runtime_seconds": result.runtime_seconds,
            "output_prefix": result.output_prefix,
        }
        meta_path = tmp_dir / self._META_FILE
        meta_path.write_text(json.dumps(meta), encoding="utf-8")

        # Atomic rename — os.rename fails if target_dir already exists
        # (another process beat us), unlike shutil.move which would nest
        # tmp_dir inside the existing target as a subdirectory.
        os.rename(str(tmp_dir), str(target_dir))
        logger.debug("Cached result for %s", key.hex_digest[:12])
    except OSError:
        # Another thread/process beat us, or a real filesystem error
        # — clean up the temporary directory.
        shutil.rmtree(tmp_dir, ignore_errors=True)

contains(key)

Check whether a cache entry exists for key.

Source code in src/idfkit/simulation/cache.py

def contains(self, key: CacheKey) -> bool:
    """Check whether a cache entry exists for *key*."""
    return (self._cache_dir / key.hex_digest / self._META_FILE).is_file()

clear()

Remove all cached entries.

Source code in src/idfkit/simulation/cache.py

def clear(self) -> None:
    """Remove all cached entries."""
    if self._cache_dir.is_dir():
        shutil.rmtree(self._cache_dir)
        logger.debug("Cleared simulation cache at %s", self._cache_dir)

CacheKey

idfkit.simulation.cache.CacheKey dataclass

Opaque cache key wrapping a hex digest string.

Source code in src/idfkit/simulation/cache.py

@dataclass(frozen=True, slots=True)
class CacheKey:
    """Opaque cache key wrapping a hex digest string."""

    hex_digest: str

hex_digest instance-attribute

default_simulation_cache_dir

idfkit.simulation.cache.default_simulation_cache_dir()

Return the platform-appropriate cache directory for simulation results.

Source code in src/idfkit/simulation/cache.py

def default_simulation_cache_dir() -> Path:
    """Return the platform-appropriate cache directory for simulation results."""
    if sys.platform == "win32":
        base = Path(os.environ.get("LOCALAPPDATA", Path.home() / "AppData" / "Local"))
        return base / "idfkit" / "cache" / "simulation"
    if sys.platform == "darwin":
        return Path.home() / "Library" / "Caches" / "idfkit" / "simulation"
    # Linux / other POSIX
    xdg = os.environ.get("XDG_CACHE_HOME")
    base = Path(xdg) if xdg else Path.home() / ".cache"
    return base / "idfkit" / "simulation"