itdominator/Newton-Editor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add debounced text change handling and modified file indicator for tabs

Browse Source 
- Debounce word completer refresh with 1500ms timeout to reduce overhead
- Add visual indicator (file-changed class) for modified files in tabs
- Refactor buffer switching into signal_mapper for DRY code
- Fix handler ID indices after adding _after_changed signal
- Move set_modified(False) after successful file write in save()
This commit is contained in:
itdominator
2026-02-22 18:22:47 -06:00
parent 3ab71d6441
commit 88a6451fa8
15 changed files with 385 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
plugins/depricated/colorize/__init__.py → plugins/code/colorize/__init__.py
View File

0
plugins/depricated/colorize/__main__.py → plugins/code/colorize/__main__.py
View File

22
plugins/depricated/colorize/color_converter_mixin.py → plugins/code/colorize/color_converter_mixin.py
View File

@@ -6,11 +6,17 @@ import colorsys
# Application imports
class ColorConverterMixinException(Exception):
...
class ColorConverterMixin:
# NOTE: HSV HSL, and Hex Alpha parsing are available in Gtk 4.0- not lower.
# So, for compatability we're gunna convert to rgba string ourselves...
def get_color_text(self, buffer, start, end):
text = buffer.get_text(start, end, include_hidden_chars = False)
def get_color_text(self, buffer, start_itr, end_itr):
text = buffer.get_text(start_itr, end_itr, include_hidden_chars = False)
try:
if "hsl" in text:
@@ -24,14 +30,14 @@ class ColorConverterMixin:
size = len(hex)
if size in [4, 8, 16]:
rgba = self.hex_to_rgba(hex, size)
print(rgba)
logger.debug(f"Colorize Plugin: RGBA = {rgba}")
except Exception as e:
except ColorConverterMixinException as e:
...
return text
def hex_to_rgba(self, hex, size):
def hex_to_rgba(self, hex: str, size: int) -> str:
rgba = []
slots = None
step = 2
@@ -60,12 +66,10 @@ class ColorConverterMixin:
rgb_sub = ','.join(map(str, tuple(rgba)))
return f"rgba({rgb_sub})"
# return tuple(rgba)
def hsl_to_rgb(self, text):
def hsl_to_rgb(self, text: str) -> str:
_h, _s , _l = text.replace("hsl", "") \
.replace("deg", "") \
.replace("(", "") \
@@ -86,7 +90,7 @@ class ColorConverterMixin:
return f"rgb({rgb_sub})"
def hsv_to_rgb(self, text):
def hsv_to_rgb(self, text: str) -> str:
_h, _s , _v = text.replace("hsv", "") \
.replace("deg", "") \
.replace("(", "") \

242
plugins/code/colorize/colorize.py Normal file
View File

