source: framspy/gui/framsutils/comm.py @ 1245

Last change on this file since 1245 was 1198, checked in by Maciej Komosinski, 22 months ago

Added simple Python GUI for Framsticks library/server
File size: 5.4 KB
Line 
1import asyncore
2import socket
3from io import StringIO
4from typing import Callable, Dict
5import threading
6import asyncio
7import queue
8import time
9
10class EventConsumer(threading.Thread):
11        def __init__(self) -> None:
12                threading.Thread.__init__(self)
13                self.condition = threading.Condition()
14                self.queue = queue.Queue()
15                self.is_running = True
16                self.runningChangeEventCallback = None
17                self.populationsGroupChangeEventCallback = None
18                self.genepoolsGroupChangeEventCallback = None
19                self.messagesEventCallback = None
20
21        def run(self):
22                while self.is_running:
23                        if self.queue.empty():
24                                time.sleep(0.0001)
25                        else:
26                                block = self.queue.get()
27                                #print("------- EVENT -------")
28                                #print(block)
29                                #print("------- END EVENT -------")
30                                header: str = block[:block.find('\n')]
31                                if header.find("running_changed") != -1 and self.runningChangeEventCallback:
32                                        self.runningChangeEventCallback(block, header)
33                                elif header.find("populations/groups_changed") != -1 and self.populationsGroupChangeEventCallback:
34                                        self.populationsGroupChangeEventCallback(block, header)
35                                elif header.find("genepools/groups_changed") != -1 and self.genepoolsGroupChangeEventCallback:
36                                        self.genepoolsGroupChangeEventCallback(block, header)
37                                elif header.find("messages") != -1 and self.messagesEventCallback:
38                                        self.messagesEventCallback(block, header)
39
40        def stop(self):
41                self.is_running = False
42
43class CommWrapper(object):
44        def start(self, host: str, port: int):
45                self.client = Comm(host, port, self.onClose)
46                self.thread = threading.Thread(target=asyncore.loop, kwargs={'timeout': 0.0005})
47                self.thread.start()
48                self.client.write_data("version 5")
49                self.client.write_data("use request_id")
50                self.client.write_data("use needfile_id")
51                self.client.write_data("use groupnames")
52                self.connected = True
53
54        def stop(self):
55                self.client.close()
56                self.thread.join()
57
58        def onClose(self):
59                self.connected = False
60
61        def close(self):
62                self.thread.join()
63
64        async def read(self, i: int, timeout: float) -> str:
65                data = ""
66                try:
67                        data = await asyncio.wait_for(self._read(i), timeout=timeout)
68                except:
69                        pass
70                #print("response")
71                #print(repr(data))
72                return data
73
74        async def _read(self, i: int) -> str:
75                while True:
76                        if i in self.client.blocks:
77                                return self.client.blocks.pop(i)
78                        try:
79                                await asyncio.sleep(0.005)
80                        except asyncio.CancelledError:
81                                break
82
83        def write(self, data: str, index = -1) -> int:
84                command = data.split(' ')[0]
85                if command in ["info", "set", "get", "call", "reg", "file"]:
86                        if index == -1:
87                                index = self.client.getIndex()
88                        data = command + ' r' + str(index) + data[len(command):]
89                self.client.write_data(data)
90                return index
91
92class Comm(asyncore.dispatcher):
93        def __init__(self, host: str, port: int, onClose: Callable[[None], None] = None) -> None:
94                self.onClose = onClose
95
96                asyncore.dispatcher.__init__(self)
97                self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
98                self.host = host
99                self.port = port
100               
101                self.connect((self.host, self.port))
102                self.out_buffer = "".encode()
103                self.in_buffer = StringIO()
104                self.blocks: Dict[str] = {}
105                self._tmpbuff = StringIO()
106                self.index = 1
107                self.is_embedded = False
108
109                self.consumer = EventConsumer()
110                self.consumer.start()
111       
112        def writable(self):
113                is_writable = (len(self.out_buffer) > 0)
114                return is_writable
115
116        def readable(self):
117                return True
118
119        def handle_close(self):
120                self.close()
121
122        def close(self) -> None:
123                self.in_buffer.close()
124                self._tmpbuff.close()
125                self.consumer.stop()
126                if self.onClose:
127                        self.onClose()
128                super().close()
129
130        def handle_connect(self):
131                pass
132
133        def handle_read(self):
134                rec_msg = self.recv(1024)
135                self.in_buffer.write(rec_msg.decode())
136                self.parse()
137
138        def handle_write(self) -> None:
139                sent = self.send(self.out_buffer)
140                self.out_buffer = self.out_buffer[sent:]
141
142        def write_data(self, data: str):
143                #print("request")
144                #print(repr(data))
145                self.out_buffer += (data + '\n' if data[-1] != '\n' else '').encode()
146
147        def read_data(self):
148                return self.in_buffer.getvalue()
149
150        def parse(self):
151                self.in_buffer.seek(0)
152                for line in self.in_buffer:
153                        if line.endswith('\n'):
154                                self._tmpbuff.write(line)
155                                if line.endswith(":~\n"):
156                                        self.is_embedded = True
157                                if self.is_embedded:
158                                        if line.endswith("~\n") and not line.endswith("\~\n"):
159                                                self.is_embedded = False
160                                else:
161                                        if line.startswith("eof"):
162                                                block = self._tmpbuff.getvalue().strip()
163                                                if block.startswith("event"):
164                                                        self.consumer.queue.put(block)
165                                                        self._tmpbuff.truncate(0)
166                                                        self._tmpbuff.seek(0)
167                                        elif line.startswith("ok") or line.startswith("error") or line.startswith("needfile"):
168                                                block = self._tmpbuff.getvalue()
169                                                if line.find(' ') > -1:
170                                                        self.blocks[int(line.split(' ')[1][1:])] = block
171                                                self._tmpbuff.truncate(0)
172                                                self._tmpbuff.seek(0)
173                self.in_buffer.truncate(0)
174                self.in_buffer.seek(0)
175                if not line.endswith('\n'):
176                        self.in_buffer.write(line)
177
178        def getIndex(self) -> int:
179                # Not every response could be fetched, because of timeout of request.
180                # Stale responses needs to be cleaned to prevent stack growing.
181                r = 100 # Remove all messagess older than r
182                if self.index % r != (self.index + 1) % r:
183                        self.blocks = {k: v for k, v in self.blocks.items() if k > self.index - r}
184                self.index += 1
185                return self.index - 1
Note: See TracBrowser for help on using the repository browser.