Repository navigation
Expand file tree
/
Copy pathpython_completion_server.py
More file actions
223 lines (211 loc) · 7.31 KB
/
Copy pathpython_completion_server.py
File metadata and controls
223 lines (211 loc) · 7.31 KB
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
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
import code
import sys
import os.path
import time
import types
import stdlib_parser
import struct
import xmlrpclib
from SimpleXMLRPCServer import SimpleXMLRPCServer
#from pysmell import idehelper
#from pysmell import tags
from re import split
type_map = {
str(types.ModuleType) : 1,
str(types.ClassType) : 2,
str(types.ObjectType) : 3,
str(types.TypeType) : 4,
str(types.InstanceType) : 5,
str(types.FunctionType) : 6,
str(types.MethodType) : 7,
str(types.UnboundMethodType) : 8,
str(types.BuiltinFunctionType) : 6,
str(types.BuiltinMethodType) : 7,
str(types.LambdaType) : 9,
str(types.GeneratorType) : 10 }
def type_suffix(symbols,s):
type_str=symbols[s][0]
if type_str in type_map:
return '?'+str(type_map[type_str])
return ''
def _uniquify(l):
found = set()
for item in l:
if item not in found:
yield item
found.add(item)
isz=struct.calcsize('@I')
class XmlRpcPipeServer:
def __init__(self):
self.fn_dict={}
self.inpipe=sys.stdin
self.outpipe=sys.stdout
sys.stdout=open(os.devnull,'wb')
def register_function(self,fn,name):
self.fn_dict[name]=fn
def register_introspection_functions(self):
pass
def handle_request(self):
##TODO: Need more error handling!
size_buf = self.inpipe.read(isz)
size = struct.unpack('@I',size_buf)[0]
call_xml = self.inpipe.read(size)
self.outpipe.write(struct.pack('c','M'))
name=''
try:
args,name = xmlrpclib.loads(call_xml)
result = self.__call__(name, *args)
if not isinstance(result,tuple):
result=(result,)
res_xml = bytes(xmlrpclib.dumps(result, methodresponse=True))
except:
import traceback
res_xml='Error running call'+name+call_xml+'\n'
res_xml+='\n'.join(traceback.format_exception(*sys.exc_info()))
size = len(res_xml)
self.outpipe.write(struct.pack('@I',size))
self.outpipe.write(res_xml)
def __call__(self,name,*args):
return self.fn_dict[name](*args)
class AsyncServer:
def __init__(self,port):
# Create XMLRPC server
self.timeout=0.2
self._quit=False
self.port=port
self.std_lib=None
def run(self):
if self.port==-1:
self.server = XmlRpcPipeServer()
else:
self.server = SimpleXMLRPCServer(("localhost", self.port))
self.server.logRequests=0
self.server.register_introspection_functions()
#self.server.socket.settimeout(self.timeout) ##timeouts cause probs
self.server.register_function(self.end,'end')
self.server.register_function(self.get_completions,'get_completions')
self.server.register_function(self.tag_gen,'tag_gen')
self.server.register_function(self.load_stdlib,'load_stdlib')
self.server.register_function(self.create_stdlib,'create_stdlib')
self.server.register_function(self.complete_phrase,'complete_phrase')
self.server.register_function(self.complete_tip,'complete_tip')
while not self._quit:
self.server.handle_request()
def load_stdlib(self,path):
# print 'loading lib',path
self.stdlib=stdlib_parser.load(path)
return self.stdlib!=None
def create_stdlib(self,path):
# print 'creating',path
self.stdlib=stdlib_parser.create(path)
return self.stdlib!=None
def complete_context(self,path,source,position):
# print 'complete context',path,position
pass
def complete_tip(self,symbol):
# print 'complete tip',symbol
context=[s.strip() for s in symbol.split('.')]
symbol=context.pop()
psymbols=self.stdlib
for s in context:
if s not in psymbols:
print 'symbol not found'
return ''
psymbols=psymbols[s][-1]
if symbol not in psymbols:
print 'symbol not found'
return ''
symbol_data=psymbols[symbol]
result=''
if symbol_data[1] is not None:
result=symbol+'('+str(symbol_data[1])+')'
else:
if symbol_data[2] is None or not symbol_data[2].startswith(symbol+'('):
result=symbol+'(...)'
doc=str(symbol_data[2])
if doc is not None:
lines=doc.split('\n')
shortened=False
if len(lines)>6:
shortened=True
doc='\n'.join(lines[:6])
if len(doc)>250:
shortened=True
doc=doc[:250]
doc=doc.strip(' \t\n')
if shortened:
doc+='...'
result+='\n'+doc
print 'result',result
return result
def complete_phrase(self,phrase):
# print 'complete phrase',phrase
context=[s.strip() for s in phrase.split('.')]
phrase=context.pop()
psymbols=self.stdlib
for s in context:
if s not in psymbols:
print 'x',[]
return []
psymbols=psymbols[s][-1]
if type(psymbols)!=types.DictType:
return []
completions=[]
try:
for s in psymbols:
if s.startswith(phrase):
completions.append(s)
completions=sorted(completions)
print 'COMPS',completions
completions=[s+type_suffix(psymbols,s) for s in completions]
return completions
except:
import traceback,sys
traceback.print_exception(*sys.exc_info())
def tool_tip(self,function):
return False
def tag_gen(self, projectDir):
args = ['pysmell', projectDir, '-o', os.path.join(projectDir, 'PYSMELLTAGS')]
sys.argv = args
tags.main()
return True
def get_completions(self, cur_file, source, line_no, cur_col):
"""arguments: fullPath, origSource, lineNo, origCol, matcher
When visiting the file at fullPath, with edited source origSource, find a list
of possible completion strings for the symbol located at origCol on orgLineNo using
matching mode matcher"""
print cur_file,line_no,cur_col
if not cur_file:
write('No filename - is the file saved?')
return TOOLTIP
source = file(cur_file,'r').read()
PYSMELLDICT = idehelper.findPYSMELLDICT(cur_file)
if PYSMELLDICT is None:
write('No PYSMELLTAGS found - you have to generate one.')
return TOOLTIP
print 'found dict'
line = source.splitlines()[line_no - 1]
index = idehelper.findBase(line, cur_col)
base = line[index:cur_col]
options = idehelper.detectCompletionType(cur_file, source, line_no, cur_col, base, PYSMELLDICT)
completions = idehelper.findCompletions(base, PYSMELLDICT, options)
return completions
def end(self):
self._quit=True
return True
def cmd_err():
print 'Correct usage: python_completion_server.py <port>'
print '<port> must be a positive integer or -1 to use stdin/stdout'
exit()
if __name__=='__main__':
if len(sys.argv)!=2:
port=-1
else:
try:
port = int(sys.argv[1])
except:
cmd_err()
if port<-1:
cmd_err()
server=AsyncServer(port)
server.run()