Newton_Editor/plugins/gtksourceview/lsp_manager/plugin.py

# Python imports
import signal
import subprocess
import json

# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk

# Application imports
from plugins.plugin_base import PluginBase
from .client_ipc import ClientIPC



class Plugin(PluginBase):
    def __init__(self):

        super().__init__()
        self.name             = "LSP Manager"  # NOTE: Need to remove after establishing private bidirectional 1-1 message bus
                                               #       where self.name should not be needed for message comms
        self.ws_config_file   = "config.json"
        self.ws_config: dict  = {}
        self.lsp_manager_proc = None
        self.lsp_window       = None


    def generate_reference_ui_element(self):
        ...

    def run(self):
        try:
            with open(self.ws_config_file) as f:
                self.ws_config = json.load(f)["websocket"]
        except Exception as e:
            raise Exception(f"Couldn't load config.json...\n{repr(e)}")

        self.lsp_window = Gtk.Window()
        box1            = Gtk.Box()
        box2            = Gtk.Box()
        start_btn       = Gtk.Button(label = "Start LSP Manager")
        stop_btn        = Gtk.Button(label = "Stop LSP Manager")
        pid_label       = Gtk.Label(label  = "LSP PID: ")

        box1.set_orientation( Gtk.Orientation.VERTICAL )

        self.lsp_window.set_deletable(False)
        self.lsp_window.set_skip_pager_hint(True)
        self.lsp_window.set_skip_taskbar_hint(True)
        self.lsp_window.set_title("LSP Manager")
        self.lsp_window.set_size_request(480, 320)

        start_btn.connect("clicked", self.start_lsp_manager)
        stop_btn.connect("clicked", self.stop_lsp_manager)

        box1.add(pid_label)
        box2.add(start_btn)
        box2.add(stop_btn)
        box1.add(box2)
        self.lsp_window.add(box1)

        box1.show_all()

        self.inner_subscribe_to_events()

    def _shutting_down(self):
        self.stop_lsp_manager()

    def _tear_down(self, widget, eve):
        return True

    def _tggl_lsp_window(self):
        if not self.lsp_window.is_visible():
            self.lsp_window.show()
        else:
            self.lsp_window.hide()

    def subscribe_to_events(self):
        self._event_system.subscribe("tggl_lsp_window", self._tggl_lsp_window)

    def inner_subscribe_to_events(self):
        self._event_system.subscribe("shutting_down", self._shutting_down)

        self._event_system.subscribe("textDocument/didOpen",    self._lsp_did_open)
        self._event_system.subscribe("textDocument/didSave",    self._lsp_did_save)
        self._event_system.subscribe("textDocument/didClose",   self._lsp_did_close)
        self._event_system.subscribe("textDocument/didChange",  self._lsp_did_change)
        self._event_system.subscribe("textDocument/definition", self._lsp_goto)
        self._event_system.subscribe("textDocument/completion", self._lsp_completion)

    def start_lsp_manager(self, button):
        if self.lsp_manager_proc: return
        self.lsp_manager_proc = subprocess.Popen(["python", "/opt/lsp-manager.zip"])
        # self.lsp_manager_proc = subprocess.Popen(["lsp-manager"])
        self._load_client_ipc_server()

    def _load_client_ipc_server(self):
        self.client_ipc = ClientIPC()
        self.client_ipc.set_event_system(self._event_system)
        self._ipc_realization_check(self.client_ipc)

        if not self.client_ipc.is_ipc_alive:
            raise AppLaunchException(f"LSP IPC Server Already Exists...")

    def _ipc_realization_check(self, ipc_server):
        try:
            ipc_server.create_ipc_listener()
        except Exception:
            ipc_server.send_test_ipc_message()

        try:
            ipc_server.create_ipc_listener()
        except Exception as e:
            ...

    def stop_lsp_manager(self, button = None):
        if not self.lsp_manager_proc: return
        if not self.lsp_manager_proc.poll() is None:
            self.lsp_manager_proc = None
            return

        self.lsp_manager_proc.terminate()
        self.client_ipc.is_ipc_alive = False
        self.lsp_manager_proc = None

    def _lsp_did_open(self, language_id: str, uri: str, text: str):
        if not self.lsp_manager_proc: return

        data = {
            "method": "textDocument/didOpen",
            "language_id": language_id,
            "uri": uri,
            "version": -1,
            "text": text,
            "line": -1,
            "column": -1,
            "char": ""
        }

        self.send_message(data)

    def _lsp_did_save(self, uri: str, text: str):
        if not self.lsp_manager_proc: return

        data = {
            "method": "textDocument/didSave",
            "language_id": "",
            "uri": uri,
            "version": -1,
            "text": text,
            "line": -1,
            "column": -1,
            "char": ""
        }

        self.send_message(data)

    def _lsp_did_close(self, uri: str):
        if not self.lsp_manager_proc: return

        data = {
            "method": "textDocument/didClose",
            "language_id": "",
            "uri": uri,
            "version": -1,
            "text": "",
            "line": -1,
            "column": -1,
            "char": ""
        }

        self.send_message(data)

    def _lsp_did_change(self, language_id: str, uri: str, buffer):
        if not self.lsp_manager_proc: return

        iter   = buffer.get_iter_at_mark( buffer.get_insert() )
        line   = iter.get_line()
        column = iter.get_line_offset()
        start  = iter.copy()
        end    = iter.copy()

        start.backward_line()
        start.forward_line()
        end.forward_to_line_end()

        text   = buffer.get_text(start, end, include_hidden_chars = False)
        data   = {
            "method": "textDocument/didChange",
            "language_id": language_id,
            "uri": uri,
            "version": buffer.version_id,
            "text": text,
            "line": line,
            "column": column,
            "char": ""
        }

        self.send_message(data)

    def _lsp_goto(self, language_id: str, uri: str, line: int, column: int):
        if not self.lsp_manager_proc: return

        data = {
            "method": "textDocument/definition",
            "language_id": language_id,
            "uri": uri,
            "version": -1,
            "text": "",
            "line": line,
            "column": column,
            "char": ""
        }

        self.send_message(data)

    def _lsp_completion(self, source_view):
        if not self.lsp_manager_proc: return

        filepath  = source_view.get_current_file()
        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()

        char      = iter.get_char()
        if iter.backward_char():
            char = iter.get_char()

        column = iter.get_line_offset()
        data   = {
            "method": "textDocument/completion",
            "language_id": source_view.get_filetype(),
            "uri": uri,
            "version": buffer.version_id,
            "text": "",
            "line": line,
            "column": column,
            "char": char
        }

        self.send_message(data)

    def send_message(self, data: dict):
        self.client_ipc.send_manager_ipc_message( json.dumps(data) )