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
|
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
"""This script is intended to be called through fzf as a preview formatter."""
import argparse
import os
import sys
here = os.path.abspath(os.path.dirname(__file__))
sys.path.insert(0, os.path.join(os.path.dirname(here), "util"))
from estimates import duration_summary
def process_args():
"""Process preview arguments."""
argparser = argparse.ArgumentParser()
argparser.add_argument(
"-s",
"--show-estimates",
action="store_true",
help="Show task duration estimates (default: False)",
)
argparser.add_argument(
"-g",
"--graph-cache",
type=str,
default=None,
help="Filename of task graph dependencies",
)
argparser.add_argument(
"-c",
"--cache_dir",
type=str,
default=None,
help="Path to cache directory containing task durations",
)
argparser.add_argument(
"-t",
"--tasklist",
type=str,
default=None,
help="Path to temporary file containing the selected tasks",
)
return argparser.parse_args()
def plain_display(taskfile):
"""Original preview window display."""
with open(taskfile) as f:
tasklist = [line.strip() for line in f]
print("\n".join(sorted(tasklist)))
def duration_display(graph_cache_file, taskfile, cache_dir):
"""Preview window display with task durations + metadata."""
with open(taskfile) as f:
tasklist = [line.strip() for line in f]
durations = duration_summary(graph_cache_file, tasklist, cache_dir)
output = ""
max_columns = int(os.environ["FZF_PREVIEW_COLUMNS"])
output += "\nSelected tasks take {}\n".format(durations["selected_duration"])
output += "+{} dependencies, total {}\n".format(
durations["dependency_count"],
durations["selected_duration"] + durations["dependency_duration"],
)
if durations.get("percentile"):
output += "This is in the top {}% of requests\n".format(
100 - durations["percentile"]
)
output += "Estimated finish in {} at {}".format(
durations["wall_duration_seconds"], durations["eta_datetime"].strftime("%H:%M")
)
duration_width = 5 # show five numbers at most.
output += "{:>{width}}\n".format("Duration", width=max_columns)
for task in tasklist:
duration = durations["task_durations"].get(task, 0.0)
output += "{:{align}{width}} {:{nalign}{nwidth}}s\n".format(
task,
duration,
align="<",
width=max_columns - (duration_width + 2), # 2: space and 's'
nalign=">",
nwidth=duration_width,
)
print(output)
if __name__ == "__main__":
args = process_args()
if args.show_estimates and os.path.isdir(args.cache_dir):
duration_display(args.graph_cache, args.tasklist, args.cache_dir)
else:
plain_display(args.tasklist)
|
