import sys
import json
import math
from PySide2 import QtCore, QtGui, QtWidgets, QtNetwork, QtWebSockets, QtSvg
from torba.workbench._output_dock import Ui_OutputDock as OutputDock
from torba.workbench._blockchain_dock import Ui_BlockchainDock as BlockchainDock
def dict_to_post_data(d):
query = QtCore.QUrlQuery()
for key, value in d.items():
query.addQueryItem(str(key), str(value))
return QtCore.QByteArray(query.toString().encode())
class LoggingOutput(QtWidgets.QDockWidget, OutputDock):
def __init__(self, title, parent):
super().__init__(parent)
self.setupUi(self)
self.setWindowTitle(title)
class BlockchainControls(QtWidgets.QDockWidget, BlockchainDock):
def __init__(self, parent):
super().__init__(parent)
self.setupUi(self)
self.generate.clicked.connect(self.on_generate)
self.transfer.clicked.connect(self.on_transfer)
def on_generate(self):
print('generating')
self.parent().run_command('generate', blocks=self.blocks.value())
def on_transfer(self):
print('transfering')
self.parent().run_command('transfer', amount=self.amount.value())
class Arrow(QtWidgets.QGraphicsLineItem):
def __init__(self, start_node, end_node, parent=None, scene=None):
super().__init__(parent, scene)
self.start_node = start_node
self.start_node.connect_arrow(self)
self.end_node = end_node
self.end_node.connect_arrow(self)
self.arrow_head = QtGui.QPolygonF()
self.setFlag(QtWidgets.QGraphicsItem.ItemIsSelectable, True)
self.setZValue(-1000.0)
self.arrow_color = QtCore.Qt.black
self.setPen(QtGui.QPen(
self.arrow_color, 2, QtCore.Qt.SolidLine, QtCore.Qt.RoundCap, QtCore.Qt.RoundJoin
))
def boundingRect(self):
extra = (self.pen().width() + 20) / 2.0
p1 = self.line().p1()
p2 = self.line().p2()
size = QtCore.QSizeF(p2.x() - p1.x(), p2.y() - p1.y())
return QtCore.QRectF(p1, size).normalized().adjusted(-extra, -extra, extra, extra)
def shape(self):
path = super().shape()
path.addPolygon(self.arrow_head)
return path
def update_position(self):
line = QtCore.QLineF(
self.mapFromItem(self.start_node, 0, 0),
self.mapFromItem(self.end_node, 0, 0)
)
self.setLine(line)
def paint(self, painter, option, widget=None):
if self.start_node.collidesWithItem(self.end_node):
return
start_node = self.start_node
end_node = self.end_node
color = self.arrow_color
pen = self.pen()
pen.setColor(self.arrow_color)
arrow_size = 20.0
painter.setPen(pen)
painter.setBrush(self.arrow_color)
end_rectangle = end_node.sceneBoundingRect()
start_center = start_node.sceneBoundingRect().center()
end_center = end_rectangle.center()
center_line = QtCore.QLineF(start_center, end_center)
end_polygon = QtGui.QPolygonF(end_rectangle)
p1 = end_polygon.at(0)
intersect_point = QtCore.QPointF()
for p2 in end_polygon:
poly_line = QtCore.QLineF(p1, p2)
intersect_type, intersect_point = poly_line.intersect(center_line)
if intersect_type == QtCore.QLineF.BoundedIntersection:
break
p1 = p2
self.setLine(QtCore.QLineF(intersect_point, start_center))
line = self.line()
angle = math.acos(line.dx() / line.length())
if line.dy() >= 0:
angle = (math.pi * 2.0) - angle
arrow_p1 = line.p1() + QtCore.QPointF(
math.sin(angle + math.pi / 3.0) * arrow_size,
math.cos(angle + math.pi / 3.0) * arrow_size
)
arrow_p2 = line.p1() + QtCore.QPointF(
math.sin(angle + math.pi - math.pi / 3.0) * arrow_size,
math.cos(angle + math.pi - math.pi / 3.0) * arrow_size
)
self.arrow_head.clear()
for point in [line.p1(), arrow_p1, arrow_p2]:
self.arrow_head.append(point)
painter.drawLine(line)
painter.drawPolygon(self.arrow_head)
if self.isSelected():
painter.setPen(QtGui.QPen(color, 1, QtCore.Qt.DashLine))
line = QtCore.QLineF(line)
line.translate(0, 4.0)
painter.drawLine(line)
line.translate(0, -8.0)
painter.drawLine(line)
ONLINE_COLOR = "limegreen"
OFFLINE_COLOR = "lightsteelblue"
class NodeItem(QtSvg.QGraphicsSvgItem):
def __init__(self, context_menu):
super().__init__()
self._port = ''
self._color = OFFLINE_COLOR
self.context_menu = context_menu
self.arrows = set()
self.renderer = QtSvg.QSvgRenderer()
self.update_svg()
self.setSharedRenderer(self.renderer)
#self.setScale(2.0)
#self.setTransformOriginPoint(24, 24)
self.setFlag(QtWidgets.QGraphicsItem.ItemIsMovable, True)
self.setFlag(QtWidgets.QGraphicsItem.ItemIsSelectable, True)
def get_svg(self):
return self.SVG.format(
port=self.port,
color=self._color
)
def update_svg(self):
self.renderer.load(QtCore.QByteArray(self.get_svg().encode()))
self.update()
@property
def port(self):
return self._port
@port.setter
def port(self, port):
self._port = port
self.update_svg()
@property
def online(self):
return self._color == ONLINE_COLOR
@online.setter
def online(self, online):
if online:
self._color = ONLINE_COLOR
else:
self._color = OFFLINE_COLOR
self.update_svg()
def connect_arrow(self, arrow):
self.arrows.add(arrow)
def disconnect_arrow(self, arrow):
self.arrows.discard(arrow)
def contextMenuEvent(self, event):
self.scene().clearSelection()
self.setSelected(True)
self.myContextMenu.exec_(event.screenPos())
def itemChange(self, change, value):
if change == QtWidgets.QGraphicsItem.ItemPositionChange:
for arrow in self.arrows:
arrow.update_position()
return value
class BlockchainNode(NodeItem):
SVG = """
"""
def __init__(self, *args):
self._block_height = ''
super().__init__(*args)
@property
def block_height(self):
return self._block_height
@block_height.setter
def block_height(self, block_height):
self._block_height = block_height
self.update_svg()
def get_svg(self):
return self.SVG.format(
port=self.port,
block=self.block_height,
color=self._color
)
class SPVNode(NodeItem):
SVG = """
"""
def __init__(self, *args):
super().__init__(*args)
class WalletNode(NodeItem):
SVG = """
"""
def __init__(self, *args):
self._coins = '--'
super().__init__(*args)
@property
def coins(self):
return self._coins
@coins.setter
def coins(self, coins):
self._coins = coins
self.update_svg()
def get_svg(self):
return self.SVG.format(
coins=self.coins,
color=self._color
)
class Stage(QtWidgets.QGraphicsScene):
def __init__(self, parent):
super().__init__(parent)
self.blockchain = b = BlockchainNode(None)
b.port = ''
b.block_height = ''
b.setZValue(0)
b.setPos(-25, -100)
self.addItem(b)
self.spv = s = SPVNode(None)
s.port = ''
s.setZValue(0)
self.addItem(s)
s.setPos(-10, -10)
self.wallet = w = WalletNode(None)
w.coins = ''
w.setZValue(0)
w.update_svg()
self.addItem(w)
w.setPos(0, 100)
self.addItem(Arrow(b, s))
self.addItem(Arrow(s, w))
class Orchstr8Workbench(QtWidgets.QMainWindow):
def __init__(self):
super().__init__()
self.stage = Stage(self)
self.view = QtWidgets.QGraphicsView(self.stage)
self.status_bar = QtWidgets.QStatusBar(self)
self.setWindowTitle('Orchstr8 Workbench')
self.setCentralWidget(self.view)
self.setStatusBar(self.status_bar)
self.block_height = self.make_status_label('Height: -- ')
self.user_balance = self.make_status_label('User Balance: -- ')
self.mining_balance = self.make_status_label('Mining Balance: -- ')
self.wallet_log = LoggingOutput('Wallet', self)
self.addDockWidget(QtCore.Qt.LeftDockWidgetArea, self.wallet_log)
self.spv_log = LoggingOutput('SPV Server', self)
self.addDockWidget(QtCore.Qt.LeftDockWidgetArea, self.spv_log)
self.blockchain_log = LoggingOutput('Blockchain', self)
self.addDockWidget(QtCore.Qt.LeftDockWidgetArea, self.blockchain_log)
self.blockchain_controls = BlockchainControls(self)
self.addDockWidget(QtCore.Qt.RightDockWidgetArea, self.blockchain_controls)
self.network = QtNetwork.QNetworkAccessManager(self)
self.socket = QtWebSockets.QWebSocket()
self.socket.connected.connect(lambda: self.run_command('start'))
self.socket.error.connect(lambda e: print(f'errored: {e}'))
self.socket.textMessageReceived.connect(self.on_message)
self.socket.open('ws://localhost:7954/log')
def make_status_label(self, text):
label = QtWidgets.QLabel(text)
label.setFrameStyle(QtWidgets.QLabel.Panel | QtWidgets.QLabel.Sunken)
self.status_bar.addPermanentWidget(label)
return label
def on_message(self, text):
msg = json.loads(text)
if msg['type'] == 'status':
self.stage.wallet.coins = msg['balance']
self.stage.blockchain.block_height = msg['height']
self.block_height.setText(f"Height: {msg['height']} ")
self.user_balance.setText(f"User Balance: {msg['balance']} ")
self.mining_balance.setText(f"Mining Balance: {msg['miner']} ")
elif msg['type'] == 'service':
node = {
'blockchain': self.stage.blockchain,
'spv': self.stage.spv,
'wallet': self.stage.wallet
}[msg['name']]
node.online = True
node.port = f":{msg['port']}"
elif msg['type'] == 'log':
log = {
'blockchain': self.blockchain_log,
'electrumx': self.spv_log,
'lbryumx': self.spv_log,
'Controller': self.spv_log,
'LBRYBlockProcessor': self.spv_log,
'LBCDaemon': self.spv_log,
}.get(msg['name'].split('.')[-1], self.wallet_log)
log.textEdit.append(msg['message'])
def run_command(self, command, **kwargs):
request = QtNetwork.QNetworkRequest(QtCore.QUrl('http://localhost:7954/'+command))
request.setHeader(QtNetwork.QNetworkRequest.ContentTypeHeader, "application/x-www-form-urlencoded")
reply = self.network.post(request, dict_to_post_data(kwargs))
# reply.finished.connect(cb)
reply.error.connect(self.on_command_error)
@staticmethod
def on_command_error(error):
print('failed executing command:')
print(error)
def main():
app = QtWidgets.QApplication(sys.argv)
workbench = Orchstr8Workbench()
workbench.setGeometry(100, 100, 1200, 600)
workbench.show()
return app.exec_()
if __name__ == "__main__":
sys.exit(main())