Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions devchat/ide/idea_services.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
from .types import LocationWithText
from .rpc import rpc_method

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)
40 changes: 40 additions & 0 deletions devchat/ide/rpc.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
# pylint: disable=W3101
# pylint: disable=W0719
# pylint: disable=R1710
# pylint: disable=W0212
import os
from functools import wraps

Expand Down Expand Up @@ -39,3 +40,42 @@ def wrapper(*args, **kwargs):
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
140 changes: 94 additions & 46 deletions devchat/ide/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,59 +2,16 @@
# pylint: disable=W0613
# pylint: disable=E1133
# pylint: disable=R1710
# pylint: disable=W0212
# pylint: disable=W0719
# pylint: disable=W3101
# pylint: disable=C0103

import os
from functools import wraps
from typing import List

import requests

from .rpc import rpc_method
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
from .vscode_services import selected_range, visible_range
from .idea_services import IdeaIDEService


class IDEService:
Expand All @@ -72,18 +29,35 @@ def __init__(self):

@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 a string result.
"""
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
Expand All @@ -95,8 +69,82 @@ def ide_logging(self, level: str, message: str) -> bool:

@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.
"""
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]:
"""
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.
"""
return [Location.parse_obj(loc) for loc in self._result]

@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.
"""
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):
"""
Determines and returns the visible range of code in the current IDE.

Returns:
A tuple denoting the visible range if the IDE is VSCode, or defers to
IdeaIDEService's get_visible_range method for other IDEs.
"""
if self.ide_name() == "vscode":
return visible_range()
return IdeaIDEService().get_visible_range()

def get_selected_range(self):
"""
Retrieves the selected range of code in the current IDE.

Returns:
Calls and returns the result of `selected_range()` if the IDE is VSCode,
otherwise, it defers to IdeaIDEService's `get_selected_range()` method.
"""
if self.ide_name() == "vscode":
return selected_range()
return IdeaIDEService().get_selected_range()
6 changes: 6 additions & 0 deletions devchat/ide/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,3 +23,9 @@ class SymbolNode(BaseModel):
kind: str
range: Range
children: List["SymbolNode"]


class LocationWithText(BaseModel):
abspath: str
range: Range
text: str
35 changes: 35 additions & 0 deletions devchat/ide/vscode_services.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import os

from .rpc import rpc_call
from .types import LocationWithText


@rpc_call
Expand Down Expand Up @@ -109,6 +110,23 @@ def 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()
Expand Down Expand Up @@ -140,3 +158,20 @@ def selected_lines():
"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],
},
}
)