@@ -0,0 +1,242 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('Gdk', '3.0')
from gi.repository import Gtk
from gi.repository import Gdk
# Application imports
from .color_converter_mixin import ColorConverterMixin
class Colorize(ColorConverterMixin):
def __init__(self):
super(Colorize, self).__init__()
self.tag_stub_name: str = "colorize-tag"
self.is_colorize_paused: bool = True
def handle_colorize(self, buffer):
if self.is_colorize_paused: return
tag_table = buffer.get_tag_table()
start_itr = None
end_itr = buffer.get_iter_at_mark( buffer.get_insert() )
i = 0
walker_iter = end_itr.copy()
working_tag = self.find_working_tag(walker_iter, i)
if working_tag:
start_itr = self.find_start_range(walker_iter, working_tag)
self.find_end_range(end_itr, working_tag)
buffer.remove_tag(working_tag, start_itr, end_itr)
else:
start_itr = self.traverse_backward_25_or_less(walker_iter)
self.traverse_forward_25_or_less(end_itr)
self.do_colorize(buffer, start_itr, end_itr)
def do_colorize(self, buffer = None, start_itr = None, end_itr = None):
if not start_itr or not end_itr:
start_itr = buffer.get_start_iter()
end_itr = buffer.get_end_iter()
# rgb(a), hsl, hsv
results = self.finalize_non_hex_matches(
self.collect_preliminary_results(
buffer, start_itr, end_itr
)
)
self.process_results(buffer, results)
# hex color search
results = self.finalize_hex_matches(
self.collect_preliminary_hex_results(
buffer, start_itr, end_itr
)
)
self.process_results(buffer, results)
def collect_preliminary_results(self, buffer = None, start_itr = None, end_itr = None):
if not buffer: return []
if not start_itr:
start_itr = buffer.get_start_iter()
results1 = self.search(start_itr, end_itr, "rgb")
results2 = self.search(start_itr, end_itr, "hsl")
results3 = self.search(start_itr, end_itr, "hsv")
return results1 + results2 + results3
def find_working_tag(self, walker_iter, i):
tags = walker_iter.get_tags()
for tag in tags:
if not tag.props.name or not self.tag_stub_name in tag.props.name: continue
return tag
result = walker_iter.backward_char()
if not result: return
if i > 25: return
return self.find_working_tag(walker_iter, i + 1)
def find_start_range(self, walker_iter, working_tag):
tags = walker_iter.get_tags()
for tag in tags:
if not tag.props.name or not working_tag.props.name in tag.props.name: continue
if not walker_iter.backward_char(): continue
self.find_start_range(walker_iter, working_tag)
return walker_iter
def find_end_range(self, end_itr, working_tag):
tags = end_itr.get_tags()
for tag in tags:
if not tag.props.name or not working_tag.props.name in tag.props.name: continue
if not end_itr.forward_char(): continue
self.find_end_range(end_itr, working_tag)
def traverse_backward_25_or_less(self, walker_itr):
i = 1
while i <= 25:
res = walker_itr.backward_char()
if not res: break
i += 1
def traverse_forward_25_or_less(self, end_itr):
i = 1
while i <= 25:
res = end_itr.forward_char()
if not res: break
i += 1
def collect_preliminary_hex_results(
self,
buffer = None,
start_itr = None,
end_itr = None
) -> list:
if not buffer: return []
if not start_itr:
start_itr = buffer.get_start_iter()
results = self.search(start_itr, end_itr, "#")
return results
def search(self, start_itr = None, end_itr = None, query: str = None) -> list:
if not start_itr or not query: return None, None
results: list = []
flags = Gtk.TextSearchFlags.VISIBLE_ONLY | Gtk.TextSearchFlags.TEXT_ONLY
while True:
result = start_itr.forward_search(query, flags, end_itr)
if not result: break
results.append(result)
start_itr = result[1]
return results
def finalize_non_hex_matches(self, result_hits: [] = []) -> list:
results: list = []
for start_itr, end_itr in result_hits:
# If one of end chars of rgb/rgba/hsv/hsl
if end_itr.get_char() in ["a", "b", "l", "v"]:
end_itr.forward_char()
# If afterwards no paren
if end_itr.get_char() != "(":
continue
end_itr.forward_chars(21) # Check if best case (255, 255, 255, 0.64)
if end_itr.get_char() == ")":
end_itr.forward_char()
results.append([start, end_itr])
continue
# Break loop if we get back to rgb/rgba/hsl/hsv -> (
while end_itr.get_char() != "(":
if end_itr.get_char() == ")":
end_itr.forward_char()
results.append([start_itr, end_itr])
break
if not end_itr.backward_char(): break
return results
def finalize_hex_matches(self, result_hits: [] = []) -> list:
results: list = []
for start_itr, end_itr in result_hits:
i = 0
_ch = end_itr.get_char()
ch = ord(end_itr.get_char()) if _ch else -1
while (
(ch >= 48 and ch <= 57) or \
(ch >= 65 and ch <= 70) or \
(ch >= 97 and ch <= 102)
):
if i > 16: break
i += 1
end_itr.forward_char()
_ch = end_itr.get_char()
ch = ord(end_itr.get_char()) if _ch else -1
if i in [3, 4, 6, 8, 9, 12, 16]:
results.append([start_itr, end_itr])
return results
def process_results(self, buffer, results):
for start_itr, end_itr in results:
text = self.get_color_text(buffer, start_itr, end_itr)
color = Gdk.RGBA()
if not color.parse(text): continue
tag = self.get_colorized_tag(buffer, text, color)
buffer.apply_tag(tag, start_itr, end_itr)
def get_colorized_tag(self, buffer, tag, color: Gdk.RGBA):
tag_table = buffer.get_tag_table()
colorize_tag = f"{self.tag_stub_name}_{tag}"
search_tag = tag_table.lookup(colorize_tag)
if not search_tag:
search_tag = buffer.create_tag(
colorize_tag, background_rgba = color
)
return search_tag
def clear_color_tags(self, buffer):
tag_table = buffer.get_tag_table()
def traverse_tags(tag, user_data):
name = tag.get_property("name")
if not name: return
if name.startswith(self.tag_stub_name):
user_data.append(tag)
tags = []
tag_table.foreach(traverse_tags, tags)
for tag in tags:
tag_table.remove(tag)

