Получать потоковые данные после реализации asyncio websockets как класс? - PullRequest
Новая
       11

Получать потоковые данные после реализации asyncio websockets как класс?

1 голос
/ 14 апреля 2020

Мой вопрос тесно связан со следующим вопросом о Stackoverflow и документацией здесь . Я определяю websockets -соединение как класс. Затем я создаю новый класс, в котором я вызываю ранее определенный класс websocket как self.ws и сообщаю, какие данные отправлять в веб-сокет с помощью self.request. Моя проблема в том, что текущий скрипт запускается только один раз, тогда как желаемый результат - непрерывные данные.

Вторая ссылка показывает, что я могу извлекать непрерывные / потоковые данные, используя 

asyncio.get_event_loop().run_until_complete(call_api(json.dumps(msg)))

Я включаю весь вышеупомянутый код в свой код (call_api определяется по-разному из-за желания напиши это как класс). Ниже приведен мой код: 

import sys, json
import asyncio
from websockets import connect

class EchoWebsocket:
   def __init__(self, URL, CLIENT_ID=None, CLIENT_SECRET=None):
      self.url = URL
      self.client_id = CLIENT_ID
      self.client_secret = CLIENT_SECRET

   async def __aenter__(self):
      self._conn = connect(self.url)
      self.websocket = await self._conn.__aenter__()
      return self

   async def __aexit__(self, *args, **kwargs):
      await self._conn.__aexit__(*args, **kwargs)

   async def send(self, message):
      await self.websocket.send(message)

   async def receive(self):
      return await self.websocket.recv()

class DERIBIT:
   def __init__(self):
      self.ws = EchoWebsocket(URL='wss://test.deribit.com/ws/api/v2')
      self.loop = asyncio.get_event_loop()
      self.request = \
                   {"jsonrpc": "2.0",
                    "method": "public/subscribe",
                    "id": 42,
                    "params": {
                        "channels": ["deribit_price_index.btc_usd"]}
                   }

   def get_ticks(self):
      return self.loop.run_until_complete(self.__async__get_ticks())

   async def __async__get_ticks(self):
      async with self.ws as echo:
         await echo.send(json.dumps(self.request))
         response = await echo.receive()
         print(response)


if __name__ == "__main__":
   deribit = DERIBIT()
   deribit.get_ticks()

Этот скрипт дает следующий вывод:

{"jsonrp c": "2.0", "method": "public / subscribe "," id ": 42," params ": {" channel ": [" deribit_price_index.btc_usd "]}}

, тогда как я хотел бы видеть

enter image description here

Пожалуйста, совет.

Ответы [ 2 ]

1 голос
/ 23 апреля 2020

проблема в функции

первый loop.run_until_complete, запущенный до завершения будущего do c run_until_complete
, означающий, что ваша функция-получатель выполнит только один ответ. run_until_complete не является callback функцией!.

, поэтому в вашем случае main:
deribit.get_ticks() -> запустить будущий экземпляр __async__get_ticks
, поэтому __async__get_ticks задача: давайте посмотрим, что задача делает:
1.открытое соединение ws:
2.отправленный запрос
3.дождаться ответа от ws
4. print (response)
here the task is done вот почему вы видите только одну строку 

   async def __async__get_ticks(self):
      async with self.ws as echo:
         await echo.send(json.dumps(self.request))
         response = await echo.receive()
         print(response)

после объяснения: решение будет простым: нужно обернуть строку response с помощью while 

async def __async__get_ticks(self):
      async with self.ws as echo:
         await echo.send(json.dumps(self.request))
         while True:
                response = await echo.receive()
                print(response)

output 

{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654476817,"price":7540.54,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654477824,"price":7540.52,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654478831,"price":7540.15,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654479838,"price":7539.83,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654480845,"price":7539.2,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654481852,"price":7538.96,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654482859,"price":7538.9,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654483866,"price":7538.89,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654484873,"price":7538.47,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654485880,"price":7537.15,"index_name":"btc_usd"}}}
0 голосов
/ 19 апреля 2020

Я работал только с веб-сокетами Торнадо, но они работают довольно хорошо, и у Торнадо есть много помощников для работы с асин c код: 

import json
import tornado
from tornado.ioloop import PeriodicCallback
from tornado.websocket import websocket_connect


class EchoWebsocket:

    def __init__(self, url, client_id=None, client_secret=None):
        self.url = url
        self.client_id = client_id
        self.client_secret = client_secret
        self.websocket = None

    async def connect(self):
        if not self.websocket:
            self.websocket = await websocket_connect(self.url)

    async def close(self):
        await self.websocket.close()
        self.websocket = None

    async def read(self):
        return await self.websocket.read_message()

    async def write(self, message):
        await self.websocket.write_message(message)


class DERIBIT:

    def __init__(self):
        self.ws = EchoWebsocket(url='wss://test.deribit.com/ws/api/v2')
        self.request = {
            "jsonrpc": "2.0",
            "method": "public/subscribe",
            "id": 42,
            "params": {
                "channels": ["deribit_price_index.btc_usd"]}
        }
        self.callback = PeriodicCallback(self.get_ticks, 1000)
        self.callback.start()

    async def get_ticks(self):
        if not self.ws.websocket:
            await self.ws.connect()
        await self.ws.write(json.dumps(self.request))
        response = await self.ws.read()
        print(response)


if __name__ == "__main__":
    deribit = DERIBIT()
    tornado.ioloop.IOLoop.current().start()

Вывод: 

{"jsonrpc":"2.0","id":42,"result":["deribit_price_index.btc_usd"],"usIn":1587298852138977,"usOut":1587298852139023,"usDiff":46,"testnet":true}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587298851526,"price":7173.46,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587298852533,"price":7173.53,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","id":42,"result":["deribit_price_index.btc_usd"],"usIn":1587298852932540,"usOut":1587298852932580,"usDiff":40,"testnet":true}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587298852533,"price":7173.53,"index_name":"btc_usd"}}}

Приведенный выше пример может быть значительно упрощен, если вы интегрируете websocket в класс DERIBIT, а не создаете для него отдельный класс.

...