gadgets.media_player.MediaTrigger.__init__() - Code Metrics - Inspection of "Merge pull request #7 from ssbm-oro/ladder" - ssbm-oro/racetime-obs - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 80a6f5...e155f9 )

by manny

created 2021-02-25 16:35 UTC

gadgets.media_player.MediaTrigger.init() A

↳ Parent: gadgets.media_player

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	4
dl	0
loc	7
rs	10
c	0
b	0
f	0

import asyncio
from asyncio.locks import Lock
from datetime import datetime, timedelta
import logging
from threading import Timer
from typing import List, Optional
from asyncio import Event
from helpers import timer_to_str
from models.race import Entrant, Race
from models.chat_message import ChatMessage


class MediaTrigger:
    place: Optional[int] = None
    entrant_count_trigger: Optional[int] = None
    media_file_path: str = ""
    triggered: bool = False
    race_started_at: datetime = None

    def __init__(
        self, media_file_path: str, place_trigger: int = None,
        entrant_count_trigger: int = None
    ):
        self.media_file_path = media_file_path
        self.place = place_trigger
        self.entrant_count_trigger = entrant_count_trigger

    def check_trigger(self, race: Race, entrant: Entrant):
        if (
            self.triggered or self.media_file_path == "" or
            race is None or entrant is None
        ):
            return False
        else:
            if (
                self.check_finish_place_entrant_count(race, entrant) or
                self.check_finish_place(entrant)
            ):
                self.triggered = True
                return True
        return False

    def check_finish_place(self, entrant: Entrant):
        if (
            self.entrant_count_trigger is not None or entrant.user is None or
            entrant.place is None
        ):
            return False
        else:
            return (
                entrant.place <= self.place
            )

    def check_finish_place_entrant_count(self, race: Race, entrant: Entrant):
        if None in [
            self.entrant_count_trigger, entrant.user, entrant.place,
            race.entrants_count
        ]:
            return False
        else:
            return (
                entrant.place <= self.place and
                race.entrants_count <= self.entrant_count_trigger
            )


class ChatTrigger:
    media_file_path: str = ""
    highlight_trigger: bool = False
    bot_trigger: bool = False
    system_trigger: bool = False
    message_plain_trigger: str = ""
    trigger_once: bool = False
    triggered: bool = False

    def __init__(self, media_file_path, highlight, bot, system, message_plain):
        self.media_file_path = media_file_path
        self.highlight_trigger = highlight
        self.bot_trigger = bot
        self.system_trigger = system
        self.message_plain_trigger = message_plain

    def check_trigger(self, chat_message: ChatMessage):
        def trigger_on_bot(bot_trigger: bool, chat_message: ChatMessage):
            return bot_trigger and chat_message.is_bot

        def trigger_on_highlight(
                highlight_trigger: bool, chat_message: ChatMessage):
            return highlight_trigger and chat_message.highlight

        def trigger_on_system(system_trigger: bool, chat_message: ChatMessage):
            return system_trigger and chat_message.is_system

        if self.trigger_once and self.triggered:
            return False
        self.triggered = (
            trigger_on_bot(self.bot_trigger, chat_message) or
            trigger_on_highlight(self.highlight_trigger, chat_message) or
            trigger_on_system(self.system_trigger, chat_message)
        )
        return self.triggered


