cengal.time_management.timer.versions.v_0.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 time import perf_counter
 41from cengal.math.numbers import RationalNumber
 42from typing import Any, Callable, Optional, Set
 43
 44TimerHandler = Callable
 45TimeInSeconds = RationalNumber
 46
 47
 48class TimerRequest:
 49    def __init__(self, timer_handler: TimerHandler, desired_time: TimeInSeconds):
 50        self.timer_handler = timer_handler
 51        self.requested_time = desired_time
 52        self.start_time = perf_counter()
 53        self.real_end_time = None
 54        self.processed = False
 55
 56    def __call__(self) -> bool:
 57        if not self.processed:
 58            current_time = perf_counter()
 59            if self.start_time > current_time:
 60                self.start_time = current_time
 61            time_delta = current_time - self.start_time
 62            if time_delta >= self.requested_time:
 63                self.timer_handler()
 64                self.processed = True
 65        return self.processed
 66
 67    def time_left(self) -> Optional[TimeInSeconds]:
 68        if self.processed:
 69            return None
 70
 71        current_time = perf_counter()
 72        if self.start_time > current_time:
 73            self.start_time = current_time
 74        time_delta = current_time - self.start_time
 75        if time_delta >= self.requested_time:
 76            return 0
 77        else:
 78            return self.requested_time - time_delta
 79
 80
 81class Timer:
 82    # TODO: improve algorithm using `cengal\time_management\repeat_for_a_time` module
 83    def __init__(self):
 84        self.requests: Set[TimerRequest] = set()  # type: Set[TimerRequest]
 85
 86    def register(self, timer_handler: TimerHandler, desired_time: TimeInSeconds) -> TimerRequest:
 87        request = TimerRequest(timer_handler, desired_time)
 88        self.requests.add(request)
 89        return request
 90
 91    def discard(self, timer_request: TimerRequest):
 92        try:
 93            self.requests.remove(timer_request)
 94            return True
 95        except KeyError:
 96            return False
 97
 98    def __call__(self):
 99        # processed_requests = set()
100        # current_requests = self.requests
101        # for request in current_requests:
102        #     if request():
103        #         processed_requests.add(request)
104
105        # if processed_requests:
106        #     self.requests -= processed_requests
107
108        # requests_buff = self.requests
109        # self.requests = type(requests_buff)()
110        # for request in requests_buff:
111        #     if not request():
112        #         self.requests.add(request)
113
114        requests_buff = self.requests
115        self.requests = set()
116        processed_requests = set()
117        try:
118            for request in requests_buff:
119                request_result: bool = True
120                try:
121                    request_result = request()
122                finally:
123                    if request_result:
124                        processed_requests.add(request)
125        finally:
126            if processed_requests:
127                requests_buff -= processed_requests
128
129            if self.requests:
130                requests_buff.update(self.requests)
131
132            self.requests = requests_buff
133
134    def nearest_event(self) -> Optional[TimeInSeconds]:
135        nearest_event_time = None
136        for request in self.requests:
137            time_left = request.time_left()
138            if time_left is None:
139                continue
140
141            if 0 == time_left:
142                return 0
143
144            if (nearest_event_time is None) or (nearest_event_time > time_left):
145                nearest_event_time = time_left
146
147        return nearest_event_time

TimerHandler = typing.Callable

TimeInSeconds = typing.Union[int, float]

class TimerRequest: View Source

49class TimerRequest:
50    def __init__(self, timer_handler: TimerHandler, desired_time: TimeInSeconds):
51        self.timer_handler = timer_handler
52        self.requested_time = desired_time
53        self.start_time = perf_counter()
54        self.real_end_time = None
55        self.processed = False
56
57    def __call__(self) -> bool:
58        if not self.processed:
59            current_time = perf_counter()
60            if self.start_time > current_time:
61                self.start_time = current_time
62            time_delta = current_time - self.start_time
63            if time_delta >= self.requested_time:
64                self.timer_handler()
65                self.processed = True
66        return self.processed
67
68    def time_left(self) -> Optional[TimeInSeconds]:
69        if self.processed:
70            return None
71
72        current_time = perf_counter()
73        if self.start_time > current_time:
74            self.start_time = current_time
75        time_delta = current_time - self.start_time
76        if time_delta >= self.requested_time:
77            return 0
78        else:
79            return self.requested_time - time_delta

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

