Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/ge/ge/_internal/artifact_utils.py: 87%

1#!/usr/bin/env python3 

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

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

4# Copyright (c) 2026 Huawei Technologies Co., Ltd. 

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

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

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

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

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

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

11# ----------------------------------------------------------------------------------------------------------- 

12 

13"""Shared artifact discovery helpers for GE Python modules.""" 

14 

15from __future__ import annotations 

16 

17import importlib.util 

18import json 

19import platform 

20import sys 

21from dataclasses import dataclass 

22from pathlib import Path 

23from types import ModuleType 

24from typing import Callable, Iterable, Optional 

25 

26 

27@dataclass(frozen=True) 

28class PythonArtifact: 

29 root: Path 

30 manifest_path: Path 

31 python_tag: str 

32 platform_tag: str 

33 abi: int 

34 native_path: Path 

35 bridge_path: Optional[Path] = None 

36 

37 

38def current_python_tag() -> str: 

39 return f"cp{sys.version_info.major}{sys.version_info.minor}" 

40 

41 

42def current_platform_tag() -> str: 

43 return f"{platform.system().lower()}-{platform.machine() or 'unknown'}" 

44 

45 

46def iter_manifest_paths(root: Path) -> Iterable[Path]: 

47 if not root.exists(): 

48 return 

49 root_manifest = root / "manifest.json" 

50 if root_manifest.is_file(): 

51 yield root_manifest 

52 for child in sorted(root.iterdir()): 

53 manifest_path = child / "manifest.json" 

54 if manifest_path.is_file(): 

55 yield manifest_path 

56 

57 

58def load_manifest_json(manifest_path: Path) -> dict: 

59 return json.loads(manifest_path.read_text(encoding="utf-8")) 

60 

61 

62def _load_artifact_manifest( 

63 manifest_path: Path, abi_key: str, require_bridge: bool 

64) -> Optional[PythonArtifact]: 

65 try: 

66 manifest = load_manifest_json(manifest_path) 

67 artifacts = manifest["artifacts"] 

68 native_path = (manifest_path.parent / artifacts["native"]).resolve() 

69 bridge_path = ( 

70 (manifest_path.parent / artifacts["bridge"]).resolve() 

71 if require_bridge 

72 else None 

73 ) 

74 artifact = PythonArtifact( 

75 root=manifest_path.parent.resolve(), 

76 manifest_path=manifest_path.resolve(), 

77 python_tag=manifest["python_tag"], 

78 platform_tag=manifest["platform"], 

79 abi=int(manifest[abi_key]), 

80 native_path=native_path, 

81 bridge_path=bridge_path, 

82 ) 

83 except (KeyError, TypeError, ValueError, OSError): 

84 return None 

85 if not artifact.native_path.is_file(): 

86 return None 

87 if require_bridge and ( 

88 artifact.bridge_path is None or not artifact.bridge_path.is_file() 

89 ): 

90 return None 

91 return artifact 

92 

93 

94def load_native_artifact_manifest(manifest_path: Path) -> Optional[PythonArtifact]: 

95 return _load_artifact_manifest(manifest_path, "native_abi", False) 

96 

97 

98def load_bridge_artifact_manifest(manifest_path: Path) -> Optional[PythonArtifact]: 

99 return _load_artifact_manifest(manifest_path, "bridge_abi", True) 

100 

101 

102def iter_artifacts( 

103 root: Path, load_manifest: Callable[[Path], Optional[PythonArtifact]] 

104) -> Iterable[PythonArtifact]: 

105 for manifest_path in iter_manifest_paths(root): 

106 artifact = load_manifest(manifest_path) 

107 if artifact is not None: 

108 yield artifact 

109 

110 

111def find_compatible_artifact( 

112 artifacts: Iterable[PythonArtifact], abi_version: int 

113) -> Optional[PythonArtifact]: 

114 python_tag = current_python_tag() 

115 platform_tag = current_platform_tag() 

116 for artifact in artifacts: 

117 if ( 

118 artifact.python_tag == python_tag 

119 and artifact.platform_tag == platform_tag 

120 and artifact.abi == abi_version 

121 ): 

122 return artifact 

123 return None 

124 

125 

126def load_module_from_path(module_name: str, module_path: Path) -> ModuleType: 

127 loaded_module = sys.modules.get(module_name) 

128 if loaded_module is not None: 

129 return loaded_module 

130 spec = importlib.util.spec_from_file_location(module_name, module_path) 

131 if spec is None or spec.loader is None: 

132 raise ImportError(f"Cannot create import spec for {module_path}") 

133 module = importlib.util.module_from_spec(spec) 

134 sys.modules[module_name] = module 

135 try: 

136 spec.loader.exec_module(module) 

137 except BaseException: 

138 sys.modules.pop(module_name, None) 

139 raise 

140 return module