Coverage for simulator/plot_utils.py: 99%
168 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 04:47 +0000
1"""
2Utilities for plotting.
3"""
4from __future__ import annotations
6import numpy as np
8from matplotlib import pyplot as plt
9from matplotlib.markers import MarkerStyle
11from typing import Optional
13from utils import get_pareto_frontier
15from sim_types import ProvisioningResult
16from sim_types import GPUType
17from sim_types import Model
18from sim_types import QualityLevel
21FIG_SIZE = (7, 5)
24def get_color_map() -> list[tuple[float, float, float]]:
25 return plt.get_cmap('tab10').colors # type: ignore[attr-defined]
28def _get_time_ticklabels(
29 xmin: Optional[float] = None,
30 xmax: Optional[float] = None,
31 exclude_x: list[int] = [],
32) -> tuple[list[int], list[str]]:
33 ticks_seconds = [1, 2, 5, 10, 15, 30]
35 # Define desired minute values (in seconds) to match the log scale
36 minute_ticks_minutes = [1, 2, 5, 10, 20, 40] # in minutes
37 minute_ticks_seconds = [mins * 60 for mins in minute_ticks_minutes] # convert to seconds
39 hour_ticks_hours = [1, 3, 5, 8, 12] # in hours
40 hour_ticks_seconds = [hours * 60 * 60 for hours in hour_ticks_hours] # convert to seconds
42 day_ticks_days: list[int] = [1] # in days
43 day_ticks_seconds = [days * 24 * 60 * 60 for days in day_ticks_days] # convert to seconds
44 tick_labels = \
45 [f"{s:g}s" for s in ticks_seconds] + \
46 [f"{m:g}m" for m in minute_ticks_minutes] + \
47 [f"{h:g}h" for h in hour_ticks_hours] + \
48 [f"{d:g}d" for d in day_ticks_days]
50 ticks: list[int] = ticks_seconds + minute_ticks_seconds + hour_ticks_seconds + day_ticks_seconds
52 if xmin is not None or xmax is not None or exclude_x:
53 filtered_ticks: list[int] = []
54 filtered_labels: list[str] = []
55 for tick, label in zip(ticks, tick_labels):
56 if tick in exclude_x:
57 continue
58 if xmin is not None and tick < xmin:
59 continue
60 if xmax is not None and tick > xmax:
61 continue
62 filtered_ticks.append(tick)
63 filtered_labels.append(label)
64 ticks = filtered_ticks
65 tick_labels = filtered_labels
67 return ticks, tick_labels
70def plot_x_vs_y(
71 x_data: list[float],
72 y_data: list[float],
73 provisions: list[dict[GPUType, int]],
74 verbose: bool = False,
75 # Plot
76 figsize: tuple[int, int] = FIG_SIZE,
77 # X
78 xlabel: str = "Latency (sec)",
79 xmin: Optional[float] = 20,
80 xmax: Optional[float] = None,
81 # Y
82 ylabel: str = "Cost ($)",
83 ymin: Optional[float] = 0,
84 ymax: Optional[float] = None,
85 # Marker
86 marker_size: int = 80,
87) -> None:
88 # Create the figure
89 plt.figure(figsize=figsize)
91 # Change the font size to 12
92 plt.rcParams.update({'font.size': 12})
94 # Define color map
95 tab10 = plt.get_cmap("tab10").colors # type: ignore[attr-defined]
96 colors = tab10
97 markers = ["o", "^", "s", "D", "P", "X"] # circle, triangle_up, square, diamond, plus, x
99 # scatter plot for mixed provisioning
100 idx_list_mixed: list[int] = []
102 # Data for each GPU type
103 idx_lists_single: dict[GPUType, list[int]] = {}
104 for gpu_idx, gpu_type in enumerate(GPUType):
105 idx_lists_single[gpu_type] = []
106 for idx, provision in enumerate(provisions):
107 if provision.get(gpu_type, 0) > 0 and len(provision) == 1:
108 idx_lists_single[gpu_type].append(idx)
109 else:
110 idx_list_mixed.append(idx)
112 # Scatter plot for mixed provisioning
113 idx_list_mixed = list(set(idx_list_mixed)) # Remove duplicates
114 idx_list_single = [
115 idx
116 for gpu_type in GPUType
117 for idx in idx_lists_single[gpu_type]
118 ]
119 idx_list_mixed = [
120 idx
121 for idx in idx_list_mixed
122 if idx not in idx_list_single
123 ]
124 x_data_mixed = [x_data[i] for i in idx_list_mixed]
125 y_data_mixed = [y_data[i] for i in idx_list_mixed]
126 if x_data_mixed and y_data_mixed:
127 plt.scatter(
128 x_data_mixed,
129 y_data_mixed,
130 s=marker_size,
131 color=tab10[0],
132 label="Mixed",
133 marker=MarkerStyle('x'),
134 alpha=0.5,
135 )
137 # Scatter plot for single GPU type provisioning
138 for gpu_idx, gpu_type in enumerate(GPUType):
139 x_data_gpu_type = [x_data[i] for i in idx_lists_single[gpu_type]]
140 y_data_gpu_type = [y_data[i] for i in idx_lists_single[gpu_type]]
141 if x_data_gpu_type and y_data_gpu_type:
142 plt.scatter(
143 x_data_gpu_type,
144 y_data_gpu_type,
145 s=marker_size,
146 color=colors[gpu_idx + 1],
147 marker=MarkerStyle(markers[gpu_idx]),
148 label=gpu_type.name,
149 alpha=0.7,
150 )
152 # Pareto frontier for all points
153 pareto_front = get_pareto_frontier(x_data, y_data)
155 # Find the provisioning options that correspond to the Pareto front points
156 pareto_provision = []
157 for point in pareto_front.tolist():
158 x_val = point[0]
159 y_val = point[1]
160 try:
161 idx = np.where((np.array(x_data) == x_val) & (np.array(y_data) == y_val))[0][0]
162 pareto_provision.append(provisions[idx])
163 except IndexError:
164 pass # Ignore artificial points
166 if verbose:
167 print("Pareto Front Provisioning Options:")
168 for point in pareto_front.tolist():
169 x_val = point[0]
170 y_val = point[1]
171 try:
172 idx = np.where((np.array(x_data) == x_val) & (np.array(y_data) == y_val))[0][0]
173 num_gpus = provisions[idx]
174 print(
175 f"{num_gpus} -> "
176 f"TTFF: {point[0]:.2f} seconds, Cost: ${point[1]:.2f}")
177 except IndexError:
178 pass # Ignore artificial points
180 # plot the parento curve
181 plt.plot(
182 pareto_front[:, 0],
183 pareto_front[:, 1],
184 color=tab10[0],
185 linewidth=3,
186 # linestyle='--',
187 label="Frontier",
188 zorder=0
189 )
191 # add arrow point to the left bottom corner with 'Better' inside the arrow
192 # plt.text(30, 20, "Better", ha="center", va="center",
193 # rotation=45, size=12,
194 # bbox=dict(boxstyle="larrow,pad=0.2", fc="white", ec="black", lw=1))
196 if xmin is None or xmin > 0:
197 plt.xscale("log")
198 plt.xlim(xmin, xmax)
199 plt.ylim(ymin, ymax)
201 # add a vertical line at x=600 seconds
202 # plt.axvline(x=600, color='red', linestyle='--', label='Realtime')
204 ticks, tick_labels = _get_time_ticklabels(
205 xmin=xmin,
206 xmax=xmax,
207 )
208 plt.xticks(
209 ticks,
210 tick_labels
211 )
213 plt.xlabel(xlabel, fontsize=12)
214 plt.ylabel(ylabel, fontsize=12)
216 plt.grid(True, linestyle='--', alpha=0.7)
218 plt.legend()
220 # Improve formatting
221 plt.tight_layout(pad=0)
223 plt.show()
224 # plt.savefig('ttff_vs_cost.pdf', dpi=300, bbox_inches='tight')
227def plot_ttff_vs_cost(
228 ttffs: list[float],
229 costs: list[float],
230 provisions: list[dict[GPUType, int]],
231 verbose: bool = False,
232 # Plot
233 figsize: tuple[int, int] = FIG_SIZE,
234 # X
235 xlabel: str = "Time to First Frame (TTFF)",
236 xmin: Optional[float] = 20,
237 xmax: Optional[float] = None,
238 # Y
239 ylabel: str = "Cost ($)",
240 ymin: Optional[float] = 0,
241 ymax: Optional[float] = None,
242) -> None:
243 """
244 Plots Time to First Frame (TTFF) against Cost for different provisioning options.
245 Args:
246 ttff_list (list): List of TTFF values corresponding to each provisioning option.
247 costs (list): List of cost values corresponding to each provisioning option.
248 provision (list): List of tuples representing the provisioning options (num_a100s, num_h100s, num_h200s).
249 """
250 plot_x_vs_y(
251 x_data=ttffs,
252 y_data=costs,
253 provisions=provisions,
254 xlabel=xlabel,
255 ylabel=ylabel,
256 verbose=verbose,
257 figsize=figsize,
258 xmin=xmin,
259 xmax=xmax,
260 ymin=ymin,
261 ymax=ymax,
262 )
265def plot_ttff_vs_energy(
266 ttff_list: list[float],
267 energy_list: list[float],
268 actual_provision: list[dict[GPUType, int]],
269 verbose: bool = False,
270 # Plot
271 figsize: tuple[int, int] = FIG_SIZE,
272 # X
273 xlabel: str = "Time to First Frame (TTFF)",
274 xmin: Optional[float] = 20,
275 xmax: Optional[float] = None,
276 # Y
277 ylabel: str = "Energy (kWh)",
278 ymin: Optional[float] = 0,
279 ymax: Optional[float] = None,
280) -> None:
281 # convert energy from Ws to kWh
282 energy_list_copy = [
283 energy / (60 * 60 * 1000) # convert from Ws to kWh
284 for energy in energy_list
285 ]
286 plot_x_vs_y(
287 x_data=ttff_list,
288 y_data=energy_list_copy,
289 provisions=actual_provision,
290 xlabel=xlabel,
291 ylabel=ylabel,
292 verbose=verbose,
293 figsize=figsize,
294 xmin=xmin,
295 xmax=xmax,
296 ymin=ymin,
297 ymax=ymax,
298 )
301def plot_adaptive_quality(
302 provisioning_result_adaptive: ProvisioningResult,
303 provisioning_qualities: dict[QualityLevel, ProvisioningResult],
304 slide_seconds: float = 2.2, # Value in seconds
305 # Plot
306 figsize: tuple[int, int] = FIG_SIZE,
307 # X: 1 second to 10 minutes
308 xmin: Optional[float] = 1,
309 xmax: Optional[float] = 10 * 60,
310 # Y: Cost ($)
311 ymin: Optional[float] = 0,
312 ymax: Optional[float] = 85,
313) -> None:
314 # Plot quality pareto frontiers
315 pareto_front_adaptive = get_pareto_frontier(
316 provisioning_result_adaptive.ttffs,
317 provisioning_result_adaptive.costs,
318 max_x=ymax,
319 max_y=xmax,
320 )
322 pareto_fronts: dict[QualityLevel, np.ndarray] = {}
323 for quality in [QualityLevel.HIGH, QualityLevel.MEDIUM, QualityLevel.LOW]:
324 pareto_fronts[quality] = get_pareto_frontier(
325 provisioning_qualities[quality].ttffs,
326 provisioning_qualities[quality].costs,
327 max_x=ymax,
328 max_y=xmax,
329 )
331 _, ax = plt.subplots(figsize=figsize)
332 for quality in [QualityLevel.HIGH, QualityLevel.MEDIUM, QualityLevel.LOW]:
333 ax.plot(
334 pareto_fronts[quality][:, 0],
335 pareto_fronts[quality][:, 1],
336 linewidth=3,
337 label=quality.name.lower().capitalize(),
338 )
340 # Adaptive quality
341 ax.plot(
342 pareto_front_adaptive[:, 0],
343 pareto_front_adaptive[:, 1],
344 linewidth=3,
345 linestyle='--',
346 label="Adaptive",
347 )
349 # Adaptive + Slide
350 plt.plot(
351 pareto_front_adaptive[:, 0] - slide_seconds,
352 pareto_front_adaptive[:, 1],
353 linewidth=3,
354 linestyle='-.',
355 label="+Slide")
357 plt.legend(loc="upper right")
358 plt.grid()
360 ax.grid(True, which='major', linestyle='--', alpha=0.7)
361 ax.grid(True, which='minor', linestyle=':', alpha=0.4)
363 ax.set_xlabel("TTFF (s)", labelpad=-8)
364 ax.set_ylabel("Cost ($)")
366 if xmin is None or xmin > 0:
367 plt.xscale("log")
368 plt.xlim(xmin, xmax)
369 plt.ylim(ymin, ymax)
371 ticks, tick_labels = _get_time_ticklabels(
372 xmin=xmin,
373 xmax=xmax,
374 )
375 plt.xticks(
376 ticks,
377 tick_labels
378 )
380 ax2 = ax.twinx()
381 if ymax is not None:
382 ax2.set_ylim(0, ymax / 10) # Scale by 1/10 cost/10 minutes
383 ax2.set_ylabel("Cost ($/minute)", rotation=-90, labelpad=15)
385 plt.tight_layout(pad=0)
386 plt.show()
389def plot_cost_vs_qpm(
390 costs: dict[GPUType, dict[Model, list[float]]],
391 qpms: list[float] = [1, 2],
392 # Plot
393 figsize: tuple[int, int] = FIG_SIZE,
394 xlabel: str = "Requests per Minute (QPM)",
395 ylabel: str = "Cost ($)",
396) -> None:
397 """Plot cost vs QPM for each component."""
398 plt.figure(figsize=figsize)
400 for gpu_type in costs.keys():
401 for model in costs[gpu_type].keys():
402 plt.plot(
403 qpms,
404 costs[gpu_type][model],
405 marker='o',
406 label=f"{model.name} ({gpu_type.name})")
408 # plt.plot(QPM_LIST, total_costs, marker='o', label='Total Cost', color='black', linewidth=2)
410 plt.xscale('log')
412 plt.xlabel(xlabel, fontsize=12)
413 plt.ylabel(ylabel, fontsize=12)
414 plt.title('Cost vs. Requests per Minute (QPM) for Each Component', fontsize=14)
416 plt.xticks(qpms, rotation=45)
417 plt.grid(True, linestyle='--', alpha=0.7)
418 if len(costs) > 0:
419 plt.legend()
420 plt.tight_layout(pad=0)
421 plt.show()
424def plot_policies_ttff_vs_cost(
425 provision_results: dict[str, ProvisioningResult],
426 points: bool = True,
427 front: bool = True,
428 # Plot
429 figsize: tuple[int, int] = FIG_SIZE,
430 # X
431 xlabel: str = "Time to first frame (TTFF)",
432 xmin: Optional[int] = 20,
433 xmax: Optional[int] = 3 * 60 * 60, # 3 hours
434 # Y
435 ylabel: str = "Cost ($)",
436 ymin: Optional[int] = 0,
437 ymax: Optional[int] = 1500,
438) -> None:
439 assert front or points, "At least one of 'front' or 'points' must be True"
441 _, ax = plt.subplots(figsize=figsize)
443 # Plot results
444 for policy_name, provision_result in provision_results.items():
445 # Plot points
446 if points:
447 ax.scatter(
448 provision_result.ttffs,
449 provision_result.costs,
450 marker=MarkerStyle("o"),
451 alpha=0.5,
452 label=policy_name
453 )
454 # Plot Pareto frontier
455 if front:
456 pareto_frontier = get_pareto_frontier(
457 provision_result.ttffs,
458 provision_result.costs,
459 max_x=ymax,
460 max_y=xmax,
461 )
462 ax.plot(
463 pareto_frontier[:, 0],
464 pareto_frontier[:, 1],
465 linewidth=3,
466 label=f"{policy_name}" if not points else None
467 )
469 if xmin is None or xmin > 0:
470 plt.xscale("log")
471 plt.xlim(xmin, xmax)
472 plt.ylim(ymin, ymax)
474 ticks, tick_labels = _get_time_ticklabels(
475 xmin=xmin,
476 xmax=xmax,
477 )
478 plt.xticks(
479 ticks,
480 tick_labels
481 )
483 plt.xlabel(xlabel)
484 plt.ylabel(ylabel)
486 plt.grid()
487 if len(provision_results) > 0:
488 plt.legend()
489 plt.tight_layout(pad=0)
490 plt.show()