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

Start updating plugins.

Browse Source
This commit is contained in:
Storm Dragon 2025-03-26 01:21:28 -04:00
parent dfb53fff89
commit 654f1acc21
9 changed files with 215 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/cthulhu/cthulhuVersion.py
View File

@ -23,5 +23,5 @@
# Fork of Orca Screen Reader (GNOME)
# Original source: https://gitlab.gnome.org/GNOME/orca
version = "2025.03.25"
version = "2025.03.26"
codeName = "testing"

2
src/cthulhu/plugins/Makefile.am
View File

@ -1,4 +1,4 @@
SUBDIRS = Clipboard DisplayVersion HelloWorld SelfVoice Time MouseReview Date ByeCthulhu HelloCthulhu PluginManager CapsLockHack SimplePluginSystem
SUBDIRS = Clipboard DisplayVersion HelloWorld self_voice Time MouseReview Date ByeCthulhu HelloCthulhu PluginManager CapsLockHack SimplePluginSystem
cthulhu_pythondir=$(pkgpythondir)/plugins

7
src/cthulhu/plugins/SelfVoice/Makefile.am
View File

@ -1,7 +0,0 @@
cthulhu_python_PYTHON = \
__init__.py \
SelfVoice.plugin \
SelfVoice.py
cthulhu_pythondir=$(pkgpythondir)/plugins/SelfVoice

6
src/cthulhu/plugins/SelfVoice/SelfVoice.plugin
View File

@ -1,6 +0,0 @@
[Plugin]
Module=SelfVoice
Loader=python3
Name=Self Voice Plugin
Description=use cthulhu text / braile from using unix sockets
Authors=Chrys chrys@linux-a11y.org

135
src/cthulhu/plugins/SelfVoice/SelfVoice.py
View File

@ -1,135 +0,0 @@
#!/usr/bin/env python3
#
# Copyright (c) 2024 Stormux
# Copyright (c) 2010-2012 The Orca Team
# Copyright (c) 2012 Igalia, S.L.
# Copyright (c) 2005-2010 Sun Microsystems Inc.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the
# Free Software Foundation, Inc., Franklin Street, Fifth Floor,
# Boston MA 02110-1301 USA.
#
# Fork of Orca Screen Reader (GNOME)
# Original source: https://gitlab.gnome.org/GNOME/orca
from cthulhu import plugin
import gi
gi.require_version('Peas', '1.0')
from gi.repository import GObject
from gi.repository import Peas
import select, socket, os, os.path
from threading import Thread, Lock
APPEND_CODE = '<#APPEND#>'
PERSISTENT_CODE = '<#PERSISTENT#>'
class SelfVoice(GObject.Object, Peas.Activatable, plugin.Plugin):
__gtype_name__ = 'SelfVoice'
object = GObject.Property(type=GObject.Object)
def __init__(self):
plugin.Plugin.__init__(self)
self.lock = Lock()
self.active = False
self.voiceThread = Thread(target=self.voiceWorker)
def do_activate(self):
API = self.object
self.activateWorker()
def do_deactivate(self):
API = self.object
self.deactivateWorker()
def do_update_state(self):
API = self.object
def deactivateWorker(self):
with self.lock:
self.active = False
self.voiceThread.join()
def activateWorker(self):
with self.lock:
self.active = True
self.voiceThread.start()
def isActive(self):
with self.lock:
return self.active
def outputMessage(self, Message):
# Prepare
API = self.object
append = Message.startswith(APPEND_CODE)
if append:
Message = Message[len(APPEND_CODE):]
if Message.endswith(PERSISTENT_CODE):
Message = Message[:len(Message)-len(PERSISTENT_CODE)]
API.app.getAPIHelper().outputMessage(Message, not append)
else:
script_manager = API.app.getDynamicApiManager().getAPI('ScriptManager')
scriptManager = script_manager.getManager()
scriptManager.getDefaultScript().presentMessage(Message, resetStyles=False)
return
try:
settings = API.app.getDynamicApiManager().getAPI('Settings')
braille = API.app.getDynamicApiManager().getAPI('Braille')
speech = API.app.getDynamicApiManager().getAPI('Speech')
# Speak
if speech != None:
if (settings.enableSpeech):
if not append:
speech.cancel()
if Message != '':
speech.speak(Message)
# Braille
if braille != None:
if (settings.enableBraille):
braille.displayMessage(Message)
except e as Exception:
print(e)
def voiceWorker(self):
socketFile = '/tmp/cthulhu.sock'
# for testing purposes
#socketFile = '/tmp/cthulhu-plugin.sock'
if os.path.exists(socketFile):
os.unlink(socketFile)
cthulhuSock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
cthulhuSock.bind(socketFile)
os.chmod(socketFile, 0o222)
cthulhuSock.listen(1)
while self.isActive():
# Check if the client is still connected and if data is available:
try:
r, _, _ = select.select([cthulhuSock], [], [], 0.8)
except select.error:
break
if r == []:
continue
if cthulhuSock in r:
client_sock, client_addr = cthulhuSock.accept()
try:
rawdata = client_sock.recv(8129)
data = rawdata.decode("utf-8").rstrip().lstrip()
self.outputMessage(data)
except:
pass
try:
client_sock.close()
except:
pass
if cthulhuSock:
cthulhuSock.close()
cthulhuSock = None
if os.path.exists(socketFile):
os.unlink(socketFile)

7
src/cthulhu/plugins/self_voice/Makefile.am Normal file
View File

@ -0,0 +1,7 @@
cthulhu_python_PYTHON = \
__init__.py \
plugin.info \
plugin.py
cthulhu_pythondir=$(pkgpythondir)/plugins/self_voice

