gadgets.ladder_timer.LadderTimer.update_ladder_status() - Code Metrics - Inspection of "join multiple ladder races in one obs session" - ssbm-oro/racetime-obs - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 6a2c9c...d2bb42 )

by manny

created 2021-09-01 08:31 UTC

LadderTimer.update_ladder_status() A

↳ Parent: gadgets.ladder_timer

Complexity

Conditions

Size

Total Lines	10
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	10
nop	1
dl	0
loc	10
rs	9.9
c	0
b	0
f	0

import asyncio
import logging
from datetime import datetime, timedelta
import pytz
from typing import List
from threading import Thread

from helpers import timer_to_str
import clients.ladder_client as ladder_client
from models.ladder import Flag, Racer, ScheduleItem, Season, Standings


def str_to_timer(str_timer):
    return timedelta(
        hours=float(str_timer[0:1]), minutes=float(str_timer[2:4]),
        seconds=float(str_timer[5:7]))


class LadderTimer:
    logger: logging.Logger = None
    enabled: bool = False
    source_name = ""
    timer_text = ""
    racer_id: int = 0
    racer_name: str = ""
    stats_source: str = ""
    stats: Standings = None
    season_for_stats: int = 0
    mode_for_stats: int = 0
    show_season_name: bool = False
    show_mode_name: bool = False
    show_rating: bool = False
    show_rank: bool = False
    show_change: bool = False
    show_win_loss_tie: bool = False
    pre_color: int = 0xFFFFFF
    racing_color: int = 0xFFFFFF
    winner_color: int = 0xFF0000
    loser_color: int = 0x00FF00
    ff_color: int = 0x00FF00
    started_at: datetime = None
    finish_time: timedelta = None
    result: str = ""
    next_race: ScheduleItem = None
    active_racers: List[Racer] = None
    current_season: Season = None
    flags: List[Flag] = []
    all_seasons: List[Season] = []
    schedule: List[ScheduleItem] = []
    last_timer_update: datetime = None
    decimals: bool = True

    @staticmethod
    def ladder_timezone():
        return pytz.timezone('US/Eastern')

    def __init__(self, logger: logging.Logger, racer_id: int = None):
        self.logger = logger
        if racer_id is not None:
            self.racer_id = racer_id

        ladder_update_t = Thread(target=self.ladder_update_thread)
        ladder_update_t.daemon = True
        ladder_update_t.start()

    def ladder_update_thread(self):
        self.logger.debug("starting ladder race update")
        loop = asyncio.new_event_loop()
        loop.run_until_complete(self.race_updater())
        loop.run_forever()

    async def race_updater(self):
        await self.update_ladder_status()
        while True:
            if not self.enabled or self.finish_time is not None:
                await asyncio.sleep(5.0)
            else:
                now = datetime.now(self.ladder_timezone())
                if self.started_at is not None and now > self.started_at:
                    self.logger.debug("checking if racer finished")
                    await self.check_racer_finish()
                await asyncio.sleep(5.0)

    async def update_ladder_status(self):
        self.started_at = None
        self.finish_time = None
        self.last_timer_update = None
        await self.update_active_racers()
        await self.update_flags()
        await self.update_season()
        await self.update_schedule(self.current_season.season_id)
        await self.update_timer()
        await self.update_stats()

    async def check_racer_finish(self):
        if ladder_client.is_currently_racing(self.racer_id):
            return
        racer_history = ladder_client.get_racer_history(self.racer_id)
        if racer_history[-1].race_id == self.next_race.race_id:
            if racer_history[-1].FinishTime == "FF":
                self.result = "FF"
                self.finish_time = timedelta(seconds=0)
                self.logger.debug("racer forfeited")
            else:
                self.finish_time = str_to_timer(racer_history[-1].FinishTime)
                self.result = racer_history[-1].Result

    def get_timer_text(self):
        if self.finish_time:
            if self.result == "W":
                return (
                    self.winner_color,
                    timer_to_str(self.finish_time, self.decimals)
                )
            elif self.result == "L":
                return (
                    self.loser_color,
                    timer_to_str(self.finish_time, self.decimals)
                )
            elif self.result == "FF":
                return self.ff_color, "-:--:--.-"
        current_timer = timedelta(seconds=0)
        color = self.pre_color
        now = datetime.now(self.ladder_timezone())
        if self.started_at is None:
            current_timer = now - self.next_race.StartTime
            if (abs(current_timer.total_seconds()) < 600.0):
                loop = asyncio.new_event_loop()
                loop.run_until_complete(self.update_timer())
                loop.close()
            color = self.pre_color
        else:
            current_timer = now - self.started_at
            if current_timer.total_seconds() > 0:
                color = self.racing_color
        return color, timer_to_str(current_timer, self.decimals)

    def get_stats(self):
        stats = ""
        if self.enabled:
            if self.show_season_name:
                stats = stats + self.stats.Season + " "
            if self.show_mode_name:
                stats = stats + self.stats.Mode + " "
            if self.show_rank:
                stats = stats + "#" + str(self.stats.Rank) + " "
            if self.show_rating:
                stats = stats + str(self.stats.Rating) + " "
            if self.show_change:
                stats = stats + str(self.stats.Change) + " "
            if self.show_win_loss_tie:
                stats = (
                    stats + str(self.stats.Wins) + "/" +
                    str(self.stats.Losses) + "/" + str(self.stats.Ties)
                )
        return stats

    def update_settings(self, racer_name: str):
        if racer_name is None or racer_name == "":
            return False
        racer_id = self.get_racer_id(racer_name)
        if racer_id == 0:
            return False
        else:
            self.racer_id = racer_id
            self.racer_name = racer_name
            loop = asyncio.new_event_loop()
            loop.run_until_complete(self.update_stats())
            loop.close()
            return True

    def get_racer_id(self, racer_name: str) -> Racer:
        loop = asyncio.new_event_loop()
        loop.run_until_complete(self.update_active_racers())
        loop.close()
        for racer in self.active_racers:
            if racer.RacerName.lower() == racer_name.lower():
                return racer.racer_id
        return 0

    async def update_active_racers(self):
        if self.active_racers is None or self.active_racers == []:
            self.active_racers = ladder_client.get_active_racers()

    async def update_season(self):
        if self.current_season is None:
            self.all_seasons = ladder_client.get_seasons()
            if self.all_seasons is not None:
                self.current_season = (
                    next(filter(
                        lambda x: x.IsCurrentSeason, self.all_seasons
                    ), None)
                )
        self.logger.info(f"current_season = {self.current_season}")

    async def update_flags(self):
        self.flags = ladder_client.get_flags()

    async def update_schedule(self, season_id: int):
        if self.schedule is None or self.schedule == []:
            self.schedule = ladder_client.get_schedule(season_id)
            self.next_race = (
                next(x for x in self.schedule if not x.HasCompleted)
            )
            self.logger.info(f"next_race = {self.next_race}")

    async def update_timer(self):
        if (
            self.last_timer_update is not None and
            (datetime.now() - self.last_timer_update) < timedelta(seconds=20)
        ):
            return
        self.logger.debug("calling get_current_race_time")
        str_timer = ladder_client.get_current_race_time()
        self.last_timer_update = datetime.now()
        self.logger.info(f"str_timer= {str_timer}")
        timer_sign = 1.0
        if str_timer[0] == '-':
            timer_sign = -1.0
            str_timer = str_timer[1:]
        timer = str_to_timer(str_timer)
        timer = timer * timer_sign
        if timer == timedelta(seconds=0):
            self.started_at = None
        else:
            self.started_at = datetime.now(self.ladder_timezone()) - timer

    async def update_stats(self):
        if self.season_for_stats == -1:
            self.season_for_stats = self.current_season.season_id
        if self.mode_for_stats == -1:
            self.mode_for_stats = self.get_mode_id_from_name(
                self.next_race.Mode)
        self.logger.info(f"season_for_stats: {self.season_for_stats}")
        self.logger.info(f"mode_for_stats: {self.mode_for_stats}")
        standings = ladder_client.get_standings(
            self.season_for_stats, self.mode_for_stats)
        for stats in standings:
            if stats.RacerName.lower() == self.racer_name.lower():
                self.stats = stats
                return
        # default standings
        self.stats = Standings(
            self.racer_name, self.current_season.SeasonName,
            self.next_race.Mode, 1600, 0, 0, 0, 0, 0)

    def get_mode_id_from_name(self, mode_name: str):
        for flag in self.flags:
            if flag.Mode == mode_name:
                return flag.flag_id


