· 4 years ago · Mar 20, 2021, 02:34 PM
1import json
2import subprocess
3from time import sleep
4
5''' Декораторы '''
6
7
8def JSON(function):
9 def to_json(*args, **kwargs):
10 ret_string = function(*args, **kwargs)
11
12 json_string = json.dumps(ret_string)
13 return json_string
14
15 return to_json
16
17
18class Engine(object): # от этого класса будем наследовать LeeloZero и API интерфейс
19 def __init__(self):
20 pass
21
22
23class LeelaZero(object):
24 def __init__(self, path_engine, path_weights):
25
26 self.inst = subprocess.Popen("{} --gtp --lagbuffer 0 --weights {}".format(path_engine, path_weights).split(),
27 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=1,
28 universal_newlines=True)
29 while True:
30 line = self.inst.stderr.readline()
31 if "max tree" in line:
32 break
33
34 def _send(self, *args, **kwargs):
35 command = kwargs.setdefault('command', 'name')
36 wait = kwargs.setdefault('wait', 0.)
37 empties = kwargs.setdefault('empties', 2) # количество пустых строк для завершения
38 self.inst.stdin.write(command + '\r\n')
39 self.inst.stdin.flush()
40
41 line = self.inst.stdout.readline()
42 print(line) # for debug
43 sleep(wait)
44
45 self.inst.stdin.write('\r\n')
46 self.inst.stdin.flush()
47 self.inst.stdout.readline()
48
49 lines = []
50 for _ in range(empties):
51 while True:
52 line = self.inst.stderr.readline()
53 line = self.remove_extra_spaces(line[:-1])
54 if not line:
55 break
56 lines.append(line)
57 return lines
58
59 def send(self, requests): # обработчик команд в соответствии с GTP, r = [{key: value (пары kwargs)}, ...]
60 responses = []
61
62 for request in requests:
63 response = self._send(**request)
64 responses.append(response)
65
66 return responses
67
68 def remove_extra_spaces(self, string):
69 if not string:
70 return None
71 return ' '.join(string.split())
72
73 def showboard(self, empties=5, wait=0.):
74 request = {
75 "command": "showboard",
76 "empties": 5,
77 "wait": 0.
78 }
79
80 responce = self.send([request])[0]
81 if not responce:
82 # error --- вообще стоит сделать словарик, характеризующий ошибку, и возвращать именно его
83 return None
84 l = len(responce)
85 Passes = responce[0].partition(' Black')[0].partition('Passes: ')[2]
86 Black_Prisoners = responce[0].partition('Prisoners: ')[2]
87 White_Prisoners = responce[1].partition('Prisoners: ')[2]
88 Who_Moves = responce[1].partition(' to move')[0]
89
90 rows = dict()
91 id_ = 0
92 for id in range(l - 5, 2, -1):
93 id_ += 1
94 row = responce[id].replace(str(id_), '')
95 dct = {id_: row}
96 rows.update(dct)
97
98 Table = {"Headers": [responce[2], responce[-4]], "Rows": rows}
99 Hash = responce[-3].partition(' Ko-Hash')[0].partition('Hash: ')[2]
100 KoHash = responce[-3].partition('Ko-Hash: ')[2]
101 Black_Time = responce[-2].partition('Black time: ')[2]
102 White_Time = responce[-1].partition('White time: ')[2]
103
104 dictionary_responce = {
105 'Passes': Passes,
106 'BlackPrisoners': Black_Prisoners,
107 'WhitePrisoners': White_Prisoners,
108 'Who_Moves': Who_Moves,
109 'Hash': Hash,
110 'KoHash': KoHash,
111 'BlackTime': Black_Time,
112 'WhiteTime': White_Time,
113 'Table': Table,
114 }
115
116 print(responce)
117 return dictionary_responce
118
119
120def run_local(*args, **kwargs):
121 leela = LeelaZero(*args, **kwargs)
122 return leela
123
124
125def run_api(*args, **kwargs):
126 pass
127
128
129@JSON
130def translate(dictionary): # чисто для проверки декоратора
131 return dictionary
132
133def translate_with_indent(dictionary): # красивый вывод в консоль
134 return json.dumps(dictionary, indent=4)
135
136if __name__ == '__main__':
137 print("Creating...")
138
139 leela = run_local(path_engine='leelazero\leelaz.exe',
140 path_weights="F:\Codes\Projects\Go\MachineLearning\Interface\lznetwork.gz")
141
142 print("Loaded.")
143
144 r_dict = leela.showboard()
145 js_dict = translate(r_dict)
146 pretty_dict = translate_with_indent(r_dict)
147 print(r_dict)
148 print(js_dict)
149 print(pretty_dict)
150