Prepare 1.10.0 release

Author: jacopotediosi

octoprint_telegram/__init__.py

@@ -0,0 +1,31 @@
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .. import TelegramPlugin
+
+
+class CommandContext:
+    def __init__(
+        self, cmd: str, chat_id: str, from_id: str, parameter: str = "", msg_id_to_update: str = "", user: str = ""
+    ):
+        self.cmd = cmd
+        self.chat_id = chat_id
+        self.from_id = from_id
+        self.parameter = parameter
+        self.msg_id_to_update = msg_id_to_update
+        self.user = user
+
+
+class BaseCommand(ABC):
+    def __init__(self, main: "TelegramPlugin"):
+        self.main = main
+        self._logger = main._logger.getChild("Commands")
+
+    def __call__(self, cmd, chat_id, from_id, parameter="", msg_id_to_update="", user=""):
+        context = CommandContext(cmd, chat_id, from_id, parameter, msg_id_to_update, user)
+        return self.execute(context)
+
+    @abstractmethod
+    def execute(self, context: CommandContext):
+        pass

octoprint_telegram/commands/__init__.py

@@ -0,0 +1,46 @@
+from ..emoji import Emoji
+from .base import BaseCommand, CommandContext
+
+render_emojis = Emoji.render_emojis
+
+
+class CmdAbort(BaseCommand):
+    def execute(self, context: CommandContext):
+        if context.parameter == "stop":
+            self.main._printer.cancel_print(user=context.user)
+
+            msg = render_emojis("{emo:check} Aborting the print.")
+
+            self.main.send_msg(
+                msg,
+                chatID=context.chat_id,
+                msg_id=context.msg_id_to_update,
+            )
+        else:
+            if self.main._printer.is_printing() or self.main._printer.is_pausing() or self.main._printer.is_paused():
+                msg = render_emojis("{emo:question} Really abort the currently running print?")
+
+                command_buttons = [
+                    [
+                        [
+                            render_emojis("{emo:check} Stop print"),
+                            f"{context.cmd}_stop",
+                        ],
+                        [render_emojis("{emo:cancel} Close"), "close"],
+                    ]
+                ]
+
+                self.main.send_msg(
+                    msg,
+                    responses=command_buttons,
+                    chatID=context.chat_id,
+                    msg_id=context.msg_id_to_update,
+                )
+            else:
+                msg = render_emojis("{emo:warning} Currently I'm not printing, so there is nothing to stop.")
+
+                self.main.send_msg(
+                    msg,
+                    chatID=context.chat_id,
+                    msg_id=context.msg_id_to_update,
+                )

octoprint_telegram/commands/base.py

@@ -0,0 +1,74 @@
+import html
+
+from ..emoji import Emoji
+from .base import BaseCommand, CommandContext
+
+render_emojis = Emoji.render_emojis
+
+
+class CmdCancelObject(BaseCommand):
+    def execute(self, context: CommandContext):
+        cancelobject_id = "cancelobject"
+
+        if not self.main._plugin_manager.get_plugin(cancelobject_id, True):
+            msg = render_emojis(
+                f"{{emo:attention}} Please install <a href='https://plugins.octoprint.org/plugins/{cancelobject_id}/'>Cancelobject</a> plugin."
+            )
+            self.main.send_msg(
+                msg,
+                chatID=context.chat_id,
+                markup="HTML",
+                msg_id=context.msg_id_to_update,
+            )
+            return
+
+        if context.parameter:
+            params = context.parameter.split("_")
+
+            id = params[0]
+            self.main.send_octoprint_simpleapi_command(cancelobject_id, "cancel", dict(cancelled=id))
+
+            msg = render_emojis("{emo:check} Command sent!")
+            command_buttons = [[[render_emojis("{emo:back} Back"), context.cmd]]]
+
+            self.main.send_msg(
+                msg,
+                chatID=context.chat_id,
+                markup="HTML",
+                responses=command_buttons,
+                msg_id=context.msg_id_to_update,
+            )
+        else:
+            objlist = self.main.send_octoprint_simpleapi_command(cancelobject_id, "objlist").json().get("list", [])
+            if objlist:
+                msg = render_emojis("{emo:question} Which object do you want to cancel?")
+
+                cancelled_objects = [obj["object"] for obj in objlist if obj.get("cancelled", False)]
+                if cancelled_objects:
+                    msg += "\n\nObjects already cancelled:\n"
+                    msg += "\n".join(f"- <code>{html.escape(object_name)}</code>" for object_name in cancelled_objects)
+
+                command_buttons = [
+                    [[obj["object"], f"{context.cmd}_{obj['id']}"]]
+                    for obj in objlist
+                    if not obj.get("cancelled", False)
+                ]
+                command_buttons.append([[render_emojis("{emo:cancel} Close"), "close"]])
+
+                self.main.send_msg(
+                    msg,
+                    chatID=context.chat_id,
+                    markup="HTML",
+                    responses=command_buttons,
+                    msg_id=context.msg_id_to_update,
+                )
+            else:
+                msg = render_emojis(
+                    "{emo:attention} No objects found. Please make sure you've selected for printing the gcode."
+                )
+                self.main.send_msg(
+                    msg,
+                    chatID=context.chat_id,
+                    markup="HTML",
+                    msg_id=context.msg_id_to_update,
+                )

octoprint_telegram/commands/cmd_abort.py

@@ -0,0 +1,17 @@
+from .base import BaseCommand, CommandContext
+
+
+class CmdClose(BaseCommand):
+    def execute(self, context: CommandContext):
+        # According to https://core.telegram.org/bots/api#deletemessage:
+        # - A message can only be deleted if it was sent less than 48 hours ago.
+        # The try-except block handles this condition.
+        try:
+            if context.msg_id_to_update:
+                self.main.telegram_utils.send_telegram_request(
+                    f"{self.main.bot_url}/deleteMessage",
+                    "post",
+                    data=dict(chat_id=context.chat_id, message_id=context.msg_id_to_update),
+                )
+        except Exception:
+            pass