pincer.utils.event_mgr._Event.wait() - Code Metrics - Inspection of "Merge pull request #176 from Lunarmagpie/main" - Pincer-org/Pincer - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — main ( 20c668...13c337 )

by Yohann

created 2021-11-16 21:56 UTC

pincer.utils.event_mgr._Event.wait() A

↳ Parent: pincer.utils.event_mgr

Complexity

Conditions

Size

Total Lines	5
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# Copyright Pincer 2021-Present

# Full MIT License can be found in `LICENSE` at the project root.

from __future__ import annotations

from abc import ABC, abstractmethod
from asyncio import Event, wait_for as _wait_for, get_running_loop, TimeoutError

from collections import deque
from typing import TYPE_CHECKING

from ..exceptions import TimeoutError as PincerTimeoutError

if TYPE_CHECKING:
    from typing import Any, List, Union
    from .types import CheckFunction


class _Processable(ABC):

    @abstractmethod
    def process(self, event_name: str, *args):
        """
        Method that is ran when an event is recieved from discord.

        Parameters
        ----------
        event_name : str
            The name of the event.
        *args : Any
            Arguments to evaluate check with.

        Returns
        -------
        bool
            Whether the event can be set
        """

    def matches_event(self, event_name: str, *args):
        """
        Parameters
        ----------
        event_name : str
            Name of event.
        *args : Any
            Arguments to evalue check with.
        """
        if self.event_name != event_name:

            return False

        if self.check:

            return self.check(*args)


        return True


def _lowest_value(*args):
    """
    Returns lowest value from list of numbers. ``None`` is not counted as a
    value. ``None`` is returned if all arguments are ``None``.
    """
    args_without_none = [n for n in args if n is not None]

    if len(args_without_none) == 0:
        return None

    return min(args_without_none)


class _Event(_Processable):
    """
    Parameters
    ----------
    event_name : str
        The name of the event.
    check : Optional[Callable[[Any], bool]]
        ``can_be_set`` only returns true if this function returns true.
        Will be ignored if set to None.

    Attributes
    ----------
    event : :class:`asyncio.Event`
        Even that is used to wait until the next valid discord event.
    return_value : Optional[str]
        Used to store the arguments from ``can_be_set`` so they can be
        returned later.
    """

    def __init__(
        self,

        event_name: str,

        check: CheckFunction

    ):
        self.event_name = event_name
        self.check = check
        self.event = Event()
        self.return_value = None
        super().__init__()

    async def wait(self):
        """
        Waits until ``self.event`` is set.
        """
        await self.event.wait()

    def process(self, event_name: str, *args) -> bool:
        if self.matches_event(event_name, *args):
            self.return_value = args
            self.event.set()


class _LoopEmptyError(Exception):
    "Raised when the _LoopMgr is empty and cannot accept new item"


class _LoopMgr(_Processable):
    """
    Parameters
    ----------
    event_name : str
        The name of the event.
    check : Optional[Callable[[Any], bool]]
        ``can_be_set`` only returns true if this function returns true.
        Will be ignored if set to None.

    Attributes
    ----------
    can_expand : bool
        Whether the queue is allowed to grow. Turned to false once the
        EventMgr's timer runs out.
    events : :class:`collections.deque`
        Qeue of events to be processed.
    wait : :class:`asyncio.Event`
        Used to make ``get_next()` wait for the next event.
    """

    def __init__(self, event_name: str, check: CheckFunction) -> None:
        self.event_name = event_name
        self.check = check

        self.can_expand = True
        self.events = deque()
        self.wait = Event()

    def process(self, event_name: str, *args):
        if not self.can_expand:
            return

        if self.matches_event(event_name, *args):
            self.events.append(args)
            self.wait.set()

    async def get_next(self):
        """
        Returns the next item if the queue. If there are no items in the queue,
        it will return the next event that happens.
        """
        if not self.events:

            if not self.can_expand:
                raise _LoopEmptyError

            self.wait.clear()
            await self.wait.wait()
            return self.events.popleft()
        else:
            return self.events.popleft()


