212 lines
7.4 KiB
Python
212 lines
7.4 KiB
Python
"""
|
|
Example plugin demonstrating Enhanced Event Bus usage.
|
|
|
|
This plugin shows how to use:
|
|
- publish_typed() - Publish typed events
|
|
- subscribe_typed() - Subscribe with filtering
|
|
- get_recent_events() - Retrieve event history
|
|
- Event filtering (min_damage, mob_types, etc.)
|
|
- Event replay for new subscribers
|
|
"""
|
|
|
|
from plugins.base_plugin import BasePlugin
|
|
from core.event_bus import (
|
|
SkillGainEvent, LootEvent, DamageEvent, GlobalEvent,
|
|
EventCategory
|
|
)
|
|
|
|
|
|
class EventBusExamplePlugin(BasePlugin):
|
|
"""Example plugin showing Enhanced Event Bus usage."""
|
|
|
|
name = "Event Bus Example"
|
|
version = "1.0.0"
|
|
author = "EU-Utility"
|
|
description = "Demonstrates Enhanced Event Bus features"
|
|
|
|
def __init__(self, overlay_window, config):
|
|
super().__init__(overlay_window, config)
|
|
self.big_hits = []
|
|
self.skill_gains = []
|
|
self.dragon_loot = []
|
|
self._subscriptions = []
|
|
|
|
def initialize(self) -> None:
|
|
"""Setup event subscriptions."""
|
|
print(f"[{self.name}] Initializing...")
|
|
|
|
# 1. Subscribe to ALL damage events
|
|
sub_id = self.subscribe_typed(
|
|
DamageEvent,
|
|
self.on_any_damage,
|
|
replay_last=5 # Replay last 5 damage events on subscribe
|
|
)
|
|
self._subscriptions.append(sub_id)
|
|
print(f"[{self.name}] Subscribed to all damage events")
|
|
|
|
# 2. Subscribe to HIGH damage events only (filtering)
|
|
sub_id = self.subscribe_typed(
|
|
DamageEvent,
|
|
self.on_big_damage,
|
|
min_damage=100, # Only events with damage >= 100
|
|
replay_last=3
|
|
)
|
|
self._subscriptions.append(sub_id)
|
|
print(f"[{self.name}] Subscribed to high damage events (≥100)")
|
|
|
|
# 3. Subscribe to skill gains for specific skills
|
|
sub_id = self.subscribe_typed(
|
|
SkillGainEvent,
|
|
self.on_combat_skill_gain,
|
|
skill_names=["Rifle", "Handgun", "Sword", "Knife"],
|
|
replay_last=10
|
|
)
|
|
self._subscriptions.append(sub_id)
|
|
print(f"[{self.name}] Subscribed to combat skill gains")
|
|
|
|
# 4. Subscribe to loot from specific mobs
|
|
sub_id = self.subscribe_typed(
|
|
LootEvent,
|
|
self.on_dragon_loot,
|
|
mob_types=["Dragon", "Drake", "Dragon Old"],
|
|
replay_last=5
|
|
)
|
|
self._subscriptions.append(sub_id)
|
|
print(f"[{self.name}] Subscribed to Dragon/Drake loot")
|
|
|
|
# 5. Subscribe to ALL globals
|
|
sub_id = self.subscribe_typed(
|
|
GlobalEvent,
|
|
self.on_global_announcement
|
|
)
|
|
self._subscriptions.append(sub_id)
|
|
print(f"[{self.name}] Subscribed to global announcements")
|
|
|
|
# 6. Demonstrate publishing events
|
|
self._publish_example_events()
|
|
|
|
# 7. Show event stats
|
|
stats = self.get_event_stats()
|
|
print(f"[{self.name}] Event Bus Stats:")
|
|
print(f" - Total published: {stats.get('total_published', 0)}")
|
|
print(f" - Active subs: {stats.get('active_subscriptions', 0)}")
|
|
|
|
def _publish_example_events(self):
|
|
"""Publish some example events to demonstrate."""
|
|
# Publish a skill gain
|
|
self.publish_typed(SkillGainEvent(
|
|
skill_name="Rifle",
|
|
skill_value=25.5,
|
|
gain_amount=0.01,
|
|
source="example_plugin"
|
|
))
|
|
|
|
# Publish some damage events
|
|
self.publish_typed(DamageEvent(
|
|
damage_amount=50.5,
|
|
damage_type="impact",
|
|
is_outgoing=True,
|
|
target_name="Berycled Young",
|
|
source="example_plugin"
|
|
))
|
|
|
|
self.publish_typed(DamageEvent(
|
|
damage_amount=150.0,
|
|
damage_type="penetration",
|
|
is_critical=True,
|
|
is_outgoing=True,
|
|
target_name="Daikiba",
|
|
source="example_plugin"
|
|
))
|
|
|
|
# Publish loot
|
|
self.publish_typed(LootEvent(
|
|
mob_name="Dragon",
|
|
items=[
|
|
{"name": "Dragon Scale", "value": 15.0},
|
|
{"name": "Animal Oil", "value": 0.05}
|
|
],
|
|
total_tt_value=15.05,
|
|
source="example_plugin"
|
|
))
|
|
|
|
print(f"[{self.name}] Published example events")
|
|
|
|
# ========== Event Handlers ==========
|
|
|
|
def on_any_damage(self, event: DamageEvent):
|
|
"""Handle all damage events."""
|
|
direction = "dealt" if event.is_outgoing else "received"
|
|
crit = " CRITICAL" if event.is_critical else ""
|
|
print(f"[{self.name}] Damage {direction}: {event.damage_amount:.1f}{crit} to {event.target_name}")
|
|
|
|
def on_big_damage(self, event: DamageEvent):
|
|
"""Handle only high damage events (filtered)."""
|
|
self.big_hits.append(event)
|
|
print(f"[{self.name}] 💥 BIG HIT! {event.damage_amount:.1f} damage to {event.target_name}")
|
|
print(f"[{self.name}] Total big hits recorded: {len(self.big_hits)}")
|
|
|
|
def on_combat_skill_gain(self, event: SkillGainEvent):
|
|
"""Handle combat skill gains."""
|
|
self.skill_gains.append(event)
|
|
print(f"[{self.name}] ⚔️ Skill up: {event.skill_name} +{event.gain_amount:.4f} = {event.skill_value:.4f}")
|
|
|
|
def on_dragon_loot(self, event: LootEvent):
|
|
"""Handle Dragon/Drake loot."""
|
|
self.dragon_loot.append(event)
|
|
items_str = ", ".join(event.get_item_names())
|
|
print(f"[{self.name}] 🐉 Dragon loot from {event.mob_name}: {items_str} (TT: {event.total_tt_value:.2f} PED)")
|
|
|
|
def on_global_announcement(self, event: GlobalEvent):
|
|
"""Handle global announcements."""
|
|
item_str = f" with {event.item_name}" if event.item_name else ""
|
|
print(f"[{self.name}] 🌍 GLOBAL: {event.player_name} - {event.achievement_type.upper()}{item_str} ({event.value:.2f} PED)")
|
|
|
|
def get_ui(self):
|
|
"""Return simple info panel."""
|
|
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QLabel, QTextEdit
|
|
|
|
widget = QWidget()
|
|
layout = QVBoxLayout()
|
|
|
|
# Title
|
|
title = QLabel(f"<h2>{self.name}</h2>")
|
|
layout.addWidget(title)
|
|
|
|
# Stats
|
|
stats_text = f"""
|
|
<b>Recorded Events:</b><br>
|
|
• Big Hits (≥100 dmg): {len(self.big_hits)}<br>
|
|
• Combat Skill Gains: {len(self.skill_gains)}<br>
|
|
• Dragon Loot: {len(self.dragon_loot)}<br>
|
|
<br>
|
|
<b>Active Subscriptions:</b> {len(self._subscriptions)}
|
|
"""
|
|
stats_label = QLabel(stats_text)
|
|
stats_label.setWordWrap(True)
|
|
layout.addWidget(stats_label)
|
|
|
|
# Recent events
|
|
layout.addWidget(QLabel("<b>Recent Combat Events:</b>"))
|
|
text_area = QTextEdit()
|
|
text_area.setReadOnly(True)
|
|
text_area.setMaximumHeight(200)
|
|
|
|
# Get recent damage events from event bus
|
|
recent = self.get_recent_events(DamageEvent, count=10)
|
|
events_text = "\\n".join([
|
|
f"• {e.damage_amount:.1f} dmg to {e.target_name}"
|
|
for e in reversed(recent)
|
|
]) or "No recent damage events"
|
|
text_area.setText(events_text)
|
|
layout.addWidget(text_area)
|
|
|
|
widget.setLayout(layout)
|
|
return widget
|
|
|
|
def shutdown(self) -> None:
|
|
"""Cleanup."""
|
|
print(f"[{self.name}] Shutting down...")
|
|
# Unsubscribe from all typed events (handled by base class)
|
|
super().shutdown()
|