darkfi/bin/dnet/view.py

# 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 <https://www.gnu.org/licenses/>.

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()))