# file_waiter.py
from __future__ import annotations
"""
Utility for blocking execution until a set of files appears on disk.
Typical use
-----------
>>> from pathlib import Path
>>> from file_waiter import wait_for_files
>>>
>>> wait_for_files(
... run_dir=Path("/scratch/sim/run001"),
... required_files=["xgc.mesh.bp", "xgc.equil.bp", "input"],
... poll_interval=2.0,
... grace_period=300, # give up after 5 min (optional)
... )
"""
import time
from pathlib import Path
from typing import Sequence
[docs]
class FileWaiter:
"""
Block execution until *all* files in *required_files* exist inside *run_dir*.
Parameters
----------
required_files
Relative paths (strings or :class:`pathlib.Path`) that must appear
inside *run_dir*.
run_dir
Directory expected to contain the files.
poll_interval
Delay **in seconds** between successive checks (default ``5.0``).
grace_period
Maximum time **in seconds** to wait before raising :class:`TimeoutError`.
``None`` (default) means “wait forever”.
Notes
-----
*Prints* a short status line on every poll. Switch to the :meth:`missing_files`
method if you want to plug this into a custom logging setup.
"""
def __init__(
self,
required_files: Sequence[str | Path],
run_dir: str | Path,
poll_interval: float = 5.0,
grace_period: float = 3.0,
timeout: float | None = 600.0) -> None:
self.run_dir = Path(run_dir).expanduser().resolve()
self.required_files: tuple[Path, ...] = tuple(
Path(p) if isinstance(p, Path) else Path(p) for p in required_files
)
if poll_interval <= 0:
raise ValueError("poll_interval must be positive")
self.poll_interval = float(poll_interval)
if grace_period <= 0:
raise ValueError("grace_period must be positive")
self.grace_period = float(grace_period)
if timeout is not None and timeout <= 0:
raise ValueError("timeout must be positive or None")
self.timeout = None if timeout is None else float(timeout)
# --------------------------------------------------------------------- helpers
def __repr__(self) -> str: # pragma: no cover
files = ", ".join(str(p) for p in self.required_files)
return (
f"{self.__class__.__name__}(required_files=[{files}], "
f"run_dir='{self.run_dir}', poll_interval={self.poll_interval}, "
f"grace_period={self.grace_period}, timeout={self.timeout})"
)
[docs]
def missing_files(self) -> list[str]:
"""Return a list of relative paths that are still absent."""
return [
str(f)
for f in self.required_files
if not (self.run_dir / f).exists()
]
# ---------------------------------------------------------------- public API
[docs]
def wait(self) -> None:
"""Block until all files are present (plus *grace_period*) or until *timeout* elapses."""
start_time = time.monotonic()
while True:
missing = self.missing_files()
if not missing:
# All files are available – wait the grace period, then exit.
print("All required files found.")
if self.grace_period:
print(
f"Holding for an additional {self.grace_period}s "
"grace period."
)
time.sleep(self.grace_period)
print("Proceeding.")
return
if self.timeout is not None and (
time.monotonic() - start_time >= self.timeout
):
raise TimeoutError(
f"Timed out after {self.timeout}s. "
f"Still missing: {', '.join(missing)}"
)
print(
f"Waiting for required files: {', '.join(missing)} "
f"(checking again in {self.poll_interval}s)"
)
time.sleep(self.poll_interval)
# ---------------------------------------------------------------- convenience wrapper
[docs]
def wait_for_files(
run_dir: str | Path,
required_files: Sequence[str | Path],
poll_interval: float = 5.0,
grace_period: float = 3.0,
timeout: float | None = 600.0,
) -> None:
"""
Convenience wrapper around :class:`FileWaiter`.
Examples
--------
>>> wait_for_files("/scratch/run42",
... ["xgc.mesh.bp", "input"],
... poll_interval=1.0,
... grace_period=0.0) # proceed immediately once ready
"""
FileWaiter(
required_files=required_files,
run_dir=run_dir,
poll_interval=poll_interval,
grace_period=grace_period,
timeout=timeout,
).wait()