Skip to content

animation — Waiting animation for Telegram UI

UIAnimationService

UIAnimationService

Service for orchestrating asynchronous UI animations during long-running tasks.

This service facilitates a responsive user experience by providing visual feedback (progress bars, infinite spinners) while the bot performs background operations or waits for state changes.

Supported Scenarios
  • Delayed Fetch: Displays an animation for a fixed duration before executing a single request. Best for "scanning" or "searching" effects.
  • Polling Loop: Executes repeated checks until a terminal state is reached or a timeout occurs. Ideal for combat or matchmaking.
  • Timed Polling: Combines immediate result checking with time-bound progress bars.

Parameters:

Name Type Description Default
sender ViewSender

Instance of ViewSender used to deliver visual updates.

 required
Source code in src/codex_bot/animation/animation_service.py 
 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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
class UIAnimationService:
    """Service for orchestrating asynchronous UI animations during long-running tasks.

    This service facilitates a responsive user experience by providing visual
    feedback (progress bars, infinite spinners) while the bot performs
    background operations or waits for state changes.

    Supported Scenarios:
        - **Delayed Fetch**: Displays an animation for a fixed duration before
          executing a single request. Best for "scanning" or "searching" effects.
        - **Polling Loop**: Executes repeated checks until a terminal state is
          reached or a timeout occurs. Ideal for combat or matchmaking.
        - **Timed Polling**: Combines immediate result checking with
          time-bound progress bars.

    Args:
        sender: Instance of `ViewSender` used to deliver visual updates.
    """

    def __init__(self, sender: "ViewSender") -> None:
        self.sender = sender

    # =========================================================================
    # Public Methods
    # =========================================================================

    async def run_delayed_fetch(
        self,
        fetch_func: PollerFunc,
        delay: float = 3.0,
        step_interval: float = 1.0,
        loading_text: str = "🔍 <b>Searching...</b>",
        animation_type: AnimationType = AnimationType.PROGRESS_BAR,
    ) -> None:
        """Display a fixed-duration animation followed by a single data fetch.

        Typically used to simulate complex processing (e.g., scanning or
        searching) to manage user expectations.

        Args:
            fetch_func: An awaitable that retrieves the final `UnifiedViewDTO`.
            delay: Total duration of the animation sequence in seconds.
            step_interval: Time between frame updates in seconds.
            loading_text: The prompt to display alongside the animation.
            animation_type: The visual style of the indicator.

        Side Effects:
            - Delivers multiple intermediate `UnifiedViewDTO` frames to the UI.
        """
        steps = max(1, int(delay / step_interval))

        for i in range(steps):
            anim_str = self._generate_animation(i, steps, loading_text, animation_type)
            temp_view = UnifiedViewDTO(content=ViewResultDTO(text=anim_str))
            await self._send(temp_view)
            await asyncio.sleep(step_interval)

        view_dto, _ = await self._poll_check(fetch_func)
        await self._send(view_dto)

    async def run_polling_loop(
        self,
        check_func: PollerFunc,
        timeout: float = 60.0,
        step_interval: float = 2.0,
        loading_text: str = "⏳ <b>Waiting...</b>",
        animation_type: AnimationType = AnimationType.INFINITE,
    ) -> None:
        """Request loop until an event occurs.

        Used for: Combat polling, Arena waiting — make requests every N seconds
        while ``is_waiting=True``.

        Args:
            check_func: Status check function, returns ``(view, is_waiting)``.
            timeout: Maximum waiting time in seconds.
            step_interval: Interval between checks.
            loading_text: Text to display while waiting.
            animation_type: Animation type (usually INFINITE).
        """
        steps = int(timeout / step_interval)

        for i in range(steps):
            view_dto, is_waiting = await self._poll_check(check_func)

            if is_waiting and view_dto.content:
                anim_str = self._generate_animation(i, steps, loading_text, animation_type)
                view_dto = self._inject_animation(view_dto, anim_str)

            await self._send(view_dto)

            if not is_waiting:
                return

            await asyncio.sleep(step_interval)

    async def run_timed_polling(
        self,
        check_func: PollerFunc,
        duration: float = 5.0,
        step_interval: float = 1.0,
        loading_text: str = "🚶 <b>Moving...</b>",
        animation_type: AnimationType = AnimationType.PROGRESS_BAR,
    ) -> None:
        """Immediate request → animation based on duration from the result.

        Used for: Move — request goes to background immediately, result is stored
        in Redis, show Progress Bar based on time.

        Args:
            check_func: Check function that reads the result from state/Redis.
            duration: Expected animation duration.
            step_interval: Interval between frames.
            loading_text: Text to display.
            animation_type: Animation type (PROGRESS_BAR for timed).
        """
        steps = max(1, int(duration / step_interval))

        for i in range(steps):
            view_dto, is_waiting = await self._poll_check(check_func)

            if not is_waiting:
                await self._send(view_dto)
                return

            anim_str = self._generate_animation(i, steps, loading_text, animation_type)
            view_dto = self._inject_animation(view_dto, anim_str)

            await self._send(view_dto)
            await asyncio.sleep(step_interval)

        # Overflow: Backend slow response → Infinite mode
        infinite_step = 0
        while True:
            view_dto, is_waiting = await self._poll_check(check_func)

            if not is_waiting:
                await self._send(view_dto)
                return

            anim_str = self._generate_animation(infinite_step, steps, loading_text, AnimationType.INFINITE)
            view_dto = self._inject_animation(view_dto, anim_str)

            await self._send(view_dto)
            await asyncio.sleep(step_interval)
            infinite_step += 1

    # =========================================================================
    # Private Helpers
    # =========================================================================

    async def _poll_check(self, func: PollerFunc) -> tuple[UnifiedViewDTO, bool]:
        """Performs status check.

        Args:
            func: Poller function.

        Returns:
            Tuple ``(view, is_waiting)``.
        """
        result: Any = await func()
        if isinstance(result, tuple):
            return result
        return result, False

    def _inject_animation(self, view_dto: UnifiedViewDTO, anim_str: str) -> UnifiedViewDTO:
        """Injects animation string into content, returns new DTO.

        Looks for ``{ANIMATION}`` in text and replaces it, otherwise adds to the end.
        Uses ``model_copy`` since DTO is frozen.

        Args:
            view_dto: Original DTO.
            anim_str: Animation string.

        Returns:
            New ``UnifiedViewDTO`` with updated text.
        """
        if not view_dto.content:
            return view_dto

        original_text = view_dto.content.text
        if "{ANIMATION}" in original_text:
            new_text = original_text.replace("{ANIMATION}", anim_str)
        else:
            new_text = original_text + f"\n\n{anim_str}"

        new_content = view_dto.content.model_copy(update={"text": new_text})
        return view_dto.model_copy(update={"content": new_content})

    async def _send(self, view_dto: UnifiedViewDTO) -> None:
        """Sends View via sender.

        Args:
            view_dto: DTO to send.
        """
        await self.sender.send(view_dto)

    # =========================================================================
    # Animation Generators
    # =========================================================================

    def _generate_animation(
        self,
        step: int,
        total_steps: int,
        text: str,
        animation_type: AnimationType,
    ) -> str:
        """Generates animation string depending on the type.

        Args:
            step: Current step.
            total_steps: Total steps.
            text: Base text.
            animation_type: Animation type.

        Returns:
            String with text and ASCII indicator.
        """
        if animation_type == AnimationType.PROGRESS_BAR:
            return self._gen_progress_bar(step, total_steps, text)
        if animation_type == AnimationType.INFINITE:
            return self._gen_infinite_bar(step, text)
        return text

    def _gen_infinite_bar(self, step: int, text: str) -> str:
        """Running indicator: ``[■□□□□] → [□■□□□] → …``

        Args:
            step: Current step.
            text: Base text.

        Returns:
            String with running indicator.
        """
        total_chars = 10
        position = step % total_chars
        bar = "□" * position + "■" + "□" * (total_chars - position - 1)
        return f"{text} [{bar}]"

    def _gen_progress_bar(self, step: int, total_steps: int, text: str) -> str:
        """Filling indicator: ``[■■□□□] 40%``

        Args:
            step: Current step.
            total_steps: Total steps.
            text: Base text.

        Returns:
            String with progress bar and percentage.
        """
        percent = 1.0 if total_steps == 0 else step / total_steps
        total_chars = 10
        filled = int(total_chars * percent)
        empty = total_chars - filled
        bar = "■" * filled + "□" * empty
        return f"{text} [{bar}] {int(percent * 100)}%"

