"""Convert a string to a lambda or return the callable."""

if callable(input_value):

return input_value

"""Create a marker function that wraps the marker string if needed."""

if callable(marker):

return marker

if marker_wrap:

return lambda progress, data: marker_wrap % marker

"""Convert a wrapper tuple or format string to a format string.

if not wrapper:

return None

elif isinstance(wrapper, tuple):

return wrapper[0] + '{}' + wrapper[1]

else:

"""Wrap the output of a function in a template string or a tuple with

wrapper_format = create_wrapper(wrapper_)

def wrap_text(progress, data):

text = function(progress, data)

return wrapper_format.format(text) if wrapper_format else text

"""Mixin to format widgets using a formatstring.

def __init__(self, format: str, new_style: bool=False, **kwargs):

self.new_style = new_style

self.format = format

def __call__(self, progress: ProgressBarMixinBase, data: Data, format: types.Optional[str]=None) -> str:

"""Formats the widget into a string."""

format_ = self.get_format(progress, data, format)

try:

if self.new_style:

return format_.format(**data)

else:

return format_ % data

except (TypeError, KeyError):

logger.exception('Error while formatting %r with data: %r', format_, data)

"""Mixing to make sure widgets are only visible if the screen is within a

def __init__(self, min_width=None, max_width=None, **kwargs):

self.min_width = min_width

"""The base class for all widgets.

copy = True

@abc.abstractmethod

def __call__(self, progress: ProgressBarMixinBase, data: Data) -> str:

"""Updates the widget.

_fixed_colors: ClassVar[TFixedColors] = TFixedColors(fg_none=None, bg_none=None)

_gradient_colors: ClassVar[TGradientColors] = TGradientColors(fg=None, bg=None)

_len: typing.Callable[[str | bytes], int] = len

@property

def uses_colors(self) -> bool:

"""Check if this widget uses colors."""

return bool(self._fixed_colors.get('fg_none') or self._fixed_colors.get('bg_none') or

self._gradient_colors.get('fg') or self._gradient_colors.get('bg'))

def __init__(self, *args, fixed_colors=None, gradient_colors=None, **kwargs):

if fixed_colors is not None:

self._fixed_colors.update(fixed_colors)

if gradient_colors is not None:

self._gradient_colors.update(gradient_colors)

if self.uses_colors:

self._len = utils.len_color

"""The base class for all variable width widgets.

@abc.abstractmethod

def __call__(self, progress: ProgressBarMixinBase, data: Data, width: int=0) -> str:

"""Updates the widget providing the total width the widget must fill.

"""The base class for all time sensitive widgets.

"""Displays a formatted label.

mapping: ClassVar[types.Dict[str, types.Tuple[str, types.Any]]] = dict(finished=('end_time', None), last_update=('last_update_time', None), max=('max_value', None), seconds=('seconds_elapsed', None), start=('start_time', None), elapsed=('total_seconds_elapsed', utils.format_time), value=('value', None))

def __init__(self, format: str, **kwargs):

FormatWidgetMixin.__init__(self, format=format, **kwargs)

WidgetBase.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, format: types.Optional[str]=None):

for name, (key, transform) in self.mapping.items():

with contextlib.suppress(KeyError, ValueError, IndexError):

if transform is None:

data[name] = data[key]

else:

data[name] = transform(data[key])

"""WidgetBase which displays the elapsed seconds."""

def __init__(self, format='Elapsed Time: %(elapsed)s', **kwargs):

if '%s' in format and '%(elapsed)s' not in format:

format = format.replace('%s', '%(elapsed)s')

FormatLabel.__init__(self, format=format, **kwargs)

TimeSensitiveWidgetBase.__init__(self, **kwargs)