4
plugins/depricated/colorize/manifest.json → plugins/code/colorize/manifest.json
View File

@@ -3,7 +3,5 @@
"author": "ITDominator",
"version": "0.0.1",
"support": "",
"requests": {
"pass_events": true
}
"requests": {}
}

56
plugins/code/colorize/plugin.py Normal file
View File

@@ -0,0 +1,56 @@
# Python imports
# Lib imports
# Application imports
from libs.event_factory import Event_Factory, Code_Event_Types
from plugins.plugin_types import PluginCode
from .colorize import Colorize
colorize = Colorize()
class Plugin(PluginCode):
def __init__(self):
super(Plugin, self).__init__()
def _controller_message(self, event: Code_Event_Types.CodeEvent):
if isinstance(event, Code_Event_Types.AddedNewFileEvent):
colorize.handle_colorize(event.file.buffer)
elif isinstance(event, Code_Event_Types.TextChangedEvent):
colorize.handle_colorize(event.buffer)
def load(self):
event = Event_Factory.create_event("register_command",
command_name = "tggle_colorize",
command = Handler,
binding_mode = "released",
binding = "<Shift><Control>c"
)
self.message_to("source_views", event)
def run(self):
...
class Handler:
@staticmethod
def execute(
view: any
):
logger.debug("Command: Toggle Colorize")
colorize.is_colorize_paused = not colorize.is_colorize_paused
if colorize.is_colorize_paused:
colorize.clear_color_tags( view.get_buffer() )
return
colorize.handle_colorize( view.get_buffer() )

14
plugins/code/completers/words_completer/provider_response_cache.py
View File

@@ -20,6 +20,7 @@ class ProviderResponseCache(ProviderResponseCacheBase):
super(ProviderResponseCache, self).__init__()
self.matchers: dict = {}
self._temp_timeout_id: int = None
def process_file_load(self, event: Code_Event_Types.AddedNewFileEvent):
@@ -36,9 +37,22 @@ class ProviderResponseCache(ProviderResponseCacheBase):
def process_file_change(self, event: Code_Event_Types.TextChangedEvent):
buffer = event.file.buffer
self._clear_temp_delay()
self._set_temp_delay(buffer)
def _clear_temp_delay(self):
if self._temp_timeout_id:
GLib.source_remove(self._temp_timeout_id)
def _set_temp_delay(self, buffer):
def run_refresh_update(buffer):
with ThreadPoolExecutor(max_workers = 1) as executor:
executor.submit(self._handle_change, buffer)
return False
self._temp_timeout_id = GLib.timeout_add(1500, run_refresh_update, buffer)
def _handle_change(self, buffer):
start_itr = buffer.get_start_iter()
end_itr = buffer.get_end_iter()

228
plugins/depricated/colorize/plugin.py
View File

