Newton_Editor/plugins/lsp_client/plugin.py

124 lines
4.3 KiB
Python
Raw Normal View History

2023-11-04 19:38:20 +00:00
# Python imports
2023-11-05 05:36:55 +00:00
import os
import json
import threading
2023-11-04 19:38:20 +00:00
# Lib imports
from gi.repository import GLib
2023-11-04 19:38:20 +00:00
# Application imports
from plugins.plugin_base import PluginBase
from .lsp_controller import LSPController
class LSPPliginException(Exception):
...
2023-11-04 19:38:20 +00:00
class Plugin(PluginBase):
def __init__(self):
super().__init__()
2023-11-05 05:36:55 +00:00
self.name = "LSP Client" # NOTE: Need to remove after establishing private bidirectional 1-1 message bus
# where self.name should not be needed for message comms
self.lsp_config_path: str = os.path.dirname(os.path.realpath(__file__)) + "/../../lsp_servers_config.json"
self.lsp_servers_config: dict = {}
self.lsp_controller = None
self.timer = None
2023-11-04 19:38:20 +00:00
def generate_reference_ui_element(self):
...
def run(self):
2023-11-05 05:36:55 +00:00
if os.path.exists(self.lsp_config_path):
with open(self.lsp_config_path, "r") as f:
self.lsp_servers_config = json.load(f)
else:
text = f"LSP NOT Enabled.\nFile:\n\t{self.lsp_config_path}\ndoes no exsist..."
2023-11-04 19:38:20 +00:00
self._event_system.emit("bubble_message", ("warning", self.name, text,))
2023-11-07 05:01:18 +00:00
return
if len(self.lsp_servers_config.keys()) > 0:
self.lsp_controller = LSPController(self.lsp_servers_config)
self.inner_subscribe_to_events()
2023-11-04 19:38:20 +00:00
def subscribe_to_events(self):
2023-11-07 05:01:18 +00:00
...
def inner_subscribe_to_events(self):
2023-11-04 19:38:20 +00:00
self._event_system.subscribe("shutting_down", self._shutting_down)
self._event_system.subscribe("textDocument/didChange", self._buffer_changed)
2023-11-07 05:01:18 +00:00
self._event_system.subscribe("textDocument/didOpen", self.lsp_controller.do_open)
self._event_system.subscribe("textDocument/didSave", self.lsp_controller.do_save)
self._event_system.subscribe("textDocument/didClose", self.lsp_controller.do_close)
self._event_system.subscribe("textDocument/definition", self._do_goto)
self._event_system.subscribe("textDocument/completion", self.completion)
2023-11-04 19:38:20 +00:00
def _shutting_down(self):
self.lsp_controller._shutting_down()
2023-11-04 19:38:20 +00:00
def cancel_timer(self):
if self.timer:
self.timer.cancel()
GLib.idle_remove_by_data(None)
2023-11-04 19:38:20 +00:00
def delay_completion_glib(self, source_view, context, callback):
GLib.idle_add(self._do_completion, source_view, context, callback)
2023-11-04 19:38:20 +00:00
def delay_completion(self, source_view, context, callback):
self.timer = threading.Timer(0.8, self.delay_completion_glib, (source_view, context, callback,))
self.timer.daemon = True
self.timer.start()
def _buffer_changed(self, language_id, buffer):
iter = buffer.get_iter_at_mark( buffer.get_insert() )
line = iter.get_line()
start = iter.copy()
end = iter.copy()
start.backward_line()
start.forward_line()
end.forward_to_line_end()
2023-11-05 05:36:55 +00:00
text = buffer.get_text(start, end, include_hidden_chars = False)
result = self.lsp_controller.do_change(language_id, line, start, end, text)
# print(result)
2023-11-07 05:01:18 +00:00
def completion(self, source_view, context, callback):
self.cancel_timer()
self.delay_completion(source_view, context, callback)
2023-11-05 05:36:55 +00:00
def _do_completion(self, source_view, context, callback):
filepath = source_view.get_current_filepath()
if not filepath: return
uri = filepath.get_uri()
buffer = source_view.get_buffer()
iter = buffer.get_iter_at_mark( buffer.get_insert() )
line = iter.get_line() + 1
_char = iter.get_char()
if iter.backward_char():
_char = iter.get_char()
offset = iter.get_line_offset()
result = self.lsp_controller.do_completion(source_view.get_filetype(), uri, line, offset, _char)
callback(context, result)
2023-11-05 05:36:55 +00:00
2023-11-07 05:01:18 +00:00
def _do_goto(self, language_id, uri, line, offset):
results = self.lsp_controller.do_goto(language_id, uri, line, offset)
if len(results) == 1:
result = results[0]
file = result.uri[7:]
line = result.range.end.line
message = f"FILE|{file}:{line}"
self._event_system.emit("post_file_to_ipc", message)