Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/ge/ge/ge_global/geapi.py: 100%

33 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-27 10:03 +0800

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"""Initialize GE for GraphEngine graph operations.""" 

15 

16import ctypes 

17 

18from ge._capi.pygeapi_wrapper import geapi_lib 

19from ge.error import raise_ge_error 

20 

21 

22class GeApi: 

23 def __init__(self): 

24 pass 

25 

26 @classmethod 

27 def ge_initialize(cls, config: dict) -> None: 

28 """Initialize GE, prepare for execution 

29 

30 Args: 

31 config: Config. 

32 

33 Returns: 

34 Initialize GE result (int res). 

35 """ 

36 if not isinstance(config, dict): 

37 raise TypeError("Ge init config must be a dictionary") 

38 keys = [k for k in config.keys()] 

39 values = [v for v in config.values()] 

40 _self = cls() 

41 c_array_key = _self._python_list_to_c_array(keys) 

42 c_array_value = _self._python_list_to_c_array(values) 

43 if len(c_array_key) == 0 or len(c_array_value) == 0: 

44 raise TypeError("Ge init config must not be empty") 

45 ret = geapi_lib.GeApiWrapper_GEInitialize( 

46 ctypes.cast(c_array_key, ctypes.POINTER(ctypes.c_char_p)), 

47 ctypes.cast(c_array_value, ctypes.POINTER(ctypes.c_char_p)), 

48 len(keys), 

49 ) 

50 if ret != 0: 

51 raise_ge_error("GEInitialize", ret) 

52 return ret 

53 

54 @classmethod 

55 def ge_finalize(cls) -> None: 

56 """ 

57 GE finalize, releasing all resources 

58 """ 

59 ret = geapi_lib.GeApiWrapper_GEFinalize() 

60 if ret != 0: 

61 raise_ge_error("GEFinalize", ret) 

62 return ret 

63 

64 def _python_list_to_c_array(self, python_list: list): 

65 """ 

66 Convert python list to c array 

67 

68 Parameters: 

69 python_list: python list. 

70 

71 Returns: 

72 c_array: c array. 

73 """ 

74 size = len(python_list) 

75 c_array = (ctypes.c_char_p * size)() 

76 for i, item in enumerate(python_list): 

77 c_array[i] = item.encode("utf-8") 

78 return c_array