@@ -1,228 +0,0 @@
# Python imports
import random
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('Gdk', '3.0')
from gi.repository import Gtk
from gi.repository import Gdk
# Application imports
from plugins.plugin_base import PluginBase
from .color_converter_mixin import ColorConverterMixin
class Plugin(ColorConverterMixin, PluginBase):
def __init__(self):
super().__init__()
self.name = "Colorize" # NOTE: Need to remove after establishing private bidirectional 1-1 message bus
# where self.name should not be needed for message comms
self.tag_stub_name = "colorize_tag"
self._buffer = None
def run(self):
...
def generate_reference_ui_element(self):
...
def subscribe_to_events(self):
self._event_system.subscribe("set_active_src_view", self._set_active_src_view)
self._event_system.subscribe("buffer_changed_first_load", self._buffer_changed_first_load)
self._event_system.subscribe("buffer_changed", self._buffer_changed)
def _set_active_src_view(self, source_view):
self._active_src_view = source_view
def _buffer_changed_first_load(self, buffer):
self._buffer = buffer
self._do_colorize(buffer)
def _buffer_changed(self, buffer):
self._event_system.emit("pause_event_processing")
self._handle_colorize(buffer)
self._event_system.emit("resume_event_processing")
def _handle_colorize(self, buffer):
self._buffer = buffer
tag_table = buffer.get_tag_table()
mark = buffer.get_insert()
start = None
end = buffer.get_iter_at_mark(mark)
i = 0
walker_iter = end.copy()
working_tag = self.find_working_tag(walker_iter, i)
if working_tag:
start = self.find_start_range(walker_iter, working_tag)
self.find_end_range(end, working_tag)
buffer.remove_tag(working_tag, start, end)
else:
start = self.traverse_backward_25_or_less(walker_iter)
self.traverse_forward_25_or_less(end)
self._do_colorize(buffer, start, end)
def find_working_tag(self, walker_iter, i):
tags = walker_iter.get_tags()
for tag in tags:
if tag.props.name and self.tag_stub_name in tag.props.name:
return tag
res = walker_iter.backward_char()
if not res: return
if i > 25: return
return self.find_working_tag(walker_iter, i + 1)
def find_start_range(self, walker_iter, working_tag):
tags = walker_iter.get_tags()
for tag in tags:
if tag.props.name and working_tag.props.name in tag.props.name:
res = walker_iter.backward_char()
if res:
self.find_start_range(walker_iter, working_tag)
return walker_iter
def find_end_range(self, end, working_tag):
tags = end.get_tags()
for tag in tags:
if tag.props.name and working_tag.props.name in tag.props.name:
res = end.forward_char()
if res:
self.find_end_range(end, working_tag)
def traverse_backward_25_or_less(self, walker_iter):
i = 1
while i <= 25:
res = walker_iter.backward_char()
if not res: break
i += 1
def traverse_forward_25_or_less(self, end):
i = 1
while i <= 25:
res = end.forward_char()
if not res: break
i += 1
def _do_colorize(self, buffer = None, start_itr = None, end_itr = None):
# rgb(a), hsl, hsv
results = self.finalize_non_hex_matches( self.collect_preliminary_results(buffer, start_itr, end_itr) )
self.process_results(buffer, results)
# hex color search
results = self.finalize_hex_matches( self.collect_preliminary_hex_results(buffer, start_itr, end_itr) )
self.process_results(buffer, results)
def collect_preliminary_results(self, buffer = None, start_itr = None, end_itr = None):
if not buffer: return []
if not start_itr:
start_itr = buffer.get_start_iter()
results1 = self.search(start_itr, end_itr, "rgb")
results2 = self.search(start_itr, end_itr, "hsl")
results3 = self.search(start_itr, end_itr, "hsv")
return results1 + results2 + results3
def collect_preliminary_hex_results(self, buffer = None, start_itr = None, end_itr = None):
if not buffer: return []
if not start_itr:
start_itr = buffer.get_start_iter()
results1 = self.search(start_itr, end_itr, "#")
return results1
def search(self, start_itr = None, end_itr = None, query = None):
if not start_itr or not query: return None, None
results = []
flags = Gtk.TextSearchFlags.VISIBLE_ONLY | Gtk.TextSearchFlags.TEXT_ONLY
while True:
result = start_itr.forward_search(query, flags, end_itr)
if not result: break
results.append(result)
start_itr = result[1]
return results
def finalize_non_hex_matches(self, result_hits: [] = []):
results = []
for start, end in result_hits:
if end.get_char() == "a":
end.forward_char()
if end.get_char() != "(":
continue
end.forward_chars(21)
if end.get_char() == ")":
end.forward_char()
results.append([start, end])
continue
while end.get_char() != "(":
if end.get_char() == ")":
end.forward_char()
results.append([start, end])
break
end.forward_chars(-1)
return results
def finalize_hex_matches(self, result_hits: [] = []):
results = []
for start, end in result_hits:
i = 0
_ch = end.get_char()
ch = ord(end.get_char()) if _ch else -1
while ((ch >= 48 and ch <= 57) or (ch >= 65 and ch <= 70) or (ch >= 97 and ch <= 102)):
if i > 16: break
i += 1
end.forward_char()
_ch = end.get_char()
ch = ord(end.get_char()) if _ch else -1
if i in [3, 4, 6, 8, 9, 12, 16]:
results.append([start, end])
return results
def process_results(self, buffer, results):
for start, end in results:
text = self.get_color_text(buffer, start, end)
color = Gdk.RGBA()
if color.parse(text):
tag = self.get_colorized_tag(buffer, text, color)
buffer.apply_tag(tag, start, end)
def get_colorized_tag(self, buffer, tag, color: Gdk.RGBA):
tag_table = buffer.get_tag_table()
colorize_tag = f"{self.tag_stub_name}_{tag}"
search_tag = tag_table.lookup(colorize_tag)
if not search_tag:
search_tag = buffer.create_tag(colorize_tag, background_rgba = color)
return search_tag

