cengal.user_interface.console.progress_meter.versions.v_0.progress_meter
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 18from cengal.time_management.repeat_for_a_time import BaseTracer, GreedyTracer 19from cengal.time_management.cpu_clock_cycles import perf_counter 20from enum import Enum 21from typing import Tuple 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 39 40class ViewIndicator: 41 def __init__(self): 42 self._view_set = ('-', '\\', '|', '/') 43 self._view_set_size = len(self._view_set) 44 self._counter = 0 45 46 def __call__(self): 47 result = self._view_set[self._counter] 48 self._counter += 1 49 if self._counter >= self._view_set_size: 50 self._counter = 0 51 return result 52 53 @property 54 def view_set(self): 55 return self._view_set 56 57 @view_set.setter 58 def view_set(self, view_set: Tuple): 59 self._view_set = view_set 60 self._view_set_size = len(self._view_set) 61 self._counter = 0 62 63 64class ProgressOutType(Enum): 65 none = 0 66 short = 1 67 full = 2 68 69 70class ProgressMeter: 71 """ 72 Example of use: 73 pm = ProgressMeter(0.1, 'files') 74 while True: 75 ... 76 pm(ProgressOutType.short) 77 pm(ProgressOutType.full, force_print=True, erasable=False) 78 79 """ 80 81 def __init__(self, update_time: float=1.0, _items_readable_name: str='iterations' 82 , default_indicator: ViewIndicator=None 83 , default_tracer: BaseTracer=GreedyTracer): 84 self._update_time = update_time 85 self._items_readable_name = _items_readable_name 86 self._last_iterations_per_second = 0.0 87 self.template_short = ' {indicator}' 88 self.template_full = ' {indicator} ## {items_count} ## {items_readable_name} per second: ' \ 89 '{iterations_per_second}' 90 self._view_indicator = default_indicator or ViewIndicator() 91 self._default_tracer = default_tracer 92 self._tracer = self._default_tracer(self._update_time) 93 self._iterations = 0 94 self._start_time = 0.0 95 self._last_tracked_time = 0.0 96 self._is_first_call = True 97 98 def __call__(self, output_type: ProgressOutType=ProgressOutType.full, 99 force_print: bool=False, erasable: bool=True) -> bool: 100 self._iterations += 1 101 if self._is_first_call: 102 self._is_first_call = False 103 self._start_time = perf_counter() 104 result = not self._tracer.iter() 105 if result or force_print: 106 self._last_tracked_time = perf_counter() 107 time_delta = self._last_tracked_time - self._start_time 108 if time_delta != 0.0: 109 self._last_iterations_per_second = self._iterations / time_delta 110 self.print(output_type, erasable) 111 self._tracer = self._default_tracer(self._update_time) 112 return result 113 114 def print(self, output_type: ProgressOutType=ProgressOutType.full, erasable: bool=True): 115 out_message = None 116 if ProgressOutType.full == output_type: 117 out_message = self.template_full.format(indicator=self._view_indicator(), 118 items_count=self._iterations, 119 items_readable_name=self._items_readable_name, 120 iterations_per_second=self._last_iterations_per_second) 121 elif ProgressOutType.short == output_type: 122 out_message = self.template_short.format(indicator=self._view_indicator()) 123 124 if out_message is not None: 125 if erasable: 126 print(out_message, end='\r', flush=True) 127 else: 128 print(out_message, flush=True) 129 130 @property 131 def iterations_per_second(self): 132 return self._last_iterations_per_second
class
ViewIndicator:
41class ViewIndicator: 42 def __init__(self): 43 self._view_set = ('-', '\\', '|', '/') 44 self._view_set_size = len(self._view_set) 45 self._counter = 0 46 47 def __call__(self): 48 result = self._view_set[self._counter] 49 self._counter += 1 50 if self._counter >= self._view_set_size: 51 self._counter = 0 52 return result 53 54 @property 55 def view_set(self): 56 return self._view_set 57 58 @view_set.setter 59 def view_set(self, view_set: Tuple): 60 self._view_set = view_set 61 self._view_set_size = len(self._view_set) 62 self._counter = 0
class
ProgressOutType(enum.Enum):
An enumeration.
none =
<ProgressOutType.none: 0>
short =
<ProgressOutType.short: 1>
full =
<ProgressOutType.full: 2>
Inherited Members
- enum.Enum
- name
- value
class
ProgressMeter:
71class ProgressMeter: 72 """ 73 Example of use: 74 pm = ProgressMeter(0.1, 'files') 75 while True: 76 ... 77 pm(ProgressOutType.short) 78 pm(ProgressOutType.full, force_print=True, erasable=False) 79 80 """ 81 82 def __init__(self, update_time: float=1.0, _items_readable_name: str='iterations' 83 , default_indicator: ViewIndicator=None 84 , default_tracer: BaseTracer=GreedyTracer): 85 self._update_time = update_time 86 self._items_readable_name = _items_readable_name 87 self._last_iterations_per_second = 0.0 88 self.template_short = ' {indicator}' 89 self.template_full = ' {indicator} ## {items_count} ## {items_readable_name} per second: ' \ 90 '{iterations_per_second}' 91 self._view_indicator = default_indicator or ViewIndicator() 92 self._default_tracer = default_tracer 93 self._tracer = self._default_tracer(self._update_time) 94 self._iterations = 0 95 self._start_time = 0.0 96 self._last_tracked_time = 0.0 97 self._is_first_call = True 98 99 def __call__(self, output_type: ProgressOutType=ProgressOutType.full, 100 force_print: bool=False, erasable: bool=True) -> bool: 101 self._iterations += 1 102 if self._is_first_call: 103 self._is_first_call = False 104 self._start_time = perf_counter() 105 result = not self._tracer.iter() 106 if result or force_print: 107 self._last_tracked_time = perf_counter() 108 time_delta = self._last_tracked_time - self._start_time 109 if time_delta != 0.0: 110 self._last_iterations_per_second = self._iterations / time_delta 111 self.print(output_type, erasable) 112 self._tracer = self._default_tracer(self._update_time) 113 return result 114 115 def print(self, output_type: ProgressOutType=ProgressOutType.full, erasable: bool=True): 116 out_message = None 117 if ProgressOutType.full == output_type: 118 out_message = self.template_full.format(indicator=self._view_indicator(), 119 items_count=self._iterations, 120 items_readable_name=self._items_readable_name, 121 iterations_per_second=self._last_iterations_per_second) 122 elif ProgressOutType.short == output_type: 123 out_message = self.template_short.format(indicator=self._view_indicator()) 124 125 if out_message is not None: 126 if erasable: 127 print(out_message, end='\r', flush=True) 128 else: 129 print(out_message, flush=True) 130 131 @property 132 def iterations_per_second(self): 133 return self._last_iterations_per_second
Example of use: pm = ProgressMeter(0.1, 'files') while True: ... pm(ProgressOutType.short) pm(ProgressOutType.full, force_print=True, erasable=False)
ProgressMeter( update_time: float = 1.0, _items_readable_name: str = 'iterations', default_indicator: ViewIndicator = None, default_tracer: cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.BaseTracer = <class 'cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.GreedyTracer'>)
82 def __init__(self, update_time: float=1.0, _items_readable_name: str='iterations' 83 , default_indicator: ViewIndicator=None 84 , default_tracer: BaseTracer=GreedyTracer): 85 self._update_time = update_time 86 self._items_readable_name = _items_readable_name 87 self._last_iterations_per_second = 0.0 88 self.template_short = ' {indicator}' 89 self.template_full = ' {indicator} ## {items_count} ## {items_readable_name} per second: ' \ 90 '{iterations_per_second}' 91 self._view_indicator = default_indicator or ViewIndicator() 92 self._default_tracer = default_tracer 93 self._tracer = self._default_tracer(self._update_time) 94 self._iterations = 0 95 self._start_time = 0.0 96 self._last_tracked_time = 0.0 97 self._is_first_call = True
115 def print(self, output_type: ProgressOutType=ProgressOutType.full, erasable: bool=True): 116 out_message = None 117 if ProgressOutType.full == output_type: 118 out_message = self.template_full.format(indicator=self._view_indicator(), 119 items_count=self._iterations, 120 items_readable_name=self._items_readable_name, 121 iterations_per_second=self._last_iterations_per_second) 122 elif ProgressOutType.short == output_type: 123 out_message = self.template_short.format(indicator=self._view_indicator()) 124 125 if out_message is not None: 126 if erasable: 127 print(out_message, end='\r', flush=True) 128 else: 129 print(out_message, flush=True)