github/darkfi
1
1
Fork 0
mirror of https://github.com/darkrenaissance/darkfi.git synced 2026-01-09 14:48:08 -05:00
Code Issues Packages Projects Releases Wiki Activity

script/dnet: implement PEP 8

Browse Source
This commit is contained in:
lunar-mining
2023-09-10 16:09:18 +02:00
parent 3be0e7a757
commit 259ecbc004
5 changed files with 59 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
script/dnet/main.py
View File

@@ -21,18 +21,18 @@ from model import Model
from rpc import JsonRpc
from view import View
class Dnetview:
def __init__(self):
self.ev = asyncio.get_event_loop()
self.queue = asyncio.Queue()
self.config = self.get_config()
self.model = Model()
self.view = View(self.model)
async def subscribe(self, rpc, name, port):
info = {}
while True:
try:
logging.debug(f"Start {name} RPC on port {port}")
@@ -77,7 +77,8 @@ class Dnetview:
async with asyncio.TaskGroup() as tg:
for i, node in enumerate(nodes):
rpc = JsonRpc()
subscribe = tg.create_task(self.subscribe(rpc, node['name'], node['port']))
subscribe = tg.create_task(self.subscribe(
rpc, node['name'], node['port']))
nodes = tg.create_task(self.update_info())
async def update_info(self):
@@ -99,7 +100,9 @@ class Dnetview:
logging.debug("update_model(): error {}", e)
def main(self):
logging.basicConfig(filename='dnet.log', encoding='utf-8', level=logging.DEBUG)
logging.basicConfig(filename='dnet.log',
encoding='utf-8',
level=logging.DEBUG)
nodes = self.config.get("nodes")
self.ev.create_task(self.start_connect_slots(nodes))
@@ -107,9 +110,9 @@ class Dnetview:
self.ev.create_task(self.view.render_info())
loop = urwid.MainLoop(self.view.ui, self.view.palette,
unhandled_input=self.unhandled_input,
event_loop=urwid.AsyncioEventLoop(loop=self.ev))
unhandled_input=self.unhandled_input,
event_loop=urwid.AsyncioEventLoop(
loop=self.ev))
loop.run()
def unhandled_input(self, key):
@@ -118,6 +121,7 @@ class Dnetview:
task.cancel()
raise urwid.ExitMainLoop()
if __name__ == '__main__':
dnet = Dnetview()
dnet.main()

13
script/dnet/model.py
View File

@@ -17,7 +17,9 @@
import logging
class Model:
def __init__(self):
self.info = Info()
self.nodes = {}
@@ -27,10 +29,8 @@ class Model:
def handle_nodes(self, node):
channel_lookup = {}
name = list(node.keys())[0]
values = list(node.values())[0]
info = values["result"]
channels = info["channels"]
@@ -70,7 +70,6 @@ class Model:
def handle_event(self, event):
name = list(event.keys())[0]
values = list(event.values())[0]
params = values.get("params")
event = params[0].get("event")
info = params[0].get("info")
@@ -92,7 +91,9 @@ class Model:
def __repr__(self):
return f"{self.nodes}"
class Info:
def __init__(self):
self.outbounds = {}
self.inbound = {}
@@ -119,10 +120,8 @@ class Info:
self.msgs[key] = [value]
def __repr__(self):
return (
f"outbound: {self.outbounds}"
return (f"outbound: {self.outbounds}"
f"inbound: {self.inbound}"
f"manual: {self.manual}"
f"seed: {self.seed}"
f"msg: {self.msgs}"
)
f"msg: {self.msgs}")

8
script/dnet/rpc.py
View File

@@ -15,9 +15,15 @@
# 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 asyncio, json, random, time, logging
import json
import time
import random
import logging
import asyncio
class JsonRpc:
async def start(self, server, port):
reader, writer = await asyncio.open_connection(server, port)
self.reader = reader

3
script/dnet/scroll.py
View File

@@ -33,6 +33,7 @@ SCROLLBAR_RIGHT = 'right'
# Add support for ScrollBar class (see stig.tui.scroll)
# https://github.com/urwid/urwid/issues/226
class ListBox_patched(urwid.ListBox):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._rows_max = None
@@ -330,6 +331,7 @@ DEFAULT_THUMB_CHAR = '\u2588'
DEFAULT_TROUGH_CHAR = " "
DEFAULT_SIDE = SCROLLBAR_RIGHT
class ScrollBar(urwid.WidgetDecoration):
_thumb_char = DEFAULT_THUMB_CHAR
@@ -570,4 +572,5 @@ class ScrollBar(urwid.WidgetDecoration):
return False
__all__ = ["Scrollable", "ScrollBar"]

