Fix bugs, add tests, and simplify WebSocket refactor

bot/klippy.py

@@ -28,23 +28,27 @@
 logger = logging.getLogger(__name__)
 
 
-class PrintState(Enum):
+class PrintState(str, Enum):
     """Klipper print job states."""
 
+    # Klipper print_stats.state values
     STANDBY = "standby"
-    START = "start"
     PRINTING = "printing"
-    FINISH = "finish"
+    PAUSED = "paused"
+    COMPLETE = "complete"
     CANCELLED = "cancelled"
     ERROR = "error"
+    # Internal bot notification states
+    NOTIFY_START = "start"
+    NOTIFY_FINISH = "finish"
 
     @property
     def is_start(self) -> bool:
-        return self == PrintState.START
+        return self == PrintState.NOTIFY_START
 
     @property
     def is_finished(self) -> bool:
-        return self in (PrintState.FINISH, PrintState.CANCELLED, PrintState.ERROR)
+        return self in (PrintState.NOTIFY_FINISH, PrintState.CANCELLED, PrintState.ERROR)
 
 
 class PowerDevice:
@@ -508,9 +512,9 @@ async def get_file_info(self, state: PrintState = PrintState.PRINTING) -> tuple[
         return await self._populate_with_thumb(self._thumbnail_path, message)
 
     _STATE_TITLES: Final = {
-        PrintState.START: "Printer started printing",
+        PrintState.NOTIFY_START: "Printer started printing",
         PrintState.PRINTING: "Printing",
-        PrintState.FINISH: "Finished printing",
+        PrintState.NOTIFY_FINISH: "Finished printing",
         PrintState.CANCELLED: "Cancelled printing",
         PrintState.ERROR: "Printing was interrupted with an error",
     }
@@ -549,7 +553,7 @@ def _get_printing_file_info(self, state: PrintState = PrintState.PRINTING) -> st
             duration_prefix = "Printed for" if state.is_finished else "Printing for"
             result += f"{duration_prefix} {timedelta(seconds=round(self.printing_duration))}\n"
 
-        if state in (PrintState.START, PrintState.PRINTING):
+        if state in (PrintState.NOTIFY_START, PrintState.PRINTING):
             eta = self._get_eta()
             if "eta" in self._message_parts:
                 result += f"Estimated time left: {eta}\n"

bot/notifications.py

@@ -426,15 +426,15 @@ async def stop_all(self) -> None:
 
     # TODO: refactor with TelegramMessageRepr class
     async def _send_print_start_info(self) -> None:
-        message, bio = await self._klippy.get_file_info(state=PrintState.START)
+        message, bio = await self._klippy.get_file_info(state=PrintState.NOTIFY_START)
 
         if not self._group_only:
             status_message = await self._bot.send_photo(
                 self._chat_id,
                 photo=bio,
                 caption=message,
                 parse_mode=ParseMode.HTML,
-                reply_markup=self.get_status_keyboard(state=PrintState.START),
+                reply_markup=self.get_status_keyboard(state=PrintState.NOTIFY_START),
                 disable_notification=self.silent_status,
             )
             self._status_message = status_message
@@ -447,7 +447,7 @@ async def _send_print_start_info(self) -> None:
                 photo=bio,
                 caption=message,
                 parse_mode=ParseMode.HTML,
-                reply_markup=self.get_status_keyboard(state=PrintState.START),
+                reply_markup=self.get_status_keyboard(state=PrintState.NOTIFY_START),
                 disable_notification=self.silent_status,
             )
         bio.close()
@@ -467,7 +467,7 @@ def send_print_start_info(self) -> None:
             )
 
     async def _send_print_finish(self) -> None:
-        self._schedule_notification(state=PrintState.FINISH)
+        self._schedule_notification(state=PrintState.NOTIFY_FINISH)
 
     def send_print_finish(self) -> None:
         if self._enabled:
@@ -646,7 +646,6 @@ def send_document(self, ws_message: str) -> None:
 
     async def parse_notification_params(self, message: str) -> None:
         mass_parts = message.split(sep=" ")
-        mass_parts.pop(0)
         response = ""
         for part in mass_parts:
             try:

bot/timelapse.py

@@ -543,7 +543,6 @@ async def restore_state(self) -> None:
 
     async def parse_timelapse_params(self, message: str) -> None:
         mass_parts = message.split(sep=" ")
-        mass_parts.pop(0)
         response = ""
         for part in mass_parts:
             try: