1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
|
"""Wn utility classes."""
import sys
from collections.abc import Callable
from typing import TextIO
def synset_id_formatter(fmt: str = "{prefix}-{offset:08}-{pos}", **kwargs) -> Callable:
"""Return a function for formatting synset ids.
The *fmt* argument can be customized. It will be formatted using
any other keyword arguments given to this function and any given
to the resulting function. By default, the format string expects a
``prefix`` string argument for the namespace (such as a lexicon
id), an ``offset`` integer argument (such as a WNDB offset), and a
``pos`` string argument.
Arguments:
fmt: A Python format string
**kwargs: Keyword arguments for the format string.
Example:
>>> pwn_synset_id = synset_id_formatter(prefix="pwn")
>>> pwn_synset_id(offset=1174, pos="n")
'pwn-00001174-n'
"""
def format_synset_id(**_kwargs) -> str:
return fmt.format(**kwargs, **_kwargs)
return format_synset_id
class ProgressHandler:
"""An interface for updating progress in long-running processes.
Long-running processes in Wn, such as :func:`wn.download` and
:func:`wn.add`, call to a progress handler object as they go. The
default progress handler used by Wn is :class:`ProgressBar`, which
updates progress by formatting and printing a textual bar to
stderr. The :class:`ProgressHandler` class may be used directly,
which does nothing, or users may create their own subclasses for,
e.g., updating a GUI or some other handler.
The initialization parameters, except for ``file``, are stored in
a :attr:`kwargs` member and may be updated after the handler is
created through the :meth:`set` method. The :meth:`update` method
is the primary way a counter is updated. The :meth:`flash` method
is sometimes called for simple messages. When the process is
complete, the :meth:`close` method is called, optionally with a
message.
"""
def __init__(
self,
*,
message: str = "",
count: int = 0,
total: int = 0,
refresh_interval: int = 0,
unit: str = "",
status: str = "",
file: TextIO = sys.stderr,
):
self.file = file
self.kwargs = {
"count": count,
"total": total,
"refresh_interval": refresh_interval,
"message": message,
"unit": unit,
"status": status,
}
self._refresh_quota: int = refresh_interval
def update(self, n: int = 1, force: bool = False) -> None:
"""Update the counter with the increment value *n*.
This method should update the ``count`` key of :attr:`kwargs`
with the increment value *n*. After this, it is expected to
update some user-facing progress indicator.
If *force* is :python:`True`, any indicator will be refreshed
regardless of the value of the refresh interval.
"""
self.kwargs["count"] += n # type: ignore
def set(self, **kwargs) -> None:
"""Update progress handler parameters.
Calling this method also runs :meth:`update` with an increment
of 0, which causes a refresh of any indicator without changing
the counter.
"""
self.kwargs.update(**kwargs)
self.update(0, force=True)
def flash(self, message: str) -> None:
"""Issue a message unrelated to the current counter.
This may be useful for multi-stage processes to indicate the
move to a new stage, or to log unexpected situations.
"""
pass
def close(self) -> None:
"""Close the progress handler.
This might be useful for closing file handles or cleaning up
resources.
"""
pass
class ProgressBar(ProgressHandler):
"""A :class:`ProgressHandler` subclass for printing a progress bar.
Example:
>>> p = ProgressBar(message="Progress: ", total=10, unit=" units")
>>> p.update(3)
Progress: [######### ] (3/10 units)
See :meth:`format` for a description of how the progress bar is
formatted.
"""
#: The default formatting template.
FMT = "\r{message}{bar}{counter}{status}"
def update(self, n: int = 1, force: bool = False) -> None:
"""Increment the count by *n* and print the reformatted bar."""
self.kwargs["count"] += n # type: ignore
self._refresh_quota -= n
if force or self._refresh_quota <= 0:
self._refresh_quota = self.kwargs["refresh_interval"] # type: ignore
s = self.format()
if self.file:
print("\r\033[K", end="", file=self.file)
print(s, end="", file=self.file)
def format(self) -> str:
"""Format and return the progress bar.
The bar is is formatted according to :attr:`FMT`, using
variables from :attr:`kwargs` and two computed variables:
- ``bar``: visualization of the progress bar, empty when
``total`` is 0
- ``counter``: display of ``count``, ``total``, and ``units``
>>> p = ProgressBar(message="Progress", count=2, total=10, unit="K")
>>> p.format()
'\\rProgress [###### ] (2/10K) '
>>> p = ProgressBar(count=2, status="Counting...")
>>> p.format()
'\\r (2) Counting...'
"""
_kw = self.kwargs
width = 30
total: int = _kw["total"] # type: ignore
count: int = _kw["count"] # type: ignore
if total > 0:
num = min(count, total) * width
fill = (num // total) * "#"
part = ((num % total) * 3) // total
if part:
fill += "-="[part - 1]
bar = f" [{fill:<{width}}]"
counter = f" ({count}/{total}{_kw['unit']}) "
else:
bar = ""
counter = f" ({count}{_kw['unit']}) "
return self.FMT.format(bar=bar, counter=counter, **_kw)
def flash(self, message: str) -> None:
"""Overwrite the progress bar with *message*."""
print(f"\r\033[K{message}", end="", file=self.file)
def close(self) -> None:
"""Print a newline so the last printed bar remains on screen."""
print(file=self.file)
|
