Skip to content

Instantly share code, notes, and snippets.

@vidjuheffex
Last active June 2, 2024 09:52
Show Gist options
  • Save vidjuheffex/a9f352334b80c4a1a2f0e14da23fce04 to your computer and use it in GitHub Desktop.
Save vidjuheffex/a9f352334b80c4a1a2f0e14da23fce04 to your computer and use it in GitHub Desktop.
QWebChannel Example Script
"""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())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment