sweep.py 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159
  1. """Sweep orchestrator: runs scenarios N times across multiple backends."""
  2. from __future__ import annotations
  3. import glob as glob_mod
  4. import json
  5. import shutil
  6. import time
  7. from dataclasses import asdict, dataclass, field
  8. from datetime import datetime
  9. from pathlib import Path
  10. from typing import Any
  11. import yaml
  12. from drill.engine import Engine, RunResult
  13. from drill.verifier import Verdict
  14. @dataclass
  15. class RunStatus:
  16. index: int
  17. status: str # "pass", "fail", "error"
  18. duration: float
  19. error: str | None = None
  20. @dataclass
  21. class RunGroup:
  22. scenario: str
  23. backend: str
  24. n: int
  25. timestamp: str
  26. sweep_id: str
  27. runs: list[RunStatus] = field(default_factory=list)
  28. partial: bool = False
  29. def write_run_group(group: RunGroup, output_dir: Path) -> None:
  30. output_dir.mkdir(parents=True, exist_ok=True)
  31. data: dict[str, Any] = {
  32. "scenario": group.scenario,
  33. "backend": group.backend,
  34. "n": group.n,
  35. "timestamp": group.timestamp,
  36. "sweep_id": group.sweep_id,
  37. "partial": group.partial,
  38. "runs": [
  39. {k: v for k, v in asdict(r).items() if k != "error" or v is not None}
  40. for r in group.runs
  41. ],
  42. }
  43. (output_dir / "run-group.json").write_text(json.dumps(data, indent=2))
  44. class Sweep:
  45. def __init__(
  46. self,
  47. scenario_path: Path,
  48. backend_names: list[str],
  49. backends_dir: Path,
  50. fixtures_dir: Path,
  51. results_dir: Path,
  52. n: int,
  53. sweep_id: str,
  54. ) -> None:
  55. self.scenario_path = scenario_path
  56. self.backend_names = backend_names
  57. self.backends_dir = backends_dir
  58. self.fixtures_dir = fixtures_dir
  59. self.results_dir = results_dir
  60. self.n = n
  61. self.sweep_id = sweep_id
  62. self._scenario_name_cache: str | None = None
  63. def validate_backends(self) -> None:
  64. for name in self.backend_names:
  65. path = self.backends_dir / f"{name}.yaml"
  66. if not path.exists():
  67. raise FileNotFoundError(f"Backend config not found: {path}")
  68. def run_all(self) -> list[RunGroup]:
  69. self.validate_backends()
  70. groups: list[RunGroup] = []
  71. for backend_name in self.backend_names:
  72. group = self._run_backend(backend_name)
  73. groups.append(group)
  74. return groups
  75. def _run_backend(self, backend_name: str) -> RunGroup:
  76. timestamp = datetime.now().strftime("%Y-%m-%dT%H-%M-%S")
  77. group_dir = (
  78. self.results_dir / self.scenario_name / backend_name / f"{timestamp}-{self.sweep_id}"
  79. )
  80. group_dir.mkdir(parents=True, exist_ok=True)
  81. group = RunGroup(
  82. scenario=self.scenario_name,
  83. backend=backend_name,
  84. n=self.n,
  85. timestamp=timestamp,
  86. sweep_id=self.sweep_id,
  87. )
  88. try:
  89. for i in range(self.n):
  90. run_status = self._run_single(backend_name, group_dir, i, timestamp)
  91. group.runs.append(run_status)
  92. except KeyboardInterrupt:
  93. group.partial = True
  94. finally:
  95. write_run_group(group, group_dir)
  96. return group
  97. def _run_single(
  98. self, backend_name: str, group_dir: Path, index: int, timestamp: str
  99. ) -> RunStatus:
  100. run_suffix = f"-run-{index:02d}"
  101. run_dir = group_dir / f"run-{index:02d}"
  102. start = time.time()
  103. try:
  104. engine = Engine(
  105. scenario_path=self.scenario_path,
  106. backend_name=backend_name,
  107. backends_dir=self.backends_dir,
  108. fixtures_dir=self.fixtures_dir,
  109. results_dir=self.results_dir,
  110. )
  111. result: RunResult = engine.run(output_dir=run_dir, run_suffix=run_suffix)
  112. verdict = Verdict.model_validate_json(result.verdict_json)
  113. duration = time.time() - start
  114. status = "pass" if verdict.passed else "fail"
  115. return RunStatus(index=index, status=status, duration=round(duration, 1))
  116. except KeyboardInterrupt:
  117. raise
  118. except Exception as e:
  119. duration = time.time() - start
  120. return RunStatus(
  121. index=index,
  122. status="error",
  123. duration=round(duration, 1),
  124. error=str(e),
  125. )
  126. finally:
  127. pattern = f"/tmp/drill-*-{timestamp}{run_suffix}"
  128. for d in glob_mod.glob(pattern):
  129. p = Path(d)
  130. if p.is_dir():
  131. shutil.rmtree(p, ignore_errors=True)
  132. @property
  133. def scenario_name(self) -> str:
  134. if self._scenario_name_cache is None:
  135. with open(self.scenario_path) as f:
  136. data = yaml.safe_load(f)
  137. self._scenario_name_cache = data["scenario"]
  138. return self._scenario_name_cache