Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/ge/ge/graph/_attr.py: 83%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ------------------------------------------------------------------- 

4# ----------------------------------------------------------------------------------------------------------- 

5# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

6# This program is free software, you can redistribute it and/or modify it under the terms and conditions of 

7# CANN Open Software License Agreement Version 2.0 (the "License"). 

8# Please refer to the License for details. You may not use this file except in compliance with the License. 

9# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, 

10# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. 

11# See LICENSE in the root of the software repository for the full text of the License. 

12# ----------------------------------------------------------------------------------------------------------- 

13 

14"""AttrValue module for attribute value operations in GraphEngine.""" 

15 

16import ctypes 

17from typing import Any, Callable, List, Type 

18 

19from ge._capi.pygraph_wrapper import graph_lib 

20 

21from .types import AttrValueType, DataType 

22 

23 

24def _clear_cache_on_success(method: Callable[..., int]) -> Callable[..., bool]: 

25 """Decorator: clear cache and return True on success (ret == 0), else return False.""" 

26 

27 def _wrapped(self, *args, **kwargs): 

28 ret = method(self, *args, **kwargs) 

29 if ret == 0: 

30 self._clear_cache() 

31 return True 

32 return False 

33 

34 return _wrapped 

35 

36 

37def _cache_with_type_check(expected_type: AttrValueType): 

38 """Decorator factory: create a cache decorator with type checking.""" 

39 

40 def decorator(method: Callable[..., Any]) -> Callable[..., Any]: 

41 def _wrapped(self, *args, **kwargs): 

42 # 检查是否有缓存且类型匹配 

43 if self._cached_value is not None and self._value_type is not None and self._value_type == expected_type: 

44 return self._cached_value 

45 

46 # 没有缓存或类型不匹配，调用原方法获取值 

47 try: 

48 result = method(self, *args, **kwargs) 

49 # 设置缓存和类型 

50 self._cached_value = result 

51 self._value_type = expected_type 

52 return result 

53 except Exception: 

54 # 如果获取失败，不设置缓存 

55 raise 

56 

57 return _wrapped 

58 

59 return decorator 

60 

61 

62def _validate_list_input(values: List[Any], expected_type: Type, type_name: str) -> None: 

63 """Validate list input parameters. 

64 

65 Args: 

66 values: List of values to validate. 

67 expected_type: Expected type for list elements. 

68 type_name: Human-readable type name for error messages. 

69 

70 Raises: 

71 TypeError: If values is not a list or elements are wrong type. 

72 ValueError: If values is empty. 

73 """ 

74 if not isinstance(values, list): 

75 raise TypeError("Values must be a list") 

76 

77 if not values: 

78 raise ValueError("Empty list is not supported") 

79 

80 if not all(isinstance(v, expected_type) for v in values): 

81 raise TypeError(f"All values must be {type_name}") 

82 

83 

84def _create_list_setter(c_type, setter_func, expected_type, type_name): 

85 """Factory function to create list setter methods. 

86 

87 Args: 

88 c_type: C type for the array (e.g., ctypes.c_float). 

89 setter_func: C API function for setting the list. 

90 expected_type: Python type for validation. 

91 type_name: Human-readable type name for error messages. 

92 

93 Returns: 

94 Decorated setter method. 

95 """ 

96 

97 @_clear_cache_on_success 

98 def setter(self, values: List[Any]) -> bool: 

99 """Set list of values. 

100 

101 Args: 

102 values: List of values. 

103 

104 Returns: 

105 True if successful, False otherwise. 

106 """ 

107 _validate_list_input(values, expected_type, type_name) 

108 

109 # Convert values to the expected type 

110 if expected_type == (int, float): # Special case for float lists 

111 converted_values = [float(v) for v in values] 

112 else: 

113 converted_values = values 

114 

115 arr = (c_type * len(converted_values))(*converted_values) 

116 return setter_func(self._av_ptr, arr, len(converted_values)) 

117 

118 return setter 

119 

120 

121def _create_list_getter(c_type, getter_func, free_func, expected_type): 

122 """Factory function to create list getter methods. 

123 

124 Args: 

125 c_type: C type for the array (e.g., ctypes.c_float). 

126 getter_func: C API function for getting the list. 

127 free_func: C API function for freeing the list. 

128 expected_type: Python type for the returned values. 

129 

130 Returns: 

131 Decorated getter method. 

132 """ 

133 

134 @_cache_with_type_check(expected_type) 

135 def getter(self) -> List[Any]: 

136 """Get list of values. 

137 

138 Returns: 

139 List of values. 

140 

141 Raises: 

142 RuntimeError: If retrieval fails. 

143 """ 

