"""PySide6 desktop UI for the DPS Coach application."""
from __future__ import annotations
import contextlib
import html
import os
import shutil
import sys
import traceback
import urllib.request
from pathlib import Path
from statistics import mean, median
from typing import Any, Dict, List, Optional
from PySide6.QtCore import Qt, QThread, Signal
from PySide6.QtGui import QPixmap
from PySide6.QtWidgets import (
QApplication,
QComboBox,
QDialog,
QFileDialog,
QFormLayout,
QGroupBox,
QHBoxLayout,
QLabel,
QLineEdit,
QMainWindow,
QMessageBox,
QProgressBar,
QPushButton,
QSizePolicy,
QSpinBox,
QStackedWidget,
QTabWidget,
QTableWidget,
QTableWidgetItem,
QTextBrowser,
QVBoxLayout,
QWidget,
)
from .constants import (
APP_NAME,
BANNER_IMAGE_PATH,
COACH_SUGGESTED_QUESTIONS,
DEFAULT_LIMIT_RUNS,
DEFAULT_MODEL_DOWNLOAD_SIZE_TEXT,
DEFAULT_MODEL_MIN_SIZE_MB,
DEFAULT_MODEL_PATH,
DEFAULT_MODEL_SHA256,
FALLBACK_MODEL_DOWNLOAD_SIZE_TEXT,
FALLBACK_MODEL_MIN_SIZE_MB,
FALLBACK_MODEL_PATH,
TL_CLASSES,
FALLBACK_MODEL_SHA256,
MAX_LIMIT_RUNS,
MIN_LIMIT_RUNS,
DEFAULT_MODEL_DOWNLOAD_URL,
resolve_default_log_dir,
)
MAX_BANNER_HEIGHT = 240
from .mcp_client import MCPAnalyzerClient
from .coach_local import SQLCoach
from .model_manager import LocalModelManager, validate_model_file, _LLAMA_IMPORT_ERROR
from .path_utils import suggest_combat_logs_path
class AnalysisWorker(QThread):
"""Background thread that executes the MCP analysis call."""
succeeded = Signal(dict)
failed = Signal(str)
status_changed = Signal(str)
def __init__(self, client: MCPAnalyzerClient, log_dir: str, limit_runs: int) -> None:
super().__init__()
self._client = client
self._log_dir = log_dir
self._limit_runs = limit_runs
def run(self) -> None: # noqa: D401 - inherited
try:
payload = self._client.analyze(
self._log_dir,
self._limit_runs,
status_callback=self._emit_status,
)
except Exception: # pragma: no cover - surfaced via signal
self.failed.emit(traceback.format_exc())
else:
self.succeeded.emit(payload)
def _emit_status(self, message: str) -> None:
self.status_changed.emit(message)
class CoachWorker(QThread):
"""Runs the local SQL-first coach without blocking the UI."""
succeeded = Signal(dict)
failed = Signal(str)
def __init__(
self,
client: MCPAnalyzerClient,
coach: SQLCoach,
payload: Dict[str, Any],
question: str,
) -> None:
super().__init__()
self._client = client
self._coach = coach
self._payload = payload
self._question = question
def run(self) -> None: # noqa: D401 - QThread contract
try:
schema = self._client.get_events_schema()
answer, tool_trace = self._coach.answer(
question=self._question,
payload=self._payload,
schema=schema,
query_callback=self._client.query,
analysis_callback=lambda: self._client.get_analysis_packet(
run_id="last",
last_n_runs=10,
top_k_skills=10,
bucket_seconds=5,
),
)
except Exception: # pragma: no cover - surfaced to UI
self.failed.emit(traceback.format_exc())
else:
self.succeeded.emit({"answer": answer, "tool_trace": tool_trace})
class ModelSelfTestWorker(QThread):
"""Ensures the bundled model responds deterministically."""
succeeded = Signal(str)
failed = Signal(str)
def __init__(self, model_manager: LocalModelManager) -> None:
super().__init__()
self._model_manager = model_manager
def run(self) -> None: # noqa: D401 - QThread contract
try:
self._model_manager.run_self_test()
self.succeeded.emit("OK")
except Exception: # pragma: no cover - emitted to UI
self.failed.emit(traceback.format_exc())
class ModelDownloadWorker(QThread):
"""Downloads the GGUF model with progress updates and SHA verification."""
progress = Signal(int)
status = Signal(str)
failed = Signal(str)
succeeded = Signal(str)
def __init__(self, url: str, target_path: Path, expected_sha256: str | None = None) -> None:
super().__init__()
self._url = url
self._target_path = Path(target_path)
self._expected_sha256 = expected_sha256
def run(self) -> None: # noqa: D401 - QThread contract
try:
self._download()
except Exception: # pragma: no cover - propagated to UI
self.failed.emit(traceback.format_exc())
else:
self.succeeded.emit(str(self._target_path))
def _download(self) -> None:
import hashlib
target = self._target_path
target.parent.mkdir(parents=True, exist_ok=True)
tmp_path = target.with_suffix(target.suffix + ".part")
with contextlib.suppress(FileNotFoundError):
tmp_path.unlink()
self.status.emit("Connecting to model host…")
downloaded = 0
chunk_size = 1024 * 1024
try:
with urllib.request.urlopen(self._url) as response:
total = int(response.headers.get("Content-Length") or 0)
with tmp_path.open("wb") as handle:
while True:
if self.isInterruptionRequested():
raise RuntimeError("Download cancelled")
chunk = response.read(chunk_size)
if not chunk:
break
handle.write(chunk)
downloaded += len(chunk)
if total:
percent = int(downloaded / total * 100)
self.progress.emit(min(max(percent, 0), 100))
else:
self.progress.emit(0)
# Verify SHA256 before finalizing
if self._expected_sha256:
self.status.emit("Verifying SHA256…")
sha = hashlib.sha256()
with tmp_path.open("rb") as handle:
for chunk in iter(lambda: handle.read(chunk_size), b""):
sha.update(chunk)
digest = sha.hexdigest()
if digest.lower() != self._expected_sha256.lower():
tmp_path.unlink()
raise ValueError(
f"SHA256 mismatch for {self._url}. Expected {self._expected_sha256.upper()}, "
f"got {digest.upper()}. Downloaded file deleted."
)
self.status.emit("SHA256 verified")
# Atomic rename on success
tmp_path.replace(target)
except Exception:
with contextlib.suppress(OSError):
tmp_path.unlink()
raise
self.progress.emit(100)
self.status.emit("Download complete")
class MainWindow(QMainWindow):
def __init__(self) -> None:
super().__init__()
self.setWindowTitle(APP_NAME)
self._client = MCPAnalyzerClient(inprocess=True)
self._worker: Optional[AnalysisWorker] = None
self._coach_worker: Optional[CoachWorker] = None
self._payload: Optional[Dict[str, Any]] = None
self._last_error: str = ""
self._llm_available = _LLAMA_IMPORT_ERROR is None
self._model_ready = False
use_fallback_model = os.getenv("DPSCOACH_USE_FALLBACK_MODEL") == "1"
coach_model_path = FALLBACK_MODEL_PATH if use_fallback_model else DEFAULT_MODEL_PATH
coach_model_min_size = FALLBACK_MODEL_MIN_SIZE_MB if use_fallback_model else DEFAULT_MODEL_MIN_SIZE_MB
coach_model_sha = FALLBACK_MODEL_SHA256 if use_fallback_model else DEFAULT_MODEL_SHA256
coach_download_url = None # Will use default URL for default model
coach_download_size = FALLBACK_MODEL_DOWNLOAD_SIZE_TEXT if use_fallback_model else DEFAULT_MODEL_DOWNLOAD_SIZE_TEXT
self._coach_model = LocalModelManager(
coach_model_path,
min_size_mb=coach_model_min_size,
expected_sha256=coach_model_sha,
)
self._active_model_url = DEFAULT_MODEL_DOWNLOAD_URL
self._active_model_size_text = coach_download_size
self._coach = SQLCoach(self._coach_model)
self._coach_ready = False
self._coach_inputs_locked = True
self._banner_pixmap: Optional[QPixmap] = None
self.banner_label: Optional[QLabel] = None
self.coach_status_label: Optional[QLabel] = None
self._self_test_worker: Optional[ModelSelfTestWorker] = None
self._download_worker: Optional[ModelDownloadWorker] = None
self.coach_stack: Optional[QStackedWidget] = None
self.coach_setup_index: int = 0
self.coach_chat_index: int = 0
self.download_model_btn: Optional[QPushButton] = None
self.choose_model_btn: Optional[QPushButton] = None
self.download_progress: Optional[QProgressBar] = None
self.download_status_label: Optional[QLabel] = None
self.quick_question_buttons: List[QPushButton] = []
self.local_coach_status: Optional[QLabel] = None
self.local_coach_transcript: Optional[QTextBrowser] = None
self.local_coach_trace: Optional[QTextBrowser] = None
self.local_coach_input: Optional[QLineEdit] = None
self.local_coach_send_btn: Optional[QPushButton] = None
self._build_ui()
self._initialize_coach_state()
# ------------------------------------------------------------------ UI setup
def _build_ui(self) -> None:
central = QWidget(self)
layout = QVBoxLayout(central)
self.banner_label = QLabel()
self.banner_label.setAlignment(Qt.AlignHCenter | Qt.AlignVCenter)
self.banner_label.setObjectName("bannerLabel")
self.banner_label.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed)
self.banner_label.setStyleSheet(
"#bannerLabel {"
"background-color: #0c0c16;"
"border-radius: 12px;"
"padding: 8px;"
"}"
)
layout.addWidget(self.banner_label)
self._load_banner_pixmap()
self._render_banner()
# Log directory picker row
picker_row = QHBoxLayout()
picker_label = QLabel("Combat Logs Folder:")
picker_row.addWidget(picker_label)
self.log_dir_edit = QLineEdit(str(resolve_default_log_dir()))
picker_row.addWidget(self.log_dir_edit, stretch=1)
browse_btn = QPushButton("Browse…")
browse_btn.clicked.connect(self._browse_for_logs)
picker_row.addWidget(browse_btn)
layout.addLayout(picker_row)
# Limit input row
limit_row = QHBoxLayout()
limit_label = QLabel("Limit Runs:")
limit_row.addWidget(limit_label)
self.limit_spin = QSpinBox()
self.limit_spin.setRange(MIN_LIMIT_RUNS, MAX_LIMIT_RUNS)
self.limit_spin.setValue(DEFAULT_LIMIT_RUNS)
limit_row.addWidget(self.limit_spin)
limit_row.addStretch()
layout.addLayout(limit_row)
# Buttons row
buttons_row = QHBoxLayout()
self.analyze_btn = QPushButton("Analyze")
self.analyze_btn.clicked.connect(self._handle_analyze)
buttons_row.addWidget(self.analyze_btn)
self.copy_error_btn = QPushButton("Copy Error")
self.copy_error_btn.setEnabled(False)
self.copy_error_btn.clicked.connect(self._copy_error)
buttons_row.addWidget(self.copy_error_btn)
buttons_row.addStretch()
layout.addLayout(buttons_row)
# Status label
self.status_label = QLabel("Idle")
layout.addWidget(self.status_label)
# Tabs
self.tabs = QTabWidget()
layout.addWidget(self.tabs, stretch=1)
self.summary_tab = QWidget()
self.runs_tab = QWidget()
self.skills_tab = QWidget()
self.coach_tab = QWidget()
self.tabs.addTab(self.summary_tab, "Summary")
self.tabs.addTab(self.runs_tab, "Runs")
self.tabs.addTab(self.skills_tab, "Top Skills")
self.tabs.addTab(self.coach_tab, "Coach")
self._build_summary_tab()
self._build_runs_tab()
self._build_skills_tab()
self._build_coach_tab()
self.setCentralWidget(central)
def _load_banner_pixmap(self) -> None:
if BANNER_IMAGE_PATH.exists():
pixmap = QPixmap(str(BANNER_IMAGE_PATH))
if pixmap and not pixmap.isNull():
self._banner_pixmap = pixmap
else:
self._banner_pixmap = None
else:
self._banner_pixmap = None
def _show_about_dialog(self) -> None:
"""Show the About dialog with project information and portfolio links."""
dialog = AboutDialog(self)
dialog.exec()
def _render_banner(self) -> None:
if not self.banner_label:
return
if self._banner_pixmap:
target_width = max(400, self.width() - 80)
scaled = self._banner_pixmap.scaled(
target_width,
MAX_BANNER_HEIGHT,
Qt.KeepAspectRatio,
Qt.SmoothTransformation,
)
self.banner_label.setPixmap(scaled)
target_height = min(MAX_BANNER_HEIGHT, scaled.height())
self.banner_label.setMinimumHeight(target_height)
self.banner_label.setMaximumHeight(target_height + 12)
self.banner_label.setVisible(True)
else:
self.banner_label.clear()
self.banner_label.setMinimumHeight(0)
self.banner_label.setMaximumHeight(MAX_BANNER_HEIGHT)
self.banner_label.setVisible(False)
def _build_summary_tab(self) -> None:
layout = QFormLayout()
self.summary_labels = {
"num_runs": QLabel("–"),
"total_damage": QLabel("–"),
"mean_dps": QLabel("–"),
"median_dps": QLabel("–"),
"mean_dpm": QLabel("–"),
"median_dpm": QLabel("–"),
"duration": QLabel("–"),
}
layout.addRow("Runs", self.summary_labels["num_runs"])
layout.addRow("Total Damage", self.summary_labels["total_damage"])
layout.addRow("Mean DPS", self.summary_labels["mean_dps"])
layout.addRow("Median DPS", self.summary_labels["median_dps"])
layout.addRow("Mean DPM", self.summary_labels["mean_dpm"])
layout.addRow("Median DPM", self.summary_labels["median_dpm"])
layout.addRow("Total Duration (s)", self.summary_labels["duration"])
self.summary_tab.setLayout(layout)
def _build_runs_tab(self) -> None:
self.runs_table = QTableWidget(0, 7)
self.runs_table.setHorizontalHeaderLabels(
[
"Run",
"Duration (s)",
"DPS",
"DPM",
"Hits",
"Crit %",
"Heavy %",
]
)
self.runs_table.horizontalHeader().setStretchLastSection(True)
self.runs_table.setEditTriggers(QTableWidget.NoEditTriggers)
self.runs_table.setSelectionBehavior(QTableWidget.SelectRows)
self.runs_table.setSelectionMode(QTableWidget.SingleSelection)
layout = QVBoxLayout()
layout.addWidget(self.runs_table)
self.runs_tab.setLayout(layout)
def _build_skills_tab(self) -> None:
self.skills_table = QTableWidget(0, 4)
self.skills_table.setHorizontalHeaderLabels([
"Skill",
"Total Damage",
"Damage Share %",
"Avg Crit %",
])
self.skills_table.horizontalHeader().setStretchLastSection(True)
self.skills_table.setEditTriggers(QTableWidget.NoEditTriggers)
self.skills_table.setSelectionBehavior(QTableWidget.SelectRows)
self.skills_table.setSelectionMode(QTableWidget.SingleSelection)
layout = QVBoxLayout()
layout.addWidget(self.skills_table)
self.skills_tab.setLayout(layout)
def _build_coach_tab(self) -> None:
self.quick_question_buttons = []
layout = QVBoxLayout()
# Header row with intro and About button
header_row = QHBoxLayout()
intro = QLabel(
"AI Coach answers any Throne & Liberty combat question by planning up to one SQL query over your logs."
)
intro.setWordWrap(True)
header_row.addWidget(intro, stretch=1)
about_btn = QPushButton("About DPSCoach")
about_btn.setMaximumWidth(150)
about_btn.clicked.connect(self._show_about_dialog)
header_row.addWidget(about_btn)
layout.addLayout(header_row)
self.coach_stack = QStackedWidget()
layout.addWidget(self.coach_stack, stretch=1)
setup_widget = QWidget()
setup_layout = QVBoxLayout(setup_widget)
setup_label = QLabel(
"A local GGUF model is required. Download it once or point the app at an existing file to unlock the Coach."
)
setup_label.setWordWrap(True)
setup_layout.addWidget(setup_label)
buttons_row = QHBoxLayout()
self.download_model_btn = QPushButton(f"Download Model ({self._active_model_size_text})")
self.download_model_btn.setMinimumHeight(48)
self.download_model_btn.clicked.connect(self._start_model_download)
buttons_row.addWidget(self.download_model_btn)
self.choose_model_btn = QPushButton("Choose Local Model File…")
self.choose_model_btn.setMinimumHeight(48)
self.choose_model_btn.clicked.connect(self._handle_choose_model)
buttons_row.addWidget(self.choose_model_btn)
buttons_row.addStretch()
setup_layout.addLayout(buttons_row)
self.download_progress = QProgressBar()
self.download_progress.setRange(0, 100)
self.download_progress.setVisible(False)
setup_layout.addWidget(self.download_progress)
self.download_status_label = QLabel("Model not detected.")
self.download_status_label.setWordWrap(True)
setup_layout.addWidget(self.download_status_label)
setup_layout.addStretch()
chat_widget = QWidget()
chat_layout = QVBoxLayout(chat_widget)
self.local_coach_status = QLabel("Model check pending…")
self.local_coach_status.setWordWrap(True)
chat_layout.addWidget(self.local_coach_status)
# Class filter for deeper analysis
filter_row = QHBoxLayout()
filter_label = QLabel("Class Filter:")
filter_row.addWidget(filter_label)
self.class_filter_combo = QComboBox()
self.class_filter_combo.addItems(TL_CLASSES)
self.class_filter_combo.setCurrentText("All Classes")
self.class_filter_combo.setToolTip("Filter analysis by character class (future enhancement)")
self.class_filter_combo.setEnabled(False) # Disabled until backend filtering implemented
filter_row.addWidget(self.class_filter_combo)
filter_row.addStretch()
chat_layout.addLayout(filter_row)
quick_label = QLabel("Quick Questions (shortcuts)")
chat_layout.addWidget(quick_label)
# Split Quick Questions into two rows for better layout
quick_row1 = QHBoxLayout()
for question in COACH_SUGGESTED_QUESTIONS[:5]:
button = QPushButton(question)
button.setEnabled(False)
button.clicked.connect(lambda _=False, text=question: self._ask_coach_question(text))
self.quick_question_buttons.append(button)
quick_row1.addWidget(button)
quick_row1.addStretch()
chat_layout.addLayout(quick_row1)
quick_row2 = QHBoxLayout()
for question in COACH_SUGGESTED_QUESTIONS[5:]:
button = QPushButton(question)
button.setEnabled(False)
button.clicked.connect(lambda _=False, text=question: self._ask_coach_question(text))
self.quick_question_buttons.append(button)
quick_row2.addWidget(button)
quick_row2.addStretch()
chat_layout.addLayout(quick_row2)
self.local_coach_transcript = QTextBrowser()
self.local_coach_transcript.setReadOnly(True)
self.local_coach_transcript.setPlaceholderText("Coach answers will appear here once the model is ready.")
chat_layout.addWidget(self.local_coach_transcript, stretch=2)
trace_group = QGroupBox("SQL Trace")
trace_layout = QVBoxLayout()
self.local_coach_trace = QTextBrowser()
self.local_coach_trace.setReadOnly(True)
trace_layout.addWidget(self.local_coach_trace)
trace_group.setLayout(trace_layout)
chat_layout.addWidget(trace_group, stretch=1)
input_row = QHBoxLayout()
self.local_coach_input = QLineEdit()
self.local_coach_input.setPlaceholderText("Ask the Coach anything about your logs…")
input_row.addWidget(self.local_coach_input, stretch=1)
self.local_coach_send_btn = QPushButton("Ask Coach")
self.local_coach_send_btn.clicked.connect(self._handle_local_coach_question)
input_row.addWidget(self.local_coach_send_btn)
chat_layout.addLayout(input_row)
footer = QLabel(f"Model path: {self._coach_model.model_path}")
footer.setWordWrap(True)
footer.setTextInteractionFlags(Qt.TextSelectableByMouse)
chat_layout.addWidget(footer)
self.coach_stack.addWidget(setup_widget)
self.coach_stack.addWidget(chat_widget)
self.coach_setup_index = 0
self.coach_chat_index = 1
self.coach_stack.setCurrentIndex(self.coach_setup_index)
self.local_coach_input.setEnabled(False)
self.local_coach_send_btn.setEnabled(False)
self.coach_tab.setLayout(layout)
def _initialize_coach_state(self) -> None:
if not self._llm_available:
message = "llama-cpp-python is missing. Reinstall the app to restore the Coach."
self._show_model_setup(message)
self._set_coach_status(message, ready=False)
return
model_path = self._coach_model.model_path
if not model_path.exists():
self._show_model_setup("Model not found. Download or select a GGUF file to continue.")
self._coach_model.set_model_path(model_path)
self._set_coach_status("Model required before the Coach can run.", ready=False)
return
try:
validate_model_file(
model_path,
min_size_mb=self._coach_model.min_size_mb,
expected_sha256=self._coach_model.expected_sha256,
)
except Exception as exc:
detail = f"Model validation failed: {exc}"
self._last_error = str(exc)
self.copy_error_btn.setEnabled(True)
self._show_model_setup(detail)
self._set_coach_status(detail, ready=False)
return
self._coach_model.set_model_path(model_path)
self._show_coach_chat()
self._set_coach_status("Running model self-test…", ready=False)
self._start_coach_self_test()
def _set_coach_status(self, message: str, *, ready: bool) -> None:
self._coach_ready = ready
if self.local_coach_status:
self.local_coach_status.setText(message)
self._apply_coach_input_state()
def _show_model_setup(self, message: str) -> None:
if self.coach_stack:
self.coach_stack.setCurrentIndex(self.coach_setup_index)
if self.download_status_label:
self.download_status_label.setText(message)
if self.local_coach_status:
self.local_coach_status.setText(message)
self._lock_coach_inputs(True)
def _show_coach_chat(self) -> None:
if self.coach_stack:
self.coach_stack.setCurrentIndex(self.coach_chat_index)
def _lock_coach_inputs(self, locked: bool) -> None:
self._coach_inputs_locked = locked
self._apply_coach_input_state()
def _apply_coach_input_state(self) -> None:
enabled = self._coach_ready and not self._coach_inputs_locked
if self.local_coach_input:
self.local_coach_input.setEnabled(enabled)
if self.local_coach_send_btn:
self.local_coach_send_btn.setEnabled(enabled)
for button in self.quick_question_buttons:
button.setEnabled(enabled)
def _start_model_download(self) -> None:
if self._download_worker and self._download_worker.isRunning():
return
target = self._coach_model.model_path
target.parent.mkdir(parents=True, exist_ok=True)
if self.download_progress:
self.download_progress.setVisible(True)
self.download_progress.setValue(0)
self._toggle_download_buttons(False)
self._download_worker = ModelDownloadWorker(
self._active_model_url,
target,
expected_sha256=self._coach_model.expected_sha256,
)
self._download_worker.progress.connect(self._handle_download_progress)
self._download_worker.status.connect(self._handle_download_status)
self._download_worker.succeeded.connect(self._handle_download_success)
self._download_worker.failed.connect(self._handle_download_failure)
self._download_worker.finished.connect(self._cleanup_download_worker)
self._download_worker.start()
if self.download_status_label:
self.download_status_label.setText(f"Downloading model to {target}…")
def _handle_download_progress(self, percent: int) -> None:
if self.download_progress:
self.download_progress.setVisible(True)
self.download_progress.setValue(max(0, min(percent, 100)))
def _handle_download_status(self, message: str) -> None:
if self.download_status_label:
self.download_status_label.setText(message)
def _handle_download_success(self, _path: str) -> None:
if self.download_status_label:
self.download_status_label.setText("Download complete. Validating model…")
if self.download_progress:
self.download_progress.setVisible(False)
self._initialize_coach_state()
def _handle_download_failure(self, error_text: str) -> None:
self._last_error = error_text
self.copy_error_btn.setEnabled(True)
if self.download_progress:
self.download_progress.setVisible(False)
self._toggle_download_buttons(True)
QMessageBox.critical(self, APP_NAME, f"Model download failed:\n{self._summarize_error(error_text)}")
def _cleanup_download_worker(self) -> None:
if self._download_worker:
self._download_worker.deleteLater()
self._download_worker = None
self._toggle_download_buttons(True)
def _toggle_download_buttons(self, enabled: bool) -> None:
for button in (self.download_model_btn, self.choose_model_btn):
if button:
button.setEnabled(enabled)
def _handle_choose_model(self) -> None:
start_dir = str(self._coach_model.model_path.parent)
file_path, _ = QFileDialog.getOpenFileName(
self,
"Select GGUF Model",
start_dir,
"GGUF files (*.gguf);;All Files (*)",
)
if not file_path:
return
self._install_model_from_path(Path(file_path))
def _install_model_from_path(self, source: Path) -> None:
try:
validate_model_file(source)
except Exception as exc:
QMessageBox.critical(self, APP_NAME, f"Invalid model file:\n{exc}")
return
destination = self._coach_model.model_path
destination.parent.mkdir(parents=True, exist_ok=True)
try:
resolved_source = source.resolve()
resolved_dest = destination.resolve()
if resolved_source != resolved_dest:
shutil.copyfile(resolved_source, resolved_dest)
except Exception as exc:
QMessageBox.critical(self, APP_NAME, f"Failed to copy model:\n{exc}")
return
self.status_label.setText(f"Model ready at {destination}")
self._initialize_coach_state()
def _start_coach_self_test(self) -> None:
if self._self_test_worker and self._self_test_worker.isRunning():
return
self._self_test_worker = ModelSelfTestWorker(self._coach_model)
self._self_test_worker.succeeded.connect(self._handle_self_test_success)
self._self_test_worker.failed.connect(self._handle_self_test_failure)
self._self_test_worker.finished.connect(self._reset_self_test_worker_state)
self._self_test_worker.start()
@staticmethod
def _summarize_error(error_text: str) -> str:
if not error_text:
return "Unknown error."
lines = [line.strip() for line in error_text.strip().splitlines() if line.strip()]
return lines[-1] if lines else "Unknown error."
# ------------------------------------------------------------------ UI actions
def _browse_for_logs(self) -> None:
directory = QFileDialog.getExistingDirectory(self, "Select TL Combat Log Folder", self.log_dir_edit.text())
if directory:
self.log_dir_edit.setText(directory)
def _copy_error(self) -> None:
if not self._last_error:
return
QApplication.clipboard().setText(self._last_error)
def _handle_self_test_success(self, _output: str) -> None:
self._set_coach_status("Local coach ready", ready=True)
self._lock_coach_inputs(False)
if self.local_coach_transcript:
self.local_coach_transcript.append("<b>Self-Test:</b> PASS (OK)")
def _handle_self_test_failure(self, error_text: str) -> None:
self._last_error = error_text
self.copy_error_btn.setEnabled(True)
detail = error_text.strip() or "Unknown error."
self._set_coach_status(f"Self-test failed: {detail}", ready=False)
self._lock_coach_inputs(True)
if self.local_coach_transcript:
safe_detail = html.escape(detail)
guidance = (
f"<b>Self-Test:</b> FAIL – {safe_detail}. Download the model again or choose a different GGUF file."
)
self.local_coach_transcript.append(guidance)
def _reset_self_test_worker_state(self) -> None:
if self._self_test_worker:
self._self_test_worker.deleteLater()
self._self_test_worker = None
def _prepare_log_dir(self) -> Optional[Path]:
raw_value = self.log_dir_edit.text().strip() if self.log_dir_edit else ""
if not raw_value:
QMessageBox.warning(self, APP_NAME, "Please select a combat log folder or file.")
return None
trimmed = raw_value.rstrip("/\\") or raw_value
candidate = Path(trimmed).expanduser()
if candidate.exists():
normalized = str(candidate)
if normalized != self.log_dir_edit.text():
self.log_dir_edit.setText(normalized)
return candidate
suggestion = suggest_combat_logs_path(candidate)
if suggestion and suggestion.exists():
corrected_text = str(suggestion)
self.log_dir_edit.setText(corrected_text)
self.status_label.setText(f"Corrected path to {corrected_text}")
return suggestion
QMessageBox.critical(self, APP_NAME, f"Folder not found:\n{candidate}")
self.status_label.setText("Folder not found")
return None
def _handle_analyze(self) -> None:
# Prevent starting a second analysis while one is running
if self._worker and self._worker.isRunning():
return
log_dir_path = self._prepare_log_dir()
if log_dir_path is None:
return
log_dir = str(log_dir_path)
limit_runs = int(self.limit_spin.value())
self.status_label.setText("Starting MCP server…")
self.analyze_btn.setEnabled(False)
self.copy_error_btn.setEnabled(False)
self.log_dir_edit.setEnabled(False)
self.limit_spin.setEnabled(False)
self._last_error = ""
self._worker = AnalysisWorker(self._client, log_dir, limit_runs)
self._worker.succeeded.connect(self._handle_analysis_success)
self._worker.failed.connect(self._handle_analysis_failure)
self._worker.status_changed.connect(self._handle_worker_status)
self._worker.finished.connect(self._reset_worker_state)
self._worker.start()
def _handle_local_coach_question(self) -> None:
if not self.local_coach_input:
return
question = self.local_coach_input.text().strip()
if not question:
return
self.local_coach_input.clear()
self._ask_coach_question(question)
def _ask_coach_question(self, question: str) -> None:
if self._coach_worker and self._coach_worker.isRunning():
return
if not self._payload:
QMessageBox.information(self, APP_NAME, "Run Analyze before using the Coach.")
return
if not self._coach_ready:
QMessageBox.warning(self, APP_NAME, "Download or validate the model to enable the Coach.")
return
if self.local_coach_trace:
self.local_coach_trace.clear()
self._append_local_coach_message("You", question)
self.status_label.setText("Coach is planning…")
self._lock_coach_inputs(True)
self._coach_worker = CoachWorker(self._client, self._coach, self._payload, question)
self._coach_worker.succeeded.connect(self._handle_coach_success)
self._coach_worker.failed.connect(self._handle_coach_failure)
self._coach_worker.finished.connect(self._reset_coach_worker_state)
self._coach_worker.start()
def _handle_coach_success(self, result: Dict[str, Any]) -> None:
answer = result.get("answer", "")
tool_trace = result.get("tool_trace") or []
for call in tool_trace:
if call.get("tool") == "get_analysis_packet":
self._append_packet_trace_entry(call)
continue
sql = call.get("sql", call.get("request", {}).get("sql", ""))
rows = int(call.get("rows_returned", 0))
fallback = bool(call.get("fallback"))
self._append_tool_trace_entry(sql, rows, fallback)
if answer:
self._append_local_coach_message("Coach", answer)
self.status_label.setText("Coach ready")
def _handle_coach_failure(self, error_text: str) -> None:
self.status_label.setText("Coach failed")
self._last_error = error_text
self.copy_error_btn.setEnabled(True)
detail = error_text.strip() or "Unknown error."
self._append_local_coach_message("Coach", f"Error: {detail}")
QMessageBox.critical(self, APP_NAME, "Coach chat failed. See Copy Error for diagnostics.")
def _reset_coach_worker_state(self) -> None:
if self._coach_worker:
self._coach_worker.deleteLater()
self._coach_worker = None
self._lock_coach_inputs(False)
@staticmethod
def _format_number(value: Any) -> str:
if value is None:
return "–"
try:
num = float(value)
except Exception:
return str(value)
if abs(num - int(num)) < 1e-6:
return f"{int(round(num)):,}"
return f"{num:.2f}"
@staticmethod
def _format_percent(value: Any) -> str:
if value is None:
return "0%"
try:
num = float(value)
except Exception:
return str(value)
return f"{num:.1f}%"
def _append_local_coach_message(self, sender: str, text: str) -> None:
if not self.local_coach_transcript:
return
safe_sender = html.escape(sender)
safe_text = html.escape(text).replace("\n", "<br>")
self.local_coach_transcript.append(f"<b>{safe_sender}:</b> {safe_text}")
def _append_tool_trace_entry(self, sql: str, rows: int, fallback: bool) -> None:
if not self.local_coach_trace:
return
flag = " (fallback)" if fallback else ""
safe_sql = html.escape(sql)
self.local_coach_trace.append(f"<code>{safe_sql}</code><br>Rows: {rows}{flag}")
def _append_packet_trace_entry(self, trace: Dict[str, Any]) -> None:
if not self.local_coach_trace:
return
run_id = trace.get("run_id", "last")
top_k = trace.get("top_k", 10)
bucket = trace.get("bucket_seconds", 5)
counts = trace.get("counts") or {}
skills = counts.get("skills", 0)
runs = counts.get("runs", 0)
timeline = counts.get("timeline", 0)
line = (
f"route=DATA tool=get_analysis_packet run_id={run_id} top_k={top_k} bucket={bucket} "
f"counts: skills={skills} runs={runs} timeline={timeline}"
)
safe_line = html.escape(line)
self.local_coach_trace.append(f"<code>{safe_line}</code>")
def _handle_analysis_success(self, payload: Dict[str, Any]) -> None:
self.status_label.setText("Analysis complete")
self._payload = payload
self._update_summary()
self._update_runs_table()
self._update_skills_table()
if self.coach_status_label:
self.coach_status_label.setText("Context ready. Ask the Coach or tap a Quick Question.")
def _handle_analysis_failure(self, error_text: str) -> None:
self.status_label.setText("Analysis failed")
self._last_error = error_text
self.copy_error_btn.setEnabled(True)
QMessageBox.critical(self, APP_NAME, "Failed to analyze logs. See details via Copy Error.")
if self.coach_status_label:
self.coach_status_label.setText("Analysis failed. Fix the issue and rerun Analyze.")
def _handle_worker_status(self, message: str) -> None:
self.status_label.setText(message)
def _reset_worker_state(self) -> None:
self.analyze_btn.setEnabled(True)
self.log_dir_edit.setEnabled(True)
self.limit_spin.setEnabled(True)
if self._worker:
self._worker.deleteLater()
self._worker = None
# ------------------------------------------------------------------ Data rendering
def _update_summary(self) -> None:
summary = (self._payload or {}).get("summary") or {}
runs = (self._payload or {}).get("runs") or []
num_runs = int(summary.get("total_runs", len(runs)))
total_damage = int(summary.get("total_damage", 0))
total_duration = float(summary.get("combined_duration_seconds", 0.0))
dps_values = [float(run.get("dps", 0.0)) for run in runs if isinstance(run, dict)]
computed_mean_dps = mean(dps_values) if dps_values else 0.0
computed_median_dps = median(dps_values) if dps_values else 0.0
mean_dps = float(summary.get("mean_dps", computed_mean_dps))
median_dps = float(summary.get("median_dps", computed_median_dps))
mean_dpm = float(summary.get("mean_dpm", mean_dps * 60.0))
median_dpm = float(summary.get("median_dpm", median_dps * 60.0))
self.summary_labels["num_runs"].setText(str(num_runs))
self.summary_labels["total_damage"].setText(f"{total_damage:,}")
self.summary_labels["mean_dps"].setText(f"{mean_dps:,.1f}")
self.summary_labels["median_dps"].setText(f"{median_dps:,.1f}")
self.summary_labels["mean_dpm"].setText(f"{mean_dpm:,.1f}")
self.summary_labels["median_dpm"].setText(f"{median_dpm:,.1f}")
self.summary_labels["duration"].setText(f"{total_duration:,.2f}")
def _update_runs_table(self) -> None:
runs: List[Dict[str, Any]] = (self._payload or {}).get("runs") or []
self.runs_table.setRowCount(len(runs))
for row_idx, run in enumerate(runs):
values = [
run.get("run_id", "?"),
f"{float(run.get('duration_seconds', 0.0)):.2f}",
f"{float(run.get('dps', 0.0)):.1f}",
f"{float(run.get('dpm', 0.0)):.1f}",
f"{int(run.get('total_hits', 0)):,}",
f"{float(run.get('crit_rate_pct', 0.0)):.1f}",
f"{float(run.get('heavy_rate_pct', 0.0)):.1f}",
]
for col_idx, value in enumerate(values):
item = QTableWidgetItem(value)
if col_idx > 0:
item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter)
self.runs_table.setItem(row_idx, col_idx, item)
self.runs_table.resizeColumnsToContents()
def _update_skills_table(self) -> None:
summary = (self._payload or {}).get("summary") or {}
total_damage = float(summary.get("total_damage", 0) or 0)
skills: List[Dict[str, Any]] = summary.get("top_skills") or summary.get("top_skills_by_damage") or []
self.skills_table.setRowCount(len(skills))
for row_idx, skill in enumerate(skills):
damage = float(skill.get("total_damage", 0))
share = (damage / total_damage * 100.0) if total_damage else 0.0
values = [
skill.get("skill", "?"),
f"{int(damage):,}",
f"{share:.2f}",
f"{float(skill.get('crit_rate_pct', 0.0)):.1f}",
]
for col_idx, value in enumerate(values):
item = QTableWidgetItem(value)
if col_idx > 0:
item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter)
self.skills_table.setItem(row_idx, col_idx, item)
self.skills_table.resizeColumnsToContents()
# ------------------------------------------------------------------ Qt overrides
def closeEvent(self, event) -> None: # noqa: D401
def stop_thread(thread, name):
if thread and thread.isRunning():
# Disable buttons while stopping
self.analyze_btn.setEnabled(False)
self.log_dir_edit.setEnabled(False)
self.limit_spin.setEnabled(False)
try:
thread.requestInterruption()
except Exception:
pass
thread.quit()
finished = thread.wait(2000)
if not finished and thread.isRunning():
print(f"WARNING: {name} did not exit after 2s, terminating.")
thread.terminate()
thread.wait(1000)
stop_thread(self._worker, "AnalysisWorker")
stop_thread(self._coach_worker, "CoachWorker")
stop_thread(self._self_test_worker, "ModelSelfTestWorker")
stop_thread(self._download_worker, "ModelDownloadWorker")
return super().closeEvent(event)
def resizeEvent(self, event) -> None: # noqa: D401
super().resizeEvent(event)
self._render_banner()
class AboutDialog(QDialog):
"""About dialog displaying project information and portfolio links."""
def __init__(self, parent: QWidget | None = None) -> None:
super().__init__(parent)
self.setWindowTitle("About DPSCoach")
self.setMinimumWidth(600)
self.setMinimumHeight(500)
layout = QVBoxLayout()
# Title
title = QLabel("<h1>DPSCoach</h1>")
title.setAlignment(Qt.AlignCenter)
layout.addWidget(title)
# Subtitle
subtitle = QLabel(
"<p><i>AI-powered, privacy-safe combat log analyzer for Throne & Liberty</i></p>"
)
subtitle.setAlignment(Qt.AlignCenter)
subtitle.setWordWrap(True)
layout.addWidget(subtitle)
# Description
description = QLabel(
"<p>DPSCoach parses Throne & Liberty combat logs into deterministic DPS/DPM summaries. "
"A local AI coach (Qwen2.5 via llama.cpp) answers natural-language questions using "
"intent routing, SQL-first planning, and DuckDB analytics — no cloud calls, no hooks, "
"no automation.</p>"
)
description.setWordWrap(True)
layout.addWidget(description)
# Key Capabilities
capabilities = QLabel(
"<h3>What It Does</h3>"
"<ul>"
"<li><b>Intent-routed coach:</b> 90% of prompts (RUNS, SKILLS, CRIT_BUCKET_TREND, etc.) "
"resolve without an LLM, guaranteeing stable answers</li>"
"<li><b>DuckDB event store:</b> Streams logs into columnar tables for fast, read-only SQL diagnostics</li>"
"<li><b>MCP parity:</b> CLI, desktop app, and FastMCP tools all consume the same payload contract</li>"
"<li><b>Strict model validation:</b> GGUF hashes, size checks, and self-tests before the coach goes live</li>"
"<li><b>Fair-play defaults:</b> SELECT-only SQL, no packet sniffing, no in-game automation</li>"
"</ul>"
)
capabilities.setWordWrap(True)
layout.addWidget(capabilities)
# Tech Stack
tech = QLabel(
"<h3>Tech Stack</h3>"
"<p><b>UI:</b> PySide6 (Qt for Python) • "
"<b>Database:</b> DuckDB • "
"<b>Protocol:</b> FastMCP • "
"<b>Model:</b> llama-cpp-python (GGUF) • "
"<b>Parser:</b> Streaming Python iterators</p>"
)
tech.setWordWrap(True)
layout.addWidget(tech)
# Engineering Signals
engineering = QLabel(
"<h3>Engineering Signals</h3>"
"<p><b>Contract-driven design:</b> MCP boundary, stable schemas, CLI/UI parity<br>"
"<b>Defensive programming:</b> Input clamps, path validation, safe fallbacks<br>"
"<b>Observability:</b> 70+ tests, trace logs, reproducible builds<br>"
"<b>Performance:</b> Streaming parsers, background workers, capped result sets<br>"
"<b>Security:</b> No shell=True, subprocess isolation, user-controlled models</p>"
)
engineering.setWordWrap(True)
layout.addWidget(engineering)
# Links section
links_label = QLabel(
"<h3>Contribute & Contact</h3>"
"<p><b>Help wanted:</b> I need per-class reference text for Throne & Liberty — benefits, skills, combos, DoT limits. "
"Share TXT/MD/CSV files via PRs or issues so we can wire class-aware insights faster.</p>"
"<p><b>GitHub:</b> <a href='https://github.com/stalcup-dev/tl-dps-mcp'>github.com/stalcup-dev/tl-dps-mcp</a><br>"
"<b>Issues:</b> <a href='https://github.com/stalcup-dev/tl-dps-mcp/issues'>Report bugs or request features</a><br>"
"<b>Email:</b> <a href='mailto:allen.stalc@gmail.com'>allen.stalc@gmail.com</a></p>"
)
links_label.setWordWrap(True)
links_label.setOpenExternalLinks(True)
links_label.setTextInteractionFlags(Qt.TextBrowserInteraction)
layout.addWidget(links_label)
# Safety note
safety = QLabel(
"<p style='color: #666;'><b>Safety Note:</b> This app never writes files from MCP tools, "
"enforces SELECT-only SQL, and runs model inference locally—no data leaves your machine.</p>"
)
safety.setWordWrap(True)
layout.addWidget(safety)
# Close button
close_btn = QPushButton("Close")
close_btn.clicked.connect(self.accept)
close_btn.setMinimumHeight(36)
layout.addWidget(close_btn)
self.setLayout(layout)
def main() -> None:
app = QApplication(sys.argv)
window = MainWindow()
window.resize(960, 600)
window.show()
sys.exit(app.exec())
if __name__ == "__main__":
main()
