Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/ge/ge/graph/tensor.py: 91%
184 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-27 10:02 +0800
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3# ----------------------------------------------------------------------------
4# Copyright (c) 2025 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# ----------------------------------------------------------------------------
13import ctypes
14from typing import TYPE_CHECKING, Any, List, Optional, Union
16from ge._capi.pyes_graph_builder_wrapper import esb_lib
17from ge._capi.pygraph_wrapper import graph_lib
18from ge._capi.pytensor_runtime_wrapper import tensor_runtime_lib
20from ._numeric import float_list_to_fp16_bits
21from .tensor_desc import Shape, TensorDesc
22from .types import DataType, Format, Placement
24if TYPE_CHECKING:
25 from ge.es.graph_builder import GraphBuilder
26 from ge.es.tensor_like import TensorLike
27UnionTensorDataType = Union[List[int], List[float], List[bool]]
30class Tensor:
31 def __init__(
32 self,
33 data: Optional[UnionTensorDataType] = None,
34 file_path: Optional[str] = None,
35 data_type: Optional[DataType] = DataType.DT_FLOAT,
36 format: Optional[Format] = Format.FORMAT_ND,
37 shape: Optional[List[int]] = None,
38 placement: Optional[Placement] = Placement.PLACEMENT_HOST,
39 ) -> None:
40 """
41 Args:
42 data: Data to read from.
43 file_path: File path to read from.
44 data_type: Data type using DataType enum, defaults to DataType.DT_FLOAT.
45 format: Data format using Format enum, defaults to Format.FORMAT_ND.
46 shape: List of shape dimensions, If None, means scalar.
47 placement: Tensor placement using Placement enum, defaults to Placement.PLACEMENT_HOST.
48 Example
49 >>> Tensor(data, None, data_type, format, shape, placement)
50 >>> Tensor(None, file_path, data_type, format, shape, placement)
51 """
52 self._handle = None
53 self._owns_handle = False
54 self._owner = None
55 if shape is not None:
56 if not isinstance(shape, list) or not all(isinstance(dim, int) for dim in shape):
57 raise TypeError("Shape must be a list of integers")
58 if not isinstance(placement, Placement):
59 raise TypeError("Placement must be a Placement")
60 if data is not None and file_path is not None:
61 raise RuntimeError("Tensor should be created either by data or by file")
62 elif data is not None and file_path is None:
63 self._create_from_data(data, data_type, format, shape)
64 elif data is None and file_path is not None:
65 self._create_from_file(file_path, data_type, format, shape)
66 else:
67 self._handle = graph_lib.GeApiWrapper_Tensor_CreateTensor()
68 if not self._handle:
69 raise RuntimeError("Failed to create Tensor")
70 self._owns_handle = True
71 self._owner = None
72 if placement == Placement.PLACEMENT_DEVICE:
73 self.to_device()
75 @staticmethod
76 def _prepare_ctypes_array(data: List[Any], data_type: DataType):
77 int_type_map = {
78 DataType.DT_INT64: ctypes.c_int64,
79 DataType.DT_UINT64: ctypes.c_uint64,
80 DataType.DT_INT32: ctypes.c_int32,
81 DataType.DT_UINT32: ctypes.c_uint32,
82 DataType.DT_INT16: ctypes.c_int16,
83 DataType.DT_UINT16: ctypes.c_uint16,
84 DataType.DT_INT8: ctypes.c_int8,
85 DataType.DT_UINT8: ctypes.c_uint8,
86 }
87 if all(isinstance(x, bool) for x in data):
88 return ctypes.c_bool, [bool(x) for x in data]
90 if data_type in int_type_map:
91 return int_type_map[data_type], [int(x) for x in data]
93 if data_type == DataType.DT_FLOAT:
94 return ctypes.c_float, [float(x) for x in data]
96 if data_type == DataType.DT_FLOAT16:
97 return ctypes.c_uint16, float_list_to_fp16_bits([float(x) for x in data])
99 if data_type == DataType.DT_DOUBLE:
100 raise RuntimeError("DT_DOUBLE is not supported in python Tensor constructor")
102 raise RuntimeError("Failed to create Tensor with data type: {}".format(data_type))
104 def _create_from_data(
105 self,
106 data: Optional[UnionTensorDataType],
107 data_type: Optional[DataType],
108 format: Optional[Format],
109 shape: Optional[List[int]],
110 ):
111 """Create Tensor from data."""
112 if not isinstance(data, list):
113 raise TypeError("data should be List")
115 c_type, normalized = self._prepare_ctypes_array(data, data_type)
116 c_data_array = (c_type * len(normalized))(*normalized)
118 dim_num = len(shape) if shape is not None else 0
119 c_dims_array = (ctypes.c_int64 * dim_num)(*shape) if dim_num > 0 else None
120 self._handle = esb_lib.EsCreateEsCTensor(c_data_array, c_dims_array, dim_num, data_type, format)
122 def _create_from_file(
123 self,
124 file_path: Optional[str],
125 data_type: Optional[DataType],
126 format: Optional[Format],
127 shape: Optional[List[int]],
128 ):
129 """Create Tensor from file."""
130 dim_num = len(shape) if shape is not None else 0
131 c_dims_array = (ctypes.c_int64 * dim_num)(*shape) if dim_num > 0 else None
132 file_path_bytes = file_path.encode("utf-8")
133 self._handle = esb_lib.EsCreateEsCTensorFromFile(file_path_bytes, c_dims_array, dim_num, data_type, format)
135 def __del__(self) -> None:
136 """Clean up resources."""
137 if self._owns_handle:
138 graph_lib.GeApiWrapper_Tensor_DestroyEsCTensor(self._handle) # _handle must be valid
139 self._handle = None
141 def __copy__(self) -> None:
142 """Copy is not supported."""
143 raise RuntimeError("Tensor does not support copy")
145 def __deepcopy__(self, memodict) -> None:
146 """Deep copy is not supported."""
147 raise RuntimeError("Tensor does not support deepcopy")
149 def __str__(self) -> str:
150 return f"""
151 Tensor format is {self.get_format()},
152 data_type is {self.get_data_type()},
153 shape is {self.get_shape()},
154 data is {self.get_data()}
155 """
157 @property
158 def placement(self) -> Placement:
159 return self.get_placement()
161 @classmethod
162 def _create_from(cls, handle: ctypes.c_void_p) -> "Tensor":
163 """Create Tensor object from C++ pointer.
165 Args:
166 handle: C++ Tensor object pointer.
168 Returns:
169 Tensor object.
171 Raises:
172 ValueError: If pointer is None.
173 """
174 if not handle:
175 raise ValueError("Tensor pointer cannot be None")
176 instance = cls.__new__(cls)
177 instance._handle = handle
178 instance._owns_handle = True
179 instance._owner = None
180 return instance
182 def _transfer_ownership_when_pass_as_attr(self, new_owner: "GraphBuilder") -> None:
183 """Transfer ownership of the C++ resource to new_owner.
185 After calling this method, Python will no longer destroy the underlying
186 C++ Tensor object. This is called automatically when a Tenensor is passed
187 as an attribute.
189 Args:
190 new_owner: The object that will manage the C++ resource (typically a GraphBuilder).
191 this Tensor will hold a reference to keep the new_owner alive.
192 """
193 if self._owner is not None:
194 raise RuntimeError(
195 "Tensor already has an new owner builder :{}, cannot transfer ownership again".format(self._owner.name)
196 )
197 self._owns_handle = False
198 self._owner = new_owner # Keep reference to prevent premature GC
200 def set_format(self, format: Format) -> "Tensor":
201 """Set format of tensor.
203 Args:
204 format: format to be set.
206 Raises:
207 TypeError: If format is not a Format.
208 RuntimeError: If setting format operation fails.
209 """
210 if not isinstance(format, Format):
211 raise TypeError("Format must be a Format")
213 format_ref = ctypes.byref(ctypes.c_int(format.value))
214 ret = graph_lib.GeApiWrapper_Tensor_SetFormat(self._handle, format_ref)
215 if ret != 0: # GRAPH_SUCCESS
216 raise RuntimeError(f"Failed to set format {format}")
217 return self
219 def get_format(self) -> Format:
220 """Get format of tensor.
222 Returns:
223 Format of tensor
225 Raises:
226 ValueError: If return format is not a Format.
227 """
228 res = graph_lib.GeApiWrapper_Tensor_GetFormat(self._handle)
229 return Format(res)
231 @property
232 def format(self) -> Format:
233 return self.get_format()
235 def set_data_type(self, data_type: DataType) -> "Tensor":
236 """Set datatype of tensor.
238 Args:
239 data_type: datatype to be set.
241 Raises:
242 TypeError: If datatype is not a DataType.
243 RuntimeError: If setting datatype operation fails.
244 """
245 if not isinstance(data_type, DataType):
246 raise TypeError("Data_type must be a DataType")
248 datatype_ref = ctypes.byref(ctypes.c_int(data_type.value))
249 ret = graph_lib.GeApiWrapper_Tensor_SetDataType(self._handle, datatype_ref)
250 if ret != 0: # GRAPH_SUCCESS
251 raise RuntimeError(f"Failed to set datatype {data_type}")
252 return self
254 def get_data_type(self) -> DataType:
255 """Get dataype of tensor.
257 Returns:
258 DataType of tensor
260 Raises:
261 ValueError: If return dataype is not a DataType.
262 """
263 res = graph_lib.GeApiWrapper_Tensor_GetDataType(self._handle)
264 return DataType(res)
266 @property
267 def data_type(self) -> DataType:
268 return self.get_data_type()
270 @property
271 def shape(self) -> Shape:
272 return self.get_shape()
274 def get_data(self) -> "TensorLike":
275 """Get tensor data.
277 Returns:
278 Tensor data.
280 Raises:
281 RuntimeError: If data retrieval fails.
282 """
283 c_str = graph_lib.GeApiWrapper_Tensor_GetData(self._handle)
284 if not c_str:
285 raise RuntimeError("Failed to get Tensor data")
287 try:
288 data_str = ctypes.string_at(c_str).decode("utf-8")
289 return unflatten_tensor_data(data_str, self.shape)
290 finally:
291 graph_lib.GeApiWrapper_FreeString(c_str)
293 @property
294 def data(self) -> "TensorLike":
295 return self.get_data()
297 def get_tensor_desc(self) -> TensorDesc:
298 """Get tensor descriptor.
300 Returns:
301 TensorDesc object.
302 """
303 desc_handle = graph_lib.GeApiWrapper_Tensor_GetTensorDesc(self._handle)
304 if not desc_handle:
305 raise RuntimeError("Failed to get tensor desc")
306 return TensorDesc._create_from(desc_handle)
308 def get_shape(self) -> Shape:
309 """Get shape of tensor.
311 Returns:
312 Shape of tensor
313 """
314 dims_num = ctypes.c_size_t()
315 dims_arr = ctypes.POINTER(ctypes.c_int64)()
316 ret = graph_lib.GeApiWrapper_Tensor_GetShape(self._handle, ctypes.byref(dims_arr), ctypes.byref(dims_num))
318 if ret != 0: # GRAPH_SUCCESS
319 raise RuntimeError("Failed to get shape of tensor")
320 try:
321 return Shape([dims_arr[i] for i in range(dims_num.value)])
322 finally:
323 graph_lib.GeApiWrapper_Tensor_FreeDimsArray(dims_arr)
325 def get_placement(self) -> Placement:
326 """Get tensor placement.
328 Returns:
329 Placement of tensor.
330 """
331 res = graph_lib.GeApiWrapper_Tensor_GetPlacement(self._handle)
332 return Placement(res)
334 def to_host(self) -> "Tensor":
335 """Move this tensor from device to host in place."""
336 if self.get_placement() != Placement.PLACEMENT_DEVICE:
337 raise ValueError("to_host() only supports device tensors")
339 ret = tensor_runtime_lib.GeApiWrapper_Tensor_ToHost(self._handle)
340 if ret != 0:
341 raise RuntimeError(f"Failed to move tensor from device to host, ret={ret}")
342 return self
344 def to_device(self) -> "Tensor":
345 """Move this tensor from host to device in place."""
346 if self.get_placement() != Placement.PLACEMENT_HOST:
347 raise ValueError("to_device() only supports host tensors")
349 ret = tensor_runtime_lib.GeApiWrapper_Tensor_ToDevice(self._handle)
350 if ret != 0:
351 raise RuntimeError(f"Failed to move tensor from host to device, ret={ret}")
352 return self
355def _parse_str_list(list_str: str) -> List[Union[int, float]]:
356 """
357 Convert string like '[1, 2, 3]' into python list.
359 Args:
360 list_str: string format of list
362 Returns:
363 List of Number
364 """
365 list_str = list_str.strip()
366 if not (list_str.startswith("[") and list_str.endswith("]")):
367 raise ValueError("Input must start with '[' and end with ']'")
368 inner = list_str[1:-1].strip()
369 if not inner:
370 return []
371 items = [x.strip() for x in inner.split(",")]
372 result = []
373 for item in items:
374 if item.isdigit() or (item.startswith("-") and item[1:].isdigit()):
375 result.append(int(item))
376 else:
377 try:
378 result.append(float(item))
379 except ValueError:
380 raise ValueError(f"Invalid item: '{item}'")
381 return result
384def unflatten_tensor_data(tensor_data: str, shape: List[int]) -> "TensorLike":
385 from ge.es.tensor_like import _unflatten
387 tensor_data_list = _parse_str_list(tensor_data)
388 if len(shape) == 0:
389 if len(tensor_data_list) != 1:
390 raise ValueError(f"Scalar tensor should contain exactly one element, got {len(tensor_data_list)}")
391 return tensor_data_list[0]
392 return _unflatten(tensor_data_list, shape)