improved a bit the web replay, disabled user movement
This commit is contained in:
parent
ca21da9f7a
commit
7e1467e7f1
6 changed files with 199 additions and 45 deletions
|
@ -1,4 +1,5 @@
|
||||||
import sys
|
import sys
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
from PyQt6.QtWidgets import QApplication
|
from PyQt6.QtWidgets import QApplication
|
||||||
|
|
||||||
|
@ -9,12 +10,16 @@ if __name__ == "__main__":
|
||||||
# create the application
|
# create the application
|
||||||
application = QApplication(sys.argv)
|
application = QApplication(sys.argv)
|
||||||
|
|
||||||
|
# TODO: cmd arguments ?
|
||||||
from source.utils import compress
|
from source.utils import compress
|
||||||
with open(r"C:\Users\RC606\Downloads\41a6268b-72e5-47a9-8106-6c15a0be366e.rsl", "rb") as file:
|
with open(r"C:\Users\RC606\PycharmProjects\M1-Recherche\results\c3376670-548d-4494-b963-dc2facf7a3d1.rsl", "rb") as file:
|
||||||
data = compress.uncompress_data(file.read())
|
data = compress.uncompress_data(file.read())
|
||||||
|
|
||||||
# create the window
|
# create the window
|
||||||
window = ReplayWindow(data["surveys"]["mission-gift-card"]["event"])
|
window = ReplayWindow(
|
||||||
|
datetime.fromtimestamp(data["time"]),
|
||||||
|
data["surveys"]["mission-game-dlc"]["event"]
|
||||||
|
)
|
||||||
window.show()
|
window.show()
|
||||||
|
|
||||||
# start the application
|
# start the application
|
||||||
|
|
|
@ -1,19 +1,25 @@
|
||||||
from PyQt6.QtCore import Qt, QUrl, QSize, QPointF
|
from datetime import datetime, timedelta
|
||||||
|
from typing import Callable
|
||||||
|
|
||||||
|
from PyQt6.QtCore import Qt, QUrl, QPointF, QTimer
|
||||||
from PyQt6.QtGui import QKeyEvent, QMouseEvent
|
from PyQt6.QtGui import QKeyEvent, QMouseEvent
|
||||||
from PyQt6.QtWebEngineWidgets import QWebEngineView
|
|
||||||
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QLabel, QApplication
|
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QLabel, QApplication
|
||||||
|
|
||||||
|
from tools.web_replay.ui import ReplayWebEngineView, ReplayInfo
|
||||||
|
|
||||||
|
|
||||||
class ReplayEngine(QWidget):
|
class ReplayEngine(QWidget):
|
||||||
"""
|
"""
|
||||||
This widget allow to replay some event that occurred on a web page
|
This widget allow to replay some event that occurred on a web page
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, replay_data: dict):
|
def __init__(self, start_time: datetime, replay_data: list):
|
||||||
super().__init__()
|
super().__init__()
|
||||||
|
|
||||||
|
self.start_time = start_time - timedelta(days=1) # remove a day to prevent archive rounding
|
||||||
self.replay_data = replay_data
|
self.replay_data = replay_data
|
||||||
self.iterator = iter(self.replay_data)
|
self.replay_index: int = 0
|
||||||
|
self.replay_time: float = 0
|
||||||
|
|
||||||
# layout
|
# layout
|
||||||
self._layout = QVBoxLayout()
|
self._layout = QVBoxLayout()
|
||||||
|
@ -25,97 +31,155 @@ class ReplayEngine(QWidget):
|
||||||
self.cursor.setStyleSheet("background-color: red; border-radius: 10px;")
|
self.cursor.setStyleSheet("background-color: red; border-radius: 10px;")
|
||||||
|
|
||||||
# web
|
# web
|
||||||
self.web = QWebEngineView()
|
self.web = ReplayWebEngineView(self.start_time)
|
||||||
self._layout.addWidget(self.web)
|
self._layout.addWidget(self.web, 1)
|
||||||
|
|
||||||
|
# information
|
||||||
|
self.information = ReplayInfo()
|
||||||
|
self._layout.addWidget(self.information)
|
||||||
|
|
||||||
|
# event timer
|
||||||
|
self.timer = QTimer()
|
||||||
|
|
||||||
|
def run_event(self, event: dict, callback: Callable):
|
||||||
|
# TODO: check if click are done correctly, check position, if event are correct, ...
|
||||||
|
|
||||||
def run_event(self, event: dict):
|
|
||||||
match event["type"]:
|
match event["type"]:
|
||||||
case "success":
|
case "success":
|
||||||
# success event
|
# success event
|
||||||
print(f"success ! ({event['time']}s)")
|
print(f"success ! ({event['time']}s)")
|
||||||
|
|
||||||
|
callback()
|
||||||
|
|
||||||
case "url":
|
case "url":
|
||||||
# changing url event
|
# changing url event
|
||||||
self.web.setUrl(QUrl(event["url"]))
|
self.web.setUrl(QUrl(event["url"]))
|
||||||
|
|
||||||
|
# callback
|
||||||
|
self.web.loadFinished.connect(lambda ok: callback()) # NOQA: connect exist
|
||||||
|
|
||||||
case "resize":
|
case "resize":
|
||||||
# changing widget size event
|
# changing widget size event
|
||||||
self.web.resize(QSize(*event["size"]))
|
w, h = event["size"]
|
||||||
|
zoom_factor: float = self.web.width() / w
|
||||||
|
self.web.setZoomFactor(zoom_factor)
|
||||||
|
|
||||||
# TODO: better way ?
|
# callback
|
||||||
self.window().resize(QSize(*event["size"]))
|
callback()
|
||||||
|
|
||||||
case "keyboard_press":
|
case "keyboard_press":
|
||||||
# keyboard key pressed event
|
# keyboard key pressed event
|
||||||
key = QKeyEvent(
|
qevent = QKeyEvent(
|
||||||
QKeyEvent.Type.KeyPress,
|
QKeyEvent.Type.KeyPress,
|
||||||
event["key"],
|
event["key"],
|
||||||
Qt.KeyboardModifier.NoModifier
|
Qt.KeyboardModifier.NoModifier
|
||||||
)
|
)
|
||||||
QApplication.sendEvent(self.web.page(), key)
|
qevent.custom = True
|
||||||
|
QApplication.postEvent(self.web.focusProxy(), qevent)
|
||||||
|
|
||||||
|
# callback
|
||||||
|
callback()
|
||||||
|
|
||||||
case "keyboard_release":
|
case "keyboard_release":
|
||||||
# keyboard key released event
|
# keyboard key released event
|
||||||
key = QKeyEvent(
|
qevent = QKeyEvent(
|
||||||
QKeyEvent.Type.KeyRelease,
|
QKeyEvent.Type.KeyRelease,
|
||||||
event["key"],
|
event["key"],
|
||||||
Qt.KeyboardModifier.NoModifier
|
Qt.KeyboardModifier.NoModifier
|
||||||
)
|
)
|
||||||
QApplication.sendEvent(self.web.page(), key)
|
qevent.custom = True
|
||||||
|
QApplication.postEvent(self.web.focusProxy(), qevent)
|
||||||
|
|
||||||
|
# callback
|
||||||
|
callback()
|
||||||
|
|
||||||
case "mouse_press":
|
case "mouse_press":
|
||||||
# mouse pressed event
|
# mouse pressed event
|
||||||
key = QMouseEvent(
|
qevent = QMouseEvent(
|
||||||
QMouseEvent.Type.KeyPress,
|
QMouseEvent.Type.KeyPress,
|
||||||
QPointF(*event["position"]),
|
QPointF(*event["position"]) / self.web.zoomFactor(),
|
||||||
Qt.MouseButton(event["button"]),
|
Qt.MouseButton(event["button"]),
|
||||||
Qt.MouseButton.NoButton,
|
Qt.MouseButton.NoButton,
|
||||||
Qt.KeyboardModifier.NoModifier
|
Qt.KeyboardModifier.NoModifier
|
||||||
)
|
)
|
||||||
QApplication.sendEvent(self.web.page(), key)
|
qevent.custom = True
|
||||||
|
QApplication.postEvent(self.web.focusProxy(), qevent)
|
||||||
|
|
||||||
|
# callback
|
||||||
|
callback()
|
||||||
|
|
||||||
case "mouse_release":
|
case "mouse_release":
|
||||||
# mouse pressed event
|
# mouse pressed event
|
||||||
key = QMouseEvent(
|
qevent = QMouseEvent(
|
||||||
QMouseEvent.Type.KeyRelease,
|
QMouseEvent.Type.KeyRelease,
|
||||||
QPointF(*event["position"]),
|
QPointF(*event["position"]) / self.web.zoomFactor(),
|
||||||
Qt.MouseButton(event["button"]),
|
Qt.MouseButton(event["button"]),
|
||||||
Qt.MouseButton.NoButton,
|
Qt.MouseButton.NoButton,
|
||||||
Qt.KeyboardModifier.NoModifier
|
Qt.KeyboardModifier.NoModifier
|
||||||
)
|
)
|
||||||
QApplication.sendEvent(self.web.page(), key)
|
qevent.custom = True
|
||||||
|
QApplication.postEvent(self.web.focusProxy(), qevent)
|
||||||
|
|
||||||
# NOTE: this event is redundant
|
# callback
|
||||||
# case "mouse_double_click":
|
callback()
|
||||||
# # mouse double-clicked event
|
|
||||||
# key = QMouseEvent(QMouseEvent.Type.MouseButtonDblClick, event["position"], event["button"])
|
|
||||||
# QApplication.sendEvent(self.page(), key)
|
|
||||||
|
|
||||||
case "mouse_move":
|
case "mouse_move":
|
||||||
# mouse moved event
|
# mouse moved event
|
||||||
key = QMouseEvent(
|
qevent = QMouseEvent(
|
||||||
QMouseEvent.Type.KeyRelease,
|
QMouseEvent.Type.KeyRelease,
|
||||||
QPointF(*event["position"]),
|
QPointF(*event["position"]) / self.web.zoomFactor(),
|
||||||
Qt.MouseButton.NoButton,
|
Qt.MouseButton.NoButton,
|
||||||
Qt.MouseButton.NoButton,
|
Qt.MouseButton.NoButton,
|
||||||
Qt.KeyboardModifier.NoModifier
|
Qt.KeyboardModifier.NoModifier
|
||||||
)
|
)
|
||||||
QApplication.sendEvent(self.web.page(), key)
|
qevent.custom = True
|
||||||
|
QApplication.postEvent(self.web.focusProxy(), qevent)
|
||||||
|
|
||||||
# move the fake cursor
|
# move the fake cursor
|
||||||
self.cursor.move(QPointF(*event["position"]).toPoint() - self.cursor.rect().center())
|
self.cursor.move(QPointF(*event["position"]).toPoint() - self.cursor.rect().center())
|
||||||
self.cursor.raise_()
|
self.cursor.raise_()
|
||||||
|
|
||||||
|
# callback
|
||||||
|
callback()
|
||||||
|
|
||||||
case "scroll":
|
case "scroll":
|
||||||
# scroll event
|
# scroll event
|
||||||
x, y = event["position"]
|
x, y = event["position"]
|
||||||
self.web.page().runJavaScript(f"window.scrollTo({x}, {y});")
|
self.web.page().runJavaScript(
|
||||||
|
f"window.scrollTo({x}, {y});",
|
||||||
|
resultCallback=lambda result: callback()
|
||||||
|
)
|
||||||
|
|
||||||
def next(self):
|
def next(self):
|
||||||
try:
|
# get event information
|
||||||
event = next(self.iterator)
|
event = self.replay_data[self.replay_index]
|
||||||
except StopIteration:
|
self.replay_time = event["time"]
|
||||||
print("end of record")
|
self.replay_index = self.replay_index + 1
|
||||||
return
|
|
||||||
|
|
||||||
self.run_event(event)
|
# set text
|
||||||
|
self.information.set_description(f"{event}")
|
||||||
|
|
||||||
|
# run the event
|
||||||
|
self.run_event(
|
||||||
|
event,
|
||||||
|
self._next_callback
|
||||||
|
)
|
||||||
|
|
||||||
|
def _next_callback(self):
|
||||||
|
# prevent the web loading to call this function again
|
||||||
|
try:
|
||||||
|
self.web.loadFinished.disconnect(self._next_callback) # NOQA: disconnect exist
|
||||||
|
except TypeError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
# if there are still events after this one
|
||||||
|
if self.replay_index < len(self.replay_data):
|
||||||
|
# next event
|
||||||
|
next_event: dict = self.replay_data[self.replay_index]
|
||||||
|
next_time: float = next_event["time"]
|
||||||
|
|
||||||
|
# prepare the timer to play the event at the corresponding time
|
||||||
|
self.timer.singleShot(
|
||||||
|
round((next_time - self.replay_time) / 1000),
|
||||||
|
self.next
|
||||||
|
)
|
||||||
|
|
19
tools/web_replay/ui/ReplayNavigation.py
Normal file
19
tools/web_replay/ui/ReplayNavigation.py
Normal file
|
@ -0,0 +1,19 @@
|
||||||
|
from PyQt6.QtCore import Qt
|
||||||
|
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QLabel
|
||||||
|
|
||||||
|
|
||||||
|
class ReplayNavigation(QWidget):
|
||||||
|
def __init__(self):
|
||||||
|
super().__init__()
|
||||||
|
|
||||||
|
# layout
|
||||||
|
layout = QVBoxLayout()
|
||||||
|
self.setLayout(layout)
|
||||||
|
|
||||||
|
# information of the replay
|
||||||
|
self._description = QLabel()
|
||||||
|
layout.addWidget(self._description)
|
||||||
|
self._description.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||||
|
|
||||||
|
def set_description(self, text: str):
|
||||||
|
self._description.setText(text)
|
59
tools/web_replay/ui/ReplayWebEngineView.py
Normal file
59
tools/web_replay/ui/ReplayWebEngineView.py
Normal file
|
@ -0,0 +1,59 @@
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
from PyQt6.QtCore import QObject, QEvent, QUrl
|
||||||
|
from PyQt6.QtWebEngineWidgets import QWebEngineView
|
||||||
|
|
||||||
|
|
||||||
|
class ReplayWebEngineView(QWebEngineView):
|
||||||
|
def __init__(self, start_time: datetime):
|
||||||
|
super().__init__()
|
||||||
|
|
||||||
|
self.start_time = start_time
|
||||||
|
|
||||||
|
self.loadFinished.connect(self._initialize_proxy_event) # NOQA: connect exist
|
||||||
|
|
||||||
|
# event filter
|
||||||
|
|
||||||
|
def setUrl(self, url: QUrl) -> None:
|
||||||
|
# get the archive.org link corresponding to that time
|
||||||
|
archive_time: str = self.start_time.strftime("%Y%m%d%H%M%S")
|
||||||
|
archive_url = f"https://web.archive.org/web/{archive_time}/{url.toString()}"
|
||||||
|
|
||||||
|
# call the super function with the archive url instead
|
||||||
|
super().setUrl(QUrl(archive_url))
|
||||||
|
|
||||||
|
# clean the archive header popup that will appear
|
||||||
|
self.loadFinished.connect(self._clean_archive_header) # NOQA: connect exist
|
||||||
|
|
||||||
|
def eventFilter(self, obj: QObject, event: QEvent) -> bool:
|
||||||
|
match event.type():
|
||||||
|
# allow scroll events (they are created automatically)
|
||||||
|
case event.Type.Scroll:
|
||||||
|
pass
|
||||||
|
|
||||||
|
# allow timed events
|
||||||
|
case event.Type.Timer:
|
||||||
|
pass
|
||||||
|
|
||||||
|
# ignore all other events
|
||||||
|
case _:
|
||||||
|
if not getattr(event, "custom", False):
|
||||||
|
return True
|
||||||
|
|
||||||
|
return super().eventFilter(obj, event)
|
||||||
|
|
||||||
|
# events
|
||||||
|
|
||||||
|
def _initialize_proxy_event(self, ok: bool):
|
||||||
|
# prevent the event from being enabled another time
|
||||||
|
self.loadFinished.disconnect(self._initialize_proxy_event) # NOQA: disconnect exist
|
||||||
|
|
||||||
|
# make self.eventFilter intercept all focusProxy events
|
||||||
|
self.focusProxy().installEventFilter(self)
|
||||||
|
|
||||||
|
def _clean_archive_header(self, ok: bool):
|
||||||
|
# prevent the event from being enabled another time
|
||||||
|
self.loadFinished.disconnect(self._clean_archive_header) # NOQA: disconnect exist
|
||||||
|
|
||||||
|
# hide archive.org header to avoid mouse movement being shifted
|
||||||
|
self.page().runJavaScript("document.getElementById('wm-ipp-base').style.display = 'none';")
|
|
@ -1,18 +1,23 @@
|
||||||
from PyQt6.QtCore import QTimer
|
from datetime import datetime
|
||||||
|
|
||||||
from PyQt6.QtWidgets import QMainWindow
|
from PyQt6.QtWidgets import QMainWindow
|
||||||
|
|
||||||
from tools.web_replay.ui import ReplayEngine
|
from tools.web_replay.ui import ReplayEngine
|
||||||
|
|
||||||
|
|
||||||
class ReplayWindow(QMainWindow):
|
class ReplayWindow(QMainWindow):
|
||||||
def __init__(self, replay_data: dict):
|
def __init__(self, start_time: datetime, replay_data: list):
|
||||||
super().__init__()
|
super().__init__()
|
||||||
|
|
||||||
self.replay_engine = ReplayEngine(replay_data)
|
# decoration
|
||||||
|
self.setWindowTitle("Survey Engine - Web Replay")
|
||||||
|
|
||||||
|
# setup the engine
|
||||||
|
self.replay_engine = ReplayEngine(start_time, replay_data)
|
||||||
self.setCentralWidget(self.replay_engine)
|
self.setCentralWidget(self.replay_engine)
|
||||||
|
|
||||||
# TODO: TEST REMOVE
|
# show the window as fullscreen
|
||||||
self.timer = QTimer()
|
self.showFullScreen()
|
||||||
self.timer.setInterval(10)
|
|
||||||
self.timer.timeout.connect(self.replay_engine.next) # NOQA: connect exist
|
# TODO: remove ?
|
||||||
self.timer.start()
|
self.replay_engine.next() # play the replay
|
||||||
|
|
|
@ -1,2 +1,4 @@
|
||||||
|
from .ReplayWebEngineView import ReplayWebEngineView
|
||||||
|
from .ReplayNavigation import ReplayNavigation
|
||||||
from .ReplayEngine import ReplayEngine
|
from .ReplayEngine import ReplayEngine
|
||||||
from .ReplayWindow import ReplayWindow
|
from .ReplayWindow import ReplayWindow
|
||||||
|
|
Loading…
Reference in a new issue