144 size = ctypes.c_int64() 

145 c_array = getter_func(self._av_ptr, ctypes.byref(size)) 

146 if not c_array: 

147 raise RuntimeError("Failed to get list value") 

148 

149 try: 

150 return [c_array[i] for i in range(size.value)] 

151 finally: 

152 free_func(c_array) 

153 

154 return getter 

155 

156 

157class _AttrValue: 

158 """AttrValue for attribute value operations in GraphEngine. 

159 

160 This class provides a Pythonic interface for attribute value operations 

161 using the GraphEngine C API. 

162 

163 Example: 

164 >>> attr = _AttrValue() 

165 >>> attr.set_string("hello") 

166 >>> value = attr.get_string() 

167 >>> attr.set_list_float([1.0, 2.0, 3.0]) 

168 >>> values = attr.get_list_float() 

169 """ 

170 

171 def __init__(self) -> None: 

172 """Initialize an AttrValue. 

173 

174 Creates a new AttrValue instance using the GraphEngine C API. 

175 """ 

176 # Create new AttrValue 

177 self._av_ptr = graph_lib.GeApiWrapper_AttrValue_Create() 

178 

179 self._value_type = None 

180 self._cached_value = None 

181 

182 def __del__(self) -> None: 

183 """Clean up resources.""" 

184 graph_lib.GeApiWrapper_AttrValue_Destroy(self._av_ptr) 

185 self._av_ptr = None 

186 

187 def __copy__(self) -> None: 

188 """Copy is not supported.""" 

189 raise RuntimeError("AttrValue does not support copy") 

190 

191 def __deepcopy__(self, memodict) -> None: 

192 """Deep copy is not supported.""" 

193 raise RuntimeError("AttrValue does not support deepcopy") 

194 

195 def _clear_cache(self) -> None: 

196 """Clear cached value and type.""" 

197 self._cached_value = None 

198 self._value_type = None 

199 

200 def _type_name(self, avt: AttrValueType) -> str: 

201 """Get human-readable type name. 

202 

203 Args: 

204 avt: AttrDataType enum value. 

205 

206 Returns: 

207 Human-readable type name. 

208 """ 

209 type_names = { 

210 AttrValueType.VT_STRING: "string", 

211 AttrValueType.VT_FLOAT: "float", 

212 AttrValueType.VT_BOOL: "bool", 

213 AttrValueType.VT_INT: "int", 

214 AttrValueType.VT_DATA_TYPE: "data_type", 

215 AttrValueType.VT_LIST_FLOAT: "list_float", 

216 AttrValueType.VT_LIST_BOOL: "list_bool", 

217 AttrValueType.VT_LIST_INT: "list_int", 

218 AttrValueType.VT_LIST_DATA_TYPE: "list_data_type", 

219 AttrValueType.VT_LIST_STRING: "list_string", 

220 } 

221 return type_names.get(avt, f"unknown({avt})") 

222 

223 @property 

224 def value_type(self) -> AttrValueType: 

225 """Get the current value type. 

226 

227 Returns: 

228 AttrDataType enum value. 

229 """ 

230 if self._value_type is None: 

231 self._value_type = AttrValueType(graph_lib.GeApiWrapper_AttrValue_GetValueType(self._av_ptr)) 

232 return self._value_type 

233 

234 def get_value_type(self) -> AttrValueType: 

235 """Get the current value type. 

236 

237 Returns: 

238 AttrDataType enum value. 

239 """ 

240 return self.value_type 

241 

242 def get_value(self) -> Any: 

243 """Get the current value. 

244 

245 Returns: 

246 Value. 

247 """ 

248 if self.value_type == AttrValueType.VT_STRING: 

249 return self.get_string() 

250 elif self.value_type == AttrValueType.VT_FLOAT: 

251 return self.get_float() 

252 elif self.value_type == AttrValueType.VT_BOOL: 

253 return self.get_bool() 

254 elif self.value_type == AttrValueType.VT_INT: 

255 return self.get_int() 

256 elif self.value_type == AttrValueType.VT_DATA_TYPE: 

257 return self.get_data_type() 

258 elif self.value_type == AttrValueType.VT_TENSOR: 

259 return self.get_tensor() 

260 elif self.value_type == AttrValueType.VT_LIST_FLOAT: 

261 return self.get_list_float() 

262 elif self.value_type == AttrValueType.VT_LIST_BOOL: 

263 return self.get_list_bool() 

264 elif self.value_type == AttrValueType.VT_LIST_INT: 