6
src/core/widgets/code/controllers/tabs_controller.py
View File

@@ -26,11 +26,11 @@ class TabsController(ControllerBase):
def _controller_message(self, event: Code_Event_Types.CodeEvent):
if isinstance(event, Code_Event_Types.FocusedViewEvent):
self.tabs_widget.view_changed(
event.view.get_buffer()
)
self.tabs_widget.view_changed( event.view.get_buffer() )
elif isinstance(event, Code_Event_Types.FilePathSetEvent):
self.update_tab_label(event)
elif isinstance(event, Code_Event_Types.ModifiedChangedEvent):
self.tabs_widget.modified_changed( event.buffer )
elif isinstance(event, Code_Event_Types.AddedNewFileEvent):
self.add_tab(event)
elif isinstance(event, Code_Event_Types.PoppedFileEvent):

4
src/core/widgets/code/controllers/views/signal_mapper.py
View File

@@ -18,6 +18,10 @@ class SourceViewSignalMapper:
def set_state_manager(self, state_manager):
self.state_manager = state_manager
def set_buffer_to_active_view(self, buffer):
self.active_view.set_buffer(buffer)
self.active_view.command.exec("update_info_bar")
def connect_signals(self, source_view: SourceView):
signal_mappings = self._get_signal_mappings()
for signal, handler in signal_mappings.items():

7
src/core/widgets/code/controllers/views/source_views_controller.py
View File

@@ -28,17 +28,14 @@ class SourceViewsController(ControllerBase, list):
if isinstance(event, Code_Event_Types.RemovedFileEvent):
self._remove_file(event)
elif isinstance(event, Code_Event_Types.RegisterCommandEvent):
self. _register_command(event)
self._register_command(event)
if not self.signal_mapper.active_view: return
if isinstance(event, Code_Event_Types.TextChangedEvent):
if not self.signal_mapper.active_view: return
self.signal_mapper.active_view.command.exec("update_info_bar")
elif isinstance(event, Code_Event_Types.SetActiveFileEvent):
self.signal_mapper.active_view.set_buffer(
event.buffer
)
self.signal_mapper.set_buffer_to_active_view(event.buffer)
elif isinstance(event, Code_Event_Types.TextInsertedEvent):
self.signal_mapper.insert_text(event.file, event.text)

16
src/core/widgets/code/source_buffer.py
View File

