cengal.time_management.sleep_tools.versions.v_0.sleep_tools

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__all__ = ['get_min_sleep_interval', 'get_usable_min_sleep_interval', 'get_countable_delta_time',
 20           'ensure_waitable_time_or_smaller', 'ensure_waitable_time_or_bigger', 'try_sleep', 'sleep', 'atry_sleep']
 21
 22
 23"""
 24Module Docstring
 25Docstrings: http://www.python.org/dev/peps/pep-0257/
 26"""
 27
 28__author__ = "ButenkoMS <gtalk@butenkoms.space>"
 29__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
 30__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
 31__license__ = "Apache License, Version 2.0"
 32__version__ = "4.4.1"
 33__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
 34__email__ = "gtalk@butenkoms.space"
 35# __status__ = "Prototype"
 36__status__ = "Development"
 37# __status__ = "Production"
 38
 39import os
 40from typing import Optional, Union
 41
 42from cengal.math.numbers import RationalNumber
 43from cengal.code_flow_control.args_manager import EntityArgsHolder
 44from cengal.time_management.high_precision_sync_sleep import hps_sleep
 45from cengal.system import IS_INSIDE_OR_FOR_WEB_BROWSER
 46
 47if IS_INSIDE_OR_FOR_WEB_BROWSER:
 48    default_sleep = hps_sleep
 49else:
 50    from time import sleep as default_sleep
 51
 52
 53if IS_INSIDE_OR_FOR_WEB_BROWSER:
 54    _default_min_sleep_interval = 0.001
 55elif 'nt' == os.name:
 56    _default_min_sleep_interval = 0.0156
 57else:
 58    _default_min_sleep_interval = 0.0001
 59
 60
 61_countable_delta_time = 0.0000001
 62
 63
 64def get_min_sleep_interval():
 65    return _default_min_sleep_interval
 66
 67
 68def get_usable_min_sleep_interval():
 69    return _default_min_sleep_interval + _countable_delta_time
 70
 71
 72def get_countable_delta_time():
 73    return _countable_delta_time
 74
 75
 76def ensure_waitable_time_or_smaller(secs: RationalNumber) -> RationalNumber:
 77    secs = abs(secs)
 78    intervals: int = secs // _default_min_sleep_interval
 79    if not intervals:
 80        return 0
 81
 82    return _default_min_sleep_interval * intervals + _countable_delta_time
 83
 84
 85def ensure_waitable_time_or_bigger(secs: RationalNumber) -> RationalNumber:
 86    if secs < get_usable_min_sleep_interval():
 87        return 0
 88
 89    intervals: int = secs // _default_min_sleep_interval
 90    if secs % _default_min_sleep_interval:
 91        intervals += 1
 92
 93    return _default_min_sleep_interval * intervals + _countable_delta_time
 94
 95
 96def try_sleep(secs: RationalNumber, max_secs: Optional[RationalNumber], sleep_func, *args, **kwargs):
 97    result = None
 98    usable_min_sleep_interval = get_usable_min_sleep_interval()
 99
100    if max_secs is not None:
101        if usable_min_sleep_interval > max_secs:
102            max_secs = usable_min_sleep_interval
103
104        if secs > max_secs:
105            secs = max_secs
106
107    secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
108    if secs >= _default_min_sleep_interval:
109        still_trying = True
110        while still_trying:
111            secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
112
113            if secs < usable_min_sleep_interval:
114                secs = usable_min_sleep_interval
115
116            try:
117                # print(f'Sleepting: {secs}...')
118                result = sleep_func(secs, *args, **kwargs)
119                # print(f'Awaken!')
120            except OverflowError:
121                still_trying = True
122                secs /= 2
123            else:
124                still_trying = False
125
126    return result, secs
127
128
129def sleep(secs: Optional[RationalNumber] = None, high_cpu_utilisation_mode: bool = False, max_secs: Optional[RationalNumber] = None, sleep_func: Optional[EntityArgsHolder] = None):
130    result = None
131    usable_min_sleep_interval = get_usable_min_sleep_interval()
132    if secs is None:
133        secs = usable_min_sleep_interval
134
135    if max_secs is not None:
136        if usable_min_sleep_interval > max_secs:
137            max_secs = usable_min_sleep_interval
138
139        if secs > max_secs:
140            secs = max_secs
141
142    secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
143    if secs >= _default_min_sleep_interval:
144        still_trying = True
145        while still_trying:
146            secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
147
148            if secs < usable_min_sleep_interval:
149                secs = usable_min_sleep_interval
150
151            try:
152                # print(f'Sleepting: {secs}...')
153                if sleep_func is None:
154                    result = default_sleep(secs)
155                else:
156                    sleep_func, args, kwargs = sleep_func.entity_args_kwargs()
157                    result = sleep_func(secs, *args, **kwargs)
158                # print(f'Awaken!')
159            except OverflowError:
160                still_trying = True
161                secs /= 2
162            else:
163                still_trying = False
164    else:
165        if high_cpu_utilisation_mode and (not IS_INSIDE_OR_FOR_WEB_BROWSER):
166            result = hps_sleep(secs)
167
168    return result, secs
169
170
171async def atry_sleep(secs: RationalNumber, max_secs: Optional[RationalNumber], awaitable_sleep_func, *args, **kwargs):
172    result = None
173    usable_min_sleep_interval = get_usable_min_sleep_interval()
174
175    if max_secs is not None:
176        if usable_min_sleep_interval > max_secs:
177            max_secs = usable_min_sleep_interval
178
179        if secs > max_secs:
180            secs = max_secs
181
182    secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
183    if secs >= _default_min_sleep_interval:
184        still_trying = True
185        while still_trying:
186            secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
187
188            if secs < usable_min_sleep_interval:
189                secs = usable_min_sleep_interval
190
191            try:
192                result = await awaitable_sleep_func(secs, *args, **kwargs)
193            except OverflowError:
194                still_trying = True
195                secs /= 2
196            else:
197                still_trying = False
198
199    return result, secs

