from __future__ import annotations
import abc
import collections.abc
import contextlib
import datetime
import functools
import logging
import typing
# Ruff is being stupid and doesn't understand `ClassVar` if it comes from the
# `types` module
from typing import ClassVar
from python_utils import containers, converters
from . import algorithms, base, terminal, utils
from .terminal import colors
if typing.TYPE_CHECKING:
from .bar import NumberT, ProgressBarMixinBase
logger = logging.getLogger(__name__)
MAX_DATE = datetime.date.max
MAX_TIME = datetime.time.max
MAX_DATETIME = datetime.datetime.max
Data = dict[str, typing.Any]
FormatString = str | None
T = typing.TypeVar('T')
[docs]
def string_or_lambda(
input_: str | collections.abc.Callable[..., str],
) -> collections.abc.Callable[..., str]:
if isinstance(input_, str):
def render_input(progress, data, width):
return input_ % data
return render_input
else:
return input_
[docs]
def create_wrapper(
wrapper: str | tuple[str | None, str | None] | None,
) -> str | None:
"""Convert a wrapper tuple or format string to a format string.
>>> create_wrapper('')
>>> print(create_wrapper('a{}b'))
a{}b
>>> print(create_wrapper(('a', 'b')))
a{}b
"""
if isinstance(wrapper, tuple) and len(wrapper) == 2:
a, b = wrapper
wrapper = (a or '') + '{}' + (b or '')
elif not wrapper:
return None
if isinstance(wrapper, str):
if '{}' not in wrapper:
raise ValueError('Expected string with {} for formatting')
else:
raise RuntimeError( # noqa: TRY004
'Pass either a begin/end string as a tuple or a template string '
'with `{}`',
)
return wrapper
[docs]
def wrapper(function, wrapper_):
"""Wrap the output of a function in a template string or a tuple with
begin/end strings.
"""
wrapper_ = create_wrapper(wrapper_)
if not wrapper_:
return function
@functools.wraps(function)
def wrap(*args: typing.Any, **kwargs: typing.Any):
return wrapper_.format(function(*args, **kwargs))
return wrap
[docs]
def create_marker(
marker: str | collections.abc.Callable[..., str],
wrap: str | tuple[str | None, str | None] | None = None,
) -> collections.abc.Callable[..., str]:
if isinstance(marker, str):
# Narrow to ``str`` once, in a fresh local, so the ``_marker`` closure
# below closes over a plain ``str`` (no cast needed). ``_marker`` is
# only ever wrapped in this branch.
marker_str = converters.to_unicode(marker)
if utils.len_color(marker_str) != 1:
raise ValueError('Markers are required to be 1 char')
def _marker(progress, data, width):
if (
progress.max_value is not base.UnknownLength
and progress.max_value > 0
):
# The fill length is based on the progress relative to
# min_value; the max() guards against a zero range and the
# min() keeps the marker within the allotted width when the
# value exceeds max_value (with max_error=False)
length = min(
width,
int(
(progress.value - progress.min_value)
/ max(progress.max_value - progress.min_value, 1e-6)
* width,
),
)
return marker_str * length
else:
return marker_str
return wrapper(_marker, wrap)
else:
return wrapper(marker, wrap)
class _WidgetKwargsSink:
"""Terminates cooperative ``__init__`` chains for widgets.
Absorbs keyword arguments no widget class consumed so a cooperative
chain never reaches ``object.__init__`` with leftovers. Tolerated
silently for backwards compatibility: third-party widgets have passed
stray kwargs to their parents for years.
"""
def __init__(self, **kwargs: typing.Any) -> None:
super().__init__()
[docs]
class TGradientColors(typing.TypedDict):
fg: terminal.OptionalColor
bg: terminal.OptionalColor
[docs]
class TFixedColors(typing.TypedDict):
fg_none: terminal.Color | None
bg_none: terminal.Color | None
[docs]
class Timer(FormatLabel, TimeSensitiveWidgetBase):
"""WidgetBase which displays the elapsed seconds."""
def __init__(
self, format='Elapsed Time: %(elapsed)s', **kwargs: typing.Any
):
# Backwards compatibility: very old configs used a bare ``%s``
# placeholder for the elapsed time. Silently rewrite it to the named
# ``%(elapsed)s`` form the widget actually formats with.
if '%s' in format and '%(elapsed)s' not in format:
format = format.replace('%s', '%(elapsed)s')
super().__init__(format=format, **kwargs)
# This is exposed as a static method for backwards compatibility
format_time = staticmethod(utils.format_time)
[docs]
class SamplesMixin(TimeSensitiveWidgetBase, metaclass=abc.ABCMeta):
"""
Mixing for widgets that average multiple measurements.
Note that samples can be either an integer or a timedelta to indicate a
certain amount of time
>>> class progress:
... last_update_time = datetime.datetime.now()
... value = 1
... extra = dict()
>>> samples = SamplesMixin(samples=2)
>>> samples(progress, None, True)
(None, None)
>>> progress.last_update_time += datetime.timedelta(seconds=1)
>>> samples(progress, None, True) == (datetime.timedelta(seconds=1), 0)
True
>>> progress.last_update_time += datetime.timedelta(seconds=1)
>>> samples(progress, None, True) == (datetime.timedelta(seconds=1), 0)
True
>>> samples = SamplesMixin(samples=datetime.timedelta(seconds=1))
>>> _, value = samples(progress, None)
>>> value
SliceableDeque([1, 1])
>>> samples(progress, None, True) == (datetime.timedelta(seconds=1), 0)
True
"""
def __init__(
self,
samples: datetime.timedelta | int = datetime.timedelta(seconds=2),
key_prefix=None,
**kwargs,
):
self.samples = samples
self.key_prefix = (key_prefix or self.__class__.__name__) + '_'
super().__init__(**kwargs)
[docs]
def get_sample_times(self, progress: ProgressBarMixinBase, data: Data):
return progress.extra.setdefault(
f'{self.key_prefix}sample_times',
containers.SliceableDeque(),
)
[docs]
def get_sample_values(self, progress: ProgressBarMixinBase, data: Data):
return progress.extra.setdefault(
f'{self.key_prefix}sample_values',
containers.SliceableDeque(),
)
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:
# Add a sample but limit the size to `num_samples`
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
while sample_times[2:] and minimum_time > sample_times[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:
return sample_times, sample_values
[docs]
class ETA(Timer):
"""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,
):
# Backwards compatibility: rewrite a legacy bare ``%s`` placeholder to
# the named ``%(eta)s`` form (see ``Timer.__init__`` for the same
# elapsed-time shim).
if '%s' in format and '%(eta)s' not in format:
format = format.replace('%s', '%(eta)s')
# ``super().__init__`` (Timer) sets ``self.format`` to the
# elapsed-time default; the ETA-specific ``self.format*`` assignments
# below MUST stay after it or ETA renders 'Elapsed Time:' not 'ETA:'.
super().__init__(**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: float,
elapsed: datetime.timedelta | None,
) -> float:
"""Updates the widget to show the ETA or total time when finished."""
if elapsed:
# The max() prevents zero division errors. ``value`` is always a
# number here (``_resolve_value_elapsed`` fills the default).
per_item = elapsed.total_seconds() / max(value, 1e-6)
remaining = progress.max_value - data['value']
return remaining * per_item
else:
return 0
def _resolve_value_elapsed(
self,
progress: ProgressBarMixinBase,
data: Data,
value,
elapsed,
):
"""Fill in the value/elapsed defaults shared by the ETA variants.
When a caller does not supply them, the per-item rate is based on the
progress relative to ``min_value`` (not the raw value) and the elapsed
time is taken from the data snapshot.
"""
if value is None:
value = data['value'] - progress.min_value
if elapsed is None:
elapsed = data['time_elapsed']
return value, elapsed
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
value=None,
elapsed=None,
):
"""Updates the widget to show the ETA or total time when finished."""
value, elapsed = self._resolve_value_elapsed(
progress, data, value, elapsed
)
# ``max_value`` is ``UnknownLength`` (or ``None``) for indeterminate
# bars. The remaining-count subtraction in ``_calculate_eta`` only
# runs (and only then fails) once ``elapsed`` is truthy, so guard on
# both to keep the ``elapsed == 0`` case rendering ``format_zero`` as
# before. Nothing else in the ETA math raises ``TypeError``, so the
# previous try/except-as-control-flow is intentionally removed.
eta_na = False
if elapsed and (
progress.max_value is None
or progress.max_value is base.UnknownLength
):
data['eta_seconds'] = None
eta_na = True
else:
data['eta_seconds'] = self._calculate_eta(
progress,
data,
value=value,
elapsed=elapsed,
)
data['eta'] = None
if data['eta_seconds']:
with contextlib.suppress(ValueError, OverflowError, OSError):
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
return Timer.__call__(self, progress, data, format=fmt)
[docs]
class AbsoluteETA(ETA):
"""Widget which attempts to estimate the absolute time of arrival."""
def _calculate_eta(
self,
progress: ProgressBarMixinBase,
data: Data,
value: float,
elapsed: datetime.timedelta | None,
) -> datetime.datetime:
eta_seconds = ETA._calculate_eta(self, progress, data, value, elapsed)
now = datetime.datetime.now()
try:
return now + datetime.timedelta(seconds=eta_seconds)
except OverflowError: # pragma: no cover
return datetime.datetime.max
def __init__(
self,
format_not_started='Estimated finish time: ----/--/-- --:--:--',
format_finished='Finished at: %(elapsed)s',
format='Estimated finish time: %(eta)s',
**kwargs,
):
super().__init__(
format_not_started=format_not_started,
format_finished=format_finished,
format=format,
**kwargs,
)
[docs]
class AdaptiveETA(ETA, SamplesMixin):
"""WidgetBase which attempts to estimate the time of arrival.
Uses a sampled average of the speed based on the 10 last updates.
Very convenient for resuming the progress halfway. For an estimate based
on an exponential moving average (EMA) of the speed instead of a windowed
sample, use `SmoothingETA`.
"""
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
super().__init__(**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
return ETA.__call__(self, progress, data, value=value, elapsed=elapsed)
[docs]
class SmoothingETA(ETA):
"""
WidgetBase which attempts to estimate the time of arrival using an
exponential moving average (EMA) of the speed.
EMA applies more weight to recent data points and less to older ones,
and doesn't require storing all past values. This approach works well
with varying data points and smooths out fluctuations effectively.
"""
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,
)
super().__init__(**kwargs)
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
value=None,
elapsed=None,
):
value, elapsed = self._resolve_value_elapsed(
progress, data, value, elapsed
)
value = self.smoothing_algorithm.update(value, elapsed)
return ETA.__call__(self, progress, data, value=value, elapsed=elapsed)
[docs]
class DataSize(FormatWidgetMixin, WidgetBase):
"""
Widget for showing an amount of data transferred/processed.
Automatically formats the value (assumed to be a count of bytes) with an
appropriate sized unit, based on the IEC binary prefixes (powers of 1024).
"""
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
super().__init__(format=format, **kwargs)
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
format: str | None = 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
return FormatWidgetMixin.__call__(self, progress, data, format)
[docs]
class FileTransferSpeed(FormatWidgetMixin, TimeSensitiveWidgetBase):
"""
Widget for showing the current transfer speed (useful for file transfers).
"""
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
super().__init__(format=format, **kwargs)
def _speed(self, value, elapsed):
speed = float(value) / elapsed
return utils.scale_1024(speed, len(self.prefixes))
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-6
and value > 2e-6
): # =~ 0
scaled, power = self._speed(value, elapsed)
else:
scaled = power = 0
data['unit'] = self.unit
if power == 0 and 0 < scaled < 0.1:
# Slow transfers are shown as seconds per unit instead. Note
# that this is only done when there is actual data; before the
# first data arrives the regular format is used.
data['scaled'] = 1 / scaled
data['prefix'] = self.prefixes[0]
return FormatWidgetMixin.__call__(
self,
progress,
data,
self.inverse_format,
)
else:
data['scaled'] = scaled
data['prefix'] = self.prefixes[power]
return FormatWidgetMixin.__call__(self, progress, data)
[docs]
class AdaptiveTransferSpeed(FileTransferSpeed, SamplesMixin):
"""Widget for showing the transfer speed based on the last X samples."""
def __init__(self, **kwargs: typing.Any):
super().__init__(**kwargs)
def __call__(
self,
progress: ProgressBarMixinBase,
data,
value=None,
total_seconds_elapsed=None,
):
elapsed, value = SamplesMixin.__call__(
self,
progress,
data,
delta=True,
)
return FileTransferSpeed.__call__(self, progress, data, value, elapsed)
[docs]
class AnimatedMarker(TimeSensitiveWidgetBase):
"""An animated marker for the progress bar which defaults to appear as if
it were rotating.
"""
def __init__(
self,
markers: str = '|/-\\',
default: str | None = None,
fill: str = '',
marker_wrap: str | tuple[str | None, str | None] | None = None,
fill_wrap: str | tuple[str | None, str | None] | None = None,
**kwargs: typing.Any,
):
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
super().__init__(**kwargs)
def __call__(self, progress: ProgressBarMixinBase, data: Data, width=None):
"""Updates the widget to show the next marker or the first marker when
finished.
"""
if progress.end_time:
# When finished, keep a filling marker full instead of
# collapsing to a single character; a plain marker has no fill
# so it falls back to its default character.
if self.fill:
return self.fill(progress, data, width)
return self.default
marker = self.markers[data['updates'] % len(self.markers)]
if self.marker_wrap:
marker = self.marker_wrap.format(marker)
if self.fill:
# Cut the last character so we can replace it with our marker
fill = self.fill(
progress,
data,
width - progress.custom_len(marker), # type: ignore
)
else:
fill = ''
# Python 3 returns an int when indexing bytes
if isinstance(marker, int): # pragma: no cover
marker = bytes(marker)
fill = fill.encode()
else:
# cast fill to the same type as marker
fill = type(marker)(fill)
return fill + marker # type: ignore
# Legacy alias for `AnimatedMarker`, kept for backwards compatibility. Kept as
# a plain alias (no DeprecationWarning) until the next major version.
RotatingMarker = AnimatedMarker
[docs]
class Counter(FormatWidgetMixin, WidgetBase):
"""Displays the current count."""
def __init__(self, format='%(value)d', **kwargs: typing.Any):
# ``format`` is consumed by ``FormatWidgetMixin``; do not leak it into
# the ``WidgetBase`` tail of the cooperative chain.
super().__init__(format=format, **kwargs)
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
format=None,
):
return FormatWidgetMixin.__call__(self, progress, data, format)
[docs]
class ColoredMixin:
# See ``WidgetBase``: class-level defaults, overridable per instance.
_fixed_colors: TFixedColors = TFixedColors(
fg_none=colors.yellow,
bg_none=None,
)
_gradient_colors: TGradientColors = TGradientColors(
fg=colors.gradient,
bg=None,
)
[docs]
class Percentage(FormatWidgetMixin, ColoredMixin, WidgetBase):
"""Displays the current percentage as a number with a percent sign."""
def __init__(
self, format='%(percentage)3d%%', na='N/A%%', **kwargs: typing.Any
):
self.na = na
super().__init__(format=format, **kwargs)
UNIT_PREFIXES = ('', 'Ki', 'Mi', 'Gi', 'Ti', 'Pi')
DEFAULT_UNIT = object()
[docs]
class UnitProgress(WidgetBase):
"""Displays progress as a count with an optional unit and 1024 scaling."""
def __init__(
self,
unit=DEFAULT_UNIT,
unit_scale=DEFAULT_UNIT,
**kwargs: typing.Any,
):
self.use_progress_unit = unit is DEFAULT_UNIT
self.use_progress_unit_scale = unit_scale is DEFAULT_UNIT
self.unit: str = (
'it' if unit is DEFAULT_UNIT else typing.cast(str, unit)
)
self.unit_scale: bool = (
False
if unit_scale is DEFAULT_UNIT
else typing.cast(bool, unit_scale)
)
WidgetBase.__init__(self, **kwargs)
def __call__(self, progress: ProgressBarMixinBase, data: Data) -> str:
unit = typing.cast(str, data.get('unit', self.unit))
unit_scale = typing.cast(bool, data.get('unit_scale', self.unit_scale))
if not self.use_progress_unit:
unit = self.unit
if not self.use_progress_unit_scale:
unit_scale = self.unit_scale
value = format_unit_value(data.get('value'), unit, unit_scale)
max_value = format_unit_value(data.get('max_value'), unit, unit_scale)
return f'{value} of {max_value}'
[docs]
class SimpleProgress(FormatWidgetMixin, ColoredMixin, WidgetBase):
"""Returns progress as a count of the total (e.g.: "5 of 47")."""
max_width_cache: dict[
str
| tuple[
NumberT | type[base.UnknownLength] | None,
NumberT | type[base.UnknownLength] | None,
],
int | None,
]
DEFAULT_FORMAT = '%(value_s)s of %(max_value_s)s'
def __init__(self, format=DEFAULT_FORMAT, **kwargs: typing.Any):
super().__init__(format=format, **kwargs)
# ``max_width_cache`` reads ``self.max_width``; keep it after super().
self.max_width_cache = dict()
# Pyright isn't happy when we set the key in the initialiser
self.max_width_cache['default'] = self.max_width or 0
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
format=None,
):
# If max_value is not available, display N/A
if data.get('max_value'):
data['max_value_s'] = data['max_value']
else:
data['max_value_s'] = 'N/A'
# if value is not available it's the zeroth iteration
if data.get('value'):
data['value_s'] = data['value']
else:
data['value_s'] = 0
formatted = FormatWidgetMixin.__call__(
self,
progress,
data,
format=format,
)
# Guess the maximum width from the min and max value
key = progress.min_value, progress.max_value
max_width: int | None = 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: # pragma: no cover
continue
temporary_data['value'] = value
if width := progress.custom_len( # pragma: no branch
FormatWidgetMixin.__call__(
self,
progress,
temporary_data,
format=format,
),
):
max_width = max(max_width or 0, width)
self.max_width_cache[key] = max_width
# Adjust the output to have a consistent size in all cases
if max_width: # pragma: no branch
formatted = formatted.rjust(max_width)
return self._apply_colors(formatted, data)
[docs]
class Bar(AutoWidthWidgetBase):
"""A progress bar which stretches to fill the line."""
fg: terminal.OptionalColor = colors.gradient
bg: terminal.OptionalColor = None
def __init__(
self,
marker='#',
left='|',
right='|',
fill=' ',
fill_left=True,
marker_wrap=None,
**kwargs,
):
"""Creates a customizable progress bar.
The callable takes the same parameters as the `__call__` method
marker - string or callable object to use as a marker
left - string or callable object to use as a left border
right - string or callable object to use as a right border
fill - character to use for the empty part of the progress bar
fill_left - whether to fill from the left or the right
"""
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
super().__init__(**kwargs)
def _render_borders(
self,
progress: ProgressBarMixinBase,
data: Data,
width: int,
) -> tuple[str, str, int]:
"""Resolve the left/right borders and the width left for the body.
The borders may be callables, so they are resolved against
``progress``/``data`` and their visible length subtracted from
``width``. Shared by every :class:`Bar` subclass' ``__call__``.
"""
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)
return left, right, width
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
width: int = 0,
color: bool = True,
):
"""Updates the progress bar and its subcomponents."""
left, right, width = self._render_borders(progress, data, width)
marker = converters.to_unicode(self.marker(progress, data, width))
fill = converters.to_unicode(self.fill(progress, data, width))
# Make sure we ignore invisible characters when filling
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)
return left + marker + right
[docs]
class ReverseBar(Bar):
"""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.
marker - string or updatable object to use as a marker
left - string or updatable object to use as a left border
right - string or updatable object to use as a right border
fill - character to use for the empty part of the progress bar
fill_left - whether to fill from the left or the right
"""
super().__init__(
marker=marker,
left=left,
right=right,
fill=fill,
fill_left=fill_left,
**kwargs,
)
[docs]
class BouncingBar(Bar, TimeSensitiveWidgetBase):
"""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: bool = True,
):
"""Updates the progress bar and its subcomponents."""
left, right, width = self._render_borders(progress, data, width)
marker = converters.to_unicode(self.marker(progress, data, width))
fill = converters.to_unicode(self.fill(progress, data, width))
if width: # pragma: no branch
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
return left + marker + right
[docs]
class FormatCustomText(FormatWidgetMixin, WidgetBase):
mapping: dict[str, typing.Any] = dict() # noqa: RUF012
copy = False
def __init__(
self,
format: str,
mapping: dict[str, typing.Any] | None = None,
**kwargs,
):
# ``self.format`` is set by FormatWidgetMixin.__init__ (via super
# below); assigning it here too would just overwrite that.
# Always build a fresh per-instance dict so update_mapping() never
# mutates the shared class-level default (which every other
# default-constructed instance would otherwise alias). Fall back to
# the class-level `mapping` so subclasses that declare defaults keep
# them when no mapping is passed.
self.mapping = dict(self.mapping if mapping is None else mapping)
super().__init__(format=format, **kwargs)
[docs]
def update_mapping(self, **mapping: typing.Any):
self.mapping.update(mapping)
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
format: str | None = None,
):
return FormatWidgetMixin.__call__(
self,
progress,
self.mapping,
format or self.format,
)
[docs]
class VariableMixin(_WidgetKwargsSink):
"""Mixin to display a custom user variable."""
def __init__(self, name, **kwargs: typing.Any):
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')
self.name = name
super().__init__(**kwargs)
[docs]
class Postfix(VariableMixin, WidgetBase):
"""Displays a live postfix string or key-value mapping."""
def __init__(
self,
name='postfix',
prefix=' ',
separator=', ',
**kwargs: typing.Any,
):
self.prefix = prefix
self.separator = separator
VariableMixin.__init__(self, name=name)
WidgetBase.__init__(self, **kwargs)
def __call__(self, progress: ProgressBarMixinBase, data: Data) -> str:
value = data['variables'].get(self.name)
if value is None or (
isinstance(value, (str, dict, list, set, tuple)) and not value
):
return ''
if isinstance(value, str):
rendered = value
elif isinstance(value, dict):
rendered = self.separator.join(
f'{key}={value[key]}' for key in sorted(value)
)
else:
rendered = str(value)
return f'{self.prefix}{rendered}'
[docs]
class MultiRangeBar(Bar, VariableMixin):
"""
A bar with multiple sub-ranges, each represented by a different symbol.
The various ranges are represented on a user-defined variable, formatted as
.. code-block:: python
[['Symbol1', amount1], ['Symbol2', amount2], ...]
"""
def __init__(self, name, markers, **kwargs: typing.Any):
# ``name`` rides through Bar's cooperative chain to VariableMixin.
super().__init__(name=name, **kwargs)
self.markers = [string_or_lambda(marker) for marker in markers]
[docs]
def get_values(self, progress: ProgressBarMixinBase, data: Data):
return data['variables'][self.name] or []
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
width: int = 0,
color: bool = True,
):
"""Updates the progress bar and its subcomponents."""
left, right, width = self._render_borders(progress, data, width)
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, strict=False):
marker = converters.to_unicode(marker(progress, data, width))
if progress.custom_len(marker) != 1:
raise ValueError('Markers are required to be 1 char')
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))
if progress.custom_len(fill) != 1:
raise ValueError(
f'Fill is required to be 1 char, got {fill!r}'
)
middle = fill * width
return left + middle + right
[docs]
class MultiProgressBar(MultiRangeBar):
def __init__(
self,
name,
# NOTE: the markers are not whitespace even though some
# terminals don't show the characters correctly!
markers=' ▁▂▃▄▅▆▇█',
**kwargs,
):
super().__init__(
name=name,
markers=list(reversed(markers)),
**kwargs,
)
[docs]
def get_values(self, progress: ProgressBarMixinBase, data: Data):
ranges = [0.0] * len(self.markers)
for value in data['variables'][self.name] or []:
if not isinstance(value, (int, float)):
# Progress is (value, max). A zero maximum means the total
# is not known (yet), so no progress can be shown.
progress_value, progress_max = value
if progress_max:
value = float(progress_value) / float(progress_max)
else:
value = 0.0
if not 0 <= value <= 1:
raise ValueError(
'Range value needs to be in the range [0..1], '
f'got {value}',
)
range_ = value * (len(ranges) - 1)
pos = int(range_)
frac = range_ % 1
ranges[pos] += 1 - frac
if frac:
ranges[pos + 1] += frac
if self.fill_left: # pragma: no branch
ranges = list(reversed(ranges))
return ranges
[docs]
class GranularMarkers:
smooth = ' ▏▎▍▌▋▊▉█'
bar = ' ▁▂▃▄▅▆▇█'
snake = ' ▖▌▛█'
fade_in = ' ░▒▓█'
dots = ' ⡀⡄⡆⡇⣇⣧⣷⣿'
growing_circles = ' .oO'
[docs]
class GranularBar(AutoWidthWidgetBase):
"""A progressbar that can display progress at a sub-character granularity
by using multiple marker characters.
Examples of markers:
- Smooth: ` ▏▎▍▌▋▊▉█` (default)
- Bar: ` ▁▂▃▄▅▆▇█`
- Snake: ` ▖▌▛█`
- Fade in: ` ░▒▓█`
- Dots: ` ⡀⡄⡆⡇⣇⣧⣷⣿`
- Growing circles: ` .oO`
The markers can be accessed through GranularMarkers. GranularMarkers.dots
for example
"""
def __init__(
self,
markers=GranularMarkers.smooth,
left='|',
right='|',
**kwargs,
):
"""Creates a customizable progress bar.
markers - string of characters to use as granular progress markers. The
first character should represent 0% and the last 100%.
Ex: ` .oO`.
left - string or callable object to use as a left border
right - string or callable object to use as a right border
"""
self.markers = markers
self.left = string_or_lambda(left)
self.right = string_or_lambda(right)
super().__init__(**kwargs)
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
width: int = 0,
):
# GranularBar descends from AutoWidthWidgetBase, not Bar, so it can't
# reach Bar._render_borders. The border preamble is intentionally
# duplicated here rather than hoisting the helper onto a shared base
# (which would put border concerns on width widgets that have none).
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 typing.cast(float, max_value) > 0
):
percent = progress.value / max_value # type: ignore
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)
# Make sure we ignore invisible characters when filling
width += len(marker) - progress.custom_len(marker)
marker = marker.ljust(width, self.markers[0])
return left + marker + right
[docs]
class PercentageLabelBar(Percentage, FormatLabelBar):
"""A bar which displays the current percentage in the center."""
# %3d adds an extra space that makes it look off-center
# %2d keeps the label somewhat consistently in-place
def __init__(
self, format='%(percentage)2d%%', na='N/A%%', **kwargs: typing.Any
):
super().__init__(format=format, na=na, **kwargs)
def __call__( # type: ignore
self,
progress: ProgressBarMixinBase,
data: Data,
width: int = 0,
format: FormatString = None,
):
return super().__call__(progress, data, width, format=format)
[docs]
class Variable(FormatWidgetMixin, VariableMixin, WidgetBase):
"""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.width = width
self.precision = precision
# FormatWidgetMixin (first in the MRO) now sets ``self.format``;
# ``name`` rides the cooperative chain to VariableMixin.
super().__init__(name=name, format=format, **kwargs)
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
format: str | None = 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:
# Make sure to try and cast the value first, otherwise the
# formatting will generate warnings/errors on newer Python releases
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
return self.format.format(**context)
[docs]
class DynamicMessage(Variable):
"""Legacy alias for `Variable`; prefer `Variable` in new code.
Kept as a plain subclass (no DeprecationWarning) until the next major
version.
"""
[docs]
class CurrentTime(FormatWidgetMixin, TimeSensitiveWidgetBase):
"""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
super().__init__(format=format, **kwargs)
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
format: str | None = None,
):
data['current_time'] = self.current_time()
data['current_datetime'] = self.current_datetime()
return FormatWidgetMixin.__call__(self, progress, data, format=format)
[docs]
def current_datetime(self):
now = datetime.datetime.now()
if not self.microseconds:
now = now.replace(microsecond=0)
return now
[docs]
def current_time(self):
return self.current_datetime().time()
[docs]
class JobStatusBar(Bar, VariableMixin):
"""
Widget which displays the job status as markers on the bar.
The status updates can be given either as a boolean or as a string. If it's
a string, it will be displayed as-is. If it's a boolean, it will be
displayed as a marker (default: '█' for success, 'X' for failure)
configurable through the `success_marker` and `failure_marker` parameters.
Args:
name: The name of the variable to use for the status updates.
left: The left border of the bar.
right: The right border of the bar.
fill: The fill character of the bar.
fill_left: Whether to fill the bar from the left or the right.
success_fg_color: The foreground color to use for successful jobs.
success_bg_color: The background color to use for successful jobs.
success_marker: The marker to use for successful jobs.
failure_fg_color: The foreground color to use for failed jobs.
failure_bg_color: The background color to use for failed jobs.
failure_marker: The marker to use for failed jobs.
"""
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]
"""Retained for backwards compatibility only.
Per-run marker state now lives in ``progress.extra`` (see
:py:meth:`get_job_markers`) so a single widget shared by multiple bars no
longer interleaves their markers. This attribute is no longer read or
updated during rendering.
"""
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,
):
# Retained for backward compatibility only; render state now lives in
# ``progress.extra`` (see get_job_markers) so a single widget reused by
# multiple bars no longer interleaves their markers.
self.job_markers = []
# Unique per-widget key so multiple JobStatusBars on the same bar do
# not share storage either.
self._markers_key = f'{type(self).__name__}_{id(self)}_job_markers'
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
# ``name`` rides Bar's cooperative chain to VariableMixin (which also
# validates it); Bar re-sets left/right/fill from the same values.
super().__init__(
name=name,
left=left,
right=right,
fill=fill,
fill_left=fill_left,
**kwargs,
)
[docs]
def get_job_markers(self, progress: ProgressBarMixinBase) -> list[str]:
# Per-bar marker history, following SamplesMixin's ``progress.extra``
# pattern so the widget itself stays stateless and reusable.
return progress.extra.setdefault(self._markers_key, [])
def __call__(
self,
progress: ProgressBarMixinBase,
data: Data,
width: int = 0,
color: bool = True,
):
left, right, width = self._render_borders(progress, data, width)
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: # pragma: no branch
marker = self.failure_marker
fg_color = self.failure_fg_color
bg_color = self.failure_bg_color
else: # pragma: no cover
marker = status
fg_color = bg_color = None
marker = converters.to_unicode(marker)
if fg_color: # pragma: no branch
marker = fg_color.fg(marker)
if bg_color: # pragma: no cover
marker = bg_color.bg(marker)
job_markers = self.get_job_markers(progress)
job_markers.append(marker)
# Drop the oldest markers when they no longer fit the
# available width
while (
len(job_markers) > 1
and progress.custom_len(''.join(job_markers)) > width
):
job_markers.pop(0)
marker = ''.join(job_markers)
width -= progress.custom_len(marker)
fill = converters.to_unicode(self.fill(progress, data, width))
fill = self._apply_colors(fill * max(width, 0), data)
if self.fill_left: # pragma: no branch
marker += fill
else: # pragma: no cover
marker = fill + marker
else:
marker = ''
return left + marker + right