class EventMgr:
    """
    Attributes
    ----------
    event_list : List[_DiscordEvent]
        The List of events that need to be processed.
    """

    def __init__(self):
        self.event_list: List[_Processable] = []

    def process_events(self, event_name, *args):
        """
        Parameters
        ----------
        event_name : str
            The name of the event to be processed.
        *args : Any
            The arguments returned from the middleware for this event.
        """
        for event in self.event_list:
            event.process(event_name, *args)

    async def wait_for(
        self,

        event_name: str,

        check: CheckFunction,

        timeout: Union[float, None]

    ) -> Any:
        """
        Parameters
        ----------
        event_name : str
            The type of event. It should start with `on_`. This is the same
            name that is used for @Client.event.
        check : Union[Callable[[Any], bool], None]
            This function only returns a value if this return true.
        timeout: Union[float, None]
            Amount of seconds before timeout. Use None for no timeout.

        Returns
        ------
        Any
            What the Discord API returns for this event.
        """

        event = _Event(event_name, check)
        self.event_list.append(event)

        try:
            await _wait_for(event.wait(), timeout=timeout)
        except TimeoutError:
            raise PincerTimeoutError(
                "wait_for() timed out while waiting for an event."
            )
        self.event_list.remove(event)
        return event.return_value

    async def loop_for(
        self,

        event_name: str,

        check: CheckFunction,

        iteration_timeout: Union[float, None],

        loop_timeout: Union[float, None],

    ) -> Any:
        """
        Parameters
        ----------
        event_name : str
            The type of event. It should start with `on_`. This is the same
            name that is used for @Client.event.
        check : Callable[[Any], bool]
            This function only returns a value if this return true.
        iteration_timeout: Union[float, None]
            Amount of seconds before timeout. Timeouts are for each loop.
        loop_timeout: Union[float, None]
            Amount of seconds before the entire loop times out. The generator
            will only raise a timeout error while it is waiting for an event.

        Yields
        ------
        Any
            What the Discord API returns for this event.
        """

        loop_mgr = _LoopMgr(event_name, check)
        self.event_list.append(loop_mgr)

        loop = get_running_loop()

        while True:
            start_time = loop.time()

            try:
                yield await _wait_for(
                    loop_mgr.get_next(),
                    timeout=_lowest_value(
                        loop_timeout, iteration_timeout
                    )
                )

            except TimeoutError:
                # Loop timed out. Loop through the remaining events recieved
                # before the timeout.
                loop_mgr.can_expand = False
                try:
                    while True:
                        yield await loop_mgr.get_next()
                except _LoopEmptyError:
                    raise PincerTimeoutError(
                        "loop_for() timed out while waiting for an event"
                    )

            # `not` can't be used here because there is a check for
            # `loop_timeout == 0`
            if loop_timeout is not None:
                loop_timeout -= loop.time() - start_time

                # loop_timeout can be below 0 if the user's code in the for loop
                # takes longer than the time left in loop_timeout
                if loop_timeout <= 0:
                    raise PincerTimeoutError(
                        "loop_for() timed out while waiting for an event"
                    )


1			# Copyright Pincer 2021-Present
			0 ignored issues – show introduced 2021-08-29 14:15 UTC by Report Bug Copy Issue Report Missing module docstring Loading history...
2			# Full MIT License can be found in `LICENSE` at the project root.
3
4			from __future__ import annotations
5
6			from abc import ABC, abstractmethod
7			from asyncio import Event, wait_for as _wait_for, get_running_loop, TimeoutError
			0 ignored issues – show Bug Best Practice introduced 2021-11-12 00:48 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `TimeoutError`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
