Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/collect/trace/trace_collect.py: 95%
220 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-19 17:46 +0800
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3# ----------------------------------------------------------------------------
4# Copyright (c) 2025 Huawei Technologies Co., Ltd.
5#
6# Licensed under the Apache License, Version 2.0 (the "License");
7# you may not use this file except in compliance with the License.
8# You may obtain a copy of the License at
9#
10# http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17# ----------------------------------------------------------------------------
19import time
20import struct
21import os.path
22from threading import Thread
23from datetime import datetime
26from common import FileOperate as f
27from common import log_error, log_warning
28from common.task_common import out_progress_bar
30MAGIC_VERSION_INFO = {
31 "2": "0xd928",
32 "3": "0xd928"
33}
35NS_TO_S = 1000000000
36FREQ_GHZ_TO_KHZ = 1000000
38TRACE_STRUCT_FIELD_TYPE_CHAR = 0
39TRACE_STRUCT_FIELD_TYPE_INT8 = 1
40TRACE_STRUCT_FIELD_TYPE_UINT8 = 2
41TRACE_STRUCT_FIELD_TYPE_INT16 = 3
42TRACE_STRUCT_FIELD_TYPE_UINT16 = 4
43TRACE_STRUCT_FIELD_TYPE_INT32 = 5
44TRACE_STRUCT_FIELD_TYPE_UINT32 = 6
45TRACE_STRUCT_FIELD_TYPE_INT64 = 7
46TRACE_STRUCT_FIELD_TYPE_UINT64 = 8
47TRACE_STRUCT_ARRAY_TYPE_CHAR = 100
48TRACE_STRUCT_ARRAY_TYPE_INT8 = 101
49TRACE_STRUCT_ARRAY_TYPE_UINT8 = 102
50TRACE_STRUCT_ARRAY_TYPE_INT16 = 103
51TRACE_STRUCT_ARRAY_TYPE_UINT16 = 104
52TRACE_STRUCT_ARRAY_TYPE_INT32 = 105
53TRACE_STRUCT_ARRAY_TYPE_UINT32 = 106
54TRACE_STRUCT_ARRAY_TYPE_INT64 = 107
55TRACE_STRUCT_ARRAY_TYPE_UINT64 = 108
56TRACE_STRUCT_BOOL = 10001
58TRACE_STRUCT_SHOW_MODE_DEC = 0
59TRACE_STRUCT_SHOW_MODE_BIN = 1
60TRACE_STRUCT_SHOW_MODE_HEX = 2
61TRACE_STRUCT_SHOW_MODE_CHAR = 3
63UNPACK = {
64 TRACE_STRUCT_FIELD_TYPE_CHAR: ['s', 1],
65 TRACE_STRUCT_FIELD_TYPE_INT8: ['b', 1],
66 TRACE_STRUCT_FIELD_TYPE_UINT8: ['B', 1],
67 TRACE_STRUCT_FIELD_TYPE_INT16: ['h', 2],
68 TRACE_STRUCT_FIELD_TYPE_UINT16: ['H', 2],
69 TRACE_STRUCT_FIELD_TYPE_INT32: ['i', 4],
70 TRACE_STRUCT_FIELD_TYPE_UINT32: ['I', 4],
71 TRACE_STRUCT_FIELD_TYPE_INT64: ['q', 8],
72 TRACE_STRUCT_FIELD_TYPE_UINT64: ['Q', 8],
73 TRACE_STRUCT_ARRAY_TYPE_CHAR: ['s', 1],
74 TRACE_STRUCT_ARRAY_TYPE_INT8: ['b', 1],
75 TRACE_STRUCT_ARRAY_TYPE_UINT8: ['B', 1],
76 TRACE_STRUCT_ARRAY_TYPE_INT16: ['h', 2],
77 TRACE_STRUCT_ARRAY_TYPE_UINT16: ['H', 2],
78 TRACE_STRUCT_ARRAY_TYPE_INT32: ['i', 4],
79 TRACE_STRUCT_ARRAY_TYPE_UINT32: ['I', 4],
80 TRACE_STRUCT_ARRAY_TYPE_INT64: ['q', 8],
81 TRACE_STRUCT_ARRAY_TYPE_UINT64: ['Q', 8],
82 TRACE_STRUCT_BOOL: ['?', 1]
83}
86def trace_show_mode(mode, value):
87 if mode == TRACE_STRUCT_SHOW_MODE_DEC:
88 return str(value)
89 elif mode == TRACE_STRUCT_SHOW_MODE_BIN:
90 return str(bin(int(value)))
91 elif mode == TRACE_STRUCT_SHOW_MODE_HEX:
92 return str(hex(int(value)))
93 else:
94 return str(value)
97class ParseTrace:
98 def __init__(self, is_file=False):
99 self.is_file = is_file
100 self.real_time = 0
101 self.tz_offset = 0
102 self.cpu_freq = 0
104 def error(self, msg):
105 if self.is_file:
106 log_error(msg)
108 def warning(self, msg):
109 if self.is_file:
110 log_warning(msg)
112 @staticmethod
113 def write_res_txt(msg_txt, trace_file):
114 trace_file = trace_file.replace(".bin", ".txt")
115 with open(trace_file, "w") as fw:
116 fw.write(msg_txt.replace('\x00', ''))
118 @staticmethod
119 def time_zone_calculation(tz_offset):
120 date_now = time.localtime()
121 date_utc = time.gmtime()
122 date_utc = datetime(date_utc.tm_year, date_utc.tm_mon, date_utc.tm_mday, date_utc.tm_hour, date_utc.tm_min)
123 date_now = datetime(date_now.tm_year, date_now.tm_mon, date_now.tm_mday, date_now.tm_hour, date_now.tm_min)
124 return ((date_now.timestamp() - date_utc.timestamp()) // 60 - tz_offset) * NS_TO_S
126 @staticmethod
127 def get_struct_data(fp, num, num_type):
128 try:
129 unpack_list = UNPACK.get(num_type)
130 data = struct.unpack(f'{num}{unpack_list[0]}', fp.read(unpack_list[1] * num))
131 except Exception as e:
132 raise ValueError('Unable to parse data, check whether the version matches '
133 'or whether the file content is complete.') from e
134 if num_type in [TRACE_STRUCT_FIELD_TYPE_CHAR, TRACE_STRUCT_ARRAY_TYPE_CHAR]:
135 return data[0].decode()
136 if num == 1:
137 return data[0]
138 return data
140 @staticmethod
141 def get_res_data(fp, byte_len):
142 try:
143 fp.read(byte_len)
144 except Exception as e:
145 raise ValueError('Unable to parse data, check whether the version matches '
146 'or whether the file content is complete.') from e
148 def parse_ctrl_head(self, fp, trace_file):
149 """
150 This is a parse control header information.
151 """
152 trace_file_name = trace_file.split(os.sep)[-1]
153 # Obtains the magic and version information.
154 magic, version = self.get_struct_data(fp, 2, TRACE_STRUCT_FIELD_TYPE_UINT32)
155 if str(version) not in MAGIC_VERSION_INFO.keys() or MAGIC_VERSION_INFO[str(version)] != str(hex(magic)):
156 raise ValueError(f'The {trace_file_name} cannot be parsed, check the version.')
158 _, _, _, trace_type = self.get_struct_data(fp, 4, TRACE_STRUCT_FIELD_TYPE_UINT8)
159 if trace_type != 0:
160 raise ValueError(f"The {trace_file_name} cannot be parsed, check trace type.")
161 # Obtains the structSize and dataSize information.
162 struct_size, data_size = self.get_struct_data(fp, 2, TRACE_STRUCT_FIELD_TYPE_UINT32)
164 # Obtains the realTime and minutestWest information.
165 self.tz_offset = self.get_struct_data(fp, 1, TRACE_STRUCT_FIELD_TYPE_UINT32)
166 self.real_time = self.get_struct_data(fp, 1, TRACE_STRUCT_FIELD_TYPE_UINT64)
167 if str(version) == "3":
168 self.cpu_freq = self.get_struct_data(fp, 1, TRACE_STRUCT_FIELD_TYPE_UINT64)
169 self.get_res_data(fp, 8)
170 else:
171 # reserve 16byte
172 self.get_res_data(fp, 16)
173 # record current location
174 current = fp.tell()
175 # skip to the end
176 fp.seek(0, 2)
177 offset_end = fp.tell() - current
178 fp.seek(current)
179 if offset_end <= struct_size:
180 raise ValueError(f"The {trace_file_name} is incomplete and cannot be parsed.")
181 if offset_end < (struct_size + data_size):
182 self.warning(f"The {trace_file_name} data is incomplete, which may cause data loss.")
184 def parse_struct_segment(self, fp):
185 """
186 This is a parse data structure body information.
187 """
188 struct_dict = dict()
189 # get struct count
190 struct_count = self.get_struct_data(fp, 1, TRACE_STRUCT_FIELD_TYPE_UINT32)
191 # reserve 36byte
192 self.get_res_data(fp, 36)
193 for _ in range(struct_count):
194 # get struct segment information
195 struct_name = self.get_struct_data(fp, 32, TRACE_STRUCT_FIELD_TYPE_CHAR).replace('\x00', '')
196 item_num = self.get_struct_data(fp, 1, TRACE_STRUCT_FIELD_TYPE_UINT32)
197 struct_type = self.get_struct_data(fp, 1, TRACE_STRUCT_FIELD_TYPE_UINT8)
198 self.get_res_data(fp, 3)
199 item_lists = []
200 for _ in range(item_num):
201 item_name = self.get_struct_data(fp, 32, TRACE_STRUCT_FIELD_TYPE_CHAR).replace('\x00', '')
202 item_type, item_mode = self.get_struct_data(fp, 2, TRACE_STRUCT_FIELD_TYPE_UINT8)
203 item_length = self.get_struct_data(fp, 1, TRACE_STRUCT_FIELD_TYPE_UINT16)
204 self.get_res_data(fp, 4)
205 item_lists.append([item_name, item_type, item_mode, item_length])
206 struct_dict[struct_type] = {"struct_name": struct_name, "item_lists": item_lists}
207 return struct_dict
209 def parse_msg_data(self, fp, item_list, txt_size):
210 """
211 Parse the data based on the item parameter.
212 """
213 item_name, item_type, item_mode, item_length = item_list
214 msg_byte = UNPACK.get(item_type)[1]
215 data_list = ""
216 use_byte = 0
217 while item_length > 0 and txt_size > 0:
218 if item_length < msg_byte or txt_size < msg_byte:
219 raise ValueError('The data type or data length is incorrect and cannot be parsed.')
220 data = self.get_struct_data(fp, 1, item_type)
221 item_length -= msg_byte
222 txt_size -= msg_byte
223 use_byte += msg_byte
224 data = trace_show_mode(item_mode, data)
225 if item_type < TRACE_STRUCT_ARRAY_TYPE_CHAR:
226 return f"{item_name}[{data}], ", use_byte
227 else:
228 if data_list == "" or item_type == TRACE_STRUCT_ARRAY_TYPE_CHAR:
229 data_list += f"{data}"
230 else:
231 data_list += f", {data}"
232 return f"{item_name}[{data_list}], ", use_byte
234 def parse_data_segment(self, fp, trace_file):
235 """
236 This is a data parsing function that returns the parsed txt string.
237 """
238 self.parse_ctrl_head(fp, trace_file)
239 struct_dict = self.parse_struct_segment(fp)
240 if not struct_dict:
241 raise ValueError('Failed to parse the data, check whether the file is complete.')
242 offset_time_ns = self.time_zone_calculation(self.tz_offset)
243 msg_size, msg_txt_size, msg_num, _ = self.get_struct_data(fp, 4, TRACE_STRUCT_FIELD_TYPE_UINT32)
244 msg_txt = ""
245 for _ in range(msg_num):
246 # data head
247 cycle = self.get_struct_data(fp, 1, TRACE_STRUCT_FIELD_TYPE_UINT64)
248 txt_size = self.get_struct_data(fp, 1, TRACE_STRUCT_FIELD_TYPE_UINT32)
249 busy = self.get_struct_data(fp, 1, TRACE_STRUCT_BOOL)
250 struct_type = self.get_struct_data(fp, 1, TRACE_STRUCT_FIELD_TYPE_UINT8)
251 self.get_res_data(fp, 2)
252 if busy:
253 self.get_res_data(fp, msg_txt_size)
254 continue
255 if self.cpu_freq != 0:
256 time_str = datetime.fromtimestamp((self.real_time + (cycle / self.cpu_freq) * FREQ_GHZ_TO_KHZ +
257 offset_time_ns) / NS_TO_S)
258 else:
259 time_str = datetime.fromtimestamp((self.real_time + cycle + offset_time_ns) / NS_TO_S)
260 time_str = time_str.strftime("%Y-%m-%d %H:%M:%S.%f")
261 # data
262 struct_info = struct_dict.get(struct_type)
263 if not struct_info:
264 self.get_res_data(fp, msg_txt_size)
265 continue
266 msg_txt += "%s.%s %s: " % (time_str[:-3], time_str[-3:], struct_info.get("struct_name"))
267 use_msg_data = 0
268 for item_list in struct_info.get("item_lists"):
269 item_txt, use_byte = self.parse_msg_data(fp, item_list, txt_size)
270 msg_txt += item_txt
271 use_msg_data += use_byte
272 # txt_size not used up, skipping byte length
273 if msg_txt_size - use_msg_data > 0:
274 self.get_res_data(fp, msg_txt_size - use_msg_data)
275 # remove end of line ', '
276 msg_txt = msg_txt[:-2] + "\n"
277 return msg_txt
279 def start_parse_file(self, trace_file, count=0, num=0):
280 out_progress_bar(count, num)
281 msg_txt = self.parse(trace_file)
282 if msg_txt:
283 self.write_res_txt(msg_txt, trace_file)
284 os.remove(trace_file)
285 return True
286 return False
288 def parse(self, trace_file):
289 msg_txt = ""
290 try:
291 with open(trace_file, "rb") as fp:
292 msg_txt = self.parse_data_segment(fp, trace_file)
293 except ValueError as e:
294 self.error(e)
295 except IOError:
296 self.error(f'The {trace_file} cannot be read or cannot be found.')
297 return msg_txt
299 def run(self, trace_path, count=0):
300 atrace_dirs = f.walk_dir(trace_path)
301 if not atrace_dirs:
302 return False
303 num = 0
304 threads = []
305 for dirs, _, files in atrace_dirs:
306 for file in files:
307 trace_file = os.path.join(dirs, file)
308 num += 1
309 if file.endswith(".bin"):
310 t = Thread(target=self.start_parse_file, args=(trace_file, count, num), daemon=True)
311 t.start()
312 threads.append(t)
314 out_progress_bar(count, count)
315 # wait for all threads to end.
316 for t in threads:
317 t.join()
318 return True
321def collect_trace(output_root_path):
322 trace_path = os.path.join(output_root_path, "dfx", "atrace")
323 parse_trace = ParseTrace()
324 parse_trace.run(trace_path)