storm/fenrir
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

More pep8 fixes. A tiny bit of refactoring.

Browse Source
This commit is contained in:
Storm Dragon
2025-07-07 00:42:23 -04:00
parent d28c18faed
commit 3390c25dfe
343 changed files with 11092 additions and 7582 deletions
Download Patch File Download Diff File Expand all files Collapse all files

235
src/fenrirscreenreader/commands/onScreenUpdate/65000-progress_detector.py
View File

@ -2,14 +2,13 @@
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributers.
from fenrirscreenreader.core.i18n import _
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.core import debug
from fenrirscreenreader.core.i18n import _
class command():
class command:
def __init__(self):
pass
@ -20,20 +19,26 @@ class command():
pass
def get_description(self):
return 'Detects progress patterns for progress bar monitoring'
return "Detects progress patterns for progress bar monitoring"
def run(self):
# Only run if progress monitoring is enabled
try:
if 'progressMonitoring' in self.env['commandBuffer'] and self.env['commandBuffer']['progressMonitoring']:
if (
"progressMonitoring" in self.env["commandBuffer"]
and self.env["commandBuffer"]["progressMonitoring"]
):
# Check if current line is a prompt - if so, reset progress
# state
if self.is_current_line_prompt():
self.reset_progress_state()
# Only check new incoming text (new_delta), but filter out
# screen changes
elif self.env['screen']['new_delta'] and self.is_real_progress_update():
self.detect_progress(self.env['screen']['new_delta'])
elif (
self.env["screen"]["new_delta"]
and self.is_real_progress_update()
):
self.detect_progress(self.env["screen"]["new_delta"])
except Exception as e:
# Silently ignore errors to avoid disrupting normal operation
pass
@ -41,26 +46,30 @@ class command():
def is_real_progress_update(self):
"""Check if this is a real progress update vs screen change/window switch"""
# If the screen/application changed, it's not a progress update
if self.env['runtime']['ScreenManager'].is_screen_change():
if self.env["runtime"]["ScreenManager"].is_screen_change():
return False
# If there was a large cursor movement, it's likely navigation, not
# progress
if self.env['runtime']['CursorManager'].is_cursor_vertical_move():
if self.env["runtime"]["CursorManager"].is_cursor_vertical_move():
x_move = abs(
self.env['screen']['new_cursor']['x'] -
self.env['screen']['old_cursor']['x'])
self.env["screen"]["new_cursor"]["x"]
- self.env["screen"]["old_cursor"]["x"]
)
y_move = abs(
self.env['screen']['new_cursor']['y'] -
self.env['screen']['old_cursor']['y'])
self.env["screen"]["new_cursor"]["y"]
- self.env["screen"]["old_cursor"]["y"]
)
# Large movements suggest navigation, not progress output
if y_move > 2 or x_move > 20:
return False
# Check if delta is too large (screen change) vs small incremental
# updates
delta_length = len(self.env['screen']['new_delta'])
if delta_length > 200: # Allow longer progress lines like Claude Code's status
delta_length = len(self.env["screen"]["new_delta"])
if (
delta_length > 200
): # Allow longer progress lines like Claude Code's status
return False
# Check if current line looks like a prompt - progress unlikely during
@ -72,11 +81,12 @@ class command():
def reset_progress_state(self):
"""Reset progress state when a prompt is detected, allowing new progress operations to start fresh"""
self.env['runtime']['DebugManager'].write_debug_out(
self.env["runtime"]["DebugManager"].write_debug_out(
"Resetting progress state due to prompt detection",
debug.DebugLevel.INFO)
self.env['commandBuffer']['lastProgressValue'] = -1
self.env['commandBuffer']['lastProgressTime'] = 0
debug.DebugLevel.INFO,
)
self.env["commandBuffer"]["lastProgressValue"] = -1
self.env["commandBuffer"]["lastProgressTime"] = 0
def detect_progress(self, text):
import re
@ -85,8 +95,9 @@ class command():
current_time = time.time()
# Debug: Print what we're checking
self.env['runtime']['DebugManager'].write_debug_out(
"Progress detector checking: '" + text + "'", debug.DebugLevel.INFO)
self.env["runtime"]["DebugManager"].write_debug_out(
"Progress detector checking: '" + text + "'", debug.DebugLevel.INFO
)
# Note: Auto-disable on 100% completion removed to respect user
# settings
@ -94,63 +105,85 @@ class command():
# Pattern 1: Percentage (50%, 25.5%, etc.)
# Filter out common non-progress percentages (weather, system stats,
# etc.)
percent_match = re.search(r'(\d+(?:\.\d+)?)\s*%', text)
percent_match = re.search(r"(\d+(?:\.\d+)?)\s*%", text)
if percent_match:
percentage = float(percent_match.group(1))
# Only trigger on realistic progress percentages (0-100%)
if 0 <= percentage <= 100:
# Filter out weather/system stats that contain percentages
if not re.search(
r'\b(?:humidity|cpu|memory|disk|usage|temp|weather|forecast)\b',
r"\b(?:humidity|cpu|memory|disk|usage|temp|weather|forecast)\b",
text,
re.IGNORECASE):
self.env['runtime']['DebugManager'].write_debug_out(
"found percentage: " + str(percentage), debug.DebugLevel.INFO)
if percentage != self.env['commandBuffer']['lastProgressValue']:
self.env['runtime']['DebugManager'].write_debug_out(
"Playing tone for: " + str(percentage), debug.DebugLevel.INFO)
re.IGNORECASE,
):
self.env["runtime"]["DebugManager"].write_debug_out(
"found percentage: " + str(percentage),
debug.DebugLevel.INFO,
)
if (
percentage
!= self.env["commandBuffer"]["lastProgressValue"]
):
self.env["runtime"]["DebugManager"].write_debug_out(
"Playing tone for: " + str(percentage),
debug.DebugLevel.INFO,
)
self.play_progress_tone(percentage)
self.env['commandBuffer']['lastProgressValue'] = percentage
self.env['commandBuffer']['lastProgressTime'] = current_time
self.env["commandBuffer"][
"lastProgressValue"
] = percentage
self.env["commandBuffer"][
"lastProgressTime"
] = current_time
return
# Pattern 1b: Time/token activity (not percentage-based, so use single
# beep)
time_match = re.search(r'(\d+)s\s', text)
token_match = re.search(r'(\d+)\s+tokens', text)
time_match = re.search(r"(\d+)s\s", text)
token_match = re.search(r"(\d+)\s+tokens", text)
# Pattern 1c: dd command output (bytes copied with transfer rate)
dd_match = re.search(r'\d+\s+bytes.*copied.*\d+\s+s.*[kMGT]?B/s', text)
dd_match = re.search(r"\d+\s+bytes.*copied.*\d+\s+s.*[kMGT]?B/s", text)
# Pattern 1d: Curl-style transfer data (bytes, speed indicators)
curl_match = re.search(
r'(\d+\s+\d+\s+\d+\s+\d+.*?(?:k|M|G)?.*?--:--:--|Speed)', text)
r"(\d+\s+\d+\s+\d+\s+\d+.*?(?:k|M|G)?.*?--:--:--|Speed)", text
)
if time_match or token_match or dd_match or curl_match:
# For non-percentage progress, use a single activity beep every 2
# seconds
if current_time - \
self.env['commandBuffer']['lastProgressTime'] >= 2.0:
self.env['runtime']['DebugManager'].write_debug_out(
"Playing activity beep for transfer progress", debug.DebugLevel.INFO)
if (
current_time - self.env["commandBuffer"]["lastProgressTime"]
>= 2.0
):
self.env["runtime"]["DebugManager"].write_debug_out(
"Playing activity beep for transfer progress",
debug.DebugLevel.INFO,
)
self.play_activity_beep()
self.env['commandBuffer']['lastProgressTime'] = current_time
self.env["commandBuffer"]["lastProgressTime"] = current_time
return
# Pattern 2: Fraction (15/100, 3 of 10, etc.)
fraction_match = re.search(r'(\d+)\s*(?:of|/)\s*(\d+)', text)
fraction_match = re.search(r"(\d+)\s*(?:of|/)\s*(\d+)", text)
if fraction_match:
current = int(fraction_match.group(1))
total = int(fraction_match.group(2))
if total > 0:
percentage = (current / total) * 100
if percentage != self.env['commandBuffer']['lastProgressValue']:
if (
percentage
!= self.env["commandBuffer"]["lastProgressValue"]
):
self.play_progress_tone(percentage)
self.env['commandBuffer']['lastProgressValue'] = percentage
self.env['commandBuffer']['lastProgressTime'] = current_time
self.env["commandBuffer"]["lastProgressValue"] = percentage
self.env["commandBuffer"][
"lastProgressTime"
] = current_time
return
# 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))
@ -158,27 +191,35 @@ class command():
# Require at least 2 progress chars total and unfilled portion must
# be spaces/dots
if total >= 2 and (
not bar_match.group(2) or re.match(
r'^[\s\.]*$',
bar_match.group(2))):
not bar_match.group(2)
or re.match(r"^[\s\.]*$", bar_match.group(2))
):
percentage = (filled / total) * 100
if percentage != self.env['commandBuffer']['lastProgressValue']:
if (
percentage
!= self.env["commandBuffer"]["lastProgressValue"]
):
self.play_progress_tone(percentage)
self.env['commandBuffer']['lastProgressValue'] = percentage
self.env['commandBuffer']['lastProgressTime'] = current_time
self.env["commandBuffer"]["lastProgressValue"] = percentage
self.env["commandBuffer"][
"lastProgressTime"
] = current_time
return
# Pattern 4: Generic activity indicators (Loading..., Working..., etc.)
activity_pattern = re.search(
r'(loading|processing|working|installing|downloading|compiling|building).*\.{2,}',
r"(loading|processing|working|installing|downloading|compiling|building).*\.{2,}",
text,
re.IGNORECASE)
re.IGNORECASE,
)
if activity_pattern:
# Play a steady beep every 2 seconds for ongoing activity
if current_time - \
self.env['commandBuffer']['lastProgressTime'] >= 2.0:
if (
current_time - self.env["commandBuffer"]["lastProgressTime"]
>= 2.0
):
self.play_activity_beep()
self.env['commandBuffer']['lastProgressTime'] = current_time
self.env["commandBuffer"]["lastProgressTime"] = current_time
def play_progress_tone(self, percentage):
# Map 0-100% to 400-1200Hz frequency range
@ -195,8 +236,8 @@ class command():
def play_quiet_tone(self, frequency, duration):
"""Play a quiet tone using Sox directly"""
import subprocess
import shlex
import subprocess
# Build the Sox command: play -qn synth <duration> tri <frequency> gain
# -8
@ -204,17 +245,20 @@ class command():
try:
# Only play if sound is enabled
if self.env['runtime']['SettingsManager'].get_setting_as_bool(
'sound', 'enabled'):
if self.env["runtime"]["SettingsManager"].get_setting_as_bool(
"sound", "enabled"
):
subprocess.Popen(
shlex.split(command),
stdin=None,
stdout=None,
stderr=None,
shell=False)
shell=False,
)
except Exception as e:
self.env['runtime']['DebugManager'].write_debug_out(
"Sox tone error: " + str(e), debug.DebugLevel.ERROR)
self.env["runtime"]["DebugManager"].write_debug_out(
"Sox tone error: " + str(e), debug.DebugLevel.ERROR
)
def is_current_line_prompt(self):
"""Check if the current line looks like a standalone prompt (not command with progress)"""
@ -222,10 +266,10 @@ class command():
try:
# Get the current screen content
if not self.env['screen']['new_content_text']:
if not self.env["screen"]["new_content_text"]:
return False
lines = self.env['screen']['new_content_text'].split('\n')
lines = self.env["screen"]["new_content_text"].split("\n")
if not lines:
return False
@ -238,34 +282,34 @@ class command():
lines_to_check.append(lines[-1])
# Add current cursor line if different from last line
if (self.env['screen']['new_cursor']['y'] < len(lines) and
self.env['screen']['new_cursor']['y'] != len(lines) - 1):
if (
self.env["screen"]["new_cursor"]["y"] < len(lines)
and self.env["screen"]["new_cursor"]["y"] != len(lines) - 1
):
lines_to_check.append(
lines[self.env['screen']['new_cursor']['y']])
lines[self.env["screen"]["new_cursor"]["y"]]
)
# Standalone prompt patterns (no commands mixed in)
standalone_prompt_patterns = [
r'^\s*\$\s*$', # Just $ (with whitespace)
r'^\s*#\s*$', # Just # (with whitespace)
r'^\s*>\s*$', # Just > (with whitespace)
r'^\[.*\]\s*[\\\$#>]\s*$', # [path]$ without commands
r'^[a-zA-Z0-9._-]+[\\\$#>]\s*$', # bash-5.1$ without commands
r"^\s*\$\s*$", # Just $ (with whitespace)
r"^\s*#\s*$", # Just # (with whitespace)
r"^\s*>\s*$", # Just > (with whitespace)
r"^\[.*\]\s*[\\\$#>]\s*$", # [path]$ without commands
r"^[a-zA-Z0-9._-]+[\\\$#>]\s*$", # bash-5.1$ without commands
# Interactive prompt patterns (these ARE standalone)
r'.*\?\s*\[[YyNn]/[YyNn]\]\s*$', # ? [Y/n] or ? [y/N] style
r'.*\?\s*\[[Yy]es/[Nn]o\]\s*$', # ? [Yes/No] style
r".*\?\s*\[[YyNn]/[YyNn]\]\s*$", # ? [Y/n] or ? [y/N] style
r".*\?\s*\[[Yy]es/[Nn]o\]\s*$", # ? [Yes/No] style
# "continue? [Y/n]" style
r'.*continue\?\s*\[[YyNn]/[YyNn]\].*$',
r'^::.*\?\s*\[[YyNn]/[YyNn]\].*$', # pacman style prompts
r".*continue\?\s*\[[YyNn]/[YyNn]\].*$",
r"^::.*\?\s*\[[YyNn]/[YyNn]\].*$", # pacman style prompts
# Authentication prompts (these ARE standalone)
r'^\[[Ss]udo\]\s*[Pp]assword\s*for\s+.*:\s*$', # [sudo] password
r'^[Pp]assword\s*:\s*$', # Password:
r'.*[Pp]assword\s*:\s*$', # general password prompts
r"^\[[Ss]udo\]\s*[Pp]assword\s*for\s+.*:\s*$", # [sudo] password
r"^[Pp]assword\s*:\s*$", # Password:
r".*[Pp]assword\s*:\s*$", # general password prompts
# Continuation prompts (these ARE standalone)
r'^[Pp]ress\s+any\s+key\s+to\s+continue.*$', # Press any key
r'^[Aa]re\s+you\s+sure\?\s*.*$', # Are you sure?
r"^[Pp]ress\s+any\s+key\s+to\s+continue.*$", # Press any key
r"^[Aa]re\s+you\s+sure\?\s*.*$", # Are you sure?
]
for line in lines_to_check:
@ -276,22 +320,21 @@ class command():
# Check if this line contains both a prompt AND other content (like commands)
# If so, don't treat it as a standalone prompt
has_prompt_marker = bool(
re.search(
r'.*@.*[\\\$#>]',
line) or re.search(
r'^\[.*\]\s*[\\\$#>]',
line))
re.search(r".*@.*[\\\$#>]", line)
or re.search(r"^\[.*\]\s*[\\\$#>]", line)
)
if has_prompt_marker:
# If line has prompt marker but also has significant content after it,
# it's a command line, not a standalone prompt
prompt_end = max(
line.rfind('$'),
line.rfind('#'),
line.rfind('>'),
line.rfind('\\')
line.rfind("$"),
line.rfind("#"),
line.rfind(">"),
line.rfind("\\"),
)
if prompt_end >= 0 and prompt_end < len(
line) - 5: # More than just whitespace after prompt
if (
prompt_end >= 0 and prompt_end < len(line) - 5
): # More than just whitespace after prompt
continue # This is a command line, not a standalone prompt
for pattern in standalone_prompt_patterns: