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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
"""
Writing to the terminal.
**Names are someone else's text.** A file or group name is chosen by whoever put
it in the group, and a raw ESC in it is an escape sequence the reader's
terminal executes; a bidirectional override makes it display as another name.
Every name printed goes through `safe()`.
"""
import json
import shutil
import sys
import time
import unicodedata
_BIDI = frozenset("")
def safe(text: str) -> str:
out = []
for c in text:
if unicodedata.category(c) == "Cc" or c in _BIDI:
out.append(f"\\x{ord(c):02x}" if ord(c) < 0x100 else f"\\u{ord(c):04x}")
else:
out.append(c)
return "".join(out)
def human_size(n: int) -> str:
size = float(n)
for unit in ("B", "KB", "MB", "GB", "TB"):
if size < 1024 or unit == "TB":
return f"{size:.0f} {unit}" if unit == "B" else f"{size:.1f} {unit}"
size /= 1024
return f"{n} B"
def date(ts: int) -> str:
return time.strftime("%Y-%m-%d %H:%M", time.localtime(ts)) if ts else ""
def columns(names: list[str]) -> str:
"""`ls`'s short form on a terminal: as many columns as fit, filled down."""
if not names:
return ""
width = shutil.get_terminal_size((80, 24)).columns
col = max(len(n) for n in names) + 2
ncols = max(1, width // col)
nrows = -(-len(names) // ncols)
lines = []
for r in range(nrows):
row = [names[c * nrows + r] for c in range(ncols) if c * nrows + r < len(names)]
lines.append("".join(n.ljust(col) for n in row).rstrip())
return "\n".join(lines)
def is_tty() -> bool:
return sys.stdout.isatty()
def err(message: str) -> None:
print(message, file=sys.stderr)
class Progress:
"""
One status line on stderr for a run of transfers, redrawn at most ten times
a second; JSON Lines with --json; nothing with -q or when stderr is not a
terminal.
"""
def __init__(self, arrow: str, *, quiet: bool = False, as_json: bool = False):
self.arrow = arrow
self.live = sys.stderr.isatty() and not quiet and not as_json
self.as_json = as_json and not quiet
self.files_total = 0
self.files_done = 0
self.bytes_total = 0
self.bytes_done = 0
self.current = ""
self._start = time.monotonic()
self._drawn = 0.0
def plan(self, files: int, nbytes: int) -> None:
self.files_total, self.bytes_total = files, nbytes
def event(self, **fields) -> None:
if self.as_json:
print(json.dumps(fields, ensure_ascii=False), file=sys.stderr, flush=True)
def start(self, name: str) -> None:
self.current = name
self.event(event="start", name=name)
self._draw(force=True)
def advance(self, nbytes: int) -> None:
self.bytes_done += nbytes
self._draw()
def queued(self, name: str, ahead: int) -> None:
self.current = f"{name} (waiting for a free slot on the node, {ahead} ahead)"
self.event(event="queued", name=name, ahead=ahead)
self._draw(force=True)
def finish(self, name: str, **fields) -> None:
self.files_done += 1
self.event(event="done", name=name, **fields)
self._draw(force=True)
def note(self, message: str) -> None:
"""A line of its own, above the status line."""
self.clear()
if not self.as_json:
print(message, file=sys.stderr)
self._draw(force=True)
def _draw(self, force: bool = False) -> None:
if not self.live:
return
now = time.monotonic()
if not force and now - self._drawn < 0.1:
return
self._drawn = now
rate = self.bytes_done / max(now - self._start, 0.001)
line = (f"{self.arrow} {self.files_done}/{self.files_total} "
f"{human_size(self.bytes_done)} / {human_size(self.bytes_total)} "
f"{human_size(int(rate))}/s {self.current}")
width = shutil.get_terminal_size((80, 24)).columns
sys.stderr.write("\r\x1b[K" + line[:width - 1])
sys.stderr.flush()
def clear(self) -> None:
if self.live:
sys.stderr.write("\r\x1b[K")
sys.stderr.flush()
def sentence(text: str) -> str:
"""The first letter up, the rest untouched: `str.capitalize` turns -R into -r."""
return text[:1].upper() + text[1:]
|