class MediaPlayer:
    logger: logging.Logger = logging.Logger("racetime-obs")
    enabled: bool = False
    triggers: List[MediaTrigger] = []
    chat_triggers: List[ChatTrigger] = []
    timers: List[Timer] = []
    triggers_lock: Lock = Lock()
    race_update_event: Event()
    play_media_callback = None
    started_at: datetime = None
    ping_chat_messages: bool = False
    chat_media_file: str = None
    last_session_race: str = ""
    monitoring_type: int = 0

    def race_updated(self, race: Race, entrant_name: str):
        # so the sound doesn't play when the user starts obs next time
        if self.last_session_race == race.name:
            return
        self.started_at = race.started_at
        for trigger in self.triggers:
            self.logger.debug(trigger)
            if trigger.check_trigger(
                race, race.get_entrant_by_name(entrant_name)
            ):
                self.play_media_callback(
                    trigger.media_file_path, self.monitoring_type)
                self.logger.debug("trigger fired")

    def chat_updated(self, message: ChatMessage):
        for trigger in self.chat_triggers:
            self.logger.debug(trigger)
            if trigger.check_trigger(message):
                self.play_media_callback(
                    trigger.media_file_path, self.monitoring_type)

    def add_trigger(
        self, media_file_path: str, place_trigger: int = None,
        entrant_count_trigger: int = None
    ):
        async def add(
            media_file_path: str, place_trigger: int = None,
            entrant_count_trigger: int = None
        ):
            async with self.triggers_lock:
                self.triggers.append(MediaTrigger(
                    media_file_path, place_trigger=place_trigger,
                    entrant_count_trigger=entrant_count_trigger
                ))

        asyncio.ensure_future(add(
                media_file_path, place_trigger,
                entrant_count_trigger
        ))

    def add_chat_trigger(
            self, media_file_path: str, highlight: bool = False,
            is_bot: bool = False, is_system: bool = False,
            message_plain_text: str = None
            ):
        async def add(
            media_file_path, highlight, is_bot, is_system, message_plain_text
        ):
            async with self.triggers_lock:
                self.chat_triggers.append(
                    ChatTrigger(media_file_path, highlight, is_bot, is_system,
                                message_plain_text))
            pass
        asyncio.ensure_future(add(
            media_file_path, highlight, is_bot, is_system, message_plain_text))

    def add_timer(self, media_file_path: str, race_time: timedelta):
        # try to wake up a little early and get ready
        timer = Timer(
            self.time_to_start_play(race_time),
            self.timer_wake_up, media_file_path, race_time
        )
        self.timers.append(timer)
        timer.start()

    def time_to_start_play(self, race_time: timedelta) -> float:
        time_to_start_play = 10000000000000.0
        return time_to_start_play

    async def timer_wake_up(self, media_file_path: str, race_time: timedelta):
        asyncio.sleep(self.time_to_start_play(race_time))
        self.logger.debug(
            f"attempting to play {media_file_path} at "
            f"{timer_to_str(race_time)}"
        )
        asyncio.ensure_future(
            self.play_media_callback(
                media_file_path, self.monitoring_type))

    def remove_trigger(self, index: int):
        async def remove(index: int):
            async with self.triggers_lock:
                if len(self.triggers) > index:
                    self.triggers.remove(self.triggers[index])
        asyncio.ensure_future(remove(index))