@@ -26,6 +26,7 @@ class SourceBuffer(GtkSource.Buffer):
def set_signals(
self,
_changed,
_after_changed,
_mark_set,
_insert_text,
_after_insert_text,
@@ -34,6 +35,7 @@ class SourceBuffer(GtkSource.Buffer):
self._handler_ids = [
self.connect("changed", _changed),
self.connect_after("changed", _after_changed),
self.connect("mark-set", _mark_set),
self.connect("insert-text", _insert_text),
self.connect_after("insert-text", _after_insert_text),
@@ -43,20 +45,26 @@ class SourceBuffer(GtkSource.Buffer):
def block_changed_signal(self):
self.handler_block(self._handler_ids[0])
def block_changed_after_signal(self):
self.handler_block(self._handler_ids[1])
def block_insert_after_signal(self):
self.handler_block(self._handler_ids[3])
self.handler_block(self._handler_ids[4])
def block_modified_changed_signal(self):
self.handler_block(self._handler_ids[4])
self.handler_block(self._handler_ids[5])
def unblock_changed_signal(self):
self.handler_unblock(self._handler_ids[0])
def unblock_changed_after_signal(self):
self.handler_unblock(self._handler_ids[1])
def unblock_insert_after_signal(self):
self.handler_unblock(self._handler_ids[3])
self.handler_unblock(self._handler_ids[4])
def unblock_modified_changed_signal(self):
self.handler_unblock(self._handler_ids[4])
self.handler_unblock(self._handler_ids[5])
def clear_signals(self):
for handle_id in self._handler_ids:

12
src/core/widgets/code/source_file.py
View File

@@ -27,7 +27,6 @@ class SourceFile(GtkSource.File):
self.fname: str = "buffer"
self.fpath: str = "buffer"
self.ftype: str = "buffer"
self.buffer: SourceBuffer = SourceBuffer()
self._set_signals()
@@ -36,6 +35,7 @@ class SourceFile(GtkSource.File):
def _set_signals(self):
self.buffer.set_signals(
self._changed,
self._after_changed,
self._mark_set,
self._insert_text,
self._after_insert_text,
@@ -43,6 +43,9 @@ class SourceFile(GtkSource.File):
)
def _changed(self, buffer: SourceBuffer):
...
def _after_changed(self, buffer: SourceBuffer):
self.check_file_on_disk()
event = Event_Factory.create_event(
@@ -53,7 +56,7 @@ class SourceFile(GtkSource.File):
self.emit(event)
if self.is_deleted():
print("deleted")
print("is_deleted")
# event = Event_Factory.create_event("file_deleted", buffer = buffer)
# event.file = self
# self.emit(event)
@@ -138,6 +141,7 @@ class SourceFile(GtkSource.File):
undo_manager.begin_not_undoable_action()
self.buffer.insert_at_cursor(data)
undo_manager.end_not_undoable_action()
self.buffer.set_modified(False)
def set_path(self, gfile: Gio.File):
if not gfile: return
@@ -150,13 +154,15 @@ class SourceFile(GtkSource.File):
self.emit(event)
def save(self):
self._write_file( self.get_location() )
self.buffer.set_modified(False)
event = Event_Factory.create_event(
"saved_file",
file = self, buffer = self.buffer
)
self.emit(event)
self._write_file( self.get_location() )
def save_as(self):
file = event_system.emit_and_await("save-file-dialog")

36
src/core/widgets/code/tabs_widget.py
View File

@@ -65,19 +65,6 @@ class TabsWidget(Gtk.Notebook):
self.message(event)
def view_changed(self, buffer):
for page_widget in self.get_children():
tab = self.get_tab_label(page_widget)
if not buffer == tab.file.buffer: continue
self.handler_block(self.switch_page_id)
self.set_current_page(
self.page_num(page_widget)
)
self.handler_unblock(self.switch_page_id)
def _bind_tab_menu(self, tab, page_widget):
def do_context_menu(tab, eve, page_widget):
if eve.type == Gdk.EventType.BUTTON_RELEASE and eve.button == 3: # r-click
@@ -116,6 +103,29 @@ class TabsWidget(Gtk.Notebook):
return context_menu
def view_changed(self, buffer):
for page_widget in self.get_children():
tab = self.get_tab_label(page_widget)
if not buffer == tab.file.buffer: continue
self.handler_block(self.switch_page_id)
self.set_current_page(
self.page_num(page_widget)
)
self.handler_unblock(self.switch_page_id)
def modified_changed(self, buffer):
for page_widget in self.get_children():
tab = self.get_tab_label(page_widget)
if not buffer == tab.file.buffer: continue
ctx = tab.label.get_style_context()
if buffer.get_modified():
ctx.add_class("file-changed")
else:
ctx.remove_class("file-changed")
def close_item(self, menu_item, page_widget):
tab = self.get_tab_label(page_widget)