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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
"""
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 terminal(stream) -> bool:
"""
Whether `stream` is a terminal a person sits at. `isatty()` alone is wrong on
Windows, where it means "a character device" and so says yes for NUL: a
script run with `< NUL`, or by the Task Scheduler, was asked questions it
could not answer instead of being told to pass --yes.
"""
try:
if not stream.isatty():
return False
except (AttributeError, ValueError):
return False
if sys.platform != "win32":
return True
try:
fd = stream.fileno()
except (OSError, ValueError):
return True # not a file at all: take its word
import ctypes
import msvcrt
try:
handle = msvcrt.get_osfhandle(fd)
except OSError:
return False
mode = ctypes.c_uint32()
return bool(ctypes.windll.kernel32.GetConsoleMode(handle, ctypes.byref(mode)))
def is_tty() -> bool:
return terminal(sys.stdout)
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 = terminal(sys.stderr) 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:]
def confirm(question: str, yes: bool) -> None:
"""
Go on only on a yes: `--yes`, or `y` typed at a terminal. With neither — a
script that did not say so — refuse before anything is sent, rather than
take silence for consent.
"""
from .errors import USAGE, CliError
if yes:
return
if not terminal(sys.stdin):
raise CliError(USAGE, f"{question} Not done: there is no terminal to ask on; "
"pass --yes to confirm.")
# On stderr, like every other word to the person: stdout is the result.
sys.stderr.write(f"{question} [y/N] ")
sys.stderr.flush()
answer = sys.stdin.readline()
if answer.strip().lower() not in ("y", "yes"):
raise CliError(USAGE, "Not done.")
|