This repository has no description
1import os
2from pathlib import Path
3from subprocess import run
4from sys import argv
5from time import time_ns
6
7from rich.console import Console
8from rich.table import Table
9
10ignored_tasks = []
11
12compare_with = ""
13if len(argv) > 1:
14 compare_with = Path(argv[1]).read_text(encoding="utf-8").strip()
15 compare_with = {
16 (line.split(",")[0], float(line.split(",")[1]), int(line.split(",")[2]))
17 for line in compare_with.splitlines()[1:]
18 }
19 print(compare_with)
20
21
22def avg(numbers: list[float]):
23 return sum(numbers) / len(numbers)
24
25
26end = 0
27start = 0
28
29if not Path("timings.log").exists():
30 start = time_ns()
31 result = run(
32 ["just", "example-video", "out.mp4", "--duration 5"],
33 capture_output=True,
34 env=os.environ | {"RUST_LOG": "debug"},
35 )
36 end = time_ns()
37
38 Path("timings.log").write_bytes(result.stdout + result.stderr)
39
40timings = [
41 line.split(" took ")
42 for line in Path("timings.log").read_text(encoding="utf-8").splitlines()
43 if " took " in line
44]
45
46
47def parse_duration(duration_string: str) -> float:
48 if " " in duration_string:
49 return sum(parse_duration(part) for part in duration_string.split(" "))
50 try:
51 figure = float(duration_string.strip("msµ"))
52 except ValueError:
53 return None
54
55 if "µs" in duration_string:
56 return figure * 1e-3
57 if "ms" in duration_string:
58 return figure
59 if "s" in duration_string:
60 return figure * 1e3
61 else:
62 return figure
63
64 raise ValueError(f"Duration string {duration_string!r} has unsupported unit")
65
66
67timings = [
68 (label.split("] ")[1], parse_duration(timing))
69 for label, timing in timings
70 if parse_duration(timing)
71]
72
73per_function: dict[str, list[float]] = {function: [] for function, _ in timings}
74
75for function, timing in timings:
76 per_function[function].append(timing)
77
78averages: list[tuple[str, float, int]] = [
79 (function, avg(timings), len(timings)) for function, timings in per_function.items()
80] + [
81 ("_Total_", (end - start) / 1e6, 1),
82]
83
84averages.sort(key=lambda item: item[1])
85
86header = ["Tâche", "Durée [ms]", "#"]
87
88if compare_with:
89 formatted_results = []
90 for function, timing_after, count_after in averages:
91 if function not in {function for function, _, _ in compare_with}:
92 continue
93 timing_before = next(
94 (timing for fn, timing, _ in compare_with if fn == function),
95 None,
96 )
97 count_before = next(
98 (count for fn, _, count in compare_with if fn == function),
99 None,
100 )
101 if timing_before is None or count_before is None:
102 continue
103 if function in ignored_tasks:
104 continue
105
106 formatted_results.append(
107 [
108 function,
109 f"{timing_after:.3f}",
110 f"{timing_before:.3f}",
111 f"{timing_after - timing_before:+.3f}"
112 if f"{timing_after - timing_before:.3f}" != "-0.000"
113 else "±0",
114 f"{count_after}",
115 f"{count_before}",
116 f"{count_after - count_before:+}"
117 if count_after != count_before
118 else "±0",
119 ]
120 )
121
122 header = [
123 "Tâche",
124 "Durée [ms]",
125 "Durée [ms] avant",
126 "Différence [±ms]",
127 "# après",
128 "# avant",
129 "Différence",
130 ]
131else:
132 formatted_results = [
133 [function, f"{timing:.3f}", f"{count}"] for function, timing, count in averages
134 ]
135
136
137def to_csv(lists: list[list[str]]):
138 return "\r\n".join(
139 ",".join(cell.replace(",", "_") for cell in line) for line in lists
140 )
141
142
143table = Table(*header)
144for row in formatted_results:
145 table.add_row(*row)
146Console().print(table)
147
148Path("results.csv").write_text(
149 to_csv(
150 [
151 header,
152 *[
153 [task, *things]
154 for task, *things in formatted_results
155 if task not in ignored_tasks
156 ],
157 ]
158 ),
159 encoding="utf8",
160)