Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / tools / model_integration / evidence_store.py: 84%
146 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-25 05:45 +0800
1# Copyright 2026 Huawei Technologies Co., Ltd
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ============================================================================
15"""Evidence directory writer and deterministic Markdown report generator."""
17from __future__ import annotations
19import csv
20import json
21import os
22import platform
23import subprocess
24import sys
25import tempfile
26from dataclasses import asdict, is_dataclass
27from enum import Enum
28from pathlib import Path
29from typing import Any, Iterable, Mapping, Sequence
31import yaml
33from hyper_parallel.tools.model_integration.schemas import IntegrationState, SCHEMA_VERSION
36_STATE_ORDER = (
37 IntegrationState.DISCOVERED,
38 IntegrationState.STRUCTURE_VALIDATED,
39 IntegrationState.MODULE_PARITY_PASSED,
40 IntegrationState.MATRIX_PASSED,
41)
44def find_repository_root() -> Path:
45 """Find the checkout root without depending on this module's depth."""
46 for candidate in Path(__file__).resolve().parents:
47 if candidate.joinpath("pyproject.toml").is_file() and candidate.joinpath(
48 "hyper_parallel"
49 ).is_dir():
50 return candidate
51 return Path.cwd().resolve()
54def _json_default(value: Any) -> Any:
55 if is_dataclass(value):
56 return asdict(value)
57 if isinstance(value, Path):
58 return str(value)
59 if isinstance(value, Enum):
60 return value.value
61 return repr(value)
64class EvidenceStore:
65 """Write versioned evidence without parsing free-form Trainer logs."""
67 def __init__(self, root: str | Path) -> None:
68 """Create an evidence writer rooted at one independent run directory."""
69 self.root = Path(root).expanduser().resolve()
70 self.root.mkdir(parents=True, exist_ok=True)
72 def path(self, relative_path: str | Path) -> Path:
73 """Resolve an evidence-relative path and reject directory traversal."""
74 resolved = (self.root / relative_path).resolve()
75 if resolved != self.root and self.root not in resolved.parents:
76 raise ValueError(f"evidence path escapes run directory: {relative_path}")
77 return resolved
79 @staticmethod
80 def _atomic_write(path: Path, content: str) -> None:
81 path.parent.mkdir(parents=True, exist_ok=True)
82 descriptor, temporary_name = tempfile.mkstemp(
83 dir=path.parent,
84 prefix=f".{path.name}.",
85 suffix=".tmp",
86 text=True,
87 )
88 try:
89 with os.fdopen(descriptor, "w", encoding="utf-8") as output_file:
90 output_file.write(content)
91 output_file.flush()
92 os.fsync(output_file.fileno())
93 os.replace(temporary_name, path)
94 except BaseException:
95 try:
96 os.unlink(temporary_name)
97 except FileNotFoundError:
98 pass
99 raise
101 def write_json(self, relative_path: str | Path, payload: Any) -> Path:
102 """Write sorted, UTF-8 JSON atomically."""
103 output_path = self.path(relative_path)
104 content = json.dumps(
105 payload,
106 indent=2,
107 ensure_ascii=False,
108 sort_keys=True,
109 default=_json_default,
110 ) + "\n"
111 self._atomic_write(output_path, content)
112 return output_path
114 def write_yaml(self, relative_path: str | Path, payload: Any) -> Path:
115 """Write a resolved YAML object atomically."""
116 output_path = self.path(relative_path)
117 content = yaml.safe_dump(payload, allow_unicode=True, sort_keys=False)
118 self._atomic_write(output_path, content)
119 return output_path
121 def append_jsonl(self, relative_path: str | Path, payload: Any) -> Path:
122 """Append one structured record to a JSON-lines evidence stream."""
123 output_path = self.path(relative_path)
124 output_path.parent.mkdir(parents=True, exist_ok=True)
125 with output_path.open("a", encoding="utf-8") as output_file:
126 output_file.write(json.dumps(payload, sort_keys=True, default=_json_default) + "\n")
127 output_file.flush()
128 return output_path
130 def write_csv(
131 self,
132 relative_path: str | Path,
133 rows: Iterable[Mapping[str, Any]],
134 fieldnames: Sequence[str],
135 ) -> Path:
136 """Write a stable-column CSV evidence table."""
137 output_path = self.path(relative_path)
138 output_path.parent.mkdir(parents=True, exist_ok=True)
139 with output_path.open("w", encoding="utf-8", newline="") as output_file:
140 writer = csv.DictWriter(output_file, fieldnames=fieldnames)
141 writer.writeheader()
142 for row in rows:
143 writer.writerow({name: row.get(name, "") for name in fieldnames})
144 return output_path
146 def capture_environment(self) -> dict[str, Any]:
147 """Capture imports and revisions that determine reproducibility."""
148 git_revision = None
149 source_root = find_repository_root()
150 try:
151 git_revision = subprocess.run(
152 ["git", "rev-parse", "HEAD"],
153 cwd=source_root,
154 check=True,
155 capture_output=True,
156 text=True,
157 ).stdout.strip()
158 except (OSError, subprocess.CalledProcessError):
159 pass
160 environment = {
161 "python": sys.version,
162 "executable": sys.executable,
163 "platform": platform.platform(),
164 "git_revision": git_revision,
165 "source_root": str(source_root),
166 "sys_path": list(sys.path),
167 }
168 try:
169 import hyper_parallel # pylint: disable=import-outside-toplevel
171 environment["hyper_parallel"] = {
172 "import_path": str(Path(hyper_parallel.__file__).resolve()),
173 }
174 except ImportError:
175 environment["hyper_parallel"] = None
176 try:
177 import torch # pylint: disable=import-outside-toplevel,forbidden-backend-import
179 environment["torch"] = {
180 "version": torch.__version__,
181 "import_path": str(Path(torch.__file__).resolve()),
182 }
183 except ImportError:
184 environment["torch"] = None
185 try:
186 import transformers # pylint: disable=import-outside-toplevel
188 environment["transformers"] = {
189 "version": transformers.__version__,
190 "import_path": str(Path(transformers.__file__).resolve()),
191 }
192 except ImportError:
193 environment["transformers"] = None
194 self.write_json("environment.json", environment)
195 return environment
197 def update_state(
198 self,
199 state: IntegrationState,
200 evidence: Iterable[str] = (),
201 reason: str = "",
202 ) -> Path:
203 """Advance the evidence-backed workflow state without skipping gates."""
204 state_path = self.path("integration_state.json")
205 previous = None
206 if state_path.exists():
207 previous = IntegrationState(json.loads(state_path.read_text(encoding="utf-8"))["state"])
208 terminal = (IntegrationState.FAILED, IntegrationState.BLOCKED)
209 if previous in terminal and state not in terminal:
210 raise ValueError(f"cannot advance terminal integration state {previous.value}")
211 if state not in terminal and previous is not None:
212 previous_index = _STATE_ORDER.index(previous)
213 next_index = _STATE_ORDER.index(state)
214 if next_index < previous_index:
215 raise ValueError(
216 f"cannot regress integration state {previous.value} -> {state.value}"
217 )
218 if next_index > previous_index + 1:
219 raise ValueError(
220 f"cannot skip integration state {previous.value} -> {state.value}"
221 )
222 payload = {
223 "schema_version": SCHEMA_VERSION,
224 "state": state.value,
225 "previous_state": previous.value if previous else None,
226 "evidence": sorted(set(evidence)),
227 "reason": reason,
228 }
229 return self.write_json("integration_state.json", payload)
231 def current_state(self) -> IntegrationState | None:
232 """Return the persisted workflow state, if this run has one."""
233 state_path = self.path("integration_state.json")
234 if not state_path.exists():
235 return None
236 return IntegrationState(json.loads(state_path.read_text(encoding="utf-8"))["state"])
238 def advance_if_before(
239 self,
240 state: IntegrationState,
241 evidence: Iterable[str] = (),
242 reason: str = "",
243 ) -> Path:
244 """Advance one adjacent gate, or preserve an already-later valid gate.
246 This is the restart-safe counterpart to :meth:`update_state`. It never
247 rewrites the persisted state backwards, while ``update_state`` remains
248 strict enough to catch accidental regression in ordinary callers.
249 """
250 current = self.current_state()
251 if current in (IntegrationState.FAILED, IntegrationState.BLOCKED):
252 raise ValueError(f"cannot advance terminal integration state {current.value}")
253 if current is not None and _STATE_ORDER.index(current) >= _STATE_ORDER.index(state):
254 return self.path("integration_state.json")
255 return self.update_state(state, evidence, reason)
257 def render_summary(
258 self,
259 status: str,
260 title: str,
261 sections: Sequence[tuple[str, Sequence[str]]],
262 relative_path: str = "summary.md",
263 ) -> Path:
264 """Write a report whose first line is PASS, FAIL, or BLOCKED.
266 Args:
267 status: Aggregate report status.
268 title: Markdown document title.
269 sections: Ordered headings and bullet entries.
270 relative_path: Evidence-relative report path.
272 Returns:
273 Path to the atomically written report.
274 """
275 if status not in ("PASS", "FAIL", "BLOCKED"):
276 raise ValueError(f"invalid report status: {status!r}")
277 lines = [status, "", f"# {title}"]
278 for heading, entries in sections:
279 lines.extend(("", f"## {heading}", ""))
280 lines.extend(f"- {entry}" for entry in entries)
281 return self._atomic_summary(relative_path, "\n".join(lines) + "\n")
283 def _atomic_summary(self, relative_path: str, content: str) -> Path:
284 output_path = self.path(relative_path)
285 self._atomic_write(output_path, content)
286 return output_path
289__all__ = ["EvidenceStore", "find_repository_root"]