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
|
"""
`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; 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.app import App
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):
super().__init__()
self.ctx = ctx
self.hub = ctx.hub
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
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(MainScreen(account))
else:
self.push_screen(MainScreen(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) -> int:
_log_to_file()
await MeshBayApp(ctx).run_async()
return 0
|