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
|
#!/usr/bin/python3
import http.client
import json
import urllib
address, port = 'localhost', 3030
class PrologEngine(object):
def __init__(self, address=address, port=port, code=''):
self.id = None
self.conn = http.client.HTTPConnection(address, port, timeout=10)
hdrs = {'Content-Type': 'application/json; charset=utf-8'}
opts = json.dumps({'destroy': False, 'src_text': code, 'format': 'json'})
self.conn.request('POST', '/pengine/create', body=opts, headers=hdrs)
response = self.conn.getresponse()
data = response.read()
reply = json.loads(str(data, encoding='utf-8'))
if reply['event'] == 'create':
self.id = reply['id']
else:
raise Exception('could not create engine: ' + reply['code'])
def send(self, event, fmt='json'):
params = urllib.parse.urlencode({
'id': self.id,
'event': event,
'format': fmt})
self.conn.request('GET', '/pengine/send?' + params)
response = self.conn.getresponse()
data = str(response.read(), encoding='utf-8')
if response.status == http.client.OK:
return json.loads(data) if fmt == 'json' else data
return None
def ask(self, query, template='', fmt='json'):
event = 'ask(({}),[template({})])'.format(query, template)
reply = self.send(event, fmt=fmt)
return reply
def next(self, n=1):
event = 'next({})'.format(n)
reply = self.send(event)
return reply
def stop(self):
return self.send('stop')
def destroy(self):
reply = self.send('destroy')
self.id = None
self.conn.close()
self.conn = None
# Basic sanity check.
if __name__ == '__main__':
engine = PrologEngine(code='dup([],[]). dup([H|T],[H,H|TT]) :- dup(T,TT).')
print('engine id is ' + engine.id)
print(engine.ask("run_tests({},'{}',Result)".format('dup/2', engine.id)))
engine.destroy()
|
