diff --git a/lib/ide_service/__init__.py b/lib/ide_service/__init__.py new file mode 100644 index 0000000..95791c8 --- /dev/null +++ b/lib/ide_service/__init__.py @@ -0,0 +1,7 @@ +from .service import IDEService +from .types import * # noqa: F403 +from .types import __all__ as types_all + +__all__ = types_all + [ + "IDEService", +] diff --git a/lib/ide_service/idea_service.py b/lib/ide_service/idea_service.py new file mode 100644 index 0000000..db5753a --- /dev/null +++ b/lib/ide_service/idea_service.py @@ -0,0 +1,15 @@ +from .rpc import rpc_method +from .types import LocationWithText + + +class IdeaIDEService: + def __init__(self): + self._result = None + + @rpc_method + def get_visible_range(self) -> LocationWithText: + return LocationWithText.parse_obj(self._result) + + @rpc_method + def get_selected_range(self) -> LocationWithText: + return LocationWithText.parse_obj(self._result) diff --git a/lib/ide_service/rpc.py b/lib/ide_service/rpc.py new file mode 100644 index 0000000..67e2923 --- /dev/null +++ b/lib/ide_service/rpc.py @@ -0,0 +1,76 @@ +import os +from functools import wraps + +import requests + +BASE_SERVER_URL = os.environ.get("DEVCHAT_IDE_SERVICE_URL", "http://localhost:3000") + + +def rpc_call(f): + @wraps(f) + def wrapper(*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, args)) + 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']}") + return response_data.get("result", None) + except ConnectionError as err: + # TODO + raise err + + return wrapper + + +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 diff --git a/lib/ide_service/service.py b/lib/ide_service/service.py new file mode 100644 index 0000000..eb72052 --- /dev/null +++ b/lib/ide_service/service.py @@ -0,0 +1,154 @@ +from typing import List + +from .idea_service import IdeaIDEService +from .rpc import rpc_method +from .types import Location, LocationWithText, SymbolNode +from .vscode_service import selected_range, visible_range + + +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: + """ + Get the LSP bridge port. + + :return: str + """ + return self._result + + @rpc_method + def install_python_env(self, command_name: str, requirements_file: str) -> str: + """ + A method to install a Python environment with the provided command name + and requirements file, returning python path installed. + Command name is the name of the environment to be installed. + """ + return self._result + + @rpc_method + def update_slash_commands(self) -> bool: + """ + Update the slash commands and return a boolean indicating the success of the operation. + """ + return self._result + + @rpc_method + def ide_language(self) -> str: + """ + Returns the current IDE language setting for the user. + - zh: Chinese + - en: English + """ + return self._result + + @rpc_method + def ide_logging(self, level: str, message: str) -> bool: + """ + Logs a message to the IDE. + level: "info" | "warn" | "error" | "debug" + """ + return self._result + + @rpc_method + def get_document_symbols(self, abspath: str) -> List[SymbolNode]: + """ + Retrieves the document symbols for a given file. + + Args: + abspath: The absolute path to the file whose symbols are to be retrieved. + + Returns: + A list of SymbolNode objects representing the symbols found in the document. + """ + try: + return [SymbolNode.parse_obj(node) for node in self._result] + except Exception: + # TODO: logging ide service error + return [] + + @rpc_method + def find_type_def_locations(self, abspath: str, line: int, character: int) -> List[Location]: + """ + Finds the location of type definitions within a file. + + Args: + abspath: The absolute path to the file to be searched. + line: The line number within the file to begin the search. + character: The character position within the line to begin the search. + + Returns: + A list of Location objects representing the locations of type definitions found. + """ + try: + return [Location.parse_obj(loc) for loc in self._result] + except Exception: + # TODO: logging ide service error + return [] + + @rpc_method + def find_def_locations(self, abspath: str, line: int, character: int) -> List[Location]: + try: + return [Location.parse_obj(loc) for loc in self._result] + except Exception: + # TODO: logging ide service error + return [] + + @rpc_method + def ide_name(self) -> str: + """Returns the name of the IDE. + + This method is a remote procedure call (RPC) that fetches the name of the IDE being used. + + Returns: + The name of the IDE as a string. For example, "vscode" or "pycharm". + """ + return self._result + + @rpc_method + def diff_apply(self, filepath, content) -> bool: + """ + Applies a given diff to a file. + + This method uses the content provided to apply changes to the file + specified by the filepath. It's an RPC method that achieves file synchronization + by updating the local version of the file with the changes described in the + content parameter. + + Args: + filepath: The path to the file that needs to be updated. + content: A string containing the new code that should be applied to the file. + + Returns: + A boolean indicating if the diff was successfully applied. + """ + return self._result + + def get_visible_range(self) -> LocationWithText: + """ + Determines and returns the visible range of code in the current IDE. + """ + # TODO: Implement vscode endpoint following the protocol in stead of using python wrapper + if self.ide_name() == "vscode": + return visible_range() + return IdeaIDEService().get_visible_range() + + def get_selected_range(self) -> LocationWithText: + """ + Retrieves the selected range of code in the current IDE. + """ + # TODO: Implement vscode endpoint following the protocol in stead of using python wrapper + if self.ide_name() == "vscode": + return selected_range() + return IdeaIDEService().get_selected_range() diff --git a/lib/ide_services/types.py b/lib/ide_service/types.py similarity index 77% rename from lib/ide_services/types.py rename to lib/ide_service/types.py index b1d8089..0e47b73 100644 --- a/lib/ide_services/types.py +++ b/lib/ide_service/types.py @@ -7,6 +7,7 @@ __all__ = [ "Range", "Location", "SymbolNode", + "LocationWithText", ] @@ -48,3 +49,15 @@ class SymbolNode(BaseModel): kind: str range: Range children: List["SymbolNode"] + + +class LocationWithText(BaseModel): + abspath: str + range: Range + text: str + + def __repr__(self): + return f"{self.abspath}::{self.range}::{self.text}" + + def __hash__(self): + return hash(self.__repr__()) diff --git a/lib/ide_services/vscode_services.py b/lib/ide_service/vscode_service.py similarity index 69% rename from lib/ide_services/vscode_services.py rename to lib/ide_service/vscode_service.py index c57dfac..c8f93a8 100644 --- a/lib/ide_services/vscode_services.py +++ b/lib/ide_service/vscode_service.py @@ -1,7 +1,7 @@ import os -import sys from .rpc import rpc_call +from .types import LocationWithText @rpc_call @@ -99,18 +99,36 @@ def visible_lines(): end_line = active_document["visibleRanges"][0][1]["line"] # read file lines from start_line to end_line - with open(file_path, "r") as file: - lines = file.readlines() - selected_lines = lines[start_line : end_line + 1] + with open(file_path, "r", encoding="utf-8") as file: + _lines = file.readlines() + _visible_lines = _lines[start_line : end_line + 1] # continue with the rest of the function return { "filePath": file_path, - "visibleText": "".join(selected_lines), + "visibleText": "".join(_visible_lines), "visibleRange": [start_line, end_line], } +def visible_range() -> LocationWithText: + visible_range_text = visible_lines() + return LocationWithText( + text=visible_range_text["visibleText"], + abspath=visible_range_text["filePath"], + range={ + "start": { + "line": visible_range_text["visibleRange"][0], + "character": 0, + }, + "end": { + "line": visible_range_text["visibleRange"][1], + "character": 0, + }, + }, + ) + + def selected_lines(): active_document = active_text_editor() fail_result = { @@ -131,13 +149,31 @@ def selected_lines(): end_col = active_document["selection"]["end"]["character"] # read file lines from start_line to end_line - with open(file_path, "r") as file: - lines = file.readlines() - selected_lines = lines[start_line : end_line + 1] + with open(file_path, "r", encoding="utf-8") as file: + _lines = file.readlines() + _selected_lines = _lines[start_line : end_line + 1] # continue with the rest of the function return { - "filePath": "", - "selectedText": "".join(selected_lines), + "filePath": file_path, + "selectedText": "".join(_selected_lines), "selectedRange": [start_line, start_col, end_line, end_col], } + + +def selected_range() -> LocationWithText: + selected_range_text = selected_lines() + return LocationWithText( + text=selected_range_text["selectedText"], + abspath=selected_range_text["filePath"], + range={ + "start": { + "line": selected_range_text["selectedRange"][0], + "character": selected_range_text["selectedRange"][1], + }, + "end": { + "line": selected_range_text["selectedRange"][2], + "character": selected_range_text["selectedRange"][3], + }, + }, + ) diff --git a/lib/ide_services/__init__.py b/lib/ide_services/__init__.py deleted file mode 100644 index f86ae78..0000000 --- a/lib/ide_services/__init__.py +++ /dev/null @@ -1,6 +0,0 @@ -from .service import IDEService -from .types import * - -__all__ = types.__all__ + [ - "IDEService", -] diff --git a/lib/ide_services/rpc.py b/lib/ide_services/rpc.py deleted file mode 100644 index eced9c3..0000000 --- a/lib/ide_services/rpc.py +++ /dev/null @@ -1,37 +0,0 @@ -import os -from functools import wraps - -import requests - -BASE_SERVER_URL = os.environ.get("DEVCHAT_IDE_SERVICE_URL", "http://localhost:3000") - - -def rpc_call(f): - @wraps(f) - def wrapper(*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, args)) - 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']}") - return response_data.get("result", None) - except ConnectionError as err: - # TODO - raise err - - return wrapper diff --git a/lib/ide_services/service.py b/lib/ide_services/service.py deleted file mode 100644 index ed0c915..0000000 --- a/lib/ide_services/service.py +++ /dev/null @@ -1,109 +0,0 @@ -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]: - try: - return [SymbolNode.parse_obj(node) for node in self._result] - except: - # TODO: loggging ide service error - return [] - - @rpc_method - def find_type_def_locations(self, abspath: str, line: int, character: int) -> List[Location]: - try: - return [Location.parse_obj(loc) for loc in self._result] - except: - # TODO: loggging ide service error - return [] - - @rpc_method - def find_def_locations(self, abspath: str, line: int, character: int) -> List[Location]: - try: - return [Location.parse_obj(loc) for loc in self._result] - except: - # TODO: loggging ide service error - return []