Improve IDE services with type annotation and casting & Add two interfaces

ask-code/ask-code.py

@@ -6,7 +6,7 @@
 sys.path.append(os.path.join(os.path.dirname(__file__), "..", "libs"))
 sys.path.append(os.path.join(os.path.dirname(__file__), "..", "..", "libs"))
 
-from ide_services import get_lsp_brige_port  # noqa: E402
+from ide_services import IDEService  # noqa: E402
 
 
 def query(question, lsp_brige_port):
@@ -59,7 +59,7 @@ def main():
             print("Usage: python index_and_query.py query [question] [port]")
             sys.exit(1)
 
-        port = get_lsp_brige_port()
+        port = IDEService().get_lsp_brige_port()
 
         question = sys.argv[2]
         query(question, port)

commit/commit.py

@@ -9,7 +9,7 @@
 sys.path.append(os.path.dirname(__file__))
 
 from chatmark import Checkbox, Form, TextEditor  # noqa: E402
-from ide_services.services import log_info
+from ide_services import IDEService  # noqa: E402
 from llm_api import chat_completion_stream  # noqa: E402
 
 diff_too_large_message_en = (
@@ -67,17 +67,19 @@ def read_prompt_from_file(filename):
     - FileNotFoundError: If the file does not exist.
     - Exception: If any other error occurs during file reading.
     """
+    s = IDEService()
     try:
         with open(filename, "r", encoding="utf-8") as file:
             return file.read().strip()
     except FileNotFoundError:
-        log_info(
+        s.ide_logging(
+            "info",
             f"File {filename} not found. "
-            "Please make sure it exists in the same directory as the script."
+            "Please make sure it exists in the same directory as the script.",
         )
         sys.exit(1)
     except Exception as e:
-        log_info(f"An error occurred while reading the file {filename}: {e}")
+        s.ide_logging("info", f"An error occurred while reading the file {filename}: {e}")
         sys.exit(1)
 
 

libs/ide_services/__init__.py

@@ -1,15 +1,5 @@
-from .services import (
-    get_lsp_brige_port,
-    ide_language,
-    ide_logging,
-    install_python_env,
-    update_slash_commands,
-)
+from .service import IDEService
 
 __all__ = [
-    "get_lsp_brige_port",
-    "install_python_env",
-    "update_slash_commands",
-    "ide_language",
-    "ide_logging",
+    "IDEService",
 ]

libs/ide_services/service.py

@@ -0,0 +1,93 @@
+import os
+from functools import wraps
+from typing import List
+
+import requests
+
+from .types import Location, SymbolNode
+
+BASE_SERVER_URL = os.environ.get("DEVCHAT_IDE_SERVICE_URL", "http://localhost:3000")
+
+
+def rpc_method(f):
+    """
+    Decorator for Service methods
+    """
+
+    @wraps(f)
+    def wrapper(self, *args, **kwargs):
+        if os.environ.get("DEVCHAT_IDE_SERVICE_URL", "") == "":
+            # maybe in a test, user don't want to mock services functions
+            return
+
+        try:
+            function_name = f.__name__
+            url = f"{BASE_SERVER_URL}/{function_name}"
+
+            data = dict(zip(f.__code__.co_varnames[1:], args))  # Exclude "self"
+            data.update(kwargs)
+            headers = {"Content-Type": "application/json"}
+
+            response = requests.post(url, json=data, headers=headers)
+
+            if response.status_code != 200:
+                raise Exception(f"Server error: {response.status_code}")
+
+            response_data = response.json()
+            if "error" in response_data:
+                raise Exception(f"Server returned an error: {response_data['error']}")
+
+            # Store the result in the _result attribute of the instance
+            self._result = response_data.get("result", None)
+            return f(self, *args, **kwargs)
+
+        except ConnectionError as err:
+            # TODO
+            raise err
+
+    return wrapper
+
+
+class IDEService:
+    """
+    Client for IDE service
+
+    Usage:
+    client = IDEService()
+    res = client.ide_language()
+    res = client.ide_logging("info", "some message")
+    """
+
+    def __init__(self):
+        self._result = None
+
+    @rpc_method
+    def get_lsp_brige_port(self) -> str:
+        return self._result
+
+    @rpc_method
+    def install_python_env(self, command_name: str, requirements_file: str) -> str:
+        return self._result
+
+    @rpc_method
+    def update_slash_commands(self) -> bool:
+        return self._result
+
+    @rpc_method
+    def ide_language(self) -> str:
+        return self._result
+
+    @rpc_method
+    def ide_logging(self, level: str, message: str) -> bool:
+        """
+        level: "info" | "warn" | "error" | "debug"
+        """
+        return self._result
+
+    @rpc_method
+    def get_document_symbols(self, abspath: str) -> List[SymbolNode]:
+        return [SymbolNode.parse_obj(node) for node in self._result]
+
+    @rpc_method
+    def find_type_def_locations(self, abspath: str, line: int, character: int) -> List[Location]:
+        return [Location.parse_obj(loc) for loc in self._result]

libs/ide_services/types.py

@@ -0,0 +1,25 @@
+from typing import List
+
+from pydantic import BaseModel
+
+
+class Position(BaseModel):
+    line: int  # 0-based
+    character: int  # 0-based
+
+
+class Range(BaseModel):
+    start: Position
+    end: Position
+
+
+class Location(BaseModel):
+    abspath: str
+    range: Range
+
+
+class SymbolNode(BaseModel):
+    name: str
+    kind: str
+    range: Range
+    children: List["SymbolNode"]

libs/llm_api/text_confirm.py

@@ -7,7 +7,6 @@
 
 
 from chatmark import Checkbox, Form, TextEditor  # noqa: #402
-from ide_services.services import log_info  # noqa: #402
 
 
 class MissEditConfirmFieldException(Exception):

libs/llm_api/tools_call.py

@@ -9,7 +9,7 @@
 sys.path.append(os.path.join(os.path.dirname(__file__), ".."))
 
 from chatmark import Checkbox, Form, Radio, TextEditor  # noqa: #402
-from ide_services.services import log_info, log_warn  # noqa: #402
+from ide_services import IDEService  # noqa: #402
 
 
 class MissToolsFieldException(Exception):
@@ -191,9 +191,10 @@ def wrapper(*args, **kwargs):
                     # call function
                     functions = {tool.function_name: tool for tool in tools}
                     for call in response["all_calls"]:
-                        log_info(
+                        IDEService().ide_logging(
+                            "info",
                             f"try to call function tool: {call['function_name']} "
-                            f"with {call['parameters']}"
+                            f"with {call['parameters']}",
                         )
                         tool = functions[call["function_name"]]
                         result = tool(**json.loads(call["parameters"]))

unit_tests/main.py

@@ -10,7 +10,7 @@
 from chatmark import Checkbox, Form, Step, TextEditor  # noqa: E402
 from find_reference_tests import find_reference_tests
 from i18n import TUILanguage, get_translation
-from ide_services import ide_language  # noqa: E402
+from ide_services import IDEService  # noqa: E402
 from model import (
     FuncToTest,
     TokenBudgetExceededException,
@@ -214,7 +214,8 @@ def main(input: str):
     user_prompt = f"Help me write unit tests for the `{func_name}` function"
 
     repo_root = os.getcwd()
-    ide_lang = ide_language()
+    ide_lang = IDEService().ide_language()
+
     tui_lang = TUILanguage.from_str(ide_lang)
     _i = get_translation(tui_lang)