Fixed some half-completed work in generic drivers. Minor -x bug fixes.

This commit is contained in:
Storm Dragon
2026-06-18 10:35:48 -04:00
parent fd5fe5b328
commit 2b7d205f06
13 changed files with 425 additions and 119 deletions
@@ -22,4 +22,7 @@ command_info = {
# 'curr_command': '',
"lastCommandExecutionTime": time.time(),
"lastCommandRequestTime": time.time(),
"lastCommand": "",
"lastCommandSection": "",
"lastCommandRunTime": 0,
}
@@ -474,6 +474,10 @@ class CommandManager:
def run_command(self, command, section="commands"):
if self.command_exists(command, section):
try:
command_time = time.time()
self.env["commandInfo"]["lastCommand"] = command
self.env["commandInfo"]["lastCommandSection"] = section
self.env["commandInfo"]["lastCommandRunTime"] = command_time
self.env["runtime"]["DebugManager"].write_debug_out(
"run_command command:" + section + "." + command,
debug.DebugLevel.INFO,
+11 -3
View File
@@ -12,8 +12,13 @@ settings_data = {
"driver": "genericDriver",
"theme": "default",
"volume": 1.0,
"generic_play_file_command": "play -q -v fenrirVolume fenrirSoundFile",
"generic_frequency_command": "play -q -v fenrirVolume -n -c1 synth fenrirDuration sine fenrirFrequence",
"generic_play_file_command": (
"play -q -v fenrir_volume fenrir_sound_file"
),
"generic_frequency_command": (
"play -q -v fenrir_volume -n -c1 synth "
"fenrir_duration sine fenrir_frequency"
),
"progress_monitoring": True,
},
"speech": {
@@ -38,7 +43,10 @@ settings_data = {
"batch_flush_interval": 0.5,
"max_batch_lines": 100,
"flood_line_threshold": 500,
"generic_speech_command": 'espeak -a fenrirVolume -s fenrirRate -p fenrirPitch -v fenrirVoice "fenrirText"',
"generic_speech_command": (
"espeak -a fenrir_volume -s fenrir_rate -p fenrir_pitch "
'-v fenrir_voice "fenrir_text"'
),
"fenrir_min_volume": 0,
"fenrir_max_volume": 200,
"fenrir_min_pitch": 0,
+2 -2
View File
@@ -4,5 +4,5 @@
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
version = "2026.06.01"
code_name = "master"
version = "2026.06.18"
code_name = "testing"
@@ -304,6 +304,11 @@ class driver(screenDriver):
"keyboard", "interrupt_on_key_press_filter"
).strip():
return
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver interrupt_output_on_stdin_input: "
+ repr(msg_bytes),
debug.DebugLevel.INFO,
)
self.start_stdin_interrupt_thread()
def start_stdin_interrupt_thread(self):
@@ -335,10 +340,43 @@ class driver(screenDriver):
return
if self.handle_vmenu_stdin_input(msg_bytes, event_queue):
return
if self.suppress_recent_review_stdin_tail(msg_bytes):
return
self.record_stdin_keypress(msg_bytes)
self.interrupt_output_on_stdin_input(msg_bytes)
self.inject_text_to_screen(msg_bytes)
def suppress_recent_review_stdin_tail(self, msg_bytes):
if not self.is_keyboard_control_sequence(msg_bytes):
return False
try:
command_info = self.env["commandInfo"]
last_command = command_info.get("lastCommand", "")
last_section = command_info.get("lastCommandSection", "")
last_run_time = command_info.get("lastCommandRunTime", 0)
except Exception:
return False
if last_section != "commands" or not last_command.startswith("REVIEW_"):
return False
if time.time() - last_run_time > 0.8:
return False
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver suppressing recent review stdin tail: "
+ repr(msg_bytes),
debug.DebugLevel.INFO,
)
return True
def is_keyboard_control_sequence(self, msg_bytes):
if not msg_bytes:
return False
if msg_bytes.startswith(b"\x1b"):
return True
if len(msg_bytes) == 1:
value = msg_bytes[0]
return value < 32 and msg_bytes not in [b"\t", b"\n", b"\r"]
return False
def handle_vmenu_stdin_input(self, msg_bytes, event_queue):
if not self.is_vmenu_active():
return False
@@ -26,15 +26,15 @@ class driver(sound_driver):
Attributes:
proc: Currently running subprocess for sound playback
soundFileCommand (str): Command template for playing sound files
frequenceCommand (str): Command template for generating frequencies
sound_file_command (str): Command template for playing sound files
frequency_command (str): Command template for generating frequencies
"""
def __init__(self):
sound_driver.__init__(self)
self.proc = None
self.soundType = ""
self.soundFileCommand = ""
self.frequenceCommand = ""
self.sound_type = ""
self.sound_file_command = ""
self.frequency_command = ""
def initialize(self, environment):
"""Initialize the generic sound driver.
@@ -46,17 +46,20 @@ class driver(sound_driver):
environment: Fenrir environment dictionary with settings
"""
self.env = environment
self.soundFileCommand = self.env["runtime"][
self.sound_file_command = self.env["runtime"][
"SettingsManager"
].get_setting("sound", "generic_play_file_command")
self.frequenceCommand = self.env["runtime"][
self.frequency_command = self.env["runtime"][
"SettingsManager"
].get_setting("sound", "generic_frequency_command")
if self.soundFileCommand == "":
self.soundFileCommand = "play -q -v fenrirVolume fenrirSoundFile"
if self.frequenceCommand == "":
self.frequenceCommand = (
"play -q -v fenrirVolume -n -c1 synth fenrirDuration sine fenrirFrequence"
if self.sound_file_command == "":
self.sound_file_command = (
"play -q -v fenrir_volume fenrir_sound_file"
)
if self.frequency_command == "":
self.frequency_command = (
"play -q -v fenrir_volume -n -c1 synth "
"fenrir_duration sine fenrir_frequency"
)
self._initialized = True
@@ -75,22 +78,22 @@ class driver(sound_driver):
return
if interrupt:
self.cancel()
popen_frequence_command = shlex.split(self.frequenceCommand)
for idx, word in enumerate(popen_frequence_command):
popen_frequency_command = shlex.split(self.frequency_command)
for idx, word in enumerate(popen_frequency_command):
word = word.replace(
"fenrirVolume", str(self.volume * adjust_volume)
"fenrir_volume", str(self.volume * adjust_volume)
)
word = word.replace("fenrirDuration", str(duration))
word = word.replace("fenrirFrequence", str(frequence))
popen_frequence_command[idx] = word
word = word.replace("fenrir_duration", str(duration))
word = word.replace("fenrir_frequency", str(frequence))
popen_frequency_command[idx] = word
self.proc = subprocess.Popen(
popen_frequence_command,
popen_frequency_command,
stdin=None,
stdout=None,
stderr=None,
shell=False,
)
self.soundType = "frequence"
self.sound_type = "frequence"
def play_sound_file(self, file_path, interrupt=True):
"""Play a sound file.
@@ -108,13 +111,13 @@ class driver(sound_driver):
if not os.path.isfile(file_path) or ".." in file_path:
return
popen_sound_file_command = shlex.split(self.soundFileCommand)
popen_sound_file_command = shlex.split(self.sound_file_command)
for idx, word in enumerate(popen_sound_file_command):
word = word.replace("fenrirVolume", str(self.volume))
word = word.replace("fenrirSoundFile", shlex.quote(str(file_path)))
word = word.replace("fenrir_volume", str(self.volume))
word = word.replace("fenrir_sound_file", str(file_path))
popen_sound_file_command[idx] = word
self.proc = subprocess.Popen(popen_sound_file_command, shell=False)
self.soundType = "file"
self.sound_type = "file"
def cancel(self):
"""Cancel currently playing sound.
@@ -123,9 +126,9 @@ class driver(sound_driver):
"""
if not self._initialized:
return
if self.soundType == "":
if self.sound_type == "":
return
if self.soundType == "file":
if self.sound_type == "file":
self.proc.kill()
try:
# Wait for process to finish to prevent zombies
@@ -134,7 +137,7 @@ class driver(sound_driver):
pass # Process already terminated
except Exception as e:
pass # Handle any other wait errors
if self.soundType == "frequence":
if self.sound_type == "frequence":
self.proc.kill()
try:
# Wait for process to finish to prevent zombies
@@ -143,4 +146,4 @@ class driver(sound_driver):
pass # Process already terminated
except Exception as e:
pass # Handle any other wait errors
self.soundType = ""
self.sound_type = ""
@@ -30,49 +30,52 @@ class driver(speech_driver):
def __init__(self):
speech_driver.__init__(self)
self.proc = None
self.speechThread = Thread(target=self.worker)
self.speech_thread = Thread(target=self.worker)
self.lock = Lock()
self.textQueue = SpeakQueue()
self.text_queue = SpeakQueue()
def initialize(self, environment):
self.env = environment
self.minVolume = self.env["runtime"][
self.min_volume = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_min_volume")
self.maxVolume = self.env["runtime"][
self.max_volume = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_max_volume")
self.minPitch = self.env["runtime"][
self.min_pitch = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_min_pitch")
self.maxPitch = self.env["runtime"][
self.max_pitch = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_max_pitch")
self.minRate = self.env["runtime"][
self.min_rate = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_min_rate")
self.maxRate = self.env["runtime"][
self.max_rate = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_max_rate")
self.speechCommand = self.env["runtime"][
self.speech_command = self.env["runtime"][
"SettingsManager"
].get_setting("speech", "generic_speech_command")
if self.speechCommand == "":
self.speechCommand = 'espeak -a fenrirVolume -s fenrirRate -p fenrirPitch -v fenrirVoice -- "fenrirText"'
if self.speech_command == "":
self.speech_command = (
"espeak -a fenrir_volume -s fenrir_rate -p fenrir_pitch "
'-v fenrir_voice -- "fenrir_text"'
)
if False: # for debugging overwrite here
# self.speechCommand = 'spd-say --wait -r 100 -i 100 "fenrirText"'
self.speechCommand = 'flite -t "fenrirText"'
# self.speech_command = 'spd-say --wait -r 100 -i 100 "fenrir_text"'
self.speech_command = 'flite -t "fenrir_text"'
self._is_initialized = True
if self._is_initialized:
self.speechThread.start()
self.speech_thread.start()
def shutdown(self):
if not self._is_initialized:
return
self.cancel()
self.textQueue.put(-1)
self.text_queue.put(-1)
def speak(self, text, queueable=True, ignore_punctuation=False):
if not self._is_initialized:
@@ -88,7 +91,7 @@ class driver(speech_driver):
"language": self.language,
"voice": self.voice,
}
self.textQueue.put(utterance.copy())
self.text_queue.put(utterance.copy())
def cancel(self):
if not self._is_initialized:
@@ -129,7 +132,7 @@ class driver(speech_driver):
def clear_buffer(self):
if not self._is_initialized:
return
self.textQueue.clear()
self.text_queue.clear()
def set_voice(self, voice):
if not self._is_initialized:
@@ -140,13 +143,13 @@ class driver(speech_driver):
if not self._is_initialized:
return
self.pitch = str(
self.minPitch + pitch * (self.maxPitch - self.minPitch)
self.min_pitch + pitch * (self.max_pitch - self.min_pitch)
)
def set_rate(self, rate):
if not self._is_initialized:
return
self.rate = str(self.minRate + rate * (self.maxRate - self.minRate))
self.rate = str(self.min_rate + rate * (self.max_rate - self.min_rate))
def set_module(self, module):
if not self._is_initialized:
@@ -162,12 +165,29 @@ class driver(speech_driver):
if not self._is_initialized:
return
self.volume = str(
self.minVolume + volume * (self.maxVolume - self.minVolume)
self.min_volume + volume * (self.max_volume - self.min_volume)
)
def _build_speech_command(self, utterance):
replacements = {
"fenrir_volume": str(utterance["volume"]),
"fenrir_module": str(utterance["module"]),
"fenrir_language": str(utterance["language"]),
"fenrir_voice": str(utterance["voice"]),
"fenrir_pitch": str(utterance["pitch"]),
"fenrir_rate": str(utterance["rate"]),
"fenrir_text": shlex.quote(str(utterance["text"])),
}
speech_command = shlex.split(self.speech_command)
for idx, word in enumerate(speech_command):
for placeholder, value in replacements.items():
word = word.replace(placeholder, value)
speech_command[idx] = word
return speech_command
def worker(self):
while True:
utterance = self.textQueue.get()
utterance = self.text_queue.get()
if isinstance(utterance, int):
if utterance == -1:
@@ -209,21 +229,7 @@ class driver(speech_driver):
if not isinstance(utterance["rate"], str):
utterance["rate"] = ""
popen_speech_command = shlex.split(self.speechCommand)
for idx, word in enumerate(popen_speech_command):
word = word.replace("fenrirVolume", str(utterance["volume"]))
word = word.replace("fenrirModule", str(utterance["module"]))
word = word.replace(
"fenrirLanguage", str(utterance["language"])
)
word = word.replace("fenrirVoice", str(utterance["voice"]))
word = word.replace("fenrirPitch", str(utterance["pitch"]))
word = word.replace("fenrirRate", str(utterance["rate"]))
# Properly quote text to prevent command injection
word = word.replace(
"fenrirText", shlex.quote(str(utterance["text"]))
)
popen_speech_command[idx] = word
popen_speech_command = self._build_speech_command(utterance)
try:
self.env["runtime"]["DebugManager"].write_debug_out(