cengal.time_management.timer.versions.v_1.timer
Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/
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__ = ['TimerHandler', 'TimeInSeconds', 'TimerRequest', 'Timer'] 38 39 40from cengal.time_management.cpu_clock import cpu_clock 41from cengal.math.numbers import RationalNumber 42from cengal.data_generation.id_generator import IDGenerator, GeneratorType 43from collections import deque 44from bisect import bisect_left 45from typing import Any, Callable, Optional, Set, Dict, Tuple, List, Deque 46 47TimerHandler = Callable 48TimeInSeconds = RationalNumber 49RequestId = int 50 51 52class TimerRequest: 53 def __init__(self, id: RequestId, timer_handler: TimerHandler, desired_time: TimeInSeconds): 54 self.request_id: RequestId = id 55 self.timer_handler = timer_handler 56 self.requested_time = desired_time 57 self.start_time = cpu_clock() 58 self.real_end_time = None 59 self.processed = False 60 61 def __call__(self, current_time = None) -> bool: 62 if not self.processed: 63 if current_time is None: current_time = cpu_clock() 64 if self.start_time > current_time: 65 self.start_time = current_time 66 time_delta = current_time - self.start_time 67 if time_delta >= self.requested_time: 68 self.timer_handler() 69 self.processed = True 70 71 return self.processed 72 73 def time_left(self, current_time = None) -> Optional[TimeInSeconds]: 74 if self.processed: 75 return None 76 77 if current_time is None: current_time = cpu_clock() 78 if self.start_time > current_time: 79 self.start_time = current_time 80 time_delta = current_time - self.start_time 81 if time_delta >= self.requested_time: 82 return 0 83 else: 84 return self.requested_time - time_delta 85 86 87class Timer: 88 # TODO: (old todo) improve algorithm using `cengal\time_management\repeat_for_a_time` module 89 def __init__(self): 90 self.request_id_generator: IDGenerator = IDGenerator(GeneratorType.integer) 91 self.requests_dict: Dict[RequestId, TimerRequest] = dict() 92 self.requests_order: Deque[Tuple[RequestId, TimeInSeconds]] = deque() 93 self.times_left_order: Deque[TimeInSeconds] = deque() 94 self.last_time: Optional[TimeInSeconds] = None 95 96 def register(self, timer_handler: TimerHandler, desired_time: TimeInSeconds) -> TimerRequest: 97 request_id = self.request_id_generator() 98 request = TimerRequest(request_id, timer_handler, desired_time) 99 self.requests_dict[request_id] = request 100 if self.requests_order: 101 self._fast_update_requests_order(request_id, desired_time) 102 else: 103 self._init_requests_order(request_id, desired_time) 104 105 return request 106 107 def _init_requests_order(self, request_id, desired_time): 108 current_time = cpu_clock() 109 self.last_time = current_time 110 self.requests_order.append((request_id, desired_time)) 111 self.times_left_order.append(desired_time) 112 113 def _fast_update_requests_order(self, request_id: RequestId, desired_time: TimeInSeconds): 114 current_time: TimeInSeconds = cpu_clock() 115 time_left: TimeInSeconds = desired_time + (current_time - self.last_time) 116 if self.requests_order[0][1] >= time_left: 117 self.requests_order.appendleft((request_id, time_left)) 118 self.times_left_order.appendleft(time_left) 119 elif self.requests_order[-1][1] <= time_left: 120 self.requests_order.append((request_id, time_left)) 121 self.times_left_order.append(time_left) 122 else: 123 insert_pos: int = bisect_left(self.times_left_order, time_left) 124 self.requests_order.insert(insert_pos, (request_id, time_left)) 125 self.times_left_order.insert(insert_pos, time_left) 126 127 # Not needed anymore since much slower 128 def _update_requests_order(self): 129 current_time = cpu_clock() 130 self.last_time = current_time 131 requests_times_left = ((request.request_id, request.time_left(current_time)) for request in self.requests_dict.values()) 132 self.requests_order = deque((request_and_time_left for request_and_time_left in sorted(requests_times_left, key=lambda request_and_time_left: request_and_time_left[1]))) 133 self.times_left_order = deque((time_left for _, time_left in self.requests_order)) 134 135 def discard(self, timer_request: TimerRequest): 136 try: 137 del self.requests_dict[timer_request.request_id] 138 return True 139 except KeyError: 140 return False 141 142 def __call__(self, current_time = None): 143 if current_time is None: current_time = cpu_clock() 144 time_spend = current_time - self.last_time 145 need_to_process_num: int = 0 146 for _, time_left in self.requests_order: 147 if time_left > time_spend: 148 break 149 150 need_to_process_num += 1 151 152 if need_to_process_num: 153 need_to_process: List[TimerRequest] = list() 154 for _ in range(need_to_process_num): 155 need_to_process.append(self.requests_dict.pop(self.requests_order.popleft()[0])) 156 157 for request in need_to_process: 158 request() 159 160 def nearest_event(self, current_time = None) -> Optional[TimeInSeconds]: 161 if current_time is None: current_time = cpu_clock() 162 if not self.requests_order: 163 return None 164 165 nearest_event_time = self.requests_order[0][1] - (current_time - self.last_time) 166 if nearest_event_time >= 0: 167 return nearest_event_time 168 else: 169 return 0
TimerHandler =
typing.Callable
TimeInSeconds =
typing.Union[int, float]
class
TimerRequest:
53class TimerRequest: 54 def __init__(self, id: RequestId, timer_handler: TimerHandler, desired_time: TimeInSeconds): 55 self.request_id: RequestId = id 56 self.timer_handler = timer_handler 57 self.requested_time = desired_time 58 self.start_time = cpu_clock() 59 self.real_end_time = None 60 self.processed = False 61 62 def __call__(self, current_time = None) -> bool: 63 if not self.processed: 64 if current_time is None: current_time = cpu_clock() 65 if self.start_time > current_time: 66 self.start_time = current_time 67 time_delta = current_time - self.start_time 68 if time_delta >= self.requested_time: 69 self.timer_handler() 70 self.processed = True 71 72 return self.processed 73 74 def time_left(self, current_time = None) -> Optional[TimeInSeconds]: 75 if self.processed: 76 return None 77 78 if current_time is None: current_time = cpu_clock() 79 if self.start_time > current_time: 80 self.start_time = current_time 81 time_delta = current_time - self.start_time 82 if time_delta >= self.requested_time: 83 return 0 84 else: 85 return self.requested_time - time_delta
def
time_left(self, current_time=None) -> Union[int, float, NoneType]:
74 def time_left(self, current_time = None) -> Optional[TimeInSeconds]: 75 if self.processed: 76 return None 77 78 if current_time is None: current_time = cpu_clock() 79 if self.start_time > current_time: 80 self.start_time = current_time 81 time_delta = current_time - self.start_time 82 if time_delta >= self.requested_time: 83 return 0 84 else: 85 return self.requested_time - time_delta
class
Timer:
88class Timer: 89 # TODO: (old todo) improve algorithm using `cengal\time_management\repeat_for_a_time` module 90 def __init__(self): 91 self.request_id_generator: IDGenerator = IDGenerator(GeneratorType.integer) 92 self.requests_dict: Dict[RequestId, TimerRequest] = dict() 93 self.requests_order: Deque[Tuple[RequestId, TimeInSeconds]] = deque() 94 self.times_left_order: Deque[TimeInSeconds] = deque() 95 self.last_time: Optional[TimeInSeconds] = None 96 97 def register(self, timer_handler: TimerHandler, desired_time: TimeInSeconds) -> TimerRequest: 98 request_id = self.request_id_generator() 99 request = TimerRequest(request_id, timer_handler, desired_time) 100 self.requests_dict[request_id] = request 101 if self.requests_order: 102 self._fast_update_requests_order(request_id, desired_time) 103 else: 104 self._init_requests_order(request_id, desired_time) 105 106 return request 107 108 def _init_requests_order(self, request_id, desired_time): 109 current_time = cpu_clock() 110 self.last_time = current_time 111 self.requests_order.append((request_id, desired_time)) 112 self.times_left_order.append(desired_time) 113 114 def _fast_update_requests_order(self, request_id: RequestId, desired_time: TimeInSeconds): 115 current_time: TimeInSeconds = cpu_clock() 116 time_left: TimeInSeconds = desired_time + (current_time - self.last_time) 117 if self.requests_order[0][1] >= time_left: 118 self.requests_order.appendleft((request_id, time_left)) 119 self.times_left_order.appendleft(time_left) 120 elif self.requests_order[-1][1] <= time_left: 121 self.requests_order.append((request_id, time_left)) 122 self.times_left_order.append(time_left) 123 else: 124 insert_pos: int = bisect_left(self.times_left_order, time_left) 125 self.requests_order.insert(insert_pos, (request_id, time_left)) 126 self.times_left_order.insert(insert_pos, time_left) 127 128 # Not needed anymore since much slower 129 def _update_requests_order(self): 130 current_time = cpu_clock() 131 self.last_time = current_time 132 requests_times_left = ((request.request_id, request.time_left(current_time)) for request in self.requests_dict.values()) 133 self.requests_order = deque((request_and_time_left for request_and_time_left in sorted(requests_times_left, key=lambda request_and_time_left: request_and_time_left[1]))) 134 self.times_left_order = deque((time_left for _, time_left in self.requests_order)) 135 136 def discard(self, timer_request: TimerRequest): 137 try: 138 del self.requests_dict[timer_request.request_id] 139 return True 140 except KeyError: 141 return False 142 143 def __call__(self, current_time = None): 144 if current_time is None: current_time = cpu_clock() 145 time_spend = current_time - self.last_time 146 need_to_process_num: int = 0 147 for _, time_left in self.requests_order: 148 if time_left > time_spend: 149 break 150 151 need_to_process_num += 1 152 153 if need_to_process_num: 154 need_to_process: List[TimerRequest] = list() 155 for _ in range(need_to_process_num): 156 need_to_process.append(self.requests_dict.pop(self.requests_order.popleft()[0])) 157 158 for request in need_to_process: 159 request() 160 161 def nearest_event(self, current_time = None) -> Optional[TimeInSeconds]: 162 if current_time is None: current_time = cpu_clock() 163 if not self.requests_order: 164 return None 165 166 nearest_event_time = self.requests_order[0][1] - (current_time - self.last_time) 167 if nearest_event_time >= 0: 168 return nearest_event_time 169 else: 170 return 0
requests_dict: Dict[int, TimerRequest]
def
register( self, timer_handler: typing.Callable, desired_time: typing.Union[int, float]) -> TimerRequest:
97 def register(self, timer_handler: TimerHandler, desired_time: TimeInSeconds) -> TimerRequest: 98 request_id = self.request_id_generator() 99 request = TimerRequest(request_id, timer_handler, desired_time) 100 self.requests_dict[request_id] = request 101 if self.requests_order: 102 self._fast_update_requests_order(request_id, desired_time) 103 else: 104 self._init_requests_order(request_id, desired_time) 105 106 return request
def
nearest_event(self, current_time=None) -> Union[int, float, NoneType]:
161 def nearest_event(self, current_time = None) -> Optional[TimeInSeconds]: 162 if current_time is None: current_time = cpu_clock() 163 if not self.requests_order: 164 return None 165 166 nearest_event_time = self.requests_order[0][1] - (current_time - self.last_time) 167 if nearest_event_time >= 0: 168 return nearest_event_time 169 else: 170 return 0