feat: Re-start threads after automatically die (#10)

* fix: Solve minor issues and reorder imports

* feat: re-start threads after die

* feat: update py related files

* refactor: apply flake8 suggestions

Author: electricalgorithm

.pylintrc

@@ -3,10 +3,10 @@
 Concretes a subject for Eye/Camera features.
 """
 import os
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime
 from threading import Lock
-from time import sleep
+from time import sleep, time
 from typing import Optional
 
 import cv2
@@ -37,6 +37,11 @@ def __init__(self, image_path: str = DEFAULT_IMAGE_LOCATIONS):
             else os.path.expanduser(image_path)
         )
 
+        # To run the eye after thread dies.
+        self.thread: Optional[Future] = None
+        self._eye_strategy: Optional[BaseEyeStrategy] = None
+        self._wifi_lock: Optional[Lock] = None
+
         # Create the default image directory if not exists.
         os.makedirs(self._image_path, exist_ok=True)
 
@@ -45,26 +50,21 @@ def get_default_state() -> EyeStates:
         """This method is called when the observer is updated."""
         return EyeStates.UNREACHABLE
 
-    def _cb_save(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 = "eyesubject_thread_died.txt"
-        with open(file_location, "w", encoding="utf-8") as file:
-            file.write("The thread died.")
-        logger.warning("[EyeSubject] The thread died. A file is created at %s.",
-                       file_location)
-
     def run(self,
             eye_strategy: BaseEyeStrategy,
             wifi_lock: Optional[Lock] = None
             ) -> None:
         """This method is called when the observer is updated."""
-        thread = ThreadPoolExecutor(
+        # Update the latest configurations.
+        self._eye_strategy = eye_strategy
+        self._wifi_lock = wifi_lock
+
+        # Run the thread.
+        self.thread = ThreadPoolExecutor(
             max_workers=1,
             thread_name_prefix="eyesubject"
         ).submit(self._run_in_loop, self, eye_strategy, wifi_lock)
-        thread.add_done_callback(self._cb_save)
+        self.thread.add_done_callback(self._cb_done)
 
     @staticmethod
     def _run_in_loop(self,
@@ -115,3 +115,14 @@ def _save_image(self, result: EyeStrategyResult) -> None:
         cv2.imwrite(file_location, result.image)
         logger.debug("[EyeSubject] Image saved to the disk with name: intruder_%s.jpg",
                      time_now)
+
+    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/eye_subject.py

@@ -2,9 +2,9 @@
 This class inherits from IBaseSubject.
 Concretes a subject WiFi features.
 """
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import Future, ThreadPoolExecutor
 from threading import Lock
-from time import sleep
+from time import sleep, time
 from typing import Optional
 
 from core.observers.subject.base_subject import BaseSubject
@@ -24,27 +24,28 @@ class WiFiSubject(BaseSubject):
     SINGLETON_LOCK: Optional[Lock] = None
     CHECK_INTERVAL: int = 5
 
+    def __init__(self):
+        super().__init__()
+        # To run the WiFi after thread dies.
+        self.thread: Optional[Future] = None
+        self._wifi_strategy: Optional[BaseWiFiStrategy] = None
+
     @staticmethod
     def get_default_state() -> WiFiStates:
         """This method is called when the observer is updated."""
         return WiFiStates.UNREACHABLE
 
-    def _cb_save(self, future) -> None:
-        """This method is called when the observer is updated."""
-        logger.warning("[WiFiSubject] The thread died.")
-        file_location = "wifisubject_thread_died.txt"
-        with open(file_location, "w", encoding="utf-8") as file:
-            file.write("The thread died.")
-        logger.warning("[WiFiSubject] The thread died. A file is created at %s.",
-                       file_location)
-
     def run(self, wifi_strategy: BaseWiFiStrategy) -> None:
         """This method is called when the observer is updated."""
-        thread = ThreadPoolExecutor(
+        # Update the latest configurations.
+        self._wifi_strategy = wifi_strategy
+
+        # Start the thread.
+        self.thread = ThreadPoolExecutor(
             max_workers=1,
             thread_name_prefix="wifisubject"
         ).submit(self._run_in_loop, self, wifi_strategy)
-        thread.add_done_callback(self._cb_save)
+        self.thread.add_done_callback(self._cb_done)
 
     @classmethod
     def get_protector_lock(cls) -> Lock:
@@ -80,3 +81,13 @@ def _run_in_loop(self, wifi_strategy: BaseWiFiStrategy) -> None:
                     protector_lock.release()
                     logger.debug("[WiFiSubject] Protector lock is released.")
             sleep(self.CHECK_INTERVAL)
+
+    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/observers/subject/wifi_subject.py

@@ -18,7 +18,6 @@ def get_logger(name: str) -> logging.Logger:
     log_dir: str = os.path.dirname(LOG_PATH)
     if not os.path.exists(log_dir):
         os.makedirs(log_dir)
-
     # Configure the log messages.
     formatter: logging.Formatter = logging.Formatter(
         fmt='[%(asctime)s] -- [%(levelname)s] -- %(name)s (%(funcName)s): %(message)s',

core/utils/logger.py

@@ -3,6 +3,7 @@
 """
 import json
 import sys
+from concurrent.futures import wait
 from typing import Any
 
 from core.observers.observer.hss_observer import HomeSecuritySystemObserver
@@ -64,6 +65,19 @@ def main():
     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.")
+
+    # Wait for the futures.
+    _, failures = wait([wifi_subject.thread, eye_subject.thread])
+    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()))
+        # Close the application to let systemd re-start it.
+        sys.exit(1)
+
 
 if __name__ == "__main__":
     try: