Compare commits
5 Commits
fe5e2c065e
...
testing
Author | SHA1 | Date | |
---|---|---|---|
8c668cc0cc | |||
579bf0f0f0 | |||
d1bad818cd | |||
ae4c418323 | |||
0c116adaf2 |
@ -99,6 +99,13 @@ class command:
|
||||
"Progress detector checking: '" + text + "'", debug.DebugLevel.INFO
|
||||
)
|
||||
|
||||
# Filter out URLs to prevent false positives
|
||||
if self.contains_url(text):
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
"Skipping progress detection - text contains URL", debug.DebugLevel.INFO
|
||||
)
|
||||
return
|
||||
|
||||
# Note: Auto-disable on 100% completion removed to respect user
|
||||
# settings
|
||||
|
||||
@ -147,8 +154,16 @@ class command:
|
||||
curl_match = re.search(
|
||||
r"(\d+\s+\d+\s+\d+\s+\d+.*?(?:k|M|G)?.*?--:--:--|Speed)", text
|
||||
)
|
||||
# Pattern 1e: General transfer progress (size, rate, time patterns)
|
||||
transfer_match = re.search(
|
||||
r"\d+\s+\d+[kMGT]?\s+\d+\s+\d+[kMGT]?.*?\d+\.\d+[kMGT].*?\d+:\d+:\d+", text
|
||||
)
|
||||
# Pattern 1f: Pacman-style transfer progress (flexible size/speed/time)
|
||||
pacman_match = re.search(
|
||||
r"\d+(?:\.\d+)?\s+[kKmMgGtT]iB\s+\d+(?:\.\d+)?\s+[kKmMgGtT]iB/s\s+\d+:\d+", text
|
||||
)
|
||||
|
||||
if time_match or token_match or dd_match or curl_match:
|
||||
if time_match or token_match or dd_match or curl_match or transfer_match or pacman_match:
|
||||
# For non-percentage progress, use a single activity beep every 2
|
||||
# seconds
|
||||
if (
|
||||
@ -183,7 +198,7 @@ class command:
|
||||
|
||||
# Pattern 3: Progress bars ([#### ], [====> ], etc.)
|
||||
# Improved pattern to avoid matching IRC channels like [#channel]
|
||||
bar_match = re.search(r"\[([#=\-\*]+)([\s\.]*)\]", text)
|
||||
bar_match = re.search(r"\[([#=\*]+)([\s\.\-]*)\]", text)
|
||||
if bar_match:
|
||||
filled = len(bar_match.group(1))
|
||||
unfilled = len(bar_match.group(2))
|
||||
@ -220,6 +235,30 @@ class command:
|
||||
):
|
||||
self.play_activity_beep()
|
||||
self.env["commandBuffer"]["lastProgressTime"] = current_time
|
||||
return
|
||||
|
||||
# Pattern 5: Braille progress indicators
|
||||
braille_match = re.search(r'[⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏⡿⣟⣯⣷⣾⣽⣻⢿]', text)
|
||||
if braille_match:
|
||||
if current_time - self.env["commandBuffer"]["lastProgressTime"] >= 1.0:
|
||||
self.play_activity_beep()
|
||||
self.env["commandBuffer"]["lastProgressTime"] = current_time
|
||||
return
|
||||
|
||||
# Pattern 6: Moon phase progress indicators
|
||||
moon_match = re.search(r'[🌑🌒🌓🌔🌕🌖🌗🌘]', text)
|
||||
if moon_match:
|
||||
moon_phases = {
|
||||
'🌑': 0, '🌒': 12.5, '🌓': 25, '🌔': 37.5,
|
||||
'🌕': 50, '🌖': 62.5, '🌗': 75, '🌘': 87.5
|
||||
}
|
||||
moon_char = moon_match.group(0)
|
||||
if moon_char in moon_phases:
|
||||
percentage = moon_phases[moon_char]
|
||||
if percentage != self.env["commandBuffer"]["lastProgressValue"]:
|
||||
self.play_progress_tone(percentage)
|
||||
self.env["commandBuffer"]["lastProgressValue"] = percentage
|
||||
return
|
||||
|
||||
def play_progress_tone(self, percentage):
|
||||
# Map 0-100% to 400-1200Hz frequency range
|
||||
@ -350,5 +389,22 @@ class command:
|
||||
# If anything fails, assume it's not a prompt to be safe
|
||||
return False
|
||||
|
||||
def contains_url(self, text):
|
||||
"""Check if text contains URLs that might cause false progress detection"""
|
||||
import re
|
||||
|
||||
# Common URL patterns that might contain progress-like patterns
|
||||
url_patterns = [
|
||||
r"https?://[^\s]+", # http:// or https:// URLs
|
||||
r"ftp://[^\s]+", # ftp:// URLs
|
||||
r"www\.[^\s]+", # www. domains
|
||||
r"[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}[/\w.-]*", # domain.com/path patterns
|
||||
]
|
||||
|
||||
for pattern in url_patterns:
|
||||
if re.search(pattern, text, re.IGNORECASE):
|
||||
return True
|
||||
return False
|
||||
|
||||
def set_callback(self, callback):
|
||||
pass
|
||||
|
@ -0,0 +1 @@
|
||||
# Emoji VMenu category
|
@ -0,0 +1 @@
|
||||
# Flags emoji subcategory
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🇨🇦"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add Canada flag emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added Canada flag to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🇬🇧"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add UK flag emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added UK flag to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🇺🇸"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add USA flag emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added USA flag to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1 @@
|
||||
# Food emoji subcategory
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🍺"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Beer emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added beer to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "☕"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add coffee emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added coffee to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🍩"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Donut emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added donut to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🍔"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add hamburger emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added hamburger to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🍕"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add pizza emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added pizza to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🌮"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Taco emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added taco to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1 @@
|
||||
# Holidays emoji subcategory
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🦇"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add bat emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added bat to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🐰"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add bunny emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added bunny to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🎄"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add Christmas tree emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added Christmas tree to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🥚"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add Easter egg emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added Easter egg to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🎆"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add fireworks emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added fireworks to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "👻"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add ghost emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added ghost to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🎁"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add gift emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added gift to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🎃"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add jack o'lantern emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added jack o'lantern to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🎅"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add Santa emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added Santa to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "☘️"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add shamrock emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added shamrock to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "💀"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add skull emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added skull to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "⛄"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add snowman emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added snowman to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🕷"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add spider emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added spider to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🦃"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add turkey emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added turkey to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1 @@
|
||||
# Nature emoji subcategory
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🐱"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Cat emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added cat to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🐶"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Dog emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added dog to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🌙"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add moon emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added moon to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🌈"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Rainbow emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added rainbow to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "☀️"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add sun emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added sun to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🌳"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add tree emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added tree to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1 @@
|
||||
# People emoji subcategory
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "😠"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Angry face emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added angry face to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "😎"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Cool face emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added cool face to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "😭"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Crying face emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added crying face to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "😈"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Devil face emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added devil face to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "😂"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add laughing face emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added laughing face to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "💩"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Poop emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added poop to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "😢"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Sad face emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added sad face to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "😱"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Shocked face emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added shocked face to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "😊"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add smiling face emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added smiling face to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "👍"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add thumbs up emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added thumbs up to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "😉"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add winking face emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added winking face to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1 @@
|
||||
# Symbols emoji subcategory
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "✅"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add checkmark emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added checkmark to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🔥"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Fire emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added fire to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "❤️"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add heart emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added heart to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "⚡"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Lightning emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added lightning to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "✌️"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Peace sign emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added peace sign to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "🤘"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Sign of the horns emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added sign of the horns to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "☠️"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Skull and crossbones emoji"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added skull and crossbones to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -0,0 +1,22 @@
|
||||
class command():
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.emoji = "⭐"
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
|
||||
def setCallback(self, callback):
|
||||
pass
|
||||
|
||||
def getDescription(self):
|
||||
return "Add star emoji to clipboard"
|
||||
|
||||
def run(self):
|
||||
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
|
||||
"clipboardHistory", self.emoji
|
||||
)
|
||||
self.env["runtime"]["OutputManager"].present_text(
|
||||
"Added star to clipboard",
|
||||
interrupt=False, flush=False
|
||||
)
|
@ -4,5 +4,6 @@
|
||||
# Fenrir TTY screen reader
|
||||
# By Chrys, Storm Dragon, and contributors.
|
||||
|
||||
version = "2025.07.09"
|
||||
code_name = "master"
|
||||
version = "2025.07.16"
|
||||
codeName = "testing"
|
||||
code_name = "testing"
|
||||
|
@ -39,7 +39,177 @@ class driver(inputDriver):
|
||||
|
||||
Args:
|
||||
environment: Fenrir environment dictionary
|
||||
|
||||
Returns:
|
||||
bool: True if initialization successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
if environment is None:
|
||||
raise ValueError("Environment cannot be None")
|
||||
|
||||
self.env = environment
|
||||
|
||||
# Validate required managers are available
|
||||
if "runtime" not in self.env:
|
||||
raise ValueError("Runtime environment missing")
|
||||
if "InputManager" not in self.env["runtime"]:
|
||||
raise ValueError("InputManager not available")
|
||||
|
||||
self.env["runtime"]["InputManager"].set_shortcut_type("BYTE")
|
||||
self._is_initialized = True
|
||||
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
"PTY inputDriver: Initialized with byte-based shortcuts",
|
||||
debug.DebugLevel.INFO
|
||||
)
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
# Log error if possible, otherwise fallback to print
|
||||
try:
|
||||
if hasattr(self, 'env') and self.env and "runtime" in self.env:
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY inputDriver: Initialization failed: {e}",
|
||||
debug.DebugLevel.ERROR
|
||||
)
|
||||
else:
|
||||
print(f"PTY inputDriver initialization error: {e}")
|
||||
except:
|
||||
print(f"PTY inputDriver initialization error: {e}")
|
||||
|
||||
self._is_initialized = False
|
||||
return False
|
||||
|
||||
def shutdown(self):
|
||||
"""Shutdown the PTY input driver.
|
||||
|
||||
Performs cleanup operations when the driver is being stopped.
|
||||
For PTY driver, this involves cleaning up any resources and
|
||||
logging the shutdown.
|
||||
"""
|
||||
if not self._is_initialized:
|
||||
return
|
||||
|
||||
try:
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
"PTY inputDriver: Shutting down",
|
||||
debug.DebugLevel.INFO
|
||||
)
|
||||
except Exception as e:
|
||||
# Fallback logging if debug manager is unavailable
|
||||
print(f"PTY inputDriver shutdown error: {e}")
|
||||
finally:
|
||||
self._is_initialized = False
|
||||
|
||||
def get_input_event(self):
|
||||
"""Get input event from PTY.
|
||||
|
||||
For PTY driver, input events are handled through the byte-based
|
||||
shortcut system rather than direct device events. This method
|
||||
returns None as PTY input is processed through the screen driver
|
||||
and InputManager's byte processing.
|
||||
|
||||
Returns:
|
||||
None: PTY driver uses byte-based processing, not event-based
|
||||
"""
|
||||
return None
|
||||
|
||||
def is_device_connected(self):
|
||||
"""Check if PTY input device is connected.
|
||||
|
||||
For PTY driver, the "device" is the terminal interface itself,
|
||||
which is considered connected if the driver is initialized.
|
||||
|
||||
Returns:
|
||||
bool: True if driver is initialized, False otherwise
|
||||
"""
|
||||
return self._is_initialized
|
||||
|
||||
def get_device_name(self):
|
||||
"""Get the name of the PTY input device.
|
||||
|
||||
Returns:
|
||||
str: Human-readable name of the PTY input device
|
||||
"""
|
||||
return "PTY (Pseudo-terminal) Input"
|
||||
|
||||
def grab_devices(self, grab=True):
|
||||
"""Grab or release input devices.
|
||||
|
||||
For PTY driver, device grabbing is not applicable since input
|
||||
is processed through terminal emulation rather than direct
|
||||
device access.
|
||||
|
||||
Args:
|
||||
grab (bool): Whether to grab (True) or release (False) devices
|
||||
|
||||
Returns:
|
||||
bool: Always returns True for PTY driver (no-op success)
|
||||
"""
|
||||
if not self._is_initialized:
|
||||
return False
|
||||
|
||||
action = "grab" if grab else "release"
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY inputDriver: {action} devices (no-op for PTY)",
|
||||
debug.DebugLevel.INFO
|
||||
)
|
||||
return True
|
||||
|
||||
def has_device_detection(self):
|
||||
"""Check if driver supports device detection.
|
||||
|
||||
PTY driver does not support dynamic device detection since
|
||||
it operates on the terminal interface directly.
|
||||
|
||||
Returns:
|
||||
bool: Always False for PTY driver
|
||||
"""
|
||||
return False
|
||||
|
||||
def get_device_list(self):
|
||||
"""Get list of available input devices.
|
||||
|
||||
For PTY driver, there is only one logical device - the terminal
|
||||
interface itself.
|
||||
|
||||
Returns:
|
||||
list: Single-item list containing PTY device info
|
||||
"""
|
||||
if not self._is_initialized:
|
||||
return []
|
||||
|
||||
return [{
|
||||
'name': 'PTY Terminal',
|
||||
'path': '/dev/pts/*',
|
||||
'type': 'terminal',
|
||||
'connected': True
|
||||
}]
|
||||
|
||||
def get_led_state(self, led_mask=None):
|
||||
"""Get LED state information.
|
||||
|
||||
PTY driver cannot access LED states since it operates through
|
||||
terminal emulation rather than direct hardware access.
|
||||
|
||||
Args:
|
||||
led_mask: LED mask parameter (ignored for PTY)
|
||||
|
||||
Returns:
|
||||
dict: Empty dict (no LED access for PTY)
|
||||
"""
|
||||
return {}
|
||||
|
||||
def set_led_state(self, led_dict):
|
||||
"""Set LED states.
|
||||
|
||||
PTY driver cannot control LEDs since it operates through
|
||||
terminal emulation rather than direct hardware access.
|
||||
|
||||
Args:
|
||||
led_dict (dict): LED state dictionary (ignored for PTY)
|
||||
|
||||
Returns:
|
||||
bool: Always False (LED control not supported)
|
||||
"""
|
||||
return False
|
||||
|
@ -13,6 +13,7 @@ import signal
|
||||
import struct
|
||||
import sys
|
||||
import termios
|
||||
import threading
|
||||
import time
|
||||
import tty
|
||||
from select import select
|
||||
@ -24,6 +25,27 @@ from fenrirscreenreader.core.eventData import FenrirEventType
|
||||
from fenrirscreenreader.core.screenDriver import ScreenDriver as screenDriver
|
||||
from fenrirscreenreader.utils import screen_utils
|
||||
|
||||
# PTY Driver Constants
|
||||
class PTYConstants:
|
||||
# Timeouts (in seconds)
|
||||
DEFAULT_READ_TIMEOUT = 0.3
|
||||
INPUT_READ_TIMEOUT = 0.01
|
||||
OUTPUT_READ_TIMEOUT = 0.05 # Faster than default but allows for network lag
|
||||
SELECT_TIMEOUT = 0.05
|
||||
PROCESS_TERMINATION_TIMEOUT = 3.0
|
||||
PROCESS_KILL_DELAY = 0.5
|
||||
|
||||
# Polling intervals (in seconds)
|
||||
MIN_POLL_INTERVAL = 0.001
|
||||
|
||||
# Limits
|
||||
MAX_TERMINAL_LINES = 10000
|
||||
DEFAULT_READ_BUFFER_SIZE = 65536
|
||||
INPUT_BUFFER_SIZE = 4096
|
||||
|
||||
# Error codes
|
||||
IO_ERROR_ERRNO = 5
|
||||
|
||||
|
||||
class FenrirScreen(pyte.Screen):
|
||||
def set_margins(self, *args, **kwargs):
|
||||
@ -32,16 +54,39 @@ class FenrirScreen(pyte.Screen):
|
||||
|
||||
|
||||
class Terminal:
|
||||
def __init__(self, columns, lines, p_in):
|
||||
def __init__(self, columns, lines, p_in, env=None):
|
||||
self.text = ""
|
||||
self.attributes = None
|
||||
self.screen = FenrirScreen(columns, lines)
|
||||
self.env = env # Environment for proper logging
|
||||
|
||||
# Pre-create default attribute template to avoid repeated allocation
|
||||
self._default_attribute = [
|
||||
"default", "default", False, False, False, False, False, False,
|
||||
"default", "default"
|
||||
]
|
||||
self.screen.write_process_input = lambda data: p_in.write(
|
||||
data.encode()
|
||||
)
|
||||
self.stream = pyte.ByteStream()
|
||||
self.stream.attach(self.screen)
|
||||
|
||||
def _log_error(self, message, level=None):
|
||||
"""Log error message using proper debug manager if available."""
|
||||
if self.env and "runtime" in self.env and "DebugManager" in self.env["runtime"]:
|
||||
try:
|
||||
log_level = level if level else debug.DebugLevel.ERROR
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY Terminal: {message}",
|
||||
log_level
|
||||
)
|
||||
return
|
||||
except Exception:
|
||||
pass # Fallback to print if debug manager fails
|
||||
|
||||
# Fallback logging when debug manager unavailable
|
||||
print(f"PTY Terminal: {message}")
|
||||
|
||||
def feed(self, data):
|
||||
self.stream.feed(data)
|
||||
|
||||
@ -52,45 +97,58 @@ class Terminal:
|
||||
lines = self.screen.dirty
|
||||
else:
|
||||
lines = range(self.screen.lines)
|
||||
try:
|
||||
self.attributes = [
|
||||
[
|
||||
list(attribute[1:]) + [False, "default", "default"]
|
||||
if len(attribute) > 1 else [False, "default", "default"]
|
||||
for attribute in line.values()
|
||||
]
|
||||
for line in buffer.values()
|
||||
]
|
||||
for y in lines:
|
||||
try:
|
||||
t = self.attributes[y]
|
||||
except Exception as e:
|
||||
# Terminal class doesn't have access to env, use fallback
|
||||
# logging
|
||||
print(
|
||||
f"ptyDriver Terminal update_attributes: Error accessing "
|
||||
f"attributes: {e}"
|
||||
self._log_error(f"Error initializing attributes: {e}")
|
||||
# Fallback to empty attributes
|
||||
self.attributes = [[] for _ in range(self.screen.lines)]
|
||||
for y in lines:
|
||||
# Validate y is within reasonable bounds (prevent memory exhaustion)
|
||||
if y >= PTYConstants.MAX_TERMINAL_LINES:
|
||||
self._log_error(
|
||||
f"Line index {y} exceeds maximum {PTYConstants.MAX_TERMINAL_LINES}, "
|
||||
f"skipping attribute update",
|
||||
debug.DebugLevel.WARNING
|
||||
)
|
||||
continue
|
||||
|
||||
# Check if line y exists in buffer before accessing it
|
||||
if y not in buffer:
|
||||
# Only log this occasionally to prevent spam
|
||||
if y % 10 == 0: # Log every 10th missing line
|
||||
# Pre-format string to avoid repeated f-string operations
|
||||
line_range = f"{y}-{y+9}"
|
||||
self._log_error(
|
||||
f"Lines {line_range} not found in buffer, skipping attribute updates",
|
||||
debug.DebugLevel.WARNING
|
||||
)
|
||||
continue
|
||||
|
||||
# Ensure attributes array is large enough for line y
|
||||
while len(self.attributes) <= y:
|
||||
self.attributes.append([])
|
||||
|
||||
try:
|
||||
self.attributes[y] = [
|
||||
list(attribute[1:]) + [False, "default", "default"]
|
||||
for attribute in (buffer[y].values())
|
||||
]
|
||||
except Exception as e:
|
||||
self._log_error(f"Error updating attributes for line {y}: {e}")
|
||||
# Initialize with empty attributes if update fails
|
||||
self.attributes[y] = []
|
||||
if len(self.attributes[y]) < self.screen.columns:
|
||||
diff = self.screen.columns - len(self.attributes[y])
|
||||
self.attributes[y] += [
|
||||
[
|
||||
"default",
|
||||
"default",
|
||||
False,
|
||||
False,
|
||||
False,
|
||||
False,
|
||||
False,
|
||||
False,
|
||||
"default",
|
||||
"default",
|
||||
]
|
||||
] * diff
|
||||
# Use pre-created template for efficiency
|
||||
self.attributes[y] += [self._default_attribute[:] for _ in range(diff)]
|
||||
|
||||
def resize(self, lines, columns):
|
||||
self.screen.resize(lines, columns)
|
||||
@ -98,31 +156,37 @@ class Terminal:
|
||||
self.update_attributes(True)
|
||||
|
||||
def set_cursor(self, x=-1, y=-1):
|
||||
x_pos = x
|
||||
y_pos = y
|
||||
if x_pos == -1:
|
||||
x_pos = self.screen.cursor.x
|
||||
if y_pos == -1:
|
||||
y_pos = self.screen.cursor.y
|
||||
self.screen.cursor.x = min(
|
||||
self.screen.cursor.x, self.screen.columns - 1
|
||||
)
|
||||
self.screen.cursor.y = min(self.screen.cursor.y, self.screen.lines - 1)
|
||||
# Determine target cursor position
|
||||
x_pos = x if x != -1 else self.screen.cursor.x
|
||||
y_pos = y if y != -1 else self.screen.cursor.y
|
||||
|
||||
# Validate and clamp cursor position to screen bounds
|
||||
max_x = max(0, self.screen.columns - 1)
|
||||
max_y = max(0, self.screen.lines - 1)
|
||||
|
||||
self.screen.cursor.x = max(0, min(x_pos, max_x))
|
||||
self.screen.cursor.y = max(0, min(y_pos, max_y))
|
||||
|
||||
def get_screen_content(self):
|
||||
cursor = self.screen.cursor
|
||||
# Only regenerate text if screen is dirty or text doesn't exist
|
||||
if not hasattr(self, 'text') or self.screen.dirty:
|
||||
self.text = "\n".join(self.screen.display)
|
||||
self.update_attributes(self.attributes is None)
|
||||
self.screen.dirty.clear()
|
||||
return {
|
||||
|
||||
# Return screen content without unnecessary copying
|
||||
# Only copy attributes if they exist and need protection
|
||||
screen_data = {
|
||||
"cursor": (cursor.x, cursor.y),
|
||||
"lines": self.screen.lines,
|
||||
"columns": self.screen.columns,
|
||||
"text": self.text,
|
||||
"attributes": self.attributes.copy(),
|
||||
"attributes": self.attributes[:] if self.attributes else [], # Shallow copy only if needed
|
||||
"screen": "pty",
|
||||
"screenUpdateTime": time.time(),
|
||||
}.copy()
|
||||
}
|
||||
return screen_data
|
||||
|
||||
|
||||
class driver(screenDriver):
|
||||
@ -132,13 +196,64 @@ class driver(screenDriver):
|
||||
self.p_out = None
|
||||
self.terminal = None
|
||||
self.p_pid = -1
|
||||
self.terminal_lock = threading.Lock() # Synchronize terminal operations
|
||||
signal.signal(signal.SIGWINCH, self.handle_sigwinch)
|
||||
|
||||
# Runtime configuration storage
|
||||
self.pty_config = {}
|
||||
|
||||
def _load_pty_settings(self):
|
||||
"""Load PTY-specific settings from configuration with fallbacks to defaults."""
|
||||
try:
|
||||
settings_manager = self.env["runtime"]["SettingsManager"]
|
||||
|
||||
# Load timeout settings with defaults
|
||||
self.pty_config = {
|
||||
'input_timeout': float(settings_manager.get_setting(
|
||||
'screen', 'ptyInputTimeout', PTYConstants.INPUT_READ_TIMEOUT
|
||||
)),
|
||||
'output_timeout': float(settings_manager.get_setting(
|
||||
'screen', 'ptyOutputTimeout', PTYConstants.OUTPUT_READ_TIMEOUT
|
||||
)),
|
||||
'select_timeout': float(settings_manager.get_setting(
|
||||
'screen', 'ptySelectTimeout', PTYConstants.SELECT_TIMEOUT
|
||||
)),
|
||||
'process_termination_timeout': float(settings_manager.get_setting(
|
||||
'screen', 'ptyProcessTimeout', PTYConstants.PROCESS_TERMINATION_TIMEOUT
|
||||
)),
|
||||
'poll_interval': float(settings_manager.get_setting(
|
||||
'screen', 'ptyPollInterval', PTYConstants.MIN_POLL_INTERVAL
|
||||
))
|
||||
}
|
||||
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY screenDriver: Loaded configuration: {self.pty_config}",
|
||||
debug.DebugLevel.INFO
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
# Fallback to constants if settings fail
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY screenDriver: Failed to load settings, using defaults: {e}",
|
||||
debug.DebugLevel.WARNING
|
||||
)
|
||||
self.pty_config = {
|
||||
'input_timeout': PTYConstants.INPUT_READ_TIMEOUT,
|
||||
'output_timeout': PTYConstants.OUTPUT_READ_TIMEOUT,
|
||||
'select_timeout': PTYConstants.SELECT_TIMEOUT,
|
||||
'process_termination_timeout': PTYConstants.PROCESS_TERMINATION_TIMEOUT,
|
||||
'poll_interval': PTYConstants.MIN_POLL_INTERVAL
|
||||
}
|
||||
|
||||
def initialize(self, environment):
|
||||
self.env = environment
|
||||
self.command = self.env["runtime"]["SettingsManager"].get_setting(
|
||||
"general", "shell"
|
||||
)
|
||||
|
||||
# Load configurable timeouts from settings
|
||||
self._load_pty_settings()
|
||||
|
||||
self.shortcutType = self.env["runtime"][
|
||||
"InputManager"
|
||||
].get_shortcut_type()
|
||||
@ -162,29 +277,57 @@ class driver(screenDriver):
|
||||
self.env["general"]["prev_user"] = getpass.getuser()
|
||||
self.env["general"]["curr_user"] = getpass.getuser()
|
||||
|
||||
def read_all(self, fd, timeout=0.3, interruptFd=None, len=65536):
|
||||
def read_all(self, fd, timeout=PTYConstants.DEFAULT_READ_TIMEOUT, interruptFd=None, len=PTYConstants.DEFAULT_READ_BUFFER_SIZE):
|
||||
"""Read all available data from file descriptor with efficient polling.
|
||||
|
||||
Uses progressively longer wait times to balance responsiveness with CPU usage.
|
||||
"""
|
||||
msg_bytes = b""
|
||||
fd_list = []
|
||||
fd_list += [fd]
|
||||
fd_list = [fd]
|
||||
if interruptFd:
|
||||
fd_list += [interruptFd]
|
||||
fd_list.append(interruptFd)
|
||||
|
||||
starttime = time.time()
|
||||
poll_timeout = self.pty_config.get('poll_interval', PTYConstants.MIN_POLL_INTERVAL) # Use configured interval
|
||||
|
||||
while True:
|
||||
r = screen_utils.has_more_what(fd_list, 0.0001)
|
||||
# nothing more to read
|
||||
# Use consistent short polling for responsiveness
|
||||
r = screen_utils.has_more_what(fd_list, poll_timeout)
|
||||
|
||||
# Nothing more to read
|
||||
if fd not in r:
|
||||
# Check overall timeout
|
||||
if (time.time() - starttime) >= timeout:
|
||||
break
|
||||
continue
|
||||
|
||||
try:
|
||||
data = os.read(fd, len)
|
||||
if data == b"":
|
||||
raise EOFError
|
||||
msg_bytes += data
|
||||
# exit on interrupt available
|
||||
if interruptFd in r:
|
||||
except OSError as e:
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY screenDriver read_all: OS error reading from fd {fd}: {e}",
|
||||
debug.DebugLevel.ERROR
|
||||
)
|
||||
# For I/O errors, exit immediately to prevent endless retry loops
|
||||
if e.errno == PTYConstants.IO_ERROR_ERRNO: # Input/output error
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
"PTY screenDriver: Terminal connection lost, stopping read loop",
|
||||
debug.DebugLevel.ERROR
|
||||
)
|
||||
raise EOFError("Terminal connection lost")
|
||||
break
|
||||
# respect timeout but wait a little bit of time to see if something
|
||||
# more is here
|
||||
|
||||
# Exit on interrupt available
|
||||
if interruptFd and interruptFd in r:
|
||||
break
|
||||
|
||||
# Check overall timeout
|
||||
if (time.time() - starttime) >= timeout:
|
||||
break
|
||||
|
||||
return msg_bytes
|
||||
|
||||
def open_terminal(self, columns, lines, command):
|
||||
@ -197,16 +340,16 @@ class driver(screenDriver):
|
||||
if env["TERM"] == "":
|
||||
env["TERM"] = "linux"
|
||||
except Exception as e:
|
||||
# Child process doesn't have access to env, use fallback
|
||||
# logging
|
||||
# Child process doesn't have access to debug manager
|
||||
# Use fallback logging with more context
|
||||
print(
|
||||
f"ptyDriver spawnTerminal: Error checking TERM environment: {e}"
|
||||
f"ptyDriver open_terminal (child): TERM environment error: {e}"
|
||||
)
|
||||
env["TERM"] = "linux"
|
||||
os.execvpe(argv[0], argv, env)
|
||||
# File-like object for I/O with the child process aka command.
|
||||
p_out = os.fdopen(master_fd, "w+b", 0)
|
||||
return Terminal(columns, lines, p_out), p_pid, p_out
|
||||
return Terminal(columns, lines, p_out, self.env), p_pid, p_out
|
||||
|
||||
def resize_terminal(self, fd):
|
||||
s = struct.pack("HHHH", 0, 0, 0, 0)
|
||||
@ -239,7 +382,7 @@ class driver(screenDriver):
|
||||
self.terminal.resize(lines, columns)
|
||||
fd_list = [sys.stdin, self.p_out, self.signalPipe[0]]
|
||||
while active.value:
|
||||
r, _, _ = select(fd_list, [], [], 1)
|
||||
r, _, _ = select(fd_list, [], [], self.pty_config.get('select_timeout', PTYConstants.SELECT_TIMEOUT)) # Configurable timeout
|
||||
# none
|
||||
if r == []:
|
||||
continue
|
||||
@ -251,7 +394,7 @@ class driver(screenDriver):
|
||||
# input
|
||||
if sys.stdin in r:
|
||||
try:
|
||||
msg_bytes = self.read_all(sys.stdin.fileno(), len=4096)
|
||||
msg_bytes = self.read_all(sys.stdin.fileno(), timeout=self.pty_config.get('input_timeout', PTYConstants.INPUT_READ_TIMEOUT), len=PTYConstants.INPUT_BUFFER_SIZE)
|
||||
except (EOFError, OSError):
|
||||
event_queue.put(
|
||||
{
|
||||
@ -289,7 +432,9 @@ class driver(screenDriver):
|
||||
if self.p_out in r:
|
||||
try:
|
||||
msg_bytes = self.read_all(
|
||||
self.p_out.fileno(), interruptFd=sys.stdin.fileno()
|
||||
self.p_out.fileno(),
|
||||
timeout=self.pty_config.get('output_timeout', PTYConstants.OUTPUT_READ_TIMEOUT),
|
||||
interruptFd=sys.stdin.fileno()
|
||||
)
|
||||
except (EOFError, OSError):
|
||||
event_queue.put(
|
||||
@ -299,34 +444,121 @@ class driver(screenDriver):
|
||||
}
|
||||
)
|
||||
break
|
||||
# feed and send event bevore write, the pyte already has the right state
|
||||
# so fenrir already can progress bevore os.write what
|
||||
# should give some better reaction time
|
||||
# Synchronize terminal operations to prevent race conditions
|
||||
with self.terminal_lock:
|
||||
# Feed data to terminal and get consistent screen state
|
||||
self.terminal.feed(msg_bytes)
|
||||
screen_content = self.terminal.get_screen_content()
|
||||
|
||||
# Send screen update event with consistent state
|
||||
event_queue.put(
|
||||
{
|
||||
"Type": FenrirEventType.screen_update,
|
||||
"data": screen_utils.create_screen_event_data(
|
||||
self.terminal.get_screen_content()
|
||||
screen_content
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
# Inject to actual screen (outside lock to avoid blocking)
|
||||
self.inject_text_to_screen(
|
||||
msg_bytes, screen=sys.stdout.fileno()
|
||||
)
|
||||
except Exception as e: # Process died?
|
||||
print(e)
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY screenDriver terminal_emulation: Exception occurred: {e}",
|
||||
debug.DebugLevel.ERROR
|
||||
)
|
||||
event_queue.put(
|
||||
{"Type": FenrirEventType.stop_main_loop, "data": None}
|
||||
)
|
||||
finally:
|
||||
os.kill(self.p_pid, signal.SIGTERM)
|
||||
self.p_out.close()
|
||||
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_attr)
|
||||
self._safe_cleanup_process()
|
||||
self._safe_cleanup_resources(old_attr)
|
||||
event_queue.put(
|
||||
{"Type": FenrirEventType.stop_main_loop, "data": None}
|
||||
)
|
||||
sys.exit(0)
|
||||
|
||||
def _safe_cleanup_process(self):
|
||||
"""Safely terminate the child process with timeout and fallback to SIGKILL."""
|
||||
if not hasattr(self, 'p_pid') or self.p_pid is None:
|
||||
return
|
||||
|
||||
try:
|
||||
# Check if process is still alive
|
||||
os.kill(self.p_pid, 0) # Signal 0 checks if process exists
|
||||
except OSError:
|
||||
# Process already dead
|
||||
self.p_pid = None
|
||||
return
|
||||
|
||||
try:
|
||||
# Try graceful termination first
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY screenDriver: Terminating process {self.p_pid} gracefully",
|
||||
debug.DebugLevel.INFO
|
||||
)
|
||||
os.kill(self.p_pid, signal.SIGTERM)
|
||||
|
||||
# Wait for graceful termination
|
||||
timeout = self.pty_config.get('process_termination_timeout', PTYConstants.PROCESS_TERMINATION_TIMEOUT)
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < timeout:
|
||||
try:
|
||||
os.kill(self.p_pid, 0) # Check if still alive
|
||||
time.sleep(0.1)
|
||||
except OSError:
|
||||
# Process terminated gracefully
|
||||
self.p_pid = None
|
||||
return
|
||||
|
||||
# Process didn't terminate gracefully, use SIGKILL
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY screenDriver: Process {self.p_pid} didn't terminate gracefully, using SIGKILL",
|
||||
debug.DebugLevel.WARNING
|
||||
)
|
||||
os.kill(self.p_pid, signal.SIGKILL)
|
||||
time.sleep(PTYConstants.PROCESS_KILL_DELAY) # Give it a moment
|
||||
|
||||
except OSError as e:
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY screenDriver: Error terminating process {self.p_pid}: {e}",
|
||||
debug.DebugLevel.ERROR
|
||||
)
|
||||
finally:
|
||||
self.p_pid = None
|
||||
|
||||
def _safe_cleanup_resources(self, old_attr=None):
|
||||
"""Safely clean up file descriptors and terminal attributes."""
|
||||
# Close output pipe safely
|
||||
if hasattr(self, 'p_out') and self.p_out is not None:
|
||||
try:
|
||||
self.p_out.close()
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
"PTY screenDriver: Closed output pipe",
|
||||
debug.DebugLevel.INFO
|
||||
)
|
||||
except Exception as e:
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY screenDriver: Error closing output pipe: {e}",
|
||||
debug.DebugLevel.ERROR
|
||||
)
|
||||
finally:
|
||||
self.p_out = None
|
||||
|
||||
# Restore terminal attributes safely
|
||||
if old_attr is not None:
|
||||
try:
|
||||
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_attr)
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
"PTY screenDriver: Restored terminal attributes",
|
||||
debug.DebugLevel.INFO
|
||||
)
|
||||
except Exception as e:
|
||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||
f"PTY screenDriver: Error restoring terminal attributes: {e}",
|
||||
debug.DebugLevel.ERROR
|
||||
)
|
||||
|
||||
def get_curr_application(self):
|
||||
pass
|
||||
|
@ -170,8 +170,9 @@ class driver(screenDriver):
|
||||
if screen is not None:
|
||||
use_screen = screen
|
||||
with open(use_screen, "w") as fd:
|
||||
for c in text:
|
||||
fcntl.ioctl(fd, termios.TIOCSTI, c)
|
||||
text_bytes = text.encode('utf-8')
|
||||
for byte in text_bytes:
|
||||
fcntl.ioctl(fd, termios.TIOCSTI, bytes([byte]))
|
||||
|
||||
def get_session_information(self):
|
||||
"""Retrieve session information via D-Bus logind interface.
|
||||
|
Reference in New Issue
Block a user