# This file is part of DarkFi (https://dark.fi) # # Copyright (C) 2020-2023 Dyne.org foundation # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import urwid import logging import asyncio import datetime as dt from scroll import ScrollBar, Scrollable from model import Model class DnetWidget(urwid.WidgetWrap): def __init__(self, node_name, session): self.node_name = node_name self.session = session def selectable(self): return True def keypress(self, size, key): return key def update(self, txt): super().__init__(txt) self._w = urwid.AttrWrap(self._w, None) self._w.focus_attr = 'line' class Node(DnetWidget): def set_txt(self): txt = urwid.Text(f"{self.node_name}") super().update(txt) class Session(DnetWidget): def set_txt(self): txt = urwid.Text(f" {self.session}") super().update(txt) class Slot(DnetWidget): def set_txt(self, i, addr): self.i = i self.addr = addr if len(self.i) == 1: txt = urwid.Text(f" {self.i}: {self.addr}") else: txt = urwid.Text(f" {self.addr}") super().update(txt) class View(): palette = [ ('body','light gray','default', 'standout'), ('line','dark cyan','default','standout'), ] def __init__(self, model): self.model = model info_text = urwid.Text("") self.pile = urwid.Pile([info_text]) scroll = ScrollBar(Scrollable(self.pile)) rightbox = urwid.LineBox(scroll) self.listbox_content = [] self.listwalker = urwid.SimpleListWalker(self.listbox_content) self.list = urwid.ListBox(self.listwalker) leftbox = urwid.LineBox(self.list) columns = urwid.Columns([leftbox, rightbox], focus_column=0) self.ui = urwid.Frame(urwid.AttrWrap( columns, 'body' )) async def update_view(self, evloop: asyncio.AbstractEventLoop, loop: urwid.MainLoop): known_nodes = [] known_inbound = [] while True: await asyncio.sleep(0.1) # Redraw the screen evloop.call_soon(loop.draw_screen) for index, item in enumerate(self.listwalker.contents): known_nodes.append(item.node_name) #----------------------------------------------------------------- # Render get_info() #----------------------------------------------------------------- for node_name, info in self.model.nodes.items(): if node_name in known_nodes: continue else: node = Node(node_name, "node") node.set_txt() self.listwalker.contents.append(node) if info['outbound']: session = Session(node_name, "outbound") session.set_txt() self.listwalker.contents.append(session) for i, addr in info['outbound'].items(): slot = Slot(node_name, "outbound-slot") slot.set_txt(i, addr) self.listwalker.contents.append(slot) if info['inbound']: session = Session(node_name, "inbound") session.set_txt() self.listwalker.contents.append(session) for i, addr in info['inbound'].items(): slot = Slot(node_name, "inbound-slot") slot.set_txt(i, addr) self.listwalker.contents.append(slot) if info['manual']: session = Session(node_name, "manual") session.set_txt() self.listwalker.contents.append(session) for i, addr in info['manual'].items(): slot = Slot(node_name, "manual-slot") slot.set_txt(i, addr) self.listwalker.contents.append(slot) if info['seed']: session = Session(node_name, "seed") session.set_txt() self.listwalker.contents.append(session) for i, info in info['seed'].items(): slot = Slot(node_name, "seed-slot") slot.set_txt(i, addr) self.listwalker.contents.append(slot) #----------------------------------------------------------------- # Render subscribe_events() (left menu) #----------------------------------------------------------------- new_inbound = {} for index, item in enumerate(self.listwalker.contents): # Update outbound slot info if item.session == "outbound-slot": key = (f"{item.node_name}", f"{item.i}") if key in self.model.nodes[item.node_name]['event']: info = self.model.nodes[item.node_name]['event'].get(key) slot = Slot(item.node_name, item.session) slot.set_txt(item.i, info) self.listwalker.contents[index] = slot # Get known inbounds if item.session == "inbound-slot": if item.i not in known_inbound: known_inbound.append(item.i) # Get unknown inbounds inbound = self.model.nodes[item.node_name]['inbound'] for i, addr in inbound.items(): if i in known_inbound: continue else: new_inbound[i] = (addr, item.node_name) # TODO: update new inbounds # Render subscribe_events() (right menu) async def render_info(self, evloop: asyncio.AbstractEventLoop, loop: urwid.MainLoop): while True: await asyncio.sleep(0.01) # Redraw the screen evloop.call_soon(loop.draw_screen) self.pile.contents.clear() focus_w = self.list.get_focus() if focus_w[0] is None: continue else: session = focus_w[0].session if session == "node": continue if session == "outbound": key = (focus_w[0].node_name, "outbound") info = self.model.nodes.get(focus_w[0].node_name) if key in info['event']: ev = info['event'].get(key) self.pile.contents.append(( urwid.Text(f" {ev}"), self.pile.options())) if (session == "outbound-slot" or session == "inbound-slot" or session == "manual-slot" or session == "seed-slot"): addr = focus_w[0].addr node_name = focus_w[0].node_name info = self.model.nodes.get(node_name) if addr in info['msgs']: msg = info['msgs'].get(addr) for m in msg: time = m[0] event = m[1] msg = m[2] self.pile.contents.append((urwid.Text( f"{time}: {event}: {msg}"), self.pile.options()))