"""

def __init__(self, samples=datetime.timedelta(seconds=2), key_prefix=None, **kwargs):

self.samples = samples

self.key_prefix = (key_prefix or self.__class__.__name__) + '_'

TimeSensitiveWidgetBase.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, delta: bool=False):

sample_times = self.get_sample_times(progress, data)

sample_values = self.get_sample_values(progress, data)

if sample_times:

sample_time = sample_times[-1]

else:

sample_time = datetime.datetime.min

if progress.last_update_time - sample_time > self.INTERVAL:

sample_times.append(progress.last_update_time)

sample_values.append(progress.value)

if isinstance(self.samples, datetime.timedelta):

minimum_time = progress.last_update_time - self.samples

minimum_value = sample_values[-1]

while sample_times[2:] and minimum_time > sample_times[1] and (minimum_value > sample_values[1]):

sample_times.pop(0)

sample_values.pop(0)

elif len(sample_times) > self.samples:

sample_times.pop(0)

sample_values.pop(0)

if delta:

if (delta_time := (sample_times[-1] - sample_times[0])):

delta_value = sample_values[-1] - sample_values[0]

return (delta_time, delta_value)

else:

return (None, None)

else:

"""WidgetBase which attempts to estimate the time of arrival."""

def __init__(self, format_not_started='ETA: --:--:--', format_finished='Time: %(elapsed)8s', format='ETA: %(eta)8s', format_zero='ETA: 00:00:00', format_na='ETA: N/A', **kwargs):

if '%s' in format and '%(eta)s' not in format:

format = format.replace('%s', '%(eta)s')

Timer.__init__(self, **kwargs)

self.format_not_started = format_not_started

self.format_finished = format_finished

self.format = format

self.format_zero = format_zero

self.format_NA = format_na

def _calculate_eta(self, progress: ProgressBarMixinBase, data: Data, value, elapsed):

"""Updates the widget to show the ETA or total time when finished."""

pass

def __call__(self, progress: ProgressBarMixinBase, data: Data, value=None, elapsed=None):

"""Updates the widget to show the ETA or total time when finished."""

if value is None:

value = data['value']

if elapsed is None:

elapsed = data['time_elapsed']

eta_na = False

try:

data['eta_seconds'] = self._calculate_eta(progress, data, value=value, elapsed=elapsed)

except TypeError:

data['eta_seconds'] = None

eta_na = True

data['eta'] = None

if data['eta_seconds']:

with contextlib.suppress(ValueError, OverflowError):

data['eta'] = utils.format_time(data['eta_seconds'])

if data['value'] == progress.min_value:

fmt = self.format_not_started

elif progress.end_time:

fmt = self.format_finished

elif data['eta']:

fmt = self.format

elif eta_na:

fmt = self.format_NA

else:

fmt = self.format_zero

"""Widget which attempts to estimate the absolute time of arrival."""

def __init__(self, format_not_started='Estimated finish time: ----/--/-- --:--:--', format_finished='Finished at: %(elapsed)s', format='Estimated finish time: %(eta)s', **kwargs):

"""WidgetBase which attempts to estimate the time of arrival.

exponential_smoothing: bool

exponential_smoothing_factor: float

def __init__(self, exponential_smoothing=True, exponential_smoothing_factor=0.1, **kwargs):

self.exponential_smoothing = exponential_smoothing

self.exponential_smoothing_factor = exponential_smoothing_factor

ETA.__init__(self, **kwargs)

SamplesMixin.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, value=None, elapsed=None):

elapsed, value = SamplesMixin.__call__(self, progress, data, delta=True)

if not elapsed:

value = None

elapsed = 0

"""

smoothing_algorithm: algorithms.SmoothingAlgorithm

smoothing_parameters: dict[str, float]

def __init__(self, smoothing_algorithm: type[algorithms.SmoothingAlgorithm]=algorithms.ExponentialMovingAverage, smoothing_parameters: dict[str, float] | None=None, **kwargs):

self.smoothing_parameters = smoothing_parameters or {}

self.smoothing_algorithm = smoothing_algorithm(**self.smoothing_parameters or {})

ETA.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, value=None, elapsed=None):

if value is None:

value = data['value']

if elapsed is None:

elapsed = data['time_elapsed']

self.smoothing_algorithm.update(value, elapsed)

"""

def __init__(self, variable='value', format='%(scaled)5.1f %(prefix)s%(unit)s', unit='B', prefixes=('', 'Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei', 'Zi', 'Yi'), **kwargs):

self.variable = variable

self.unit = unit

self.prefixes = prefixes

FormatWidgetMixin.__init__(self, format=format, **kwargs)

WidgetBase.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, format: types.Optional[str]=None):

value = data[self.variable]

if value is not None:

scaled, power = utils.scale_1024(value, len(self.prefixes))

else:

scaled = power = 0

data['scaled'] = scaled

data['prefix'] = self.prefixes[power]

data['unit'] = self.unit