1			import asyncio
2			from asyncio.locks import Lock
3			from datetime import datetime, timedelta
4			import logging
5			from threading import Timer
6			from typing import List, Optional
7			from asyncio import Event
8			from helpers import timer_to_str
9			from models.race import Entrant, Race
10			from models.chat_message import ChatMessage
11
12
13			class MediaTrigger:
14			place: Optional[int] = None
15			entrant_count_trigger: Optional[int] = None
16			media_file_path: str = ""
17			triggered: bool = False
18			race_started_at: datetime = None
19
20			def __init__(
21			self, media_file_path: str, place_trigger: int = None,
22			entrant_count_trigger: int = None
23			):
24			self.media_file_path = media_file_path
25			self.place = place_trigger
26			self.entrant_count_trigger = entrant_count_trigger
27
28			def check_trigger(self, race: Race, entrant: Entrant):
29			if (
30			self.triggered or self.media_file_path == "" or
31			race is None or entrant is None
32			):
33			return False
34			else:
35			if (
36			self.check_finish_place_entrant_count(race, entrant) or
37			self.check_finish_place(entrant)
38			):
39			self.triggered = True
40			return True
41			return False
42
43			def check_finish_place(self, entrant: Entrant):
44			if (
45			self.entrant_count_trigger is not None or entrant.user is None or
46			entrant.place is None
47			):
48			return False
49			else:
50			return (
51			entrant.place <= self.place
52			)
53
54			def check_finish_place_entrant_count(self, race: Race, entrant: Entrant):
55			if None in [
56			self.entrant_count_trigger, entrant.user, entrant.place,
57			race.entrants_count
58			]:
59			return False
60			else:
61			return (
62			entrant.place <= self.place and
63			race.entrants_count <= self.entrant_count_trigger
64			)
65
66
67			class ChatTrigger:
68			media_file_path: str = ""
69			highlight_trigger: bool = False
70			bot_trigger: bool = False
71			system_trigger: bool = False
72			message_plain_trigger: str = ""
73			trigger_once: bool = False
74			triggered: bool = False
75
76			def __init__(self, media_file_path, highlight, bot, system, message_plain):
77			self.media_file_path = media_file_path
78			self.highlight_trigger = highlight
79			self.bot_trigger = bot
80			self.system_trigger = system
81			self.message_plain_trigger = message_plain
82
83			def check_trigger(self, chat_message: ChatMessage):
84			def trigger_on_bot(bot_trigger: bool, chat_message: ChatMessage):
85			return bot_trigger and chat_message.is_bot
86
87			def trigger_on_highlight(
88			highlight_trigger: bool, chat_message: ChatMessage):
89			return highlight_trigger and chat_message.highlight
90
91			def trigger_on_system(system_trigger: bool, chat_message: ChatMessage):
92			return system_trigger and chat_message.is_system
93
94			if self.trigger_once and self.triggered:
95			return False
96			self.triggered = (
97			trigger_on_bot(self.bot_trigger, chat_message) or
98			trigger_on_highlight(self.highlight_trigger, chat_message) or
99			trigger_on_system(self.system_trigger, chat_message)
100			)
101			return self.triggered
102
103
104			class MediaPlayer:
105			logger: logging.Logger = logging.Logger("racetime-obs")
106			enabled: bool = False
107			triggers: List[MediaTrigger] = []
108			chat_triggers: List[ChatTrigger] = []
109			timers: List[Timer] = []
110			triggers_lock: Lock = Lock()
111			race_update_event: Event()
112			play_media_callback = None
113			started_at: datetime = None
114			ping_chat_messages: bool = False
115			chat_media_file: str = None
116			last_session_race: str = ""
117			monitoring_type: int = 0
118
119			def race_updated(self, race: Race, entrant_name: str):
120			# so the sound doesn't play when the user starts obs next time
121			if self.last_session_race == race.name:
122			return
123			self.started_at = race.started_at
124			for trigger in self.triggers:
125			self.logger.debug(trigger)
126			if trigger.check_trigger(
127			race, race.get_entrant_by_name(entrant_name)
128			):
129			self.play_media_callback(
130			trigger.media_file_path, self.monitoring_type)
131			self.logger.debug("trigger fired")
132
133			def chat_updated(self, message: ChatMessage):
134			for trigger in self.chat_triggers:
135			self.logger.debug(trigger)
136			if trigger.check_trigger(message):
137			self.play_media_callback(
138			trigger.media_file_path, self.monitoring_type)
139
140			def add_trigger(
141			self, media_file_path: str, place_trigger: int = None,
142			entrant_count_trigger: int = None
143			):
144			async def add(
145			media_file_path: str, place_trigger: int = None,
146			entrant_count_trigger: int = None
147			):
148			async with self.triggers_lock:
149			self.triggers.append(MediaTrigger(
150			media_file_path, place_trigger=place_trigger,
151			entrant_count_trigger=entrant_count_trigger
152			))
153
154			asyncio.ensure_future(add(
155			media_file_path, place_trigger,
156			entrant_count_trigger
157			))
158
159			def add_chat_trigger(
160			self, media_file_path: str, highlight: bool = False,
161			is_bot: bool = False, is_system: bool = False,
162			message_plain_text: str = None
163			):
164			async def add(
165			media_file_path, highlight, is_bot, is_system, message_plain_text
166			):
167			async with self.triggers_lock:
168			self.chat_triggers.append(
169			ChatTrigger(media_file_path, highlight, is_bot, is_system,
170			message_plain_text))
171			pass
172			asyncio.ensure_future(add(
173			media_file_path, highlight, is_bot, is_system, message_plain_text))
174
175			def add_timer(self, media_file_path: str, race_time: timedelta):
176			# try to wake up a little early and get ready
177			timer = Timer(
178			self.time_to_start_play(race_time),
179			self.timer_wake_up, media_file_path, race_time
180			)
181			self.timers.append(timer)
182			timer.start()
183
184			def time_to_start_play(self, race_time: timedelta) -> float:
185			time_to_start_play = 10000000000000.0
186			return time_to_start_play
187
188			async def timer_wake_up(self, media_file_path: str, race_time: timedelta):
189			asyncio.sleep(self.time_to_start_play(race_time))
190			self.logger.debug(
191			f"attempting to play {media_file_path} at "
192			f"{timer_to_str(race_time)}"
193			)
194			asyncio.ensure_future(
195			self.play_media_callback(
196			media_file_path, self.monitoring_type))
197
198			def remove_trigger(self, index: int):
199			async def remove(index: int):
200			async with self.triggers_lock:
201			if len(self.triggers) > index:
202			self.triggers.remove(self.triggers[index])
203			asyncio.ensure_future(remove(index))
204

ssbm-oro / racetime-obs

Push — master ( 80a6f5...e155f9 )

gadgets.media_player.MediaTrigger.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

gadgets.media_player.MediaTrigger.init() A