50    def __init__(self, timer_handler: TimerHandler, desired_time: TimeInSeconds):
51        self.timer_handler = timer_handler
52        self.requested_time = desired_time
53        self.start_time = perf_counter()
54        self.real_end_time = None
55        self.processed = False

timer_handler

requested_time

start_time

real_end_time

processed

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

68    def time_left(self) -> Optional[TimeInSeconds]:
69        if self.processed:
70            return None
71
72        current_time = perf_counter()
73        if self.start_time > current_time:
74            self.start_time = current_time
75        time_delta = current_time - self.start_time
76        if time_delta >= self.requested_time:
77            return 0
78        else:
79            return self.requested_time - time_delta

class Timer: View Source

 82class Timer:
 83    # TODO: improve algorithm using `cengal\time_management\repeat_for_a_time` module
 84    def __init__(self):
 85        self.requests: Set[TimerRequest] = set()  # type: Set[TimerRequest]
 86
 87    def register(self, timer_handler: TimerHandler, desired_time: TimeInSeconds) -> TimerRequest:
 88        request = TimerRequest(timer_handler, desired_time)
 89        self.requests.add(request)
 90        return request
 91
 92    def discard(self, timer_request: TimerRequest):
 93        try:
 94            self.requests.remove(timer_request)
 95            return True
 96        except KeyError:
 97            return False
 98
 99    def __call__(self):
100        # processed_requests = set()
101        # current_requests = self.requests
102        # for request in current_requests:
103        #     if request():
104        #         processed_requests.add(request)
105
106        # if processed_requests:
107        #     self.requests -= processed_requests
108
109        # requests_buff = self.requests
110        # self.requests = type(requests_buff)()
111        # for request in requests_buff:
112        #     if not request():
113        #         self.requests.add(request)
114
115        requests_buff = self.requests
116        self.requests = set()
117        processed_requests = set()
118        try:
119            for request in requests_buff:
120                request_result: bool = True
121                try:
122                    request_result = request()
123                finally:
124                    if request_result:
125                        processed_requests.add(request)
126        finally:
127            if processed_requests:
128                requests_buff -= processed_requests
129
130            if self.requests:
131                requests_buff.update(self.requests)
132
133            self.requests = requests_buff
134
135    def nearest_event(self) -> Optional[TimeInSeconds]:
136        nearest_event_time = None
137        for request in self.requests:
138            time_left = request.time_left()
139            if time_left is None:
140                continue
141
142            if 0 == time_left:
143                return 0
144
145            if (nearest_event_time is None) or (nearest_event_time > time_left):
146                nearest_event_time = time_left
147
148        return nearest_event_time

requests: Set[TimerRequest]

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

87    def register(self, timer_handler: TimerHandler, desired_time: TimeInSeconds) -> TimerRequest:
88        request = TimerRequest(timer_handler, desired_time)
89        self.requests.add(request)
90        return request

def discard( self, timer_request: TimerRequest): View Source

92    def discard(self, timer_request: TimerRequest):
93        try:
94            self.requests.remove(timer_request)
95            return True
96        except KeyError:
97            return False

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

135    def nearest_event(self) -> Optional[TimeInSeconds]:
136        nearest_event_time = None
137        for request in self.requests:
138            time_left = request.time_left()
139            if time_left is None:
140                continue
141
142            if 0 == time_left:
143                return 0
144
145            if (nearest_event_time is None) or (nearest_event_time > time_left):
146                nearest_event_time = time_left
147
148        return nearest_event_time