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