1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
|
import os
import subprocess
import uuid
from typing import Any, Callable, Coroutine, List, Optional
from dotenv import load_dotenv
from fastapi import WebSocket
from ..models.filesystem import (
FileSystem,
RangeInFile,
RangeInFileWithContents,
RealFileSystem,
)
from ..models.filesystem_edit import EditDiff, FileEdit, FileSystemEdit
from ..server.ide_protocol import AbstractIdeProtocolServer
load_dotenv()
def get_mac_address():
mac_num = hex(uuid.getnode()).replace("0x", "").upper()
mac = "-".join(mac_num[i : i + 2] for i in range(0, 11, 2))
return mac
class LocalIdeProtocol(AbstractIdeProtocolServer):
websocket: WebSocket = None
session_id: Optional[str]
workspace_directory: str = os.getcwd()
unique_id: str = get_mac_address()
filesystem: FileSystem = RealFileSystem()
async def handle_json(self, data: Any):
"""Handle a json message"""
pass
def showSuggestion(self, file_edit: FileEdit):
"""Show a suggestion to the user"""
pass
async def setFileOpen(self, filepath: str, open: bool = True):
"""Set whether a file is open"""
pass
async def showMessage(self, message: str):
"""Show a message to the user"""
print(message)
async def showVirtualFile(self, name: str, contents: str):
"""Show a virtual file"""
pass
async def setSuggestionsLocked(self, filepath: str, locked: bool = True):
"""Set whether suggestions are locked"""
pass
async def getSessionId(self):
"""Get a new session ID"""
pass
async def showSuggestionsAndWait(self, suggestions: List[FileEdit]) -> bool:
"""Show suggestions to the user and wait for a response"""
pass
def onAcceptRejectSuggestion(self, accepted: bool):
"""Called when the user accepts or rejects a suggestion"""
pass
def onFileSystemUpdate(self, update: FileSystemEdit):
"""Called when a file system update is received"""
pass
def onCloseGUI(self, session_id: str):
"""Called when a GUI is closed"""
pass
def onOpenGUIRequest(self):
"""Called when a GUI is requested to be opened"""
pass
async def getOpenFiles(self) -> List[str]:
"""Get a list of open files"""
pass
async def getVisibleFiles(self) -> List[str]:
"""Get a list of visible files"""
pass
async def getHighlightedCode(self) -> List[RangeInFile]:
"""Get a list of highlighted code"""
pass
async def readFile(self, filepath: str) -> str:
"""Read a file"""
return self.filesystem.read(filepath)
async def readRangeInFile(self, range_in_file: RangeInFile) -> str:
"""Read a range in a file"""
return self.filesystem.read_range_in_file(range_in_file)
async def editFile(self, edit: FileEdit):
"""Edit a file"""
self.filesystem.apply_file_edit(edit)
async def applyFileSystemEdit(self, edit: FileSystemEdit) -> EditDiff:
"""Apply a file edit"""
return self.filesystem.apply_edit(edit)
async def saveFile(self, filepath: str):
"""Save a file"""
pass
async def getUserSecret(self, key: str):
"""Get a user secret"""
return os.environ.get(key)
async def highlightCode(self, range_in_file: RangeInFile, color: str):
"""Highlight code"""
pass
async def runCommand(self, command: str) -> str:
"""Run a command using subprocess (don't pass, actually implement)"""
return subprocess.check_output(command, shell=True).decode("utf-8")
def onHighlightedCodeUpdate(self, range_in_files: List[RangeInFileWithContents]):
"""Called when highlighted code is updated"""
pass
def onDeleteAtIndex(self, index: int):
"""Called when a step is deleted at a given index"""
pass
async def showDiff(self, filepath: str, replacement: str, step_index: int):
"""Show a diff"""
pass
def subscribeToFilesCreated(self, callback: Callable[[List[str]], None]):
"""Subscribe to files created event"""
pass
def subscribeToFilesDeleted(self, callback: Callable[[List[str]], None]):
"""Subscribe to files deleted event"""
pass
def subscribeToFilesRenamed(self, callback: Callable[[List[str], List[str]], None]):
"""Subscribe to files renamed event"""
pass
def subscribeToFileSaved(self, callback: Callable[[str, str], None]):
"""Subscribe to file saved event"""
pass
def onFilesCreated(self, filepaths: List[str]):
"""Called when files are created"""
pass
def onFilesDeleted(self, filepaths: List[str]):
"""Called when files are deleted"""
pass
def onFilesRenamed(self, old_filepaths: List[str], new_filepaths: List[str]):
"""Called when files are renamed"""
pass
def onFileSaved(self, filepath: str, contents: str):
"""Called when a file is saved"""
pass
async def fileExists(self, filepath: str) -> Coroutine[Any, Any, str]:
"""Check if a file exists"""
return self.filesystem.exists(filepath)
async def getTerminalContents(self) -> Coroutine[Any, Any, str]:
return ""
async def listDirectoryContents(
self, directory: str, recursive: bool = False
) -> List[str]:
return self.filesystem.list_directory_contents(directory, recursive=recursive)
|
