17009917031

Committed 16 Aug 2025 03:18PM UTC coverage: 40.7% (+0.02%) from 40.683%

Build # 17009917031

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #563 from ChristianTremblay/develop

Bug fixes

Run Details

15 of 28 new or added lines in 3 files covered. (53.57%)

1 existing line in 1 file now uncovered.

2234 of 5489 relevant lines covered (40.7%)

0.41 hits per line

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

67.44

/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
import functools
import inspect
from typing import Any, Callable, Coroutine, Tuple, Union

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:
        # Prefer awaiting async callables; offload sync callables with to_thread.
        def _is_async_callable(fn: Callable) -> bool:
            if inspect.iscoroutinefunction(fn):
                return True
            call = getattr(fn, "__call__", None)
            if call and inspect.iscoroutinefunction(call):
                return True
            if isinstance(fn, functools.partial):
                return _is_async_callable(fn.func)
            return False

        if self.fnc_args is not None:
            if _is_async_callable(self.func):
                await self.func(self.fnc_args)
            else:
                try:
                    await asyncio.to_thread(self.func, self.fnc_args)
                except (RuntimeError, AttributeError):
                    # Fallback for environments without to_thread or loop quirks
                    loop = asyncio.get_running_loop()
                    await loop.run_in_executor(None, self.func, self.fnc_args)
        else:
            if _is_async_callable(self.func):
                await self.func()
            else:
                try:
                    await asyncio.to_thread(self.func)
                except (RuntimeError, AttributeError):
                    loop = asyncio.get_running_loop()
                    await loop.run_in_executor(None, 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	import functools	1✔
12	import inspect	1✔
13	from typing import Any, Callable, Coroutine, Tuple, Union	1✔
14
15	from ..core.utils.notes import note_and_log	1✔
16	from .TaskManager import Task	1✔
17
18
19	@note_and_log	1✔
20	class RecurringTask(Task):	1✔
21	"""
22	Start a recurring task (a function passed)
23	"""
24
25	def __init__(	1✔
26	self,
27	fnc: Union[Tuple[Callable, Any], Callable, Coroutine],
28	delay: int = 60,
29	name: str = "recurring",
30	) -> None:
31	"""
32	:param fnc: a function or a tuple (function, args)
33	:param delay: (int) Delay between reads executions
34
35	:returns: Nothing
36	"""
37	self.fnc_args = None	1✔
38	self.delay = delay	1✔
39	if isinstance(fnc, tuple):	1✔
40	self.func, self.fnc_args = fnc	×
41	elif hasattr(fnc, "__call__"):	1✔
42	self.func = fnc	1✔
43	else:
44	raise ValueError(	×
45	"You must pass a function or a tuple (function,args) to this..."
46	)
47	Task.__init__(self, name=name, delay=delay)	1✔
48
49	async def task(self) -> None:	1✔
50	# Prefer awaiting async callables; offload sync callables with to_thread.
51	def _is_async_callable(fn: Callable) -> bool:	1✔
52	if inspect.iscoroutinefunction(fn):	1✔
53	return True	1✔
54	call = getattr(fn, "__call__", None)	1✔
55	if call and inspect.iscoroutinefunction(call):	1✔
NEW 56	return True	×
57	if isinstance(fn, functools.partial):	1✔
NEW 58	return _is_async_callable(fn.func)	×
59	return False	1✔
60
61	if self.fnc_args is not None:	1✔
NEW 62	if _is_async_callable(self.func):	×
UNCOV 63	await self.func(self.fnc_args)	×
64	else:
NEW 65	try:	×
NEW 66	await asyncio.to_thread(self.func, self.fnc_args)	×
NEW 67	except (RuntimeError, AttributeError):	×
68	# Fallback for environments without to_thread or loop quirks
NEW 69	loop = asyncio.get_running_loop()	×
NEW 70	await loop.run_in_executor(None, self.func, self.fnc_args)	×
71	else:
72	if _is_async_callable(self.func):	1✔
73	await self.func()	1✔
74	else:
75	try:	1✔
76	await asyncio.to_thread(self.func)	1✔
NEW 77	except (RuntimeError, AttributeError):	×
NEW 78	loop = asyncio.get_running_loop()	×
NEW 79	await loop.run_in_executor(None, self.func)	×
80	await asyncio.sleep(self.delay)	1✔

ChristianTremblay / BAC0 / 17009917031

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