cengal.time_management.timer.versions.v_1.timer

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__ = ['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: View Source

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

TimerRequest( id: int, timer_handler: typing.Callable, desired_time: typing.Union[int, float]) View Source

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

request_id: int

timer_handler

requested_time

start_time

real_end_time

processed

def time_left(self, current_time=None) -> Union[int, float, NoneType]: View Source

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: View Source

 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

request_id_generator: <cyfunction IDGenerator at 0x7f5316b55ad0>

requests_dict: Dict[int, TimerRequest]

requests_order: Deque[Tuple[int, Union[int, float]]]

times_left_order: Deque[Union[int, float]]

last_time: Union[int, float, NoneType]

def register( self, timer_handler: typing.Callable, desired_time: typing.Union[int, float]) -> TimerRequest: View Source

 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 discard( self, timer_request: TimerRequest): View Source

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

def nearest_event(self, current_time=None) -> Union[int, float, NoneType]: View Source

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