simplify API

Author: dsammaruga

src/arduino/app_bricks/telegram_bot/__init__.py

@@ -2,9 +2,6 @@
 #
 # SPDX-License-Identifier: MPL-2.0
 
-from .telegram_bot import TelegramBot
-from telegram import Update
-from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
+from .telegram_bot import TelegramBot, Message
 
-
-__all__ = ["TelegramBot", "Update", "Application", "CommandHandler", "MessageHandler", "filters", "ContextTypes"]
+__all__ = ["TelegramBot", "Message"]

src/arduino/app_bricks/telegram_bot/telegram_bot.py

@@ -5,9 +5,9 @@
 import os
 import asyncio
 import threading
-import inspect
 import time
 from typing import Callable, Optional
+from dataclasses import dataclass
 from arduino.app_utils import brick, Logger
 from telegram import Update, BotCommand
 from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
@@ -16,6 +16,30 @@
 logger = Logger("TelegramBot")
 
 
+@dataclass
+class Message:
+    """Simplified message object with only essential attributes.
+
+    This object is passed to user callbacks, containing all the information
+    needed to process a message and respond to it.
+
+    Attributes:
+        chat_id: Telegram chat ID (used to send responses).
+        text: Text content of the message (None for photos).
+        user_id: ID of the user who sent the message.
+        user_name: First name of the user.
+        username: Username of the user (None if not set).
+        photo_bytes: Photo data as bytes (None for text messages).
+    """
+
+    chat_id: int
+    text: Optional[str] = None
+    user_id: Optional[int] = None
+    user_name: Optional[str] = None
+    username: Optional[str] = None
+    photo_bytes: Optional[bytearray] = None
+
+
 @brick
 class TelegramBot:
     """A brick to manage Telegram Bot interactions with synchronous API.
@@ -66,73 +90,121 @@ def __init__(
         self._scheduled_tasks: dict[str, threading.Timer] = {}
         self._commands_registry: dict[str, str] = {}
 
-    def _make_async_handler(self, callback: Callable) -> Callable:
-        """Convert a synchronous callback to an async handler if needed.
+    def _create_message_handler(self, callback: Callable) -> Callable:
+        """Create a Telegram handler from user's simple callback.
+
+        This method extracts essential information from Telegram's Update object,
+        creates a simplified Message object, and handles async-to-sync conversion
+        automatically. Photos are downloaded automatically if present.
+
+        User's callback receives only a simplified Message object, not Update/Context.
 
         Args:
-            callback: User-defined callback function (sync or async).
+            callback: User's synchronous callback(message: Message) -> None
 
         Returns:
-            Async-compatible callback function.
+            Async handler compatible with python-telegram-bot
         """
-        if inspect.iscoroutinefunction(callback):
-            # Already async, use as-is
-            return callback
 