"""

def __init__(self, format='%(scaled)5.1f %(prefix)s%(unit)-s/s', inverse_format='%(scaled)5.1f s/%(prefix)s%(unit)-s', unit='B', prefixes=('', 'Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei', 'Zi', 'Yi'), **kwargs):

self.unit = unit

self.prefixes = prefixes

self.inverse_format = inverse_format

FormatWidgetMixin.__init__(self, format=format, **kwargs)

TimeSensitiveWidgetBase.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data, value=None, total_seconds_elapsed=None):

"""Updates the widget with the current SI prefixed speed."""

if value is None:

value = data['value']

elapsed = utils.deltas_to_seconds(total_seconds_elapsed, data['total_seconds_elapsed'])

if value is not None and elapsed is not None and (elapsed > 2e-06) and (value > 2e-06):

scaled, power = self._speed(value, elapsed)

else:

scaled = power = 0

data['unit'] = self.unit

if power == 0 and scaled < 0.1:

if scaled > 0:

scaled = 1 / scaled

data['scaled'] = scaled

data['prefix'] = self.prefixes[0]

return FormatWidgetMixin.__call__(self, progress, data, self.inverse_format)

else:

data['scaled'] = scaled

data['prefix'] = self.prefixes[power]

"""Widget for showing the transfer speed based on the last X samples."""

def __init__(self, **kwargs):

FileTransferSpeed.__init__(self, **kwargs)

SamplesMixin.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data, value=None, total_seconds_elapsed=None):

elapsed, value = SamplesMixin.__call__(self, progress, data, delta=True)

"""An animated marker for the progress bar which defaults to appear as if

def __init__(self, markers='|/-\\', default=None, fill='', marker_wrap=None, fill_wrap=None, **kwargs):

self.markers = markers

self.marker_wrap = create_wrapper(marker_wrap)

self.default = default or markers[0]

self.fill_wrap = create_wrapper(fill_wrap)

self.fill = create_marker(fill, self.fill_wrap) if fill else None

WidgetBase.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, width=None):

"""Updates the widget to show the next marker or the first marker when

if progress.end_time:

return self.default

marker = self.markers[data['updates'] % len(self.markers)]

if self.marker_wrap:

marker = self.marker_wrap.format(marker)

if self.fill:

fill = self.fill(progress, data, width - progress.custom_len(marker))

else:

fill = ''

if isinstance(marker, int):

marker = bytes(marker)

fill = fill.encode()

else:

fill = type(marker)(fill)

"""Displays the current count."""

def __init__(self, format='%(value)d', **kwargs):

FormatWidgetMixin.__init__(self, format=format, **kwargs)

WidgetBase.__init__(self, format=format, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, format=None):

"""Displays the current percentage as a number with a percent sign."""

def __init__(self, format='%(percentage)3d%%', na='N/A%%', **kwargs):

self.na = na

FormatWidgetMixin.__init__(self, format=format, **kwargs)

"""Returns progress as a count of the total (e.g.: "5 of 47")."""

max_width_cache: dict[types.Union[str, tuple[float, float | types.Type[base.UnknownLength]]], types.Optional[int]]

DEFAULT_FORMAT = '%(value_s)s of %(max_value_s)s'

def __init__(self, format=DEFAULT_FORMAT, **kwargs):

FormatWidgetMixin.__init__(self, format=format, **kwargs)

WidgetBase.__init__(self, format=format, **kwargs)

self.max_width_cache = dict(default=self.max_width or 0)

def __call__(self, progress: ProgressBarMixinBase, data: Data, format=None):

if data.get('max_value'):

data['max_value_s'] = data['max_value']

else:

data['max_value_s'] = 'N/A'

if data.get('value'):

data['value_s'] = data['value']

else:

data['value_s'] = 0

formatted = FormatWidgetMixin.__call__(self, progress, data, format=format)

key = (progress.min_value, progress.max_value)

max_width: types.Optional[int] = self.max_width_cache.get(key, self.max_width)

if not max_width:

temporary_data = data.copy()

for value in key:

if value is None:

continue

temporary_data['value'] = value

if (width := progress.custom_len(FormatWidgetMixin.__call__(self, progress, temporary_data, format=format))):

max_width = max(max_width or 0, width)

self.max_width_cache[key] = max_width

if max_width:

formatted = formatted.rjust(max_width)

"""A progress bar which stretches to fill the line."""

fg: terminal.OptionalColor | None = colors.gradient

bg: terminal.OptionalColor | None = None

def __init__(self, marker='#', left='|', right='|', fill=' ', fill_left=True, marker_wrap=None, **kwargs):

"""Creates a customizable progress bar.

self.marker = create_marker(marker, marker_wrap)

self.left = string_or_lambda(left)

self.right = string_or_lambda(right)

self.fill = string_or_lambda(fill)

self.fill_left = fill_left

AutoWidthWidgetBase.__init__(self, **kwargs)

def _apply_colors(self, text: str, data: Data) -> str:

