implement command registration and execution

Author: d-biehl

robotcode/language_server/common/decorators.py

@@ -79,3 +79,23 @@ def filter(c: Any) -> bool:
         )
 
     return filter
+
+
+@runtime_checkable
+class IsCommand(Protocol):
+    __command_name__: List[str]
+
+
+def command(name: str) -> Callable[[_F], _F]:
+    def decorator(func: _F) -> _F:
+        setattr(func, "__command_name__", name)
+        return func
+
+    return decorator
+
+
+def get_command_name(func: _F) -> str:
+    if isinstance(func, IsCommand):
+        return func.__command_name__
+
+    raise TypeError(f"{func} is not a command.")

robotcode/language_server/common/lsp_types.py

@@ -13,6 +13,9 @@
 ProgressToken = Union[str, int]
 
 
+LSPAny = Any
+
+
 class ErrorCodes:
     PARSE_ERROR = -32700
     INVALID_REQUEST = -32600
@@ -981,6 +984,22 @@ class ExecuteCommandOptions(WorkDoneProgressOptions, _ExecuteCommandOptions):
     pass
 
 
+@dataclass(repr=False)
+class ExecuteCommandRegistrationOptions(ExecuteCommandOptions):
+    pass
+
+
+@dataclass(repr=False)
+class _ExecuteCommandParams(Model):
+    command: str
+    arguments: Optional[List[LSPAny]] = None
+
+
+@dataclass(repr=False)
+class ExecuteCommandParams(WorkDoneProgressParams, _ExecuteCommandParams):
+    pass
+
+
 @dataclass(repr=False)
 class SemanticTokensLegend(Model):
     token_types: List[str]
@@ -2368,3 +2387,16 @@ class CodeAction:
     edit: Optional[WorkspaceEdit] = None
     command: Optional[Command] = None
     data: Any = None
+
+
+@dataclass(repr=False)
+class ApplyWorkspaceEditParams(Model):
+    edit: WorkspaceEdit
+    label: Optional[str] = None
+
+
+@dataclass(repr=False)
+class ApplyWorkspaceEditResult(Model):
+    applied: bool
+    failure_reason: Optional[str] = None
+    failed_change: Optional[int] = None

robotcode/language_server/common/parts/commands.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Optional
+
+from ....jsonrpc2.protocol import JsonRPCErrorException, rpc_method
+from ....utils.async_tools import threaded
+from ....utils.logging import LoggingDescriptor
+from ..decorators import get_command_name
+from ..has_extend_capabilities import HasExtendCapabilities
+from ..lsp_types import (
+    ErrorCodes,
+    ExecuteCommandOptions,
+    ExecuteCommandParams,
+    LSPAny,
+    ServerCapabilities,
+)
+
+if TYPE_CHECKING:
+    from ..protocol import LanguageServerProtocol
+
+from .protocol_part import LanguageServerProtocolPart
+
+_FUNC_TYPE = Callable[..., Awaitable[Optional[LSPAny]]]
+
+
+class CommandsProtocolPart(LanguageServerProtocolPart, HasExtendCapabilities):
+
+    _logger = LoggingDescriptor()
+
+    def __init__(self, parent: LanguageServerProtocol) -> None:
+        super().__init__(parent)
+        self.commands: Dict[str, _FUNC_TYPE] = {}
+
+    def register(self, callback: _FUNC_TYPE, name: Optional[str] = None) -> None:
+        command = name or get_command_name(callback)
+
+        if command in self.commands:
+            self._logger.critical(f"command '{command}' already registered.")
+            return
+
+        self.commands[command] = callback
+
+    def extend_capabilities(self, capabilities: ServerCapabilities) -> None:
+
+        capabilities.execute_command_provider = ExecuteCommandOptions(list(self.commands.keys()))
+
+    @rpc_method(name="workspace/executeCommand", param_type=ExecuteCommandParams)
+    @threaded()
+    async def _text_document_code_lens(
+        self, command: str, arguments: Optional[List[LSPAny]], *args: Any, **kwargs: Any
+    ) -> Optional[LSPAny]:
+        self._logger.info(f"execute command {command}")
+
+        callback = self.commands.get(command, None)
+        if callback is None:
+            raise JsonRPCErrorException(ErrorCodes.INVALID_PARAMS, f"Command '{command}' unknown.")
+
+        return await callback(*(arguments or ()))