265 return self.get_list_int() 

266 elif self.value_type == AttrValueType.VT_LIST_DATA_TYPE: 

267 return self.get_list_data_type() 

268 elif self.value_type == AttrValueType.VT_LIST_STRING: 

269 return self.get_list_string() 

270 else: 

271 raise RuntimeError(f"Unsupported attribute type: {self.value_type}") 

272 

273 def set_value(self, value: Any) -> None: 

274 """Set the current value. 

275 

276 Args: 

277 value: Value to set. 

278 """ 

279 from .tensor import Tensor 

280 

281 if isinstance(value, str): 

282 self.set_string(value) 

283 elif isinstance(value, float): 

284 self.set_float(value) 

285 elif isinstance(value, bool): 

286 self.set_bool(value) 

287 elif isinstance(value, DataType): 

288 self.set_data_type(value) 

289 elif isinstance(value, Tensor): 

290 self.set_tensor(value) 

291 elif isinstance(value, int): 

292 self.set_int(value) 

293 elif isinstance(value, list) and all(isinstance(v, float) for v in value): 

294 self.set_list_float(value) 

295 elif isinstance(value, list) and all(isinstance(v, bool) for v in value): 

296 self.set_list_bool(value) 

297 elif isinstance(value, list) and all(isinstance(v, int) for v in value): 

298 self.set_list_int(value) 

299 elif isinstance(value, list) and all(isinstance(v, DataType) for v in value): 

300 self.set_list_data_type(value) 

301 elif isinstance(value, list) and all(isinstance(v, str) for v in value): 

302 self.set_list_string(value) 

303 else: 

304 raise ValueError(f"Unsupported attribute type: {type(value)} for value: {value}") 

305 

306 # String operations 

307 @_clear_cache_on_success 

308 def set_string(self, value: str) -> bool: 

309 """Set string value. 

310 

311 Args: 

312 value: String value to set. 

313 

314 Returns: 

315 True if successful, False otherwise. 

316 """ 

317 if not isinstance(value, str): 

318 raise TypeError("Value must be a string") 

319 

320 value_bytes = value.encode("utf-8") 

321 return graph_lib.GeApiWrapper_AttrValue_SetString(self._av_ptr, value_bytes) 

322 

323 @_cache_with_type_check(AttrValueType.VT_STRING) 

324 def get_string(self) -> str: 

325 """Get string value. 

326 

327 Returns: 

328 String value. 

329 

330 Raises: 

331 RuntimeError: If value is not a string or retrieval fails. 

332 """ 

333 c_str = graph_lib.GeApiWrapper_AttrValue_GetString(self._av_ptr) 

334 if not c_str: 

335 raise RuntimeError("Failed to get string value") 

336 

337 try: 

338 return ctypes.string_at(c_str).decode("utf-8") 

339 finally: 

340 graph_lib.GeApiWrapper_FreeString(c_str) 

341 

342 # Float operations 

343 @_clear_cache_on_success 

344 def set_float(self, value: float) -> bool: 

345 """Set float value. 

346 

347 Args: 

348 value: Float value to set. 

349 

350 Returns: 

351 True if successful, False otherwise. 

352 """ 

353 if not isinstance(value, (int, float)): 

354 raise TypeError("Value must be a number") 

355 

356 return graph_lib.GeApiWrapper_AttrValue_SetFloat(self._av_ptr, ctypes.c_float(float(value))) 

357 

358 @_cache_with_type_check(AttrValueType.VT_FLOAT) 

359 def get_float(self) -> float: 

360 """Get float value. 

361 

362 Returns: 

363 Float value. 

364 

365 Raises: 

366 RuntimeError: If value is not a float or retrieval fails. 

367 """ 

368 value = ctypes.c_float() 

369 ret = graph_lib.GeApiWrapper_AttrValue_GetFloat(self._av_ptr, ctypes.byref(value)) 

370 if ret != 0: 

371 raise RuntimeError("Failed to get float value") 

372 return value.value 

373 

374 # Bool operations 

375 @_clear_cache_on_success 

376 def set_bool(self, value: bool) -> bool: 

377 """Set bool value. 

378 

379 Args: 

380 value: Bool value to set. 

381 

382 Returns: 

383 True if successful, False otherwise. 

384 """ 

385 if not isinstance(value, bool): 

386 raise TypeError("Value must be a boolean") 

387 

388 return graph_lib.GeApiWrapper_AttrValue_SetBool(self._av_ptr, ctypes.c_bool(value)) 

389 

390 @_cache_with_type_check(AttrValueType.VT_BOOL) 

