fix some logic around status_inline_update keyboard

Author: nlef

bot/klippy.py

@@ -392,7 +392,7 @@ async def check_connection(self) -> str:
             except Exception as ex:
                 logger.error(ex, exc_info=True)
             retries += 1
-            time.sleep(1)
+            await asyncio.sleep(1)
         return f"Connection failed. {last_reason}"
 
     def update_sensor(self, name: str, value) -> None:

bot/main.py

@@ -17,7 +17,6 @@
 import subprocess
 import sys
 import tarfile
-import time
 from typing import Any, Dict, List, Optional, Union
 from zipfile import ZipFile
 
@@ -151,24 +150,9 @@ async def status_no_confirm(effective_message: Message) -> None:
     is_inline_button_press = effective_message.from_user is not None and effective_message.from_user.id == effective_message.get_bot().id
     if klippy.printing and not configWrap.notifications.group_only:
         notifier.update_status()
-        time.sleep(configWrap.camera.light_timeout + 1.5)
-        if not is_inline_button_press:
-            await effective_message.delete()
     else:
         text = await klippy.get_status()
-        inline_keyboard = None
-        if configWrap.telegram_ui.status_update_button:
-            inline_keyboard = InlineKeyboardMarkup(
-                [
-                    [
-                        InlineKeyboardButton(
-                            text="Update",
-                            callback_data="updstatus",
-                        )
-                    ]
-                ]
-            )
-        message = TelegramMessageRepr(text, parse_mode=ParseMode.HTML, silent=notifier.silent_commands, reply_markup=inline_keyboard)
+        message = TelegramMessageRepr(text, parse_mode=ParseMode.HTML, silent=notifier.silent_commands, reply_markup=notifier.get_status_keyboard())
         if cameraWrap.enabled:
             loop_loc = asyncio.get_running_loop()
             with await loop_loc.run_in_executor(executors_pool, cameraWrap.take_photo) as bio:

bot/notifications.py

@@ -129,22 +129,38 @@ def interval(self, new_value: int) -> None:
             self._interval = new_value
             self._reschedule_notifier_timer()
 
+    def get_status_keyboard(self, finish: bool = False) -> Optional[InlineKeyboardMarkup]:
+        inline_keyboard = None
+        if self._use_status_update_button and not finish:
+            inline_keyboard = InlineKeyboardMarkup(
+                [
+                    [
+                        InlineKeyboardButton(
+                            text="Update",
+                            callback_data="updstatus",
+                        )
+                    ]
+                ]
+            )
+        return inline_keyboard
+
+    async def _send_bzz_message(self, message: TelegramMessageRepr) -> None:
+        if self._progress_update_message:
+            if self._bzz_mess_id != 0:
+                try:
+                    await self._bot.delete_message(self._chat_id, self._bzz_mess_id)
+                except BadRequest as badreq:
+                    logger.warning("Failed deleting bzz message \n%s", badreq)
+                    self._bzz_mess_id = 0
+            mes = await self._bot.send_message(self._chat_id, text="Status has been updated\nThis message will be deleted", disable_notification=message.is_silent())
+            self._bzz_mess_id = mes.message_id
+
     async def _send_message(self, message: TelegramMessageRepr, group_only: bool = False, manual: bool = False) -> None:
         if not group_only:
             if self._status_message and not manual:
-                if self._bzz_mess_id != 0:
-                    try:
-                        await self._bot.delete_message(self._chat_id, self._bzz_mess_id)
-                    except BadRequest as badreq:
-                        logger.warning("Failed deleting bzz message \n%s", badreq)
-                        self._bzz_mess_id = 0
                 await message.update_existing(self._status_message)
-
-                if self._progress_update_message:
-                    mes = await self._bot.send_message(self._chat_id, text="Status has been updated\nThis message will be deleted", disable_notification=message.is_silent())
-                    self._bzz_mess_id = mes.message_id
+                await self._send_bzz_message(message)
             else:
-
                 sent_message = await message.send(self._bot, self._chat_id)
                 if not self._status_message and not manual:
                     self._status_message = sent_message