"""Apply colors to the text based on the progress percentage."""

percentage = data.get('percentage')

if percentage is None:

return text

return terminal.apply_colors(text, percentage / 100.0,

fg=self.fg, bg=self.bg,

**self._fixed_colors)

def __call__(self, progress: ProgressBarMixinBase, data: Data, width: int=0, color=True):

"""Updates the progress bar and its subcomponents."""

left = converters.to_unicode(self.left(progress, data, width))

right = converters.to_unicode(self.right(progress, data, width))

width -= progress.custom_len(left) + progress.custom_len(right)

marker = converters.to_unicode(self.marker(progress, data, width))

fill = converters.to_unicode(self.fill(progress, data, width))

width += len(marker) - progress.custom_len(marker)

if self.fill_left:

marker = marker.ljust(width, fill)

else:

marker = marker.rjust(width, fill)

if color:

marker = self._apply_colors(marker, data)

"""A bar which has a marker that goes from right to left."""

def __init__(self, marker='#', left='|', right='|', fill=' ', fill_left=False, **kwargs):

"""Creates a customizable progress bar.

"""A bar which has a marker which bounces from side to side."""

INTERVAL = datetime.timedelta(milliseconds=100)

def __call__(self, progress: ProgressBarMixinBase, data: Data, width: int=0, color=True):

"""Updates the progress bar and its subcomponents."""

left = converters.to_unicode(self.left(progress, data, width))

right = converters.to_unicode(self.right(progress, data, width))

width -= progress.custom_len(left) + progress.custom_len(right)

marker = converters.to_unicode(self.marker(progress, data, width))

fill = converters.to_unicode(self.fill(progress, data, width))

if width:

value = int(data['total_seconds_elapsed'] / self.INTERVAL.total_seconds())

a = value % width

b = width - a - 1

if value % (width * 2) >= width:

a, b = (b, a)

if self.fill_left:

marker = a * fill + marker + b * fill

else:

marker = b * fill + marker + a * fill

mapping: types.Dict[str, types.Any] = dict()

copy = False

def __init__(self, format: str, mapping: types.Optional[types.Dict[str, types.Any]]=None, **kwargs):

self.format = format

self.mapping = mapping or self.mapping

FormatWidgetMixin.__init__(self, format=format, **kwargs)

WidgetBase.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, format: types.Optional[str]=None):

"""Mixin to display a custom user variable."""

def __init__(self, name, **kwargs):

if not isinstance(name, str):

raise TypeError('Variable(): argument must be a string')

if len(name.split()) > 1:

raise ValueError('Variable(): argument must be single word')

"""

def __init__(self, name, markers, **kwargs):

VariableMixin.__init__(self, name)

Bar.__init__(self, **kwargs)

self.markers = [string_or_lambda(marker) for marker in markers]

def __call__(self, progress: ProgressBarMixinBase, data: Data, width: int=0, color=True):

"""Updates the progress bar and its subcomponents."""

left = converters.to_unicode(self.left(progress, data, width))

right = converters.to_unicode(self.right(progress, data, width))

width -= progress.custom_len(left) + progress.custom_len(right)

values = self.get_values(progress, data)

values_sum = sum(values)

if width and values_sum:

middle = ''

values_accumulated = 0

width_accumulated = 0

for marker, value in zip(self.markers, values):

marker = converters.to_unicode(marker(progress, data, width))

assert progress.custom_len(marker) == 1

values_accumulated += value

item_width = int(values_accumulated / values_sum * width)

item_width -= width_accumulated

width_accumulated += item_width

middle += item_width * marker

else:

fill = converters.to_unicode(self.fill(progress, data, width))

assert progress.custom_len(fill) == 1

middle = fill * width

smooth = ' ▏▎▍▌▋▊▉█'

bar = ' ▁▂▃▄▅▆▇█'

snake = ' ▖▌▛█'

fade_in = ' ░▒▓█'

dots = ' ⡀⡄⡆⡇⣇⣧⣷⣿'

"""A progressbar that can display progress at a sub-character granularity

def __init__(self, markers=GranularMarkers.smooth, left='|', right='|', **kwargs):

"""Creates a customizable progress bar.

self.markers = markers

self.left = string_or_lambda(left)

self.right = string_or_lambda(right)

AutoWidthWidgetBase.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, width: int=0):

left = converters.to_unicode(self.left(progress, data, width))

right = converters.to_unicode(self.right(progress, data, width))

width -= progress.custom_len(left) + progress.custom_len(right)

max_value = progress.max_value

