| from functools import wraps |
|
|
|
|
| class Callback: |
| """ |
| Base class and interface for callback mechanism |
| |
| This class can be used directly for monitoring file transfers by |
| providing ``callback=Callback(hooks=...)`` (see the ``hooks`` argument, |
| below), or subclassed for more specialised behaviour. |
| |
| Parameters |
| ---------- |
| size: int (optional) |
| Nominal quantity for the value that corresponds to a complete |
| transfer, e.g., total number of tiles or total number of |
| bytes |
| value: int (0) |
| Starting internal counter value |
| hooks: dict or None |
| A dict of named functions to be called on each update. The signature |
| of these must be ``f(size, value, **kwargs)`` |
| """ |
|
|
| def __init__(self, size=None, value=0, hooks=None, **kwargs): |
| self.size = size |
| self.value = value |
| self.hooks = hooks or {} |
| self.kw = kwargs |
|
|
| def __enter__(self): |
| return self |
|
|
| def __exit__(self, *exc_args): |
| self.close() |
|
|
| def close(self): |
| """Close callback.""" |
|
|
| def branched(self, path_1, path_2, **kwargs): |
| """ |
| Return callback for child transfers |
| |
| If this callback is operating at a higher level, e.g., put, which may |
| trigger transfers that can also be monitored. The function returns a callback |
| that has to be passed to the child method, e.g., put_file, |
| as `callback=` argument. |
| |
| The implementation uses `callback.branch` for compatibility. |
| When implementing callbacks, it is recommended to override this function instead |
| of `branch` and avoid calling `super().branched(...)`. |
| |
| Prefer using this function over `branch`. |
| |
| Parameters |
| ---------- |
| path_1: str |
| Child's source path |
| path_2: str |
| Child's destination path |
| **kwargs: |
| Arbitrary keyword arguments |
| |
| Returns |
| ------- |
| callback: Callback |
| A callback instance to be passed to the child method |
| """ |
| self.branch(path_1, path_2, kwargs) |
| |
| return kwargs.pop("callback", DEFAULT_CALLBACK) |
|
|
| def branch_coro(self, fn): |
| """ |
| Wraps a coroutine, and pass a new child callback to it. |
| """ |
|
|
| @wraps(fn) |
| async def func(path1, path2: str, **kwargs): |
| with self.branched(path1, path2, **kwargs) as child: |
| return await fn(path1, path2, callback=child, **kwargs) |
|
|
| return func |
|
|
| def set_size(self, size): |
| """ |
| Set the internal maximum size attribute |
| |
| Usually called if not initially set at instantiation. Note that this |
| triggers a ``call()``. |
| |
| Parameters |
| ---------- |
| size: int |
| """ |
| self.size = size |
| self.call() |
|
|
| def absolute_update(self, value): |
| """ |
| Set the internal value state |
| |
| Triggers ``call()`` |
| |
| Parameters |
| ---------- |
| value: int |
| """ |
| self.value = value |
| self.call() |
|
|
| def relative_update(self, inc=1): |
| """ |
| Delta increment the internal counter |
| |
| Triggers ``call()`` |
| |
| Parameters |
| ---------- |
| inc: int |
| """ |
| self.value += inc |
| self.call() |
|
|
| def call(self, hook_name=None, **kwargs): |
| """ |
| Execute hook(s) with current state |
| |
| Each function is passed the internal size and current value |
| |
| Parameters |
| ---------- |
| hook_name: str or None |
| If given, execute on this hook |
| kwargs: passed on to (all) hook(s) |
| """ |
| if not self.hooks: |
| return |
| kw = self.kw.copy() |
| kw.update(kwargs) |
| if hook_name: |
| if hook_name not in self.hooks: |
| return |
| return self.hooks[hook_name](self.size, self.value, **kw) |
| for hook in self.hooks.values() or []: |
| hook(self.size, self.value, **kw) |
|
|
| def wrap(self, iterable): |
| """ |
| Wrap an iterable to call ``relative_update`` on each iterations |
| |
| Parameters |
| ---------- |
| iterable: Iterable |
| The iterable that is being wrapped |
| """ |
| for item in iterable: |
| self.relative_update() |
| yield item |
|
|
| def branch(self, path_1, path_2, kwargs): |
| """ |
| Set callbacks for child transfers |
| |
| If this callback is operating at a higher level, e.g., put, which may |
| trigger transfers that can also be monitored. The passed kwargs are |
| to be *mutated* to add ``callback=``, if this class supports branching |
| to children. |
| |
| Parameters |
| ---------- |
| path_1: str |
| Child's source path |
| path_2: str |
| Child's destination path |
| kwargs: dict |
| arguments passed to child method, e.g., put_file. |
| |
| Returns |
| ------- |
| |
| """ |
| return None |
|
|
| def no_op(self, *_, **__): |
| pass |
|
|
| def __getattr__(self, item): |
| """ |
| If undefined methods are called on this class, nothing happens |
| """ |
| return self.no_op |
|
|
| @classmethod |
| def as_callback(cls, maybe_callback=None): |
| """Transform callback=... into Callback instance |
| |
| For the special value of ``None``, return the global instance of |
| ``NoOpCallback``. This is an alternative to including |
| ``callback=DEFAULT_CALLBACK`` directly in a method signature. |
| """ |
| if maybe_callback is None: |
| return DEFAULT_CALLBACK |
| return maybe_callback |
|
|
|
|
| class NoOpCallback(Callback): |
| """ |
| This implementation of Callback does exactly nothing |
| """ |
|
|
| def call(self, *args, **kwargs): |
| return None |
|
|
|
|
| class DotPrinterCallback(Callback): |
| """ |
| Simple example Callback implementation |
| |
| Almost identical to Callback with a hook that prints a char; here we |
| demonstrate how the outer layer may print "#" and the inner layer "." |
| """ |
|
|
| def __init__(self, chr_to_print="#", **kwargs): |
| self.chr = chr_to_print |
| super().__init__(**kwargs) |
|
|
| def branch(self, path_1, path_2, kwargs): |
| """Mutate kwargs to add new instance with different print char""" |
| kwargs["callback"] = DotPrinterCallback(".") |
|
|
| def call(self, **kwargs): |
| """Just outputs a character""" |
| print(self.chr, end="") |
|
|
|
|
| class TqdmCallback(Callback): |
| """ |
| A callback to display a progress bar using tqdm |
| |
| Parameters |
| ---------- |
| tqdm_kwargs : dict, (optional) |
| Any argument accepted by the tqdm constructor. |
| See the `tqdm doc <https://tqdm.github.io/docs/tqdm/#__init__>`_. |
| Will be forwarded to `tqdm_cls`. |
| tqdm_cls: (optional) |
| subclass of `tqdm.tqdm`. If not passed, it will default to `tqdm.tqdm`. |
| |
| Examples |
| -------- |
| >>> import fsspec |
| >>> from fsspec.callbacks import TqdmCallback |
| >>> fs = fsspec.filesystem("memory") |
| >>> path2distant_data = "/your-path" |
| >>> fs.upload( |
| ".", |
| path2distant_data, |
| recursive=True, |
| callback=TqdmCallback(), |
| ) |
| |
| You can forward args to tqdm using the ``tqdm_kwargs`` parameter. |
| |
| >>> fs.upload( |
| ".", |
| path2distant_data, |
| recursive=True, |
| callback=TqdmCallback(tqdm_kwargs={"desc": "Your tqdm description"}), |
| ) |
| |
| You can also customize the progress bar by passing a subclass of `tqdm`. |
| |
| .. code-block:: python |
| |
| class TqdmFormat(tqdm): |
| '''Provides a `total_time` format parameter''' |
| @property |
| def format_dict(self): |
| d = super().format_dict |
| total_time = d["elapsed"] * (d["total"] or 0) / max(d["n"], 1) |
| d.update(total_time=self.format_interval(total_time) + " in total") |
| return d |
| |
| >>> with TqdmCallback( |
| tqdm_kwargs={ |
| "desc": "desc", |
| "bar_format": "{total_time}: {percentage:.0f}%|{bar}{r_bar}", |
| }, |
| tqdm_cls=TqdmFormat, |
| ) as callback: |
| fs.upload(".", path2distant_data, recursive=True, callback=callback) |
| """ |
|
|
| def __init__(self, tqdm_kwargs=None, *args, **kwargs): |
| try: |
| from tqdm import tqdm |
|
|
| except ImportError as exce: |
| raise ImportError( |
| "Using TqdmCallback requires tqdm to be installed" |
| ) from exce |
|
|
| self._tqdm_cls = kwargs.pop("tqdm_cls", tqdm) |
| self._tqdm_kwargs = tqdm_kwargs or {} |
| self.tqdm = None |
| super().__init__(*args, **kwargs) |
|
|
| def call(self, *args, **kwargs): |
| if self.tqdm is None: |
| self.tqdm = self._tqdm_cls(total=self.size, **self._tqdm_kwargs) |
| self.tqdm.total = self.size |
| self.tqdm.update(self.value - self.tqdm.n) |
|
|
| def close(self): |
| if self.tqdm is not None: |
| self.tqdm.close() |
| self.tqdm = None |
|
|
| def __del__(self): |
| return self.close() |
|
|
|
|
| DEFAULT_CALLBACK = _DEFAULT_CALLBACK = NoOpCallback() |
|
|
