324 lines
12 KiB
Python
324 lines
12 KiB
Python
"""
|
|
EU-Utility - Skill Scanner Plugin
|
|
|
|
Uses core OCR and Log services via PluginAPI.
|
|
"""
|
|
|
|
import re
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from PyQt6.QtWidgets import (
|
|
QWidget, QVBoxLayout, QHBoxLayout, QLabel,
|
|
QPushButton, QTableWidget, QTableWidgetItem, QProgressBar,
|
|
QFrame, QGroupBox, QTextEdit, QSplitter
|
|
)
|
|
from PyQt6.QtCore import Qt, QThread, pyqtSignal
|
|
|
|
from plugins.base_plugin import BasePlugin
|
|
|
|
|
|
class SkillOCRThread(QThread):
|
|
"""OCR scan using core service."""
|
|
scan_complete = pyqtSignal(dict)
|
|
scan_error = pyqtSignal(str)
|
|
progress_update = pyqtSignal(str)
|
|
|
|
def __init__(self, ocr_service):
|
|
super().__init__()
|
|
self.ocr_service = ocr_service
|
|
|
|
def run(self):
|
|
"""Perform OCR using core service."""
|
|
try:
|
|
self.progress_update.emit("Capturing screen...")
|
|
|
|
# Use core OCR service
|
|
result = self.ocr_service.recognize()
|
|
|
|
if 'error' in result and result['error']:
|
|
self.scan_error.emit(result['error'])
|
|
return
|
|
|
|
self.progress_update.emit("Parsing skills...")
|
|
|
|
# Parse skills from text
|
|
skills_data = self._parse_skills(result.get('text', ''))
|
|
self.scan_complete.emit(skills_data)
|
|
|
|
except Exception as e:
|
|
self.scan_error.emit(str(e))
|
|
|
|
def _parse_skills(self, text):
|
|
"""Parse skill data from OCR text with improved handling for 3-column layout."""
|
|
skills = {}
|
|
|
|
# Ranks in Entropia Universe (in order)
|
|
RANKS = [
|
|
'Newbie', 'Inept', 'Beginner', 'Amateur', 'Average',
|
|
'Skilled', 'Expert', 'Professional', 'Master', 'Grand Master',
|
|
'Champion', 'Legendary', 'Guru', 'Astonishing', 'Remarkable',
|
|
'Outstanding', 'Marvelous', 'Prodigious', 'Amazing', 'Incredible', 'Awesome'
|
|
]
|
|
rank_pattern = '|'.join(RANKS)
|
|
|
|
# Clean up the text - remove common headers and junk
|
|
text = text.replace('SKILLS', '').replace('ALL CATEGORIES', '')
|
|
text = text.replace('SKILL NAME', '').replace('RANK', '').replace('POINTS', '')
|
|
text = text.replace('Attributes', '').replace('COMBAT', '').replace('Design', '')
|
|
text = text.replace('Construction', '').replace('Defense', '').replace('General', '')
|
|
text = text.replace('Handgun', '').replace('Heavy Melee Weapons', '')
|
|
text = text.replace('Information', '').replace('Inflict Melee Damage', '')
|
|
text = text.replace('Inflict Ranged Damage', '').replace('Light Melee Weapons', '')
|
|
text = text.replace('Longblades', '').replace('Medical', '').replace('Mining', '')
|
|
text = text.replace('Science', '').replace('Social', '').replace('Beauty', '')
|
|
text = text.replace('Mindforce', '')
|
|
|
|
lines = text.split('\n')
|
|
|
|
for line in lines:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
# Skip category headers and short lines
|
|
if len(line) < 10:
|
|
continue
|
|
|
|
# Try pattern: SkillName Rank Points
|
|
# More flexible pattern to handle merged text
|
|
match = re.search(
|
|
rf'([A-Za-z][A-Za-z\s]{{2,50}}?)\s+({rank_pattern})\s+(\d{{1,6}})(?:\s|$)',
|
|
line, re.IGNORECASE
|
|
)
|
|
|
|
if match:
|
|
skill_name = match.group(1).strip()
|
|
rank = match.group(2)
|
|
points = int(match.group(3))
|
|
|
|
# Clean up skill name
|
|
skill_name = skill_name.strip()
|
|
|
|
# Validate - points should be reasonable (not too small)
|
|
if points > 0:
|
|
skills[skill_name] = {
|
|
'rank': rank,
|
|
'points': points,
|
|
'scanned_at': datetime.now().isoformat()
|
|
}
|
|
print(f"[SkillScanner] Parsed: {skill_name} = {rank} ({points})")
|
|
|
|
# Alternative parsing: try to find skill-rank-points triplets
|
|
if not skills:
|
|
skills = self._parse_skills_alternative(text, RANKS)
|
|
|
|
return skills
|
|
|
|
def _parse_skills_alternative(self, text, ranks):
|
|
"""Alternative parser for when text is heavily merged."""
|
|
skills = {}
|
|
|
|
# Find all rank positions in the text
|
|
for rank in ranks:
|
|
# Look for pattern: [text] [Rank] [number]
|
|
pattern = rf'([A-Z][a-z]{{2,}}(?:\s+[A-Z][a-z]{{2,}}){{0,3}})\s+{rank}\s+(\d{{1,6}})'
|
|
matches = re.finditer(pattern, text, re.IGNORECASE)
|
|
|
|
for match in matches:
|
|
skill_name = match.group(1).strip()
|
|
points = int(match.group(2))
|
|
|
|
if points > 0 and len(skill_name) > 2:
|
|
skills[skill_name] = {
|
|
'rank': rank,
|
|
'points': points,
|
|
'scanned_at': datetime.now().isoformat()
|
|
}
|
|
|
|
return skills
|
|
|
|
|
|
class SkillScannerPlugin(BasePlugin):
|
|
"""Scan skills using core OCR and track gains via core Log service."""
|
|
|
|
name = "Skill Scanner"
|
|
version = "2.1.0"
|
|
author = "ImpulsiveFPS"
|
|
description = "Uses core OCR and Log services"
|
|
hotkey = "ctrl+shift+s"
|
|
|
|
def initialize(self):
|
|
"""Setup skill scanner."""
|
|
self.data_file = Path("data/skill_tracker.json")
|
|
self.data_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Load saved data
|
|
self.skills_data = {}
|
|
self.skill_gains = []
|
|
self._load_data()
|
|
|
|
# Subscribe to skill gain events from core Log service
|
|
try:
|
|
from core.plugin_api import get_api
|
|
api = get_api()
|
|
|
|
# Check if log service available
|
|
log_service = api.services.get('log')
|
|
if log_service:
|
|
print(f"[SkillScanner] Connected to core Log service")
|
|
|
|
except Exception as e:
|
|
print(f"[SkillScanner] Could not connect to Log service: {e}")
|
|
|
|
def _load_data(self):
|
|
"""Load saved skill data."""
|
|
if self.data_file.exists():
|
|
try:
|
|
with open(self.data_file, 'r') as f:
|
|
data = json.load(f)
|
|
self.skills_data = data.get('skills', {})
|
|
self.skill_gains = data.get('gains', [])
|
|
except:
|
|
pass
|
|
|
|
def _save_data(self):
|
|
"""Save skill data."""
|
|
with open(self.data_file, 'w') as f:
|
|
json.dump({
|
|
'skills': self.skills_data,
|
|
'gains': self.skill_gains
|
|
}, f, indent=2)
|
|
|
|
def get_ui(self):
|
|
"""Create skill scanner UI."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
layout.setSpacing(15)
|
|
layout.setContentsMargins(0, 0, 0, 0)
|
|
|
|
# Header
|
|
header = QLabel("Skill Tracker")
|
|
header.setStyleSheet("font-size: 18px; font-weight: bold; color: white;")
|
|
layout.addWidget(header)
|
|
|
|
# Info about core services
|
|
info = self._get_service_status()
|
|
info_label = QLabel(info)
|
|
info_label.setStyleSheet("color: rgba(255,255,255,100); font-size: 11px;")
|
|
layout.addWidget(info_label)
|
|
|
|
# Splitter
|
|
splitter = QSplitter(Qt.Orientation.Vertical)
|
|
|
|
# Scan section
|
|
scan_group = QGroupBox("OCR Scan (Core Service)")
|
|
scan_layout = QVBoxLayout(scan_group)
|
|
|
|
scan_btn = QPushButton("Scan Skills Window")
|
|
scan_btn.setStyleSheet("""
|
|
QPushButton {
|
|
background-color: #ff8c42;
|
|
color: white;
|
|
padding: 12px;
|
|
border: none;
|
|
border-radius: 4px;
|
|
font-weight: bold;
|
|
}
|
|
""")
|
|
scan_btn.clicked.connect(self._scan_skills)
|
|
scan_layout.addWidget(scan_btn)
|
|
|
|
self.scan_progress = QLabel("Ready to scan")
|
|
self.scan_progress.setStyleSheet("color: rgba(255,255,255,150);")
|
|
scan_layout.addWidget(self.scan_progress)
|
|
|
|
self.skills_table = QTableWidget()
|
|
self.skills_table.setColumnCount(3)
|
|
self.skills_table.setHorizontalHeaderLabels(["Skill", "Rank", "Points"])
|
|
self.skills_table.horizontalHeader().setStretchLastSection(True)
|
|
scan_layout.addWidget(self.skills_table)
|
|
|
|
splitter.addWidget(scan_group)
|
|
|
|
# Log section
|
|
log_group = QGroupBox("Log Tracking (Core Service)")
|
|
log_layout = QVBoxLayout(log_group)
|
|
|
|
log_status = QLabel("Skill gains tracked from chat log")
|
|
log_status.setStyleSheet("color: #4ecdc4;")
|
|
log_layout.addWidget(log_status)
|
|
|
|
self.gains_text = QTextEdit()
|
|
self.gains_text.setReadOnly(True)
|
|
self.gains_text.setMaximumHeight(150)
|
|
self.gains_text.setPlaceholderText("Recent skill gains from core Log service...")
|
|
log_layout.addWidget(self.gains_text)
|
|
|
|
self.total_gains_label = QLabel(f"Total gains: {len(self.skill_gains)}")
|
|
self.total_gains_label.setStyleSheet("color: rgba(255,255,255,150);")
|
|
log_layout.addWidget(self.total_gains_label)
|
|
|
|
splitter.addWidget(log_group)
|
|
|
|
layout.addWidget(splitter)
|
|
|
|
self._refresh_skills_table()
|
|
|
|
return widget
|
|
|
|
def _get_service_status(self) -> str:
|
|
"""Get status of core services."""
|
|
try:
|
|
from core.ocr_service import get_ocr_service
|
|
from core.log_reader import get_log_reader
|
|
|
|
ocr = get_ocr_service()
|
|
log = get_log_reader()
|
|
|
|
ocr_status = "✓" if ocr.is_available() else "✗"
|
|
log_status = "✓" if log.is_available() else "✗"
|
|
|
|
return f"Core Services - OCR: {ocr_status} Log: {log_status}"
|
|
except:
|
|
return "Core Services - status unknown"
|
|
|
|
def _scan_skills(self):
|
|
"""Start OCR scan using core service."""
|
|
try:
|
|
from core.ocr_service import get_ocr_service
|
|
ocr_service = get_ocr_service()
|
|
|
|
if not ocr_service.is_available():
|
|
self.scan_progress.setText("Error: OCR service not available")
|
|
return
|
|
|
|
self.scanner = SkillOCRThread(ocr_service)
|
|
self.scanner.scan_complete.connect(self._on_scan_complete)
|
|
self.scanner.scan_error.connect(self._on_scan_error)
|
|
self.scanner.progress_update.connect(self._on_scan_progress)
|
|
self.scanner.start()
|
|
|
|
except Exception as e:
|
|
self.scan_progress.setText(f"Error: {e}")
|
|
|
|
def _on_scan_progress(self, message):
|
|
self.scan_progress.setText(message)
|
|
|
|
def _on_scan_complete(self, skills_data):
|
|
self.skills_data.update(skills_data)
|
|
self._save_data()
|
|
self._refresh_skills_table()
|
|
self.scan_progress.setText(f"Found {len(skills_data)} skills")
|
|
|
|
def _on_scan_error(self, error):
|
|
self.scan_progress.setText(f"Error: {error}")
|
|
|
|
def _refresh_skills_table(self):
|
|
self.skills_table.setRowCount(len(self.skills_data))
|
|
for i, (name, data) in enumerate(sorted(self.skills_data.items())):
|
|
self.skills_table.setItem(i, 0, QTableWidgetItem(name))
|
|
self.skills_table.setItem(i, 1, QTableWidgetItem(data.get('rank', '-')))
|
|
self.skills_table.setItem(i, 2, QTableWidgetItem(str(data.get('points', 0))))
|