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
|
"""
Sends updates to a Discord bot.
Usage:
>>> from tqdm.contrib.discord import tqdm, trange
>>> for i in trange(10, token='{token}', channel_id='{channel_id}'):
... ...
"""
from os import getenv
from warnings import warn
from requests import Session
from requests.utils import default_user_agent
from ..auto import tqdm as tqdm_auto
from ..std import TqdmWarning
from ..version import __version__
from .utils_worker import MonoWorker
__author__ = {"github.com/": ["casperdcl", "guigoruiz1"]}
__all__ = ['DiscordIO', 'tqdm_discord', 'tdrange', 'tqdm', 'trange']
class DiscordIO(MonoWorker):
"""Non-blocking file-like IO using a Discord Bot."""
API = "https://discord.com/api/v10"
UA = f"tqdm (https://tqdm.github.io, {__version__}) {default_user_agent()}"
def __init__(self, token, channel_id):
"""Creates a new message in the given `channel_id`."""
super().__init__()
self.token = token
self.channel_id = channel_id
self.session = Session()
self.text = self.__class__.__name__
self.message_id
@property
def message_id(self):
if hasattr(self, '_message_id'):
return self._message_id
try:
res = self.session.post(
f'{self.API}/channels/{self.channel_id}/messages',
headers={'Authorization': f'Bot {self.token}', 'User-Agent': self.UA},
json={'content': f"`{self.text}`"}).json()
except Exception as e:
tqdm_auto.write(str(e))
else:
if res.get('error_code') == 429:
warn("Creation rate limit: try increasing `mininterval`.",
TqdmWarning, stacklevel=2)
else:
self._message_id = res['id']
return self._message_id
def write(self, s):
"""Replaces internal `message_id`'s text with `s`."""
if not s:
s = "..."
s = s.replace('\r', '').strip()
if s == self.text:
return # avoid duplicate message Bot error
message_id = self.message_id
if message_id is None:
return
self.text = s
try:
future = self.submit(
self.session.patch,
f'{self.API}/channels/{self.channel_id}/messages/{message_id}',
headers={'Authorization': f'Bot {self.token}', 'User-Agent': self.UA},
json={'content': f"`{self.text}`"})
except Exception as e:
tqdm_auto.write(str(e))
else:
return future
def delete(self):
"""Deletes internal `message_id`."""
try:
future = self.submit(
self.session.delete,
f'{self.API}/channels/{self.channel_id}/messages/{self.message_id}',
headers={'Authorization': f'Bot {self.token}', 'User-Agent': self.UA})
except Exception as e:
tqdm_auto.write(str(e))
else:
return future
class tqdm_discord(tqdm_auto):
"""
Standard `tqdm.auto.tqdm` but also sends updates to a Discord Bot.
May take a few seconds to create (`__init__`).
- create a discord bot (not public, no requirement of OAuth2 code
grant, only send message permissions) & invite it to a channel:
<https://discordpy.readthedocs.io/en/latest/discord.html>
- copy the bot `{token}` & `{channel_id}` and paste below
>>> from tqdm.contrib.discord import tqdm, trange
>>> for i in tqdm(iterable, token='{token}', channel_id='{channel_id}'):
... ...
"""
def __init__(self, *args, **kwargs):
"""
Parameters
----------
token : str, required. Discord bot token
[default: ${TQDM_DISCORD_TOKEN}].
channel_id : int, required. Discord channel ID
[default: ${TQDM_DISCORD_CHANNEL_ID}].
See `tqdm.auto.tqdm.__init__` for other parameters.
"""
if not kwargs.get('disable'):
kwargs = kwargs.copy()
self.dio = DiscordIO(
kwargs.pop('token', getenv('TQDM_DISCORD_TOKEN')),
kwargs.pop('channel_id', getenv('TQDM_DISCORD_CHANNEL_ID')))
super().__init__(*args, **kwargs)
def display(self, **kwargs):
super().display(**kwargs)
fmt = self.format_dict
if fmt.get('bar_format', None):
fmt['bar_format'] = fmt['bar_format'].replace(
'<bar/>', '{bar:10u}').replace('{bar}', '{bar:10u}')
else:
fmt['bar_format'] = '{l_bar}{bar:10u}{r_bar}'
self.dio.write(self.format_meter(**fmt))
def clear(self, *args, **kwargs):
super().clear(*args, **kwargs)
if not self.disable:
self.dio.write("")
def close(self):
if self.disable:
return
super().close()
if not (self.leave or (self.leave is None and self.pos == 0)):
self.dio.delete()
def tdrange(*args, **kwargs):
"""Shortcut for `tqdm.contrib.discord.tqdm(range(*args), **kwargs)`."""
return tqdm_discord(range(*args), **kwargs)
# Aliases
tqdm = tqdm_discord
trange = tdrange
|
