Build has been canceled!

10761719058

Committed 08 Sep 2024 05:13PM UTC coverage: 38.955% (+0.04%) from 38.92%

Build # 10761719058

Build Type

push

github

Committed by

ChristianTremblay

Commit Message

Merge branch 'async' of https://github.com/ChristianTremblay/BAC0 into async

Run Details

90 of 171 new or added lines in 18 files covered. (52.63%)

1008 existing lines in 18 files now uncovered.

2072 of 5319 relevant lines covered (38.95%)

0.39 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.78

/BAC0/tasks/RecurringTask.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 by Christian Tremblay, P.Eng <christian.tremblay@servisys.com>
# Licensed under LGPLv3, see file LICENSE in this source tree.
#
"""
RecurringTask.py - execute a recurring task
"""
import asyncio
from typing import Any, Callable, Tuple, Union, Coroutine
from concurrent.futures import ThreadPoolExecutor

from ..core.utils.notes import note_and_log
from .TaskManager import Task


@note_and_log
class RecurringTask(Task):
    """
    Start a recurring task (a function passed)
    """

    def __init__(
        self,
        fnc: Union[Tuple[Callable, Any], Callable, Coroutine],
        delay: int = 60,
        name: str = "recurring",
    ) -> None:
        """
        :param fnc: a function or a tuple (function, args)
        :param delay: (int) Delay between reads executions

        :returns: Nothing
        """
        self.fnc_args = None
        self.delay = delay
        if isinstance(fnc, tuple):
            self.func, self.fnc_args = fnc
        elif hasattr(fnc, "__call__"):
            self.func = fnc
        else:
            raise ValueError(
                "You must pass a function or a tuple (function,args) to this..."
            )
        Task.__init__(self, name=name, delay=delay)

    async def task(self) -> None:
        loop = asyncio.get_event_loop()
        executor = ThreadPoolExecutor()
        if self.fnc_args:
            if asyncio.iscoroutinefunction(self.func):
                await self.func(self.fnc_args)
            else:
                await loop.run_in_executor(executor, self.func, self.fnc_args)
                #self.func(self.fnc_args)
        else:
            if asyncio.iscoroutinefunction(self.func):
                await self.func()
            else:
                await loop.run_in_executor(executor, self.func)
                #self.func()
        await asyncio.sleep(self.delay)

1	#!/usr/bin/python
2	# -- coding: utf-8 --
3	#
4	# Copyright (C) 2015 by Christian Tremblay, P.Eng <christian.tremblay@servisys.com>
5	# Licensed under LGPLv3, see file LICENSE in this source tree.
6	#
7	"""
8	RecurringTask.py - execute a recurring task
9	"""
10	import asyncio	1✔
11	from typing import Any, Callable, Tuple, Union, Coroutine	1✔
12	from concurrent.futures import ThreadPoolExecutor	1✔
13
14	from ..core.utils.notes import note_and_log	1✔
15	from .TaskManager import Task	1✔
16
17
18	@note_and_log	1✔
19	class RecurringTask(Task):	1✔
20	"""
21	Start a recurring task (a function passed)
22	"""
23
24	def __init__(	1✔
25	self,
26	fnc: Union[Tuple[Callable, Any], Callable, Coroutine],
27	delay: int = 60,
28	name: str = "recurring",
29	) -> None:
30	"""
31	:param fnc: a function or a tuple (function, args)
32	:param delay: (int) Delay between reads executions
33
34	:returns: Nothing
35	"""
36	self.fnc_args = None	1✔
37	self.delay = delay	1✔
38	if isinstance(fnc, tuple):	1✔
UNCOV 39	self.func, self.fnc_args = fnc	×
40	elif hasattr(fnc, "__call__"):	1✔
41	self.func = fnc	1✔
42	else:
UNCOV 43	raise ValueError(	×
44	"You must pass a function or a tuple (function,args) to this..."
45	)
46	Task.__init__(self, name=name, delay=delay)	1✔
47
48	async def task(self) -> None:	1✔
49	loop = asyncio.get_event_loop()	1✔
50	executor = ThreadPoolExecutor()	1✔
51	if self.fnc_args:	1✔
52	if asyncio.iscoroutinefunction(self.func):	×
UNCOV 53	await self.func(self.fnc_args)	×
54	else:
UNCOV 55	await loop.run_in_executor(executor, self.func, self.fnc_args)	×
56	#self.func(self.fnc_args)
57	else:
58	if asyncio.iscoroutinefunction(self.func):	1✔
59	await self.func()	1✔
60	else:
UNCOV 61	await loop.run_in_executor(executor, self.func)	×
62	#self.func()
63	await asyncio.sleep(self.delay)	1✔

ChristianTremblay / BAC0 / 10761719058

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous