mirror of
https://github.com/chibicitiberiu/ytsm.git
synced 2024-02-24 05:43:31 +00:00
84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
from typing import Callable, List, Any, Optional
|
|
|
|
|
|
class ProgressTracker(object):
|
|
"""
|
|
Class which helps keep track of complex operation progress.
|
|
"""
|
|
|
|
def __init__(self, total_steps: float = 100, initial_steps: float = 0,
|
|
listener: Callable[[float, str], None] = None,
|
|
listener_args: List[Any] = None,
|
|
parent: Optional["ProgressTracker"] = None):
|
|
"""
|
|
Constructor
|
|
:param total_steps: Total number of steps required by this operation
|
|
:param initial_steps: Starting steps
|
|
:param parent: Parent progress tracker
|
|
:param listener: Callable which is called when any progress happens
|
|
"""
|
|
|
|
self.total_steps = total_steps
|
|
self.steps = initial_steps
|
|
|
|
self.__subtask: ProgressTracker = None
|
|
self.__subtask_steps = 0
|
|
|
|
self.__parent = parent
|
|
self.__listener = listener
|
|
self.__listener_args = listener_args or []
|
|
|
|
def __on_progress(self, progress_msg):
|
|
if self.__listener is not None:
|
|
self.__listener(*self.__listener_args, self.compute_progress(), progress_msg)
|
|
|
|
if self.__parent is not None:
|
|
self.__parent.__on_progress(progress_msg)
|
|
|
|
def advance(self, steps: float = 1, progress_msg: str = ''):
|
|
"""
|
|
Advances a number of steps.
|
|
:param steps: Number of steps to advance
|
|
:param progress_msg: A message which will be passed to a listener
|
|
:return:
|
|
"""
|
|
|
|
# We can assume previous subtask is now completed
|
|
if self.__subtask is not None:
|
|
self.steps += self.__subtask_steps
|
|
self.__subtask = None
|
|
|
|
self.steps += steps
|
|
self.__on_progress(progress_msg)
|
|
|
|
def subtask(self, steps: float = 1, subtask_total_steps: float = 100, subtask_initial_steps: float = 0):
|
|
"""
|
|
Creates a 'subtask' which has its own progress, which will be used in the calculation of the final progress.
|
|
:param steps: Number of steps the subtask is 'worth'
|
|
:param subtask_total_steps: Total number of steps for subtask
|
|
:param subtask_initial_steps: Initial steps for subtask
|
|
:return: ProgressTracker for subtask
|
|
"""
|
|
|
|
# We can assume previous subtask is now completed
|
|
if self.__subtask is not None:
|
|
self.steps += self.__subtask_steps
|
|
|
|
self.__subtask = ProgressTracker(total_steps=subtask_total_steps,
|
|
initial_steps=subtask_initial_steps,
|
|
parent=self)
|
|
self.__subtask_steps = steps
|
|
|
|
return self.__subtask
|
|
|
|
def compute_progress(self):
|
|
"""
|
|
Calculates final progress value in percent.
|
|
:return: value in [0,1] interval representing progress
|
|
"""
|
|
base = float(self.steps) / self.total_steps
|
|
if self.__subtask is not None:
|
|
base += self.__subtask.compute_progress() * self.__subtask_steps / self.total_steps
|
|
|
|
return min(base, 1.0)
|