7 Commits
30 changed files with 1321 additions and 190 deletions
+4 -2
View File
@@ -9,7 +9,8 @@ driver=gstreamerDriver
#driver=genericDriver #driver=genericDriver
# Sound themes. These are the pack of sounds used for sound alerts. # Sound themes. These are the pack of sounds used for sound alerts.
# Sound packs may be located at /usr/share/sounds # Sound packs may be located at ~/.local/stormux/fenrir/sounds,
# /usr/share/sounds/fenrir, or /usr/share/sounds/fenrirscreenreader.
theme=default theme=default
# Sound volume controls how loud the sounds for your selected soundpack are. # Sound volume controls how loud the sounds for your selected soundpack are.
@@ -201,7 +202,8 @@ date_format=%%A, %%B %%d, %%Y
auto_spell_check=True auto_spell_check=True
# Language for spell checking (format: language_COUNTRY, e.g., en_US, en_GB, es_ES) # Language for spell checking (format: language_COUNTRY, e.g., en_US, en_GB, es_ES)
spell_check_language=en_US spell_check_language=en_US
# path for your scripts "script_keys" functionality # path for your scripts "script_keys" functionality.
# User-local scripts in ~/.local/stormux/fenrir are loaded first.
script_path=/usr/share/fenrirscreenreader/scripts script_path=/usr/share/fenrirscreenreader/scripts
# Override default commands or add custom commands without modifying Fenrir installation # Override default commands or add custom commands without modifying Fenrir installation
# Leave empty to use default commands only # Leave empty to use default commands only
+1 -1
View File
@@ -41,7 +41,7 @@ def create_argument_parser():
argumentParser.add_argument( argumentParser.add_argument(
'-s', '--setting', '-s', '--setting',
metavar='SETTING-FILE', metavar='SETTING-FILE',
default='/etc/fenrir/settings/settings.conf', default=None,
help='Path to custom settings file' help='Path to custom settings file'
) )
argumentParser.add_argument( argumentParser.add_argument(
@@ -22,7 +22,10 @@ class command:
return _("sends the following keypress to the terminal or application") return _("sends the following keypress to the terminal or application")
def run(self): def run(self):
self.env["input"]["key_forward"] = 3 if self.env["runtime"]["InputManager"].no_key_pressed():
self.env["input"]["key_forward"] = 1
else:
self.env["input"]["key_forward"] = -1
self.env["runtime"]["OutputManager"].present_text( self.env["runtime"]["OutputManager"].present_text(
_("Forward next keypress"), interrupt=True _("Forward next keypress"), interrupt=True
) )
@@ -22,10 +22,7 @@ class command:
return _("Saves your current Fenrir settings so they are the default.") return _("Saves your current Fenrir settings so they are the default.")
def run(self): def run(self):
settings_file = self.env["runtime"][ self.env["runtime"]["SettingsManager"].save_settings()
"SettingsManager"
].get_settings_file()
self.env["runtime"]["SettingsManager"].save_settings(settings_file)
self.env["runtime"]["OutputManager"].present_text( self.env["runtime"]["OutputManager"].present_text(
_("Settings saved."), interrupt=True _("Settings saved."), interrupt=True
) )
@@ -6,7 +6,6 @@
import _thread import _thread
import os import os
import subprocess
from subprocess import PIPE from subprocess import PIPE
from subprocess import Popen from subprocess import Popen
@@ -53,10 +52,11 @@ class command:
def _thread_run(self): def _thread_run(self):
try: try:
callstring = ( p = Popen(
self.script_path + " " + self.env["general"]["curr_user"] [self.script_path, self.env["general"]["curr_user"]],
stdout=PIPE,
stderr=PIPE,
) )
p = Popen(callstring, stdout=PIPE, stderr=PIPE, shell=True)
stdout, stderr = p.communicate() stdout, stderr = p.communicate()
stdout = stdout.decode("utf-8") stdout = stdout.decode("utf-8")
stderr = stderr.decode("utf-8") stderr = stderr.decode("utf-8")
@@ -48,7 +48,7 @@ class command:
for curr_key in self.env["input"]["curr_input"]: for curr_key in self.env["input"]["curr_input"]:
if curr_key not in filter_list: if curr_key not in filter_list:
return return
self.env["runtime"]["OutputManager"].interrupt_output() self.env["runtime"]["OutputManager"].interrupt_output_async()
def set_callback(self, callback): def set_callback(self, callback):
pass pass
@@ -28,15 +28,21 @@ class command:
# Only announce numlock changes if an actual numlock key was pressed # Only announce numlock changes if an actual numlock key was pressed
# AND the LED state actually changed (some numpads send spurious NUMLOCK events) # AND the LED state actually changed (some numpads send spurious NUMLOCK events)
current_input = self.env["input"]["curr_input"] current_input = self.env["input"]["curr_input"]
previous_input = self.env["input"]["prev_input"]
relevant_input = current_input or previous_input
# Check if this is a genuine numlock key press by verifying: # Check if this is a genuine numlock key press by verifying:
# 1. KEY_NUMLOCK is in the current input sequence # 1. KEY_NUMLOCK is in the current input sequence
# 2. The LED state has actually changed # 2. The LED state has actually changed
# 3. This isn't just a side effect from a KP_ key (which some buggy numpads do) # 3. This isn't just a side effect from a KP_ key (which some buggy numpads do)
is_genuine_numlock = ( is_genuine_numlock = (
current_input and relevant_input and
"KEY_NUMLOCK" in current_input and "KEY_NUMLOCK" in relevant_input and
not any(key.startswith("KEY_KP") for key in current_input if isinstance(key, str)) not any(
key.startswith("KEY_KP")
for key in relevant_input
if isinstance(key, str)
)
) )
if is_genuine_numlock: if is_genuine_numlock:
@@ -22,23 +22,24 @@ class command(config_command):
def run(self): def run(self):
current_theme = self.get_setting("sound", "theme", "default") current_theme = self.get_setting("sound", "theme", "default")
current_theme_name = os.path.basename(
os.path.normpath(current_theme)
)
# Present current theme # Present current theme
self.present_text(f"Current sound theme: {current_theme}") self.present_text(f"Current sound theme: {current_theme}")
# Look for available sound themes # Look for available sound themes
sound_paths = [ sound_paths = self.env[
"/usr/share/sounds", "runtime"
"/usr/share/fenrirscreenreader/sounds", ]["SettingsManager"].get_sound_theme_roots()
os.path.expanduser("~/.local/share/fenrirscreenreader/sounds"),
]
available_themes = self.get_available_themes(sound_paths) available_themes = self.get_available_themes(sound_paths)
if len(available_themes) > 1: if len(available_themes) > 1:
# For this implementation, cycle through available themes # For this implementation, cycle through available themes
try: try:
current_index = available_themes.index(current_theme) current_index = available_themes.index(current_theme_name)
next_index = (current_index + 1) % len(available_themes) next_index = (current_index + 1) % len(available_themes)
new_theme = available_themes[next_index] new_theme = available_themes[next_index]
except ValueError: except ValueError:
+115 -55
View File
@@ -174,90 +174,121 @@ class CommandManager:
) )
continue continue
def get_script_paths(self, script_path=""):
if script_path:
candidate_paths = [script_path]
else:
settings_manager = self.env["runtime"]["SettingsManager"]
candidate_paths = [
settings_manager.get_user_script_path(),
settings_manager.get_setting("general", "script_path"),
os.path.join(fenrir_path, "../../config/scripts/"),
]
script_paths = []
seen_paths = set()
for path in candidate_paths:
if not path:
continue
normalized_path = os.path.abspath(os.path.expanduser(path))
if normalized_path in seen_paths:
continue
script_paths.append(normalized_path)
seen_paths.add(normalized_path)
return script_paths
def load_script_commands(self, section="commands", script_path=""): def load_script_commands(self, section="commands", script_path=""):
if script_path == "": script_paths = self.get_script_paths(script_path)
script_path = self.env["runtime"]["SettingsManager"].get_setting( loaded_any_path = False
"general", "script_path" loaded_script_names = set()
) for path in script_paths:
if not script_path.endswith("/"): loaded_any_path = (
script_path += "/" self.load_script_commands_from_path(
if not os.path.exists(script_path): section, path, loaded_script_names
if os.path.exists(fenrir_path + "/../../config/scripts/"):
script_path = fenrir_path + "/../../config/scripts/"
else:
self.env["runtime"]["DebugManager"].write_debug_out(
"scriptpath not exists:" + script_path,
debug.DebugLevel.WARNING,
) )
return or loaded_any_path
)
if not loaded_any_path:
self.env["runtime"]["DebugManager"].write_debug_out(
"No script paths available:" + str(script_paths),
debug.DebugLevel.WARNING,
)
def load_script_commands_from_path(
self, section, script_path, loaded_script_names=None
):
if loaded_script_names is None:
loaded_script_names = set()
if not os.path.exists(script_path):
return False
if not os.path.isdir(script_path): if not os.path.isdir(script_path):
self.env["runtime"]["DebugManager"].write_debug_out( self.env["runtime"]["DebugManager"].write_debug_out(
"scriptpath not a directory:" + script_path, "scriptpath not a directory:" + script_path,
debug.DebugLevel.ERROR, debug.DebugLevel.ERROR,
) )
return return False
if not os.access(script_path, os.R_OK): if not os.access(script_path, os.R_OK):
self.env["runtime"]["DebugManager"].write_debug_out( self.env["runtime"]["DebugManager"].write_debug_out(
"scriptpath not readable:" + script_path, "scriptpath not readable:" + script_path,
debug.DebugLevel.ERROR, debug.DebugLevel.ERROR,
) )
return return False
command_list = glob.glob(script_path + "*") command_list = sorted(glob.glob(os.path.join(script_path, "*")))
sub_command = fenrir_path + "/commands/commands/subprocess.py" sub_command = fenrir_path + "/commands/commands/subprocess.py"
for command in command_list: for command in command_list:
invalid = False
try: try:
if not os.path.isfile(command):
continue
file_name, file_extension = os.path.splitext(command) file_name, file_extension = os.path.splitext(command)
file_name = file_name.split("/")[-1] file_name = file_name.split("/")[-1]
if file_name.startswith("__"): if file_name.startswith("__"):
continue continue
if file_name.upper() in self.env["commands"][section]: command_name = file_name.upper()
script_name = self.get_script_name(file_name)
if script_name in loaded_script_names:
self.env["runtime"]["DebugManager"].write_debug_out(
"Skip script with duplicate script name:"
+ command_name,
debug.DebugLevel.INFO,
)
continue
if command_name in self.env["commands"][section]:
self.env["runtime"]["DebugManager"].write_debug_out(
"Skip script with duplicate command name:"
+ command_name,
debug.DebugLevel.INFO,
)
continue
shortcut = self.get_script_shortcut(file_name)
if not shortcut:
continue
shortcut_key = str(shortcut)
if shortcut_key in self.env["bindings"]:
self.env["runtime"]["DebugManager"].write_debug_out(
"Skip script with duplicate shortcut:"
+ command_name
+ " "
+ shortcut_key,
debug.DebugLevel.INFO,
)
continue continue
command_mod = module_utils.import_module( command_mod = module_utils.import_module(
file_name, sub_command file_name, sub_command
) )
self.env["commands"][section][ self.env["commands"][section][command_name] = (
file_name.upper() command_mod.command()
] = command_mod.command() )
self.env["commands"][section][file_name.upper()].initialize( self.env["commands"][section][command_name].initialize(
self.env, command self.env, command
) )
self.env["bindings"][shortcut_key] = command_name
self.env["rawBindings"][shortcut_key] = shortcut
loaded_script_names.add(script_name)
self.env["runtime"]["DebugManager"].write_debug_out( self.env["runtime"]["DebugManager"].write_debug_out(
"Load script:" + section + "." + file_name.upper(), "Load script:" + section + "." + command_name,
debug.DebugLevel.INFO, debug.DebugLevel.INFO,
on_any_level=True, on_any_level=True,
) )
comm_settings = file_name.upper().split("__-__")
if len(comm_settings) == 1:
keys = comm_settings[0]
elif len(comm_settings) == 2:
keys = comm_settings[1]
elif len(comm_settings) > 2:
continue
keys = keys.split("__+__")
shortcut_keys = []
shortcut = []
for key in keys:
if not self.env["runtime"]["InputManager"].is_valid_key(
key.upper()
):
self.env["runtime"]["DebugManager"].write_debug_out(
"invalid key : "
+ key.upper()
+ " script:"
+ file_name,
debug.DebugLevel.WARNING,
)
invalid = True
break
shortcut_keys.append(key.upper())
if invalid:
continue
if "KEY_SCRIPT" not in shortcut_keys:
shortcut_keys.append("KEY_SCRIPT")
shortcut.append(1)
shortcut.append(sorted(shortcut_keys))
self.env["bindings"][str(shortcut)] = file_name.upper()
except Exception as e: except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out( self.env["runtime"]["DebugManager"].write_debug_out(
"Loading script:" + file_name, debug.DebugLevel.ERROR "Loading script:" + file_name, debug.DebugLevel.ERROR
@@ -266,6 +297,35 @@ class CommandManager:
str(e), debug.DebugLevel.ERROR str(e), debug.DebugLevel.ERROR
) )
continue continue
return True
def get_script_name(self, file_name):
return file_name.upper().split("__-__", 1)[0]
def get_script_shortcut(self, file_name):
comm_settings = file_name.upper().split("__-__")
if len(comm_settings) == 1:
keys = comm_settings[0]
elif len(comm_settings) == 2:
keys = comm_settings[1]
else:
return None
shortcut_keys = []
for key in keys.split("__+__"):
key = key.upper()
if not key:
return None
if not self.env["runtime"]["InputManager"].is_valid_key(key):
self.env["runtime"]["DebugManager"].write_debug_out(
"invalid key : " + key + " script:" + file_name,
debug.DebugLevel.WARNING,
)
return None
shortcut_keys.append(key)
if "KEY_SCRIPT" not in shortcut_keys:
shortcut_keys.append("KEY_SCRIPT")
return [1, sorted(shortcut_keys)]
def shutdown_commands(self, section): def shutdown_commands(self, section):
# Check if the section exists in the commands dictionary # Check if the section exists in the commands dictionary
@@ -48,8 +48,7 @@ class DynamicKeyboardLayoutCommand:
) )
# Save to the actual config file # Save to the actual config file
configFilePath = settingsManager.get_settings_file() settingsManager.save_settings()
settingsManager.save_settings(configFilePath)
self.env["runtime"]["OutputManager"].present_text( self.env["runtime"]["OutputManager"].present_text(
f"Keyboard layout set to {self.layoutName}. Please restart Fenrir for this change to take effect." f"Keyboard layout set to {self.layoutName}. Please restart Fenrir for this change to take effect."
+10 -3
View File
@@ -126,8 +126,7 @@ class FenrirManager:
self.environment["runtime"]["InputManager"].write_event_buffer() self.environment["runtime"]["InputManager"].write_event_buffer()
self.environment["runtime"]["InputManager"].handle_device_grab() self.environment["runtime"]["InputManager"].handle_device_grab()
if self.environment["input"]["key_forward"] > 0: self.update_key_forward()
self.environment["input"]["key_forward"] -= 1
self.environment["runtime"]["CommandManager"].execute_default_trigger( self.environment["runtime"]["CommandManager"].execute_default_trigger(
"onKeyInput" "onKeyInput"
@@ -260,7 +259,7 @@ class FenrirManager:
) )
def detect_shortcut_command(self): def detect_shortcut_command(self):
if self.environment["input"]["key_forward"] > 0: if self.environment["input"]["key_forward"] != 0:
return return
if len(self.environment["input"]["prev_input"]) > len( if len(self.environment["input"]["prev_input"]) > len(
@@ -324,6 +323,14 @@ class FenrirManager:
) )
self.command = "" self.command = ""
def update_key_forward(self):
key_forward = self.environment["input"]["key_forward"]
input_manager = self.environment["runtime"]["InputManager"]
if key_forward == -1 and input_manager.no_key_pressed():
self.environment["input"]["key_forward"] = 1
elif key_forward == 1 and input_manager.no_key_pressed():
self.environment["input"]["key_forward"] = 0
def set_process_name(self, name="fenrir"): def set_process_name(self, name="fenrir"):
"""Attempts to set the process name to 'fenrir'.""" """Attempts to set the process name to 'fenrir'."""
try: try:
@@ -163,6 +163,14 @@ class InputManager:
def get_last_event(self): def get_last_event(self):
return self.lastEvent return self.lastEvent
def record_unmanaged_keypress(self, event_name):
self.lastEvent = {
"event_name": event_name,
"event_state": 1,
}
self.set_last_deepest_input([event_name])
self.lastInputTime = time.time()
def handle_input_event(self, event_data): def handle_input_event(self, event_data):
if not event_data: if not event_data:
return return
+51 -2
View File
@@ -6,6 +6,7 @@
import re import re
import string import string
import threading
import time import time
from fenrirscreenreader.core import debug from fenrirscreenreader.core import debug
@@ -16,6 +17,11 @@ from fenrirscreenreader.utils import line_utils
class OutputManager: class OutputManager:
def __init__(self): def __init__(self):
self.last_echo = "" self.last_echo = ""
self.interrupt_lock = threading.Lock()
self.interrupt_running = False
self.interrupt_thread = None
self.interrupt_done = None
self.interrupt_wait_timeout = 0.1
def initialize(self, environment): def initialize(self, environment):
self.env = environment self.env = environment
@@ -45,7 +51,7 @@ class OutputManager:
announce_capital=False, announce_capital=False,
flush=True, flush=True,
): ):
if text == "": if text == "" and sound_icon == "":
return return
if ( if (
self.env["runtime"]["SettingsManager"].get_setting_as_bool( self.env["runtime"]["SettingsManager"].get_setting_as_bool(
@@ -63,6 +69,8 @@ class OutputManager:
"sound_icon found", debug.DebugLevel.INFO "sound_icon found", debug.DebugLevel.INFO
) )
return return
if text == "":
return
if (len(text) > 1) and (text.strip(string.whitespace) == ""): if (len(text) > 1) and (text.strip(string.whitespace) == ""):
return return
is_capital = self._should_announce_capital(text, announce_capital) is_capital = self._should_announce_capital(text, announce_capital)
@@ -278,7 +286,40 @@ class OutputManager:
str(e), debug.DebugLevel.ERROR str(e), debug.DebugLevel.ERROR
) )
def interrupt_output(self): def interrupt_output(self, wait=True):
interrupt_done, started = self.start_interrupt_output()
if wait and started and interrupt_done:
interrupt_done.wait(timeout=self.interrupt_wait_timeout)
def interrupt_output_async(self):
self.start_interrupt_output()
def start_interrupt_output(self):
with self.interrupt_lock:
if self.interrupt_running:
return self.interrupt_done, False
self.interrupt_running = True
self.interrupt_done = threading.Event()
self.interrupt_thread = threading.Thread(
target=self.run_interrupt_output,
args=(self.interrupt_done,),
daemon=True,
)
interrupt_thread = self.interrupt_thread
interrupt_done = self.interrupt_done
interrupt_thread.start()
return interrupt_done, True
def run_interrupt_output(self, interrupt_done):
try:
self.cancel_speech()
finally:
interrupt_done.set()
with self.interrupt_lock:
if self.interrupt_done is interrupt_done:
self.interrupt_running = False
def cancel_speech(self):
try: try:
self.env["runtime"]["SpeechDriver"].cancel() self.env["runtime"]["SpeechDriver"].cancel()
self.env["runtime"]["DebugManager"].write_debug_out( self.env["runtime"]["DebugManager"].write_debug_out(
@@ -349,6 +390,14 @@ class OutputManager:
return False return False
def play_sound(self, sound_icon="", interrupt=True):
aliases = {
"ERROR": "ERRORSCREEN",
"ERRORSOUND": "ERRORSCREEN",
}
normalized_icon = aliases.get(str(sound_icon).upper(), sound_icon)
return self.play_sound_icon(normalized_icon, interrupt)
def play_frequence(self, frequence, duration, interrupt=True): def play_frequence(self, frequence, duration, interrupt=True):
if not self.env["runtime"]["SettingsManager"].get_setting_as_bool( if not self.env["runtime"]["SettingsManager"].get_setting_as_bool(
"sound", "enabled" "sound", "enabled"
@@ -464,12 +464,6 @@ class RemoteManager:
) )
def save_settings(self, setting_config_path=None): def save_settings(self, setting_config_path=None):
if not setting_config_path:
setting_config_path = self.env["runtime"][
"SettingsManager"
].get_settings_file()
if setting_config_path == "":
return
self.env["runtime"]["SettingsManager"].save_settings( self.env["runtime"]["SettingsManager"].save_settings(
setting_config_path setting_config_path
) )
+159 -80
View File
@@ -42,11 +42,23 @@ fenrir_path = os.path.dirname(currentdir)
class SettingsManager: class SettingsManager:
system_settings_root = "/etc/fenrirscreenreader/"
system_settings_file = "/etc/fenrirscreenreader/settings/settings.conf"
user_settings_file = (
"~/.local/share/stormux/fenrirscreenreader/settings/settings.conf"
)
user_resource_root = "~/.local/stormux/fenrir/"
system_sound_roots = [
"/usr/share/sounds/fenrir/",
"/usr/share/sounds/fenrirscreenreader/",
]
def __init__(self): def __init__(self):
self.settings = settings_data self.settings = settings_data
self.settingArgDict = {} self.settingArgDict = {}
self.bindingsBackup = None self.bindingsBackup = None
self.settings_file = "" self.settings_file = ""
self.save_settings_path = ""
def initialize(self, environment): def initialize(self, environment):
self.env = environment self.env = environment
@@ -104,12 +116,116 @@ class SettingsManager:
return self.settings_file return self.settings_file
def set_settings_file(self, settings_file): def set_settings_file(self, settings_file):
if not os.path.exists(settings_file): if os.path.exists(settings_file) and not os.access(
return settings_file, os.R_OK
if not os.access(settings_file, os.R_OK): ):
return return
self.settings_file = settings_file self.settings_file = settings_file
def get_user_settings_file(self):
return os.path.expanduser(self.user_settings_file)
def get_system_settings_file(self):
return self.system_settings_file
def is_user_settings_mode(self):
return os.geteuid() != 0
def get_default_save_settings_file(self):
if self.is_user_settings_mode():
return self.get_user_settings_file()
return self.get_system_settings_file()
def get_bundled_settings_root(self):
return os.path.abspath(os.path.join(fenrir_path, "../../config/")) + "/"
def get_user_resource_root(self):
return os.path.expanduser(self.user_resource_root)
def get_user_script_path(self):
return self.get_user_resource_root()
def get_user_sound_root(self):
return os.path.join(self.get_user_resource_root(), "sounds/")
def get_resource_settings_root(self):
if os.path.exists(self.system_settings_root):
return self.system_settings_root
bundled_settings_root = self.get_bundled_settings_root()
if os.path.exists(bundled_settings_root):
return bundled_settings_root
return ""
def get_sound_theme_roots(self):
candidate_roots = [
self.get_user_sound_root(),
*self.system_sound_roots,
os.path.join(self.get_bundled_settings_root(), "sound/"),
]
sound_roots = []
seen_roots = set()
for root in candidate_roots:
normalized_root = os.path.abspath(os.path.expanduser(root))
if normalized_root in seen_roots:
continue
sound_roots.append(normalized_root + "/")
seen_roots.add(normalized_root)
return sound_roots
def resolve_sound_theme_path(self, theme):
if not theme:
return ""
theme = os.path.expanduser(theme)
if os.path.exists(os.path.join(theme, "soundicons.conf")):
return theme
for sound_root in self.get_sound_theme_roots():
theme_path = os.path.join(sound_root, theme)
if os.path.exists(os.path.join(theme_path, "soundicons.conf")):
return theme_path
return ""
def load_keyboard_layout(self, environment):
settings_root = self.get_resource_settings_root()
keyboard_layout = self.get_setting("keyboard", "keyboard_layout")
if os.path.exists(keyboard_layout):
environment["runtime"]["InputManager"].load_shortcuts(
keyboard_layout
)
return
layout_path = os.path.join(settings_root, "keyboard", keyboard_layout)
if os.path.exists(layout_path):
self.set_setting("keyboard", "keyboard_layout", layout_path)
environment["runtime"]["InputManager"].load_shortcuts(layout_path)
return
layout_path = os.path.join(
settings_root, "keyboard", keyboard_layout + ".conf"
)
if os.path.exists(layout_path):
self.set_setting("keyboard", "keyboard_layout", layout_path)
environment["runtime"]["InputManager"].load_shortcuts(layout_path)
def resolve_settings_file(self, requested_settings_file=None):
if requested_settings_file and os.path.exists(requested_settings_file):
return requested_settings_file
if self.is_user_settings_mode() and os.path.exists(
self.get_user_settings_file()
):
return self.get_user_settings_file()
if os.path.exists(self.get_system_settings_file()):
return self.get_system_settings_file()
bundled_settings_file = os.path.join(
self.get_bundled_settings_root(), "settings/settings.conf"
)
if os.path.exists(bundled_settings_file):
return bundled_settings_file
return ""
def load_settings(self, setting_config_path): def load_settings(self, setting_config_path):
if not os.path.exists(setting_config_path): if not os.path.exists(setting_config_path):
return False return False
@@ -128,7 +244,29 @@ class SettingsManager:
self.set_settings_file(setting_config_path) self.set_settings_file(setting_config_path)
return True return True
def save_settings(self, setting_config_path): def save_settings(self, setting_config_path=None):
if setting_config_path is None:
if self.save_settings_path:
setting_config_path = self.save_settings_path
else:
setting_config_path = self.get_default_save_settings_file()
# Ensure directory exists for user-local settings
if setting_config_path and setting_config_path.startswith(
os.path.expanduser("~/.local/")
):
config_dir = os.path.dirname(setting_config_path)
if not os.path.exists(config_dir):
try:
os.makedirs(config_dir, mode=0o755, exist_ok=True)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"save_settings: failed to create directory "
+ config_dir
+ ": "
+ str(e),
debug.DebugLevel.ERROR,
)
# set opt dict here # set opt dict here
# save file # save file
try: try:
@@ -226,7 +364,7 @@ class SettingsManager:
try: try:
if self.env["runtime"][driverType] is not None: if self.env["runtime"][driverType] is not None:
self.env["runtime"][driverType].shutdown(self.env) self.env["runtime"][driverType].shutdown()
except Exception as e: except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out( self.env["runtime"]["DebugManager"].write_debug_out(
"settings_manager load_driver: Error shutting down driver: " "settings_manager load_driver: Error shutting down driver: "
@@ -458,25 +596,16 @@ class SettingsManager:
def init_fenrir_config( def init_fenrir_config(
self, cliArgs, fenrir_manager=None, environment=environment.environment self, cliArgs, fenrir_manager=None, environment=environment.environment
): ):
settings_root = "/etc/fenrirscreenreader/" self.cliArgs = cliArgs
settings_file = cliArgs.setting
sound_root = "/usr/share/sounds/fenrirscreenreader/" settings_root = self.get_resource_settings_root()
# get fenrir settings root if not settings_root:
if not os.path.exists(settings_root): return None
if os.path.exists(fenrir_path + "/../../config/"):
settings_root = fenrir_path + "/../../config/" self.save_settings_path = self.get_default_save_settings_file()
else: settings_file = self.resolve_settings_file(cliArgs.setting)
return None if not settings_file:
# get settings file return None
if settings_file is None or not os.path.exists(settings_file):
if os.path.exists(settings_root + "/settings/settings.conf"):
settings_file = settings_root + "/settings/settings.conf"
else:
return None
# get sound themes root
if not os.path.exists(sound_root):
if os.path.exists(fenrir_path + "/../../config/sound/"):
sound_root = fenrir_path + "/../../config/sound/"
environment["runtime"]["SettingsManager"] = self environment["runtime"]["SettingsManager"] = self
environment["runtime"]["SettingsManager"].initialize(environment) environment["runtime"]["SettingsManager"].initialize(environment)
@@ -526,22 +655,11 @@ class SettingsManager:
"screen", "ignore_screen", ",".join(ignore_screens) "screen", "ignore_screen", ",".join(ignore_screens)
) )
if not os.path.exists( sound_theme_path = self.resolve_sound_theme_path(
self.get_setting("sound", "theme") + "/soundicons.conf" self.get_setting("sound", "theme")
): )
if os.path.exists(sound_root + self.get_setting("sound", "theme")): if sound_theme_path:
self.set_setting( self.set_setting("sound", "theme", sound_theme_path)
"sound",
"theme",
sound_root + self.get_setting("sound", "theme"),
)
if os.path.exists(
self.get_setting("sound", "theme") + "/soundicons.conf"
):
environment["runtime"]["SettingsManager"].load_sound_icons(
self.get_setting("sound", "theme"), environment
)
else:
environment["runtime"]["SettingsManager"].load_sound_icons( environment["runtime"]["SettingsManager"].load_sound_icons(
self.get_setting("sound", "theme"), environment self.get_setting("sound", "theme"), environment
) )
@@ -618,6 +736,7 @@ class SettingsManager:
environment["runtime"]["InputManager"] = inputManager.InputManager() environment["runtime"]["InputManager"] = inputManager.InputManager()
environment["runtime"]["InputManager"].initialize(environment) environment["runtime"]["InputManager"].initialize(environment)
self.load_keyboard_layout(environment)
environment["runtime"]["ScreenManager"] = screenManager.ScreenManager() environment["runtime"]["ScreenManager"] = screenManager.ScreenManager()
environment["runtime"]["ScreenManager"].initialize(environment) environment["runtime"]["ScreenManager"].initialize(environment)
@@ -638,46 +757,6 @@ class SettingsManager:
] = diffReviewManager.DiffReviewManager() ] = diffReviewManager.DiffReviewManager()
environment["runtime"]["DiffReviewManager"].initialize(environment) environment["runtime"]["DiffReviewManager"].initialize(environment)
if not os.path.exists(
self.get_setting("keyboard", "keyboard_layout")
):
if os.path.exists(
settings_root
+ "keyboard/"
+ self.get_setting("keyboard", "keyboard_layout")
):
self.set_setting(
"keyboard",
"keyboard_layout",
settings_root
+ "keyboard/"
+ self.get_setting("keyboard", "keyboard_layout"),
)
environment["runtime"]["InputManager"].load_shortcuts(
self.get_setting("keyboard", "keyboard_layout")
)
if os.path.exists(
settings_root
+ "keyboard/"
+ self.get_setting("keyboard", "keyboard_layout")
+ ".conf"
):
self.set_setting(
"keyboard",
"keyboard_layout",
settings_root
+ "keyboard/"
+ self.get_setting("keyboard", "keyboard_layout")
+ ".conf",
)
environment["runtime"]["InputManager"].load_shortcuts(
self.get_setting("keyboard", "keyboard_layout")
)
else:
environment["runtime"]["InputManager"].load_shortcuts(
self.get_setting("keyboard", "keyboard_layout")
)
environment["runtime"]["CursorManager"] = cursorManager.CursorManager() environment["runtime"]["CursorManager"] = cursorManager.CursorManager()
environment["runtime"]["CursorManager"].initialize(environment) environment["runtime"]["CursorManager"].initialize(environment)
environment["runtime"][ environment["runtime"][
+1 -1
View File
@@ -20,7 +20,7 @@ class sound_driver:
if not self._initialized: if not self._initialized:
return return
self.cancel() self.cancel()
self._is_initialized = False self._initialized = False
def play_frequence( def play_frequence(
self, frequence, duration, adjust_volume=0.0, interrupt=True self, frequence, duration, adjust_volume=0.0, interrupt=True
+1 -1
View File
@@ -4,5 +4,5 @@
# Fenrir TTY screen reader # Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors. # By Chrys, Storm Dragon, and contributors.
version = "2026.05.10" version = "2026.05.12"
code_name = "testing" code_name = "testing"
+53 -11
View File
@@ -437,8 +437,8 @@ class driver(inputDriver):
self.refresh_interesting_keys() self.refresh_interesting_keys()
passive_grabs = self.build_passive_grabs() passive_grabs = self.build_passive_grabs()
failed_before = self.failed_grabs failed_before = self.failed_grabs
for key_name, modifier_mask in passive_grabs: for key_name, modifier_mask, include_num_lock in passive_grabs:
self.grab_key_name(key_name, modifier_mask) self.grab_key_name(key_name, modifier_mask, include_num_lock)
self.display.flush() self.display.flush()
self.grab_signature = signature self.grab_signature = signature
self.write_debug( self.write_debug(
@@ -454,9 +454,10 @@ class driver(inputDriver):
def build_passive_grabs(self): def build_passive_grabs(self):
grabs = set() grabs = set()
for fenrir_key in self.env["input"]["fenrir_key"]: for fenrir_key in self.env["input"]["fenrir_key"]:
grabs.add((fenrir_key, 0)) grabs.add((fenrir_key, 0, True))
for script_key in self.env["input"]["script_key"]: for script_key in self.env["input"]["script_key"]:
grabs.add((script_key, 0)) grabs.add((script_key, 0, True))
grabs.add(("KEY_NUMLOCK", 0, True))
for shortcut in self.env.get("rawBindings", {}).values(): for shortcut in self.env.get("rawBindings", {}).values():
keys = shortcut[1] keys = shortcut[1]
expanded_keys = self.expand_special_keys(keys) expanded_keys = self.expand_special_keys(keys)
@@ -470,12 +471,40 @@ class driver(inputDriver):
final_key = non_modifier_keys[-1] final_key = non_modifier_keys[-1]
if "KEY_FENRIR" in keys: if "KEY_FENRIR" in keys:
for fenrir_key in self.env["input"]["fenrir_key"]: for fenrir_key in self.env["input"]["fenrir_key"]:
grabs.add((fenrir_key, modifier_mask)) grabs.add((fenrir_key, modifier_mask, True))
fenrir_modifier_mask = self.modifier_masks.get(
fenrir_key, 0
)
if fenrir_modifier_mask:
grabs.add(
(
final_key,
modifier_mask | fenrir_modifier_mask,
not final_key.startswith("KEY_KP"),
)
)
elif "KEY_SCRIPT" in keys: elif "KEY_SCRIPT" in keys:
for script_key in self.env["input"]["script_key"]: for script_key in self.env["input"]["script_key"]:
grabs.add((script_key, modifier_mask)) grabs.add((script_key, modifier_mask, True))
script_modifier_mask = self.modifier_masks.get(
script_key, 0
)
if script_modifier_mask:
grabs.add(
(
final_key,
modifier_mask | script_modifier_mask,
not final_key.startswith("KEY_KP"),
)
)
else: else:
grabs.add((final_key, modifier_mask)) grabs.add(
(
final_key,
modifier_mask,
not final_key.startswith("KEY_KP"),
)
)
return grabs return grabs
def expand_special_keys(self, keys): def expand_special_keys(self, keys):
@@ -495,7 +524,9 @@ class driver(inputDriver):
modifier_mask |= self.modifier_masks.get(key_name, 0) modifier_mask |= self.modifier_masks.get(key_name, 0)
return modifier_mask return modifier_mask
def grab_key_name(self, key_name, modifier_mask=0): def grab_key_name(
self, key_name, modifier_mask=0, include_num_lock=True
):
keysym_names = self.key_name_to_keysym_names(key_name) keysym_names = self.key_name_to_keysym_names(key_name)
for keysym_name in keysym_names: for keysym_name in keysym_names:
keysym = XK.string_to_keysym(keysym_name) keysym = XK.string_to_keysym(keysym_name)
@@ -504,7 +535,9 @@ class driver(inputDriver):
keycode = self.display.keysym_to_keycode(keysym) keycode = self.display.keysym_to_keycode(keysym)
if not keycode: if not keycode:
continue continue
for effective_mask in self.optional_modifier_masks(modifier_mask): for effective_mask in self.optional_modifier_masks(
modifier_mask, include_num_lock
):
try: try:
self.window.grab_key( self.window.grab_key(
keycode, keycode,
@@ -524,9 +557,9 @@ class driver(inputDriver):
debug.DebugLevel.WARNING, debug.DebugLevel.WARNING,
) )
def optional_modifier_masks(self, modifier_mask): def optional_modifier_masks(self, modifier_mask, include_num_lock=True):
optional_masks = [0, X.LockMask] optional_masks = [0, X.LockMask]
if self.num_lock_mask: if include_num_lock and self.num_lock_mask:
optional_masks += [self.num_lock_mask, self.num_lock_mask | X.LockMask] optional_masks += [self.num_lock_mask, self.num_lock_mask | X.LockMask]
return {modifier_mask | optional for optional in optional_masks} return {modifier_mask | optional for optional in optional_masks}
@@ -610,6 +643,15 @@ class driver(inputDriver):
self.ungrab_all_devices() self.ungrab_all_devices()
def get_led_state(self, led=0): def get_led_state(self, led=0):
try:
pointer = self.root.query_pointer()
mask = getattr(pointer, "mask", 0)
if led == 0:
return bool(self.num_lock_mask and mask & self.num_lock_mask)
if led == 1:
return bool(mask & X.LockMask)
except Exception:
pass
return False return False
def set_led_state(self, led_dict): def set_led_state(self, led_dict):
@@ -9,6 +9,7 @@ import getpass
import os import os
import pty import pty
import shlex import shlex
from queue import Full
import signal import signal
import struct import struct
import sys import sys
@@ -201,6 +202,9 @@ class driver(screenDriver):
self.terminal = None self.terminal = None
self.p_pid = -1 self.p_pid = -1
self.terminal_lock = threading.Lock() # Synchronize terminal operations self.terminal_lock = threading.Lock() # Synchronize terminal operations
self.stdin_interrupt_lock = threading.Lock()
self.stdin_interrupt_running = False
self.stdin_interrupt_thread = None
signal.signal(signal.SIGWINCH, self.handle_sigwinch) signal.signal(signal.SIGWINCH, self.handle_sigwinch)
# Runtime configuration storage # Runtime configuration storage
@@ -288,6 +292,94 @@ class driver(screenDriver):
msg_bytes = bytes(msg_bytes, "UTF-8") msg_bytes = bytes(msg_bytes, "UTF-8")
os.write(screen, msg_bytes) os.write(screen, msg_bytes)
def interrupt_output_on_stdin_input(self, msg_bytes):
if not msg_bytes:
return
settings_manager = self.env["runtime"]["SettingsManager"]
if not settings_manager.get_setting_as_bool(
"keyboard", "interrupt_on_key_press"
):
return
if settings_manager.get_setting(
"keyboard", "interrupt_on_key_press_filter"
).strip():
return
self.start_stdin_interrupt_thread()
def start_stdin_interrupt_thread(self):
with self.stdin_interrupt_lock:
if self.stdin_interrupt_running:
return
self.stdin_interrupt_running = True
self.stdin_interrupt_thread = threading.Thread(
target=self.run_stdin_interrupt,
daemon=True,
)
self.stdin_interrupt_thread.start()
def run_stdin_interrupt(self):
try:
self.env["runtime"]["OutputManager"].interrupt_output()
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver interrupt_output_on_stdin_input: "
+ str(e),
debug.DebugLevel.ERROR,
)
finally:
with self.stdin_interrupt_lock:
self.stdin_interrupt_running = False
def handle_stdin_input(self, msg_bytes, event_queue):
if self.synthesize_backspace_shortcut(msg_bytes, event_queue):
return
self.record_stdin_keypress(msg_bytes)
self.interrupt_output_on_stdin_input(msg_bytes)
self.inject_text_to_screen(msg_bytes)
def record_stdin_keypress(self, msg_bytes):
if msg_bytes != b"\t":
return
try:
self.env["runtime"]["InputManager"].record_unmanaged_keypress(
"KEY_TAB"
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver record_stdin_keypress: " + str(e),
debug.DebugLevel.ERROR,
)
def synthesize_backspace_shortcut(self, msg_bytes, event_queue):
if msg_bytes not in [b"\x7f", b"\x08"]:
return False
if "KEY_FENRIR" not in self.env["input"]["curr_input"]:
return False
event_time = time.time()
for event_state in [1, 0]:
try:
event_queue.put(
{
"Type": FenrirEventType.keyboard_input,
"data": {
"event_name": "KEY_BACKSPACE",
"event_value": 0,
"event_sec": int(event_time),
"event_usec": int((event_time % 1) * 1000000),
"event_state": event_state,
"event_type": 0,
},
},
block=False,
)
except Full:
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver synthesize_backspace_shortcut: Event queue full, dropping backspace events",
debug.DebugLevel.WARNING,
)
return True
def get_session_information(self): def get_session_information(self):
self.env["screen"]["autoIgnoreScreens"] = [] self.env["screen"]["autoIgnoreScreens"] = []
self.env["general"]["prev_user"] = getpass.getuser() self.env["general"]["prev_user"] = getpass.getuser()
@@ -420,7 +512,7 @@ class driver(screenDriver):
) )
break break
try: try:
self.inject_text_to_screen(msg_bytes) self.handle_stdin_input(msg_bytes, event_queue)
except Exception as e: except Exception as e:
self.env["runtime"][ self.env["runtime"][
"DebugManager" "DebugManager"
@@ -36,7 +36,7 @@ class driver(sound_driver):
self._initialized = _gstreamerAvailable self._initialized = _gstreamerAvailable
if not self._initialized: if not self._initialized:
global _availableError global _availableError
self.environment["runtime"]["DebugManager"].write_debug_out( self.env["runtime"]["DebugManager"].write_debug_out(
"Gstreamer not available " + _availableError, "Gstreamer not available " + _availableError,
debug.DebugLevel.ERROR, debug.DebugLevel.ERROR,
) )
+76
View File
@@ -0,0 +1,76 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.commands.commands.forward_keypress import command
from fenrirscreenreader.core.fenrirManager import FenrirManager
@pytest.mark.unit
def test_forward_keypress_arms_when_command_keys_are_still_down():
env = {
"input": {"key_forward": 0},
"runtime": {
"InputManager": Mock(no_key_pressed=Mock(return_value=False)),
"OutputManager": Mock(present_text=Mock()),
},
}
forward_command = command()
forward_command.initialize(env)
forward_command.run()
assert env["input"]["key_forward"] == -1
@pytest.mark.unit
def test_forward_keypress_activates_immediately_when_keyboard_is_idle():
env = {
"input": {"key_forward": 0},
"runtime": {
"InputManager": Mock(no_key_pressed=Mock(return_value=True)),
"OutputManager": Mock(present_text=Mock()),
},
}
forward_command = command()
forward_command.initialize(env)
forward_command.run()
assert env["input"]["key_forward"] == 1
@pytest.mark.unit
def test_forward_keypress_waits_for_command_release_before_forwarding():
input_manager = Mock(no_key_pressed=Mock(return_value=False))
manager = FenrirManager.__new__(FenrirManager)
manager.environment = {
"input": {"key_forward": -1},
"runtime": {"InputManager": input_manager},
}
manager.update_key_forward()
assert manager.environment["input"]["key_forward"] == -1
input_manager.no_key_pressed.return_value = True
manager.update_key_forward()
assert manager.environment["input"]["key_forward"] == 1
@pytest.mark.unit
def test_forward_keypress_stays_active_until_forwarded_key_release():
input_manager = Mock(no_key_pressed=Mock(return_value=False))
manager = FenrirManager.__new__(FenrirManager)
manager.environment = {
"input": {"key_forward": 1},
"runtime": {"InputManager": input_manager},
}
manager.update_key_forward()
assert manager.environment["input"]["key_forward"] == 1
input_manager.no_key_pressed.return_value = True
manager.update_key_forward()
assert manager.environment["input"]["key_forward"] == 0
+122
View File
@@ -0,0 +1,122 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.commandManager import CommandManager
from fenrirscreenreader.core.settingsManager import SettingsManager
def write_script(path):
path.write_text("#!/bin/sh\n", encoding="utf-8")
path.chmod(0o755)
def build_command_environment(local_path, system_path):
settings_manager = Mock()
settings_manager.get_user_script_path.return_value = str(local_path)
settings_manager.get_setting.return_value = str(system_path)
valid_keys = {
"KEY_A",
"KEY_B",
"KEY_C",
"KEY_D",
"KEY_E",
"KEY_SCRIPT",
}
return {
"commands": {"commands": {}},
"commandsIgnore": {"commands": {}},
"bindings": {
str([1, sorted(["KEY_C", "KEY_SCRIPT"])]): "EXISTING"
},
"rawBindings": {},
"general": {"curr_user": "Username"},
"runtime": {
"SettingsManager": settings_manager,
"DebugManager": Mock(write_debug_out=Mock()),
"InputManager": Mock(
is_valid_key=Mock(side_effect=lambda key: key in valid_keys)
),
},
}
@pytest.mark.unit
def test_local_scripts_load_before_non_conflicting_system_scripts(tmp_path):
local_path = tmp_path / "local"
system_path = tmp_path / "system"
local_path.mkdir()
system_path.mkdir()
write_script(local_path / "local_only__-__KEY_A")
write_script(local_path / "same_name__-__KEY_B")
write_script(system_path / "same_name__-__KEY_D")
write_script(system_path / "system_only__-__KEY_E")
env = build_command_environment(local_path, system_path)
command_manager = CommandManager()
command_manager.initialize = lambda environment: None
command_manager.env = env
command_manager.load_script_commands()
commands = env["commands"]["commands"]
assert "LOCAL_ONLY__-__KEY_A" in commands
assert "SYSTEM_ONLY__-__KEY_E" in commands
assert commands["SAME_NAME__-__KEY_B"].script_path == str(
local_path / "same_name__-__KEY_B"
)
assert "SAME_NAME__-__KEY_D" not in commands
@pytest.mark.unit
def test_system_scripts_skip_existing_shortcut_bindings(tmp_path):
local_path = tmp_path / "local"
system_path = tmp_path / "system"
local_path.mkdir()
system_path.mkdir()
write_script(system_path / "conflicting_key__-__KEY_C")
write_script(system_path / "system_only__-__KEY_E")
env = build_command_environment(local_path, system_path)
command_manager = CommandManager()
command_manager.env = env
command_manager.load_script_commands()
commands = env["commands"]["commands"]
assert "CONFLICTING_KEY__-__KEY_C" not in commands
assert "SYSTEM_ONLY__-__KEY_E" in commands
@pytest.mark.unit
def test_sound_theme_resolution_prefers_local_soundpacks(tmp_path):
manager = SettingsManager()
local_root = tmp_path / "local" / "fenrir"
system_root = tmp_path / "system" / "sounds" / "fenrir"
local_theme = local_root / "sounds" / "default"
system_theme = system_root / "default"
local_theme.mkdir(parents=True)
system_theme.mkdir(parents=True)
(local_theme / "soundicons.conf").write_text(
"Accept=Accept.wav\n", encoding="utf-8"
)
(system_theme / "soundicons.conf").write_text(
"Accept=SystemAccept.wav\n", encoding="utf-8"
)
manager.user_resource_root = str(local_root)
manager.system_sound_roots = [str(system_root)]
assert manager.resolve_sound_theme_path("default") == str(local_theme)
@pytest.mark.unit
def test_sound_theme_resolution_accepts_absolute_theme_path(tmp_path):
manager = SettingsManager()
theme_path = tmp_path / "custom"
theme_path.mkdir()
(theme_path / "soundicons.conf").write_text(
"Accept=Accept.wav\n", encoding="utf-8"
)
assert manager.resolve_sound_theme_path(str(theme_path)) == str(theme_path)
+65
View File
@@ -0,0 +1,65 @@
import importlib.util
from pathlib import Path
from unittest.mock import Mock
import pytest
def load_numlock_command():
command_path = (
Path(__file__).parents[2]
/ "src"
/ "fenrirscreenreader"
/ "commands"
/ "onKeyInput"
/ "80500-numlock.py"
)
spec = importlib.util.spec_from_file_location(
"numlock_command", command_path
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module.command
@pytest.mark.unit
def test_numlock_off_can_be_announced_from_release_event():
output_manager = Mock()
command = load_numlock_command()()
command.initialize(
{
"input": {
"old_num_lock": True,
"new_num_lock": False,
"curr_input": [],
"prev_input": ["KEY_NUMLOCK"],
},
"runtime": {"OutputManager": output_manager},
}
)
command.run()
output_manager.present_text.assert_called_once()
assert output_manager.present_text.call_args.args[0] == "Numlock off"
@pytest.mark.unit
def test_numlock_command_ignores_non_numlock_release():
output_manager = Mock()
command = load_numlock_command()()
command.initialize(
{
"input": {
"old_num_lock": True,
"new_num_lock": False,
"curr_input": [],
"prev_input": ["KEY_KP1"],
},
"runtime": {"OutputManager": output_manager},
}
)
command.run()
output_manager.present_text.assert_not_called()
+150
View File
@@ -0,0 +1,150 @@
import importlib.util
import threading
import time
from pathlib import Path
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.outputManager import OutputManager
def build_output_manager():
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting_as_float.return_value = 1.0
sound_driver = Mock()
speech_driver = Mock()
output_manager = OutputManager()
output_manager.env = {
"soundIcons": {
"ACCEPT": "/tmp/Accept.wav",
"ERRORSCREEN": "/tmp/ErrorScreen.wav",
},
"runtime": {
"SettingsManager": settings_manager,
"SoundDriver": sound_driver,
"SpeechDriver": speech_driver,
"DebugManager": Mock(write_debug_out=Mock()),
},
}
return output_manager, sound_driver, speech_driver
def load_key_interrupt_module():
module_path = (
Path(__file__).resolve().parents[2]
/ "src"
/ "fenrirscreenreader"
/ "commands"
/ "onKeyInput"
/ "10000-shut_up.py"
)
spec = importlib.util.spec_from_file_location(
"fenrir_key_interrupt", module_path
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
@pytest.mark.unit
def test_present_text_allows_sound_only_feedback():
output_manager, sound_driver, _speech_driver = build_output_manager()
output_manager.present_text("", sound_icon="Accept", interrupt=False)
sound_driver.play_sound_file.assert_called_once_with(
"/tmp/Accept.wav", False
)
@pytest.mark.unit
def test_play_sound_supports_error_alias():
output_manager, sound_driver, _speech_driver = build_output_manager()
assert output_manager.play_sound("Error") is True
sound_driver.play_sound_file.assert_called_once_with(
"/tmp/ErrorScreen.wav", True
)
@pytest.mark.unit
def test_interrupt_output_async_does_not_block_on_slow_cancel():
output_manager, _sound_driver, speech_driver = build_output_manager()
interrupt_started = threading.Event()
release_interrupt = threading.Event()
def slow_cancel():
interrupt_started.set()
release_interrupt.wait(timeout=1.0)
speech_driver.cancel.side_effect = slow_cancel
start_time = time.monotonic()
output_manager.interrupt_output_async()
elapsed = time.monotonic() - start_time
try:
assert interrupt_started.wait(timeout=0.2)
assert elapsed < 0.2
output_manager.interrupt_output_async()
assert speech_driver.cancel.call_count == 1
finally:
release_interrupt.set()
output_manager.interrupt_thread.join(timeout=1.0)
@pytest.mark.unit
def test_interrupt_output_waits_only_briefly_for_slow_cancel():
output_manager, _sound_driver, speech_driver = build_output_manager()
interrupt_started = threading.Event()
release_interrupt = threading.Event()
def slow_cancel():
interrupt_started.set()
release_interrupt.wait(timeout=1.0)
speech_driver.cancel.side_effect = slow_cancel
start_time = time.monotonic()
output_manager.interrupt_output()
elapsed = time.monotonic() - start_time
try:
assert interrupt_started.wait(timeout=0.2)
assert elapsed < 0.2
output_manager.interrupt_output()
assert speech_driver.cancel.call_count == 1
finally:
release_interrupt.set()
output_manager.interrupt_thread.join(timeout=1.0)
@pytest.mark.unit
def test_key_interrupt_command_uses_nonblocking_interrupt():
module = load_key_interrupt_module()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting.return_value = ""
output_manager = Mock()
env = {
"input": {
"curr_input": ["KEY_A"],
"prev_input": [],
},
"runtime": {
"InputManager": Mock(no_key_pressed=Mock(return_value=False)),
"OutputManager": output_manager,
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
"SettingsManager": settings_manager,
},
}
command = module.command()
command.initialize(env)
command.run()
output_manager.interrupt_output_async.assert_called_once_with()
output_manager.interrupt_output.assert_not_called()
+177
View File
@@ -1,5 +1,10 @@
import threading
import time
from unittest.mock import Mock
import pytest import pytest
from fenrirscreenreader.core.eventData import FenrirEventType
from fenrirscreenreader.screenDriver.ptyDriver import PTYConstants from fenrirscreenreader.screenDriver.ptyDriver import PTYConstants
from fenrirscreenreader.screenDriver.ptyDriver import Terminal from fenrirscreenreader.screenDriver.ptyDriver import Terminal
from fenrirscreenreader.screenDriver.ptyDriver import driver as PtyDriver from fenrirscreenreader.screenDriver.ptyDriver import driver as PtyDriver
@@ -52,3 +57,175 @@ def test_optional_float_setting_uses_default_when_missing():
) )
== PTYConstants.OUTPUT_READ_TIMEOUT == PTYConstants.OUTPUT_READ_TIMEOUT
) )
@pytest.mark.unit
def test_pty_stdin_input_interrupts_output_when_all_keys_interrupt_enabled():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting.return_value = ""
output_manager = Mock()
pty_driver.env = {
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
}
}
pty_driver.interrupt_output_on_stdin_input(b"a")
pty_driver.stdin_interrupt_thread.join(timeout=1.0)
output_manager.interrupt_output.assert_called_once_with()
@pytest.mark.unit
def test_pty_stdin_input_interrupt_does_not_block_input_injection():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting.return_value = ""
interrupt_started = threading.Event()
release_interrupt = threading.Event()
def slow_interrupt():
interrupt_started.set()
release_interrupt.wait(timeout=1.0)
output_manager = Mock(interrupt_output=Mock(side_effect=slow_interrupt))
pty_driver.env = {
"input": {"curr_input": []},
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
"DebugManager": Mock(write_debug_out=Mock()),
},
}
pty_driver.inject_text_to_screen = Mock()
start_time = time.monotonic()
pty_driver.handle_stdin_input(b"a", Mock())
elapsed = time.monotonic() - start_time
try:
assert interrupt_started.wait(timeout=0.2)
assert elapsed < 0.2
pty_driver.inject_text_to_screen.assert_called_once_with(b"a")
finally:
release_interrupt.set()
pty_driver.stdin_interrupt_thread.join(timeout=1.0)
@pytest.mark.unit
def test_pty_raw_tab_records_recent_tab_keypress():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = False
input_manager = Mock()
pty_driver.env = {
"input": {"curr_input": []},
"runtime": {
"DebugManager": Mock(write_debug_out=Mock()),
"InputManager": input_manager,
"SettingsManager": settings_manager,
},
}
pty_driver.inject_text_to_screen = Mock()
pty_driver.handle_stdin_input(b"\t", Mock())
input_manager.record_unmanaged_keypress.assert_called_once_with("KEY_TAB")
pty_driver.inject_text_to_screen.assert_called_once_with(b"\t")
@pytest.mark.unit
def test_pty_plain_stdin_does_not_record_tab_keypress():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = False
input_manager = Mock()
pty_driver.env = {
"input": {"curr_input": []},
"runtime": {
"DebugManager": Mock(write_debug_out=Mock()),
"InputManager": input_manager,
"SettingsManager": settings_manager,
},
}
pty_driver.inject_text_to_screen = Mock()
pty_driver.handle_stdin_input(b"a", Mock())
input_manager.record_unmanaged_keypress.assert_not_called()
pty_driver.inject_text_to_screen.assert_called_once_with(b"a")
@pytest.mark.unit
def test_pty_stdin_input_honors_interrupt_disabled():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = False
output_manager = Mock()
pty_driver.env = {
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
}
}
pty_driver.interrupt_output_on_stdin_input(b"a")
output_manager.interrupt_output.assert_not_called()
@pytest.mark.unit
def test_pty_stdin_input_leaves_filtered_interrupts_to_key_events():
pty_driver = PtyDriver()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting.return_value = "KEY_ENTER"
output_manager = Mock()
pty_driver.env = {
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
}
}
pty_driver.interrupt_output_on_stdin_input(b"a")
output_manager.interrupt_output.assert_not_called()
@pytest.mark.unit
def test_pty_backspace_with_fenrir_key_synthesizes_shortcut_events():
pty_driver = PtyDriver()
event_queue = Mock()
pty_driver.env = {
"input": {"curr_input": ["KEY_FENRIR"]},
}
handled = pty_driver.synthesize_backspace_shortcut(b"\x7f", event_queue)
assert handled is True
assert event_queue.put.call_count == 2
first_event = event_queue.put.call_args_list[0].args[0]
second_event = event_queue.put.call_args_list[1].args[0]
assert first_event["Type"] == FenrirEventType.keyboard_input
assert first_event["data"]["event_name"] == "KEY_BACKSPACE"
assert first_event["data"]["event_state"] == 1
assert second_event["data"]["event_state"] == 0
@pytest.mark.unit
def test_pty_plain_backspace_is_not_synthesized():
pty_driver = PtyDriver()
event_queue = Mock()
pty_driver.env = {
"input": {"curr_input": []},
}
handled = pty_driver.synthesize_backspace_shortcut(b"\x7f", event_queue)
assert handled is False
event_queue.put.assert_not_called()
+92
View File
@@ -8,10 +8,12 @@ for all configurable settings that could cause crashes or accessibility issues.
import pytest import pytest
import sys import sys
from pathlib import Path from pathlib import Path
from unittest.mock import Mock
# Import the settings manager # Import the settings manager
from fenrirscreenreader.core.settingsData import settings_data from fenrirscreenreader.core.settingsData import settings_data
from fenrirscreenreader.core.settingsManager import SettingsManager from fenrirscreenreader.core.settingsManager import SettingsManager
from fenrirscreenreader.commands.commands import save_settings
@pytest.mark.unit @pytest.mark.unit
@@ -198,3 +200,93 @@ class TestValidationSkipsUnknownSettings:
def test_focus_settings_define_tui_toggle(): def test_focus_settings_define_tui_toggle():
"""Focus settings should include the TUI toggle used by on-screen handlers.""" """Focus settings should include the TUI toggle used by on-screen handlers."""
assert settings_data["focus"]["tui"] is False assert settings_data["focus"]["tui"] is False
@pytest.mark.unit
@pytest.mark.settings
class TestSettingsPathSelection:
"""Test root/user settings load and save path selection."""
def setup_method(self):
self.manager = SettingsManager()
def configure_paths(self, tmp_path):
system_root = tmp_path / "etc" / "fenrirscreenreader"
system_file = system_root / "settings" / "settings.conf"
user_file = (
tmp_path
/ "home"
/ "Username"
/ ".local"
/ "share"
/ "stormux"
/ "fenrirscreenreader"
/ "settings"
/ "settings.conf"
)
self.manager.system_settings_root = str(system_root) + "/"
self.manager.system_settings_file = str(system_file)
self.manager.user_settings_file = str(user_file)
return system_file, user_file
def test_non_root_loads_user_settings_when_present(
self, tmp_path, monkeypatch
):
system_file, user_file = self.configure_paths(tmp_path)
system_file.parent.mkdir(parents=True)
system_file.write_text("[general]\n", encoding="utf-8")
user_file.parent.mkdir(parents=True)
user_file.write_text("[general]\n", encoding="utf-8")
monkeypatch.setattr("os.geteuid", lambda: 1000)
assert self.manager.resolve_settings_file() == str(user_file)
def test_non_root_falls_back_to_system_settings(
self, tmp_path, monkeypatch
):
system_file, _user_file = self.configure_paths(tmp_path)
system_file.parent.mkdir(parents=True)
system_file.write_text("[general]\n", encoding="utf-8")
monkeypatch.setattr("os.geteuid", lambda: 1000)
assert self.manager.resolve_settings_file() == str(system_file)
def test_root_uses_system_settings_even_when_user_settings_exists(
self, tmp_path, monkeypatch
):
system_file, user_file = self.configure_paths(tmp_path)
system_file.parent.mkdir(parents=True)
system_file.write_text("[general]\n", encoding="utf-8")
user_file.parent.mkdir(parents=True)
user_file.write_text("[general]\n", encoding="utf-8")
monkeypatch.setattr("os.geteuid", lambda: 0)
assert self.manager.resolve_settings_file() == str(system_file)
def test_save_default_path_follows_effective_user(
self, tmp_path, monkeypatch
):
system_file, user_file = self.configure_paths(tmp_path)
monkeypatch.setattr("os.geteuid", lambda: 1000)
assert self.manager.get_default_save_settings_file() == str(user_file)
monkeypatch.setattr("os.geteuid", lambda: 0)
assert self.manager.get_default_save_settings_file() == str(system_file)
def test_save_settings_command_uses_default_save_path(self):
manager = Mock()
output_manager = Mock()
command = save_settings.command()
command.initialize(
{
"runtime": {
"SettingsManager": manager,
"OutputManager": output_manager,
}
}
)
command.run()
manager.save_settings.assert_called_once_with()
+29
View File
@@ -0,0 +1,29 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.soundDriver import sound_driver
from fenrirscreenreader.soundDriver import gstreamerDriver
@pytest.mark.unit
def test_base_sound_driver_shutdown_clears_initialized_flag():
driver = sound_driver()
driver.initialize({})
driver.shutdown()
assert driver._initialized is False
@pytest.mark.unit
def test_gstreamer_driver_unavailable_logs_without_crashing(monkeypatch):
monkeypatch.setattr(gstreamerDriver, "_gstreamerAvailable", False)
monkeypatch.setattr(gstreamerDriver, "_availableError", "missing", raising=False)
debug_manager = Mock(write_debug_out=Mock())
driver = gstreamerDriver.driver()
driver.initialize({"runtime": {"DebugManager": debug_manager}})
debug_manager.write_debug_out.assert_called()
assert driver._initialized is False
+33
View File
@@ -0,0 +1,33 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.commands.commands import subprocess as subprocess_command
@pytest.mark.unit
def test_script_command_executes_without_shell(monkeypatch):
process = Mock()
process.communicate.return_value = (b"done", b"")
popen = Mock(return_value=process)
monkeypatch.setattr(subprocess_command, "Popen", popen)
output_manager = Mock()
command = subprocess_command.command()
command.initialize(
{
"general": {"curr_user": "Username"},
"runtime": {"OutputManager": output_manager},
},
"/tmp/script with spaces",
)
command._thread_run()
popen.assert_called_once_with(
["/tmp/script with spaces", "Username"],
stdout=subprocess_command.PIPE,
stderr=subprocess_command.PIPE,
)
output_manager.present_text.assert_called_once_with(
"done", sound_icon="", interrupt=False
)
+21
View File
@@ -212,6 +212,27 @@ def test_recent_tab_screen_update_works_without_key_input_snapshot():
assert manager.process_update() == "cuments/" assert manager.process_update() == "cuments/"
@pytest.mark.unit
def test_recent_raw_pty_tab_speaks_short_completion_without_capture():
manager, env, input_manager = _build_env(
"cd Documents/".ljust(20), {"x": 13, "y": 0}
)
input_manager.get_last_deepest_input.return_value = ["KEY_TAB"]
input_manager.get_last_event.return_value = None
env["screen"]["old_content_text"] = "cd Docume".ljust(20)
env["screen"]["old_cursor"] = {"x": 9, "y": 0}
env["commandBuffer"]["tabCompletion"]["pending"] = None
_set_screen_update(
env,
"cd Documents/".ljust(20),
{"x": 13, "y": 0},
delta="nts/",
typing=True,
)
assert manager.process_update() == "nts/"
@pytest.mark.unit @pytest.mark.unit
def test_large_insertion_echo_speaks_pasted_cursor_text(): def test_large_insertion_echo_speaks_pasted_cursor_text():
large_insertion_module = _load_large_insertion_module() large_insertion_module = _load_large_insertion_module()
+32 -5
View File
@@ -167,23 +167,50 @@ def test_x11_build_passive_grabs_for_fenrir_keys_and_shortcuts():
input_manager = Mock(convert_event_name=lambda key: key) input_manager = Mock(convert_event_name=lambda key: key)
x11.env = { x11.env = {
"input": { "input": {
"fenrir_key": ["KEY_KP0", "KEY_CAPSLOCK"], "fenrir_key": ["KEY_KP0", "KEY_META"],
"script_key": [], "script_key": [],
}, },
"rawBindings": { "rawBindings": {
"fenrir_combo": [1, ["KEY_FENRIR", "KEY_KP8"]], "fenrir_combo": [1, ["KEY_FENRIR", "KEY_KP8"]],
"bare_keypad": [1, ["KEY_KP5"]], "bare_keypad": [1, ["KEY_KP5"]],
"ctrl_keypad": [1, ["KEY_CTRL", "KEY_KP2"]], "ctrl_keypad": [1, ["KEY_CTRL", "KEY_KP2"]],
"meta_combo": [1, ["KEY_FENRIR", "KEY_BACKSPACE"]],
}, },
"runtime": {"InputManager": input_manager}, "runtime": {"InputManager": input_manager},
} }
grabs = x11.build_passive_grabs() grabs = x11.build_passive_grabs()
assert ("KEY_KP0", 0) in grabs assert ("KEY_KP0", 0, True) in grabs
assert ("KEY_CAPSLOCK", 0) in grabs assert ("KEY_META", 0, True) in grabs
assert ("KEY_KP5", 0) in grabs assert ("KEY_NUMLOCK", 0, True) in grabs
assert ("KEY_KP2", X.ControlMask) in grabs assert ("KEY_KP5", 0, False) in grabs
assert ("KEY_KP2", X.ControlMask, False) in grabs
assert ("KEY_BACKSPACE", X.Mod4Mask, True) in grabs
@pytest.mark.unit
def test_x11_optional_modifier_masks_can_exclude_numlock():
x11 = X11Driver()
x11.num_lock_mask = X.Mod2Mask
masks = x11.optional_modifier_masks(0, include_num_lock=False)
assert X.Mod2Mask not in masks
assert X.Mod2Mask | X.LockMask not in masks
@pytest.mark.unit
def test_x11_get_led_state_reads_lock_modifiers_from_pointer_mask():
x11 = X11Driver()
x11.num_lock_mask = X.Mod2Mask
pointer = Mock(mask=X.Mod2Mask | X.LockMask)
x11.root = Mock()
x11.root.query_pointer.return_value = pointer
assert x11.get_led_state(0) is True
assert x11.get_led_state(1) is True
assert x11.get_led_state(2) is False
@pytest.mark.unit @pytest.mark.unit