1			import asyncio
2			import logging
3			from datetime import datetime, timedelta
4			import pytz
5			from typing import List
6			from threading import Thread
7
8			from helpers import timer_to_str
9			import clients.ladder_client as ladder_client
10			from models.ladder import Flag, Racer, ScheduleItem, Season, Standings
11
12
13			def str_to_timer(str_timer):
14			return timedelta(
15			hours=float(str_timer[0:1]), minutes=float(str_timer[2:4]),
16			seconds=float(str_timer[5:7]))
17
18
19			class LadderTimer:
20			logger: logging.Logger = None
21			enabled: bool = False
22			source_name = ""
23			timer_text = ""
24			racer_id: int = 0
25			racer_name: str = ""
26			stats_source: str = ""
27			stats: Standings = None
28			season_for_stats: int = 0
29			mode_for_stats: int = 0
30			show_season_name: bool = False
31			show_mode_name: bool = False
32			show_rating: bool = False
33			show_rank: bool = False
34			show_change: bool = False
35			show_win_loss_tie: bool = False
36			pre_color: int = 0xFFFFFF
37			racing_color: int = 0xFFFFFF
38			winner_color: int = 0xFF0000
39			loser_color: int = 0x00FF00
40			ff_color: int = 0x00FF00
41			started_at: datetime = None
42			finish_time: timedelta = None
43			result: str = ""
44			next_race: ScheduleItem = None
45			active_racers: List[Racer] = None
46			current_season: Season = None
47			flags: List[Flag] = []
48			all_seasons: List[Season] = []
49			schedule: List[ScheduleItem] = []
50			last_timer_update: datetime = None
51			decimals: bool = True
52
53			@staticmethod
54			def ladder_timezone():
55			return pytz.timezone('US/Eastern')
56
57			def __init__(self, logger: logging.Logger, racer_id: int = None):
58			self.logger = logger
59			if racer_id is not None:
60			self.racer_id = racer_id
61
62			ladder_update_t = Thread(target=self.ladder_update_thread)
63			ladder_update_t.daemon = True
64			ladder_update_t.start()
65
66			def ladder_update_thread(self):
67			self.logger.debug("starting ladder race update")
68			loop = asyncio.new_event_loop()
69			loop.run_until_complete(self.race_updater())
70			loop.run_forever()
71
72			async def race_updater(self):
73			await self.update_ladder_status()
74			while True:
75			if not self.enabled or self.finish_time is not None:
76			await asyncio.sleep(5.0)
77			else:
78			now = datetime.now(self.ladder_timezone())
79			if self.started_at is not None and now > self.started_at:
80			self.logger.debug("checking if racer finished")
81			await self.check_racer_finish()
82			await asyncio.sleep(5.0)
83
84			async def update_ladder_status(self):
85			self.started_at = None
86			self.finish_time = None
87			self.last_timer_update = None
88			await self.update_active_racers()
89			await self.update_flags()
90			await self.update_season()
91			await self.update_schedule(self.current_season.season_id)
92			await self.update_timer()
93			await self.update_stats()
94
95			async def check_racer_finish(self):
96			if ladder_client.is_currently_racing(self.racer_id):
97			return
98			racer_history = ladder_client.get_racer_history(self.racer_id)
99			if racer_history[-1].race_id == self.next_race.race_id:
100			if racer_history[-1].FinishTime == "FF":
101			self.result = "FF"
102			self.finish_time = timedelta(seconds=0)
103			self.logger.debug("racer forfeited")
104			else:
105			self.finish_time = str_to_timer(racer_history[-1].FinishTime)
106			self.result = racer_history[-1].Result
107
108			def get_timer_text(self):
109			if self.finish_time:
110			if self.result == "W":
111			return (
112			self.winner_color,
113			timer_to_str(self.finish_time, self.decimals)
114			)
115			elif self.result == "L":
116			return (
117			self.loser_color,
118			timer_to_str(self.finish_time, self.decimals)
119			)
120			elif self.result == "FF":
121			return self.ff_color, "-:--:--.-"
122			current_timer = timedelta(seconds=0)
123			color = self.pre_color
124			now = datetime.now(self.ladder_timezone())
125			if self.started_at is None:
126			current_timer = now - self.next_race.StartTime
127			if (abs(current_timer.total_seconds()) < 600.0):
128			loop = asyncio.new_event_loop()
129			loop.run_until_complete(self.update_timer())
130			loop.close()
131			color = self.pre_color
132			else:
133			current_timer = now - self.started_at
134			if current_timer.total_seconds() > 0:
135			color = self.racing_color
136			return color, timer_to_str(current_timer, self.decimals)
137
138			def get_stats(self):
139			stats = ""
140			if self.enabled:
141			if self.show_season_name:
142			stats = stats + self.stats.Season + " "
143			if self.show_mode_name:
144			stats = stats + self.stats.Mode + " "
145			if self.show_rank:
146			stats = stats + "#" + str(self.stats.Rank) + " "
147			if self.show_rating:
148			stats = stats + str(self.stats.Rating) + " "
149			if self.show_change:
150			stats = stats + str(self.stats.Change) + " "
151			if self.show_win_loss_tie:
152			stats = (
153			stats + str(self.stats.Wins) + "/" +
154			str(self.stats.Losses) + "/" + str(self.stats.Ties)
155			)
156			return stats
157
158			def update_settings(self, racer_name: str):
159			if racer_name is None or racer_name == "":
160			return False
161			racer_id = self.get_racer_id(racer_name)
162			if racer_id == 0:
163			return False
164			else:
165			self.racer_id = racer_id
166			self.racer_name = racer_name
167			loop = asyncio.new_event_loop()
168			loop.run_until_complete(self.update_stats())
169			loop.close()
170			return True
171
172			def get_racer_id(self, racer_name: str) -> Racer:
173			loop = asyncio.new_event_loop()
174			loop.run_until_complete(self.update_active_racers())
175			loop.close()
176			for racer in self.active_racers:
177			if racer.RacerName.lower() == racer_name.lower():
178			return racer.racer_id
179			return 0
180
181			async def update_active_racers(self):
182			if self.active_racers is None or self.active_racers == []:
183			self.active_racers = ladder_client.get_active_racers()
184
185			async def update_season(self):
186			if self.current_season is None:
187			self.all_seasons = ladder_client.get_seasons()
188			if self.all_seasons is not None:
189			self.current_season = (
190			next(filter(
191			lambda x: x.IsCurrentSeason, self.all_seasons
192			), None)
193			)
194			self.logger.info(f"current_season = {self.current_season}")
195
196			async def update_flags(self):
197			self.flags = ladder_client.get_flags()
198
199			async def update_schedule(self, season_id: int):
200			if self.schedule is None or self.schedule == []:
201			self.schedule = ladder_client.get_schedule(season_id)
202			self.next_race = (
203			next(x for x in self.schedule if not x.HasCompleted)
204			)
205			self.logger.info(f"next_race = {self.next_race}")
206
207			async def update_timer(self):
208			if (
209			self.last_timer_update is not None and
210			(datetime.now() - self.last_timer_update) < timedelta(seconds=20)
211			):
212			return
213			self.logger.debug("calling get_current_race_time")
214			str_timer = ladder_client.get_current_race_time()
215			self.last_timer_update = datetime.now()
216			self.logger.info(f"str_timer= {str_timer}")
217			timer_sign = 1.0
218			if str_timer[0] == '-':
219			timer_sign = -1.0
220			str_timer = str_timer[1:]
221			timer = str_to_timer(str_timer)
222			timer = timer * timer_sign
223			if timer == timedelta(seconds=0):
224			self.started_at = None
225			else:
226			self.started_at = datetime.now(self.ladder_timezone()) - timer
227
228			async def update_stats(self):
229			if self.season_for_stats == -1:
230			self.season_for_stats = self.current_season.season_id
231			if self.mode_for_stats == -1:
232			self.mode_for_stats = self.get_mode_id_from_name(
233			self.next_race.Mode)
234			self.logger.info(f"season_for_stats: {self.season_for_stats}")
235			self.logger.info(f"mode_for_stats: {self.mode_for_stats}")
236			standings = ladder_client.get_standings(
237			self.season_for_stats, self.mode_for_stats)
238			for stats in standings:
239			if stats.RacerName.lower() == self.racer_name.lower():
240			self.stats = stats
241			return
242			# default standings
243			self.stats = Standings(
244			self.racer_name, self.current_season.SeasonName,
245			self.next_race.Mode, 1600, 0, 0, 0, 0, 0)
246
247			def get_mode_id_from_name(self, mode_name: str):
248			for flag in self.flags:
249			if flag.Mode == mode_name:
250			return flag.flag_id
251

ssbm-oro / racetime-obs

Push — master ( 6a2c9c...d2bb42 )

LadderTimer.update_ladder_status() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like