feat: Telegram Support, High-Resolution Capture (#14)

* feat: better camera resolution

* feat: new telegram method

* refactor: remove unnecessary stuff

* feat: send initial image

* refactor: configure for telegram send

* docs: update version

* refactor: apply flake8 suggestions

* fix: force reviewer to use toml file

* fix: install flake8pyproject

Author: electricalgorithm

.github/workflows/reviewer.yaml

@@ -17,7 +17,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install flake8 pylint
+        pip install flake8 pylint Flake8-pyproject
     - name: Checking the code using Flake8
       run: |
         flake8 $(git ls-files '*.py')

core/observers/observer/hss_observer.py

@@ -6,6 +6,8 @@
 from core.observers.subject.eye_subject import EyeSubject
 from core.observers.subject.wifi_subject import WiFiSubject
 from core.strategies.notifier.base_notifier_strategy import BaseNotifierStrategy
+from core.strategies.notifier.telegram_strategy import TelegramStrategy
+from core.strategies.notifier.whatsapp_strategy import WhatsappStrategy
 from core.utils.datatypes import EyeStates, WiFiStates
 from core.utils.fileio_adaptor import read_latest_file, upload_to_fileio
 from core.utils.logger import get_logger
@@ -35,10 +37,20 @@ def update(self, subject: BaseSubject) -> None:
 
         if self.wifi_state == WiFiStates.DISCONNECTED and self.eye_state == EyeStates.DETECTED:
             logger.info("There is an intruder!")
-            fileio_link = upload_to_fileio(
-                read_latest_file("~/.home-security-system/images")
-            )
-            self._notifier.notify_all(f"There is an intruder! Here is the image: {fileio_link}.")
+            if isinstance(self._notifier, WhatsappStrategy):
+                fileio_link = upload_to_fileio(
+                    read_latest_file("~/.home-security-system/images")
+                )
+                self._notifier.notify_all(
+                    f"There is an intruder! Here is the image: {fileio_link}."
+                )
+            elif isinstance(self._notifier, TelegramStrategy):
+                latest_file = read_latest_file("~/.home-security-system/images")
+                with open(latest_file, 'rb') as intruder_image:
+                    self._notifier.notify_all("There is an intruder! Here is the image:")
+                    self._notifier.send_image_all(intruder_image)
+            else:
+                logger.error("Notifier is not set!")
 
     def set_notifier(self, notifier: BaseNotifierStrategy) -> None:
         """This method is called when the observer is updated."""

core/observers/subject/eye_subject.py

@@ -6,7 +6,7 @@
 from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime
 from threading import Lock
-from time import sleep, time
+from time import sleep
 from typing import Optional
 
 import cv2
@@ -80,6 +80,12 @@ def _run_in_loop(self,
             logger.debug("[EyeSubject] A lock instance is generated.")
             wifi_lock = Lock()
 
+        # Save an initial image.
+        initial_frame = eye_strategy.get_frame()
+        file_location = f"{self._image_path}/initial_frame.jpg"
+        cv2.imwrite(file_location, initial_frame)
+        logger.debug("[EyeSubject] Initial frame has been saved.")
+
         while True:
             # If WiFi subject would give rights to use camera,
             # Check if any intruders detected.
@@ -119,10 +125,6 @@ def _save_image(self, result: EyeStrategyResult) -> None:
     def _cb_done(self, future) -> None:
         """This method is called when the observer is updated."""
         logger.warning("[EyeSubject] The thread died.")
-        # Create a txt file to indicate the thread died.
-        file_location = "thread_die.txt"
-        with open(file_location, "a", encoding="utf-8") as file:
-            file.write(f"The EyeSubject thread died. Time: {time()}")
 
         # Start the thread again.
         self.run(self._eye_strategy, self._wifi_lock)

core/observers/subject/wifi_subject.py

@@ -4,7 +4,7 @@
 """
 from concurrent.futures import Future, ThreadPoolExecutor
 from threading import Lock
-from time import sleep, time
+from time import sleep
 from typing import Optional
 
 from core.observers.subject.base_subject import BaseSubject
@@ -85,9 +85,6 @@ def _run_in_loop(self, wifi_strategy: BaseWiFiStrategy) -> None:
     def _cb_done(self, future) -> None:
         """This method is called when the observer is updated."""
         logger.warning("[WiFiSubject] The thread died.")
-        file_location = "thread_die.txt"
-        with open(file_location, "a", encoding="utf-8") as file:
-            file.write(f"The WiFiSubject thread died. Time: {time()}")
 
         # Run the thread again.
         self.run(self._wifi_strategy)

core/strategies/detectors/efficientdet_strategy.py

@@ -61,16 +61,16 @@ def detect_humans(cls, frame: numpy.ndarray) -> DetectorResult:
                 continue
 
             if (labels[int(pred_class)]) == 'person':
-                min_y = round(box[0] * image_height)
-                min_x = round(box[1] * image_width)
-                max_y = round(box[2] * image_height)
-                max_x = round(box[3] * image_width)
+                frame_height, frame_width = frame.shape[:2]
+                min_y = round(box[0] * frame_height)
+                min_x = round(box[1] * frame_width)
+                max_y = round(box[2] * frame_height)
+                max_x = round(box[3] * frame_width)
                 detection_regions.append((min_x, max_x, min_y, max_y))
-
-                cv2.rectangle(image, (min_x, min_y), (max_x, max_y), (0, 255, 0), 2)
+                cv2.rectangle(frame, (min_x, min_y), (max_x, max_y), (0, 255, 0), 5)
 
         result = DetectorResult(
-            image=image,
+            image=frame,
             human_found=len(detection_regions) > 0,
             regions=detection_regions,
             num_detections=len(detection_regions),

core/strategies/eye/picamera_strategy.py

@@ -1,6 +1,7 @@
 """
 The Camera strategy for eye strategies.
 """
+import cv2
 import numpy
 from picamera2 import Picamera2
 
@@ -32,11 +33,10 @@ def get_frame(self) -> numpy.ndarray:
         """This method returns the frame from the camera."""
         # Internal attributes
         picam2 = Picamera2()
-        picam2.configure(picam2.create_preview_configuration(
-            main={"format": 'XRGB8888', "size": (640, 480)}
-        ))
+        still_configuration = picam2.create_still_configuration(main={"format": 'XRGB8888'})
+        picam2.configure(still_configuration)
         picam2.start()
         frame = picam2.capture_array()
         picam2.close()
         del picam2
-        return frame
+        return cv2.rotate(frame, cv2.ROTATE_90_COUNTERCLOCKWISE)

core/strategies/notifier/telegram_strategy.py

@@ -0,0 +1,38 @@
+"""
+This module contains the Telegram notifier strategy.
+"""
+import telebot
+
+from core.strategies.notifier.base_notifier_strategy import BaseNotifierStrategy
+from core.utils.datatypes import TelegramReciever
+from core.utils.logger import get_logger
+
+# Get the logger instance.
+logger = get_logger(__name__)
+
+
+class TelegramStrategy(BaseNotifierStrategy):
+    """
+    The Telegram notifier strategy.
+    """
+
+    def __init__(self, api_key: str) -> 'TelegramStrategy':
+        """Initilizes the Telegram modifier strategy."""
+        self._bot = telebot.TeleBot(token=api_key,
+                                    threaded=True,
+                                    num_threads=2)
+        super().__init__()
+
+    def notify_all(self, message: str) -> None:
+        """This method is called when the notifier is updated."""
+        for reciever in self._recievers:
+            if isinstance(reciever, TelegramReciever):
+                logger.debug("Sending Telegram message (%s) to %s", message, reciever.chat_id)
+                self._bot.send_message(reciever.chat_id, message)
+
+    def send_image_all(self, image: bytes) -> None:
+        """This method is called when the notifier is updated."""
+        for reciever in self._recievers:
+            if isinstance(reciever, TelegramReciever):
+                logger.debug("Sending image to %s", reciever.chat_id)
+                self._bot.send_photo(reciever.chat_id, image)

core/utils/datatypes.py

@@ -64,3 +64,9 @@ class WhatsappReciever(NotifierReciever):
     """This class represents WhatsApp reciever."""
     telephone_number: str
     api_key: str
+
+
+@dataclass
+class TelegramReciever(NotifierReciever):
+    """This class represents Telegram reciever."""
+    chat_id: str

main.py

@@ -4,16 +4,19 @@
 import json
 import sys
 from concurrent.futures import wait
+from time import sleep
 from typing import Any
 
 from core.observers.observer.hss_observer import HomeSecuritySystemObserver
 from core.observers.subject.eye_subject import EyeSubject
 from core.observers.subject.wifi_subject import WiFiSubject
 from core.strategies.detectors.efficientdet_strategy import EfficientdetStrategy
 from core.strategies.eye.picamera_strategy import PiCameraStrategy
+from core.strategies.notifier.telegram_strategy import TelegramStrategy
 from core.strategies.notifier.whatsapp_strategy import WhatsappStrategy
 from core.strategies.wifi.admin_panel_strategy import AdminPanelStrategy
-from core.utils.datatypes import Protector, WhatsappReciever
+from core.utils.datatypes import Protector, TelegramReciever
+from core.utils.fileio_adaptor import upload_to_fileio
 
 
 def read_configurations() -> dict[str, Any]:
@@ -35,21 +38,20 @@ def main():
     # Read configurations.
     config, strategy_config = read_configurations()
     # Create a WhatsApp notifier.
-    whatsapp_notifier = WhatsappStrategy()
+    notifier = TelegramStrategy(strategy_config['telegram_strategy']['bot_key'])
     for reciever in config['recievers']:
-        whatsapp_notifier.add_reciever(WhatsappReciever(reciever['name'],
-                                                        reciever['tel_no'],
-                                                        reciever['callmebot_key']))
+        notifier.add_reciever(TelegramReciever(reciever['name'],
+                                               reciever['chat_id']))
 
     # Create a Protector within IpAddressStrategy.
-    ip_address_strategy = AdminPanelStrategy(strategy_config)
+    network_strategy = AdminPanelStrategy(strategy_config['admin_panel_strategy'])
     for protector in config['protectors']:
-        ip_address_strategy.add_protector(Protector(protector['name'],
-                                                    protector['address']))
+        network_strategy.add_protector(Protector(protector['name'],
+                                                 protector['address']))
 
     # Create observer.
     hss_observer = HomeSecuritySystemObserver()
-    hss_observer.set_notifier(whatsapp_notifier)
+    hss_observer.set_notifier(notifier)
 
     #  Create subjects to observe.
     wifi_subject = WiFiSubject()
@@ -58,23 +60,31 @@ def main():
     eye_subject.attach(hss_observer)
 
     # Run subjects.
-    wifi_subject.run(ip_address_strategy)
+    wifi_subject.run(network_strategy)
 
     # Set-up the camera to detect humans.
     camera = PiCameraStrategy()
     camera.set_detector(EfficientdetStrategy())
     eye_subject.run(camera, wifi_subject.get_protector_lock())
 
     # Notify that the system is running.
-    whatsapp_notifier.notify_all("Home Security System is started.")
+    notifier.notify_all("Home Security System is started.")
+    sleep(5)
+
+    if isinstance(notifier, TelegramStrategy):
+        with open(f"{eye_subject._image_path}/initial_frame.jpg", "rb") as initial_frame:
+            notifier.send_image_all(initial_frame)
+    elif isinstance(notifier, WhatsappStrategy):
+        fileio_link = upload_to_fileio(initial_frame)
+        notifier.notify_all(f"Here is the initial frame: {fileio_link}.")
 
     # Wait for the futures.
-    _, failures = wait([wifi_subject.thread, eye_subject.thread])
+    _, failures = wait([wifi_subject.thread, eye_subject.thread], return_when="FIRST_COMPLETED")
     for failure in failures:
-        whatsapp_notifier.notify_all("Home Security System has failed to run. "
-                                     "Please check the logs.")
-        whatsapp_notifier.notify_all("Error: " + str(failure.exception()))
-        whatsapp_notifier.notify_all("Result: " + str(failure.result()))
+        notifier.notify_all("Home Security System has failed to run. Please check the logs.")
+        notifier.notify_all("Failure: " + str(failure))
+        notifier.notify_all("Error: " + str(failure.exception()))
+        notifier.notify_all("Result: " + str(failure.result()))
         # Close the application to let systemd re-start it.
         sys.exit(1)
 

pyproject.toml

@@ -2,7 +2,7 @@
 name = "home-security-system"
 description = "Check human presence within a camera in your Raspberry Pi if your phone is not connected to the WiFi network."
 keywords = ["security-system", "tinyml", "camera", "embedded-linux"]
-version = "2024.3.0"
+version = "2024.4.21"
 readme = { file = "README.md", content-type = "text/markdown" }
 license = { file = "LICENSE" }