@@ -165,18 +181,8 @@ async def _send_photo(self, message: TelegramMessageRepr, group_only: bool = Fal
         with await loop.run_in_executor(self._executors_pool, self._cam_wrap.take_photo) as photo:
             if not group_only:
                 if self._status_message and not manual:
-                    if self._bzz_mess_id != 0:
-                        try:
-                            await self._bot.delete_message(self._chat_id, self._bzz_mess_id)
-                        except BadRequest as badreq:
-                            logger.warning("Failed deleting bzz message \n%s", badreq)
-                            self._bzz_mess_id = 0
                     await message.update_existing(self._status_message, photo=photo)
-
-                    if self._progress_update_message:
-                        mes = await self._bot.send_message(self._chat_id, text="Status has been updated\nThis message will be deleted", disable_notification=message.is_silent())
-                        self._bzz_mess_id = mes.message_id
-
+                    await self._send_bzz_message(message)
                 else:
                     sent_message = await message.send(self._bot, self._chat_id, photo=photo)
                     if not self._status_message and not manual:
@@ -198,12 +204,12 @@ async def _send_photo(self, message: TelegramMessageRepr, group_only: bool = Fal
 
     async def _notify(self, message: TelegramMessageRepr, group_only: bool = False, manual: bool = False, finish: bool = False) -> None:
         try:
-            if not self._cam_wrap.enabled:
-                await self._send_message(message, manual)
+            if self._cam_wrap.enabled:
+                await self._send_photo(message, group_only=group_only, manual=manual)
             else:
-                await self._send_photo(message, group_only, manual)
+                await self._send_message(message, group_only=group_only, manual=manual)
         except Exception as ex:
-            logger.error(ex)
+            logger.error(ex, exc_info=True, stack_info=True)
         finally:
             if finish:
                 await self.reset_notifications()
@@ -326,23 +332,11 @@ def _schedule_notification(self, message: str = "", schedule: bool = False, fini
         if "last_update_time" in self._message_parts:
             mess += f"_Last update at {datetime.now():%H:%M:%S}_"
 
-        inline_keyboard = None
-        if self._use_status_update_button and not finish:
-            inline_keyboard = InlineKeyboardMarkup(
-                [
-                    [
-                        InlineKeyboardButton(
-                            text="Update",
-                            callback_data="updstatus",
-                        )
-                    ]
-                ]
-            )
         tg_message = TelegramMessageRepr(
             text=mess,
             silent=self._silent_progress,
             suppress_escaping=True,
-            reply_markup=inline_keyboard,
+            reply_markup=self.get_status_keyboard(finish=finish),
         )
 
         self._sched.add_job(
@@ -425,34 +419,45 @@ async def stop_all(self) -> None:
         await self.reset_notifications()
         self.remove_notifier_timer()
 
+    # Todo: refactor with TelegramMessageRepr class
     async def _send_print_start_info(self) -> None:
         message, bio = await self._klippy.get_file_info("Printer started printing")
+
         if bio is not None:
-            status_message = await self._bot.send_photo(
-                self._chat_id,
-                photo=bio,
-                caption=message,
-                disable_notification=self.silent_status,
-            )
+            if not self._group_only:
+                status_message = await self._bot.send_photo(
+                    self._chat_id,
+                    photo=bio,
+                    caption=message,
+                    reply_markup=self.get_status_keyboard(),
+                    disable_notification=self.silent_status,
+                )
+                self._status_message = status_message
+
             for group_, message_thread_id in self._notify_groups:
                 bio.seek(0)
                 self._groups_status_messages[group_] = await self._bot.send_photo(
                     chat_id=group_,
                     message_thread_id=message_thread_id,
                     photo=bio,
                     caption=message,
+                    reply_markup=self.get_status_keyboard(),
                     disable_notification=self.silent_status,
                 )
             bio.close()
         else:
-            status_message = await self._bot.send_message(chat_id=self._chat_id, text=message, disable_notification=self.silent_status)
+            if not self._group_only:
+                status_message = await self._bot.send_message(chat_id=self._chat_id, text=message, reply_markup=self.get_status_keyboard(), disable_notification=self.silent_status)
+                self._status_message = status_message
+
             for group_, message_thread_id in self._notify_groups:
-                self._groups_status_messages[group_] = await self._bot.send_message(chat_id=group_, message_thread_id=message_thread_id, text=message, disable_notification=self.silent_status)
-        self._status_message = status_message
+                self._groups_status_messages[group_] = await self._bot.send_message(
+                    chat_id=group_, message_thread_id=message_thread_id, text=message, reply_markup=self.get_status_keyboard(), disable_notification=self.silent_status
+                )
 
-        if self._pin_status_single_message:
+        if self._pin_status_single_message and self._status_message is not None:
             await self._bot.unpin_all_chat_messages(self._chat_id)
-            await self._bot.pin_chat_message(self._chat_id, status_message.message_id, disable_notification=self.silent_status)
+            await self._bot.pin_chat_message(self._chat_id, self._status_message.message_id, disable_notification=self.silent_status)
 
     def send_print_start_info(self) -> None:
         if self._enabled:

bot/websocket_helper.py

@@ -310,7 +310,7 @@ def power_device_state(self, device):
             self._klippy.light_device.device_state = device_state
 
     async def websocket_to_message(self, ws_message):
-        logger.debug(ws_message)
+        # logger.debug(ws_message)
         json_message = orjson.loads(ws_message)
 
         if "error" in json_message:

scripts/requirements.dev.txt

@@ -3,7 +3,7 @@ black==25.1.0
 identify==2.6.6
 isort==6.0.0
 memory_profiler==0.61.0
-mypy==1.14.1
+mypy==1.15.0
 numpy~=2.2.2 ; python_version>='3.10'
 numpy~=2.0.2 ; python_version=='3.9'
 opencv-python~=4.11.0.86