robotcode/language_server/common/parts/workspace.py

@@ -36,6 +36,8 @@
 from ....utils.uri import Uri
 from ...common.has_extend_capabilities import HasExtendCapabilities
 from ..lsp_types import (
+    ApplyWorkspaceEditParams,
+    ApplyWorkspaceEditResult,
     ConfigurationItem,
     ConfigurationParams,
     CreateFilesParams,
@@ -482,3 +484,12 @@ async def remove_file_watcher_entry(self, entry: FileWatcherEntry) -> None:
                 ):
                     await self.parent.unregister_capability(entry.id, "workspace/didChangeWatchedFiles")
                 # TODO: implement own filewatcher if not supported by language server client
+
+    async def apply_edit(self, edit: WorkspaceEdit, label: Optional[str] = None) -> ApplyWorkspaceEditResult:
+        r = await self.parent.send_request_async(
+            "workspace/applyEdit", ApplyWorkspaceEditParams(edit, label), return_type=ApplyWorkspaceEditResult
+        )
+
+        assert r is not None
+
+        return r

robotcode/language_server/common/protocol.py

@@ -40,6 +40,7 @@
 )
 from .parts.code_action import CodeActionProtocolPart
 from .parts.code_lens import CodeLensProtocolPart
+from .parts.commands import CommandsProtocolPart
 from .parts.completion import CompletionProtocolPart
 from .parts.declaration import DeclarationProtocolPart
 from .parts.definition import DefinitionProtocolPart
@@ -72,6 +73,7 @@ class LanguageServerProtocol(JsonRPCProtocol):
 
     _logger = LoggingDescriptor()
 
+    commands = ProtocolPartDescriptor(CommandsProtocolPart)
     window = ProtocolPartDescriptor(WindowProtocolPart)
     documents = ProtocolPartDescriptor(TextDocumentProtocolPart)
     diagnostics = ProtocolPartDescriptor(DiagnosticsProtocolPart)

robotcode/language_server/robotframework/parts/code_action.py

@@ -17,14 +17,28 @@
 from ....utils.logging import LoggingDescriptor
 from ....utils.net import find_free_port
 from ....utils.uri import Uri
-from ...common.decorators import code_action_kinds, language_id
+from ...common.decorators import (
+    code_action_kinds,
+    command,
+    get_command_name,
+    language_id,
+)
 from ...common.lsp_types import (
+    AnnotatedTextEdit,
+    ChangeAnnotation,
     CodeAction,
     CodeActionContext,
     CodeActionKinds,
     Command,
+    CreateFile,
+    DeleteFile,
+    MessageType,
     Model,
+    OptionalVersionedTextDocumentIdentifier,
     Range,
+    RenameFile,
+    TextDocumentEdit,
+    WorkspaceEdit,
 )
 from ...common.text_document import TextDocument
 from ..configuration import DocumentationServerConfig
@@ -186,6 +200,8 @@ def __init__(self, parent: RobotLanguageServerProtocol) -> None:
         self._documentation_server_lock = threading.RLock()
         self._documentation_server_port = 0
 
+        self.parent.commands.register(self.translate_suite)
+
     async def initialized(self, sender: Any) -> None:
         await self._ensure_http_server_started()
 
