Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / auto_parallel / sapp_nd / nd / debug.py: 92%
719 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-25 05:45 +0800
1# Copyright 2025-2026 Huawei Technologies Co., Ltd
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ============================================================================
15"""Debugging utilities"""
17import os
18import colorsys
19import csv
20from contextlib import contextmanager
21from enum import Enum, auto
22from pathlib import Path
23from math import isnan, sqrt
24from typing import Any, Iterator, Optional
26import pandas as pd
27import matplotlib.pyplot as plt
28import matplotlib.colors as mc
29from matplotlib.font_manager import FontProperties
30from matplotlib.backends.backend_pdf import PdfPages
31from scipy.stats import pearsonr
32import yaml
34from hyper_parallel.auto_parallel.sapp_nd.nd.logger import logger
35import hyper_parallel.auto_parallel.sapp_nd.nd.dimensions as Dim
36from hyper_parallel.auto_parallel.sapp_nd.nd.recompute_dimension import read_recompute_modes
38# The column a configuration's activation checkpoint mode goes in, written by a
39# search with a recompute dimension and read from a measured CSV that states it.
40RECOMPUTE_COLUMN = "recompute"
42# The results table's recompute rows: the activation checkpoint mode a
43# configuration runs, and the plan per layer that names the exceptions to it.
44MODE_ROW = "RMOD"
45LAYER_ROW = "RLAYER"
46# Rows whose cells hold text, which carries no shading and no largest value.
47_TEXT_ROWS = (MODE_ROW, LAYER_ROW)
50def _recompute_of(row: dict) -> Optional[str]:
51 """Take a measured row's activation checkpoint mode out of it, None where the CSV states none."""
52 stated = (row.pop(RECOMPUTE_COLUMN, None) or "").strip()
53 if not stated:
54 return None
55 modes = read_recompute_modes(stated, f"the {RECOMPUTE_COLUMN} column")
56 if len(modes) != 1:
57 raise ValueError(f"the {RECOMPUTE_COLUMN} column: a measured run ran one mode, not {stated!r}")
58 return modes[0]
61def _has_recompute(entries: list) -> bool:
62 """Whether any entry's configuration states an activation checkpoint mode."""
63 return any(getattr(entry[0], "recompute", None) for entry in entries)
66# Where the CSVs and the plots go. Defaults to a directory beside this file,
67# which is unwritable in an installed package and mixes consecutive runs
68# together, so ``run_nd -o`` overrides it.
69_OUTPUT_DIR = None
70# The CSVs Debug.write appends a row a candidate to.
71DEBUG_CSVS = ("debug.csv", "debug_mem.csv")
74def set_output_dir(path: Optional[str], fresh: bool = False) -> None:
75 """Send the debug artifacts to *path* instead of the package directory.
77 Args:
78 path: the run's output folder, or None for the package's own.
79 fresh: remove the debug CSVs an earlier run left there. ``Debug.write`` appends and writes the column
80 titles only into a new file, so a run into a folder another run wrote would add its rows under that
81 run's. The process that starts a run passes True; the pool's workers, which append to the files
82 the run started, do not.
83 """
84 global _OUTPUT_DIR # pylint: disable=global-statement
85 _OUTPUT_DIR = str(path) if path else None
86 if _OUTPUT_DIR:
87 os.makedirs(_OUTPUT_DIR, exist_ok=True)
88 if fresh:
89 for name in DEBUG_CSVS:
90 stale = os.path.join(output_dir(), name)
91 if os.path.isfile(stale):
92 os.remove(stale)
95def output_dir() -> str:
96 """Return the directory the debug artifacts are written to."""
97 if _OUTPUT_DIR:
98 return _OUTPUT_DIR
99 return os.path.join(os.path.dirname(os.path.abspath(__file__)), "output")
102class PerfParts(Enum):
103 """decomposition of performance"""
105 FW_COMPUTE = auto()
106 BW_COMPUTE = auto()
107 RECOMPUTE = auto()
108 # FSDP's gathers and reduce-scatters, and a run's other DP traffic.
109 DP_COMM = auto()
110 # The all-reduce among the copies of an FSDP shard: HSDP's once a step,
111 # every gradient's where FSDP shards nothing.
112 DP_REDUCE = auto()
113 MP_COMM = auto()
114 EP_COMM = auto()
115 CP_COMM = auto()
116 PP_COMM = auto()
117 BUBBLE = auto()
118 TOTAL = auto()
119 MEMORY = auto()
121 def __str__(self):
122 return self.name
124 def short_name(self):
125 """Returns short component name"""
126 name = "Perf"
127 if self == self.FW_COMPUTE:
128 name = "FW"
129 elif self == self.BW_COMPUTE:
130 name = "BW"
131 elif self == self.RECOMPUTE:
132 name = "Rec"
133 elif self == self.DP_COMM:
134 name = "DP"
135 elif self == self.DP_REDUCE:
136 name = "AR"
137 elif self == self.MP_COMM:
138 name = "MP"
139 elif self == self.EP_COMM:
140 name = "EP"
141 elif self == self.CP_COMM:
142 name = "CP"
143 elif self == self.PP_COMM:
144 name = "P2P"
145 elif self == self.BUBBLE:
146 name = "BBL"
147 elif self == self.MEMORY:
148 name = "MEM"
149 return name
152class RealParts(Enum):
153 """decomposition of performance"""
155 COMP = auto()
156 DP_WAIT = auto()
157 MP_WAIT = auto()
158 EP_WAIT = auto()
159 CP_WAIT = auto()
160 PP_WAIT = auto()
161 IDLE = auto()
162 TOTAL = auto()
164 def __str__(self):
165 return self.name.lower()
168# An optional column of a classified CSV, and the key it is read into: the
169# trainer's own step time in milliseconds, measured on steps the profiler left
170# alone. It is not a RealParts member, so every consumer that walks that enum
171# ignores it. ``examples/training_demo/sweep_qwen3_5_moe.py`` writes the column
172# and spells the name again rather than importing it, because that launcher
173# runs on the cluster's control node with nothing but the standard library.
174TRAINER_STEP = "step_trainer"
177class MemParts(Enum):
178 """decomposition of memory"""
180 TOTAL = auto()
182 def __str__(self):
183 return self.name
186class Debug:
187 """Debugging tools"""
189 def __init__(
190 self,
191 parallel_dimensions,
192 info_type,
193 enable=True,
194 output_file="debug.csv",
195 ):
196 self.enable = enable
197 if self.enable:
198 self.parallel_dimensions = parallel_dimensions
199 self.info = {p: 0 for p in info_type}
200 self.output_file = os.path.join(output_dir(), output_file)
202 def is_enabled(self):
203 """Check whether debugging is enabled"""
204 return self.enable
206 def column_titles(self):
207 """Parameters to debug"""
208 titles = self.parallel_dimensions.keys() + list(self.info.keys())
209 titles = [str(t) for t in titles]
210 return ",".join(titles) + "\n"
212 def values(self):
213 """values debugged"""
214 str_dims = [str(v) for v in self.parallel_dimensions.values()]
215 str_score = [str(int(v)) for v in self.info.values()]
216 return ",".join(str_dims + str_score) + "\n"
218 def write(self):
219 """Append this row to the CSV, under the column titles where the file is new.
221 Inside :func:`holding_open` the file stays open for the next row.
222 """
223 if self.enable:
224 logger.info("debug written")
225 held = _HELD.files.get(self.output_file)
226 if held is not None:
227 held.write(self.values())
228 return
229 os.makedirs(os.path.dirname(self.output_file), exist_ok=True)
230 is_new = not os.path.exists(self.output_file)
231 outfile = open(self.output_file, "a", encoding="utf-8") # pylint: disable=consider-using-with
232 if _HELD.depth:
233 _HELD.files[self.output_file] = outfile
234 try:
235 if is_new:
236 outfile.write(self.column_titles())
237 outfile.write(self.values())
238 finally:
239 if not _HELD.depth:
240 outfile.close()
243class _HeldFiles:
244 """The debug CSVs a search holds open while it writes them, by path."""
246 def __init__(self) -> None:
247 """No block open and no file held."""
248 self.depth = 0
249 self.files = {}
252_HELD = _HeldFiles()
255@contextmanager
256def holding_open() -> Iterator[None]:
257 """Keep each debug CSV written inside the block open until the block ends.
259 A search writes a row a candidate, and opening and closing the file for
260 each row cost 3.3 s of the Qwen3.5 example's 14507 rows in a temporary
261 folder and 6.5 s inside OneDrive. The files are closed, so whole on
262 disk, when the outermost block ends; a row written outside any block
263 still opens and closes its file.
264 """
265 _HELD.depth += 1
266 try:
267 yield
268 finally:
269 _HELD.depth -= 1
270 if not _HELD.depth:
271 while _HELD.files:
272 _HELD.files.popitem()[1].close()
275def pastel(color, l_delta=0.0, lbl=None, sat=None):
276 "Pastel (lighter) color of the input"
277 if color == "white":
278 return (1.0, 1.0, 1.0)
279 if color == "black":
280 return (0.5, 0.5, 0.5)
281 try:
282 color = mc.cnames[color]
283 except KeyError:
284 pass
285 color_hls = colorsys.rgb_to_hls(*mc.to_rgb(color))
286 lgt = 0.7
287 if lbl is not None:
288 lgt = lbl
289 lgt = lgt + l_delta
291 if sat is None:
292 sat = 0.6
293 return colorsys.hls_to_rgb(color_hls[0], lgt, sat)
296def near_white(color, ratio):
297 "Very light color of the input for background"
298 rgb = mc.to_rgb(color)
299 if rgb is None:
300 return "white"
301 (red, green, blue) = rgb
302 red += (1 - red) * ratio
303 green += (1 - green) * ratio
304 blue += (1 - blue) * ratio
305 return (red, green, blue)
308def dim_color(dim, default="black"):
309 """Color of parallel dimensions for plot"""
310 color = {
311 Dim.DP: "orange",
312 Dim.OP: "orange",
313 Dim.TP: "red",
314 Dim.EP: "blue",
315 Dim.CP: "teal",
316 Dim.PP: "green",
317 Dim.VPP: "green",
318 Dim.MBN: "green",
319 }
320 try:
321 dim = Dim.get_dim(dim)
322 if dim in color:
323 return color[dim]
324 return default
325 except ValueError:
326 return default
329def gen_colors(categories):
330 """Color of each time component"""
331 compute_color = "purple"
332 idle_color = "grey"
333 col_d = {
334 str(PerfParts.FW_COMPUTE): pastel(compute_color, -0.2),
335 str(PerfParts.BW_COMPUTE): pastel(compute_color, -0.1),
336 str(PerfParts.RECOMPUTE): pastel(compute_color),
337 str(PerfParts.DP_COMM): pastel(dim_color(Dim.DP)),
338 str(PerfParts.DP_REDUCE): pastel(dim_color(Dim.DP), -0.15),
339 str(PerfParts.MP_COMM): pastel(dim_color(Dim.TP), -0.1),
340 str(PerfParts.EP_COMM): pastel(dim_color(Dim.EP)),
341 str(PerfParts.CP_COMM): pastel(dim_color(Dim.CP)),
342 str(PerfParts.PP_COMM): pastel(dim_color(Dim.PP)),
343 str(PerfParts.BUBBLE): pastel(dim_color(Dim.PP), -0.15),
344 "IDLE": idle_color,
345 "COMPUTATION": pastel(compute_color, -0.2),
346 }
347 return [col_d.get(cat) for cat in categories]
350def set_twin_handles(ax1, data_frame, dbg_cols):
351 """Set legend for estimation and real"""
352 handle1, label1 = ax1.get_legend_handles_labels()
353 ax2 = plt.twinx()
354 data_frame[dbg_cols].plot.bar(
355 stacked=True,
356 sharex=True,
357 ax=ax2,
358 position=0,
359 color=gen_colors(dbg_cols),
360 width=0.4,
361 rot=0,
362 )
364 handle2, label2 = ax2.get_legend_handles_labels() # type: ignore
365 # Patches compare by identity, so testing the handle admitted every one of
366 # them while the labels deduplicated by name: the legend then drew more
367 # swatches than it had names and mislabelled every entry past the first
368 # shared one. Deduplicate on the label and keep its handle in step.
369 for handle, lbl in zip(handle2, label2):
370 if lbl not in label1:
371 handle1.append(handle)
372 label1.append(lbl)
373 handles = handle1
374 labels = label1
375 plt.legend(handles, labels, loc="upper left", bbox_to_anchor=(1, 1))
376 leg = ax2.get_legend()
377 pp_color = gen_colors(["PP_COMM"])[0]
378 leg.legend_handles[-1].set_facecolor(pp_color) # type: ignore
381# The measured parts of the comparison plot, named by the ND part each is set
382# against: FSDP waits count as DP and sequence-parallel waits as MP, as they
383# do in the correlations.
384MEASURED_BARS = ("COMPUTATION", "DP_COMM", "MP_COMM", "EP_COMM", "CP_COMM", "BUBBLE")
387def measured_bars(waits: dict, plot_idle: bool = False) -> list:
388 """Return one configuration's measured parts in ``MEASURED_BARS`` order.
390 Every wait column of the classified CSV lands in one bar, ``op_wait`` in
391 DP's and ``sp_wait`` in MP's as ``real_in_parts`` counts them, so with idle
392 the stack adds up to the measured step. A part the CSV lacks is zero.
394 Args:
395 waits: A configuration's measured parts, as ``get_comm_classified_data`` reads them.
396 plot_idle: Whether to append the idle remainder.
397 """
398 def part(name: str) -> float:
399 """The measured part *name*, zero when the CSV does not have it."""
400 return waits.get(name) or 0.0
402 bars = [
403 part("comp"),
404 part("dp_wait") + part("op_wait"),
405 part("mp_wait") + part("sp_wait"),
406 part("ep_wait"),
407 part("cp_wait"),
408 part("BUBBLE"),
409 ]
410 if plot_idle:
411 bars.append(part("IDLE"))
412 return bars
415def _cell_number(text):
416 """Read one cell of the degree table as a number.
418 A boolean dimension such as SP prints as True or False, which float()
419 refuses, so every search that varied SP failed as its plot was drawn.
420 """
421 if text in ("True", "False"):
422 return float(text == "True")
423 return float(text)
426class Plot:
427 """plot ND top configs"""
429 title: str
430 col_title: list[str]
431 row_title: list[str]
432 cell_text: list[list[str]]
433 data: list[tuple]
434 dbg_cols: list[str]
435 top: int
437 def __init__(
438 self,
439 title: Optional[str],
440 rows: list,
441 debug_parts: list,
442 top: Optional[int] = None,
443 show_top: bool = False,
444 ) -> None:
445 """Initialize the plot table, optionally displaying each configuration's rank."""
446 self.title = title
447 self.top = top if top is not None else 20
448 self.row_title = rows + ["MEM"]
449 if show_top:
450 self.row_title.insert(0, "TOP")
451 self.dbg_cols = list(map(str, debug_parts))
452 self.col_title = []
453 self.cell_text = []
454 self.data = []
456 def make_table(self):
457 """Make the table below the plot with each parallelism degree and the recompute rows.
459 A recompute row holds text rather than a degree, so it keeps one
460 shade and its cells grow with the lines its plan takes.
461 """
462 self.cell_text = list(map(list, zip(*self.cell_text))) # transpose
463 max_rows = [
464 None if title in _TEXT_ROWS else max(map(_cell_number, cells))
465 for title, cells in zip(self.row_title, self.cell_text)
466 ]
467 the_table = plt.table(
468 cellText=self.cell_text,
469 rowLabels=self.row_title,
470 colLabels=self.col_title,
471 cellLoc="center",
472 loc="bottom",
473 )
474 row_colors = list(map(dim_color, self.row_title))
475 for row in range(len(self.row_title)):
476 lines = max(str(text).count("\n") + 1 for text in self.cell_text[row])
477 cell = the_table[row + 1, -1]
478 cell.set_edgecolor("none")
479 cell.get_text().set_color(row_colors[row])
480 cell.set_text_props(fontproperties=FontProperties(weight="bold"))
481 cell.set_height(cell.get_height() * lines)
482 for col in range(len(self.cell_text[0])):
483 self._shade_cell(the_table[row + 1, col], row, max_rows[row], lines, row_colors[row])
485 for col in range(len(self.cell_text[0])):
486 the_table[0, col].set_edgecolor("none")
488 the_table.scale(xscale=1, yscale=1.2) # +len(rows)/5)
490 def _shade_cell(self, cell: Any, row: int, largest: Optional[float], lines: int, color: Any) -> None:
491 """Shade one cell of the table by its value against the row's largest.
493 Args:
494 cell: The cell to shade.
495 row: Its row, which names the degree or the recompute setting.
496 largest: The largest value the row holds; ``None`` for a row of
497 text, which keeps the row's lightest shade.
498 lines: The lines the row's tallest cell takes.
499 color: The row's colour.
500 """
501 cell.set_height(cell.get_height() * lines)
502 if largest is None:
503 ratio = 1
504 else:
505 try:
506 ratio = 1 - (_cell_number(str(cell.get_text().get_text())) / largest)
507 except ZeroDivisionError:
508 ratio = 0
509 if self.row_title[row] == LAYER_ROW:
510 cell.set_text_props(ha="center", fontfamily="monospace")
511 logger.debug(
512 "tmax = %s, ratio = %f, col=%s, newcolor=%s",
513 str(largest),
514 ratio,
515 str(mc.to_rgb(color)),
516 str(near_white(color, ratio)),
517 )
518 cell.set_facecolor(near_white(pastel(color), ratio))
519 cell.set_edgecolor("none")
521 def close(self, output_path, filename):
522 """Plot closing statements"""
523 plt.gca().set_xticklabels([])
524 plt.gca().set_yticklabels([])
525 plt.xlim([-0.5, len(self.data) - 0.5])
526 if self.title is not None:
527 plt.title(self.title)
528 plt.subplots_adjust(left=0.1, bottom=0.047 * (2 + len(self.row_title)))
529 plotfile = os.path.join(output_path, filename + ".pdf")
530 plt.savefig(plotfile, bbox_inches="tight")
531 plt.clf()
533 def parse_data(
534 self,
535 configs_estimated,
536 **kwargs,
537 ):
538 """Parse test data for plot"""
539 real_data = kwargs.get("real_data", None)
540 plot_idle = kwargs.get("plot_idle", False)
541 include_all = kwargs.get("include_all", False)
542 recompute_plans = kwargs.get("recompute_plans")
543 min_e = configs_estimated[0][2]
544 i = 0
545 for index, cfg_e in enumerate(configs_estimated):
546 cells = cfg_e[0].values() + self._recompute_cells(cfg_e[0], recompute_plans) + [cfg_e[1]]
547 if self.row_title[0] == "TOP":
548 cells.insert(0, cfg_e[0].rank or index + 1)
549 self.cell_text.append(cells)
550 self.col_title.append("")
551 try:
552 self.data.append(
553 tuple([cfg_e[0], cfg_e[2], cfg_e[3]] + cfg_e[4])
554 )
555 if real_data is not None:
556 logger.info(cfg_e[5])
557 real_data.append(tuple(measured_bars(cfg_e[5], plot_idle)))
558 except IndexError:
559 score = cfg_e[2]
560 if not include_all and (
561 i >= self.top or (min_e is not None and score > min_e * 20)
562 ):
563 self.cell_text.pop()
564 break
565 self.data.append(tuple([cfg_e[0], score] + cfg_e[3]))
566 i += 1
568 def _recompute_cells(self, config: Any, recompute_plans: Optional[dict]) -> list:
569 """The recompute cells of one configuration's column, empty where the table has no such row.
571 Args:
572 config: The configuration the column states.
573 recompute_plans: ``{configuration: (mode, layers)}`` from a
574 choice among named modes per layer; ``None`` where the search
575 made none.
577 Returns:
578 The mode alone, the mode and its plan per layer, or nothing.
579 """
580 if MODE_ROW not in self.row_title:
581 return []
582 mode, layers = (recompute_plans or {}).get(config, (getattr(config, "recompute", None), None))
583 cells = [mode.upper() if mode else "-"]
584 if LAYER_ROW in self.row_title:
585 cells.append(
586 yaml.safe_dump(layers, default_flow_style=False, sort_keys=False).strip()
587 if layers is not None else "-"
588 )
589 return cells
592def top_plot_configs(configs_estimated: list, max_num: Optional[int] = None) -> list:
593 """Return the configurations the standard ND plot would include."""
594 if not configs_estimated:
595 return []
596 top = max_num if max_num is not None else 20
597 if top <= 0:
598 return []
599 best_score = configs_estimated[0][2]
600 selected = []
601 for config in configs_estimated:
602 if len(selected) >= top or config[2] > best_score * 20:
603 break
604 selected.append(config)
605 return selected
608def plot_nd(
609 configs_estimated: list,
610 output_path: str,
611 debug_parts: list,
612 title: Optional[str] = None,
613 max_num: Optional[int] = None,
614 include_all: bool = False,
615 top_result_count: Optional[int] = None,
616 recompute_plans: Optional[dict] = None,
617) -> None:
618 """Plot estimation with optional recompute rows and a divider before additions.
620 Args:
621 configs_estimated: Ranked configurations with memory, score and its parts.
622 TOP uses the rank stored on each configuration, falling back to its
623 position in the supplied list when it has not been ranked.
624 output_path: Directory in which to save results.pdf.
625 debug_parts: Performance components to plot.
626 title: Optional plot title.
627 max_num: Maximum number of configurations in the normal plot.
628 include_all: Include every supplied configuration without normal plot limits.
629 top_result_count: Number of normal results preceding the additions.
630 recompute_plans: ``{configuration: (mode, layers)}``, the activation
631 checkpoint mode and the plan per layer a choice among named modes
632 made. ``None`` omits the plan row; a configuration that states a
633 mode of its own still shows it.
634 """
635 plot = Plot(
636 title, configs_estimated[0][0].keys(), debug_parts,
637 top=max_num, show_top=True,
638 )
639 if _has_recompute(configs_estimated) or recompute_plans is not None:
640 plot.row_title.insert(-1, MODE_ROW)
641 if recompute_plans is not None:
642 plot.row_title.insert(-1, LAYER_ROW)
643 plot.parse_data(configs_estimated, include_all=include_all, recompute_plans=recompute_plans)
645 data_frame = pd.DataFrame(
646 plot.data, columns=(["config", "estim"] + plot.dbg_cols)
647 )
648 axis = data_frame[plot.dbg_cols].plot.bar(
649 stacked=True, color=gen_colors(plot.dbg_cols), width=0.4, rot=0
650 )
651 axis.set_ylim(ymin=1)
652 axis.legend(loc="upper left", bbox_to_anchor=(1, 1))
653 if top_result_count is not None and 0 < top_result_count < len(plot.data):
654 axis.axvline(top_result_count - 0.5, color="black", linewidth=1.5)
656 plot.make_table()
657 plot.close(output_path, "results")
660def plot_recompute(
661 groups: list,
662 output_path: str,
663 debug_parts: list,
664 filename: str,
665 title: Optional[str] = None,
666 columns_per_page: int = 12,
667) -> None:
668 """Write a summary of every selected recompute plan, then a detail page per strategy.
670 Each bar is the cost model's estimate of the step under that plan, split
671 into its parts, so the step between two bars is what the cost model
672 expects the extra recomputed layer to cost.
674 Args:
675 groups: The ranked parallel configurations and the records of their
676 selected plans, as ``recompute_plot_data`` prepares them.
677 output_path: Directory in which to save the PDF.
678 debug_parts: The performance components in each record's parts.
679 filename: The PDF's basename, without its extension.
680 title: Optional description of the model and the selection.
681 columns_per_page: The most plans on one detail page. The summary
682 holds every selected plan, with a divider between strategies.
684 Raises:
685 ValueError: The page size is not positive, or a record's score parts
686 do not cover every component.
687 """
688 if columns_per_page <= 0:
689 raise ValueError("columns_per_page must be positive")
690 if not groups:
691 return
692 os.makedirs(output_path, exist_ok=True)
693 heading = title or "ND recompute"
694 with PdfPages(os.path.join(output_path, filename + ".pdf")) as document:
695 _save_page(document, _recompute_summary(groups, debug_parts, heading))
696 for config, records in groups:
697 _save_detail_pages(document, config, records, debug_parts, heading, columns_per_page)
700def _save_page(document: Any, figure: Any) -> None:
701 """Add one figure to the PDF and release it."""
702 document.savefig(figure, bbox_inches="tight")
703 plt.close(figure)
706def _recompute_summary(groups: list, debug_parts: list, heading: str) -> Any:
707 """The page that holds every selected plan, the strategies it covers named where it has none."""
708 plans = [(config, record) for config, records in groups for record in records]
709 labels = [f"{config.rank}.{index + 1}" for config, records in groups for index in range(len(records))]
710 page_title = f"{heading}\nSummary | {len(plans)} plans across {len(groups)} configurations"
711 if not plans:
712 return _empty_recompute_page([config for config, _ in groups], page_title)
713 figure = _recompute_page(plans, debug_parts, page_title, labels)
714 empty = [str(config.rank) for config, records in groups if not records]
715 if empty:
716 figure.text(0.5, 0.01, "No qualifying plans for TOP " + ", ".join(empty), ha="center", fontsize=8)
717 return figure
720def _save_detail_pages(
721 document: Any, config: Any, records: list, debug_parts: list, heading: str, columns_per_page: int
722) -> None:
723 """Add one strategy's plans to the PDF, paginated, or a page saying it has none."""
724 if not records:
725 _save_page(document, _empty_recompute_page([config], f"{heading}\nTOP {config.rank}"))
726 return
727 for start in range(0, len(records), columns_per_page):
728 page = records[start:start + columns_per_page]
729 page_title = (
730 f"{heading}\n"
731 f"TOP {config.rank} | plans {start + 1}-{start + len(page)} of {len(records)}"
732 )
733 labels = [str(start + index + 1) for index in range(len(page))]
734 _save_page(document, _recompute_page([(config, record) for record in page], debug_parts, page_title, labels))
737def _empty_recompute_page(configs: list, title: str) -> Any:
738 """Say that a selection is empty, and name the strategies it covers."""
739 figure = plt.figure(figsize=(12, max(5, 3 + 0.3 * len(configs))))
740 figure.text(0.5, 0.8, title, ha="center", fontsize=12)
741 figure.text(0.5, 0.6, "No recompute plan satisfies the selection for these configurations.", ha="center")
742 descriptions = [
743 f"TOP {config.rank}: " + " ".join(f"{dim}={value}" for dim, value in config.dims_val.items())
744 for config in configs
745 ]
746 figure.text(0.5, 0.45, "\n".join(descriptions), ha="center", va="top")
747 return figure
750def _recompute_page(plans: list, debug_parts: list, title: str, labels: list) -> Any:
751 """Draw one page of plans, with a layer table under the chart and a divider between strategies."""
752 if any(len(record["parts"]) != len(debug_parts) for _, record in plans):
753 raise ValueError("recompute plot records must include every performance component")
754 rows = ["TOP"] + plans[0][0].keys() + ["RFULL", MODE_ROW, LAYER_ROW, "MEM", "SCORE"]
755 cells = list(map(list, zip(*[_recompute_column(config, record) for config, record in plans])))
756 lines = [max(str(text).count("\n") + 1 for text in row) for row in cells]
757 heights = [0.20] + [0.18 * count + 0.06 for count in lines]
758 table_height = sum(heights)
759 figure = plt.figure(figsize=(max(12, 1.3 * len(plans) + 3), 4.5 + table_height))
760 grid = figure.add_gridspec(2, 1, height_ratios=[3.5, table_height], hspace=0.10)
761 axis = _recompute_chart(figure.add_subplot(grid[0]), plans, debug_parts, title, labels)
762 table_axis = figure.add_subplot(grid[1])
763 _recompute_table(table_axis, rows, cells, labels, heights)
764 for index in range(1, len(plans)):
765 if plans[index][0] != plans[index - 1][0]:
766 axis.axvline(index - 0.5, color="black", linewidth=1.5)
767 table_axis.axvline(index / len(plans), color="black", linewidth=1.5, zorder=10)
768 figure.subplots_adjust(left=0.09, right=0.84, top=0.88, bottom=0.035)
769 return figure
772def _recompute_column(config: Any, record: dict) -> list:
773 """One plan's column: its strategy, how many layers it fully recomputes, its plan, memory and estimate."""
774 layers = record["layers"]
775 return [str(config.rank)] + config.values() + [
776 str(record["full_layers"]) if record["full_layers"] is not None else "-",
777 record["mode"].upper() if record["mode"] else "-",
778 yaml.safe_dump(layers, default_flow_style=False, sort_keys=False).strip() if layers is not None else "-",
779 f"{record['memory']:.0f} MB",
780 f"{record['score']:.6g}",
781 ]
784def _recompute_chart(axis: Any, plans: list, debug_parts: list, title: str, labels: list) -> Any:
785 """Stack each plan's estimated parts on one bar, and return the axis."""
786 parts = list(map(str, debug_parts))
787 frame = pd.DataFrame([record["parts"] for _, record in plans], columns=parts, index=labels)
788 frame.plot.bar(ax=axis, stacked=True, color=gen_colors(parts), width=0.65, rot=0)
789 axis.set_xlim(-0.5, len(plans) - 0.5)
790 axis.set_ylim(
791 min(0, float(frame.clip(upper=0).sum(axis=1).min()) * 1.05),
792 max(1, float(frame.clip(lower=0).sum(axis=1).max()) * 1.05),
793 )
794 axis.set_ylabel("Estimated step (lower is better)")
795 axis.set_xlabel("")
796 axis.legend(loc="upper left", bbox_to_anchor=(1, 1), fontsize=8)
797 axis.set_title(title, fontsize=11)
798 return axis
801def _recompute_table(table_axis: Any, rows: list, cells: list, labels: list, heights: list) -> None:
802 """Draw the table of plans under the chart, each row shaded like its dimension."""
803 table_height = sum(heights)
804 table_axis.axis("off")
805 table = table_axis.table(cellText=cells, rowLabels=list(map(str, rows)), colLabels=labels,
806 cellLoc="center", loc="center", bbox=[0, 0, 1, 1])
807 table.auto_set_font_size(False)
808 table.set_fontsize(8)
809 for (row, column), cell in table.get_celld().items():
810 cell.set_height(heights[row] / table_height)
811 cell.set_edgecolor("white")
812 if row == 0:
813 cell.set_facecolor("#eeeeee")
814 continue
815 row_title = rows[row - 1]
816 if column == -1:
817 cell.get_text().set_color(dim_color(row_title))
818 cell.get_text().set_fontweight("bold")
819 else:
820 cell.set_facecolor(near_white(pastel(dim_color(row_title)), 0.9))
821 if row_title == LAYER_ROW:
822 cell.get_text().set_fontfamily("monospace")
823 cell.get_text().set_ha("center" if cells[row - 1][column] == "{}" else "left")
826def _score_parts() -> list:
827 """The parts a score splits into, in the order the debugger fills them."""
828 return [part for part in PerfParts if part not in {PerfParts.TOTAL, PerfParts.MEMORY}]
831def _make_parent(path: str) -> None:
832 """Create the directory *path* is to be written in, when it is missing."""
833 directory = os.path.dirname(path)
834 if directory:
835 os.makedirs(directory, exist_ok=True)
838def write_ranking_csv(scored_space: list, path: str) -> None:
839 """Write a search's configurations in ND's order, best first.
841 One row per configuration that fits memory: its rank, its degrees, the
842 activation checkpoint mode where the search had a recompute dimension,
843 the peak memory in MB, the score and the parts the score splits into,
844 which are blank when the search ran without debug output. Scores keep
845 full precision so that a reader can tell a tie from a near miss.
847 Args:
848 scored_space: ``(config, memory, score, parts)`` entries, as
849 ``ParallelizeLayer.order_search_space`` sorts them. Stored ranks
850 are preserved; unranked configurations use their input positions.
851 path: CSV file to write; its directory is created when missing.
852 """
853 parts = _score_parts()
854 _make_parent(path)
855 dims = [str(dim) for dim in scored_space[0][0].keys()] if scored_space else []
856 moded = _has_recompute(scored_space)
857 with open(path, "w", newline="", encoding="utf-8") as handle:
858 writer = csv.writer(handle)
859 writer.writerow(["rank"] + dims + ([RECOMPUTE_COLUMN] if moded else []) + ["memory_mb", "score"]
860 + [str(part) for part in parts])
861 for rank, (config, memory, score, values) in enumerate(scored_space, start=1):
862 split = [repr(float(value)) for value in values] if values else [""] * len(parts)
863 writer.writerow([config.rank or rank] + config.values() + ([config.recompute] if moded else [])
864 + [memory, repr(float(score))] + split)
867def write_estimates_csv(configs_estimated: list, path: str) -> None:
868 """Write ND's estimate of every configuration of a classified comparison.
870 One row per measured configuration, in the comparison's order: its
871 degrees, the activation checkpoint mode where the measured CSV states
872 one, the measured step, ND's peak memory in MB, its score and the parts
873 of the score. The plots show these only as bars; a sweep needs ND's
874 memory as a number, to set it beside the peak the trainer logged.
876 Args:
877 configs_estimated: ``(config, peak_mem, real_time, score, parts,
878 real_parts)`` entries, as ``ParallelizeLayer.compare_with_csv``
879 returns them.
880 path: CSV file to write; its directory is created when missing.
881 """
882 parts = _score_parts()
883 _make_parent(path)
884 dims = [str(dim) for dim in configs_estimated[0][0].keys()] if configs_estimated else []
885 moded = _has_recompute(configs_estimated)
886 with open(path, "w", newline="", encoding="utf-8") as handle:
887 writer = csv.writer(handle)
888 writer.writerow(dims + ([RECOMPUTE_COLUMN] if moded else []) + ["time", "memory_mb", "score"]
889 + [str(part) for part in parts])
890 for config, memory, step, score, values, _ in configs_estimated:
891 split = [repr(float(value)) for value in values[:len(parts)]]
892 writer.writerow(config.values() + ([config.recompute] if moded else [])
893 + [step, memory, repr(float(score))] + split)
896def busy_time(entry: tuple) -> float:
897 """The measured step of a comparison entry less its idle remainder."""
898 return entry[2] - (entry[5].get("IDLE") or 0.0)
901def plot_vs_real(
902 configs_estimated, csv_f, output_path, debug_parts, title=None
903):
904 """Plot estimation vs real global time"""
905 plot = Plot(title, configs_estimated[0][0].keys(), debug_parts)
906 plot.parse_data(configs_estimated)
908 data_frame = pd.DataFrame(
909 plot.data, columns=(["config", "Real", "estim"] + plot.dbg_cols)
910 )
911 ax1 = data_frame["Real"].plot.bar(
912 position=1.1, width=0.4, secondary_y="real", color="grey", rot=0
913 )
915 set_twin_handles(ax1, data_frame, plot.dbg_cols)
916 plot.make_table()
917 plot.close(output_path, Path(os.path.basename(csv_f)).stem)
920def plot_vs_real_comm_classified(
921 configs_estimated,
922 csv_f,
923 output_path,
924 debug_parts,
925 **kwargs,
926):
927 """Plot estimation vs real detailed time.
929 Written to ``<csv stem><suffix>.pdf`` in *output_path*: ``plot_idle`` adds
930 the measured idle remainder to the measured bars, and ``suffix`` (empty
931 by default) tells apart several plots of one CSV.
932 """
933 plot_idle = kwargs.get("plot_idle", False)
934 title = kwargs.get("title", None)
935 suffix = kwargs.get("suffix", "")
936 real_data = []
938 plot = Plot(title, configs_estimated[0][0].keys(), debug_parts)
939 plot.parse_data(
940 configs_estimated,
941 real_data=real_data,
942 plot_idle=plot_idle,
943 )
945 data_frame = pd.DataFrame(
946 plot.data, columns=(["config", "real", "estim"] + plot.dbg_cols)
947 )
948 real_cols = list(MEASURED_BARS)
949 if plot_idle:
950 real_cols.append("IDLE")
951 real_df = pd.DataFrame(real_data, columns=real_cols)
953 ax1 = real_df[real_cols].plot.bar(
954 stacked=True,
955 sharex=True,
956 position=1,
957 secondary_y="real",
958 color=gen_colors(real_cols),
959 width=0.4,
960 rot=0,
961 legend=False,
962 )
964 set_twin_handles(ax1, data_frame, plot.dbg_cols)
965 plot.make_table()
966 plot.close(
967 output_path,
968 Path(os.path.basename(csv_f)).stem + suffix,
969 )
972def correlation_topk(configs_estimated, csv_f):
973 """Computes correlation & top-k between real & estimation"""
974 times = []
975 estims = []
976 for _, _, time, score, _ in configs_estimated:
977 times.append(time)
978 estims.append(score)
979 correl = pearsonr(times, estims).statistic # type: ignore
980 if isnan(correl):
981 logger.critical(
982 "An input array is constant: %s or %s", str(times), str(estims)
983 )
984 topk = 0
985 for i, score in enumerate(estims):
986 if not score == min(estims[i:]):
987 break
988 topk += 1
989 if topk == 0:
990 for i, score in enumerate(estims):
991 if score == min(estims[i:]):
992 break
993 topk -= 1
995 logger.info("Correlation for file %s is: %.3f", csv_f, correl * 100)
996 return correl, topk
999def get_real_data(csv_f):
1000 """Read execution time of different configurations on a given csv file"""
1001 configs = []
1002 row_num = 0
1003 with open(csv_f, newline="", encoding="utf-8") as csv_file:
1004 rows = csv.DictReader(csv_file)
1005 for row in rows:
1006 row_num += 1
1007 logger.info(row)
1008 real_time = float(row.pop("time"))
1009 recompute = _recompute_of(row)
1010 config = []
1011 for dim_str, value in row.items():
1012 try:
1013 dim = Dim.get_dim(dim_str)
1014 logger.debug("%s : %s", str(dim), str(dim.from_str(value)))
1015 config.append((dim, dim.from_str(value)))
1016 except ValueError:
1017 pass
1018 configs.append((Dim.Dimensions(config, recompute=recompute), real_time))
1019 return configs, row_num
1022def get_diff_dims(csv_f):
1023 """Read execution time of different configurations on a given csv file"""
1024 dims = []
1025 data_frame = pd.read_csv(csv_f)
1026 for dim_str, degrees in data_frame.items():
1027 try:
1028 dim = Dim.get_dim(dim_str)
1029 diff_values = len(set(degrees))
1030 if diff_values > 1:
1031 dims.append(dim)
1032 except ValueError:
1033 pass
1034 return dims
1037def get_comm_classified_data(csv_f, plot_idle=False):
1038 """Read time components of different configurations on a given csv file.
1040 ``time`` is the mean of the PROFILED steps, which is what the instrument
1041 read and not what the model costs: on Ascend the profiler adds 0.33 to
1042 0.84 s a step at EP 1 and 0.01 to 0.10 at EP 8, which is enough to reverse
1043 a ranking. A CSV may therefore carry ``step_trainer`` as well, the
1044 trainer's own step time over steps it did not profile, and this reads it
1045 into the parts under the same name for whoever ranks the round.
1047 It is deliberately NOT substituted for ``time`` here. The parts carry the
1048 profiler's cost too, so the step less its parts goes negative against the
1049 honest total on 4 of the 15 points of the 2 October round, to -499 ms at
1050 EP 1, where against the profiled step the same residual is +96 to +454.
1051 Idle stays a profiled quantity until the parts have a correction of their
1052 own; only the total is honest enough to rank on.
1053 """
1054 configs = []
1055 with open(csv_f, newline="", encoding="utf-8") as csv_file:
1056 rows = csv.DictReader(csv_file)
1057 for row in rows:
1058 logger.info(row)
1059 time = float(row.pop("time"))
1060 trainer_step = (row.pop(TRAINER_STEP, "") or "").strip()
1061 recompute = _recompute_of(row)
1062 config = []
1063 comm_wait_time_classified = {}
1064 total_wait = 0
1066 for component, value_str in row.items():
1067 if "wait" in component:
1068 value_float = float(value_str)
1069 logger.info(
1070 "Comm_wait = %s, v = %f", component, value_float
1071 )
1072 comm_wait_time_classified[component] = value_float
1073 total_wait += value_float
1074 elif "comp" in component:
1075 value_float = float(value_str)
1076 logger.info("Computation = %f", value_float)
1077 comm_wait_time_classified["comp"] = value_float
1078 total_wait += value_float
1079 else:
1080 logger.info("d = %s, v = %s", component, value_str)
1081 dim = Dim.get_dim(component)
1082 config.append((dim, dim.from_str(value_str)))
1083 comm_wait_time_classified["BUBBLE"] = comm_wait_time_classified.get(str(RealParts.PP_WAIT))
1084 if trainer_step:
1085 comm_wait_time_classified[TRAINER_STEP] = float(trainer_step)
1086 if plot_idle:
1087 comm_wait_time_classified["IDLE"] = time - total_wait
1088 logger.info(
1089 "idle = total time - total waits = %.3f - %.3f",
1090 time,
1091 total_wait,
1092 )
1093 configs.append(
1094 (Dim.Dimensions(config, recompute=recompute), time, comm_wait_time_classified)
1095 )
1096 return configs
1099def estimation_in_real_parts(
1100 estimations_in_real_components, estimations, score
1101):
1102 """Transform the estimation components into the RealParts components for comparison with real time"""
1103 estimations_in_real_components[RealParts.TOTAL].append(score)
1104 estimations_in_real_components[RealParts.COMP].append(
1105 estimations[PerfParts.FW_COMPUTE.value - 1]
1106 + estimations[PerfParts.BW_COMPUTE.value - 1]
1107 + estimations[PerfParts.RECOMPUTE.value - 1]
1108 )
1109 estimations_in_real_components[RealParts.DP_WAIT].append(
1110 estimations[PerfParts.DP_COMM.value - 1]
1111 + estimations[PerfParts.DP_REDUCE.value - 1]
1112 )
1113 estimations_in_real_components[RealParts.MP_WAIT].append(
1114 estimations[PerfParts.MP_COMM.value - 1]
1115 )
1116 estimations_in_real_components[RealParts.CP_WAIT].append(
1117 estimations[PerfParts.CP_COMM.value - 1]
1118 )
1119 estimations_in_real_components[RealParts.EP_WAIT].append(
1120 estimations[PerfParts.EP_COMM.value - 1]
1121 )
1122 estimations_in_real_components[RealParts.PP_WAIT].append(
1123 estimations[PerfParts.BUBBLE.value - 1]
1124 + estimations[PerfParts.PP_COMM.value - 1]
1125 )
1126 return estimations_in_real_components
1129def real_in_parts(parts, real, time):
1130 """Transform the real time components into the RealParts components for comparison with estimation"""
1131 parts[RealParts.TOTAL].append(time)
1132 for part in RealParts:
1133 if part not in {RealParts.TOTAL, RealParts.IDLE}:
1134 if str(part) in real.keys():
1135 parts[part].append(real[str(part)])
1136 else:
1137 logger.warning(
1138 "part = %s not in real keys = %s", part, real.keys()
1139 )
1140 parts[part].append(0)
1142 op = "op_wait"
1143 if op in real.keys():
1144 parts[RealParts.DP_WAIT][-1] += real["op_wait"]
1146 sp = "sp_wait"
1147 if sp in real.keys():
1148 parts[RealParts.MP_WAIT][-1] += real["sp_wait"]
1150 return parts
1153def correlation_with_classified_comms(configs_estimated: list) -> tuple:
1154 """Computes correlation and distance between components time & estimation."""
1155 score_classified = {}
1156 time_classified = {}
1157 distances = {}
1159 for wait in RealParts:
1160 if wait not in {RealParts.IDLE}:
1161 score_classified[wait] = []
1162 time_classified[wait] = []
1163 distances[wait] = []
1165 topk = 0
1166 still_top_k = True
1168 for i, (_, _, time, score, values, real_values) in enumerate(
1169 configs_estimated
1170 ):
1171 if (
1172 still_top_k
1173 and score == (min(configs_estimated[i:], key=lambda t: t[3]))[3]
1174 ):
1175 topk += 1
1176 else:
1177 still_top_k = False
1179 score_classified = estimation_in_real_parts(
1180 score_classified, values, score
1181 )
1183 time_classified = real_in_parts(time_classified, real_values, time)
1185 square_distances_sum = 0
1186 for wait in RealParts:
1187 if wait not in {RealParts.TOTAL, RealParts.IDLE}:
1188 distance = (
1189 time_classified[wait][-1]
1190 / time_classified[RealParts.TOTAL][-1]
1191 - score_classified[wait][-1]
1192 / score_classified[RealParts.TOTAL][-1]
1193 )
1194 square_distances_sum += distance * distance
1195 distances[wait].append(abs(distance))
1196 distances[RealParts.TOTAL].append(sqrt(square_distances_sum))
1198 correls = {}
1199 for wait in RealParts:
1200 pearson_wait(correls, time_classified, score_classified, wait)
1201 return correls, distances, topk, len(configs_estimated)
1204def _share(value, total):
1205 """Fraction of a total, 0 when the total is 0"""
1206 return value / total if total else 0.0
1209def format_classified_comparison(configs_estimated: list) -> str:
1210 """Side-by-side measured and estimated parts of every configuration.
1212 Shares are over each side's own total. ND has no idle term, so the measured
1213 idle share is the part of the step the estimate does not account for.
1215 Args:
1216 configs_estimated: Entries of ``ParallelizeLayer.order_space_test_comm_classified``.
1218 Returns:
1219 One table per configuration: measured value and share, ND share, and their difference.
1220 """
1221 parts = [part for part in RealParts if part not in {RealParts.IDLE, RealParts.TOTAL}]
1222 lines = []
1223 for config, _, time, score, values, real_values in configs_estimated:
1224 estim = estimation_in_real_parts({part: [] for part in RealParts}, values, score)
1225 real = real_in_parts({part: [] for part in RealParts}, real_values, time)
1226 lines.append(f"{config}: measured {time:.6g}, ND score {score:.6g}")
1227 lines.append(f" {'part':8s} {'measured':>12s} {'share':>8s} {'ND share':>9s} {'diff':>8s}")
1228 for part in parts:
1229 real_share = _share(real[part][-1], time)
1230 estim_share = _share(estim[part][-1], score)
1231 lines.append(
1232 f" {str(part):8s} {real[part][-1]:12.6g} {real_share:8.1%} {estim_share:9.1%} "
1233 f"{real_share - estim_share:+8.1%}"
1234 )
1235 idle = time - sum(real[part][-1] for part in parts)
1236 lines.append(f" {str(RealParts.IDLE):8s} {idle:12.6g} {_share(idle, time):8.1%} {'-':>9s}")
1237 return "\n".join(lines)
1240def color_diff(diff):
1241 """Color difference"""
1242 if diff > 0:
1243 return f"\033[92m improved by {diff:.3f}%\033[00m"
1244 return f"\033[91m worsened by {-diff:.3f}%\033[00m"
1247def color_correl(correlation):
1248 """Color correlation"""
1249 res = f"{correlation*100:.3f}%"
1250 if correlation > 0.9:
1251 res = f" \033[92m{res}\033[00m "
1252 elif correlation < 0:
1253 res = f"\033[91m{res}\033[00m "
1254 elif correlation < 0.5:
1255 res = f" \033[91m{res}\033[00m "
1256 else:
1257 res = f" \033[00m{res}\033[00m "
1258 return res
1261def print_diff(case, prev, new, **kwargs):
1262 """Print difference of correlation"""
1263 topk = kwargs.get("topk", None)
1264 total = kwargs.get("total", None)
1265 tabsize = kwargs.get("tabsize", 40)
1266 diff = (new - prev) * 100
1267 msg = ""
1268 if -0.1 < diff < 0.1:
1269 msg = f"{case} \tcorrelation :{color_correl(new)} \033[00m\033[00m"
1270 else:
1271 msg = (
1272 f"{case} \tcorrelation ({color_correl(new)}) is{color_diff(diff)}"
1273 )
1274 if topk is not None and total is not None:
1275 msg += f" topk = {topk}/{total}"
1276 logger.output(msg.expandtabs(tabsize))
1279def get_distance_i(part: RealParts, data_i: tuple) -> float:
1280 """get the average distance of a given part"""
1281 _, distance, _, _ = data_i
1282 return sum(distance[part]) / len(distance[part])
1285def get_correl_i(part, data_i):
1286 """get the correlation of a given part"""
1287 f_correl, _, _, _ = data_i
1288 return f_correl[part]
1291def print_part_x_file(data, fun):
1292 """Prints a metric computed by fun for each couple (part, file)"""
1293 msg = ""
1294 for part in RealParts:
1295 if part is not RealParts.IDLE:
1296 msg += "\n" + str(part) + "\t"
1297 col_sum = 0
1298 col_num = 0
1299 for data_i in data:
1300 try:
1301 info = fun(part, data_i)
1302 msg += f"\t{(info*100):.1f}%"
1303 col_sum += info
1304 col_num += 1
1305 except KeyError:
1306 msg += "\t -"
1307 if col_num > 0:
1308 msg += f"\t\t{(col_sum/col_num)*100:.1f}%"
1309 return msg
1312def print_correlations_classified(data):
1313 """Printer for estimation vs detailed profiling"""
1314 msg = "\n\t"
1315 for i, _ in enumerate(data):
1316 msg += "\tFile " + str(i + 1)
1317 msg += "\t\tavg"
1319 msg += "\nCorrelation (higher is better)"
1320 msg += print_part_x_file(data, get_correl_i)
1322 msg += "\ntop_k\t"
1323 for _, _, top_k, total in data:
1324 msg += "\t" + str(top_k) + "/" + str(total)
1326 msg += "\n\nEuclidean Distance (lower is better)"
1327 msg += print_part_x_file(data, get_distance_i)
1329 logger.output(msg)
1332def is_constant(array):
1333 """Whether the given array only has the same elements"""
1334 if len(array) == 0:
1335 return True
1336 value = array[0]
1337 return all(vi == value for vi in array)
1340def pearson_wait(correls, real, estim, wait):
1341 """Compute Pearson correlation if inputs are not empty"""
1342 if wait not in {RealParts.IDLE}:
1343 logger.debug(
1344 "correlation for %s between real = %s && estim = %s",
1345 str(wait),
1346 str(real[wait]),
1347 str(estim[wait]),
1348 )
1349 if not is_constant(real[wait]) and not is_constant(estim[wait]):
1350 pearson = pearsonr(
1351 real[wait], estim[wait]
1352 ).statistic # type: ignore
1353 logger.info(
1354 "correlation[%s] of real %s vs estim %s = %f",
1355 wait,
1356 real[wait],
1357 estim[wait],
1358 pearson,
1359 )
1360 correls[wait] = pearson
1361 else:
1362 logger.warning(
1363 "either estim[%s] = %s is constant", wait, str(estim[wait])
1364 )
1365 logger.warning(
1366 "or real[%s] = %s is constant", wait, str(real[wait])
1367 )