gadgets.ladder_timer.LadderTimer.get_timer_text() - Code Metrics - ssbm-oro/racetime-obs - Measure and Improve Code Quality continuously with Scrutinizer

gadgets.ladder_timer.LadderTimer.get_timer_text() B
last analyzed 2021-11-05 18:13 UTC

↳ Parent: gadgets.ladder_timer

Complexity

Conditions

Size

Total Lines	29
Code Lines	26

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	8
eloc	26
nop	1
dl	0
loc	29
rs	7.3333
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
        self.next_race = None
        self.schedule = 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			self.next_race = None
89			self.schedule = None
90			await self.update_active_racers()
91			await self.update_flags()
92			await self.update_season()
93			await self.update_schedule(self.current_season.season_id)
94			await self.update_timer()
95			await self.update_stats()
96
97			async def check_racer_finish(self):
98			if ladder_client.is_currently_racing(self.racer_id):
99			return
100			racer_history = ladder_client.get_racer_history(self.racer_id)
101			if racer_history[-1].race_id == self.next_race.race_id:
102			if racer_history[-1].FinishTime == "FF":
103			self.result = "FF"
104			self.finish_time = timedelta(seconds=0)
105			self.logger.debug("racer forfeited")
106			else:
107			self.finish_time = str_to_timer(racer_history[-1].FinishTime)
108			self.result = racer_history[-1].Result
109
110			def get_timer_text(self):
111			if self.finish_time:
112			if self.result == "W":
113			return (
114			self.winner_color,
115			timer_to_str(self.finish_time, self.decimals)
116			)
117			elif self.result == "L":
118			return (
119			self.loser_color,
120			timer_to_str(self.finish_time, self.decimals)
121			)
122			elif self.result == "FF":
123			return self.ff_color, "-:--:--.-"
124			current_timer = timedelta(seconds=0)
125			color = self.pre_color
126			now = datetime.now(self.ladder_timezone())
127			if self.started_at is None:
128			current_timer = now - self.next_race.StartTime
129			if (abs(current_timer.total_seconds()) < 600.0):
130			loop = asyncio.new_event_loop()
131			loop.run_until_complete(self.update_timer())
132			loop.close()
133			color = self.pre_color
134			else:
135			current_timer = now - self.started_at
136			if current_timer.total_seconds() > 0:
137			color = self.racing_color
138			return color, timer_to_str(current_timer, self.decimals)
139
140			def get_stats(self):
141			stats = ""
142			if self.enabled:
143			if self.show_season_name:
144			stats = stats + self.stats.Season + " "
145			if self.show_mode_name:
146			stats = stats + self.stats.Mode + " "
147			if self.show_rank:
148			stats = stats + "#" + str(self.stats.Rank) + " "
149			if self.show_rating:
150			stats = stats + str(self.stats.Rating) + " "
151			if self.show_change:
152			stats = stats + str(self.stats.Change) + " "
153			if self.show_win_loss_tie:
154			stats = (
155			stats + str(self.stats.Wins) + "/" +
156			str(self.stats.Losses) + "/" + str(self.stats.Ties)
157			)
158			return stats
159
160			def update_settings(self, racer_name: str):
161			if racer_name is None or racer_name == "":
162			return False
163			racer_id = self.get_racer_id(racer_name)
164			if racer_id == 0:
165			return False
166			else:
167			self.racer_id = racer_id
168			self.racer_name = racer_name
169			loop = asyncio.new_event_loop()
170			loop.run_until_complete(self.update_stats())
171			loop.close()
172			return True
173
174			def get_racer_id(self, racer_name: str) -> Racer:
175			loop = asyncio.new_event_loop()
176			loop.run_until_complete(self.update_active_racers())
177			loop.close()
178			for racer in self.active_racers:
179			if racer.RacerName.lower() == racer_name.lower():
180			return racer.racer_id
181			return 0
182
183			async def update_active_racers(self):
184			if self.active_racers is None or self.active_racers == []:
185			self.active_racers = ladder_client.get_active_racers()
186
187			async def update_season(self):
188			if self.current_season is None:
189			self.all_seasons = ladder_client.get_seasons()
190			if self.all_seasons is not None:
191			self.current_season = (
192			next(filter(
193			lambda x: x.IsCurrentSeason, self.all_seasons
194			), None)
195			)
196			self.logger.info(f"current_season = {self.current_season}")
197
198			async def update_flags(self):
199			self.flags = ladder_client.get_flags()
200
201			async def update_schedule(self, season_id: int):
202			if self.schedule is None or self.schedule == []:
203			self.schedule = ladder_client.get_schedule(season_id)
204			self.next_race = (
205			next(x for x in self.schedule if not x.HasCompleted)
206			)
207			self.logger.info(f"next_race = {self.next_race}")
208
209			async def update_timer(self):
210			if (
211			self.last_timer_update is not None and
212			(datetime.now() - self.last_timer_update) < timedelta(seconds=20)
213			):
214			return
215			self.logger.debug("calling get_current_race_time")
216			str_timer = ladder_client.get_current_race_time()
217			self.last_timer_update = datetime.now()
218			self.logger.info(f"str_timer= {str_timer}")
219			timer_sign = 1.0
220			if str_timer[0] == '-':
221			timer_sign = -1.0
222			str_timer = str_timer[1:]
223			timer = str_to_timer(str_timer)
224			timer = timer * timer_sign
225			if timer == timedelta(seconds=0):
226			self.started_at = None
227			else:
228			self.started_at = datetime.now(self.ladder_timezone()) - timer
229
230			async def update_stats(self):
231			if self.season_for_stats == -1:
232			self.season_for_stats = self.current_season.season_id
233			if self.mode_for_stats == -1:
234			self.mode_for_stats = self.get_mode_id_from_name(
235			self.next_race.Mode)
236			self.logger.info(f"season_for_stats: {self.season_for_stats}")
237			self.logger.info(f"mode_for_stats: {self.mode_for_stats}")
238			standings = ladder_client.get_standings(
239			self.season_for_stats, self.mode_for_stats)
240			for stats in standings:
241			if stats.RacerName.lower() == self.racer_name.lower():
242			self.stats = stats
243			return
244			# default standings
245			self.stats = Standings(
246			self.racer_name, self.current_season.SeasonName,
247			self.next_race.Mode, 1600, 0, 0, 0, 0, 0)
248
249			def get_mode_id_from_name(self, mode_name: str):
250			for flag in self.flags:
251			if flag.Mode == mode_name:
252			return flag.flag_id
253

ssbm-oro / racetime-obs

gadgets.ladder_timer.LadderTimer.get_timer_text() B last analyzed 2021-11-05 18:13 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

gadgets.ladder_timer.LadderTimer.get_timer_text() B
last analyzed 2021-11-05 18:13 UTC