0
src/cthulhu/plugins/SelfVoice/__init__.py → src/cthulhu/plugins/self_voice/__init__.py
View File

12
src/cthulhu/plugins/self_voice/plugin.info Normal file
View File

@ -0,0 +1,12 @@
[Plugin]
name=Self Voice
module_name=self_voice
version=1.0.0
description=Use Cthulhu speech and braille from external applications via Unix sockets
authors=Stormux
copyright=Copyright (c) 2024 Stormux
website=https://stormux.org
icon_name=audio-speakers
builtin=false
hidden=false
help_uri=https://stormux.org

194
src/cthulhu/plugins/self_voice/plugin.py Normal file
View File

@ -0,0 +1,194 @@
#!/usr/bin/env python3
#
# Copyright (c) 2024 Stormux
# Copyright (c) 2010-2012 The Orca Team
# Copyright (c) 2012 Igalia, S.L.
# Copyright (c) 2005-2010 Sun Microsystems Inc.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the
# Free Software Foundation, Inc., Franklin Street, Fifth Floor,
# Boston MA 02110-1301 USA.
#
# Fork of Orca Screen Reader (GNOME)
# Original source: https://gitlab.gnome.org/GNOME/orca
"""Self Voice plugin for Cthulhu screen reader."""
import os
import socket
import select
import logging
import threading
from threading import Thread, Lock
from cthulhu.plugin import Plugin, cthulhu_hookimpl
logger = logging.getLogger(__name__)
# Special codes for message handling
APPEND_CODE = '<#APPEND#>'
PERSISTENT_CODE = '<#PERSISTENT#>'
class SelfVoice(Plugin):
"""Plugin that provides a socket interface for external applications to send text to Cthulhu."""
def __init__(self):
"""Initialize the plugin."""
super().__init__()
self.lock = Lock()
self.active = False
self.voiceThread = Thread(target=self.voiceWorker)
self.voiceThread.daemon = True # Make thread exit when main thread exits
@cthulhu_hookimpl
def activate(self):
"""Activate the self-voice plugin."""
super().activate()
logger.info("Activating Self Voice Plugin")
self.activateWorker()
@cthulhu_hookimpl
def deactivate(self):
"""Deactivate the self-voice plugin."""
logger.info("Deactivating Self Voice Plugin")
self.deactivateWorker()
super().deactivate()
def activateWorker(self):
"""Start the voice worker thread."""
with self.lock:
self.active = True
# Only start if not already running
if not self.voiceThread.is_alive():
self.voiceThread = Thread(target=self.voiceWorker)
self.voiceThread.daemon = True
self.voiceThread.start()
def deactivateWorker(self):
"""Stop the voice worker thread."""
with self.lock:
self.active = False
# Try to join the thread if it's alive, with a timeout
if self.voiceThread.is_alive():
try:
self.voiceThread.join(timeout=2.0)
except Exception as e:
logger.warning(f"Error stopping voice worker thread: {e}")
def isActive(self):
"""Check if the worker is active."""
with self.lock:
return self.active
def outputMessage(self, message):
"""Output a message through Cthulhu's speech and braille systems.
Args:
message: The message to output. May include special codes.
"""
# Process special codes
append = message.startswith(APPEND_CODE)
if append:
message = message[len(APPEND_CODE):]
persistent = False
if message.endswith(PERSISTENT_CODE):
message = message[:len(message)-len(PERSISTENT_CODE)]
persistent = True
# Output through appropriate channel
if persistent:
# Use the APIHelper for persistent messages
self.app.getAPIHelper().outputMessage(message, not append)
else:
# Use the script manager for standard messages
script_manager = self.app.getDynamicApiManager().getAPI('ScriptManager')
scriptManager = script_manager.getManager()
scriptManager.getDefaultScript().presentMessage(message, resetStyles=False)
def voiceWorker(self):
"""Worker thread that listens on a socket for messages to speak."""
socketFile = '/tmp/cthulhu.sock'
# For testing purposes
# socketFile = '/tmp/cthulhu-plugin.sock'
# Clean up any existing socket file
if os.path.exists(socketFile):
try:
os.unlink(socketFile)
except Exception as e:
logger.error(f"Error removing existing socket file: {e}")
return
try:
# Create and set up the socket
cthulhu_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
cthulhu_sock.bind(socketFile)
os.chmod(socketFile, 0o222) # Write-only for everyone
cthulhu_sock.listen(1)
logger.info(f"Self Voice plugin listening on {socketFile}")
# Main loop - listen for connections
while self.isActive():
# Check if data is available with a timeout
try:
r, _, _ = select.select([cthulhu_sock], [], [], 0.8)
except select.error as e:
logger.error(f"Select error: {e}")
break
if not r: # No data available
continue
# Accept connection
if cthulhu_sock in r:
try:
client_sock, _ = cthulhu_sock.accept()
client_sock.settimeout(0.5) # Set a timeout for receiving data
try:
# Receive and process data
raw_data = client_sock.recv(8192)
if raw_data:
data = raw_data.decode("utf-8").strip()
if data:
self.outputMessage(data)
except socket.timeout:
pass
except Exception as e:
logger.error(f"Error receiving data: {e}")
finally:
client_sock.close()
except Exception as e:
logger.error(f"Error accepting connection: {e}")
except Exception as e:
logger.error(f"Socket error: {e}")
finally:
# Clean up
if 'cthulhu_sock' in locals():
try:
cthulhu_sock.close()
except Exception:
pass
if os.path.exists(socketFile):
try:
os.unlink(socketFile)
except Exception as e:
logger.error(f"Error removing socket file: {e}")
logger.info("Self Voice plugin socket closed")