391 def get_bool(self) -> bool: 

392 """Get bool value. 

393 

394 Returns: 

395 Bool value. 

396 

397 Raises: 

398 RuntimeError: If value is not a bool or retrieval fails. 

399 """ 

400 value = ctypes.c_bool() 

401 ret = graph_lib.GeApiWrapper_AttrValue_GetBool(self._av_ptr, ctypes.byref(value)) 

402 if ret != 0: 

403 raise RuntimeError("Failed to get bool value") 

404 return value.value 

405 

406 # Int operations 

407 @_clear_cache_on_success 

408 def set_int(self, value: int) -> bool: 

409 """Set int value. 

410 

411 Args: 

412 value: Int value to set. 

413 

414 Returns: 

415 True if successful, False otherwise. 

416 """ 

417 if not isinstance(value, int): 

418 raise TypeError("Value must be an integer") 

419 

420 return graph_lib.GeApiWrapper_AttrValue_SetInt(self._av_ptr, ctypes.c_int64(value)) 

421 

422 @_cache_with_type_check(AttrValueType.VT_INT) 

423 def get_int(self) -> int: 

424 """Get int value. 

425 

426 Returns: 

427 Int value. 

428 

429 Raises: 

430 RuntimeError: If value is not an int or retrieval fails. 

431 """ 

432 value = ctypes.c_int64() 

433 ret = graph_lib.GeApiWrapper_AttrValue_GetInt(self._av_ptr, ctypes.byref(value)) 

434 if ret != 0: 

435 raise RuntimeError("Failed to get int value") 

436 return value.value 

437 

438 # DataType operations 

439 @_clear_cache_on_success 

440 def set_data_type(self, value: DataType) -> bool: 

441 """Set DataType value. 

442 

443 Args: 

444 value: DataType value to set. 

445 

446 Returns: 

447 True if successful, False otherwise. 

448 """ 

449 if not isinstance(value, DataType): 

450 raise TypeError("Value must be a DataType") 

451 

452 return graph_lib.GeApiWrapper_AttrValue_SetDataType(self._av_ptr, ctypes.c_int(value.value)) 

453 

454 @_cache_with_type_check(AttrValueType.VT_DATA_TYPE) 

455 def get_data_type(self) -> DataType: 

456 """Get DataType value. 

457 

458 Returns: 

459 DataType value. 

460 

461 Raises: 

462 RuntimeError: If value is not a DataType or retrieval fails. 

463 """ 

464 value = ctypes.c_int() 

465 ret = graph_lib.GeApiWrapper_AttrValue_GetDataType(self._av_ptr, ctypes.byref(value)) 

466 if ret != 0: 

467 raise RuntimeError("Failed to get DataType value") 

468 return DataType(value.value) 

469 

470 @_clear_cache_on_success 

471 def set_tensor(self, value) -> bool: 

472 """Set Tensor value.""" 

473 from .tensor import Tensor 

474 

475 if not isinstance(value, Tensor): 

476 raise TypeError("Value must be a Tensor") 

477 return graph_lib.GeApiWrapper_AttrValue_SetTensor(self._av_ptr, value._handle) 

478 

479 @_cache_with_type_check(AttrValueType.VT_TENSOR) 

480 def get_tensor(self): 

481 """Get Tensor value.""" 

482 from .tensor import Tensor 

483 

484 value = graph_lib.GeApiWrapper_AttrValue_GetTensor(self._av_ptr) 

485 if not value: 

486 raise RuntimeError("Failed to get Tensor value") 

487 return Tensor._create_from(value) 

488 

489 # List operations 

490 set_list_float = _create_list_setter( 

491 ctypes.c_float, 

492 graph_lib.GeApiWrapper_AttrValue_SetListFloat, 

493 (int, float), 

494 "numbers", 

495 ) 

496 

497 get_list_float = _create_list_getter( 

498 ctypes.c_float, 

499 graph_lib.GeApiWrapper_AttrValue_GetListFloat, 

500 graph_lib.GeApiWrapper_FreeListFloat, 

501 AttrValueType.VT_LIST_FLOAT, 

502 ) 

503 

504 set_list_int = _create_list_setter(ctypes.c_int64, graph_lib.GeApiWrapper_AttrValue_SetListInt, int, "integers") 

505 

506 get_list_int = _create_list_getter( 

507 ctypes.c_int64, 

508 graph_lib.GeApiWrapper_AttrValue_GetListInt, 

509 graph_lib.GeApiWrapper_FreeListInt, 

510 AttrValueType.VT_LIST_INT, 

511 ) 

512 

