gadgets.ladder_timer.LadderTimer.check_racer_finish() - Code Metrics - Inspection of "Merge ladder branch into master for Season 6" - ssbm-oro/racetime-obs - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#7)

by manny

created 2021-02-25 16:29 UTC

LadderTimer.check_racer_finish() A

↳ Parent: gadgets.ladder_timer

Complexity

Conditions

Size

Total Lines	10
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	9
nop	1
dl	0
loc	10
rs	9.95
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

    @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_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()
        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 check_racer_finish(self):
        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)
            elif self.result == "L":
                return self.loser_color, timer_to_str(self.finish_time)
            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)

    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
52			@staticmethod
53			def ladder_timezone():
54			return pytz.timezone('US/Eastern')
55
56			def __init__(self, logger: logging.Logger, racer_id: int = None):
57			self.logger = logger
58			if racer_id is not None:
59			self.racer_id = racer_id
60
61			ladder_update_t = Thread(target=self.ladder_update_thread)
62			ladder_update_t.daemon = True
63			ladder_update_t.start()
64
65			def ladder_update_thread(self):
66			self.logger.debug("starting ladder race update")
67			loop = asyncio.new_event_loop()
68			loop.run_until_complete(self.race_updater())
69			loop.run_forever()
70
71			async def race_updater(self):
72			await self.update_active_racers()
73			await self.update_flags()
74			await self.update_season()
75			await self.update_schedule(self.current_season.season_id)
76			await self.update_timer()
77			await self.update_stats()
78			while True:
79			if not self.enabled or self.finish_time is not None:
80			await asyncio.sleep(5.0)
81			else:
82			now = datetime.now(self.ladder_timezone())
83			if self.started_at is not None and now > self.started_at:
84			self.logger.debug("checking if racer finished")
85			await self.check_racer_finish()
86			await asyncio.sleep(5.0)
87
88			async def check_racer_finish(self):
89			racer_history = ladder_client.get_racer_history(self.racer_id)
90			if racer_history[-1].race_id == self.next_race.race_id:
91			if racer_history[-1].FinishTime == "FF":
92			self.result = "FF"
93			self.finish_time = timedelta(seconds=0)
94			self.logger.debug("racer forfeited")
95			else:
96			self.finish_time = str_to_timer(racer_history[-1].FinishTime)
97			self.result = racer_history[-1].Result
98
99			def get_timer_text(self):
100			if self.finish_time:
101			if self.result == "W":
102			return self.winner_color, timer_to_str(self.finish_time)
103			elif self.result == "L":
104			return self.loser_color, timer_to_str(self.finish_time)
105			elif self.result == "FF":
106			return self.ff_color, "-:--:--.-"
107			current_timer = timedelta(seconds=0)
108			color = self.pre_color
109			now = datetime.now(self.ladder_timezone())
110			if self.started_at is None:
111			current_timer = now - self.next_race.StartTime
112			if (abs(current_timer.total_seconds()) < 600.0):
113			loop = asyncio.new_event_loop()
114			loop.run_until_complete(self.update_timer())
115			loop.close()
116			color = self.pre_color
117			else:
118			current_timer = now - self.started_at
119			if current_timer.total_seconds() > 0:
120			color = self.racing_color
121			return color, timer_to_str(current_timer)
122
123			def get_stats(self):
124			stats = ""
125			if self.enabled:
126			if self.show_season_name:
127			stats = stats + self.stats.Season + " "
128			if self.show_mode_name:
129			stats = stats + self.stats.Mode + " "
130			if self.show_rank:
131			stats = stats + "#" + str(self.stats.Rank) + " "
132			if self.show_rating:
133			stats = stats + str(self.stats.Rating) + " "
134			if self.show_change:
135			stats = stats + str(self.stats.Change) + " "
136			if self.show_win_loss_tie:
137			stats = (
138			stats + str(self.stats.Wins) + "/" +
139			str(self.stats.Losses) + "/" + str(self.stats.Ties)
140			)
141			return stats
142
143			def update_settings(self, racer_name: str):
144			if racer_name is None or racer_name == "":
145			return False
146			racer_id = self.get_racer_id(racer_name)
147			if racer_id == 0:
148			return False
149			else:
150			self.racer_id = racer_id
151			self.racer_name = racer_name
152			loop = asyncio.new_event_loop()
153			loop.run_until_complete(self.update_stats())
154			loop.close()
155			return True
156
157			def get_racer_id(self, racer_name: str) -> Racer:
158			loop = asyncio.new_event_loop()
159			loop.run_until_complete(self.update_active_racers())
160			loop.close()
161			for racer in self.active_racers:
162			if racer.RacerName.lower() == racer_name.lower():
163			return racer.racer_id
164			return 0
165
166			async def update_active_racers(self):
167			if self.active_racers is None or self.active_racers == []:
168			self.active_racers = ladder_client.get_active_racers()
169
170			async def update_season(self):
171			if self.current_season is None:
172			self.all_seasons = ladder_client.get_seasons()
173			if self.all_seasons is not None:
174			self.current_season = (
175			next(filter(
176			lambda x: x.IsCurrentSeason, self.all_seasons
177			), None)
178			)
179			self.logger.info(f"current_season = {self.current_season}")
180
181			async def update_flags(self):
182			self.flags = ladder_client.get_flags()
183
184			async def update_schedule(self, season_id: int):
185			if self.schedule is None or self.schedule == []:
186			self.schedule = ladder_client.get_schedule(season_id)
187			self.next_race = (
188			next(x for x in self.schedule if not x.HasCompleted)
189			)
190			self.logger.info(f"next_race = {self.next_race}")
191
192			async def update_timer(self):
193			if (
194			self.last_timer_update is not None and
195			(datetime.now() - self.last_timer_update) < timedelta(seconds=20)
196			):
197			return
198			self.logger.debug("calling get_current_race_time")
199			str_timer = ladder_client.get_current_race_time()
200			self.last_timer_update = datetime.now()
201			self.logger.info(f"str_timer= {str_timer}")
202			timer_sign = 1.0
203			if str_timer[0] == '-':
204			timer_sign = -1.0
205			str_timer = str_timer[1:]
206			timer = str_to_timer(str_timer)
207			timer = timer * timer_sign
208			if timer == timedelta(seconds=0):
209			self.started_at = None
210			else:
211			self.started_at = datetime.now(self.ladder_timezone()) - timer
212
213			async def update_stats(self):
214			if self.season_for_stats == -1:
215			self.season_for_stats = self.current_season.season_id
216			if self.mode_for_stats == -1:
217			self.mode_for_stats = self.get_mode_id_from_name(
218			self.next_race.Mode)
219			self.logger.info(f"season_for_stats: {self.season_for_stats}")
220			self.logger.info(f"mode_for_stats: {self.mode_for_stats}")
221			standings = ladder_client.get_standings(
222			self.season_for_stats, self.mode_for_stats)
223			for stats in standings:
224			if stats.RacerName.lower() == self.racer_name.lower():
225			self.stats = stats
226			return
227			# default standings
228			self.stats = Standings(
229			self.racer_name, self.current_season.SeasonName,
230			self.next_race.Mode, 1600, 0, 0, 0, 0, 0)
231
232			def get_mode_id_from_name(self, mode_name: str):
233			for flag in self.flags:
234			if flag.Mode == mode_name:
235			return flag.flag_id
236

ssbm-oro / racetime-obs

Pull Request — master (#7)

LadderTimer.check_racer_finish() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like