storm/cthulhu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Visual speech monitor added to speech history plugin. Toggle with cthulhu+shift+D

Browse Source
This commit is contained in:
Storm Dragon
2026-02-17 08:40:51 -05:00
parent 40e63150a6
commit ed78ffc248
3 changed files with 240 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/cthulhu/plugins/SpeechHistory/plugin.info
View File

@@ -1,9 +1,8 @@
name = Speech History
version = 1.0.0
description = Shows a searchable history of the last 50 unique utterances spoken by Cthulhu
version = 1.2.0
description = Shows speech history and a live floating monitor of spoken text from Cthulhu
authors = Stormux
website = https://git.stormux.org/storm/cthulhu
copyright = Copyright 2025 Stormux
builtin = false
hidden = false

298
src/cthulhu/plugins/SpeechHistory/plugin.py
View File

@@ -7,39 +7,47 @@
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
"""Speech History plugin for Cthulhu."""
"""Speech history and speech monitor plugin for Cthulhu."""
import logging
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk, Gdk
from gi.repository import Gtk, Gdk, GLib
from cthulhu.plugin import Plugin, cthulhu_hookimpl
from cthulhu import debug
from cthulhu import speech
from cthulhu import speech_history
logger = logging.getLogger(__name__)
class SpeechHistory(Plugin):
"""Plugin that displays a window containing recent spoken output."""
"""Plugin that provides speech history and a live speech monitor."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._activated = False
self._kbOpenWindow = None
self._kbToggleMonitor = None
# Speech history window state
self._window = None
self._filterEntry = None
self._filterText = ""
self._listStore = None
self._filterModel = None
self._treeView = None
self._capturePaused = False
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Plugin initialized", True)
# Live monitor window state
self._monitorWindow = None
self._monitorTextView = None
self._monitorTextBuffer = None
self._monitorEndMark = None
self._monitorLineCount = 0
self._monitorMaxLines = 500
@cthulhu_hookimpl
def activate(self, plugin=None):
@@ -47,17 +55,13 @@ class SpeechHistory(Plugin):
return
if self._activated:
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Already activated, skipping", True)
return True
try:
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Activating plugin", True)
self._register_keybinding()
self._register_keybindings()
self._activated = True
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Activated successfully", True)
return True
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR during activate: {e}", True)
except Exception:
logger.exception("Error activating SpeechHistory plugin")
return False
@@ -67,46 +71,208 @@ class SpeechHistory(Plugin):
return
try:
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Deactivating plugin", True)
self._close_window()
self._disable_monitor()
self._kbOpenWindow = None
self._kbToggleMonitor = None
self._activated = False
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Deactivated successfully", True)
return True
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR during deactivate: {e}", True)
except Exception:
logger.exception("Error deactivating SpeechHistory plugin")
return False
def _register_keybinding(self):
try:
if not self.app:
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: No app reference; cannot register keybinding", True)
return
def _register_keybindings(self):
if not self.app:
debug.printMessage(
debug.LEVEL_INFO,
"SpeechHistory: No app reference; cannot register keybindings",
True,
)
return
gestureString = "kb:cthulhu+control+h"
description = "Open speech history"
historyGesture = "kb:cthulhu+control+h"
historyDescription = "Open speech history"
self._kbOpenWindow = self.registerGestureByString(
self._open_window,
historyDescription,
historyGesture,
)
self._kbOpenWindow = self.registerGestureByString(
self._open_window,
description,
gestureString,
monitorGesture = "kb:cthulhu+shift+d"
monitorDescription = "Toggle speech monitor"
self._kbToggleMonitor = self.registerGestureByString(
self._toggle_monitor,
monitorDescription,
monitorGesture,
)
if not self._kbOpenWindow:
debug.printMessage(
debug.LEVEL_INFO,
f"SpeechHistory: Failed to register keybinding {historyGesture}",
True,
)
if self._kbOpenWindow:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: Registered keybinding {gestureString}", True)
else:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: Failed to register keybinding {gestureString}", True)
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR registering keybinding: {e}", True)
logger.exception("Error registering keybinding for SpeechHistory")
if not self._kbToggleMonitor:
debug.printMessage(
debug.LEVEL_INFO,
f"SpeechHistory: Failed to register keybinding {monitorGesture}",
True,
)
# Live monitor methods
def _toggle_monitor(self, script=None, inputEvent=None):
try:
if self._monitorWindow is not None:
self._disable_monitor()
self._present_message("Speech monitor disabled.")
return True
self._enable_monitor()
self._present_message("Speech monitor enabled.")
return True
except Exception:
logger.exception("Error toggling speech monitor")
self._disable_monitor()
self._present_message("Error toggling speech monitor.")
return False
def _enable_monitor(self):
if self._monitorWindow is not None:
self._monitorWindow.show_all()
return
self._create_monitor_window()
speech.set_monitor_callbacks(writeText=self._on_spoken_text)
self._monitorWindow.show_all()
def _disable_monitor(self):
speech.set_monitor_callbacks(writeText=None)
if self._monitorWindow is not None:
self._monitorWindow.destroy()
def _create_monitor_window(self):
self._monitorWindow = Gtk.Window(title="Speech Monitor - Cthulhu")
self._monitorWindow.set_default_size(900, 320)
self._monitorWindow.set_modal(False)
self._monitorWindow.set_border_width(8)
self._monitorWindow.set_keep_above(True)
self._monitorWindow.set_accept_focus(False)
self._monitorWindow.set_focus_on_map(False)
self._monitorWindow.set_skip_taskbar_hint(True)
self._monitorWindow.set_skip_pager_hint(True)
self._monitorWindow.set_type_hint(Gdk.WindowTypeHint.UTILITY)
self._monitorWindow.connect("destroy", self._on_monitor_window_destroy)
self._monitorWindow.connect("key-press-event", self._on_monitor_window_key_press)
mainBox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
scrolledWindow = Gtk.ScrolledWindow()
scrolledWindow.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
self._monitorTextView = Gtk.TextView()
self._monitorTextView.set_editable(False)
self._monitorTextView.set_cursor_visible(False)
self._monitorTextView.set_can_focus(False)
self._monitorTextView.set_wrap_mode(Gtk.WrapMode.WORD_CHAR)
self._monitorTextView.set_accepts_tab(False)
self._monitorTextView.set_left_margin(6)
self._monitorTextView.set_right_margin(6)
self._monitorTextView.set_top_margin(6)
self._monitorTextView.set_bottom_margin(6)
textAccessible = self._monitorTextView.get_accessible()
if textAccessible:
textAccessible.set_name("Speech monitor output")
self._monitorTextBuffer = self._monitorTextView.get_buffer()
self._monitorEndMark = self._monitorTextBuffer.create_mark(
"monitorEnd",
self._monitorTextBuffer.get_end_iter(),
False,
)
self._monitorLineCount = 0
scrolledWindow.add(self._monitorTextView)
mainBox.pack_start(scrolledWindow, True, True, 0)
buttonRow = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
buttonRow.set_halign(Gtk.Align.END)
clearButton = Gtk.Button(label="Clear")
closeButton = Gtk.Button(label="Close")
clearButton.connect("clicked", self._on_monitor_clear_clicked)
closeButton.connect("clicked", self._on_monitor_close_clicked)
buttonRow.pack_start(clearButton, False, False, 0)
buttonRow.pack_start(closeButton, False, False, 0)
mainBox.pack_start(buttonRow, False, False, 0)
self._monitorWindow.add(mainBox)
def _on_monitor_close_clicked(self, button):
self._disable_monitor()
def _on_monitor_clear_clicked(self, button):
if self._monitorTextBuffer is None:
return
self._monitorTextBuffer.set_text("")
self._monitorLineCount = 0
def _on_monitor_window_destroy(self, widget):
speech.set_monitor_callbacks(writeText=None)
self._monitorWindow = None
self._monitorTextView = None
self._monitorTextBuffer = None
self._monitorEndMark = None
self._monitorLineCount = 0
def _on_monitor_window_key_press(self, widget, event):
if event.keyval == Gdk.KEY_Escape:
self._disable_monitor()
return True
return False
def _on_spoken_text(self, text):
if not text or not isinstance(text, str):
return
GLib.idle_add(self._append_text_on_main_thread, text)
def _append_text_on_main_thread(self, text):
if self._monitorTextBuffer is None:
return False
self._monitorTextBuffer.insert(self._monitorTextBuffer.get_end_iter(), f"{text}\n")
self._monitorLineCount += 1
if self._monitorLineCount > self._monitorMaxLines:
excess = self._monitorLineCount - self._monitorMaxLines
startIter = self._monitorTextBuffer.get_start_iter()
cutIter = self._monitorTextBuffer.get_iter_at_line(excess)
self._monitorTextBuffer.delete(startIter, cutIter)
self._monitorLineCount = self._monitorMaxLines
if self._monitorEndMark is not None and self._monitorTextView is not None:
self._monitorTextBuffer.move_mark(
self._monitorEndMark,
self._monitorTextBuffer.get_end_iter(),
)
self._monitorTextView.scroll_mark_onscreen(self._monitorEndMark)
return False
# Speech history methods
def _open_window(self, script=None, inputEvent=None):
try:
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Open window requested", True)
if self._window:
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Window already open; presenting", True)
self._window.present()
return True
@@ -120,10 +286,8 @@ class SpeechHistory(Plugin):
elif self._filterEntry:
self._filterEntry.grab_focus()
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Window shown", True)
return True
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR opening window: {e}", True)
except Exception:
logger.exception("Error opening SpeechHistory window")
self._resume_capture()
return False
@@ -141,7 +305,6 @@ class SpeechHistory(Plugin):
self._filterText = ""
# Filter row
filterRow = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10)
filterLabel = Gtk.Label(label="_Filter:")
filterLabel.set_use_underline(True)
@@ -156,7 +319,6 @@ class SpeechHistory(Plugin):
filterRow.pack_start(self._filterEntry, True, True, 0)
mainBox.pack_start(filterRow, False, False, 0)
# List
self._listStore = Gtk.ListStore(str)
self._filterModel = self._listStore.filter_new()
self._filterModel.set_visible_func(self._filter_visible_func)
@@ -169,7 +331,7 @@ class SpeechHistory(Plugin):
textRenderer = Gtk.CellRendererText()
textRenderer.set_property("wrap-width", 640)
textRenderer.set_property("wrap-mode", 2) # Pango.WrapMode.WORD_CHAR
textRenderer.set_property("wrap-mode", 2)
textColumn = Gtk.TreeViewColumn("Spoken Text", textRenderer, text=0)
textColumn.set_resizable(True)
textColumn.set_expand(True)
@@ -180,7 +342,6 @@ class SpeechHistory(Plugin):
scrolled.add(self._treeView)
mainBox.pack_start(scrolled, True, True, 0)
# Buttons
buttonRow = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10)
buttonRow.set_halign(Gtk.Align.END)
@@ -201,16 +362,12 @@ class SpeechHistory(Plugin):
self._refresh_list(selectFirst=True)
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Window created", True)
def _on_filter_changed(self, entry):
try:
self._filterText = entry.get_text() or ""
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: Filter changed '{self._filterText}'", True)
if self._filterModel:
self._filterModel.refilter()
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR filtering: {e}", True)
except Exception:
logger.exception("Error updating speech history filter")
def _filter_visible_func(self, model, treeIter, data=None):
@@ -221,8 +378,7 @@ class SpeechHistory(Plugin):
spokenText = model[treeIter][0] or ""
return spokenText.lower().startswith(filterText)
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR in filter func: {e}", True)
except Exception:
return True
def _refresh_list(self, selectFirst=False):
@@ -230,31 +386,19 @@ class SpeechHistory(Plugin):
if not self._listStore:
return
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Refreshing list", True)
self._listStore.clear()
items = speech_history.get_items()
debug.printMessage(
debug.LEVEL_INFO,
f"SpeechHistory: Retrieved {len(items)} items (paused={speech_history.is_capture_paused()})",
True,
)
for item in items:
self._listStore.append([item])
if self._filterModel:
self._filterModel.refilter()
debug.printMessage(
debug.LEVEL_INFO,
f"SpeechHistory: Filtered items visible={len(self._filterModel)} filter='{self._filterText}'",
True,
)
if selectFirst and self._treeView and len(self._filterModel) > 0:
selection = self._treeView.get_selection()
selection.select_path(Gtk.TreePath.new_first())
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR refreshing list: {e}", True)
except Exception:
logger.exception("Error refreshing speech history list")
def _get_selected_text(self):
@@ -268,8 +412,7 @@ class SpeechHistory(Plugin):
return None
return model[treeIter][0]
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR getting selection: {e}", True)
except Exception:
logger.exception("Error getting selected speech history item")
return None
@@ -284,11 +427,8 @@ class SpeechHistory(Plugin):
clipboard.set_text(selectedText, -1)
clipboard.store()
preview = selectedText[:60] + ("..." if len(selectedText) > 60 else "")
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: Copied to clipboard '{preview}'", True)
self._present_message("Copied to clipboard.")
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR copying: {e}", True)
except Exception:
logger.exception("Error copying speech history item to clipboard")
self._present_message("Error copying to clipboard.")
@@ -300,33 +440,27 @@ class SpeechHistory(Plugin):
return
removed = speech_history.remove(selectedText)
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: Remove requested removed={removed}", True)
if removed:
self._refresh_list(selectFirst=True)
self._present_message("Removed from history.")
else:
self._present_message("Item not found in history.")
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR removing: {e}", True)
except Exception:
logger.exception("Error removing speech history item")
self._present_message("Error removing item from history.")
def _on_close_clicked(self, button):
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Close button clicked", True)
self._close_window()
def _close_window(self):
try:
if self._window:
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Closing window", True)
self._window.destroy()
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR closing window: {e}", True)
except Exception:
logger.exception("Error closing SpeechHistory window")
self._resume_capture()
def _on_window_destroy(self, widget):
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Window destroyed", True)
self._window = None
self._filterEntry = None
self._listStore = None
@@ -338,14 +472,12 @@ class SpeechHistory(Plugin):
def _on_window_key_press(self, widget, event):
try:
if event.keyval == Gdk.KEY_Escape:
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Escape pressed; closing window", True)
self._close_window()
return True
if not self._filterEntry or self._filterEntry.is_focus():
return False
# If user starts typing anywhere, move focus to filter and update it.
if event.keyval == Gdk.KEY_BackSpace:
currentText = self._filterEntry.get_text() or ""
if currentText:
@@ -376,8 +508,7 @@ class SpeechHistory(Plugin):
self._filterEntry.set_text(currentText + charTyped)
self._filterEntry.set_position(-1)
return True
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR in key handler: {e}", True)
except Exception:
logger.exception("Error handling key press in SpeechHistory window")
return False
@@ -385,7 +516,6 @@ class SpeechHistory(Plugin):
if self._capturePaused:
return
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Pausing capture while window is open", True)
speech_history.pause_capture(reason="SpeechHistory window open")
self._capturePaused = True
@@ -393,7 +523,6 @@ class SpeechHistory(Plugin):
if not self._capturePaused:
return
debug.printMessage(debug.LEVEL_INFO, "SpeechHistory: Resuming capture (window closed)", True)
speech_history.resume_capture(reason="SpeechHistory window closed")
self._capturePaused = False
@@ -408,6 +537,5 @@ class SpeechHistory(Plugin):
state.activeScript.presentMessage(message, resetStyles=False)
else:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: {message}", True)
except Exception as e:
debug.printMessage(debug.LEVEL_INFO, f"SpeechHistory: ERROR presenting message: {e}", True)
except Exception:
logger.exception("Error presenting message from SpeechHistory")

26
src/cthulhu/speech.py
View File

@@ -36,7 +36,7 @@ __license__ = "LGPL"
import importlib
import time
from typing import TYPE_CHECKING, Optional, List, Dict, Any, Union
from typing import TYPE_CHECKING, Optional, List, Dict, Any, Union, Callable
from . import debug
from . import logger
@@ -73,6 +73,9 @@ _speechserver: Optional[SpeechServer] = None
# The last time something was spoken.
_timestamp: float = 0.0
# Optional callback for live monitoring of spoken text.
_monitorWriteTextCallback: Optional[Callable[[str], None]] = None
def _initSpeechServer(moduleName: Optional[str], speechServerInfo: Optional[Any]) -> None:
global _speechserver
@@ -168,6 +171,21 @@ def setSpeechServer(speechServer: SpeechServer) -> None:
global _speechserver
_speechserver = speechServer
def set_monitor_callbacks(writeText: Optional[Callable[[str], None]] = None) -> None:
"""Sets runtime callbacks for live speech monitoring."""
global _monitorWriteTextCallback
_monitorWriteTextCallback = writeText
def _write_to_monitor(text: str) -> None:
"""Writes text to the active speech monitor callback if set."""
if _monitorWriteTextCallback is None:
return
try:
_monitorWriteTextCallback(text)
except Exception:
debug.printException(debug.LEVEL_INFO)
def __resolveACSS(acss: Optional[Any] = None) -> ACSS:
if isinstance(acss, ACSS):
family = acss.get(acss.FAMILY)
@@ -237,6 +255,8 @@ def _speak(text: str, acss: Optional[Any], interrupt: bool) -> None:
except Exception:
debug.printException(debug.LEVEL_INFO)
_write_to_monitor(text)
if not _speechserver:
logLine = f"SPEECH OUTPUT: '{text}' {acss}"
debug.printMessage(debug.LEVEL_INFO, logLine, True)
@@ -393,6 +413,8 @@ def speakKeyEvent(event: Any, acss: Optional[Any] = None) -> None:
if log:
log.info(logLine)
_write_to_monitor(msg.strip())
if _speechserver:
_speechserver.speakKeyEvent(event, acss) # type: ignore
@@ -425,6 +447,8 @@ def speakCharacter(character: str, acss: Optional[Any] = None) -> None:
if log:
log.info(f"SPEECH OUTPUT: '{character}'")
_write_to_monitor(character)
if _speechserver:
_speechserver.speakCharacter(character, acss=acss) # type: ignore