63
script/dnet/view.py
View File

@@ -18,26 +18,33 @@
import urwid
import logging
import asyncio
import datetime
import datetime as dt
from scroll import ScrollBar, Scrollable
from model import Model
event_loop = asyncio.get_event_loop()
class LeftList(urwid.ListBox):
def focus_next(self):
try:
self.body.set_focus(self.body.get_next(self.body.get_focus()[1])[1])
self.body.set_focus(self.body.get_next(
self.body.get_focus()[1])[1])
except:
pass
def focus_previous(self):
try:
self.body.set_focus(self.body.get_prev(self.body.get_focus()[1])[1])
self.body.set_focus(self.body.get_prev(
self.body.get_focus()[1])[1])
except:
pass
class NodeView(urwid.WidgetWrap):
def __init__(self, info):
self.name = info
self.text = urwid.Text(f"{self.name}")
@@ -62,7 +69,9 @@ class NodeView(urwid.WidgetWrap):
def get_name(self):
return self.name
class ConnectView(urwid.WidgetWrap):
def __init__(self, info):
self.name = info
self.text = urwid.Text(f"{self.name}")
@@ -74,8 +83,6 @@ class ConnectView(urwid.WidgetWrap):
return True
def keypress(self, size, key):
#if key in ('q'):
# raise urwid.ExitMainLoop()
return key
def update_w(self):
@@ -87,7 +94,9 @@ class ConnectView(urwid.WidgetWrap):
def get_name(self):
return self.name
class SlotView(urwid.WidgetWrap):
def __init__(self, info):
self.name = info
self.text = urwid.Text(f"{self.name}")
@@ -99,8 +108,6 @@ class SlotView(urwid.WidgetWrap):
return True
def keypress(self, size, key):
#if key in ('q'):
# raise urwid.ExitMainLoop()
return key
def update_w(self):
@@ -112,6 +119,7 @@ class SlotView(urwid.WidgetWrap):
def get_name(self):
return self.name
class View():
palette = [
('body','light gray','black', 'standout'),
@@ -124,18 +132,17 @@ class View():
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 = LeftList(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):
names = []
while True:
names = []
await asyncio.sleep(0.1)
for item in self.listwalker.contents:
name = item.get_name()
names.append(name)
@@ -143,7 +150,6 @@ class View():
for name, values in self.model.nodes.items():
if name in names:
continue
else:
widget = NodeView(name)
self.listwalker.contents.append(widget)
@@ -172,44 +178,39 @@ class View():
widget = ConnectView(" manual")
self.listwalker.contents.append(widget)
await asyncio.sleep(0.1)
async def render_info(self):
while True:
await asyncio.sleep(0.1)
self.pile.contents.clear()
focus_w = self.list.get_focus()
match focus_w[0].get_widget():
case "NodeView":
self.pile.contents.append((
urwid.Text(f"Node selected"),
self.pile.options()))
case "ConnectView":
self.pile.contents.append((
urwid.Text("Connection selected"),
self.pile.options()))
case "SlotView":
name = focus_w[0].get_name()
# Remove the prepend
name = name[7:]
numbered_name = focus_w[0].get_name()
# Remove numbering
name = numbered_name[7:]
if name in self.model.info.msgs.keys():
values = (
self.model.info.msgs.get(name)
)
values = (self.model.info.msgs.get(name))
for value in values:
nanotime = (
int(value[0])
)
time = (
datetime.datetime.fromtimestamp(
nanotime/1000000000).strftime(
'%Y-%m-%d %H:%M:%S.%f')
)
nano = (int(value[0]))
time = (dt.datetime
.fromtimestamp(nano/1000000000)
.strftime('%Y-%m-%d %H:%M:%S.%f'))
event = value[1]
msg = value[2]
#logging.debug(values)
self.pile.contents.append((
urwid.Text(
self.pile.contents.append((urwid.Text(
f"{time}: {event}: {msg}"),
self.pile.options()))
self.pile.options()))