#!/usr/bin/python3 # This file is part of DarkFi (https://dark.fi) # # Copyright (C) 2020-2024 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 json, urwid, asyncio, logging import src.util from os.path import join from src.model import Model from src.rpc import JsonRpc from src.view import View class Degview: def __init__(self): self.ev = asyncio.new_event_loop() asyncio.set_event_loop(self.ev) self.queue = asyncio.Queue() os = src.util.get_os() config_path = src.util.user_config_dir('darkfi', os) suffix = '.toml' filename = 'deg_config' path = join(config_path, filename + suffix) self.config = src.util.spawn_config(path) self.model = Model() self.view = View(self.model) async def subscribe(self, rpc, node): name = node['name'] host = node['host'] port = node['port'] type = node['type'] info = {} while True: try: await rpc.start(host, port) logging.debug(f'Started {name} RPC on port {port}') break except Exception as e: info[name] = {} await self.queue.put(info) continue if type == 'NORMAL': data = await rpc._make_request('eventgraph.get_info', []) info[name] = data await self.queue.put(info) await rpc.deg_switch(True) await rpc.deg_subscribe_events() while True: await asyncio.sleep(0.01) data = await rpc.reader.readline() try: data = json.loads(data) info[name] = data await self.queue.put(info) except: info[name] = {} await self.queue.put(info) await rpc.deg_switch(False) await rpc.stop() async def start_connect_slots(self, nodes): tasks = [] async with asyncio.TaskGroup() as tg: for i, node in enumerate(nodes): rpc = JsonRpc() subscribe = tg.create_task(self.subscribe( rpc, node)) nodes = tg.create_task(self.update_info()) async def update_info(self): while True: info = await self.queue.get() values = list(info.values())[0] if not values: self.model.add_offline(info) if 'result' in values: result = values.get('result') if 'eventgraph_info' in result: self.model.add_eg(info) if 'params' in values: self.model.add_event(info) self.queue.task_done() def main(self): logging.basicConfig(filename='deg.log', encoding='utf-8', level=logging.DEBUG) nodes = self.config.get('nodes') loop = urwid.MainLoop(self.view.ui, self.view.palette, unhandled_input=self.unhandled_input, event_loop=urwid.AsyncioEventLoop( loop=self.ev)) self.ev.create_task(self.start_connect_slots(nodes)) self.ev.create_task(self.view.update_view(self.ev, loop)) loop.run() def unhandled_input(self, key): if isinstance(key, tuple): return if key in ('q'): for task in asyncio.all_tasks(): task.cancel() raise urwid.ExitMainLoop() if __name__ == '__main__': deg = Degview() deg.main()