cengal.parallel_execution.coroutines.coro_standard_services.timer_func_runner.versions.v_0.timer_func_runner

Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/

View Source

  1#!/usr/bin/env python
  2# coding=utf-8
  3
  4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>
  5# 
  6# Licensed under the Apache License, Version 2.0 (the "License");
  7# you may not use this file except in compliance with the License.
  8# You may obtain a copy of the License at
  9# 
 10#     http://www.apache.org/licenses/LICENSE-2.0
 11# 
 12# Unless required by applicable law or agreed to in writing, software
 13# distributed under the License is distributed on an "AS IS" BASIS,
 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 15# See the License for the specific language governing permissions and
 16# limitations under the License.
 17
 18
 19"""
 20Module Docstring
 21Docstrings: http://www.python.org/dev/peps/pep-0257/
 22"""
 23
 24
 25__author__ = "ButenkoMS <gtalk@butenkoms.space>"
 26__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
 27__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
 28__license__ = "Apache License, Version 2.0"
 29__version__ = "4.4.1"
 30__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
 31__email__ = "gtalk@butenkoms.space"
 32# __status__ = "Prototype"
 33__status__ = "Development"
 34# __status__ = "Production"
 35
 36
 37__all__ = ['TimerFuncRunner', 'add_timer_func_run_from_other_service', 'discard_timer_func_run_from_other_service', 'timer_func_run_on', 'try_timer_func_run_on', 'atimer_func_run_on', 'atry_timer_func_run_on', 'timer_func_run', 'try_timer_func_run', 'atimer_func_run', 'atry_timer_func_run']
 38
 39from cengal.parallel_execution.coroutines.coro_scheduler import *
 40from cengal.parallel_execution.coroutines.coro_standard_services_internal_lib.service_with_a_direct_request import *
 41from cengal.code_flow_control.smart_values import ValueExistence
 42from cengal.time_management.timer import Timer, TimerRequest
 43from functools import partial
 44from typing import Tuple, Callable, Optional, Union, overload
 45
 46
 47class TimerFuncRunnerRequest(ServiceRequest):
 48    def add(self, delay: float, handler: Callable, *args, **kwargs) -> TimerRequest:
 49        return self._save(0, delay, handler, args, kwargs)
 50    def discard(self, timer_request: TimerRequest) -> bool:
 51        return self._save(1, timer_request)
 52
 53
 54class TimerFuncRunner(DualImmediateProcessingServiceMixin, ServiceWithADirectRequestMixin, TypedService[TimerRequest]):
 55    def __init__(self, loop: CoroSchedulerType):
 56        super(TimerFuncRunner, self).__init__(loop)
 57        self.timer = Timer()
 58        self.pending_tasks_number = 0
 59        self.pending_foreground_tasks_number = 0
 60        self.direct_requests = list()
 61        self._request_workers = {
 62            0: self._on_add,
 63            1: self._on_discard,
 64        }
 65
 66    def single_task_registration_or_immediate_processing_single(
 67            self, delay: float, handler: Callable, *args, **kwargs) -> Tuple[bool, TimerRequest, None]:
 68        timer_request: TimerRequest = self._add_request_impl(not self.current_caller_coro_info.coro.is_background_coro, delay, handler, *args, **kwargs)
 69        self.make_live()
 70        return True, timer_request, None
 71    
 72    def _add_request_impl(self, foreground: bool, delay: float, handler: Callable, *args, **kwargs) -> TimerRequest:
 73        def timer_handler_func(foreground: bool, handler_: Callable, *args_, **kwargs_):
 74            try:
 75                handler_(*args_, **kwargs_)
 76            except:
 77                self._loop.logger.exception('TimerFuncRunner. Event handler error')
 78            finally:
 79                self.task_triggered(foreground)
 80
 81        timer_handler = partial(timer_handler_func, foreground, handler, *args, **kwargs)
 82        self.task_added(foreground)
 83        timer_request: TimerRequest = self.timer.register(timer_handler, delay)
 84        timer_request.foreground = foreground
 85        return timer_request
 86    
 87    def _on_add(self, delay: float, handler: Callable, *args, **kwargs) -> Tuple[bool, TimerRequest, None]:
 88        timer_request: TimerRequest = self._add_request_impl(
 89            not self.current_caller_coro_info.coro.is_background_coro, delay, handler, *args, **kwargs)
 90        self.make_live()
 91        return True, timer_request, None
 92    
 93    def _on_discard(self, timer_request: TimerRequest) -> Tuple[bool, TimerRequest, None]:
 94        result: bool = self.timer.discard(timer_request)
 95        if result:
 96            self.task_triggered(timer_request.foreground)
 97        
 98        return True, result, None
 99    