8			from collections import deque
9			from typing import TYPE_CHECKING
10
11			from ..exceptions import TimeoutError as PincerTimeoutError
12
13			if TYPE_CHECKING:
14			from typing import Any, List, Union
15			from .types import CheckFunction
16
17
18			class _Processable(ABC):
19
20			@abstractmethod
21			def process(self, event_name: str, *args):
22			"""
23			Method that is ran when an event is recieved from discord.
24
25			Parameters
26			----------
27			event_name : str
28			The name of the event.
29			*args : Any
30			Arguments to evaluate check with.
31
32			Returns
33			-------
34			bool
35			Whether the event can be set
36			"""
37
38			def matches_event(self, event_name: str, *args):
39			"""
40			Parameters
41			----------
42			event_name : str
43			Name of event.
44			*args : Any
45			Arguments to evalue check with.
46			"""
47			if self.event_name != event_name:
			0 ignored issues – show Bug introduced 2021-11-12 00:48 UTC by Report Bug Copy Issue Report The Instance of `_Processable` does not seem to have a member named `event_name`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
48			return False
49
50			if self.check:
			0 ignored issues – show Bug introduced 2021-11-12 00:48 UTC by Report Bug Copy Issue Report The Instance of `_Processable` does not seem to have a member named `check`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
51			return self.check(*args)
			0 ignored issues – show Bug introduced 2021-11-12 00:48 UTC by Report Bug Copy Issue Report The Instance of `_Processable` does not seem to have a member named `check`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
52
53			return True
54
55
56			def _lowest_value(*args):
57			"""
58			Returns lowest value from list of numbers. ``None`` is not counted as a
59			value. ``None`` is returned if all arguments are ``None``.
60			"""
61			args_without_none = [n for n in args if n is not None]
62
63			if len(args_without_none) == 0:
64			return None
65
66			return min(args_without_none)
67
68
69			class _Event(_Processable):
70			"""
71			Parameters
72			----------
73			event_name : str
74			The name of the event.
75			check : Optional[Callable[[Any], bool]]
76			``can_be_set`` only returns true if this function returns true.
77			Will be ignored if set to None.
78
79			Attributes
80			----------
81			event : :class:`asyncio.Event`
82			Even that is used to wait until the next valid discord event.
83			return_value : Optional[str]
84			Used to store the arguments from ``can_be_set`` so they can be
85			returned later.
86			"""
87
88			def __init__(
89			self,
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
90			event_name: str,
			0 ignored issues – show Coding Style introduced 2021-11-06 00:55 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
91			check: CheckFunction
			0 ignored issues – show Coding Style introduced 2021-11-06 00:55 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
92			):
93			self.event_name = event_name
94			self.check = check
95			self.event = Event()
96			self.return_value = None
97			super().__init__()
98
99			async def wait(self):
100			"""
101			Waits until ``self.event`` is set.
102			"""
103			await self.event.wait()
104
105			def process(self, event_name: str, *args) -> bool:
106			if self.matches_event(event_name, *args):
107			self.return_value = args
108			self.event.set()
109
110
111			class _LoopEmptyError(Exception):
112			"Raised when the _LoopMgr is empty and cannot accept new item"
113
114
115			class _LoopMgr(_Processable):
116			"""
117			Parameters
118			----------
119			event_name : str
120			The name of the event.
121			check : Optional[Callable[[Any], bool]]
122			``can_be_set`` only returns true if this function returns true.
123			Will be ignored if set to None.
124
125			Attributes
126			----------
127			can_expand : bool
128			Whether the queue is allowed to grow. Turned to false once the
129			EventMgr's timer runs out.
130			events : :class:`collections.deque`
131			Qeue of events to be processed.
132			wait : :class:`asyncio.Event`
133			Used to make ``get_next()` wait for the next event.
134			"""
135
136			def __init__(self, event_name: str, check: CheckFunction) -> None:
137			self.event_name = event_name
138			self.check = check
139
140			self.can_expand = True
141			self.events = deque()
142			self.wait = Event()
143
144			def process(self, event_name: str, *args):
145			if not self.can_expand:
146			return
147
148			if self.matches_event(event_name, *args):
149			self.events.append(args)
150			self.wait.set()
151
152			async def get_next(self):
153			"""
154			Returns the next item if the queue. If there are no items in the queue,
155			it will return the next event that happens.
156			"""
157			if not self.events:
			0 ignored issues – show unused-code introduced 2021-11-14 22:56 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
