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
|
"""
`meshbay-cli` with no argument: a full-screen terminal interface.
Textual, because it runs on the same asyncio loop as aiortc and httpx rather
than beside them, and on Linux terminals and Windows' consoles alike. Signed in
already, it opens on the two panes — or, for `meshbay-cli chat` with no group,
on the groups to chat in (`tui/chat.py`); otherwise on the sign-in screen first.
Nothing is printed while it runs — a line on stderr would tear the screen — so
the log goes to a file in the CLI's data directory instead.
"""
import logging
from textual import events
from textual.app import App
from textual.errors import NoWidget
from ..client.account import Account, no_passphrase
from ..client.hub import Hub
from ..client.store import data_dir
from ..commands.context import Context
from ..errors import CliError
from .screens import DeviceCode, MainScreen, SignInScreen
class MeshBayApp(App):
TITLE = "MeshBay"
def __init__(self, ctx: Context, home=MainScreen):
super().__init__()
self.ctx = ctx
self.hub = ctx.hub
# The screen a signed-in account lands on, made from the account.
self.home = home
self._right_click_taken = False
async def use_hub(self, url: str) -> None:
if url != self.hub.url:
await self.hub.aclose()
self.hub = Hub(url)
self.ctx.hub = self.hub
self.ctx.hub_url = url
async def on_event(self, event: events.Event) -> None:
"""A right-click goes to the widget under it when it has a use for one
(the chat's copy and paste), before Textual takes it for a selection."""
if (isinstance(event, (events.MouseDown, events.MouseUp)) and event.button == 3
and not event.is_forwarded):
if isinstance(event, events.MouseDown):
try:
widget, _ = self.get_widget_at(event.x, event.y)
except NoWidget:
widget = None
handler = getattr(widget, "right_click", None)
self._right_click_taken = handler is not None
if handler is not None:
await handler(event)
return
elif self._right_click_taken:
self._right_click_taken = False
return
await super().on_event(event)
def on_mount(self) -> None:
account = None
if self.ctx.store.peek().get("hub") == self.hub.url:
try:
account = Account.open(self.hub, None, no_passphrase, self.ctx.store)
except CliError as e:
self.notify(e.message, severity="warning", markup=False)
if account is not None and account.remembered:
self.start(account)
else:
self.push_screen(SignInScreen())
def start(self, account: Account) -> None:
account.on_device_code = DeviceCode(self)
if isinstance(self.screen, SignInScreen):
self.switch_screen(self.home(account))
else:
self.push_screen(self.home(account))
def _log_to_file() -> None:
root = logging.getLogger()
for h in list(root.handlers):
root.removeHandler(h)
try:
directory = data_dir()
directory.mkdir(parents=True, exist_ok=True)
handler = logging.FileHandler(directory / "interactive.log", encoding="utf-8")
except OSError:
handler = logging.NullHandler()
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
root.addHandler(handler)
async def run_interactive(ctx: Context, home=MainScreen) -> int:
_log_to_file()
await MeshBayApp(ctx, home).run_async()
return 0
|