rtgg_obs

Complexity

Total Complexity

27

Size/Duplication

Total Lines	117
Duplicated Lines	0 %

Importance

Changes

0

6 Methods

Rating	Name	Duplication	Size	Complexity
A	RacetimeObs.__init__()	0	2	1
A	RacetimeObs.process_ws_message()	0	14	5
B	RacetimeObs.process_messages()	0	19	6
A	RacetimeObs.update_logger()	0	17	5
B	RacetimeObs.race_updater()	0	19	8
A	RacetimeObs.race_update_thread()	0	5	1

import json
import websockets
from websockets.client import WebSocketClientProtocol
from datetime import datetime, timedelta, timezone
import racetime_client
from models.race import Race, race_from_dict
import asyncio
import websockets
import dateutil
import logging
from helpers.LogFormatter import LogFormatter
from gadgets.timer import Timer
from gadgets.coop import Coop
from gadgets.qualifier import Qualifier

def script_description():
    return "<p>You've loaded the incorrect script.<br><br>Please remove this file and add 'racetime_obs.py' instead</p>"

class RacetimeObs():
    logger = logging.Logger("racetime-obs")
    race: Race = None
    selected_race = ""
    check_race_updates = False
    race_changed = False
    full_name = ""
    category = ""
    timer = Timer()
    coop = Coop()
    qualifier = Qualifier()

    def __init__(self):
        self.timer.logger = self.coop.logger = self.qualifier.logger = self.logger

    

    def race_update_thread(self):
        self.logger.debug("starting race update")
        race_event_loop = asyncio.new_event_loop()
        race_event_loop.run_until_complete(self.race_updater())
        race_event_loop.run_forever()


    async def race_updater(self):
        headers = {
            'User-Agent': "oro-obs-bot_alpha"
        }
        host = "racetime.gg"

        while True:
            if not self.timer.enabled:
                await asyncio.sleep(5.0)
            else:
                if self.race is None and self.selected_race != "":
                    self.race = racetime_client.get_race(self.selected_race)
                if self.race is not None and self.race.websocket_url != "":
                    async with websockets.connect("wss://racetime.gg" + self.race.websocket_url, host=host, extra_headers=headers) as ws:
                        self.race_changed = False
                        self.logger.info(
                            f"connected to websocket: {self.race.websocket_url}")
                        await self.process_messages(ws)
            await asyncio.sleep(5.0)


    async def process_messages(self, ws: WebSocketClientProtocol):
        last_pong = datetime.now(timezone.utc)
        while True:
            try:
                if self.race_changed:
                    self.logger.info("new race selected")
                    self.race_changed = False
                    break
                message = await asyncio.wait_for(ws.recv(), 5.0)
                self.logger.info(f"received message from websocket: {message}")
                data = json.loads(message)
                last_pong = self.process_ws_message(data, last_pong)
            except asyncio.TimeoutError:
                if datetime.now(timezone.utc) - last_pong > timedelta(seconds=20):
                    await ws.send(json.dumps({"action": "ping"}))
            except websockets.ConnectionClosed:
                self.logger.error(f"websocket connection closed")
                self.race = None
                break

    def process_ws_message(self, data, last_pong):
        if data.get("type") == "race.data":
            r = race_from_dict(data.get("race"))
            self.logger.debug(f"race data parsed: {r}")
            self.logger.debug(f"current race is {self.race}")
            if r is not None and r.version > self.race.version:
                self.race = r
                self.logger.debug(f"self.race is {self.race}")
                self.coop.update_coop_text(self.race, self.full_name)
                self.qualifier.update_qualifier_text(self.race, self.full_name)
        elif data.get("type") == "pong":
            last_pong = dateutil.parser.parse(data.get("date"))
            pass
        return last_pong


    def update_logger(self, enabled: bool, log_to_file: bool, log_file: str, level: str):
        self.logger.disabled = not enabled
        self.logger.handlers = []
        handler = logging.StreamHandler()
        if log_to_file:
            try:
                handler = logging.FileHandler(log_file)
            except:
                self.logger.error(f"Unable to open {log_file}")
        elif level == "Debug":
            handler.setLevel(logging.DEBUG)
        elif level == "Info":
            handler.setLevel(logging.INFO)
        else:
            handler.setLevel(logging.ERROR)
        handler.setFormatter(LogFormatter())
        self.logger.addHandler(handler)