ed/manual_slop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(gui): Implement event-driven API metrics updates and decouple from render loop

Browse Source
This commit is contained in:
Edward R. Gonzalez
2026-02-23 16:38:23 -05:00
parent 0c27aa6c6b
commit 2dd6145bd8
4 changed files with 138 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
gui.py
View File

@@ -496,6 +496,11 @@ class App:
self._is_script_blinking = False
self._script_blink_start_time = 0.0
# Subscribe to API lifecycle events
ai_client.events.on("request_start", self._on_api_event)
ai_client.events.on("response_received", self._on_api_event)
ai_client.events.on("tool_execution", self._on_api_event)
self.perf_monitor = PerformanceMonitor()
self.perf_history = {
"frame_time": [0.0] * 100,
@@ -822,12 +827,16 @@ class App:
total = usage["input_tokens"] + usage["output_tokens"]
dpg.set_value("ai_token_usage", f"Tokens: {total} (In: {usage['input_tokens']} Out: {usage['output_tokens']})")
def _update_telemetry_panel(self):
"""Updates the token budget visualizer in the Provider panel."""
# Update history bleed stats for all providers (throttled)
now = time.time()
if now - self._last_bleed_update_time > 2.0:
self._last_bleed_update_time = now
def _on_api_event(self, *args, **kwargs):
"""Callback for ai_client events. Queues a telemetry refresh on the main thread."""
with self._pending_gui_tasks_lock:
self._pending_gui_tasks.append({"action": "refresh_api_metrics"})
def _refresh_api_metrics(self):
"""Updates the token budget and cache stats visualizers."""
self._last_bleed_update_time = time.time()
# History bleed
stats = ai_client.get_history_bleed_stats()
if dpg.does_item_exist("token_budget_bar"):
percentage = stats.get("percentage", 0.0)
@@ -837,10 +846,7 @@ class App:
limit = stats.get("limit", 0)
dpg.set_value("token_budget_label", f"{current:,} / {limit:,}")
# Update Gemini-specific cache stats (throttled with diagnostics)
if now - self._last_diag_update_time > 10.0:
self._last_diag_update_time = now
# Gemini cache
if dpg.does_item_exist("gemini_cache_label"):
if self.current_provider == "gemini":
try:
@@ -851,12 +857,15 @@ class App:
text = f"Gemini Caches: {count} ({size_kb:.1f} KB)"
dpg.set_value("gemini_cache_label", text)
dpg.configure_item("gemini_cache_label", show=True)
except Exception as e:
# If the API call fails, just hide the label
except Exception:
dpg.configure_item("gemini_cache_label", show=False)
else:
dpg.configure_item("gemini_cache_label", show=False)
def _update_performance_diagnostics(self):
"""Updates performance diagnostics displays (throttled)."""
now = time.time()
# Update Diagnostics panel (throttled for smoothness)
if now - self._last_perf_update_time > 0.5:
self._last_perf_update_time = now
@@ -2221,6 +2230,34 @@ class App:
dpg.add_line_series(list(range(100)), self.perf_history["cpu"], label="cpu usage", tag="perf_cpu_plot")
dpg.set_axis_limits("axis_cpu_y", 0, 100)
def _process_pending_gui_tasks(self):
"""Processes tasks queued from background threads on the main thread."""
if not self._pending_gui_tasks:
return
with self._pending_gui_tasks_lock:
gui_tasks = self._pending_gui_tasks[:]
self._pending_gui_tasks.clear()
for task in gui_tasks:
try:
action = task.get("action")
if action == "set_value":
item = task.get("item")
val = task.get("value")
if item and dpg.does_item_exist(item):
dpg.set_value(item, val)
elif action == "click":
item = task.get("item")
if item and dpg.does_item_exist(item):
cb = dpg.get_item_callback(item)
if cb:
cb()
elif action == "refresh_api_metrics":
self._refresh_api_metrics()
except Exception as e:
print(f"Error executing GUI hook task: {e}")
def run(self):
dpg.create_context()
dpg.configure_app(docking=True, docking_space=True, init_file="dpg_layout.ini")
@@ -2272,25 +2309,7 @@ class App:
# Process queued API GUI tasks
self.perf_monitor.start_component("GUI_Tasks")
with self._pending_gui_tasks_lock:
gui_tasks = self._pending_gui_tasks[:]
self._pending_gui_tasks.clear()
for task in gui_tasks:
try:
action = task.get("action")
if action == "set_value":
item = task.get("item")
val = task.get("value")
if item and dpg.does_item_exist(item):
dpg.set_value(item, val)
elif action == "click":
item = task.get("item")
if item and dpg.does_item_exist(item):
cb = dpg.get_item_callback(item)
if cb:
cb()
except Exception as e:
print(f"Error executing GUI hook task: {e}")
self._process_pending_gui_tasks()
self.perf_monitor.end_component("GUI_Tasks")
# Handle retro arcade blinking effect
@@ -2394,7 +2413,7 @@ class App:
self.perf_monitor.end_component("Comms")
self.perf_monitor.start_component("Telemetry")
self._update_telemetry_panel()
self._update_performance_diagnostics()
self.perf_monitor.end_component("Telemetry")
self.perf_monitor.end_frame()

2
tests/test_gui_diagnostics.py
View File

@@ -53,7 +53,7 @@ def test_diagnostics_panel_updates(app_instance):
# We also need to mock ai_client stats
with patch('ai_client.get_history_bleed_stats', return_value={}):
app_instance._update_telemetry_panel()
app_instance._update_performance_diagnostics()
# Verify UI updates
mock_set_value.assert_any_call("perf_fps_text", "100.0")

62
tests/test_gui_events.py Normal file
View File

@@ -0,0 +1,62 @@
import pytest
from unittest.mock import MagicMock, patch
import dearpygui.dearpygui as dpg
import gui
from gui import App
import ai_client
@pytest.fixture
def app_instance():
"""
Fixture to create an instance of the App class for testing.
It creates a real DPG context but mocks functions that would
render a window or block execution.
"""
dpg.create_context()
with patch('dearpygui.dearpygui.create_viewport'), \
patch('dearpygui.dearpygui.setup_dearpygui'), \
patch('dearpygui.dearpygui.show_viewport'), \
patch('dearpygui.dearpygui.start_dearpygui'), \
patch('gui.load_config', return_value={}), \
patch('gui.PerformanceMonitor'), \
patch('gui.shell_runner'), \
patch('gui.project_manager'), \
patch.object(App, '_load_active_project'), \
patch.object(App, '_rebuild_files_list'), \
patch.object(App, '_rebuild_shots_list'), \
patch.object(App, '_rebuild_disc_list'), \
patch.object(App, '_rebuild_disc_roles_list'), \
patch.object(App, '_rebuild_discussion_selector'), \
patch.object(App, '_refresh_project_widgets'):
app = App()
yield app
dpg.destroy_context()
def test_gui_updates_on_event(app_instance):
# Patch dependencies for the test
with patch('dearpygui.dearpygui.set_value') as mock_set_value, \
patch('dearpygui.dearpygui.does_item_exist', return_value=True), \
patch('dearpygui.dearpygui.configure_item'), \
patch('ai_client.get_history_bleed_stats') as mock_stats:
mock_stats.return_value = {"percentage": 50.0, "current": 500, "limit": 1000}
# We'll use patch.object to see if _refresh_api_metrics is called
with patch.object(app_instance, '_refresh_api_metrics', wraps=app_instance._refresh_api_metrics) as mock_refresh:
# Simulate event
ai_client.events.emit("response_received", payload={})
# Process tasks manually
app_instance._process_pending_gui_tasks()
# Verify that _refresh_api_metrics was called
mock_refresh.assert_called_once()
# Verify that dpg.set_value was called for the metrics widgets
calls = [call.args[0] for call in mock_set_value.call_args_list]
assert "token_budget_bar" in calls
assert "token_budget_label" in calls

8
tests/test_gui_updates.py
View File

@@ -41,7 +41,7 @@ def app_instance():
def test_telemetry_panel_updates_correctly(app_instance):
"""
Tests that the _update_telemetry_panel method correctly updates
Tests that the _update_performance_diagnostics method correctly updates
DPG widgets based on the stats from ai_client.
"""
# 1. Set the provider to anthropic
@@ -64,7 +64,7 @@ def test_telemetry_panel_updates_correctly(app_instance):
patch('dearpygui.dearpygui.does_item_exist', return_value=True) as mock_does_item_exist:
# 4. Call the method under test
app_instance._update_telemetry_panel()
app_instance._refresh_api_metrics()
# 5. Assert the results
mock_get_stats.assert_called_once()
@@ -78,7 +78,7 @@ def test_telemetry_panel_updates_correctly(app_instance):
def test_cache_data_display_updates_correctly(app_instance):
"""
Tests that the _update_telemetry_panel method correctly updates the
Tests that the _update_performance_diagnostics method correctly updates the
GUI with Gemini cache statistics when the provider is set to Gemini.
"""
# 1. Set the provider to Gemini
@@ -104,7 +104,7 @@ def test_cache_data_display_updates_correctly(app_instance):
with patch('ai_client.get_history_bleed_stats', return_value={}):
# 4. Call the method under test
app_instance._update_telemetry_panel()
app_instance._refresh_api_metrics()
# 5. Assert the results
mock_get_cache_stats.assert_called_once()