import asyncio import os import time import traceback import warnings from concurrent.futures import ThreadPoolExecutor from datetime import datetime from typing import cast, List import aiohttp import cv2 import numpy as np import telegram from aiohttp import BasicAuth from cv2 import aruco from pytapo import Tapo from telegram import Update, Message from telegram.ext import Updater from urllib3.exceptions import InsecureRequestWarning warnings.filterwarnings("ignore", category=InsecureRequestWarning) class Bot: def __init__(self): self.token = os.environ["BOT_TOKEN"] self.camera_ip = os.environ["CAMERA_IP"] self.camera_user = os.environ["CAMERA_USER"] self.camera_password = os.environ["CAMERA_PASSWORD"] self.chat_id = int(os.environ["CHAT_ID"]) self.profile_name = os.environ.get("CAMERA_PROFILE_NAME", "board") self.openhab_url = os.environ["OPENHAB_URL"] self.openhab_token = os.environ["OPENHAB_TOKEN"] self.openhab_item = os.environ["OPENHAB_ITEM"] self.bot = telegram.Bot(token=self.token) self.me = None self.tapo = Tapo(self.camera_ip, self.camera_user, self.camera_password) self.executor = ThreadPoolExecutor(max_workers=len(os.sched_getaffinity(0))) self.last_aruco_corners = {} def _get_presets(self): presets = self.tapo.getPresets() return {v: k for k, v in presets.items()} async def _get_item_state(self): url = f"{self.openhab_url}/rest/items/{self.openhab_item}/state" # use aiohttp instead of requests to avoid blocking async with aiohttp.ClientSession() as session: async with session.get(url, auth=BasicAuth(self.openhab_token, "")) as resp: return await resp.text() async def _send_item_command(self, command): print(f"Sending command {command}") url = f"{self.openhab_url}/rest/items/{self.openhab_item}" async with aiohttp.ClientSession() as session: async with session.post( url, auth=BasicAuth(self.openhab_token, ""), data=command, headers={"Content-Type": "text/plain", "Accept": "*/*"}, ) as resp: return await resp.text() def _take_photo_blocking( self, adjust_perspective=True, timeout: float = 5.0, ) -> List[cv2.typing.MatLike]: privacy_mode = self.tapo.getPrivacyMode() # Prepare the camera print("Disabling privacy mode and setting auto day/night mode...") self.tapo.setPrivacyMode(False) self.tapo.setDayNightMode("auto") time.sleep(1) # Take the color image vcap = cv2.VideoCapture( f"rtsp://{self.camera_user}:{self.camera_password}@{self.camera_ip}:554/stream1" ) print("Taking color image...") ret, pretty_image = vcap.read() while pretty_image is None: ret, pretty_image = vcap.read() if not adjust_perspective: self.tapo.setPrivacyMode(privacy_mode) return [pretty_image] self.tapo.setDayNightMode("on") # Iterate until we find all 4 aruco markers or timeout aruco_corners = {} annotated_image = None aruco_dict = aruco.getPredefinedDictionary(aruco.DICT_6X6_250) aruco_params = aruco.DetectorParameters() t0 = time.time() print("Taking image with ArUco markers...") while len(aruco_corners) < 4: if time.time() - t0 > timeout: a = self.last_aruco_corners.copy() a.update(aruco_corners) aruco_corners = a if len(aruco_corners) == 4: print("Timeout waiting for ArUco markers, using cached corners") break print( "Timeout waiting for ArUco markers, returning only original image" ) self.tapo.setPrivacyMode(privacy_mode) self.tapo.setDayNightMode("auto") return [pretty_image] ret, annotated_image = vcap.read() if not ret: continue # Detect the markers corners, ids, rejected = aruco.detectMarkers( annotated_image, aruco_dict, parameters=aruco_params ) if corners is not None and ids is not None: for corner, i in zip(corners, ids): aruco_corners[i[0]] = corner assert annotated_image is not None del vcap print( "Found all ArUco markers, enabled privacy mode and set auto day/night mode..." ) self.tapo.setDayNightMode("auto") self.tapo.setPrivacyMode(privacy_mode) corners = [aruco_corners[i] for i in range(1, 5)] ids = np.array([[i] for i in range(1, 5)]) aruco.drawDetectedMarkers(annotated_image, corners, ids) # Annotate the image with the detected markers and apply the perspective transform to the pretty image # Get the outermost points of each marker bl_marker = corners[0].squeeze()[3] # bottom left marker tl_marker = corners[1].squeeze()[0] # top left marker tr_marker = corners[2].squeeze()[1] # top right marker bc_marker = corners[3].squeeze()[3] # bottom center marker # Calculate the fourth point by computing the line through the bottom markers and intersecting with the vertical # line through the top right marker slope = (bc_marker[1] - bl_marker[1]) / (bc_marker[0] - bl_marker[0]) y_intersection = slope * (tr_marker[0] - bc_marker[0]) + bc_marker[1] fourth_point = [tr_marker[0], y_intersection] rectangle_points = np.array( [tl_marker, bl_marker, fourth_point, tr_marker], dtype="float32" ) # Expand the rectangle slightly centroid = np.mean(rectangle_points, axis=0) expanded_rectangle_points = ( rectangle_points + (rectangle_points - centroid) * 0.025 ) # Draw the expanded rectangle on the annotated image cv2.polylines( annotated_image, [np.int32(expanded_rectangle_points)], isClosed=True, color=(0, 255, 0), thickness=3, ) # Define destination points for perspective transform, maintaining a 3:2 aspect ratio width, height = 300 * 5, 200 * 5 dst_pts = np.array( [[0, 0], [0, height], [width, height], [width, 0]], dtype="float32" ) matrix = cv2.getPerspectiveTransform(expanded_rectangle_points, dst_pts) warped = cv2.warpPerspective(pretty_image, matrix, (width, height)) self.last_aruco_corners = aruco_corners return [warped, annotated_image] async def _photo_command(self, chat_id: int, adjust_perspective: bool = True): await self.bot.send_chat_action(chat_id=chat_id, action="upload_photo") photos = await self.take_photo(adjust_perspective=adjust_perspective) jpegs = [cv2.imencode(".jpg", photo)[1].tobytes() for photo in photos] media = [ telegram.InputMediaPhoto( media=telegram.InputFile(jpeg, filename=f"photo{i}.jpg", attach=True), filename=f"photo{i}.jpg", ) for i, jpeg in enumerate(jpegs) ] await self.bot.send_media_group( chat_id=chat_id, media=media, caption=str(datetime.now().strftime("%A, %B %d, %Y %H:%M:%S")), ) async def parse_message(self, msg: Message): match msg.text: case "/start": await self.bot.send_message( chat_id=msg.chat_id, text="Hello, I'm a bot that can send you a photo from the camera.", ) case "/reposition": privacy_mode = self.tapo.getPrivacyMode() self.tapo.setPrivacyMode(False) presets = self._get_presets() if self.profile_name not in presets: await self.bot.send_message( chat_id=msg.chat_id, text=f"Profile '{self.profile_name}' not found", ) return self.tapo.setPreset(presets[self.profile_name]) message = await self.bot.send_message( chat_id=msg.chat_id, text=f"Repositioning to profile '{self.profile_name}'...", ) await asyncio.sleep(5) self.tapo.setPrivacyMode(privacy_mode) await self.bot.edit_message_text( chat_id=msg.chat_id, message_id=message.message_id, text=f"Repositioning complete", ) case "/calibrate": privacy_mode = self.tapo.getPrivacyMode() self.tapo.setPrivacyMode(False) await asyncio.sleep(0.3) self.tapo.calibrateMotor() message = await self.bot.send_message( chat_id=msg.chat_id, text=f"Calibrating, this will take ~25s...", ) await asyncio.sleep(26) await self.bot.edit_message_text( chat_id=msg.chat_id, message_id=message.message_id, text=f"Calibration complete", ) self.tapo.setPrivacyMode(privacy_mode) case "/light_on": await self._send_item_command("ON") await self.bot.send_message( chat_id=msg.chat_id, text=f"Light turned on", ) case "/light_off": await self._send_item_command("OFF") await self.bot.send_message( chat_id=msg.chat_id, text=f"Light turned off", ) case "/light_status": state = await self._get_item_state() await self.bot.send_message( chat_id=msg.chat_id, text=f"Light is {state}", ) case "/photo": await self._photo_command(msg.chat_id, adjust_perspective=True) case "/photo_unprocessed": await self._photo_command(msg.chat_id, adjust_perspective=False) case "/privacy_on": self.tapo.setPrivacyMode(True) await self.bot.send_message( chat_id=msg.chat_id, text=f"Privacy mode turned on", ) case "/privacy_off": self.tapo.setPrivacyMode(False) await self.bot.send_message( chat_id=msg.chat_id, text=f"Privacy mode turned off", ) case "/privacy_status": state = self.tapo.getPrivacyMode() await self.bot.send_message( chat_id=msg.chat_id, text=f"Privacy mode is {state and 'enabled' or 'disabled'}", ) async def take_photo( self, adjust_perspective=True, timeout=5.0 ) -> List[cv2.typing.MatLike]: item_state = await self._get_item_state() if item_state == "OFF": print("Turning light on") await self._send_item_command("ON") try: return await asyncio.get_event_loop().run_in_executor( self.executor, self._take_photo_blocking, adjust_perspective, timeout ) finally: if item_state == "OFF": print("Turning light back off") await self._send_item_command("OFF") async def run(self): async with self.bot: self.me = await self.bot.get_me() updater = Updater(bot=self.bot, update_queue=asyncio.Queue()) async with updater: queue = await updater.start_polling(allowed_updates=[Update.MESSAGE]) print("Bot is up and running") while True: # noinspection PyBroadException try: upd = cast(Update, await queue.get()) print(upd) if not upd.message or upd.message.chat_id != self.chat_id: print("Ignoring message") continue # noinspection PyBroadException try: await self.parse_message(upd.message) except Exception: traceback.print_exc() exc = traceback.format_exc() await self.bot.send_message( chat_id=self.chat_id, text=f"Error: {exc}", ) except Exception: traceback.print_exc() if __name__ == "__main__": bot = Bot() asyncio.run(bot.run())