381 lines
13 KiB
Python
381 lines
13 KiB
Python
#!/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.
|
|
#
|
|
# Forked from Orca screen reader.
|
|
# Cthulhu project: https://git.stormux.org/storm/cthulhu
|
|
|
|
import argparse
|
|
import gi
|
|
gi.require_version("Atspi", "2.0")
|
|
from gi.repository import Atspi
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
def setup_paths():
|
|
"""Configure paths for both installed and source directory execution."""
|
|
currentDir = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
# Check if running from source
|
|
if os.path.exists(os.path.join(currentDir, 'plugins')):
|
|
# Running from source directory
|
|
sys.path.insert(0, os.path.dirname(currentDir))
|
|
pythondir = currentDir
|
|
datadir = currentDir
|
|
else:
|
|
# Running installed - determine if local or system based on actual path
|
|
if currentDir.startswith(os.path.expanduser('~/.local')):
|
|
# Local installation (~/.local/bin/cthulhu)
|
|
prefix = os.path.expanduser('~/.local')
|
|
elif currentDir.startswith('/usr/local'):
|
|
# /usr/local installation
|
|
prefix = '/usr/local'
|
|
else:
|
|
# System installation (/usr/bin/cthulhu)
|
|
prefix = '/usr'
|
|
|
|
# Try to find Python modules in multiple possible locations
|
|
python_version = f'python{sys.version_info.major}.{sys.version_info.minor}'
|
|
possible_pythondirs = [
|
|
os.path.join(prefix, 'lib', python_version, 'site-packages'),
|
|
os.path.join('/usr', 'lib', python_version, 'site-packages'), # System fallback
|
|
os.path.join('/usr/local', 'lib', python_version, 'site-packages') # /usr/local fallback
|
|
]
|
|
|
|
# Use the first directory that contains the cthulhu module
|
|
pythondir = None
|
|
for candidate_dir in possible_pythondirs:
|
|
if os.path.exists(os.path.join(candidate_dir, 'cthulhu', '__init__.py')):
|
|
pythondir = candidate_dir
|
|
break
|
|
|
|
if pythondir is None:
|
|
# Fallback to prefix-relative path if module not found
|
|
pythondir = os.path.join(prefix, 'lib', python_version, 'site-packages')
|
|
|
|
datadir = os.path.join(prefix, 'share', 'cthulhu')
|
|
|
|
sys.path.insert(1, pythondir)
|
|
|
|
# Set environment variables for resource paths
|
|
if 'CTHULHU_DATA_DIR' not in os.environ:
|
|
os.environ['CTHULHU_DATA_DIR'] = datadir
|
|
|
|
# Set up paths before importing Cthulhu modules
|
|
setup_paths()
|
|
|
|
from cthulhu import debug
|
|
from cthulhu import messages
|
|
from cthulhu import settings
|
|
from cthulhu.ax_object import AXObject
|
|
from cthulhu.ax_utilities import AXUtilities
|
|
from cthulhu.cthulhu_platform import version
|
|
|
|
class ListApps(argparse.Action):
|
|
def __call__(self, parser, namespace, values, option_string=None):
|
|
desktop = AXUtilities.get_desktop()
|
|
for app in AXObject.iter_children(desktop):
|
|
pid = AXObject.get_process_id(app)
|
|
try:
|
|
name = Atspi.Accessible.get_name(app) or "(none)"
|
|
except Exception:
|
|
name = "[DEAD]"
|
|
|
|
try:
|
|
cmdline = subprocess.getoutput('cat /proc/%s/cmdline' % pid)
|
|
except Exception:
|
|
cmdline = '(exception encountered)'
|
|
else:
|
|
cmdline = cmdline.replace('\x00', ' ')
|
|
|
|
print(time.strftime('%H:%M:%S', time.localtime()),
|
|
' pid: %5s %-25s %s' % (pid, name, cmdline))
|
|
|
|
parser.exit()
|
|
|
|
class Settings(argparse.Action):
|
|
def __call__(self, parser, namespace, values, option_string=None):
|
|
settingsDict = getattr(namespace, 'settings', {})
|
|
invalid = getattr(namespace, 'invalid', [])
|
|
for value in values.split(','):
|
|
item = str.title(value).replace('-', '')
|
|
try:
|
|
test = 'enable%s' % item
|
|
eval('settings.%s' % test)
|
|
except AttributeError:
|
|
try:
|
|
test = 'show%s' % item
|
|
eval('settings.%s' % test)
|
|
except AttributeError:
|
|
invalid.append(value)
|
|
continue
|
|
settingsDict[test] = self.const
|
|
setattr(namespace, 'settings', settingsDict)
|
|
setattr(namespace, 'invalid', invalid)
|
|
|
|
class HelpFormatter(argparse.HelpFormatter):
|
|
def __init__(self, prog, indent_increment=2, max_help_position=32,
|
|
width=None):
|
|
|
|
super().__init__(prog, indent_increment, max_help_position, width)
|
|
|
|
def add_usage(self, usage, actions, groups, prefix=None):
|
|
super().add_usage(usage, actions, groups, messages.CLI_USAGE)
|
|
|
|
class Parser(argparse.ArgumentParser):
|
|
def __init__(self, *args, **kwargs):
|
|
super(Parser, self).__init__(
|
|
epilog=messages.CLI_EPILOG, formatter_class=HelpFormatter, add_help=False)
|
|
self.add_argument(
|
|
"-h", "--help", action="help", help=messages.CLI_HELP)
|
|
self.add_argument(
|
|
"-v", "--version", action="version", version=version, help=messages.CLI_VERSION)
|
|
self.add_argument(
|
|
"-r", "--replace", action="store_true", help=messages.CLI_REPLACE)
|
|
self.add_argument(
|
|
"-s", "--setup", action="store_true", help=messages.CLI_GUI_SETUP)
|
|
self.add_argument(
|
|
"-l", "--list-apps", action=ListApps, nargs=0,
|
|
help=messages.CLI_LIST_APPS)
|
|
self.add_argument(
|
|
"-e", "--enable", action=Settings, const=True,
|
|
help=messages.CLI_ENABLE_OPTION, metavar=messages.CLI_OPTION)
|
|
self.add_argument(
|
|
"-d", "--disable", action=Settings, const=False,
|
|
help=messages.CLI_DISABLE_OPTION, metavar=messages.CLI_OPTION)
|
|
self.add_argument(
|
|
"-p", "--profile", action="store",
|
|
help=messages.CLI_LOAD_PROFILE, metavar=messages.CLI_PROFILE_NAME)
|
|
self.add_argument(
|
|
"-u", "--user-prefs", action="store",
|
|
help=messages.CLI_LOAD_PREFS, metavar=messages.CLI_PREFS_DIR)
|
|
self.add_argument(
|
|
"--debug-file", action="store",
|
|
help=messages.CLI_DEBUG_FILE, metavar=messages.CLI_DEBUG_FILE_NAME)
|
|
self.add_argument(
|
|
"--debug", action="store_true", help=messages.CLI_ENABLE_DEBUG)
|
|
|
|
self._optionals.title = messages.CLI_OPTIONAL_ARGUMENTS
|
|
|
|
def parse_known_args(self, *args, **kwargs):
|
|
opts, invalid = super(Parser, self).parse_known_args(*args, **kwargs)
|
|
try:
|
|
invalid.extend(opts.invalid)
|
|
except Exception:
|
|
pass
|
|
if invalid:
|
|
print((messages.CLI_INVALID_OPTIONS + " ".join(invalid)))
|
|
|
|
if opts.debug_file:
|
|
opts.debug = True
|
|
elif opts.debug:
|
|
opts.debug_file = time.strftime('debug-%Y-%m-%d-%H:%M:%S.out')
|
|
|
|
return opts, invalid
|
|
|
|
def setProcessName(name):
|
|
"""Attempts to set the process name to the specified name."""
|
|
|
|
sys.argv[0] = name
|
|
|
|
try:
|
|
from setproctitle import setproctitle
|
|
except ImportError:
|
|
pass
|
|
else:
|
|
setproctitle(name)
|
|
return True
|
|
|
|
try:
|
|
from ctypes import cdll, byref, create_string_buffer
|
|
libc = cdll.LoadLibrary('libc.so.6')
|
|
stringBuffer = create_string_buffer(len(name) + 1)
|
|
stringBuffer.value = bytes(name, 'UTF-8')
|
|
libc.prctl(15, byref(stringBuffer), 0, 0, 0)
|
|
return True
|
|
except Exception:
|
|
pass
|
|
|
|
return False
|
|
|
|
def inGraphicalDesktop():
|
|
"""Returns True if we are in a graphical desktop."""
|
|
|
|
# TODO - JD: Make this desktop environment agnostic
|
|
try:
|
|
import gi
|
|
gi.require_version("Gdk", "3.0")
|
|
from gi.repository import Gdk
|
|
display = Gdk.Display.get_default()
|
|
except Exception:
|
|
return False
|
|
|
|
return display is not None
|
|
|
|
def getSessionType():
|
|
sessionType = (os.environ.get("XDG_SESSION_TYPE") or "").strip().lower()
|
|
if sessionType:
|
|
return sessionType
|
|
if os.environ.get("WAYLAND_DISPLAY"):
|
|
return "wayland"
|
|
if os.environ.get("DISPLAY"):
|
|
return "x11"
|
|
return "unknown"
|
|
|
|
def getXServerVendor():
|
|
display = os.environ.get("DISPLAY")
|
|
if not display:
|
|
return None
|
|
|
|
xdpyinfoPath = shutil.which("xdpyinfo")
|
|
if not xdpyinfoPath:
|
|
return None
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
[xdpyinfoPath, "-display", display],
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
timeout=1,
|
|
)
|
|
except Exception:
|
|
return None
|
|
|
|
for line in result.stdout.splitlines():
|
|
if "vendor string:" in line:
|
|
return line.split("vendor string:", 1)[1].strip()
|
|
return None
|
|
|
|
def otherCthulhus():
|
|
"""Returns the pid of any other instances of Cthulhu owned by this user."""
|
|
|
|
openFile = subprocess.Popen('pgrep -u %s -x cthulhu' % os.getuid(),
|
|
shell=True,
|
|
stdout=subprocess.PIPE).stdout
|
|
pids = openFile.read()
|
|
openFile.close()
|
|
cthulhus = [int(p) for p in pids.split()]
|
|
|
|
pid = os.getpid()
|
|
return [p for p in cthulhus if p != pid]
|
|
|
|
def cleanup(sigval):
|
|
"""Tries to clean up any other running Cthulhu instances owned by this user."""
|
|
|
|
cthulhusToKill = otherCthulhus()
|
|
debug.printMessage(debug.LEVEL_INFO, "INFO: Cleaning up these PIDs: %s" % cthulhusToKill)
|
|
|
|
def onTimeout(signum, frame):
|
|
cthulhusToKill = otherCthulhus()
|
|
debug.printMessage(debug.LEVEL_INFO, "INFO: Timeout cleaning up: %s" % cthulhusToKill)
|
|
for pid in cthulhusToKill:
|
|
os.kill(pid, signal.SIGKILL)
|
|
|
|
for pid in cthulhusToKill:
|
|
os.kill(pid, sigval)
|
|
signal.signal(signal.SIGALRM, onTimeout)
|
|
signal.alarm(2)
|
|
while otherCthulhus():
|
|
time.sleep(0.5)
|
|
|
|
def main():
|
|
setProcessName('cthulhu')
|
|
|
|
parser = Parser()
|
|
args, invalid = parser.parse_known_args()
|
|
|
|
if args.debug:
|
|
debug.debugLevel = debug.LEVEL_ALL
|
|
debug.eventDebugLevel = debug.LEVEL_OFF
|
|
debug.debugFile = open(args.debug_file, 'w')
|
|
|
|
if args.replace:
|
|
cleanup(signal.SIGKILL)
|
|
|
|
settingsDict = getattr(args, 'settings', {})
|
|
|
|
if not inGraphicalDesktop():
|
|
print(messages.CLI_NO_DESKTOP_ERROR)
|
|
return 1
|
|
|
|
sessionType = getSessionType()
|
|
sessionDetails = []
|
|
xdgSessionType = os.environ.get("XDG_SESSION_TYPE")
|
|
if xdgSessionType:
|
|
sessionDetails.append(f"XDG_SESSION_TYPE={xdgSessionType}")
|
|
if sessionType == "wayland":
|
|
waylandDisplay = os.environ.get("WAYLAND_DISPLAY")
|
|
if waylandDisplay:
|
|
sessionDetails.append(f"WAYLAND_DISPLAY={waylandDisplay}")
|
|
elif sessionType == "x11":
|
|
display = os.environ.get("DISPLAY")
|
|
if display:
|
|
sessionDetails.append(f"DISPLAY={display}")
|
|
vendor = getXServerVendor()
|
|
if vendor:
|
|
sessionDetails.append(f"X server vendor={vendor}")
|
|
|
|
if sessionDetails:
|
|
msg = f"INFO: Session: {sessionType} ({', '.join(sessionDetails)})"
|
|
else:
|
|
msg = f"INFO: Session: {sessionType}"
|
|
debug.printMessage(debug.LEVEL_INFO, msg, True)
|
|
|
|
debug.printMessage(debug.LEVEL_INFO, "INFO: Preparing to launch.", True)
|
|
|
|
from cthulhu import cthulhu
|
|
manager = cthulhu.getSettingsManager()
|
|
|
|
if not manager:
|
|
print(messages.CLI_SETTINGS_MANAGER_ERROR)
|
|
return 1
|
|
|
|
debug.printMessage(debug.LEVEL_INFO, "INFO: About to activate settings manager.", True)
|
|
manager.activate(args.user_prefs, settingsDict)
|
|
sys.path.insert(0, manager.getPrefsDir())
|
|
|
|
if args.profile:
|
|
try:
|
|
manager.setProfile(args.profile)
|
|
except Exception:
|
|
print(messages.CLI_LOAD_PROFILE_ERROR % args.profile)
|
|
manager.setProfile()
|
|
|
|
if args.setup:
|
|
cleanup(signal.SIGKILL)
|
|
cthulhu.showPreferencesGUI()
|
|
|
|
if otherCthulhus():
|
|
print(messages.CLI_OTHER_CTHULHUS_ERROR)
|
|
return 1
|
|
|
|
debug.printMessage(debug.LEVEL_INFO, "INFO: About to launch Cthulhu.", True)
|
|
return cthulhu.main()
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|