58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
import asyncio
|
|
import enum
|
|
import json
|
|
import logging
|
|
|
|
import websockets
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class BaseData:
|
|
def __init__(self, host: str = '127.0.0.1', port: int = 2946,
|
|
path: str = '/BSDataPuller'):
|
|
self.host = host
|
|
self.port = port
|
|
self.path = path
|
|
|
|
async def listen(self):
|
|
try:
|
|
async with websockets.connect(
|
|
'ws://{}:{}{}'.format(self.host, self.port, self.path)) as websocket:
|
|
async for message in websocket:
|
|
await self._handle(message)
|
|
except Exception as e:
|
|
logger.critical(e)
|
|
|
|
|
|
class LiveData(BaseData):
|
|
TRIGGERS = enum.Enum('Triggers',
|
|
['Unknown', 'TimerElapsed', 'NoteMissed', 'EnergyChange', 'ScoreChange'])
|
|
|
|
def __init__(self, host: str = '127.0.0.1', port: int = 2946,
|
|
path: str = '/BSDataPuller/LiveData'):
|
|
super().__init__(host, port, path)
|
|
self.events = {}
|
|
|
|
def on(self, event: str, function):
|
|
if event not in self.events:
|
|
self.events[event] = []
|
|
self.events[event].append(function)
|
|
|
|
def _handle(self, message: str):
|
|
for event in events[TRIGGERS(message['EventTrigger']).name]:
|
|
event(message)
|
|
|
|
|
|
class MapData(BaseData):
|
|
def __init__(self, host: str = '127.0.0.1', port: int = 2946,
|
|
path: str = '/BSDataPuller/MapData'):
|
|
super().__init__(host, port, path)
|
|
|
|
def on_update(self, function):
|
|
self.function = function
|
|
|
|
def _handle(self, message: str):
|
|
self.function(message)
|