Functions

run_delayed_fetch(fetch_func, delay=3.0, step_interval=1.0, loading_text='🔍 <b>Searching...</b>', animation_type=AnimationType.PROGRESS_BAR) async

Display a fixed-duration animation followed by a single data fetch.

Typically used to simulate complex processing (e.g., scanning or searching) to manage user expectations.

Parameters:

Name Type Description Default
fetch_func PollerFunc

An awaitable that retrieves the final UnifiedViewDTO.

 required
delay float

Total duration of the animation sequence in seconds.

 3.0
step_interval float

Time between frame updates in seconds.

 1.0
loading_text str

The prompt to display alongside the animation.

 '🔍 <b>Searching...</b>'
animation_type AnimationType

The visual style of the indicator.

 PROGRESS_BAR
Side Effects
  • Delivers multiple intermediate UnifiedViewDTO frames to the UI.
Source code in src/codex_bot/animation/animation_service.py 
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
async def run_delayed_fetch(
    self,
    fetch_func: PollerFunc,
    delay: float = 3.0,
    step_interval: float = 1.0,
    loading_text: str = "🔍 <b>Searching...</b>",
    animation_type: AnimationType = AnimationType.PROGRESS_BAR,
) -> None:
    """Display a fixed-duration animation followed by a single data fetch.

    Typically used to simulate complex processing (e.g., scanning or
    searching) to manage user expectations.

    Args:
        fetch_func: An awaitable that retrieves the final `UnifiedViewDTO`.
        delay: Total duration of the animation sequence in seconds.
        step_interval: Time between frame updates in seconds.
        loading_text: The prompt to display alongside the animation.
        animation_type: The visual style of the indicator.

    Side Effects:
        - Delivers multiple intermediate `UnifiedViewDTO` frames to the UI.
    """
    steps = max(1, int(delay / step_interval))

    for i in range(steps):
        anim_str = self._generate_animation(i, steps, loading_text, animation_type)
        temp_view = UnifiedViewDTO(content=ViewResultDTO(text=anim_str))
        await self._send(temp_view)
        await asyncio.sleep(step_interval)

    view_dto, _ = await self._poll_check(fetch_func)
    await self._send(view_dto)

