plugin.py

from __future__ import annotations

import os
import shutil
from typing import Any, TypedDict, cast

import sublime
from LSP.plugin import AbstractPlugin, LspTextCommand, Request, register_plugin, unregister_plugin
from LSP.plugin.core.views import (
    extract_variables,
    first_selection_region,
    offset_to_point,
    text_document_identifier,
    text_document_position_params,
)
from LSP.protocol import Position, TextDocumentIdentifier
from typing_extensions import NotRequired

from .const import ARCH, PLATFORM, PLATFORM_ARCH, PLATFORM_ARCH_TO_TARBALL, PLUGIN_NAME, SERVER_VERSION
from .server import get_default_server_bin_path, get_plugin_storage_dir, get_server_dir, get_server_download_url
from .tarball import decompress, download


class TextDocumentBuildParams(TypedDict):
    textDocument: TextDocumentIdentifier
    position: NotRequired[Position]


def plugin_loaded() -> None:
    register_plugin(LspTexLabPlugin)


def plugin_unloaded() -> None:
    unregister_plugin(LspTexLabPlugin)


class LspTexLabPlugin(AbstractPlugin):
    @classmethod
    def name(cls) -> str:
        return PLUGIN_NAME

    @classmethod
    def configuration(cls) -> tuple[sublime.Settings, str]:
        name = cls.name()
        basename = f"{name}.sublime-settings"
        filepath = f"Packages/{name}/{basename}"
        return sublime.load_settings(basename), filepath

    @classmethod
    def additional_variables(cls) -> dict[str, str]:
        return {
            "texlab_bin": get_default_server_bin_path()
            if PLATFORM_ARCH in PLATFORM_ARCH_TO_TARBALL
            else "texlab",
        }

    @classmethod
    def needs_update_or_installation(cls) -> bool:
        command = cast('list[str]', cls.configuration()[0].get("command"))
        server_bin = command[0]

        # only auto manage platforms which the official server supports
        if PLATFORM_ARCH in PLATFORM_ARCH_TO_TARBALL and server_bin in {
            "${texlab_bin}",
            "$texlab_bin",
        }:
            variables = extract_variables(sublime.active_window())
            variables.update(cls.additional_variables())
            server_bin = sublime.expand_variables(server_bin, variables)
            return not os.path.isfile(server_bin)

        # for unofficial supported platforms, users have to compile texlab by themselves
        # and adjust the "command" to use the executable
        return False

    @classmethod
    def install_or_update(cls) -> None:
        cls._cleanup_cache()

        is_download_ok = cls._prepare_server_bin()
        if not is_download_ok:
            raise RuntimeError("Unable to download the server binary...")

    @classmethod
    def _prepare_server_bin(cls) -> bool:
        """Download the LSP server binary."""

        server_dir = get_server_dir()
        download_url = get_server_download_url(SERVER_VERSION, PLATFORM, ARCH)
        if not download_url:
            return False

        tarball_name = download_url.split("/")[-1]
        tarball_path = os.path.join(server_dir, tarball_name)

        download(download_url, tarball_path)
        decompress(tarball_path, server_dir)

        return True

    @classmethod
    def _cleanup_cache(cls) -> None:
        """Clean up this plugin's cache directory."""

        shutil.rmtree(get_plugin_storage_dir(), ignore_errors=True)


class LspTexlabForwardSearchCommand(LspTextCommand):
    session_name = PLUGIN_NAME

    def run(self, _: sublime.Edit) -> None:
        session = self.session_by_name(PLUGIN_NAME)
        if not session:
            return
        params = text_document_position_params(self.view, next(iter(self.view.sel())).a)
        session.send_request(
            Request("textDocument/forwardSearch", params),
            self.on_response_async,
            self.on_error_async,
        )

    def on_response_async(self, response: Any) -> None:
        status = response["status"]
        window = self.view.window()
        if window is None:
            return
        if status == 0:
            pass  # success
        elif status == 1:
            window.status_message(PLUGIN_NAME + ": Previewer exited with errors")
        elif status == 2:
            window.status_message(
                PLUGIN_NAME + ": Previewer failed to start or crashed"
            )
        elif status == 3:
            window.status_message(PLUGIN_NAME + ": Previewer is not configured")

    def on_error_async(self, error: Any) -> None:
        pass


class LspTexlabBuildCommand(LspTextCommand):
    session_name = PLUGIN_NAME

    def run(self, _: sublime.Edit) -> None:
        session = self.session_by_name(PLUGIN_NAME)
        if not session:
            return
        params: TextDocumentBuildParams = {"textDocument": text_document_identifier(self.view)}
        region = first_selection_region(self.view)
        if region is not None:
            params["position"] = offset_to_point(self.view, region.b).to_lsp()

        session.send_request(
            Request("textDocument/build", params),
            self.on_response_async,
            self.on_error_async,
        )

    def on_response_async(self, response: Any) -> None:
        status = response["status"]
        window = self.view.window()
        if window is None:
            return
        if status == 0:
            pass  # success
        elif status == 1:
            window.status_message(PLUGIN_NAME + ": Build error")
        elif status == 2:
            window.status_message(PLUGIN_NAME + ": Build failure")
        elif status == 3:
            window.status_message(PLUGIN_NAME + ": Build cancelled")

    def on_error_async(self, error: Any) -> None:
        pass