1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
|
# Copyright 2018 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""This script implements a few IntelPowerGadget related helper functions.
This script only works on Windows/Mac with Intel CPU. Intel Power Gadget needs
to be installed on the machine before this script works. The software can be
downloaded from:
https://software.intel.com/en-us/articles/intel-power-gadget
An easy way to use the APIs are:
1) Launch your program.
2) Call RunIPG() with no args. It will automatically locate the IPG installed
on the machine.
3) Call AnalyzeIPGLogFile() with no args. It will analyze the default IPG log
file, which is PowerLog.csv at current dir; then it will print out the power
usage summary. If you want to skip a few seconds of the power log data, say,
5 seconds, call AnalyzeIPGLogFile(skip_in_sec=5).
"""
import dataclasses
import datetime
import json
import logging
import os
import subprocess
from typing import Any
from gpu_tests.util import host_information
SummaryType = dict[str, dict[str, float]]
ResultType = dict[str, Any]
MetricType = dict[str, list[str] | list[float]]
@dataclasses.dataclass
class _LogFileColumn:
"""Represents the parsed data from a column in an IPG log file."""
# The index of this column within the file.
index: int
# The name of the column.
label: str
# The sum of all rows within the column.
total: float = 0.0
def LocateIPG() -> str:
if host_information.IsWindows():
ipg_dir = os.getenv('IPG_Dir')
if not ipg_dir:
raise Exception('No env IPG_Dir')
gadget_path = os.path.join(ipg_dir, 'PowerLog3.0.exe')
if not os.path.isfile(gadget_path):
raise Exception("Can't locate Intel Power Gadget at " + gadget_path)
return gadget_path
if host_information.IsMac():
return '/Applications/Intel Power Gadget/PowerLog'
raise Exception('Only supported on Windows/Mac')
def GenerateIPGLogFilename(log_prefix: str = 'PowerLog',
log_dir: str | None = None,
current_run: int = 1,
total_runs: int = 1,
timestamp: bool = False) -> str:
# If all args take default value, it is the IPG's default log path.
log_dir = log_dir or os.getcwd()
log_dir = os.path.abspath(log_dir)
if total_runs > 1:
log_prefix = f'{log_prefix}_{current_run}_{total_runs}'
if timestamp:
now = datetime.datetime.now()
log_prefix = f'{log_prefix}_{now.strftime("%Y%m%d%H%M%S")}'
return os.path.join(log_dir, log_prefix + '.csv')
def RunIPG(duration_in_s: int = 60,
resolution_in_ms: int = 100,
logfile: str | None = None) -> None:
intel_power_gadget_path = LocateIPG()
command = (f'"{intel_power_gadget_path}" -duration {duration_in_s} '
f'-resolution {resolution_in_ms}')
if not logfile:
# It is not necessary but allows to print out the log path for debugging.
logfile = GenerateIPGLogFilename()
command = f'{command} -file {logfile}'
logging.debug('Running: %s', command)
try:
output = subprocess.check_output(command,
shell=True,
stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
logging.error('Running Intel Power Gadget failed. Output: %s', e.output)
raise
logging.debug('Running: DONE')
logging.debug(output)
def AnalyzeIPGLogFile(logfile: str | None = None,
skip_in_sec: int = 0) -> ResultType:
if not logfile:
logfile = GenerateIPGLogFilename()
if not os.path.isfile(logfile):
raise Exception(f"Can't locate logfile at {logfile}")
first_line = True
samples = 0
total_columns = 0
columns = []
col_time = None
with open(logfile, encoding='utf-8') as infile:
contents = infile.read()
for line in contents.splitlines(keepends=True):
tokens = [token.strip('" ') for token in line.split(',')]
if first_line:
first_line = False
total_columns = len(tokens)
for ii in range(total_columns):
token = tokens[ii]
if token.startswith('Elapsed Time'):
col_time = ii
elif token.endswith('(Watt)'):
columns.append(_LogFileColumn(index=ii, label=token[:-len('(Watt)')]))
assert col_time
assert total_columns > 0
assert len(columns) > 0
continue
if len(tokens) != total_columns:
continue
if skip_in_sec > 0 and float(tokens[col_time]) < skip_in_sec:
continue
samples += 1
for c in columns:
c.total += float(tokens[c.index])
results = {'samples': samples}
if samples > 0:
for c in columns:
results[c.label] = c.total / samples
return results
def ProcessResultsFromMultipleIPGRuns(
logfiles: list[str],
skip_in_seconds: int = 0,
outliers: int = 0,
output_json: str | None = None) -> SummaryType:
def _ScrapeDataFromIPGLogFiles() -> tuple[dict[str, ResultType], MetricType]:
"""Scrapes data from IPG log files.
Returns:
A tuple (per_core_results, metrics). |output| is a dictionary containing
per-core results extracted from the IPG log files. |metrics| is a
dictionary mapping metrics found in the logs to all found data points.
"""
per_core_results = {}
metrics = {}
for logfile in logfiles:
results = AnalyzeIPGLogFile(logfile, skip_in_seconds)
results['log'] = logfile
(_, filename) = os.path.split(logfile)
(core, _) = os.path.splitext(filename)
prefix = 'PowerLog_'
if core.startswith(prefix):
core = core[len(prefix):]
per_core_results[core] = results
for key, value in results.items():
if key in ('samples', 'log'):
continue
metrics.setdefault(key, []).append(value)
return per_core_results, metrics
def _CalculateSummaryStatistics(metrics: MetricType) -> SummaryType:
"""Calculates summary statistics for the given metrics.
Args:
metrics: A dictionary mapping metrics to lists of data points.
Returns:
A dictionary mapping the same metrics in |metrics| to dicts containing
the 'mean' and 'stdev' for the metric.
"""
summary = {}
for key, data in metrics.items():
assert data and len(data) > 1
n = len(data)
if outliers > 0:
assert outliers * 2 < n
data.sort()
data = data[outliers:(n - outliers)]
n = len(data)
logging.debug('%s: valid samples = %d', key, n)
mean = sum(data) / float(n)
ss = sum((x - mean)**2 for x in data)
stdev = (ss / float(n))**0.5
summary[key] = {
'mean': mean,
'stdev': stdev,
}
return summary
assert len(logfiles) > 1
output, metrics = _ScrapeDataFromIPGLogFiles()
summary = _CalculateSummaryStatistics(metrics)
output['summary'] = summary
if output_json:
with open(output_json, 'w', encoding='utf-8') as json_file:
json_file.write(json.dumps(output, indent=4))
return summary
|