100    def add_timer_func_run_from_other_service(self, foreground: bool, delay: float, handler: Callable, *args, **kwargs) -> TimerRequest:
101        timer_request: TimerRequest = self._add_request_impl(foreground, delay, handler, *args, **kwargs)
102        self.make_live()
103        return timer_request
104    
105    def discard_timer_func_run_from_other_service(self, timer_request: TimerRequest) -> bool:
106        result: bool = self.timer.discard(timer_request)
107        if result:
108            self.task_triggered(timer_request.foreground)
109        
110        return result
111
112    def full_processing_iteration(self):
113        if self.direct_requests:
114            direct_requests_buff = self.direct_requests
115            self.direct_requests = type(direct_requests_buff)()
116            for delay, handler, args, kwargs in direct_requests_buff:
117                self._add_request_impl(True, delay, handler, *args, **kwargs)
118        
119        self.timer()
120        if 0 == self.pending_tasks_number:
121            self.make_dead()
122    
123    def _add_direct_request(self, delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[None]:
124        self.direct_requests.append((delay, handler, args, kwargs))
125        self.make_live()
126        return (False, None)
127
128    def task_added(self, foreground: bool):
129        self.pending_tasks_number += 1
130        self.pending_foreground_tasks_number += 1 if foreground else 0
131
132    def task_triggered(self, foreground: bool):
133        self.pending_tasks_number -= 1
134        self.pending_foreground_tasks_number -= 1 if foreground else 0
135
136    def in_work(self) -> bool:
137        result: bool = (self.pending_tasks_number != 0) or bool(self.direct_requests)
138        return self.thrifty_in_work(result)
139    
140    def in_forground_work(self) -> bool:
141        return self.pending_foreground_tasks_number or bool(self.direct_requests)
142    
143    def time_left_before_next_event(self) -> Tuple[bool, Optional[Union[int, float]]]:
144        return True, self.timer.nearest_event()
145
146
147TimerFuncRunnerRequest.default_service_type = TimerFuncRunner
148
149
150def add_timer_func_run_from_other_service(current_service: Service, foreground: bool, delay: float, handler: Callable, *args, **kwargs) -> TimerRequest:
151    timer_func_runner: TimerFuncRunner = current_service._loop.get_service_instance(TimerFuncRunner)
152    return timer_func_runner.add_timer_func_run_from_other_service(foreground, delay, handler, *args, **kwargs)
153
154
155def discard_timer_func_run_from_other_service(current_service: Service, timer_request: TimerRequest) -> bool:
156    timer_func_runner: TimerFuncRunner = current_service._loop.get_service_instance(TimerFuncRunner)
157    return timer_func_runner.discard_timer_func_run_from_other_service(timer_request)
158
159
160def timer_func_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
161    """_summary_
162        context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module
163        
164        An example:
165
166        from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
167        from cengal.parallel_execution.coroutines.coro_standard_services.timer_func_runner import timer_func_run_on
168        from typing import Optional, Union
169
170        def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
171            try:
172                def print_hello_world(name: str):
173                    print(f'Hello Wrold from {name}!)
174                
175                timer_func_run_on(10, print_hello_world, 'John Doe')
176            except CoroSchedulerContextIsNotAvailable:
177                print('We are outside of the loop AND no loop was selected as a Primary AND our given `loop` var is None)
178        
179    Args:
180        context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_
181        delay (float): delay in seconds
182        handler (Callable): handler
183
184    Returns:
185        ValueExistence[Optional[CoroID]]: _description_
186    """
187    return make_request_to_service_with_context(context, TimerFuncRunner, delay, handler, *args, **kwargs)
188
189
190def try_timer_func_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
191    """_summary_
192        context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module
193        
194        An example:
195
196        from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
197        from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import try_put_coro_to
198        from typing import Optional, Union
199
200        def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
201            def print_hello_world(name: str):
202                print(f'Hello Wrold from {name}!)
203            
204            try_timer_func_run_on(10, print_hello_world, 'John Doe')
205        
206    Args:
207        context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_
208        delay (float): delay in seconds
209        handler (Callable): handler
210
211    Returns:
212        ValueExistence[Optional[CoroID]]: _description_
213    """
214    return try_make_request_to_service_with_context(context, TimerFuncRunner, delay, handler, *args, **kwargs)
215
216
217async def atimer_func_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[CoroID]:
218    return await amake_request_to_service_with_context(context, TimerFuncRunner, delay, handler, *args, **kwargs)
219
220
221async def atry_timer_func_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
222    return await atry_make_request_to_service_with_context(context, TimerFuncRunner, delay, handler, *args, **kwargs)
223
224
225def timer_func_run(delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[CoroID]:
226    return make_request_to_service(TimerFuncRunner, delay, handler, *args, **kwargs)
227
228
229def try_timer_func_run(delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
230    return try_make_request_to_service(TimerFuncRunner, delay, handler, *args, **kwargs)
231
232
233async def atimer_func_run(delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[CoroID]:
234    return await amake_request_to_service(TimerFuncRunner, delay, handler, *args, **kwargs)
235
236
237async def atry_timer_func_run(delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
238    return await atry_make_request_to_service(TimerFuncRunner, delay, handler, *args, **kwargs)

class TimerFuncRunner(cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.DualImmediateProcessingServiceMixin, cengal.parallel_execution.coroutines.coro_standard_services_internal_lib.service_with_a_direct_request.versions.v_0.service_with_a_direct_request.ServiceWithADirectRequestMixin, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.TypedService[cengal.time_management.timer.versions.v_0.timer.TimerRequest]): View Source

 55class TimerFuncRunner(DualImmediateProcessingServiceMixin, ServiceWithADirectRequestMixin, TypedService[TimerRequest]):
 56    def __init__(self, loop: CoroSchedulerType):
 57        super(TimerFuncRunner, self).__init__(loop)
 58        self.timer = Timer()
 59        self.pending_tasks_number = 0
 60        self.pending_foreground_tasks_number = 0
 61        self.direct_requests = list()
 62        self._request_workers = {
 63            0: self._on_add,
 64            1: self._on_discard,
 65        }
 66
 67    def single_task_registration_or_immediate_processing_single(
 68            self, delay: float, handler: Callable, *args, **kwargs) -> Tuple[bool, TimerRequest, None]:
 69        timer_request: TimerRequest = self._add_request_impl(not self.current_caller_coro_info.coro.is_background_coro, delay, handler, *args, **kwargs)
 70        self.make_live()
 71        return True, timer_request, None
 72    
 73    def _add_request_impl(self, foreground: bool, delay: float, handler: Callable, *args, **kwargs) -> TimerRequest:
 74        def timer_handler_func(foreground: bool, handler_: Callable, *args_, **kwargs_):
 75            try:
 76                handler_(*args_, **kwargs_)
 77            except:
 78                self._loop.logger.exception('TimerFuncRunner. Event handler error')
 79            finally:
 80                self.task_triggered(foreground)
 81
 82        timer_handler = partial(timer_handler_func, foreground, handler, *args, **kwargs)
 83        self.task_added(foreground)
 84        timer_request: TimerRequest = self.timer.register(timer_handler, delay)
 85        timer_request.foreground = foreground
 86        return timer_request
 87    
 88    def _on_add(self, delay: float, handler: Callable, *args, **kwargs) -> Tuple[bool, TimerRequest, None]:
 89        timer_request: TimerRequest = self._add_request_impl(
 90            not self.current_caller_coro_info.coro.is_background_coro, delay, handler, *args, **kwargs)
 91        self.make_live()
 92        return True, timer_request, None
 93    
 94    def _on_discard(self, timer_request: TimerRequest) -> Tuple[bool, TimerRequest, None]:
 95        result: bool = self.timer.discard(timer_request)
 96        if result:
 97            self.task_triggered(timer_request.foreground)
 98        
 99        return True, result, None
100    
101    def add_timer_func_run_from_other_service(self, foreground: bool, delay: float, handler: Callable, *args, **kwargs) -> TimerRequest:
102        timer_request: TimerRequest = self._add_request_impl(foreground, delay, handler, *args, **kwargs)
103        self.make_live()
104        return timer_request
105    
106    def discard_timer_func_run_from_other_service(self, timer_request: TimerRequest) -> bool:
107        result: bool = self.timer.discard(timer_request)
108        if result:
109            self.task_triggered(timer_request.foreground)
110        
111        return result
112
113    def full_processing_iteration(self):
114        if self.direct_requests:
115            direct_requests_buff = self.direct_requests
116            self.direct_requests = type(direct_requests_buff)()
117            for delay, handler, args, kwargs in direct_requests_buff:
118                self._add_request_impl(True, delay, handler, *args, **kwargs)
119        
120        self.timer()
121        if 0 == self.pending_tasks_number:
122            self.make_dead()
123    
124    def _add_direct_request(self, delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[None]:
125        self.direct_requests.append((delay, handler, args, kwargs))
126        self.make_live()
127        return (False, None)
128
129    def task_added(self, foreground: bool):
130        self.pending_tasks_number += 1
131        self.pending_foreground_tasks_number += 1 if foreground else 0
132
133    def task_triggered(self, foreground: bool):
134        self.pending_tasks_number -= 1
135        self.pending_foreground_tasks_number -= 1 if foreground else 0
136
137    def in_work(self) -> bool:
138        result: bool = (self.pending_tasks_number != 0) or bool(self.direct_requests)
139        return self.thrifty_in_work(result)
140    
141    def in_forground_work(self) -> bool:
142        return self.pending_foreground_tasks_number or bool(self.direct_requests)
143    
144    def time_left_before_next_event(self) -> Tuple[bool, Optional[Union[int, float]]]:
145        return True, self.timer.nearest_event()

Abstract base class for generic types.

A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as::

class Mapping(Generic[KT, VT]): def __getitem__(self, key: KT) -> VT: ... # Etc.

This class can then be used as follows::

def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT: try: return mapping[key] except KeyError: return default

TimerFuncRunner( loop: Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable]) View Source

56    def __init__(self, loop: CoroSchedulerType):
57        super(TimerFuncRunner, self).__init__(loop)
58        self.timer = Timer()
59        self.pending_tasks_number = 0
60        self.pending_foreground_tasks_number = 0
61        self.direct_requests = list()
62        self._request_workers = {
63            0: self._on_add,
64            1: self._on_discard,
65        }

timer

pending_tasks_number

pending_foreground_tasks_number

direct_requests

def add_timer_func_run_from_other_service( self, foreground: bool, delay: float, handler: Callable, *args, **kwargs) -> cengal.time_management.timer.versions.v_0.timer.TimerRequest: View Source

101    def add_timer_func_run_from_other_service(self, foreground: bool, delay: float, handler: Callable, *args, **kwargs) -> TimerRequest:
102        timer_request: TimerRequest = self._add_request_impl(foreground, delay, handler, *args, **kwargs)
103        self.make_live()
104        return timer_request

def discard_timer_func_run_from_other_service( self, timer_request: cengal.time_management.timer.versions.v_0.timer.TimerRequest) -> bool: View Source

106    def discard_timer_func_run_from_other_service(self, timer_request: TimerRequest) -> bool:
107        result: bool = self.timer.discard(timer_request)
108        if result:
109            self.task_triggered(timer_request.foreground)
110        
111        return result

def full_processing_iteration(self): View Source

113    def full_processing_iteration(self):
114        if self.direct_requests:
115            direct_requests_buff = self.direct_requests
116            self.direct_requests = type(direct_requests_buff)()
117            for delay, handler, args, kwargs in direct_requests_buff:
118                self._add_request_impl(True, delay, handler, *args, **kwargs)
119        
120        self.timer()
121        if 0 == self.pending_tasks_number:
122            self.make_dead()

def task_added(self, foreground: bool): View Source

129    def task_added(self, foreground: bool):
130        self.pending_tasks_number += 1
131        self.pending_foreground_tasks_number += 1 if foreground else 0

def task_triggered(self, foreground: bool): View Source

133    def task_triggered(self, foreground: bool):
134        self.pending_tasks_number -= 1
135        self.pending_foreground_tasks_number -= 1 if foreground else 0

def in_work(self) -> bool: View Source

137    def in_work(self) -> bool:
138        result: bool = (self.pending_tasks_number != 0) or bool(self.direct_requests)
139        return self.thrifty_in_work(result)

Will be executed twice per iteration: once before and once after the full_processing_iteration() execution

Raises: NotImplementedError: _description_

Returns: bool: _description_

def in_forground_work(self) -> bool: View Source

141    def in_forground_work(self) -> bool:
142        return self.pending_foreground_tasks_number or bool(self.direct_requests)

def time_left_before_next_event(self) -> Tuple[bool, Union[int, float, NoneType]]: View Source

144    def time_left_before_next_event(self) -> Tuple[bool, Optional[Union[int, float]]]:
145        return True, self.timer.nearest_event()

Inherited Members

cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.DualImmediateProcessingServiceMixin

single_task_registration_or_immediate_processing

single_task_registration_or_immediate_processing_multiple

single_task_registration_or_immediate_processing_single

cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Service

current_caller_coro_info

iteration

make_response

register_response

put_task

resolve_request

try_resolve_request

thrifty_in_work

is_low_latency

make_live

make_dead

service_id_impl

service_id

destroy

def add_timer_func_run_from_other_service( current_service: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Service, foreground: bool, delay: float, handler: Callable, *args, **kwargs) -> cengal.time_management.timer.versions.v_0.timer.TimerRequest: View Source

151def add_timer_func_run_from_other_service(current_service: Service, foreground: bool, delay: float, handler: Callable, *args, **kwargs) -> TimerRequest:
152    timer_func_runner: TimerFuncRunner = current_service._loop.get_service_instance(TimerFuncRunner)
153    return timer_func_runner.add_timer_func_run_from_other_service(foreground, delay, handler, *args, **kwargs)

def discard_timer_func_run_from_other_service( current_service: cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Service, timer_request: cengal.time_management.timer.versions.v_0.timer.TimerRequest) -> bool: View Source

156def discard_timer_func_run_from_other_service(current_service: Service, timer_request: TimerRequest) -> bool:
157    timer_func_runner: TimerFuncRunner = current_service._loop.get_service_instance(TimerFuncRunner)
158    return timer_func_runner.discard_timer_func_run_from_other_service(timer_request)

def timer_func_run_on( context: Tuple[Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable, NoneType], Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, NoneType], bool], delay: float, handler: Callable, *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[typing.Union[int, NoneType]]: View Source

161def timer_func_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
162    """_summary_
163        context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module
164        
165        An example:
166
167        from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
168        from cengal.parallel_execution.coroutines.coro_standard_services.timer_func_runner import timer_func_run_on
169        from typing import Optional, Union
170
171        def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
172            try:
173                def print_hello_world(name: str):
174                    print(f'Hello Wrold from {name}!)
175                
176                timer_func_run_on(10, print_hello_world, 'John Doe')
177            except CoroSchedulerContextIsNotAvailable:
178                print('We are outside of the loop AND no loop was selected as a Primary AND our given `loop` var is None)
179        
180    Args:
181        context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_
182        delay (float): delay in seconds
183        handler (Callable): handler
184
185    Returns:
186        ValueExistence[Optional[CoroID]]: _description_
187    """
188    return make_request_to_service_with_context(context, TimerFuncRunner, delay, handler, *args, **kwargs)

_summary_ context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module

An example:

from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
from cengal.parallel_execution.coroutines.coro_standard_services.timer_func_runner import timer_func_run_on
from typing import Optional, Union

def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
    try:
        def print_hello_world(name: str):
            print(f'Hello Wrold from {name}!)

        timer_func_run_on(10, print_hello_world, 'John Doe')
    except CoroSchedulerContextIsNotAvailable:
        print('We are outside of the loop AND no loop was selected as a Primary AND our given `loop` var is None)

Args: context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_ delay (float): delay in seconds handler (Callable): handler

Returns: ValueExistence[Optional[CoroID]]: _description_

def try_timer_func_run_on( context: Tuple[Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable, NoneType], Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, NoneType], bool], delay: float, handler: Callable, *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[typing.Union[int, NoneType]]: View Source

191def try_timer_func_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
192    """_summary_
193        context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module
194        
195        An example:
196
197        from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
198        from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import try_put_coro_to
199        from typing import Optional, Union
200
201        def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
202            def print_hello_world(name: str):
203                print(f'Hello Wrold from {name}!)
204            
205            try_timer_func_run_on(10, print_hello_world, 'John Doe')
206        
207    Args:
208        context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_
209        delay (float): delay in seconds
210        handler (Callable): handler
211
212    Returns:
213        ValueExistence[Optional[CoroID]]: _description_
214    """
215    return try_make_request_to_service_with_context(context, TimerFuncRunner, delay, handler, *args, **kwargs)

_summary_ context can be generated by one of the [interface_and_loop_with_backup_loop, get_interface_and_loop_with_backup_loop, interface_and_loop_with_explicit_loop, get_interface_and_loop_with_explicit_loop, interface_for_an_explicit_loop, get_interface_for_an_explicit_loop] functions from the cengal/parallel_execution/coroutines/coro_scheduler module

An example:

from cengal.parallel_execution.coroutines.coro_scheduler import get_interface_and_loop_with_explicit_loop, CoroSchedulerType, ExplicitWorker, Worker, CoroID
from cengal.parallel_execution.coroutines.coro_standard_services.put_coro import try_put_coro_to
from typing import Optional, Union

def my_func(loop: CoroSchedulerType, coro_worker: AnyWorker, a, b) -> Optional[CoroID]:
    def print_hello_world(name: str):
        print(f'Hello Wrold from {name}!)

    try_timer_func_run_on(10, print_hello_world, 'John Doe')

Args: context (Tuple[Optional[CoroSchedulerType], Optional[Interface], bool]): _description_ delay (float): delay in seconds handler (Callable): handler

Returns: ValueExistence[Optional[CoroID]]: _description_

async def atimer_func_run_on( context: Tuple[Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable, NoneType], Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, NoneType], bool], delay: float, handler: Callable, *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[int]: View Source

218async def atimer_func_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[CoroID]:
219    return await amake_request_to_service_with_context(context, TimerFuncRunner, delay, handler, *args, **kwargs)

async def atry_timer_func_run_on( context: Tuple[Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable, NoneType], Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Interface, NoneType], bool], delay: float, handler: Callable, *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[typing.Union[int, NoneType]]: View Source

222async def atry_timer_func_run_on(context: Tuple[Optional[CoroSchedulerType], Optional[Interface], bool], delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
223    return await atry_make_request_to_service_with_context(context, TimerFuncRunner, delay, handler, *args, **kwargs)

def timer_func_run( delay: float, handler: Callable, *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[int]: View Source

226def timer_func_run(delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[CoroID]:
227    return make_request_to_service(TimerFuncRunner, delay, handler, *args, **kwargs)

def try_timer_func_run( delay: float, handler: Callable, *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[typing.Union[int, NoneType]]: View Source

230def try_timer_func_run(delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
231    return try_make_request_to_service(TimerFuncRunner, delay, handler, *args, **kwargs)

async def atimer_func_run( delay: float, handler: Callable, *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[int]: View Source

234async def atimer_func_run(delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[CoroID]:
235    return await amake_request_to_service(TimerFuncRunner, delay, handler, *args, **kwargs)

async def atry_timer_func_run( delay: float, handler: Callable, *args, **kwargs) -> cengal.code_flow_control.smart_values.versions.v_2.smart_values.ValueExistence[typing.Union[int, NoneType]]: View Source

238async def atry_timer_func_run(delay: float, handler: Callable, *args, **kwargs) -> ValueExistence[Optional[CoroID]]:
239    return await atry_make_request_to_service(TimerFuncRunner, delay, handler, *args, **kwargs)