ishangsf/workflows
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adjust lib/ide_service

Browse Source
This commit is contained in:
kagami 2024-05-09 21:30:09 +08:00
parent b87414e38d
commit dabd12f5cf
9 changed files with 311 additions and 162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
lib/ide_service/__init__.py Normal file
View File

@ -0,0 +1,7 @@
from .service import IDEService
from .types import * # noqa: F403
from .types import __all__ as types_all
__all__ = types_all + [
"IDEService",
]

15
lib/ide_service/idea_service.py Normal file
View File

@ -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)

76
lib/ide_service/rpc.py Normal file
View File

@ -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

154
lib/ide_service/service.py Normal file
View File

@ -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()

13
lib/ide_services/types.py → lib/ide_service/types.py
View File

@ -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__())

56
lib/ide_services/vscode_services.py → lib/ide_service/vscode_service.py
View File

@ -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],
},
},
)

6
lib/ide_services/__init__.py
View File

@ -1,6 +0,0 @@
from .service import IDEService
from .types import *
__all__ = types.__all__ + [
"IDEService",
]

37
lib/ide_services/rpc.py
View File

@ -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

109
lib/ide_services/service.py
View File

@ -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 []