if max_value is not base.UnknownLength and max_value > 0:

percent = progress.value / max_value

else:

percent = 0

num_chars = percent * width

marker = self.markers[-1] * int(num_chars)

if (marker_idx := int(num_chars % 1 * (len(self.markers) - 1))):

marker += self.markers[marker_idx]

marker = converters.to_unicode(marker)

width += len(marker) - progress.custom_len(marker)

marker = marker.ljust(width, self.markers[0])

"""A bar which has a formatted label in the center."""

def __init__(self, format, **kwargs):

FormatLabel.__init__(self, format, **kwargs)

Bar.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, width: int=0, format: FormatString=None):

center = FormatLabel.__call__(self, progress, data, format=format)

bar = Bar.__call__(self, progress, data, width, color=False)

center_len = progress.custom_len(center)

center_left = int((width - center_len) / 2)

center_right = center_left + center_len

"""A bar which displays the current percentage in the center."""

def __init__(self, format='%(percentage)2d%%', na='N/A%%', **kwargs):

Percentage.__init__(self, format, na=na, **kwargs)

FormatLabelBar.__init__(self, format, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, width: int=0, format: FormatString=None):

"""Displays a custom variable."""

def __init__(self, name, format='{name}: {formatted_value}', width=6, precision=3, **kwargs):

"""Creates a Variable associated with the given name."""

self.format = format

self.width = width

self.precision = precision

VariableMixin.__init__(self, name=name)

WidgetBase.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, format: types.Optional[str]=None):

value = data['variables'][self.name]

context = data.copy()

context['value'] = value

context['name'] = self.name

context['width'] = self.width

context['precision'] = self.precision

try:

value = float(value)

fmt = '{value:{width}.{precision}}'

context['formatted_value'] = fmt.format(**context)

except (TypeError, ValueError):

if value:

context['formatted_value'] = '{value:{width}}'.format(**context)

else:

context['formatted_value'] = '-' * self.width

"""Widget which displays the current (date)time with seconds resolution."""

INTERVAL = datetime.timedelta(seconds=1)

def __init__(self, format='Current Time: %(current_time)s', microseconds=False, **kwargs):

self.microseconds = microseconds

FormatWidgetMixin.__init__(self, format=format, **kwargs)

TimeSensitiveWidgetBase.__init__(self, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, format: types.Optional[str]=None):

data['current_time'] = self.current_time()

data['current_datetime'] = self.current_datetime()

"""

success_fg_color: terminal.Color | None = colors.green

success_bg_color: terminal.Color | None = None

success_marker: str = '█'

failure_fg_color: terminal.Color | None = colors.red

failure_bg_color: terminal.Color | None = None

failure_marker: str = 'X'

job_markers: list[str]

def __init__(self, name: str, left='|', right='|', fill=' ', fill_left=True, success_fg_color=colors.green, success_bg_color=None, success_marker='█', failure_fg_color=colors.red, failure_bg_color=None, failure_marker='X', **kwargs):

VariableMixin.__init__(self, name)

self.name = name

self.job_markers = []

self.left = string_or_lambda(left)

self.right = string_or_lambda(right)

self.fill = string_or_lambda(fill)

self.success_fg_color = success_fg_color

self.success_bg_color = success_bg_color

self.success_marker = success_marker

self.failure_fg_color = failure_fg_color

self.failure_bg_color = failure_bg_color

self.failure_marker = failure_marker

Bar.__init__(self, left=left, right=right, fill=fill, fill_left=fill_left, **kwargs)

def __call__(self, progress: ProgressBarMixinBase, data: Data, width: int=0, color=True):

left = converters.to_unicode(self.left(progress, data, width))

right = converters.to_unicode(self.right(progress, data, width))

width -= progress.custom_len(left) + progress.custom_len(right)

status: str | bool | None = data['variables'].get(self.name)

if width and status is not None:

if status is True:

marker = self.success_marker

fg_color = self.success_fg_color

bg_color = self.success_bg_color

elif status is False:

marker = self.failure_marker

fg_color = self.failure_fg_color

bg_color = self.failure_bg_color

else:

marker = status

fg_color = bg_color = None

marker = converters.to_unicode(marker)

if fg_color:

marker = fg_color.fg(marker)

if bg_color:

marker = bg_color.bg(marker)

self.job_markers.append(marker)

marker = ''.join(self.job_markers)

width -= progress.custom_len(marker)

fill = converters.to_unicode(self.fill(progress, data, width))

fill = self._apply_colors(fill * width, data)

if self.fill_left:

marker += fill

else: