Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / codegen / snapshot.py: 78%

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"""Isolated in-memory observations captured from a native model build.""" 

16 

17from __future__ import annotations 

18 

19from collections.abc import Mapping, Sequence 

20from copy import deepcopy 

21from dataclasses import dataclass, field 

22from enum import Enum 

23from typing import Any, Optional 

24 

25 

26class CodegenStatus(str, Enum): 

27 """Lifecycle state of one codegen collection session.""" 

28 

29 NOT_REQUIRED = "not_required" 

30 COLLECTING = "collecting" 

31 DERIVED = "derived" 

32 GENERATING = "generating" 

33 GENERATED = "generated" 

34 FAILED = "failed" 

35 SKIPPED = "skipped" 

36 

37 

38def copy_mapping(value: Mapping[str, Any], path: str) -> dict[str, Any]: 

39 """Return a deep copy of a string-keyed observation mapping.""" 

40 if not isinstance(value, Mapping): 

41 raise TypeError(f"{path} must be a mapping") 

42 if not all(isinstance(key, str) for key in value): 

43 raise ValueError(f"{path} mapping keys must be strings") 

44 return deepcopy(dict(value)) 

45 

46 

47def _copy_records( 

48 values: Sequence[Mapping[str, Any]], path: str 

49) -> tuple[dict[str, Any], ...]: 

50 """Copy a sequence of observation records.""" 

51 return tuple( 

52 copy_mapping(item, f"{path}[{index}]") 

53 for index, item in enumerate(values) 

54 ) 

55 

56 

57@dataclass(frozen=True) 

58class ReplacementSnapshot: 

59 """Declared replacement rules and their observed native applications.""" 

60 

61 declarations: tuple[dict[str, Any], ...] = () 

62 applications: tuple[dict[str, Any], ...] = () 

63 unmatched: tuple[dict[str, Any], ...] = () 

64 

65 @classmethod 

66 def from_data( 

67 cls, 

68 *, 

69 declarations: Sequence[Mapping[str, Any]] = (), 

70 applications: Sequence[Mapping[str, Any]] = (), 

71 unmatched: Sequence[Mapping[str, Any]] = (), 

72 ) -> "ReplacementSnapshot": 

73 """Copy replacement observations into the session boundary.""" 

74 return cls( 

75 declarations=_copy_records( 

76 declarations, "replacements.declarations" 

77 ), 

78 applications=_copy_records( 

79 applications, "replacements.applications" 

80 ), 

81 unmatched=_copy_records(unmatched, "replacements.unmatched"), 

82 ) 

83 

84 def to_dict(self) -> dict[str, Any]: 

85 """Return an isolated mutable representation for emission.""" 

86 return deepcopy( 

87 { 

88 "declarations": list(self.declarations), 

89 "applications": list(self.applications), 

90 "unmatched": list(self.unmatched), 

91 } 

92 ) 

93 

94 

95@dataclass(frozen=True) 

96class FailureSnapshot: 

97 """Stable failure information without exception or traceback objects.""" 

98 

99 stage: str 

100 reason: str 

101 details: dict[str, Any] = field(default_factory=dict) 

102 

103 def __post_init__(self) -> None: 

104 if not isinstance(self.stage, str) or not self.stage: 

105 raise ValueError("failure stage must be a non-empty string") 

106 if not isinstance(self.reason, str) or not self.reason: 

107 raise ValueError("failure reason must be a non-empty string") 

108 object.__setattr__( 

109 self, 

110 "details", 

111 copy_mapping(self.details, "failure.details"), 

112 ) 

113 

114 @classmethod 

115 def from_data( 

116 cls, 

117 stage: str, 

118 reason: str, 

119 details: Optional[Mapping[str, Any]] = None, 

120 ) -> "FailureSnapshot": 

121 """Build an isolated failure record.""" 

122 if not stage or not reason: 

123 raise ValueError("failure stage and reason must be non-empty") 

124 return cls(stage, reason, details or {}) 

125 

126 def to_dict(self) -> dict[str, Any]: 

127 """Return an isolated mutable representation for emission.""" 

128 return { 

129 "stage": self.stage, 

130 "reason": self.reason, 

131 "details": deepcopy(self.details), 

132 } 

133 

134 

135@dataclass(frozen=True) 

136class CodegenSnapshot: 

137 """Isolated observation of one native model build.""" 

138 

139 status: CodegenStatus 

140 source: dict[str, Any] = field(default_factory=dict) 

141 replacements: ReplacementSnapshot = field(default_factory=ReplacementSnapshot) 

142 plan: Optional[dict[str, Any]] = None 

143 diagnostics: dict[str, Any] = field(default_factory=dict) 

144 reason: Optional[str] = None 

145 failure: Optional[FailureSnapshot] = None 

146 

147 def __post_init__(self) -> None: 

148 if not isinstance(self.status, CodegenStatus): 

149 raise TypeError("snapshot status must be a CodegenStatus") 

150 if not isinstance(self.replacements, ReplacementSnapshot): 

151 raise TypeError("snapshot replacements must be a ReplacementSnapshot") 

152 if self.failure is not None and not isinstance(self.failure, FailureSnapshot): 

153 raise TypeError("snapshot failure must be a FailureSnapshot or None") 

154 object.__setattr__(self, "source", copy_mapping(self.source, "snapshot.source")) 

155 object.__setattr__( 

156 self, 

157 "diagnostics", 

158 copy_mapping(self.diagnostics, "snapshot.diagnostics"), 

159 ) 

160 for name in ("plan",): 

161 value = getattr(self, name) 

162 if value is not None: 

163 object.__setattr__( 

164 self, 

165 name, 

166 copy_mapping(value, f"snapshot.{name}"), 

167 ) 

168 

169 def to_dict(self) -> dict[str, Any]: 

170 """Return isolated mutable containers for the emission boundary.""" 

171 return deepcopy(self._emission_view()) 

172 

173 def _emission_view(self) -> dict[str, Any]: 

174 """Borrow observation data internally; callers must copy before mutation.""" 

175 return { 

176 "status": self.status.value, 

177 "source": self.source, 

178 "replacements": { 

179 "declarations": list(self.replacements.declarations), 

180 "applications": list(self.replacements.applications), 

181 "unmatched": list(self.replacements.unmatched), 

182 }, 

183 "plan": self.plan, 

184 "diagnostics": self.diagnostics, 

185 "reason": self.reason, 

186 "failure": None if self.failure is None else { 

187 "stage": self.failure.stage, 

188 "reason": self.failure.reason, 

189 "details": self.failure.details, 

190 }, 

191 } 

192 

193 @classmethod 

194 def _from_owned_data(cls, **values: Any) -> "CodegenSnapshot": 

195 """Adopt data already isolated by a session without copying it again.""" 

196 snapshot = object.__new__(cls) 

197 for name, value in values.items(): 

198 object.__setattr__(snapshot, name, value) 

199 return snapshot 

200 

201 

202__all__ = [ 

203 "CodegenSnapshot", 

204 "CodegenStatus", 

205 "FailureSnapshot", 

206 "ReplacementSnapshot", 

207 "copy_mapping", 

208]