17868482439

Committed 19 Sep 2025 07:59PM UTC coverage: 39.741% (-1.0%) from 40.7%

Build # 17868482439

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #567 from ChristianTremblay/develop

Task fix + Mypy

Run Details

150 of 437 new or added lines in 24 files covered. (34.32%)

19 existing lines in 9 files now uncovered.

2270 of 5712 relevant lines covered (39.74%)

0.4 hits per line

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

52.38

/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: Union[int, float] = 60,
        name: str = "recurring",
        minimum_delay: Union[int, float] = 5,
    ) -> 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, minimum_delay=minimum_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)

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

ChristianTremblay / BAC0 / 17868482439

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