492 lines
20 KiB
Python
492 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
# Fenrir TTY screen reader
|
|
# By Chrys, Storm Dragon, and contributors.
|
|
|
|
import difflib
|
|
import os
|
|
import re
|
|
import time
|
|
|
|
from fenrirscreenreader.core import debug
|
|
from fenrirscreenreader.utils import screen_utils
|
|
|
|
|
|
class ScreenManager:
|
|
def __init__(self):
|
|
self.differ = difflib.Differ()
|
|
self.currScreenIgnored = False
|
|
self.prevScreenIgnored = False
|
|
self.prevScreenText = ""
|
|
self.currScreenText = ""
|
|
self.colums = None
|
|
self.rows = None
|
|
# Compile regex once for better performance
|
|
self._space_normalize_regex = re.compile(" +")
|
|
|
|
def get_rows(self):
|
|
return self.rows
|
|
|
|
def get_columns(self):
|
|
return self.colums
|
|
|
|
def initialize(self, environment):
|
|
self.env = environment
|
|
self.env["runtime"]["SettingsManager"].load_driver(
|
|
self.env["runtime"]["SettingsManager"].get_setting(
|
|
"screen", "driver"
|
|
),
|
|
"ScreenDriver",
|
|
)
|
|
self.get_curr_screen()
|
|
self.get_curr_screen()
|
|
self.get_session_information()
|
|
self.update_screen_ignored()
|
|
self.update_screen_ignored()
|
|
|
|
def reset_screen_text(self, screen_text):
|
|
self.prevScreenText = ""
|
|
self.currScreenText = screen_text
|
|
|
|
def set_screen_text(self, screen_text):
|
|
self.prevScreenText = self.currScreenText
|
|
self.currScreenText = screen_text
|
|
|
|
def get_screen_text(self):
|
|
return self.currScreenText
|
|
|
|
def get_line_text(self, line_number):
|
|
"""Get text from a specific line (0-based index)"""
|
|
if not self.env["screen"]["new_content_text"]:
|
|
return ""
|
|
|
|
lines = self.env["screen"]["new_content_text"].split("\n")
|
|
if line_number < 0 or line_number >= len(lines):
|
|
return ""
|
|
|
|
return lines[line_number]
|
|
|
|
def get_curr_screen(self):
|
|
try:
|
|
self.env["runtime"]["ScreenDriver"].get_curr_screen()
|
|
except Exception as e:
|
|
self.env["runtime"]["DebugManager"].write_debug_out(
|
|
"ScreenManager get_curr_screen: Error getting current screen: "
|
|
+ str(e),
|
|
debug.DebugLevel.ERROR,
|
|
)
|
|
|
|
def get_session_information(self):
|
|
try:
|
|
self.env["runtime"]["ScreenDriver"].get_session_information()
|
|
except Exception as e:
|
|
self.env["runtime"]["DebugManager"].write_debug_out(
|
|
"ScreenManager get_session_information: Error getting session info: "
|
|
+ str(e),
|
|
debug.DebugLevel.ERROR,
|
|
)
|
|
|
|
def shutdown(self):
|
|
self.env["runtime"]["SettingsManager"].shutdown_driver("ScreenDriver")
|
|
|
|
def is_curr_screen_ignored_changed(self):
|
|
return self.get_curr_screen_ignored() != self.get_prev_screen_ignored()
|
|
|
|
def handle_screen_change(self, event_data):
|
|
self.get_curr_screen()
|
|
self.get_session_information()
|
|
self.update_screen_ignored()
|
|
if self.is_curr_screen_ignored_changed():
|
|
self.env["runtime"]["InputManager"].set_execute_device_grab()
|
|
self.env["runtime"]["InputManager"].handle_device_grab()
|
|
if not self.is_ignored_screen(self.env["screen"]["newTTY"]):
|
|
self.update(event_data, "onScreenChange")
|
|
self.env["screen"]["lastScreenUpdate"] = time.time()
|
|
else:
|
|
self.env["runtime"]["OutputManager"].interrupt_output()
|
|
|
|
def handle_screen_update(self, event_data):
|
|
self.env["screen"]["oldApplication"] = self.env["screen"][
|
|
"new_application"
|
|
]
|
|
self.update_screen_ignored()
|
|
if self.is_curr_screen_ignored_changed():
|
|
self.env["runtime"]["InputManager"].set_execute_device_grab()
|
|
self.env["runtime"]["InputManager"].handle_device_grab()
|
|
if not self.get_curr_screen_ignored():
|
|
self.update(event_data, "onScreenUpdate")
|
|
self.env["screen"]["lastScreenUpdate"] = time.time()
|
|
elif self.is_curr_screen_ignored_changed():
|
|
self.env["runtime"]["OutputManager"].interrupt_output()
|
|
|
|
def get_curr_screen_ignored(self):
|
|
return self.currScreenIgnored
|
|
|
|
def get_prev_screen_ignored(self):
|
|
return self.prevScreenIgnored
|
|
|
|
def update_screen_ignored(self):
|
|
self.prevScreenIgnored = self.currScreenIgnored
|
|
self.currScreenIgnored = self.is_ignored_screen(
|
|
self.env["screen"]["newTTY"]
|
|
)
|
|
|
|
def update(self, event_data, trigger="onUpdate"):
|
|
# set new "old" values
|
|
self.env["screen"]["oldContentBytes"] = self.env["screen"][
|
|
"newContentBytes"
|
|
]
|
|
self.env["screen"]["old_content_text"] = self.env["screen"][
|
|
"new_content_text"
|
|
]
|
|
self.env["screen"]["old_cursor"] = self.env["screen"][
|
|
"new_cursor"
|
|
].copy()
|
|
self.env["screen"]["oldDelta"] = self.env["screen"]["new_delta"]
|
|
self.env["screen"]["oldNegativeDelta"] = self.env["screen"][
|
|
"newNegativeDelta"
|
|
]
|
|
self.env["screen"]["newContentBytes"] = event_data["bytes"]
|
|
|
|
# get metadata like cursor or screensize
|
|
self.env["screen"]["lines"] = int(event_data["lines"])
|
|
self.env["screen"]["columns"] = int(event_data["columns"])
|
|
self.colums = int(event_data["columns"])
|
|
self.rows = int(event_data["lines"])
|
|
self.env["screen"]["new_cursor"]["x"] = int(
|
|
event_data["textCursor"]["x"]
|
|
)
|
|
self.env["screen"]["new_cursor"]["y"] = int(
|
|
event_data["textCursor"]["y"]
|
|
)
|
|
self.env["screen"]["newTTY"] = event_data["screen"]
|
|
self.env["screen"]["new_content_text"] = event_data["text"]
|
|
|
|
# screen change
|
|
if self.is_screen_change():
|
|
self.env["screen"]["oldContentBytes"] = b""
|
|
self.reset_screen_text(event_data["text"])
|
|
self.env["runtime"]["AttributeManager"].reset_attributes(
|
|
event_data["attributes"]
|
|
)
|
|
self.env["runtime"]["AttributeManager"].reset_attribute_cursor()
|
|
self.env["screen"]["old_content_text"] = ""
|
|
self.env["screen"]["old_cursor"]["x"] = 0
|
|
self.env["screen"]["old_cursor"]["y"] = 0
|
|
self.env["screen"]["oldDelta"] = ""
|
|
self.env["screen"]["oldNegativeDelta"] = ""
|
|
else:
|
|
self.set_screen_text(event_data["text"])
|
|
self.env["runtime"]["AttributeManager"].set_attributes(
|
|
event_data["attributes"]
|
|
)
|
|
# initialize current deltas
|
|
self.env["screen"]["newNegativeDelta"] = ""
|
|
self.env["screen"]["new_delta"] = ""
|
|
self.env["runtime"]["AttributeManager"].reset_attribute_delta()
|
|
|
|
# Diff generation - critical for screen reader functionality
|
|
# This code detects and categorizes screen content changes to provide appropriate
|
|
# speech feedback (typing echo vs incoming text vs screen updates)
|
|
|
|
# Track whether this appears to be typing (user input) vs other screen
|
|
# changes
|
|
typing = False
|
|
diff_list = []
|
|
|
|
if (
|
|
self.env["screen"]["old_content_text"]
|
|
!= self.env["screen"]["new_content_text"]
|
|
):
|
|
# Special case: Initial screen content (going from empty to populated)
|
|
# This handles first screen load or TTY switch scenarios
|
|
if (
|
|
self.env["screen"]["new_content_text"] != ""
|
|
and self.env["screen"]["old_content_text"] == ""
|
|
):
|
|
# Pre-process screen text for comparison - collapse multiple spaces to single space
|
|
# This normalization prevents spurious diffs from spacing
|
|
# inconsistencies
|
|
old_screen_text = self._space_normalize_regex.sub(
|
|
" ",
|
|
self.env["runtime"][
|
|
"ScreenManager"
|
|
].get_window_area_in_text(
|
|
self.env["screen"]["old_content_text"]
|
|
),
|
|
)
|
|
new_screen_text = self._space_normalize_regex.sub(
|
|
" ",
|
|
self.env["runtime"][
|
|
"ScreenManager"
|
|
].get_window_area_in_text(
|
|
self.env["screen"]["new_content_text"]
|
|
),
|
|
)
|
|
if old_screen_text == "" and new_screen_text != "":
|
|
self.env["screen"]["new_delta"] = new_screen_text
|
|
else:
|
|
# Calculate byte positions for the current cursor's line in the flat text buffer
|
|
# Formula: (line_number * columns) + line_number accounts for newlines
|
|
# Each line contributes 'columns' chars + 1 newline char
|
|
cursor_line_start = (
|
|
self.env["screen"]["new_cursor"]["y"]
|
|
* self.env["screen"]["columns"]
|
|
+ self.env["screen"]["new_cursor"]["y"]
|
|
)
|
|
cursor_line_end = (
|
|
cursor_line_start + self.env["screen"]["columns"]
|
|
)
|
|
|
|
# TYPING DETECTION ALGORITHM
|
|
# Determines if this screen change is likely user typing vs other content changes
|
|
# All conditions must be met for typing detection:
|
|
if (
|
|
abs(
|
|
self.env["screen"]["old_cursor"]["x"]
|
|
- self.env["screen"]["new_cursor"]["x"]
|
|
)
|
|
>= 1
|
|
and self.env["screen"]["old_cursor"]["y"]
|
|
== self.env["screen"]["new_cursor"]["y"]
|
|
and self.env["screen"]["new_content_text"][
|
|
:cursor_line_start
|
|
]
|
|
== self.env["screen"]["old_content_text"][
|
|
:cursor_line_start
|
|
]
|
|
and self.env["screen"]["new_content_text"][
|
|
cursor_line_end:
|
|
]
|
|
== self.env["screen"]["old_content_text"][cursor_line_end:]
|
|
):
|
|
# Condition 1: Cursor moved horizontally by at least 1 position (typical of typing)
|
|
# Condition 2: Cursor stayed on same line (typing doesn't usually change lines)
|
|
# Condition 3: Content BEFORE cursor line is unchanged (text above typing area)
|
|
# Condition 4: Content AFTER cursor line is unchanged (text below typing area)
|
|
# Together: only the current line changed, cursor moved
|
|
# horizontally = likely typing
|
|
|
|
# Optimize diff calculation for typing by focusing on a
|
|
# small window around cursor
|
|
cursor_line_start_offset = cursor_line_start
|
|
cursor_line_end_offset = cursor_line_end
|
|
|
|
# Limit analysis window to avoid processing entire long lines
|
|
# +3 provides safety buffer beyond cursor position to catch edge cases
|
|
if (
|
|
cursor_line_end
|
|
> cursor_line_start
|
|
+ self.env["screen"]["new_cursor"]["x"]
|
|
+ 3
|
|
):
|
|
cursor_line_end_offset = (
|
|
cursor_line_start
|
|
+ self.env["screen"]["new_cursor"]["x"]
|
|
+ 3
|
|
)
|
|
|
|
# Extract just the relevant text sections for
|
|
# character-level diff
|
|
old_screen_text = self.env["screen"]["old_content_text"][
|
|
cursor_line_start_offset:cursor_line_end_offset
|
|
]
|
|
new_screen_text = self.env["screen"]["new_content_text"][
|
|
cursor_line_start_offset:cursor_line_end_offset
|
|
]
|
|
|
|
# Character-level diff for typing detection (not
|
|
# line-level)
|
|
diff = self.differ.compare(
|
|
old_screen_text, new_screen_text
|
|
)
|
|
diff_list = list(diff)
|
|
typing = True
|
|
|
|
# Validate typing assumption by checking if detected
|
|
# changes match cursor movement
|
|
temp_new_delta = "".join(
|
|
x[2:] for x in diff_list if x[0] == "+"
|
|
)
|
|
if temp_new_delta.strip() != "":
|
|
# Compare diff result against expected typing pattern
|
|
# Expected: characters between old and new cursor
|
|
# positions
|
|
expected_typing = "".join(
|
|
new_screen_text[
|
|
self.env["screen"]["old_cursor"][
|
|
"x"
|
|
] : self.env["screen"]["new_cursor"]["x"]
|
|
].rstrip()
|
|
)
|
|
|
|
# If diff doesn't match expected typing pattern, treat
|
|
# as general screen change
|
|
if temp_new_delta != expected_typing:
|
|
# Fallback: treat entire current line as new
|
|
# content
|
|
diff_list = [
|
|
"+ "
|
|
+ self.env["screen"]["new_content_text"].split(
|
|
"\n"
|
|
)[self.env["screen"]["new_cursor"]["y"]]
|
|
+ "\n"
|
|
]
|
|
typing = False
|
|
else:
|
|
# GENERAL SCREEN CHANGE DETECTION
|
|
# Not typing - handle as line-by-line content change
|
|
# This catches: incoming messages, screen updates,
|
|
# application output, etc.
|
|
|
|
# Pre-process screen text for comparison - collapse multiple spaces to single space
|
|
# This normalization prevents spurious diffs from spacing
|
|
# inconsistencies
|
|
old_screen_text = self._space_normalize_regex.sub(
|
|
" ",
|
|
self.env["runtime"][
|
|
"ScreenManager"
|
|
].get_window_area_in_text(
|
|
self.env["screen"]["old_content_text"]
|
|
),
|
|
)
|
|
new_screen_text = self._space_normalize_regex.sub(
|
|
" ",
|
|
self.env["runtime"][
|
|
"ScreenManager"
|
|
].get_window_area_in_text(
|
|
self.env["screen"]["new_content_text"]
|
|
),
|
|
)
|
|
|
|
diff = self.differ.compare(
|
|
old_screen_text.split("\n"),
|
|
new_screen_text.split("\n"),
|
|
)
|
|
diff_list = list(diff)
|
|
|
|
# Extract added and removed content from diff results
|
|
# Output format depends on whether this was detected as typing
|
|
# or general change
|
|
if not typing:
|
|
# Line-based changes: join with newlines for proper speech
|
|
# cadence
|
|
self.env["screen"]["new_delta"] = "\n".join(
|
|
x[2:] for x in diff_list if x[0] == "+"
|
|
)
|
|
else:
|
|
# Character-based changes: no newlines for smooth typing
|
|
# echo
|
|
self.env["screen"]["new_delta"] = "".join(
|
|
x[2:] for x in diff_list if x[0] == "+"
|
|
)
|
|
|
|
# Negative delta (removed content) - used for backspace/delete
|
|
# detection
|
|
self.env["screen"]["newNegativeDelta"] = "".join(
|
|
x[2:] for x in diff_list if x[0] == "-"
|
|
)
|
|
|
|
# track highlighted
|
|
try:
|
|
if self.env["runtime"]["AttributeManager"].is_attribute_change():
|
|
if self.env["runtime"]["SettingsManager"].get_setting_as_bool(
|
|
"focus", "highlight"
|
|
):
|
|
attribute_delta, attributeCursor = self.env["runtime"][
|
|
"AttributeManager"
|
|
].track_highlights()
|
|
if attributeCursor:
|
|
self.env["runtime"][
|
|
"AttributeManager"
|
|
].set_attribute_cursor(attributeCursor)
|
|
self.env["runtime"][
|
|
"AttributeManager"
|
|
].set_attribute_delta(attribute_delta)
|
|
except Exception as e:
|
|
self.env["runtime"]["DebugManager"].write_debug_out(
|
|
"ScreenManager:update:highlight: " + str(e),
|
|
debug.DebugLevel.ERROR,
|
|
)
|
|
|
|
def is_ignored_screen(self, screen=None):
|
|
if screen is None:
|
|
screen = self.env["screen"]["newTTY"]
|
|
# Check if force all screens flag is set
|
|
if self.env["runtime"].get("force_all_screens", False):
|
|
return False
|
|
ignore_screens = []
|
|
fix_ignore_screens = self.env["runtime"][
|
|
"SettingsManager"
|
|
].get_setting("screen", "ignoreScreen")
|
|
if fix_ignore_screens != "":
|
|
ignore_screens.extend(fix_ignore_screens.split(","))
|
|
if self.env["runtime"]["SettingsManager"].get_setting_as_bool(
|
|
"screen", "autodetectIgnoreScreen"
|
|
):
|
|
ignore_screens.extend(self.env["screen"]["autoIgnoreScreens"])
|
|
self.env["runtime"]["DebugManager"].write_debug_out(
|
|
"ScreenManager:is_ignored_screen ignore:"
|
|
+ str(ignore_screens)
|
|
+ " current:"
|
|
+ str(screen),
|
|
debug.DebugLevel.INFO,
|
|
)
|
|
return screen in ignore_screens
|
|
|
|
def is_screen_change(self):
|
|
if not self.env["screen"]["oldTTY"]:
|
|
return False
|
|
return self.env["screen"]["newTTY"] != self.env["screen"]["oldTTY"]
|
|
|
|
def is_delta(self, ignoreSpace=False):
|
|
new_delta = self.env["screen"]["new_delta"]
|
|
if ignoreSpace:
|
|
new_delta = new_delta.strip()
|
|
return new_delta != ""
|
|
|
|
def is_negative_delta(self):
|
|
return self.env["screen"]["newNegativeDelta"] != ""
|
|
|
|
def get_window_area_in_text(self, text):
|
|
if not self.env["runtime"][
|
|
"CursorManager"
|
|
].is_application_window_set():
|
|
return text
|
|
window_text = ""
|
|
window_list = text.split("\n")
|
|
curr_app = self.env["runtime"][
|
|
"ApplicationManager"
|
|
].get_current_application()
|
|
window_list = window_list[
|
|
self.env["commandBuffer"]["windowArea"][curr_app]["1"][
|
|
"y"
|
|
] : self.env["commandBuffer"]["windowArea"][curr_app]["2"]["y"]
|
|
+ 1
|
|
]
|
|
for line in window_list:
|
|
window_text += (
|
|
line[
|
|
self.env["commandBuffer"]["windowArea"][curr_app]["1"][
|
|
"x"
|
|
] : self.env["commandBuffer"]["windowArea"][curr_app]["2"][
|
|
"x"
|
|
]
|
|
+ 1
|
|
]
|
|
+ "\n"
|
|
)
|
|
return window_text
|
|
|
|
def inject_text_to_screen(self, text, screen=None):
|
|
try:
|
|
self.env["runtime"]["ScreenDriver"].inject_text_to_screen(
|
|
text, screen
|
|
)
|
|
except Exception as e:
|
|
self.env["runtime"]["DebugManager"].write_debug_out(
|
|
"ScreenManager:inject_text_to_screen " + str(e),
|
|
debug.DebugLevel.ERROR,
|
|
)
|