Files

1431 lines
56 KiB
Python
Executable File

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Self-voiced Terminal Menu System for Stormux game image
import os
import platform
import sys
import time
import threading
import signal
import curses
import subprocess
import speechd # Python bindings for Speech Dispatcher
import configparser
import pathlib
import re
import simpleaudio as sa
import json
import shlex
from dataclasses import dataclass
from pathlib import Path
from stormux_speech_settings import load_speech_settings, save_speech_settings
INSTALL_GAME_RUNNER = Path("/home/stormux/.local/.functions/install_game.sh")
SERVICE_STATE_HELPER = Path("/home/stormux/.local/bin/stormux_service_state.sh")
SERVICE_ACTION_LOG = Path.home() / ".cache" / "stormux" / "service-actions.log"
@dataclass(frozen=True)
class GameEntry:
section: str
name: str
installScript: Path
launchScript: Path | None = None
@property
def isInstalled(self):
return self.launchScript is not None
@property
def defaultAction(self):
return "Play" if self.isInstalled else "Install"
def find_launch_script(launchRoot, _section, gameName):
candidate = Path(launchRoot) / f"{gameName}.sh"
if candidate.exists():
return candidate
return None
def discover_game_entries(installRoot=None, launchRoot=None):
installRoot = Path(installRoot or Path.home() / ".local" / ".install")
launchRoot = Path(launchRoot or Path.home() / ".local" / ".launch")
entries = []
if not installRoot.is_dir():
return entries
for sectionRoot in sorted(path for path in installRoot.iterdir() if path.is_dir()):
section = sectionRoot.name
for installScript in sorted(sectionRoot.glob("*.sh")):
gameName = installScript.stem
launchScript = find_launch_script(launchRoot, section, gameName)
entries.append(GameEntry(section, gameName, installScript, launchScript))
return entries
def build_game_command(entry):
if entry.launchScript is not None:
launchPath = str(entry.launchScript)
return f"STORMUX_LAUNCH_SCRIPT={launchPath!r} startx"
return shlex.quote(str(entry.installScript))
def build_game_install_command(entry):
return f"{shlex.quote(str(INSTALL_GAME_RUNNER))} {shlex.quote(str(entry.installScript))}"
def build_game_uninstall_command(entry):
if entry.launchScript is None:
return ""
launchPath = shlex.quote(str(entry.launchScript))
gameName = shlex.quote(entry.name)
return f"rm -f {launchPath}; echo {gameName} uninstalled"
def build_game_actions(entry):
actions = []
if entry.isInstalled:
actions.append(("Play", build_game_command(entry)))
actions.append(("Uninstall", build_game_uninstall_command(entry)))
else:
actions.append(("Install", build_game_install_command(entry)))
return actions
def submenu_command(callback):
def wrapped():
return callback()
wrapped.opensSubmenu = True
return wrapped
def run_service_action(action, serviceName, helper=SERVICE_STATE_HELPER, logPath=SERVICE_ACTION_LOG):
command = ["sudo", "-n", str(helper), action, serviceName]
result = subprocess.run(command, capture_output=True, text=True, check=False)
logPath = Path(logPath)
logPath.parent.mkdir(parents=True, exist_ok=True)
with logPath.open("a", encoding="utf-8") as logFile:
logFile.write(f"Service action {' '.join(command)} [{time.ctime()}]\n")
logFile.write(f"Return code: {result.returncode}\n")
if result.stdout:
logFile.write("stdout:\n")
logFile.write(result.stdout)
if not result.stdout.endswith("\n"):
logFile.write("\n")
if result.stderr:
logFile.write("stderr:\n")
logFile.write(result.stderr)
if not result.stderr.endswith("\n"):
logFile.write("\n")
return result
class VoicedMenu:
def __init__(self, title="Stormux Game Menu"):
self.title = title
self.menuSections = {} # Dictionary to hold sections and their items
self.sectionNames = [] # List to maintain section order
self.currentSection = 0 # Index of current section
self.currentItemIndices = {} # Current item index for each section
self.stdscr = None
# Check if system is installed or running from USB
self.is_installed = os.path.exists("/home/stormux/.baremetal")
# System services dictionary - maps friendly names to service names
self.systemMenuServices = {
'Braille': 'brltty.path',
'D L N A Server': 'minidlna.service',
'Fenrir Screen Reader': 'fenrirscreenreader-tty.service',
'Bluetooth': 'bluetooth.service',
'SSH': 'sshd.service',
'Battery Monitoring': 'battery-monitor.service'
}
# Config settings
self.configDir = os.path.expanduser("~/.config/stormux")
self.configFile = os.path.join(self.configDir, "game_launcher.conf")
self.config = configparser.ConfigParser()
# Default settings
self.speechRate = 0 # Normal speech rate (0 is default in speechd)
self.speechPitch = 0 # Normal speech pitch
self.volume = 50 # Default volume level
# Load settings
self.load_settings()
# Initialize speech client
self.speechClient = None
self.init_speech()
# Track playing sound
self.currentSound = None
# Load downloadable games registry
self.downloadable_games = self.load_downloadable_games()
self.savedMenuState = None
self.menuStateStack = []
self.activeGameEntry = None
def load_downloadable_games(self):
"""Load downloadable games registry from JSON file"""
registry_path = "/usr/share/stormux/downloadable_games.json"
try:
if os.path.exists(registry_path):
with open(registry_path, 'r') as f:
data = json.load(f)
return data.get('downloadable_games', {})
return {}
except Exception as e:
print(f"Error loading downloadable games registry: {e}")
return {}
def is_game_installed(self, game_id):
"""Check if a downloadable game is installed"""
if game_id not in self.downloadable_games:
return True # Not a downloadable game, assume installed
game_info = self.downloadable_games[game_id]
game_dir = os.path.expanduser(f"~/.local/games/{game_info['directory']}")
return os.path.exists(game_dir) and os.path.exists(os.path.join(game_dir, game_info['executable']))
def get_display_name(self, game_name, game_id=None):
"""Get display name for menu item, adding (not installed) if needed"""
if game_id and game_id in self.downloadable_games and not self.is_game_installed(game_id):
return f"{game_name} (not installed)"
return game_name
def init_speech(self):
"""Initialize the speech client"""
try:
self.speechClient = speechd.SSIPClient('stormux_menu')
self.speechClient.set_priority(speechd.Priority.IMPORTANT)
self.speechClient.set_punctuation(speechd.PunctuationMode.SOME)
# Apply speech settings from saved values
self.speechClient.set_rate(self.speechRate)
self.speechClient.set_pitch(self.speechPitch)
except Exception as e:
print(f"Could not initialize speech: {e}")
# Fallback to None - the speak method will handle this
def load_settings(self):
"""Load settings from config file"""
speechSettings = load_speech_settings()
self.speechRate = speechSettings.rate
self.speechPitch = speechSettings.pitch
# Create default settings if they don't exist
if not os.path.exists(self.configFile):
self.save_settings()
return
try:
self.config.read(self.configFile)
# Load volume settings
if 'Volume' in self.config:
self.volume = self.config.getint('Volume', 'level', fallback=50)
except Exception as e:
print(f"Error loading settings: {e}")
# If loading fails, we'll use default values
def save_settings(self):
"""Save menu-local settings to config file."""
# Ensure config directory exists
os.makedirs(self.configDir, exist_ok=True)
# Save volume settings
if 'Volume' not in self.config:
self.config['Volume'] = {}
self.config['Volume']['level'] = str(self.volume)
# Write to file
try:
with open(self.configFile, 'w') as f:
self.config.write(f)
except Exception as e:
print(f"Error saving settings: {e}")
def save_speech_settings(self):
"""Save speech settings to the shared Speech Dispatcher config."""
speechSettings = load_speech_settings()
speechSettings.rate = self.speechRate
speechSettings.pitch = self.speechPitch
save_speech_settings(speechSettings)
def increase_speech_rate(self):
"""Increase speech rate"""
self.speechRate = min(100, self.speechRate + 10) # Max is 100
if self.speechClient:
try:
self.speechClient.set_rate(self.speechRate)
self.speak(f"System speech rate: {self.speechRate}")
except Exception as e:
print(f"Error adjusting speech rate: {e}")
# Save the new setting
self.save_speech_settings()
def decrease_speech_rate(self):
"""Decrease speech rate"""
self.speechRate = max(-100, self.speechRate - 10) # Min is -100
if self.speechClient:
try:
self.speechClient.set_rate(self.speechRate)
self.speak(f"System speech rate: {self.speechRate}")
except Exception as e:
print(f"Error adjusting speech rate: {e}")
# Save the new setting
self.save_speech_settings()
def increase_speech_pitch(self):
"""Increase speech pitch"""
self.speechPitch = min(100, self.speechPitch + 10) # Max is 100
if self.speechClient:
try:
self.speechClient.set_pitch(self.speechPitch)
self.speak(f"System speech pitch: {self.speechPitch}")
except Exception as e:
print(f"Error adjusting speech pitch: {e}")
# Save the new setting
self.save_speech_settings()
def decrease_speech_pitch(self):
"""Decrease speech pitch"""
self.speechPitch = max(-100, self.speechPitch - 10) # Min is -100
if self.speechClient:
try:
self.speechClient.set_pitch(self.speechPitch)
self.speak(f"System speech pitch: {self.speechPitch}")
except Exception as e:
print(f"Error adjusting speech pitch: {e}")
# Save the new setting
self.save_speech_settings()
def get_current_volume(self):
"""Get the current system volume percentage"""
try:
result = subprocess.run(
['pactl', 'get-sink-volume', '@DEFAULT_SINK@'],
capture_output=True, text=True, check=True
)
output = result.stdout
# Extract percentage from output like "Volume: front-left: 27111 / 41% / -23.00 dB"
match = re.search(r'(\d+)%', output)
if match:
return int(match.group(1))
return self.volume # Return saved volume if parsing fails
except Exception as e:
print(f"Error getting volume: {e}")
return self.volume # Return saved volume if command fails
def set_volume(self, volumePercent):
"""Set the system volume to the specified percentage"""
# Ensure volume is between 0 and 150%
volumePercent = max(0, min(150, volumePercent))
try:
subprocess.run(
['pactl', 'set-sink-volume', '@DEFAULT_SINK@', f'{volumePercent}%'],
check=True
)
subprocess.run(
['sudo', '-n', '/usr/lib/stormux/stormux_alsa_state.sh', 'store'],
check=False
)
self.volume = volumePercent
self.save_settings()
return True
except Exception as e:
print(f"Error setting volume: {e}")
return False
def increase_volume(self):
"""Increase the system volume by 5%"""
currentVolume = self.get_current_volume()
newVolume = min(150, currentVolume + 5) # Max 150%
if self.set_volume(newVolume):
self.speak(f"Volume {newVolume} percent")
self.draw_menu()
def decrease_volume(self):
"""Decrease the system volume by 5%"""
currentVolume = self.get_current_volume()
newVolume = max(0, currentVolume - 5) # Min 0%
if self.set_volume(newVolume):
self.speak(f"Volume {newVolume} percent")
self.draw_menu()
def play_sound(self, soundName):
"""Play a sound effect using simpleaudio with ability to cancel previous sounds"""
try:
# Cancel any currently playing sound
self.stop_sound()
soundsDir = "/usr/share/sounds/stormux"
soundFile = os.path.join(soundsDir, f"{soundName}.wav")
# Check if the file exists
if not os.path.exists(soundFile):
print(f"Sound file not found: {soundFile}")
return False
# Play the sound and store the play_obj
waveObj = sa.WaveObject.from_wave_file(soundFile)
self.currentSound = waveObj.play()
return True
except Exception as e:
print(f"Error playing sound {soundName}: {e}")
return False
def stop_sound(self):
"""Stop any currently playing sound"""
if self.currentSound is not None and self.currentSound.is_playing():
self.currentSound.stop()
self.currentSound = None
def check_service_status(self, serviceName):
"""Check if a system service is active"""
try:
result = subprocess.run(
['systemctl', 'is-active', serviceName],
capture_output=True, text=True
)
return result.stdout.strip() == 'active'
except Exception as e:
print(f"Error checking {serviceName} status: {e}")
return False
def toggle_service(self, friendlyName):
"""Toggle a system service on/off"""
# Get the actual service name from the dictionary
serviceName = self.systemMenuServices.get(friendlyName)
if not serviceName:
print(f"Unknown service: {friendlyName}")
return
isActive = self.check_service_status(serviceName)
# Clean up curses before running the command with sudo
curses.endwin()
# Clean up speech client
if self.speechClient:
self.speechClient.close()
self.speechClient = None
action = "disable" if isActive else "enable"
result = None
try:
result = run_service_action(action, serviceName)
if result.returncode == 0:
print(f"{friendlyName} {action}d successfully")
else:
print(f"Error {action}ing {friendlyName}: see {SERVICE_ACTION_LOG}")
except Exception as e:
print(f"Error {action}ing {friendlyName}: {e}")
# Restart speech client
self.init_speech()
# Restart curses
self.stdscr = curses.initscr()
curses.noecho()
curses.cbreak()
self.stdscr.keypad(True)
if result is not None and result.returncode == 0:
self.speak(f"{friendlyName} {action}d")
# Update the menu with the new service status
self.update_service_menu_items()
# Update Bluetooth menu items if Bluetooth service was toggled
if friendlyName == "Bluetooth":
self.update_bluetooth_menu_items()
# Update Media menu items if DLNA Server was toggled
if friendlyName == "D L N A Server":
self.update_media_menu_items()
else:
self.speak(f"Could not {action} {friendlyName}. Check service action log.")
# Redraw the menu
self.draw_menu()
def toggle_screen(self, screenType):
"""Toggle between screen output and headless mode"""
try:
# Platform-specific handling
is_x86 = platform.machine() == "x86_64"
if screenType == "headless":
if is_x86:
# On x86, remove all 10-* files for best compatibility
os.system("sudo rm -f /etc/X11/xorg.conf.d/10-*.conf")
else:
# On Pi, only remove non-essential files, preserve fbdev
os.system("sudo rm -f /etc/X11/xorg.conf.d/10-screen.conf")
configFile = "/home/stormux/.local/files/10-headless.conf"
message = "Screen disabled."
else:
# For enabling screen, remove any existing configuration first
os.system("sudo rm -f /etc/X11/xorg.conf.d/10-*.conf")
configFile = "/home/stormux/.local/files/10-screen.conf"
message = "Screen enabled."
# Copy the configuration file
os.system(f"sudo cp {configFile} /etc/X11/xorg.conf.d/")
self.speak(message, interrupt=False)
except Exception as e:
message = f"Error changing screen configuration: {e}"
print(message)
self.speak(message)
def report_battery_status(self):
"""Report current battery status"""
try:
import pathlib
power_supply_dir = pathlib.Path('/sys/class/power_supply')
# Check if power supply directory exists
if not power_supply_dir.exists():
self.speak("No battery found.")
return
battery_found = False
battery_level = None
is_charging = False
# Look for battery
for item in power_supply_dir.iterdir():
type_file = item / 'type'
if type_file.exists():
try:
if type_file.read_text().strip() == 'Battery':
battery_found = True
# Get battery level
capacity_file = item / 'capacity'
if capacity_file.exists():
battery_level = int(capacity_file.read_text().strip())
# Check charging status
status_file = item / 'status'
if status_file.exists():
status = status_file.read_text().strip()
is_charging = status in ['Charging', 'Full']
break
except Exception:
continue
# Check for AC power
ac_connected = False
for item in power_supply_dir.iterdir():
type_file = item / 'type'
online_file = item / 'online'
if type_file.exists() and online_file.exists():
try:
device_type = type_file.read_text().strip()
if device_type in ['ADP', 'Mains', 'AC']:
online = int(online_file.read_text().strip())
if online == 1:
ac_connected = True
break
except Exception:
continue
if not battery_found:
self.speak("No battery found.")
return
# Build status message
if battery_level is not None:
message = f"Battery level: {battery_level} percent"
if is_charging:
message += ", charging"
elif ac_connected:
message += ", plugged in"
else:
message += ", on battery power"
# Add warning context for low levels
if battery_level <= 5:
message += ". Critical level!"
elif battery_level <= 10:
message += ". Low battery!"
else:
message = "Battery detected but unable to read level"
self.speak(message)
except Exception as e:
self.speak(f"Error reading battery status: {e}")
def handle_sigint(self, _signum, _frame):
"""Make Ctrl+C behave like Escape inside the menu."""
self.handle_escape_or_restart()
def update_service_menu_items(self):
"""Update all service-related menu items based on their current status"""
if self.sectionNames != ["Services"]:
return
self.menuSections["Services"] = self.build_services_menu_actions() + [("Back", self.restore_saved_menu)]
def build_service_menu_items(self):
"""Build service toggle items based on current service state."""
items = []
for friendlyName, serviceName in self.systemMenuServices.items():
isActive = self.check_service_status(serviceName)
startLabel = "Start" if friendlyName == "Fenrir Screen Reader" else "Enable"
stopLabel = "Stop" if friendlyName == "Fenrir Screen Reader" else "Disable"
if isActive:
items.append((f"{stopLabel} {friendlyName}", lambda fn=friendlyName: self.toggle_service(fn)))
else:
items.append((f"{startLabel} {friendlyName}", lambda fn=friendlyName: self.toggle_service(fn)))
return items
def build_services_menu_actions(self):
"""Build the system services submenu."""
return self.build_service_menu_items() + [
("Enable Screen", lambda: self.toggle_screen("screen")),
("Disable Screen", lambda: self.toggle_screen("headless")),
("Battery Status", self.report_battery_status),
]
def show_services_menu(self):
"""Show service toggles and related runtime controls."""
self.show_action_menu("Services", self.build_services_menu_actions())
def show_settings_menu(self):
"""Show system settings and configuration actions."""
actions = []
if platform.machine() == "x86_64" and not os.path.exists("/home/stormux/.baremetal"):
actions.append(("Install System to Hard Drive", "GAME='Install to Disk' /home/stormux/.clirc"))
actions.extend([
("Internet Configuration", "GAME=\"Network Configuration\" /home/stormux/.clirc"),
("Set System Speech Settings", "/home/stormux/.local/bin/speechd_rate.py"),
("Set Default Voice", "/home/stormux/.local/bin/set-voice.py"),
("Set Timezone", "GAME='Set Timezone' /home/stormux/.clirc"),
("Download Files", "/home/stormux/.local/download_server.py"),
("Resize to fill empty space on disk", "sudo growpartfs $(df --output='source' / | tail -1)"),
])
self.show_action_menu("Settings", actions)
def show_power_menu(self):
"""Show power and update actions."""
self.show_action_menu("Power", [
("System Update", "/home/stormux/.local/bin/system-update.sh"),
("Restart: Can Take Several Minutes", "sudo reboot"),
("Power Off: Wait 2 Minutes Before Disconnecting Power", "sudo poweroff"),
])
def install_and_launch(self, executable_name, launch_mode="gui"):
"""Launch application via xinitrc (which handles installation if needed)"""
try:
# Launch the application - xinitrc will handle installation
if launch_mode == "gui":
command = f"GAME='{executable_name}' startx"
else: # cli mode
command = f"GAME='{executable_name}' /home/stormux/.clirc"
# Use the existing execute_current_item infrastructure by temporarily setting command
# Save current state
original_sections = self.sectionNames.copy()
original_current_section = self.currentSection
original_items = {}
for section in self.menuSections:
original_items[section] = self.menuSections[section].copy()
# Create temporary item to execute
temp_section = "temp_install_launch"
self.add_section(temp_section)
self.add_item(temp_section, f"Launch {executable_name}", command)
# Set to the temporary section and item
self.currentSection = len(self.sectionNames) - 1
self.currentItemIndices[temp_section] = 0
# Execute the command using existing infrastructure
self.execute_current_item()
# Restore original state
self.sectionNames = original_sections
self.currentSection = original_current_section
self.menuSections = original_items
except Exception as e:
error_msg = f"Error launching {executable_name}: {e}"
self.speak(error_msg, interrupt=False)
def update_bluetooth_menu_items(self):
"""Update Bluetooth-related menu items in Accessories section"""
if "Accessories" in self.menuSections:
# Remove any existing Bluetooth-related items
self.menuSections["Accessories"] = [item for item in self.menuSections["Accessories"]
if "Bluetooth" not in item[0]]
# Add Bluetooth management item only if Bluetooth is enabled
if self.check_service_status('bluetooth.service'):
self.add_item("Accessories", "Manage Bluetooth Devices",
"GAME=blueman-manager startx")
def update_media_menu_items(self):
"""Update Media menu items including DLNA server toggle"""
if "Media" in self.menuSections:
# Remove any existing DLNA-related items
self.menuSections["Media"] = [item for item in self.menuSections["Media"]
if "D L N A" not in item[0]]
# Check DLNA server status and add appropriate toggle
isActive = self.check_service_status('minidlna.service')
if isActive:
self.add_item("Media", "Disable D L N A Server",
lambda: self.toggle_service('D L N A Server'))
else:
self.add_item("Media", "Enable D L N A Server",
lambda: self.toggle_service('D L N A Server'))
def scan_documentation_files(self):
"""Scan Documents directory for .md files and add them to help menu"""
docs_dir = os.path.expanduser("~/Documents")
if not os.path.exists(docs_dir):
return
try:
# Get all .md files in Documents directory
md_files = []
for file in os.listdir(docs_dir):
if file.endswith('.md'):
file_path = os.path.join(docs_dir, file)
if os.path.isfile(file_path):
# Create a friendly display name from filename
# Remove .md extension and replace underscores with spaces
display_name = file[:-3].replace('_', ' ').title()
md_files.append((display_name, file))
# Sort files alphabetically by display name
md_files.sort(key=lambda x: x[0])
# Add each markdown file to the help section
for display_name, filename in md_files:
file_path = f"~/Documents/{filename}"
self.add_item("Help and Documentation", display_name, f"GAME={file_path} /home/stormux/.clirc")
except Exception as e:
print(f"Error scanning documentation files: {e}")
def add_section(self, sectionName):
"""Add a new section to the menu"""
if sectionName not in self.menuSections:
self.menuSections[sectionName] = []
self.sectionNames.append(sectionName)
self.currentItemIndices[sectionName] = 0
def add_item(self, sectionName, name, command):
"""Add a menu item to a specific section"""
# Create section if it doesn't exist
if sectionName not in self.menuSections:
self.add_section(sectionName)
self.menuSections[sectionName].append((name, command))
def save_menu_state(self):
"""Save the current menu state so a temporary submenu can restore it."""
menuState = {
"sectionNames": self.sectionNames.copy(),
"currentSection": self.currentSection,
"currentItemIndices": self.currentItemIndices.copy(),
"menuSections": {
sectionName: items.copy()
for sectionName, items in self.menuSections.items()
},
}
self.menuStateStack.append(menuState)
self.savedMenuState = menuState
def restore_saved_menu(self):
"""Restore the menu state saved before entering a temporary submenu."""
if not self.menuStateStack:
return
menuState = self.menuStateStack.pop()
self.sectionNames = menuState["sectionNames"]
self.currentSection = menuState["currentSection"]
self.currentItemIndices = menuState["currentItemIndices"]
self.menuSections = menuState["menuSections"]
self.savedMenuState = self.menuStateStack[-1] if self.menuStateStack else None
if not self.menuStateStack:
self.activeGameEntry = None
self.draw_menu()
self.announce_current_section()
time.sleep(0.5)
self.announce_current_item(interrupt=False)
def show_action_menu(self, sectionName, actions):
"""Replace the current menu with a temporary action submenu."""
self.save_menu_state()
self.sectionNames = [sectionName]
self.currentSection = 0
self.currentItemIndices = {sectionName: 0}
self.menuSections = {sectionName: actions + [("Back", self.restore_saved_menu)]}
self.draw_menu()
self.announce_current_item()
def show_game_actions(self, gameEntry):
"""Show available actions for a game."""
self.activeGameEntry = gameEntry
self.show_action_menu(gameEntry.name, build_game_actions(gameEntry))
def refresh_game_entry(self, gameEntry):
"""Return a fresh game entry with current launcher state."""
launchRoot = gameEntry.launchScript.parent if gameEntry.launchScript else Path.home() / ".local" / ".launch"
if gameEntry.installScript.parent.parent.name == ".install":
launchRoot = gameEntry.installScript.parent.parent.parent / ".launch"
launchScript = find_launch_script(launchRoot, gameEntry.section, gameEntry.name)
return GameEntry(gameEntry.section, gameEntry.name, gameEntry.installScript, launchScript)
def refresh_game_action_menu(self):
"""Refresh the current game action submenu after install state changes."""
if self.activeGameEntry is None:
return
self.activeGameEntry = self.refresh_game_entry(self.activeGameEntry)
sectionName = self.activeGameEntry.name
self.sectionNames = [sectionName]
self.currentSection = 0
self.currentItemIndices = {sectionName: 0}
self.menuSections = {sectionName: build_game_actions(self.activeGameEntry) + [("Back", self.restore_saved_menu)]}
self.draw_menu()
self.announce_current_item()
def handle_escape(self):
"""Handle Escape based on whether a temporary submenu is active."""
if not self.menuStateStack:
return False
self.restore_saved_menu()
return True
def handle_escape_or_restart(self):
"""Go back from submenus, otherwise restart the menu."""
if self.handle_escape():
return
try:
# Announce restart first while speech still works
self.speak("Restarting menu")
# Wait for speech to complete
time.sleep(1)
# Now clean up resources
self.cleanup()
# Try to kill the user's speech-dispatcher if needed
subprocess.run(["killall", "speech-dispatcher"], check=False)
# Restart the application
os.execv(sys.argv[0], sys.argv)
except Exception as e:
print(f"Error during restart: {e}")
# If restart fails, we need to recover the UI
self.stdscr = curses.initscr()
curses.noecho()
curses.cbreak()
self.stdscr.keypad(True)
self.draw_menu()
def jump_to_item_by_letter(self, letter):
"""Move to the next item in the current menu starting with letter."""
if not letter or not self.sectionNames:
return False
letter = letter.lower()
items = self.get_current_items()
if not items:
return False
currentIndex = self.get_current_item_index()
for offset in range(1, len(items) + 1):
candidateIndex = (currentIndex + offset) % len(items)
itemName = items[candidateIndex][0].strip().lower()
if itemName.startswith(letter):
self.set_current_item_index(candidateIndex)
return True
return False
def jump_to_section_by_letter(self, letter):
"""Move to the next top-level section starting with letter."""
if not letter or self.menuStateStack or not self.sectionNames:
return False
letter = letter.lower()
for offset in range(1, len(self.sectionNames) + 1):
candidateIndex = (self.currentSection + offset) % len(self.sectionNames)
sectionName = self.sectionNames[candidateIndex].strip().lower()
if sectionName.startswith(letter):
self.currentSection = candidateIndex
return True
return False
def handle_control_shortcut(self, key):
"""Handle reliable control-key shortcuts before printable navigation."""
if key == 2: # Ctrl+B
self.report_battery_status()
return True
if key == 8: # Ctrl+H
self.speak_help()
return True
return False
def handle_letter_navigation(self, key):
"""Handle printable letter navigation."""
if key < 0 or key > 255:
return False
keyChar = chr(key)
if not keyChar.isalpha():
return False
moved = False
if keyChar.isupper() and not self.menuStateStack:
moved = self.jump_to_section_by_letter(keyChar)
if moved:
self.draw_menu()
self.play_sound("menu_category")
self.announce_current_section()
time.sleep(0.5)
self.announce_current_item(interrupt=False)
else:
moved = self.jump_to_item_by_letter(keyChar)
if moved:
self.draw_menu()
self.play_sound("menu_move")
self.announce_current_item()
return moved
def speak(self, text, interrupt=True):
"""Speak the given text with option to interrupt existing speech"""
if self.speechClient is None:
return
try:
if interrupt:
self.stop_speech()
self.speechClient.speak(text)
except Exception as e:
# If speech fails, try to reinitialize and try once more
try:
self.init_speech()
if self.speechClient:
self.speechClient.speak(text)
except:
# If reinitializing fails, just give up silently
pass
def stop_speech(self):
"""Stop any ongoing speech"""
if self.speechClient is None:
return
try:
self.speechClient.cancel()
except Exception as e:
# If cancel fails, try to reinitialize
self.init_speech()
def get_current_items(self):
"""Get items from the current section"""
currentSectionName = self.sectionNames[self.currentSection]
return self.menuSections[currentSectionName]
def get_current_item_index(self):
"""Get the current item index in the current section"""
currentSectionName = self.sectionNames[self.currentSection]
return self.currentItemIndices[currentSectionName]
def set_current_item_index(self, index):
"""Set the current item index for the current section"""
currentSectionName = self.sectionNames[self.currentSection]
self.currentItemIndices[currentSectionName] = index
def announce_current_section(self, interrupt=True):
"""Announce the currently selected section"""
if 0 <= self.currentSection < len(self.sectionNames):
sectionName = self.sectionNames[self.currentSection]
self.speak(sectionName, interrupt=interrupt)
def announce_current_item(self, interrupt=True):
"""Announce the currently selected menu item"""
if len(self.sectionNames) > 0:
items = self.get_current_items()
index = self.get_current_item_index()
if 0 <= index < len(items):
name = items[index][0]
self.speak(name, interrupt=interrupt)
def execute_current_item(self):
"""Execute the currently selected menu item"""
if len(self.sectionNames) > 0:
items = self.get_current_items()
index = self.get_current_item_index()
if 0 <= index < len(items):
name, command = items[index]
# Check if command is a function (for service toggles and other functions)
if callable(command):
if not getattr(command, "opensSubmenu", False):
self.speak(f"Launching {name}")
command()
return
# Announce we're launching the program
self.speak(f"Launching {name}")
# Save current terminal state before any changes
savedTerminalState = None
try:
result = subprocess.run(['stty', '-g'], capture_output=True, text=True, check=True)
savedTerminalState = result.stdout.strip()
except Exception as e:
print(f"Warning: Could not save terminal state: {e}")
# Cleanup before running the command
self.stop_speech()
# Close speech client
if self.speechClient:
try:
self.speechClient.close()
except:
pass
self.speechClient = None
# Curses cleanup - be thorough
if self.stdscr:
try:
curses.nocbreak()
self.stdscr.keypad(False)
curses.echo()
curses.endwin()
except:
pass
self.stdscr = None
# Complete terminal reset to ensure clean state for games
try:
subprocess.run(['reset'], check=False, stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except:
pass
# Ensure we're in a new process group to avoid signal conflicts
# Execute the command with proper terminal isolation
try:
# For applications that need dialog/curses, we need to ensure they
# become the controlling process with full terminal access
# Use os.system with proper terminal restoration as it gives
# the subprocess complete control over the terminal
# First, save our current process group
original_pgrp = os.getpgrp()
# Execute using os.system which gives the subprocess complete terminal control
exit_code = os.system(command)
# Try to regain terminal control after the subprocess exits
try:
# Get the current terminal file descriptor
tty_fd = os.open('/dev/tty', os.O_RDWR)
# Try to make our process group the foreground group again
os.tcsetpgrp(tty_fd, original_pgrp)
os.close(tty_fd)
except:
# If we can't regain control, that's okay
pass
except Exception as e:
print(f"Error launching {name}: {e}")
# Another terminal reset after game exits to clean up any changes
try:
subprocess.run(['reset'], check=False, stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except:
pass
# Restore the saved terminal state if we have it
if savedTerminalState:
try:
subprocess.run(['stty', savedTerminalState], check=False)
except Exception as e:
print(f"Warning: Could not restore terminal state: {e}")
# Brief delay to let terminal stabilize
time.sleep(0.5)
# Reinitialize speech and display
self.init_speech()
# Reinitialize curses with better error handling
try:
self.stdscr = curses.initscr()
curses.noecho()
curses.cbreak()
self.stdscr.keypad(True)
self.stdscr.clear()
self.stdscr.refresh()
except Exception as e:
print(f"Error reinitializing display: {e}")
# Try a more aggressive recovery
try:
curses.endwin()
time.sleep(1)
self.stdscr = curses.initscr()
curses.noecho()
curses.cbreak()
self.stdscr.keypad(True)
self.stdscr.clear()
self.stdscr.refresh()
except:
print("Could not recover display. Please restart the menu.")
sys.exit(1)
# Update service menu items and redraw
self.update_service_menu_items()
if self.activeGameEntry is not None:
self.refresh_game_action_menu()
else:
self.draw_menu()
# Let the user know the menu is activated
self.speak("Game menu activated")
def speak_help(self):
"""Speak help information"""
helpText = """
Navigation controls:
Up arrow: Previous menu item.
Down arrow: Next menu item.
Left arrow: Previous section.
Right arrow: Next section.
Lowercase letters: Jump to menu items.
Uppercase letters: Jump to categories from the main menu.
Enter: Launch selected item.
Control H: Hear these instructions again.
Control B: Report battery status.
Left bracket: Decrease system speech rate.
Right bracket: Increase system speech rate.
Left brace: Decrease system speech pitch.
Right brace: Increase system speech pitch.
9 key: Decrease volume.
0 key: Increase volume.
Escape: Go back from submenus or refresh the main menu.
Q: Exit the menu.
Control C behaves like Escape.
Any key will interrupt speech.
"""
self.speak(helpText)
def draw_menu(self):
"""Draw the menu on the screen"""
self.stdscr.clear()
h, w = self.stdscr.getmaxyx()
# Draw title
title = f" {self.title} "
x = max(0, w // 2 - len(title) // 2)
self.stdscr.addstr(1, x, title, curses.A_BOLD)
# Draw help line
helpText = "Navigate | Letters: Jump | Enter: Select | Ctrl+H: Help | Ctrl+B: Battery | Esc: Back"
x = max(0, w // 2 - len(helpText) // 2)
self.stdscr.addstr(3, x, helpText)
# Draw current section
if len(self.sectionNames) > 0:
currentSectionName = self.sectionNames[self.currentSection]
sectionText = f"== {currentSectionName} =="
x = max(0, w // 2 - len(sectionText) // 2)
self.stdscr.addstr(5, x, sectionText, curses.A_BOLD)
# Draw menu items for current section
items = self.get_current_items()
currentItemIndex = self.get_current_item_index()
for i, (name, _) in enumerate(items):
y = i + 7 # Start items 2 lines below section header
if y < h - 1: # Ensure we don't draw outside the window
# Highlight the selected item
if i == currentItemIndex:
text = f" > {name} "
attr = curses.A_REVERSE
else:
text = f" {name} "
attr = curses.A_NORMAL
x = max(0, w // 2 - len(text) // 2)
self.stdscr.addstr(y, x, text, attr)
# Draw speech rate indicator
rateText = f"System Speech Rate: {self.speechRate}"
self.stdscr.addstr(h-2, 2, rateText)
# Draw speech pitch indicator
pitchText = f"System Pitch: {self.speechPitch}"
pitchX = max(0, w // 2 - len(pitchText) // 2)
self.stdscr.addstr(h-2, pitchX, pitchText)
# Draw volume indicator
volumeText = f"Volume: {self.get_current_volume()}%"
self.stdscr.addstr(h-2, w-len(volumeText)-2, volumeText)
self.stdscr.refresh()
def cleanup(self):
"""Clean up resources before exiting"""
# Stop any speech
self.stop_speech()
# Close speech client
if self.speechClient:
try:
self.speechClient.close()
except:
pass
self.speechClient = None
# Restore terminal settings
curses.nocbreak()
self.stdscr.keypad(False)
curses.echo()
curses.endwin()
def run(self):
"""Run the menu system"""
if not self.sectionNames:
print("Menu is empty. Exiting.")
return
try:
# Initialize curses
self.stdscr = curses.initscr()
curses.noecho() # Don't echo keypresses
curses.cbreak() # React to keys instantly
self.stdscr.keypad(True) # Enable special keys
signal.signal(signal.SIGINT, self.handle_sigint)
# Update all service menu items based on current status
self.update_service_menu_items()
# Update Bluetooth menu items based on current status
self.update_bluetooth_menu_items()
# Update Media menu items based on current status
self.update_media_menu_items()
# Initial draw
self.draw_menu()
# Welcome message - don't interrupt this initial speech
boot_source = "internal disk" if self.is_installed else "USB drive"
self.speak(f"Welcome to {self.title}, booted from {boot_source}. Use left and right arrows to navigate sections. Up and down for items. Press Control H for help.")
# Wait for initial speech to finish before announcing section
time.sleep(1)
# Announce initial section and item without interrupting welcome speech
self.announce_current_section(interrupt=False)
time.sleep(0.5) # Wait before announcing first item
self.announce_current_item(interrupt=False)
# Main loop
while True:
key = self.stdscr.getch()
# Stop any speech when a key is pressed
self.stop_speech()
# Handle navigation
if key == curses.KEY_UP:
# Move to previous item in current section
items = self.get_current_items()
if items:
prevIndex = self.get_current_item_index()
self.set_current_item_index((prevIndex - 1) % len(items))
newIndex = self.get_current_item_index()
self.draw_menu()
# Play sound if selection actually changed
if newIndex != prevIndex:
self.play_sound("menu_move")
self.announce_current_item()
elif key == curses.KEY_DOWN:
# Move to next item in current section
items = self.get_current_items()
if items:
prevIndex = self.get_current_item_index()
self.set_current_item_index((prevIndex + 1) % len(items))
newIndex = self.get_current_item_index()
self.draw_menu()
# Play sound if selection actually changed
if newIndex != prevIndex:
self.play_sound("menu_move")
self.announce_current_item()
elif key == curses.KEY_LEFT:
# Move to previous section
if self.sectionNames:
prevSection = self.currentSection
self.currentSection = (self.currentSection - 1) % len(self.sectionNames)
self.draw_menu()
# Play category change sound if section actually changed
if prevSection != self.currentSection:
self.play_sound("menu_category")
# Announce section and current item without interruption between them
self.announce_current_section()
time.sleep(0.5) # Brief pause between section and item announcement
self.announce_current_item(interrupt=False) # Don't interrupt the section announcement
elif key == curses.KEY_RIGHT:
# Move to next section
if self.sectionNames:
prevSection = self.currentSection
self.currentSection = (self.currentSection + 1) % len(self.sectionNames)
self.draw_menu()
# Play category change sound if section actually changed
if prevSection != self.currentSection:
self.play_sound("menu_category")
# Announce section and current item without interruption between them
self.announce_current_section()
time.sleep(0.5) # Brief pause between section and item announcement
self.announce_current_item(interrupt=False) # Don't interrupt the section announcement
elif key == curses.KEY_ENTER or key == 10 or key == 13: # Enter key
self.play_sound("menu_select")
self.execute_current_item()
elif key == ord('['): # Decrease speech rate
self.decrease_speech_rate()
self.draw_menu()
elif key == ord(']'): # Increase speech rate
self.increase_speech_rate()
self.draw_menu()
elif key == ord('{'): # Decrease speech pitch
self.decrease_speech_pitch()
self.draw_menu()
elif key == ord('}'): # Increase speech pitch
self.increase_speech_pitch()
self.draw_menu()
elif key == ord('9'): # Decrease volume
self.decrease_volume()
elif key == ord('0'): # Increase volume
self.increase_volume()
elif key == ord('q') or key == ord('Q'): # Quit
self.speak("Exiting menu")
time.sleep(0.5)
break
elif self.handle_control_shortcut(key):
pass
elif self.handle_letter_navigation(key):
pass
elif key == 27: # Esc key - go back from submenus, otherwise restart the application
self.handle_escape_or_restart()
except Exception as e:
# End curses in case of error
curses.endwin()
print(f"An error occurred: {e}")
finally:
# Clean up
self.cleanup()
def configure_system_menu(menu):
"""Add the nested System menu."""
menu.add_section("System")
menu.add_item("System", "Settings", submenu_command(menu.show_settings_menu))
menu.add_item("System", "Services", submenu_command(menu.show_services_menu))
menu.add_item("System", "Power", submenu_command(menu.show_power_menu))
# Example usage
if __name__ == "__main__":
# Create the menu with sections
menu = VoicedMenu(title="Stormux Gaming Menu")
for gameEntry in discover_game_entries():
menu.add_item(
gameEntry.section,
gameEntry.name,
submenu_command(lambda entry=gameEntry: menu.show_game_actions(entry)),
)
menu.add_section("Emulators")
menu.add_item("Emulators", "Apple 2e", "/home/stormux/.local/bin/apple_2e.py")
menu.add_item("Emulators", "Game Console Menu", "/home/stormux/.local/bin/rom_launcher.py")
# Add help and documentation section
menu.add_section("Help and Documentation")
# Dynamically scan and add all .md files from Documents directory
menu.scan_documentation_files()
# Add the IRC help item
menu.add_item("Help and Documentation", "Get help on IRC", "GAME=IRC /home/stormux/.clirc")
# Add media section
menu.add_section("Media")
menu.add_item("Media", "Music Player", "/home/stormux/.local/bin/music_player.py")
menu.add_item("Media", "BookStorm", "GAME=BookStorm startx")
menu.add_item("Media", "Upload Files", "/home/stormux/.local/upload_server/uploader.py")
# Add accessories section
menu.add_section("Accessories")
menu.add_item("Accessories", "Local IP Address", "/home/stormux/.local/bin/ip_info.py local")
menu.add_item("Accessories", "Remote IP Address", "/home/stormux/.local/bin/ip_info.py remote")
menu.add_item("Accessories", "Sound and Volume", "/home/stormux/.local/bin/audio_manager.py")
menu.add_item("Accessories", "Web Browser", "GAME=Brave startx")
menu.add_item("Accessories", "LibreOffice", lambda: menu.install_and_launch("libreoffice", "gui"))
menu.add_item("Accessories", "Thunderbird", lambda: menu.install_and_launch("thunderbird", "gui"))
configure_system_menu(menu)
# Run the menu
menu.run()