513 set_list_bool = _create_list_setter(ctypes.c_bool, graph_lib.GeApiWrapper_AttrValue_SetListBool, bool, "booleans") 

514 

515 get_list_bool = _create_list_getter( 

516 ctypes.c_bool, 

517 graph_lib.GeApiWrapper_AttrValue_GetListBool, 

518 graph_lib.GeApiWrapper_FreeListBool, 

519 AttrValueType.VT_LIST_BOOL, 

520 ) 

521 

522 set_list_data_type = _create_list_setter( 

523 ctypes.c_int, 

524 graph_lib.GeApiWrapper_AttrValue_SetListDataType, 

525 DataType, 

526 "DataTypes", 

527 ) 

528 

529 get_list_data_type = _create_list_getter( 

530 ctypes.c_int, 

531 graph_lib.GeApiWrapper_AttrValue_GetListDataType, 

532 graph_lib.GeApiWrapper_FreeListDataType, 

533 AttrValueType.VT_LIST_DATA_TYPE, 

534 ) 

535 

536 # String list needs special handling due to encoding 

537 @_clear_cache_on_success 

538 def set_list_string(self, values: List[str]) -> bool: 

539 """Set list of string values. 

540 

541 Args: 

542 values: List of string values. 

543 

544 Returns: 

545 True if successful, False otherwise. 

546 """ 

547 _validate_list_input(values, str, "strings") 

548 

549 arr = (ctypes.c_char_p * len(values))(*[v.encode("utf-8") for v in values]) 

550 return graph_lib.GeApiWrapper_AttrValue_SetListString(self._av_ptr, arr, len(values)) 

551 

552 @_cache_with_type_check(AttrValueType.VT_LIST_STRING) 

553 def get_list_string(self) -> List[str]: 

554 """Get list of string values. 

555 

556 Returns: 

557 List of string values. 

558 

559 Raises: 

560 RuntimeError: If value is not a list of strings or retrieval fails. 

561 """ 

562 size = ctypes.c_int64() 

563 c_array = graph_lib.GeApiWrapper_AttrValue_GetListString(self._av_ptr, ctypes.byref(size)) 

564 if not c_array: 

565 raise RuntimeError("Failed to get list_string value") 

566 

567 try: 

568 return [ctypes.string_at(c_array[i]).decode("utf-8") for i in range(size.value)] 

569 finally: 

570 graph_lib.GeApiWrapper_FreeListString(c_array) 

571 

572 def __str__(self) -> str: 

573 """String representation of AttrValue.""" 

574 try: 

575 value_type = self.get_value_type() 

576 type_name = self._type_name(value_type) 

577 

578 if value_type == AttrValueType.VT_STRING: 

579 return f"AttrValue(type: {type_name}, value: '{self.get_string()}')" 

580 elif value_type == AttrValueType.VT_FLOAT: 

581 return f"AttrValue(type: {type_name}, value: {self.get_float()})" 

582 elif value_type == AttrValueType.VT_BOOL: 

583 return f"AttrValue(type: {type_name}, value: {self.get_bool()})" 

584 elif value_type == AttrValueType.VT_INT: 

585 return f"AttrValue(type: {type_name}, value: {self.get_int()})" 

586 elif value_type == AttrValueType.VT_DATA_TYPE: 

587 return f"AttrValue(type: {type_name}, value: {self.get_data_type()})" 

588 elif value_type == AttrValueType.VT_TENSOR: 

589 return f"AttrValue(type: {type_name}, value: {self.get_tensor()})" 

590 elif value_type == AttrValueType.VT_LIST_FLOAT: 

591 return f"AttrValue(type: {type_name}, value: {self.get_list_float()})" 

592 elif value_type == AttrValueType.VT_LIST_BOOL: 

593 return f"AttrValue(type: {type_name}, value: {self.get_list_bool()})" 

594 elif value_type == AttrValueType.VT_LIST_INT: 

595 return f"AttrValue(type: {type_name}, value: {self.get_list_int()})" 

596 elif value_type == AttrValueType.VT_LIST_DATA_TYPE: 

597 return f"AttrValue(type: {type_name}, value: {self.get_list_data_type()})" 

598 elif value_type == AttrValueType.VT_LIST_STRING: 

599 return f"AttrValue(type: {type_name}, value: {self.get_list_string()})" 

600 else: 

601 return f"AttrValue(type: {type_name}, value: <unknown>)" 

602 except Exception: 

603 return "AttrValue(<error reading value>)" 

604 

605 def __repr__(self) -> str: 

606 """Detailed string representation of AttrValue.""" 

607 return self.__str__()