| 115 | |
| 116 | |
| 117 | class TqdmGit(Tqdm): |
| 118 | BAR_FMT = ( |
| 119 | "{desc}|{bar}|{postfix[info]}{n_fmt}/{total_fmt} [{elapsed}, {rate_fmt:>11}]" |
| 120 | ) |
| 121 | |
| 122 | def __init__(self, *args, **kwargs): |
| 123 | kwargs.setdefault("unit", "obj") |
| 124 | kwargs.setdefault("bar_format", self.BAR_FMT) |
| 125 | super().__init__(*args, **kwargs) |
| 126 | self._last_phase = None |
| 127 | |
| 128 | def update_git(self, event: "GitProgressEvent") -> None: |
| 129 | phase, completed, total, message, *_ = event |
| 130 | if phase: |
| 131 | message = (phase + " | " + message) if message else phase |
| 132 | if message: |
| 133 | self.set_msg(message) |
| 134 | force_refresh = ( # force-refresh progress bar when: |
| 135 | (total and completed and completed >= total) # the task completes |
| 136 | or total != self.total # the total changes |
| 137 | or phase != self._last_phase # or, the phase changes |
| 138 | ) |
| 139 | if completed is not None: |
| 140 | self.update_to(completed, total) |
| 141 | if force_refresh: |
| 142 | self.refresh() |
| 143 | self._last_phase = phase |
| 144 | |
| 145 | |
| 146 | def clone(url: str, to_path: str, **kwargs): |
no outgoing calls
no test coverage detected