run_polling_loop(check_func, timeout=60.0, step_interval=2.0, loading_text='⏳ <b>Waiting...</b>', animation_type=AnimationType.INFINITE) async

Request loop until an event occurs.

Used for: Combat polling, Arena waiting — make requests every N seconds while is_waiting=True.

Parameters:

Name Type Description Default
check_func PollerFunc

Status check function, returns (view, is_waiting).

 required
timeout float

Maximum waiting time in seconds.

 60.0
step_interval float

Interval between checks.

 2.0
loading_text str

Text to display while waiting.

 '⏳ <b>Waiting...</b>'
animation_type AnimationType

Animation type (usually INFINITE).

 INFINITE
Source code in src/codex_bot/animation/animation_service.py 
 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
async def run_polling_loop(
    self,
    check_func: PollerFunc,
    timeout: float = 60.0,
    step_interval: float = 2.0,
    loading_text: str = "⏳ <b>Waiting...</b>",
    animation_type: AnimationType = AnimationType.INFINITE,
) -> None:
    """Request loop until an event occurs.

    Used for: Combat polling, Arena waiting — make requests every N seconds
    while ``is_waiting=True``.

    Args:
        check_func: Status check function, returns ``(view, is_waiting)``.
        timeout: Maximum waiting time in seconds.
        step_interval: Interval between checks.
        loading_text: Text to display while waiting.
        animation_type: Animation type (usually INFINITE).
    """
    steps = int(timeout / step_interval)

    for i in range(steps):
        view_dto, is_waiting = await self._poll_check(check_func)

        if is_waiting and view_dto.content:
            anim_str = self._generate_animation(i, steps, loading_text, animation_type)
            view_dto = self._inject_animation(view_dto, anim_str)

        await self._send(view_dto)

        if not is_waiting:
            return

        await asyncio.sleep(step_interval)

