QWebChannel Example Script

qwebchannelexample.py

"""create a remote connection to a webapp."""

import sys
import json

from PySide6.QtWidgets import QApplication, QMainWindow
from PySide6.QtNetwork import QHostAddress, QSslSocket
from PySide6.QtWebChannel import QWebChannel, QWebChannelAbstractTransport
from PySide6.QtWebSockets import QWebSocketServer
from PySide6.QtWebEngineWidgets import QWebEngineView
from PySide6.QtCore import QObject, Signal, QByteArray, QJsonDocument, Slot, QUrl


class WebSocketTransport(QWebChannelAbstractTransport):
    """QWebChannelAbstractSocket implementation using a QWebSocket internally
        The transport delegates all messages received over the QWebSocket over
        its textMessageReceived signal. Analogously, all calls to
        sendTextMessage will be sent over the QWebSocket to the remote client.
    """

    def __init__(self, socket):
        """Construct the transport object and wrap the given socket.
           The socket is also set as the parent of the transport object."""
        super().__init__(socket)
        self._socket = socket
        self._socket.textMessageReceived.connect(self.text_message_received)
        self._socket.disconnected.connect(self._disconnected)

    def __del__(self):
        """Destroys the WebSocketTransport."""
        self._socket.deleteLater()

    def _disconnected(self):
        self.deleteLater()

    def sendMessage(self, message):
        """Serialize the JSON message and send it as a text message via the
           WebSocket to the client."""
        doc = QJsonDocument(message)
        json_message = str(doc.toJson(QJsonDocument.Compact), "utf-8")
        self._socket.sendTextMessage(json_message)

    @Slot(str)
    def text_message_received(self, message_data_in):
        """Deserialize the stringified JSON messageData and emit
           messageReceived."""
        message_data = QByteArray(bytes(message_data_in, encoding='utf8'))
        message = QJsonDocument.fromJson(message_data)
        if message.isNull():
            print("Failed to parse text message as JSON object:", message_data)
            return
        if not message.isObject():
            print("Received JSON message that is not an object: ", message_data)
            return
        self.messageReceived.emit(message.object(), self)



class WebSocketClientWrapper(QObject):
    """Wraps connected QWebSockets clients in WebSocketTransport objects.

       This code is all that is required to connect incoming WebSockets to
       the WebChannel. Any kind of remote JavaScript client that supports
       WebSockets can thus receive messages and access the published objects.
    """
    client_connected = Signal(WebSocketTransport)

    def __init__(self, server, parent=None):
        """Construct the client wrapper with the given parent. All clients
           connecting to the QWebSocketServer will be automatically wrapped
           in WebSocketTransport objects."""
        super().__init__(parent)
        self._server = server
        self._server.newConnection.connect(self.handle_new_connection)
        self._transports = []

    @Slot()
    def handle_new_connection(self):
        """Wrap an incoming WebSocket connection in a WebSocketTransport
           object."""
        socket = self._server.nextPendingConnection()
        transport = WebSocketTransport(socket)
        self._transports.append(transport)
        self.client_connected.emit(transport)


class Core(QObject):
    """An instance of this class gets published over the WebChannel."""

    sendText = Signal(str)

    def __init__(self):
        """Initialize the QObject."""
        super().__init__()

    @Slot(str)
    def receiveText(self, text):
        """Slot to interactively send messages from client."""
        data = json.loads(text)
        action = (data["action"])

        if action == "ADD_TOOL":
            tool = data["payload"]["tool"]

            # fusion is injected into the code by the host software in this case    
            comp = fusion.GetCurrentComp()
            comp.AddTool(tool)
            self.sendText.emit(
                json.dumps({"data": {},
                            "status": 'success',
                            "message": 'Successfully added a tool'}))


class MainWindow(QMainWindow):
    """Create main window and load webapp."""

    def __init__(self):
        """Initialize main window."""
        super().__init__()
        self.webEngineView = QWebEngineView()
        self.setCentralWidget(self.webEngineView)
        self.webEngineView.load(
            QUrl("https://622d303119d5fca85c85049d.clutch.host/"))

if __name__ == '__main__':
    app = QApplication(sys.argv)

    if not QSslSocket.supportsSsl():
        print('This app requires SSL support.')
        sys.exit(-1)

    # setup the QWebSocketServer
    server = QWebSocketServer("QWebChannel PySide Example",
                              QWebSocketServer.NonSecureMode)

    if not server.listen(QHostAddress.LocalHost, 12345):
        print("Failed to open web socket server.")
        sys.exit(-1)

    # wrap WebSocket clients in QWebChannelAbstractTransport objects
    client_wrapper = WebSocketClientWrapper(server)

    # setup the channel
    channel = QWebChannel()
    client_wrapper.client_connected.connect(channel.connectTo)

    core = Core()
    channel.registerObject("core", core)

    # setup the WebView
    mainWindow = MainWindow()

    mainWindow.show()

    sys.exit(app.exec())