from __future__ import annotations
import time
from dataclasses import dataclass
from typing import Any
from aarecommon.config.logger import setup_logger
from PySide6.QtCore import (
Property,
QEasingCurve,
QObject,
QPropertyAnimation,
QRect,
Qt,
QTimer,
Signal,
)
from PySide6.QtGui import QPainter, QPen
from PySide6.QtWidgets import QLabel, QPushButton, QWidget
from aare.gui.constants import LOGGER_NAME
from aare.gui.styles import (
DEFAULT_TEXT,
NOTE_TEXT,
SHADOW,
TUTORIAL_BORDER,
TUTORIAL_HIGHLIGHT,
WHITE,
qcolor,
)
from aare.gui.tutorials.tutorial_models import (
StepFlow,
StepStatus,
TutorialContext,
TutorialScenario,
TutorialStepDefinition,
)
from aare.gui.tutorials.tutorial_runtime import (
CompletionEvaluator,
ResolvedTutorialTarget,
TutorialActionExecutor,
TutorialEvent,
TutorialEventBus,
TutorialTargetResolver,
TutorialTextResolver,
build_runtime_state_for_scenario,
ensure_step_runtime_state,
)
logger = setup_logger(LOGGER_NAME)
def _now_ms() -> int:
return int(time.time() * 1000)
@dataclass(slots=True)
class OverlayStepViewModel:
title: str
body: str
hint: str | None = None
rect: QRect | None = None
can_go_next: bool = True
can_go_back: bool = False
can_skip: bool = False
next_label: str = "Next"
waiting_for_target_click: bool = False
class TutorialOverlay(QWidget):
"""
Presentation-only overlay.
It knows how to:
- dim the app
- highlight a target rect
- show tutorial copy
- expose buttons/signals
It does NOT know anything about tutorial scenarios or completion rules.
"""
next_requested = Signal()
back_requested = Signal()
skip_requested = Signal()
cancelled = Signal()
highlight_clicked = Signal()
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self.setAttribute(Qt.WA_TransparentForMouseEvents, False)
self.setWindowFlags(Qt.FramelessWindowHint)
self.setAttribute(Qt.WA_TranslucentBackground, True)
self.setFocusPolicy(Qt.StrongFocus)
self.current_rect = QRect()
self.target_rect = QRect()
self.opacity = 1.0
self._waiting_for_click = False
self._highlight_padding_px = 6
self._dummy = 0.0
self.anim = QPropertyAnimation(self, b"dummy")
self.anim.valueChanged.connect(self.update)
self.callout = QLabel(self)
self.callout.setStyleSheet(f"""
background: {WHITE};
color: {DEFAULT_TEXT};
padding: 16px;
border: 2px solid {TUTORIAL_BORDER};
border-radius: 10px;
font-size: 16px;
""")
self.callout.setWordWrap(True)
self.callout.hide()
button_style = """
QPushButton {
font-size: 15px;
font-weight: 600;
padding: 10px 16px;
}
"""
self.back_button = QPushButton("Back", self)
self.back_button.setStyleSheet(button_style)
self.back_button.setMinimumHeight(44)
self.back_button.clicked.connect(self.back_requested.emit)
self.back_button.hide()
self.skip_button = QPushButton("Skip", self)
self.skip_button.setStyleSheet(button_style)
self.skip_button.setMinimumHeight(44)
self.skip_button.clicked.connect(self.skip_requested.emit)
self.skip_button.hide()
self.next_button = QPushButton("Next", self)
self.next_button.setStyleSheet(button_style)
self.next_button.setMinimumHeight(44)
self.next_button.clicked.connect(self.next_requested.emit)
self.next_button.hide()
self.end_button = QPushButton("End tutorial", self)
self.end_button.setStyleSheet(button_style)
self.end_button.setMinimumHeight(44)
self.end_button.clicked.connect(self.cancelled.emit)
self.end_button.hide()
self.hide()
def get_dummy(self) -> float:
return float(self._dummy)
def set_dummy(self, value: float) -> None:
self._dummy = float(value)
t = self._dummy
self.current_rect = self._interpolate_rect(self.current_rect, self.target_rect, t)
self.opacity = t
self.update()
dummy = Property(float, get_dummy, set_dummy)
def showEvent(self, event) -> None:
super().showEvent(event)
if self.parentWidget() is not None:
self.setGeometry(self.parentWidget().rect())
self._position_end_button()
def resizeEvent(self, event) -> None:
super().resizeEvent(event)
if self.parentWidget() is not None:
self.setGeometry(self.parentWidget().rect())
self._position_end_button()
self._reposition_action_buttons()
def keyPressEvent(self, event) -> None:
if event.key() == Qt.Key_Escape:
self.cancelled.emit()
event.accept()
return
super().keyPressEvent(event)
def mousePressEvent(self, event) -> None:
if (
self._waiting_for_click
and self.current_rect.isValid()
and self.current_rect.contains(event.pos())
):
self.highlight_clicked.emit()
event.accept()
return
super().mousePressEvent(event)
def paintEvent(self, event) -> None:
painter = QPainter(self)
painter.fillRect(self.rect(), qcolor(SHADOW, int(150 * self.opacity)))
if self.current_rect.isValid():
pen = QPen(qcolor(TUTORIAL_HIGHLIGHT), 4)
painter.setPen(pen)
painter.setBrush(Qt.NoBrush)
painter.drawRoundedRect(self.current_rect, 8, 8)
def animate_to(self, rect: QRect, duration: int = 300) -> None:
self.target_rect = rect.intersected(self.rect())
self.anim.stop()
self.anim.setStartValue(0.0)
self.anim.setEndValue(1.0)
self.anim.setDuration(duration)
self.anim.setEasingCurve(QEasingCurve.InOutQuad)
self.anim.start()
def set_step_view(self, view: OverlayStepViewModel) -> None:
rect = view.rect if view.rect is not None else QRect()
if rect.isValid():
pad = int(self._highlight_padding_px)
rect = QRect(rect)
rect.adjust(-pad, -pad, pad, pad)
rect = rect.intersected(self.rect())
self.current_rect = rect
self.target_rect = rect
text_parts = []
if view.title.strip():
text_parts.append(f"{view.title}")
if view.body.strip():
text_parts.append(view.body)
if view.hint:
text_parts.append(f"{view.hint}")
self.callout.setText("
".join(text_parts))
self.callout.setMaximumWidth(460)
self.callout.adjustSize()
self._waiting_for_click = bool(view.waiting_for_target_click)
self.back_button.setVisible(True)
self.back_button.setEnabled(view.can_go_back)
self.skip_button.setVisible(view.can_skip)
self.skip_button.setEnabled(view.can_skip)
self.next_button.setVisible(True)
self.next_button.setEnabled(view.can_go_next)
self.next_button.setText(view.next_label)
self.callout.show()
self._position_end_button()
self._position_callout(rect)
self._reposition_action_buttons()
self.update()
def clear_view(self) -> None:
self.current_rect = QRect()
self.target_rect = QRect()
self.callout.hide()
self.back_button.hide()
self.skip_button.hide()
self.next_button.hide()
self.end_button.hide()
self._waiting_for_click = False
self.update()
def _position_end_button(self) -> None:
self.end_button.adjustSize()
margin = 10
x = max(margin, self.width() - self.end_button.width() - margin)
y = margin
self.end_button.move(x, y)
self.end_button.setVisible(True)
def _position_callout(self, rect: QRect) -> None:
margin = 10
if rect.isValid():
bubble_x = rect.x()
bubble_y_above = rect.y() - self.callout.height() - 12
bubble_y_below = rect.y() + rect.height() + 12
bubble_y = bubble_y_above if bubble_y_above >= margin else bubble_y_below
else:
bubble_x = margin
bubble_y = margin + 48
bubble_x = min(
max(margin, bubble_x), max(margin, self.width() - self.callout.width() - margin)
)
bubble_y = min(
max(margin, bubble_y), max(margin, self.height() - self.callout.height() - margin - 60)
)
self.callout.move(bubble_x, bubble_y)
def _reposition_action_buttons(self) -> None:
buttons = [
btn for btn in (self.back_button, self.skip_button, self.next_button) if btn.isVisible()
]
if not buttons:
return
margin = 10
spacing = 8
y = self.callout.y() + self.callout.height() + 10
widths = []
total_width = 0
for btn in buttons:
btn.adjustSize()
widths.append(btn.width())
total_width += btn.width()
total_width += spacing * (len(buttons) - 1)
x = min(
max(margin, self.callout.x() + self.callout.width() - total_width),
max(margin, self.width() - total_width - margin),
)
y = min(
max(margin, y),
max(margin, self.height() - max(btn.height() for btn in buttons) - margin),
)
for btn in buttons:
btn.move(x, y)
x += btn.width() + spacing
@staticmethod
def _interpolate_rect(r1: QRect, r2: QRect, t: float) -> QRect:
x = r1.x() + (r2.x() - r1.x()) * t
y = r1.y() + (r2.y() - r1.y()) * t
w = r1.width() + (r2.width() - r1.width()) * t
h = r1.height() + (r2.height() - r1.height()) * t
return QRect(int(x), int(y), int(w), int(h))
class TutorialManager(QObject):
"""
Scenario-driven tutorial controller.
This replaces the old step model with:
- declarative scenarios
- runtime context/state
- target resolution
- text resolution
- action execution
- event-driven completion
"""
tutorial_started = Signal(str)
tutorial_stopped = Signal(str)
tutorial_completed = Signal(str)
step_changed = Signal(str, str) # scenario_id, step_id
def __init__(
self,
parent_window: QWidget,
target_resolver: TutorialTargetResolver,
text_resolver: TutorialTextResolver,
action_executor: TutorialActionExecutor,
event_bus: TutorialEventBus | None = None,
parent: QObject | None = None,
):
super().__init__(parent)
self.window = parent_window
self.overlay = TutorialOverlay(parent_window)
self.target_resolver = target_resolver
self.text_resolver = text_resolver
self.action_executor = action_executor
self.event_bus = event_bus or TutorialEventBus(self)
self.completion_evaluator = CompletionEvaluator(self.event_bus)
self.scenarios: dict[str, TutorialScenario] = {}
self.current_scenario: TutorialScenario | None = None
self.context: TutorialContext | None = None
self.runtime_state = None
self._current_resolved_target: ResolvedTutorialTarget | None = None
self._step_timeout_timer = QTimer(self)
self._step_timeout_timer.setSingleShot(True)
self._step_timeout_timer.timeout.connect(self._on_step_timeout)
self.overlay.next_requested.connect(self._on_next_requested)
self.overlay.back_requested.connect(self._on_back_requested)
self.overlay.skip_requested.connect(self._on_skip_requested)
self.overlay.cancelled.connect(self.stop)
self.overlay.highlight_clicked.connect(self._on_highlight_clicked)
self.event_bus.event_emitted.connect(self._on_tutorial_event)
def add_scenario(self, scenario: TutorialScenario) -> None:
self.scenarios[scenario.id] = scenario
def start(self, scenario_id: str, language: str = "en") -> None:
scenario = self.scenarios.get(scenario_id)
if scenario is None:
raise ValueError(f"Tutorial scenario not found: {scenario_id}")
if self.current_scenario is not None:
self.stop()
self.current_scenario = scenario
self.context = TutorialContext(scenario_id=scenario.id, language=language)
self.runtime_state = build_runtime_state_for_scenario(scenario)
self.runtime_state.started_at_ms = _now_ms()
self.runtime_state.active_step_index = -1
self._current_resolved_target = None
self.overlay.setGeometry(self.window.rect())
self.overlay.show()
self.overlay.raise_()
self.overlay.activateWindow()
self.overlay.setFocus()
self.tutorial_started.emit(scenario.id)
self._advance_to_index(0)
def stop(self) -> None:
scenario_id = self.current_scenario.id if self.current_scenario else None
self._step_timeout_timer.stop()
if self.current_scenario and self.context:
current_step = self.get_current_step()
if current_step is not None:
self._run_cleanup_actions(current_step)
if self.runtime_state is not None:
self.runtime_state.stopped_at_ms = _now_ms()
self.overlay.clear_view()
self.overlay.hide()
self.current_scenario = None
self.context = None
self.runtime_state = None
self._current_resolved_target = None
if scenario_id is not None:
self.tutorial_stopped.emit(scenario_id)
def emit_event(self, name: str, payload: dict[str, Any] | None = None) -> TutorialEvent:
event = self.event_bus.emit_event(name, payload or {})
if self.context is not None:
self.context.event_log.append(event)
return event
def get_current_step(self) -> TutorialStepDefinition | None:
if self.current_scenario is None or self.runtime_state is None:
return None
index = self.runtime_state.active_step_index
if not (0 <= index < len(self.current_scenario.steps)):
return None
return self.current_scenario.steps[index]
def _advance_to_index(self, new_index: int) -> None:
if self.current_scenario is None or self.context is None or self.runtime_state is None:
return
previous_step = self.get_current_step()
if previous_step is not None:
self._run_cleanup_actions(previous_step)
if new_index >= len(self.current_scenario.steps):
scenario_id = self.current_scenario.id
self.runtime_state.completed_at_ms = _now_ms()
self.overlay.clear_view()
self.overlay.hide()
self.current_scenario = None
self.context = None
self.runtime_state = None
self._current_resolved_target = None
self.tutorial_completed.emit(scenario_id)
self.tutorial_stopped.emit(scenario_id)
return
new_index = max(new_index, 0)
self.runtime_state.active_step_index = new_index
step = self.current_scenario.steps[new_index]
self.context.current_step_index = new_index
self.context.current_step_id = step.id
step_state = ensure_step_runtime_state(self.runtime_state, step.id)
step_state.status = StepStatus.ACTIVE
if step_state.started_at_ms is None:
step_state.started_at_ms = _now_ms()
self._run_setup_actions(step)
self._enter_current_step()
def _enter_current_step(self) -> None:
if self.current_scenario is None or self.context is None or self.runtime_state is None:
return
step = self.get_current_step()
if step is None:
return
self._step_timeout_timer.stop()
self._current_resolved_target = self._resolve_step_target(step)
title = self.text_resolver.resolve_text(step.title, self.context.language)
body = self.text_resolver.resolve_text(step.body, self.context.language)
hint = (
self.text_resolver.resolve_text(step.hint, self.context.language)
if step.hint is not None
else None
)
rect = self._resolved_target_rect(self._current_resolved_target)
if rect is not None and rect.isValid():
self.overlay.animate_to(rect)
is_first = self.runtime_state.active_step_index <= 0
is_last = self.runtime_state.active_step_index == (len(self.current_scenario.steps) - 1)
waiting_for_target_click = bool(
step.completion is not None and step.completion.kind.value == "target_clicked"
)
can_go_next = step.flow == StepFlow.NEXT_ONLY
next_label = "Finish" if is_last and can_go_next else "Next"
view = OverlayStepViewModel(
title=title,
body=body,
hint=hint,
rect=rect,
can_go_next=can_go_next,
can_go_back=self.current_scenario.allow_back and not is_first,
can_skip=self.current_scenario.allow_skip and step.skippable,
next_label=next_label,
waiting_for_target_click=waiting_for_target_click,
)
self.overlay.set_step_view(view)
if step.timeout_ms is not None and step.timeout_ms > 0:
self._step_timeout_timer.start(int(step.timeout_ms))
self.step_changed.emit(self.current_scenario.id, step.id)
self._try_auto_advance_current_step()
def _resolve_step_target(self, step: TutorialStepDefinition) -> ResolvedTutorialTarget | None:
if step.target is None or self.context is None:
return None
try:
return self.target_resolver.resolve_target(step.target, self.context)
except Exception:
logger.debug("Could not resolve the tutorial step target", exc_info=True)
return None
def _resolved_target_rect(self, resolved: ResolvedTutorialTarget | None) -> QRect | None:
if resolved is None:
return None
if resolved.widget is not None:
widget = resolved.widget
top_left_global = widget.mapToGlobal(widget.rect().topLeft())
bottom_right_global = widget.mapToGlobal(widget.rect().bottomRight())
top_left = self.overlay.mapFromGlobal(top_left_global)
bottom_right = self.overlay.mapFromGlobal(bottom_right_global)
return QRect(top_left, bottom_right).normalized().intersected(self.overlay.rect())
if resolved.rect is not None:
return QRect(resolved.rect).intersected(self.overlay.rect())
return None
def _run_setup_actions(self, step: TutorialStepDefinition) -> None:
if self.context is None:
return
for action in step.setup_actions:
self.action_executor.execute_action(action, self.context)
def _run_cleanup_actions(self, step: TutorialStepDefinition) -> None:
if self.context is None:
return
for action in step.cleanup_actions:
self.action_executor.execute_action(action, self.context)
def _mark_current_step_completed(self) -> None:
if self.current_scenario is None or self.context is None or self.runtime_state is None:
return
step = self.get_current_step()
if step is None:
return
step_state = ensure_step_runtime_state(self.runtime_state, step.id)
step_state.status = StepStatus.COMPLETED
step_state.completed_at_ms = _now_ms()
if step.id not in self.context.completed_step_ids:
self.context.completed_step_ids.append(step.id)
def _mark_current_step_skipped(self) -> None:
if self.current_scenario is None or self.context is None or self.runtime_state is None:
return
step = self.get_current_step()
if step is None:
return
step_state = ensure_step_runtime_state(self.runtime_state, step.id)
step_state.status = StepStatus.SKIPPED
step_state.skipped_at_ms = _now_ms()
def _mark_current_step_failed(self, message: str | None = None) -> None:
if self.current_scenario is None or self.runtime_state is None:
return
step = self.get_current_step()
if step is None:
return
step_state = ensure_step_runtime_state(self.runtime_state, step.id)
step_state.status = StepStatus.FAILED
step_state.failed_at_ms = _now_ms()
step_state.last_error = message
def _is_current_step_complete(self, *, target_clicked: bool = False) -> bool:
if self.context is None:
return False
step = self.get_current_step()
if step is None:
return False
if step.flow == StepFlow.NEXT_ONLY:
return False
return self.completion_evaluator.is_step_complete(
step, self.context, target_clicked=target_clicked
)
def _try_auto_advance_current_step(self, *, target_clicked: bool = False) -> None:
step = self.get_current_step()
if step is None:
return
if step.flow == StepFlow.NEXT_ONLY:
return
if self._is_current_step_complete(target_clicked=target_clicked):
self._mark_current_step_completed()
assert self.runtime_state is not None
self._advance_to_index(self.runtime_state.active_step_index + 1)
def _on_next_requested(self) -> None:
step = self.get_current_step()
if step is None or self.runtime_state is None:
return
if step.flow == StepFlow.NEXT_ONLY:
self._mark_current_step_completed()
self._advance_to_index(self.runtime_state.active_step_index + 1)
return
if self._is_current_step_complete():
self._mark_current_step_completed()
self._advance_to_index(self.runtime_state.active_step_index + 1)
def _on_back_requested(self) -> None:
if self.current_scenario is None or self.runtime_state is None:
return
if not self.current_scenario.allow_back:
return
target_index = self.runtime_state.active_step_index - 1
if target_index < 0:
return
self._advance_to_index(target_index)
def _on_skip_requested(self) -> None:
step = self.get_current_step()
if step is None or self.current_scenario is None or self.runtime_state is None:
return
if not self.current_scenario.allow_skip or not step.skippable:
return
self._mark_current_step_skipped()
self._advance_to_index(self.runtime_state.active_step_index + 1)
def _on_highlight_clicked(self) -> None:
self._try_auto_advance_current_step(target_clicked=True)
def _on_tutorial_event(self, event: TutorialEvent) -> None:
if self.current_scenario is None or self.context is None:
return
self._try_auto_advance_current_step()
def _on_step_timeout(self) -> None:
step = self.get_current_step()
if step is None:
return
self._mark_current_step_failed("Step timed out.")
if (
step.skippable
and self.current_scenario is not None
and self.current_scenario.allow_skip
) and self.runtime_state is not None:
self._advance_to_index(self.runtime_state.active_step_index + 1)
return
self.overlay.next_button.setEnabled(True)