-        # Sync callback, wrap it
-        async def async_wrapper(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        async def wrapper(update: Update, context: ContextTypes.DEFAULT_TYPE):
+            # Extract essential info from Update into simplified Message object
+            message = Message(
+                chat_id=update.message.chat_id,
+                text=update.message.text if update.message.text else None,
+                user_id=update.effective_user.id,
+                user_name=update.effective_user.first_name,
+                username=update.effective_user.username,
+            )
+
+            # Download photo if present
+            if update.message.photo:
+                try:
+                    photo_file = await update.message.photo[-1].get_file()
+                    message.photo_bytes = await photo_file.download_as_bytearray()
+                except Exception as e:
+                    logger.error(f"Failed to download photo: {e}")
+
+            # Run user's callback in executor (sync)
             loop = asyncio.get_event_loop()
-            await loop.run_in_executor(None, callback, update, context)
+            await loop.run_in_executor(None, callback, message)
 
-        return async_wrapper
+        return wrapper
 
-    def add_command(self, command: str, callback: Callable, description: str = "") -> None:
-        """Register a slash command handler with optional description.
+    def add_command(self, command: str, callback: Callable[[Message], None], description: str = "") -> None:
+        """Register a command handler (e.g., /start).
 
-        The callback can be either synchronous or asynchronous. If synchronous,
-        it will be automatically converted to async. If a description is provided
-        and auto_set_commands is enabled, the command will appear in Telegram's
-        command menu when users type '/'.
+        The callback function receives a simplified Message object containing
+        all essential information about the message and user.
 
         Args:
-            command: Command name (without the leading slash, e.g., "start" for /start).
-            callback: Handler function (can be sync or async). Receives Update and ContextTypes.
-            description: Optional description shown in Telegram's command menu. If empty,
-                the command will still work but won't appear in the '/' menu.
+            command: Command name without '/' (e.g., "start", "hello").
+            callback: Function that receives a Message object.
+            description: Optional description shown in Telegram's command menu.
 
         Example:
-            >>> bot.add_command("start", start_handler, "Start the bot")
-            >>> bot.add_command("help", help_handler, "Show available commands")
+            >>> def greet(msg: Message):
+            ...     bot.send(msg.chat_id, f"Hello {msg.user_name}!")
+            >>> bot.add_command("hello", greet, "Greet the user")
         """
-        async_callback = self._make_async_handler(callback)
-        self.application.add_handler(CommandHandler(command, async_callback))
-        
-        # Track command for auto-sync with Telegram
+        handler = self._create_message_handler(callback)
+        self.application.add_handler(CommandHandler(command, handler))
+
         if description:
             self._commands_registry[command] = description
-            logger.info(f"Registered command /{command}: {description}")
-        else:
-            logger.info(f"Registered command /{command}")
 
-    def on_text(self, callback: Callable) -> None:
+        logger.info(f"Registered command: /{command}" + (f" - {description}" if description else ""))
+
+    def on_text(self, callback: Callable[[Message], None]) -> None:
         """Register a handler for text messages.
 
+        The callback function receives a simplified Message object containing
+        the text and user information.
+
         Args:
-            callback: Handler function (can be sync or async). Receives Update and ContextTypes.
-                Called for all text messages except commands.
+            callback: Function that receives a Message object.
+
+        Example:
+            >>> def echo(msg: Message):
+            ...     bot.send(msg.chat_id, f"You said: {msg.text}")
+            >>> bot.on_text(echo)
         """
-        async_callback = self._make_async_handler(callback)
-        self.application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, async_callback))
+        handler = self._create_message_handler(callback)
+        self.application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handler))
+        logger.info("Registered text message handler")
 
-    def on_photo(self, callback: Callable) -> None:
+    def on_photo(self, callback: Callable[[Message], None]) -> None:
         """Register a handler for photo messages.
 
+        The callback function receives a simplified Message object with the
+        photo already downloaded as photo_bytes.
+
         Args:
-            callback: Handler function (can be sync or async). Receives Update and ContextTypes.
-                Called when user sends a photo.
+            callback: Function that receives a Message object with photo_bytes.
+
+        Example:
+            >>> def handle_photo(msg: Message):
+            ...     if msg.photo_bytes:
+            ...         bot.send(msg.chat_id, "Got your photo!")
+            >>> bot.on_photo(handle_photo)
+        """
+        handler = self._create_message_handler(callback)
+        self.application.add_handler(MessageHandler(filters.PHOTO, handler))
+        logger.info("Registered photo message handler")
+
+    def send(self, chat_id: int, text: str) -> bool:
+        """Send a text message to a chat (simplified method).
+
+        Args:
+            chat_id: Telegram chat ID.
+            text: Message text.
+
+        Returns:
+            True if message was sent successfully, False otherwise.
+
+        Example:
+            >>> bot.send(123456, "Hello from Arduino!")
         """
-        async_callback = self._make_async_handler(callback)
-        self.application.add_handler(MessageHandler(filters.PHOTO & ~filters.COMMAND, async_callback))
+        return self.send_message(chat_id, text)
 
     def send_message(self, chat_id: int, message_text: str) -> bool:
         """Send a text message to a specific chat (synchronous with automatic retry).
@@ -193,24 +265,28 @@ async def _send_message_async(self, chat_id: int, message_text: str) -> None:
             logger.error(f"An error occurred: {e}")
             raise
 
-    def send_photo(self, chat_id: int, photo, caption: Optional[str] = None) -> bool:
-        """Send a photo to a specific chat (synchronous with automatic retry).
+    def send_photo(self, chat_id: int, photo_bytes: bytes, caption: str = "") -> bool:
+        """Send a photo to a chat.
 
         Args:
-            chat_id: Telegram chat ID to send the photo to.
-            photo: Photo to send (file path, URL, or file-like object).
-            caption: Optional caption for the photo.
+            chat_id: Telegram chat ID.
+            photo_bytes: Photo as bytes.
+            caption: Optional caption text.
 
         Returns:
-            True if photo was sent successfully, False otherwise.
+            True if successful, False otherwise.
+
+        Example:
+            >>> with open("image.jpg", "rb") as f:
+            ...     bot.send_photo(123456, f.read(), "Check this out!")
         """
         if not self._running or not self._loop or not self._initialized:
             logger.error("Bot not properly initialized, cannot send photo")
             return False
 
         for attempt in range(self.max_retries):
             try:
-                future = asyncio.run_coroutine_threadsafe(self._send_photo_async(chat_id, photo, caption), self._loop)
+                future = asyncio.run_coroutine_threadsafe(self._send_photo_async(chat_id, photo_bytes, caption), self._loop)
                 future.result(timeout=self.photo_timeout)
                 return True
             except TimeoutError:
@@ -225,13 +301,13 @@ def send_photo(self, chat_id: int, photo, caption: Optional[str] = None) -> bool
         logger.error(f"Failed to send photo after {self.max_retries} attempts")
         return False
 
-    async def _send_photo_async(self, chat_id: int, photo, caption: Optional[str] = None) -> None:
+    async def _send_photo_async(self, chat_id: int, photo_bytes: bytes, caption: str) -> None:
         """Internal async method to send a photo with network error handling.
 
         Args:
             chat_id: Telegram chat ID.
-            photo: Photo to send.
-            caption: Optional caption.
+            photo_bytes: Photo bytes to send.
+            caption: Photo caption.
 
         Raises:
             NetworkError: If network issues occur.
@@ -242,7 +318,7 @@ async def _send_photo_async(self, chat_id: int, photo, caption: Optional[str] =
         try:
             await self.application.bot.send_photo(
                 chat_id=chat_id,
-                photo=photo,
+                photo=photo_bytes,
                 caption=caption,
                 read_timeout=self.photo_timeout,
                 write_timeout=self.photo_timeout,
@@ -255,63 +331,33 @@ async def _send_photo_async(self, chat_id: int, photo, caption: Optional[str] =
             logger.error(f"An error occurred: {e}")
             raise
 
-    def get_photo(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> Optional[bytearray]:
-        """Download photo from an update (synchronous with automatic retry).
+    def schedule(self, chat_id: int, text: str, interval_seconds: int) -> str:
+        """Schedule recurring messages to a chat.
 
         Args:
-            update: Telegram update object containing the photo.
-            context: Telegram context object.
+            chat_id: Telegram chat ID.
+            text: Message text to send.
+            interval_seconds: Interval between messages in seconds.
 
         Returns:
-            Photo bytes as bytearray, or None if download fails after all retries.
-        """
-        if not self._running or not self._loop or not self._initialized:
-            logger.error("Bot not properly initialized, cannot get photo")
-            return None
-
-        for attempt in range(self.max_retries):
-            try:
-                future = asyncio.run_coroutine_threadsafe(self._get_photo_async(update, context), self._loop)
-                return future.result(timeout=self.photo_timeout)
-            except TimeoutError:
-                logger.warning(f"Photo download timeout (attempt {attempt + 1}/{self.max_retries})")
-                if attempt < self.max_retries - 1:
-                    time.sleep(2 * (attempt + 1))  # Backoff for downloads
-                    continue
-            except Exception as e:
-                logger.error(f"Failed to get photo: {e}")
-                return None
+            Task ID that can be used to cancel the scheduled message.
 
-        logger.error(f"Failed to download photo after {self.max_retries} attempts")
-        return None
+        Example:
+            >>> task_id = bot.schedule(123456, "Reminder!", 60)  # Every minute
+            >>> # Later: bot.cancel_schedule(task_id)
+        """
+        return self.schedule_message(chat_id, text, interval_seconds)
 
-    async def _get_photo_async(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> bytearray:
-        """Internal async method to download a photo with network error handling.
+    def cancel_schedule(self, task_id: str) -> bool:
+        """Cancel a scheduled message.
 
         Args:
-            update: Telegram update object.
-            context: Telegram context object.
+            task_id: Task ID returned by schedule().
 
         Returns:
-            Photo bytes as bytearray.
-
-        Raises:
-            NetworkError: If network issues occur.
-            TimedOut: If request times out.
-            Exception: If photo download fails for other reasons.
+            True if task was cancelled, False if task_id not found.
         """
-        logger.info("Downloading photo from Telegram...")
-        try:
-            photo_file = await update.message.photo[-1].get_file()
-            photo_bytes = await photo_file.download_as_bytearray()
-            logger.info("Photo downloaded successfully!")
-            return photo_bytes
-        except (NetworkError, TimedOut) as e:
-            logger.warning(f"Network issue while downloading photo: {e}")
-            raise
-        except Exception as e:
-            logger.error(f"An error occurred while downloading photo: {e}")
-            raise
+        return self.cancel_scheduled_message(task_id)
 
     async def _set_bot_commands(self) -> None:
         """Internal method to sync registered commands with Telegram.
@@ -327,10 +373,7 @@ async def _set_bot_commands(self) -> None:
             return
 
         try:
-            bot_commands = [
-                BotCommand(command=cmd, description=desc)
-                for cmd, desc in self._commands_registry.items()
-            ]
+            bot_commands = [BotCommand(command=cmd, description=desc) for cmd, desc in self._commands_registry.items()]
             await self.application.bot.set_my_commands(bot_commands)
             logger.info(f"Successfully registered {len(bot_commands)} command(s) with Telegram's menu")
         except Exception as e: