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
|
"""A question with a few answers, and the strip that shows transfers."""
import time
from dataclasses import dataclass
from rich.text import Text
from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical
from textual.screen import ModalScreen
from textual.widgets import Button, Static
from ..output import human_size, safe
class Choice(ModalScreen[str]):
"""One question, asked once for a whole batch — never once per file."""
DEFAULT_CSS = """
Choice { align: center middle; }
Choice > Vertical { width: 72; height: auto; max-height: 80%;
border: thick $accent; background: $surface; padding: 1 2; }
Choice Static { height: auto; margin-bottom: 1; }
Choice Horizontal { height: auto; align: center middle; }
Choice Button { margin: 0 1; }
"""
BINDINGS = [("escape", "dismiss('cancel')", "Cancel")]
def __init__(self, message: str, options: list[tuple[str, str]]):
super().__init__()
self.message = message
self.options = options
def compose(self) -> ComposeResult:
with Vertical():
yield Static(Text(self.message))
with Horizontal():
for value, label in self.options:
yield Button(label, id=f"choice-{value}")
def on_mount(self) -> None:
self.query(Button).first().focus()
def on_button_pressed(self, event: Button.Pressed) -> None:
self.dismiss(event.button.id.removeprefix("choice-"))
@dataclass
class Job:
arrow: str
name: str
size: int
done: int = 0
state: str = "waiting" # waiting, queued, running, done, failed
note: str = ""
started: float = 0.0
class TransferStrip(Static):
"""The running transfers, then the latest finished ones."""
DEFAULT_CSS = """
TransferStrip { height: 6; border-top: solid $panel; padding: 0 1; }
"""
SHOWN = 5
def __init__(self, **kwargs):
super().__init__("", markup=False, **kwargs)
self.jobs: list[Job] = []
def on_mount(self) -> None:
self.refresh_text()
def add(self, arrow: str, name: str, size: int) -> Job:
job = Job(arrow, name, size)
self.jobs.append(job)
self.refresh_text()
return job
def update_job(self, job: Job, **fields) -> None:
for k, v in fields.items():
setattr(job, k, v)
if job.state == "running" and not job.started:
job.started = time.monotonic()
self.refresh_text()
def refresh_text(self) -> None:
active = [j for j in self.jobs if j.state in ("running", "queued", "waiting")]
ended = [j for j in self.jobs if j.state in ("done", "failed")]
lines = [self._line(j) for j in active[:self.SHOWN]]
if len(active) > self.SHOWN:
lines.append(f" … and {len(active) - self.SHOWN} more waiting")
for j in reversed(ended[-(self.SHOWN - len(lines)):] if len(lines) < self.SHOWN
else []):
lines.append(self._line(j))
self.update(Text("\n".join(lines) or "No transfers."))
def _line(self, j: Job) -> str:
name = safe(j.name)
if j.state == "running":
pct = 100 * j.done // j.size if j.size else 100
rate = j.done / max(time.monotonic() - j.started, 0.001)
return f"{j.arrow} {name} {pct}% {human_size(int(rate))}/s"
if j.state == "queued":
return f"{j.arrow} {name} waiting for a free slot on the node {j.note}"
if j.state == "waiting":
return f"{j.arrow} {name} waiting"
if j.state == "done":
return f"{j.arrow} {name} done{(' — ' + j.note) if j.note else ''}"
return f"{j.arrow} {name} FAILED: {j.note}"
|