def get_min_sleep_interval(): View Source

65def get_min_sleep_interval():
66    return _default_min_sleep_interval

def get_usable_min_sleep_interval(): View Source

69def get_usable_min_sleep_interval():
70    return _default_min_sleep_interval + _countable_delta_time

def get_countable_delta_time(): View Source

73def get_countable_delta_time():
74    return _countable_delta_time

def ensure_waitable_time_or_smaller(secs: Union[int, float]) -> Union[int, float]: View Source

77def ensure_waitable_time_or_smaller(secs: RationalNumber) -> RationalNumber:
78    secs = abs(secs)
79    intervals: int = secs // _default_min_sleep_interval
80    if not intervals:
81        return 0
82
83    return _default_min_sleep_interval * intervals + _countable_delta_time

def ensure_waitable_time_or_bigger(secs: Union[int, float]) -> Union[int, float]: View Source

86def ensure_waitable_time_or_bigger(secs: RationalNumber) -> RationalNumber:
87    if secs < get_usable_min_sleep_interval():
88        return 0
89
90    intervals: int = secs // _default_min_sleep_interval
91    if secs % _default_min_sleep_interval:
92        intervals += 1
93
94    return _default_min_sleep_interval * intervals + _countable_delta_time

def try_sleep( secs: Union[int, float], max_secs: Union[int, float, NoneType], sleep_func, *args, **kwargs): View Source

 97def try_sleep(secs: RationalNumber, max_secs: Optional[RationalNumber], sleep_func, *args, **kwargs):
 98    result = None
 99    usable_min_sleep_interval = get_usable_min_sleep_interval()
100
101    if max_secs is not None:
102        if usable_min_sleep_interval > max_secs:
103            max_secs = usable_min_sleep_interval
104
105        if secs > max_secs:
106            secs = max_secs
107
108    secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
109    if secs >= _default_min_sleep_interval:
110        still_trying = True
111        while still_trying:
112            secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
113
114            if secs < usable_min_sleep_interval:
115                secs = usable_min_sleep_interval
116
117            try:
118                # print(f'Sleepting: {secs}...')
119                result = sleep_func(secs, *args, **kwargs)
120                # print(f'Awaken!')
121            except OverflowError:
122                still_trying = True
123                secs /= 2
124            else:
125                still_trying = False
126
127    return result, secs

def sleep( secs: Union[int, float, NoneType] = None, high_cpu_utilisation_mode: bool = False, max_secs: Union[int, float, NoneType] = None, sleep_func: Union[cengal.code_flow_control.args_manager.versions.v_0.args_manager.EntityArgsHolder, NoneType] = None): View Source

130def sleep(secs: Optional[RationalNumber] = None, high_cpu_utilisation_mode: bool = False, max_secs: Optional[RationalNumber] = None, sleep_func: Optional[EntityArgsHolder] = None):
131    result = None
132    usable_min_sleep_interval = get_usable_min_sleep_interval()
133    if secs is None:
134        secs = usable_min_sleep_interval
135
136    if max_secs is not None:
137        if usable_min_sleep_interval > max_secs:
138            max_secs = usable_min_sleep_interval
139
140        if secs > max_secs:
141            secs = max_secs
142
143    secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
144    if secs >= _default_min_sleep_interval:
145        still_trying = True
146        while still_trying:
147            secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
148
149            if secs < usable_min_sleep_interval:
150                secs = usable_min_sleep_interval
151
152            try:
153                # print(f'Sleepting: {secs}...')
154                if sleep_func is None:
155                    result = default_sleep(secs)
156                else:
157                    sleep_func, args, kwargs = sleep_func.entity_args_kwargs()
158                    result = sleep_func(secs, *args, **kwargs)
159                # print(f'Awaken!')
160            except OverflowError:
161                still_trying = True
162                secs /= 2
163            else:
164                still_trying = False
165    else:
166        if high_cpu_utilisation_mode and (not IS_INSIDE_OR_FOR_WEB_BROWSER):
167            result = hps_sleep(secs)
168
169    return result, secs

async def atry_sleep( secs: Union[int, float], max_secs: Union[int, float, NoneType], awaitable_sleep_func, *args, **kwargs): View Source

172async def atry_sleep(secs: RationalNumber, max_secs: Optional[RationalNumber], awaitable_sleep_func, *args, **kwargs):
173    result = None
174    usable_min_sleep_interval = get_usable_min_sleep_interval()
175
176    if max_secs is not None:
177        if usable_min_sleep_interval > max_secs:
178            max_secs = usable_min_sleep_interval
179
180        if secs > max_secs:
181            secs = max_secs
182
183    secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
184    if secs >= _default_min_sleep_interval:
185        still_trying = True
186        while still_trying:
187            secs = _default_min_sleep_interval * (secs // _default_min_sleep_interval) + _countable_delta_time
188
189            if secs < usable_min_sleep_interval:
190                secs = usable_min_sleep_interval
191
192            try:
193                result = await awaitable_sleep_func(secs, *args, **kwargs)
194            except OverflowError:
195                still_trying = True
196                secs /= 2
197            else:
198                still_trying = False
199
200    return result, secs