Python 3 Каскадный WebSockets сервера

Я строй приложения на WebSocket сервера в Python 3. Я использую эту реализацию: https://websockets.readthedocs.io/Python 3 Каскадный WebSockets сервера

В основном я хочу управлять несколько клиентом. Кроме того, я хочу, чтобы отправить данные из 2 разных потока (один для GPS + один для ИДУ) GPS поток обновляется 1Гц, в то время как ИДУ нить обновления на 25Hz

Моя проблема заключается в методе MSGWorker.sendData: как только Я раскомментирую строку @ asyncio.coroutine и выйду из websocket.send ('{"GPS": "% s"}'% data) весь метод ничего не делает (нет печати («Отправить данные: foo») в терминале)

Однако, если эти две строки прокомментированы, мой код работает так, как я ожидал, за исключением того, что я ничего не посылаю через websocket.

Но, конечно, моя цель - отправить данные через websocket, я просто не понимаю, почему это не работает? Есть идеи ?

server.py

#!/usr/bin/env python3 
import signal, sys 
sys.path.append('.') 
import time 
import websockets 
import asyncio 
import threading 

connected = set() 
stopFlag = False 



class GPSWorker (threading.Thread): 
    def __init__(self): 
    threading.Thread.__init__(self) 
    self.data = 0 
    self.lastData = 0 
    self.inc = 0 

    # Simulate GPS data 
    def run(self): 
    while not stopFlag: 
     self.data = self.inc 
     self.inc += 1 
     time.sleep(1) 

    def get(self): 
    if self.lastData is not self.data: 
     self.lastData = self.data 
     return self.data 



class IMUWorker (threading.Thread): 
    def __init__(self): 
    threading.Thread.__init__(self) 
    self.data = 0 
    self.lastData = 0 
    self.inc = 0 

    # Simulate IMU data 
    def run(self): 
    while not stopFlag: 
     self.data = self.inc 
     self.inc += 1 
     time.sleep(0.04) 

    def get(self): 
    if self.lastData is not self.data: 
     self.lastData = self.data 
     return self.data 



class MSGWorker (threading.Thread): 
    def __init__(self): 
    threading.Thread.__init__(self) 

    def run(self): 
    while not stopFlag: 
     data = gpsWorker.get() 
     if data: 
     self.sendData('{"GPS": "%s"}' % data)   

     data = imuWorker.get() 
     if data: 
     self.sendData('{"IMU": "%s"}' % data) 

     time.sleep(0.04) 

    #@asyncio.coroutine 
    def sendData(self, data): 
    for websocket in connected.copy(): 
     print("Sending data: %s" % data) 
     #yield from websocket.send('{"GPS": "%s"}' % data) 



@asyncio.coroutine 
def handler(websocket, path): 
    global connected 
    connected.add(websocket) 
    #TODO: handle client disconnection 
    # i.e connected.remove(websocket) 



if __name__ == "__main__": 
    print('aeroPi server') 
    gpsWorker = GPSWorker() 
    imuWorker = IMUWorker() 
    msgWorker = MSGWorker() 

    try: 
    gpsWorker.start() 
    imuWorker.start() 
    msgWorker.start() 

    start_server = websockets.serve(handler, 'localhost', 7700) 
    loop = asyncio.get_event_loop() 
    loop.run_until_complete(start_server) 
    loop.run_forever() 

    except KeyboardInterrupt: 
    stopFlag = True 
    loop.close() 
    print("Exiting program...")

client.html

<!doctype html> 
<html> 
<head> 
    <meta charset="UTF-8" /> 
</head> 
<body> 
</body> 
</html> 
<script type="text/javascript"> 
    var ws = new WebSocket("ws://localhost:7700", 'json'); 
    ws.onmessage = function (e) { 
    var data = JSON.parse(e.data); 
    console.log(data); 
    }; 
</script>

Спасибо вам за вашу помощь

источник

2016-05-11 Clément de l' Hamaide

Наконец я получил его! Он требовал Python 3.5.1 (в то время как мой дистрибутив предоставил только 3.4.3) и некоторую помощь от Aymeric, автора библиотеки websockets (спасибо ему).

#!/usr/bin/env python3 
import signal, sys 
sys.path.append('.') 
import time 
import websockets 
import asyncio 
import threading 


stopFlag = False 



class GPSWorker (threading.Thread): 
    def __init__(self): 
    threading.Thread.__init__(self) 
    self.data = 0 
    self.lastData = 0 
    self.inc = 0 

    # Simulate GPS data 
    def run(self): 
    while not stopFlag: 
     self.data = self.inc 
     self.inc += 1 
     time.sleep(1) 

    def get(self): 
    if self.lastData is not self.data: 
     self.lastData = self.data 
     return self.data 



class IMUWorker (threading.Thread): 
    def __init__(self): 
    threading.Thread.__init__(self) 
    self.data = 0 
    self.lastData = 0 
    self.inc = 0 

    # Simulate IMU data 
    def run(self): 
    while not stopFlag: 
     self.data = self.inc 
     self.inc += 1 
     time.sleep(0.04) 

    def get(self): 
    if self.lastData is not self.data: 
     self.lastData = self.data 
     return self.data 



class MSGWorker (threading.Thread): 
    def __init__(self): 
    threading.Thread.__init__(self) 
    self.connected = set() 

    def run(self): 
    while not stopFlag: 
     data = gpsWorker.get() 
     if data: 
     self.sendData('{"GPS": "%s"}' % data) 

     data = imuWorker.get() 
     if data: 
     self.sendData('{"IMU": "%s"}' % data) 

     time.sleep(0.04) 

    async def handler(self, websocket, path): 
    self.connected.add(websocket) 
    try: 
     await websocket.recv() 
    except websockets.exceptions.ConnectionClosed: 
     pass 
    finally: 
     self.connected.remove(websocket) 

    def sendData(self, data): 
    for websocket in self.connected.copy(): 
     print("Sending data: %s" % data) 
     coro = websocket.send(data) 
     future = asyncio.run_coroutine_threadsafe(coro, loop) 



if __name__ == "__main__": 
    print('aeroPi server') 
    gpsWorker = GPSWorker() 
    imuWorker = IMUWorker() 
    msgWorker = MSGWorker() 

    try: 
    gpsWorker.start() 
    imuWorker.start() 
    msgWorker.start() 

    ws_server = websockets.serve(msgWorker.handler, '0.0.0.0', 7700) 
    loop = asyncio.get_event_loop() 
    loop.run_until_complete(ws_server) 
    loop.run_forever() 
    except KeyboardInterrupt: 
    stopFlag = True 
    #TODO: close ws server and loop correctely 
    print("Exiting program...")

С уважением, Климент

источник

2016-05-13 18:18:20

Python 3 Каскадный WebSockets сервера

ответ

Смежные вопросы