158			if not self.can_expand:
159			raise _LoopEmptyError
160
161			self.wait.clear()
162			await self.wait.wait()
163			return self.events.popleft()
164			else:
165			return self.events.popleft()
166
167
168			class EventMgr:
169			"""
170			Attributes
171			----------
172			event_list : List[_DiscordEvent]
173			The List of events that need to be processed.
174			"""
175
176			def __init__(self):
177			self.event_list: List[_Processable] = []
178
179			def process_events(self, event_name, *args):
180			"""
181			Parameters
182			----------
183			event_name : str
184			The name of the event to be processed.
185			*args : Any
186			The arguments returned from the middleware for this event.
187			"""
188			for event in self.event_list:
189			event.process(event_name, *args)
190
191			async def wait_for(
192			self,
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
193			event_name: str,
			0 ignored issues – show Coding Style introduced 2021-11-06 00:55 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
194			check: CheckFunction,
			0 ignored issues – show Coding Style introduced 2021-11-06 00:55 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
195			timeout: Union[float, None]
			0 ignored issues – show Coding Style introduced 2021-11-06 00:55 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
196			) -> Any:
197			"""
198			Parameters
199			----------
200			event_name : str
201			The type of event. It should start with `on_`. This is the same
202			name that is used for @Client.event.
203			check : Union[Callable[[Any], bool], None]
204			This function only returns a value if this return true.
205			timeout: Union[float, None]
206			Amount of seconds before timeout. Use None for no timeout.
207
208			Returns
209			------
210			Any
211			What the Discord API returns for this event.
212			"""
213
214			event = _Event(event_name, check)
215			self.event_list.append(event)
216
217			try:
218			await _wait_for(event.wait(), timeout=timeout)
219			except TimeoutError:
220			raise PincerTimeoutError(
221			"wait_for() timed out while waiting for an event."
222			)
223			self.event_list.remove(event)
224			return event.return_value
225
226			async def loop_for(
227			self,
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
228			event_name: str,
			0 ignored issues – show Coding Style introduced 2021-11-14 22:56 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
229			check: CheckFunction,
			0 ignored issues – show Coding Style introduced 2021-11-14 22:56 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
230			iteration_timeout: Union[float, None],
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
231			loop_timeout: Union[float, None],
			0 ignored issues – show Coding Style introduced 2021-11-04 02:21 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
232			) -> Any:
233			"""
234			Parameters
235			----------
236			event_name : str
237			The type of event. It should start with `on_`. This is the same
238			name that is used for @Client.event.
239			check : Callable[[Any], bool]
240			This function only returns a value if this return true.
241			iteration_timeout: Union[float, None]
242			Amount of seconds before timeout. Timeouts are for each loop.
243			loop_timeout: Union[float, None]
244			Amount of seconds before the entire loop times out. The generator
245			will only raise a timeout error while it is waiting for an event.
246
247			Yields
248			------
249			Any
250			What the Discord API returns for this event.
251			"""
252
253			loop_mgr = _LoopMgr(event_name, check)
254			self.event_list.append(loop_mgr)
255
256			loop = get_running_loop()
257
258			while True:
259			start_time = loop.time()
260
261			try:
262			yield await _wait_for(
263			loop_mgr.get_next(),
264			timeout=_lowest_value(
265			loop_timeout, iteration_timeout
266			)
267			)
268
269			except TimeoutError:
270			# Loop timed out. Loop through the remaining events recieved
271			# before the timeout.
272			loop_mgr.can_expand = False
273			try:
274			while True:
275			yield await loop_mgr.get_next()
276			except _LoopEmptyError:
277			raise PincerTimeoutError(
278			"loop_for() timed out while waiting for an event"
279			)
280
281			# `not` can't be used here because there is a check for
282			# `loop_timeout == 0`
283			if loop_timeout is not None:
284			loop_timeout -= loop.time() - start_time
285
286			# loop_timeout can be below 0 if the user's code in the for loop
287			# takes longer than the time left in loop_timeout
288			if loop_timeout <= 0:
289			raise PincerTimeoutError(
290			"loop_for() timed out while waiting for an event"
291			)
292

Pincer-org / Pincer

Push — main ( 20c668...13c337 )

pincer.utils.event_mgr._Event.wait() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like