@@ -265,11 +281,11 @@ async def collect(
 
                             return [
                                 CodeAction(
-                                    f"Translate Suite to {language.name}",
+                                    f"Translate file to `{language.name}`",
                                     kind=CodeActionKinds.SOURCE + ".openDocumentation",
                                     command=Command(
                                         f"Translate Suite to {lang}",
-                                        "robotcode.translateSuite",
+                                        get_command_name(self.translate_suite),
                                         [document.document_uri, lang],
                                     ),
                                 )
@@ -394,3 +410,87 @@ async def _convert_uri(self, uri: str, *args: Any, **kwargs: Any) -> Optional[st
             return f"http://localhost:{self._documentation_server_port}/{path.as_posix()}"
 
         return None
+
+    @command("robotcode.translateSuite")
+    async def translate_suite(self, document_uri: str, lang: str) -> None:
+        from robot.conf.languages import Language
+        from robot.parsing.lexer.tokens import Token as RobotToken
+
+        try:
+            language = Language.from_name(lang)
+        except ValueError:
+            self.parent.window.show_message(f"Invalid language {lang}", MessageType.ERROR)
+            return
+
+        document = await self.parent.documents.get(document_uri)
+
+        if document is None:
+            return
+
+        changes: List[Union[TextDocumentEdit, CreateFile, RenameFile, DeleteFile]] = []
+
+        header_translations = {
+            RobotToken.SETTING_HEADER: language.settings_header,
+            RobotToken.VARIABLE_HEADER: language.variables_header,
+            RobotToken.TESTCASE_HEADER: language.test_cases_header,
+            RobotToken.TASK_HEADER: language.tasks_header,
+            RobotToken.KEYWORD_HEADER: language.keywords_header,
+            RobotToken.COMMENT_HEADER: language.comments_header,
+        }
+        settings_translations = {
+            RobotToken.LIBRARY: language.library_setting,
+            RobotToken.DOCUMENTATION: language.documentation_setting,
+            RobotToken.SUITE_SETUP: language.suite_setup_setting,
+            RobotToken.SUITE_TEARDOWN: language.suite_teardown_setting,
+            RobotToken.METADATA: language.metadata_setting,
+            RobotToken.KEYWORD_TAGS: language.keyword_tags_setting,
+            RobotToken.LIBRARY: language.library_setting,
+            RobotToken.RESOURCE: language.resource_setting,
+            RobotToken.VARIABLES: language.variables_setting,
+            RobotToken.SETUP: f"[{language.setup_setting}]",
+            RobotToken.TEARDOWN: f"[{language.teardown_setting}]",
+            RobotToken.TEMPLATE: f"[{language.template_setting}]",
+            RobotToken.TIMEOUT: f"[{language.timeout_setting}]",
+            RobotToken.TAGS: f"[{language.tags_setting}]",
+            RobotToken.ARGUMENTS: f"[{language.arguments_setting}]",
+        }
+
+        for token in await self.parent.documents_cache.get_tokens(document):
+            if token.type in header_translations.keys():
+                changes.append(
+                    TextDocumentEdit(
+                        OptionalVersionedTextDocumentIdentifier(str(document.uri), document.version),
+                        [
+                            AnnotatedTextEdit(
+                                range_from_token(token),
+                                f"*** { header_translations[token.type]} ***",
+                                annotation_id="translate_settings",
+                            )
+                        ],
+                    )
+                )
+            elif token.type in settings_translations.keys():
+                changes.append(
+                    TextDocumentEdit(
+                        OptionalVersionedTextDocumentIdentifier(str(document.uri), document.version),
+                        [
+                            AnnotatedTextEdit(
+                                range_from_token(token),
+                                settings_translations[token.type],
+                                annotation_id="translate_settings",
+                            )
+                        ],
+                    )
+                )
+            else:
+                pass
+
+        if not changes:
+            return
+
+        edit = WorkspaceEdit(
+            document_changes=changes,
+            change_annotations={"translate_settings": ChangeAnnotation("Translate Settings", False)},
+        )
+
+        await self.parent.workspace.apply_edit(edit, "Translate")