Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / tools / model_integration / schemas.py: 80%
82 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"""Versioned machine-readable schemas for model-integration diagnostics."""
17from __future__ import annotations
19from dataclasses import asdict, dataclass, field
20from enum import Enum
21from typing import Any, Iterable, Mapping, Optional
24SCHEMA_VERSION = 2
27def tolerance_value(
28 tolerance: Mapping[str, Any],
29 name: str,
30 fallback: str | None = None,
31) -> float:
32 """Resolve an optional tolerance, treating YAML null as unspecified."""
33 value = tolerance.get(name)
34 if value is None and fallback is not None:
35 value = tolerance.get(fallback)
36 return 0.0 if value is None else float(value)
39class FindingSeverity(str, Enum):
40 """Stable severity levels consumed by CLI and Agent workflows."""
42 INFO = "INFO"
43 WARNING = "WARNING"
44 ERROR = "ERROR"
47class IntegrationState(str, Enum):
48 """Evidence-backed validation gates, excluding derived/reporting states."""
50 DISCOVERED = "DISCOVERED"
51 STRUCTURE_VALIDATED = "STRUCTURE_VALIDATED"
52 MODULE_PARITY_PASSED = "MODULE_PARITY_PASSED"
53 MATRIX_PASSED = "MATRIX_PASSED"
54 FAILED = "FAILED"
55 BLOCKED = "BLOCKED"
58@dataclass(frozen=True)
59class ModelIntegrationFinding:
60 """One structured diagnostic with facts and an actionable remediation."""
62 phase: str
63 owner_fqn: str
64 severity: FindingSeverity
65 message: str
66 facts: dict[str, Any] = field(default_factory=dict)
67 why_unsafe: str = ""
68 remediation: str = ""
69 related_config: tuple[str, ...] = ()
71 def __post_init__(self) -> None:
72 if not self.phase or not self.message:
73 raise ValueError("model integration finding phase and message must be non-empty")
74 if self.severity is FindingSeverity.ERROR and (
75 not self.why_unsafe or not self.remediation
76 ):
77 raise ValueError("ERROR findings require why_unsafe and remediation")
79 def to_dict(self) -> dict[str, Any]:
80 """Return a JSON-serializable representation."""
81 payload = asdict(self)
82 payload["severity"] = self.severity.value
83 return payload
85 def format_human(self) -> str:
86 """Format a stable human-readable diagnostic block."""
87 owner = f"\nowner: {self.owner_fqn}" if self.owner_fqn else ""
88 facts = ""
89 if self.facts:
90 facts = "\nfacts:" + "".join(
91 f"\n {name}: {value}" for name, value in sorted(self.facts.items())
92 )
93 unsafe = f"\nwhy unsafe: {self.why_unsafe}" if self.why_unsafe else ""
94 remediation = f"\nfix: {self.remediation}" if self.remediation else ""
95 return (
96 f"{self.severity.value} phase={self.phase}: {self.message}"
97 f"{owner}{facts}{unsafe}{remediation}"
98 )
101class ModelIntegrationValidationError(RuntimeError):
102 """Raised when a model integration report contains one or more ERROR findings."""
104 def __init__(self, findings: Iterable[ModelIntegrationFinding]) -> None:
105 """Build one exception from ordered structured findings."""
106 self.findings = tuple(findings)
107 super().__init__("\n\n".join(finding.format_human() for finding in self.findings))
110@dataclass
111class ModelIntegrationReport:
112 """Aggregate findings for one model/build/run while preserving schema version."""
114 findings: list[ModelIntegrationFinding] = field(default_factory=list)
115 metadata: dict[str, Any] = field(default_factory=dict)
116 schema_version: int = SCHEMA_VERSION
118 @property
119 def errors(self) -> tuple[ModelIntegrationFinding, ...]:
120 """Return ERROR findings."""
121 return tuple(
122 finding for finding in self.findings
123 if finding.severity is FindingSeverity.ERROR
124 )
126 def add(self, finding: ModelIntegrationFinding) -> None:
127 """Append a finding."""
128 self.findings.append(finding)
130 def extend(self, findings: Iterable[ModelIntegrationFinding]) -> None:
131 """Append findings from another validator."""
132 self.findings.extend(findings)
134 def raise_for_errors(self) -> None:
135 """Raise a structured exception when any ERROR is present."""
136 if self.errors:
137 raise ModelIntegrationValidationError(self.errors)
139 def to_dict(self) -> dict[str, Any]:
140 """Return the stable JSON payload."""
141 return {
142 "schema_version": self.schema_version,
143 "status": "FAIL" if self.errors else "PASS",
144 "metadata": self.metadata,
145 "findings": [finding.to_dict() for finding in self.findings],
146 }
149@dataclass(frozen=True)
150class ComparisonMetric:
151 """Numerical comparison result for one tensor observation."""
153 name: str
154 status: str
155 max_abs: Optional[float]
156 relative_l2: Optional[float]
157 finite: bool
158 reference_shape: tuple[int, ...]
159 candidate_shape: tuple[int, ...]
160 reason: str = ""
162 def to_dict(self) -> dict[str, Any]:
163 """Return a JSON-serializable representation."""
164 return asdict(self)
167__all__ = [
168 "ComparisonMetric",
169 "ModelIntegrationReport",
170 "FindingSeverity",
171 "IntegrationState",
172 "ModelIntegrationFinding",
173 "ModelIntegrationValidationError",
174 "SCHEMA_VERSION",
175 "tolerance_value",
176]