run_timed_polling(check_func, duration=5.0, step_interval=1.0, loading_text='🚶 <b>Moving...</b>', animation_type=AnimationType.PROGRESS_BAR) async

Immediate request → animation based on duration from the result.

Used for: Move — request goes to background immediately, result is stored in Redis, show Progress Bar based on time.

Parameters:

Name Type Description Default
check_func PollerFunc

Check function that reads the result from state/Redis.

 required
duration float

Expected animation duration.

 5.0
step_interval float

Interval between frames.

 1.0
loading_text str

Text to display.

 '🚶 <b>Moving...</b>'
animation_type AnimationType

Animation type (PROGRESS_BAR for timed).

 PROGRESS_BAR
Source code in src/codex_bot/animation/animation_service.py 
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
async def run_timed_polling(
    self,
    check_func: PollerFunc,
    duration: float = 5.0,
    step_interval: float = 1.0,
    loading_text: str = "🚶 <b>Moving...</b>",
    animation_type: AnimationType = AnimationType.PROGRESS_BAR,
) -> None:
    """Immediate request → animation based on duration from the result.

    Used for: Move — request goes to background immediately, result is stored
    in Redis, show Progress Bar based on time.

    Args:
        check_func: Check function that reads the result from state/Redis.
        duration: Expected animation duration.
        step_interval: Interval between frames.
        loading_text: Text to display.
        animation_type: Animation type (PROGRESS_BAR for timed).
    """
    steps = max(1, int(duration / step_interval))

    for i in range(steps):
        view_dto, is_waiting = await self._poll_check(check_func)

        if not is_waiting:
            await self._send(view_dto)
            return

        anim_str = self._generate_animation(i, steps, loading_text, animation_type)
        view_dto = self._inject_animation(view_dto, anim_str)

        await self._send(view_dto)
        await asyncio.sleep(step_interval)

    # Overflow: Backend slow response → Infinite mode
    infinite_step = 0
    while True:
        view_dto, is_waiting = await self._poll_check(check_func)

        if not is_waiting:
            await self._send(view_dto)
            return

        anim_str = self._generate_animation(infinite_step, steps, loading_text, AnimationType.INFINITE)
        view_dto = self._inject_animation(view_dto, anim_str)

        await self._send(view_dto)
        await asyncio.sleep(step_interval)
        infinite_step += 1

AnimationType

AnimationType

Bases: Enum

Animation type for displaying progress.

Attributes:

Name Type Description
PROGRESS_BAR

Fills from 0% to 100%.
INFINITE

Running indicator (snake).
NONE

No animation.
Source code in src/codex_bot/animation/animation_service.py 
20
21
22
23
24
25
26
27
28
29
30
31
class AnimationType(Enum):
    """Animation type for displaying progress.

    Attributes:
        PROGRESS_BAR: Fills from 0% to 100%.
        INFINITE: Running indicator (snake).
        NONE: No animation.
    """

    PROGRESS_BAR = "progress_bar"
    INFINITE = "infinite"
    NONE = "none"