import time from typing import Optional, Any from PyQt6.QtCore import Qt, QTimer, pyqtSignal, QUrl, QEvent, QObject, QPointF from PyQt6.QtGui import QFont, QMouseEvent, QResizeEvent, QKeyEvent from PyQt6.QtWidgets import QLabel, QVBoxLayout, QSizePolicy from source import translate, widget from source.survey.base import BaseSurvey class WebMission(BaseSurvey): def __init__( self, title: translate.Translatable, url: str, check_condition: Optional[str] = None, skip_time: Optional[float] = None, signals: dict[str, pyqtSignal] = None ): super().__init__() self.check_condition = check_condition self.default_url = url self.skip_time = skip_time self.signals = signals if signals is not None else {} self._finished = False # set layout self._layout = QVBoxLayout() self.setLayout(self._layout) # data collection self.start_time = time.time() self._collected_events: list[dict[str, Any]] = [] # mission title self.label_title = QLabel() self._layout.addWidget(self.label_title) self.label_title.setText(translate.translate(title)) self.label_title.setAlignment(Qt.AlignmentFlag.AlignCenter) font_title = self.label_title.font() font_title.setPointSize(24) font_title.setWeight(QFont.Weight.Bold) self.label_title.setFont(font_title) # web page self.browser = widget.Browser() self._layout.addWidget(self.browser) self.browser.web.focusProxy().installEventFilter(self) # capture the event in eventFilter self.browser.web.urlChanged.connect(self._on_url_changed) # NOQA: connect exist self.browser.web.setUrl(QUrl(self.default_url)) self.browser.web.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) # setup the timer for the check self.timer_check = None if self.check_condition is not None: self.timer_check = QTimer() self.timer_check.setInterval(100) self.timer_check.timeout.connect(self.check) # NOQA: connect exist # setup the timer for skipping the mission self.timer_skip = None if self.skip_time is not None: self.timer_skip = QTimer() self.timer_skip.setInterval(self.skip_time * 1000) self.timer_skip.timeout.connect(self._on_time_skip) # NOQA: connect exist # setup the events self.browser.web.page().scrollPositionChanged.connect(self._on_scroll_position_changed) # navigation self.navigation = widget.SurveyNavigation(signals=signals) self._layout.addWidget(self.navigation) # initialize the start time self.start_time = time.time() # check timer if self.timer_check is not None: # enable the timer self.timer_check.start() else: self._success() # call directly the success method # skip timer if self.timer_skip is not None: # enable the timer for skipping the question self.timer_skip.start() @classmethod def from_dict(cls, data: dict[str, Any], signals: dict[str, pyqtSignal]) -> "WebMission": return cls( title=data["title"], url=data.get("url"), check_condition=data.get("check"), skip_time=data.get("skip_time"), signals=signals ) # events def _on_scroll_position_changed(self, position: QPointF): self._save_event(type="scroll", position=[position.x(), position.y()]) def eventFilter(self, obj: QObject, event: QEvent) -> bool: if obj is self.browser.web.focusProxy() and not self._finished: # if the object is the content of the web engine widget match event.type(): case QEvent.Type.MouseMove: # if this is a mouse movement event: QMouseEvent position = event.position() self._save_event( type="mouse_move", position=[position.x(), position.y()], ) case QEvent.Type.MouseButtonPress: # if this is a mouse click press event: QMouseEvent position = event.position() self._save_event( type="mouse_press", position=[position.x(), position.y()], button=event.button().value, ) case QEvent.Type.MouseButtonRelease: # if this is a mouse click release event: QMouseEvent position = event.position() self._save_event( type="mouse_release", position=[position.x(), position.y()], button=event.button().value, ) case QEvent.Type.MouseButtonDblClick: # if this is a mouse double click event: QMouseEvent position = event.position() self._save_event( type="mouse_double_click", position=[position.x(), position.y()], ) case QEvent.Type.KeyPress: # when the keyboard is pressed event: QKeyEvent self._save_event( type="keyboard_press", key=event.key(), ) case QEvent.Type.KeyRelease: # when the keyboard is released event: QKeyEvent self._save_event( type="keyboard_release", key=event.key(), ) case QEvent.Type.Resize: # if the window got resized event: QResizeEvent size = event.size() self._save_event( type="resize", size=[size.width(), size.height()], ) return super().eventFilter(obj, event) def _success(self): if self._finished: return # mark the success in the events self._save_event(type="check") # emit on the success signal if "success" in self.signals: self.signals["success"].emit() # NOQA: emit exist # mark the mission as finished self._finished = True def _on_url_changed(self): # log the new url self._save_event( type="url", url=self.browser.web.url().toString() ) def _on_time_skip(self): # when the timer to allow skip have run out if "skip" in self.signals: self.navigation.show_skip() # condition def check(self) -> None: """ Check if the checking condition have been completed """ def check_callback(result: bool): if result: self._success() page = self.browser.web.page() # preprocess the condition condition = self.check_condition condition = condition.replace("#LANGUAGE_CODE#", translate.get_language()) # run the condition page.runJavaScript(condition, resultCallback=check_callback) # data collection def _save_event(self, **data) -> None: # if the mission is already finished, ignore if self._finished: return # save the data of the event and add the current time data["time"] = round(time.time() - self.start_time, 3) self._collected_events.append(data) def get_collected_data(self) -> dict: return { "event": self._collected_events, } # survey events def on_ready(self) -> None: # enable the maximized mode self.window().showMaximized() def on_finalize(self) -> None: # disable the maximized mode self.window().showNormal()