# Copyright 2018-2021 the .NET Foundation
# Licensed under the BSD license
"""
This file defines the WWT Qt widget.
Most of the code here deals with differences between WebEngine and WebKit
(either of which may be available in Qt) and also deals with figuring out how we
know once WWT is set up.
"""
import json
import time
from qtpy.QtWebEngineWidgets import QWebEngineView, QWebEnginePage, WEBENGINE
from qtpy import QtWidgets, QtGui, QtCore, PYQT6, PYSIDE6
from .app import get_qapp
from .core import BaseWWTWidget
from .logger import logger
from .data_server import get_data_server
__all__ = ['WWTQtClient']
APP_LIVENESS_DEADLINE = 10 # seconds
class WWTWebEngineView(QWebEngineView):
# Pass drag and drop events back up to the parent
# as this is needed for cases where applications
# embed a PyWWT Qt widget.
def dragEnterEvent(self, event):
if self.parent() is None:
super(WWTWebEngineView, self).dragEnterEvent(event)
else:
return self.parent().dragEnterEvent(event)
def dragMoveEvent(self, event):
if self.parent() is None:
super(WWTWebEngineView, self).dragMoveEvent(event)
else:
return self.parent().dragMoveEvent(event)
def dragLeaveEvent(self, event):
if self.parent() is None:
super(WWTWebEngineView, self).dragLeaveEvent(event)
else:
return self.parent().dragLeaveEvent(event)
def dropEvent(self, event):
if self.parent() is None:
super(WWTWebEngineView, self).dropEvent(event)
else:
return self.parent().dropEvent(event)
class WWTQWebEnginePage(QWebEnginePage):
"""
Subclass of QWebEnginePage that abstracts the JavaScript invocation
mechanism.
"""
app_message_callback = None
wwt_ready = QtCore.Signal()
"""A signal raised when the WWT app first becomes ready.
The glue-wwt plugin requires this signal, so we can't move or remove it.
"""
def __init__(self, parent=None):
super(WWTQWebEnginePage, self).__init__(parent=parent)
if WEBENGINE:
self._js_response_received = False
self._js_response = None
else:
self._frame = self.mainFrame()
if WEBENGINE:
def javaScriptConsoleMessage(
self,
level=None,
message=None,
line_number=None,
source_id=None
):
try:
context = 'level={0}, line_number={1}, source_id={2}'.format(level, line_number, source_id)
self._common_console_handler(message, context)
except: # noqa: E722
logger.exception('unhandled Python exception in Qt webengine javaScriptConsoleMessage') # noqa
def _process_js_response(self, result):
self._js_response_received = True
self._js_response = result
def runJavaScript(self, code):
app = get_qapp()
self._js_response_received = False
self._js_response = None
super(WWTQWebEnginePage, self).runJavaScript(code, self._process_js_response)
while not self._js_response_received:
app.processEvents()
return self._js_response
else:
def javaScriptConsoleMessage(
self,
message=None,
line_number=None,
source_id=None
):
try:
context = 'line_number={0}, source_id={1}'.format(line_number, source_id)
self._common_console_handler(message, context)
except: # noqa: E722
logger.exception('unhandled Python exception in Qt webkit javaScriptConsoleMessage') # noqa
def runJavaScript(self, code):
return self._frame.evaluateJavaScript(code)
def _common_console_handler(self, message, context):
if message.startswith('pywwtMessage:'):
try:
payload = json.loads(message[13:])
except Exception as e:
logger.warning('invalid pywwtMessage JSON: %s', e)
return
if self.app_message_callback is None:
logger.warning('received app message, but no handler available; message: %s', payload)
else:
try:
self.app_message_callback(payload)
except Exception:
logger.exception('error handling app message; payload: %s', payload) # noqa
else:
logger.debug('JS console message: %s (%s)', message, context)
class WWTQtWidget(QtWidgets.QWidget):
def __init__(self, url, parent=None):
super(WWTQtWidget, self).__init__(parent=parent)
self.page = WWTQWebEnginePage()
if PYQT6 or PYSIDE6:
self.web = WWTWebEngineView(self.page)
else:
self.web = WWTWebEngineView()
self.page.setView(self.web)
self.web.setPage(self.page)
self.web.setUrl(QtCore.QUrl(url))
layout = QtWidgets.QVBoxLayout()
layout.setContentsMargins(0, 0, 0, 0)
self.setLayout(layout)
layout.addWidget(self.web)
# More mouse-drag helpers
def dragEnterEvent(self, event):
if self.parent() is None:
super(WWTQtWidget, self).dragEnterEvent(event)
else:
return self.parent().dragEnterEvent(event)
def dragMoveEvent(self, event):
if self.parent() is None:
super(WWTQtWidget, self).dragMoveEvent(event)
else:
return self.parent().dragMoveEvent(event)
def dragLeaveEvent(self, event):
if self.parent() is None:
super(WWTQtWidget, self).dragLeaveEvent(event)
else:
return self.parent().dragLeaveEvent(event)
def dropEvent(self, event):
if self.parent() is None:
super(WWTQtWidget, self).dropEvent(event)
else:
return self.parent().dropEvent(event)
[docs]
class WWTQtClient(BaseWWTWidget):
"""
A client to create and drive the Qt widget.
Parameters
----------
block_until_ready : `bool`
Tells Python to wait for WorldWide Telescope to open before
proceeding with any following script (default: `True`).
size : `tuple`
Sets size of widget in pixels (default: (600, 600)).
hide_all_chrome : optional `bool`
Configures the WWT frontend to hide all user-interface "chrome".
Defaults to true to maintain compatibility with the historical
pywwt user experience.
"""
def __init__(self, block_until_ready=False, size=None, hide_all_chrome=True):
app = get_qapp()
self._data_server = get_data_server()
wwt_url = self._data_server.static_url('qtwrapper.html')
self.widget = WWTQtWidget(url=wwt_url)
if size is not None:
self.widget.resize(*size)
self.widget.page.app_message_callback = self._on_app_message
self.widget.show()
super(WWTQtClient, self).__init__(
hide_all_chrome=hide_all_chrome,
)
# Start polling for the app to start responding to messages
self._last_pong_timestamp = 0
self._timer = QtCore.QTimer()
self._timer.timeout.connect(self._check_ready)
self._timer.start(1000)
# TODO: this should be more generic
if block_until_ready:
while True:
app.processEvents()
if self._appAlive:
break
def _check_ready(self):
# If this Qt signal callback function raises an unhandled exception, it
# can crash the whole process! Which is ridiculous but let's try to do
# our part to make that not happen.
#
# Cf: https://doc.qt.io/qt-5/exceptionsafety.html#signals-and-slots
try:
self._check_ready_inner()
except: # noqa: E722
logger.exception('unhandled exception in Qt check-ready callback')
def _check_ready_inner(self):
# Send the ping. We have some extra paranoia here in case funky
# sequencing can happen in the stop() method.
if self.widget is not None and self.widget.page is not None:
self._actually_send_msg({
'type': 'wwt_ping_pong',
'sessionId': 'qt',
'threadId': str(time.time()),
})
# Evaluate pong status, with a hack for glue-wwt -- we need to emit the
# wwt_ready signal on our `page` field. It hardcodes the location of the
# field so we can't rationalize this API.
alive = (time.time() - self._last_pong_timestamp) < APP_LIVENESS_DEADLINE
if not self._appAlive and alive:
self.widget.page.wwt_ready.emit()
self._on_app_status_change(alive=alive)
def _on_app_message(self, payload):
ptype = payload.get('type')
if ptype == 'wwt_ping_pong':
try:
ts = float(payload['threadId'])
except Exception:
logger.exception('invalid timestamp in pywwt Qt pingpong response')
else:
self._last_pong_timestamp = ts
else:
self._on_app_message_received(payload)
[docs]
def wait(self, duration=None):
"""
Prevents WorldWide Telescope from closing once Python reaches the
end of a given script.
Parameters
----------
duration : int or float or None
How many seconds to wait for. By default, this waits until the
Qt window is closed.
"""
app = get_qapp()
if duration is None:
app.exec_()
else:
time1 = time.time()
while time.time() - time1 < duration:
app.processEvents()
def _actually_send_msg(self, payload):
jmsg = json.dumps(payload)
return self.widget.page.runJavaScript("pywwtSendMessage({0});".format(jmsg))
def _serve_file(self, filename, extension=''):
return self._data_server.serve_file(filename, extension=extension)
[docs]
def render(self, filename):
"""
Saves a screenshot of the viewer's current state.
Parameters
----------
filename : `str`
The desired name of the image file to be saved.
"""
image = QtGui.QImage(self.widget.size(), QtGui.QImage.Format_RGB32)
painter = QtGui.QPainter(image)
self.widget.render(painter)
image.save(filename)
painter.end()
[docs]
def close(self):
self._timer.stop()
self.widget.page = None
self.widget.web = None
self.widget.close()
self.widget = None
