feat: Apply minor changes (#8)

* feat: apply minor changes on project

* feat: reorder imports

* chore: add settings.json to ignore

Author: electricalgorithm

.gitignore

@@ -8,4 +8,5 @@ __pycache__/
 .vscode-test/
 .vagra
 DS_Store
-*.log
+*.log
+.vscode/*

core/observers/observer/hss_observer.py

@@ -1,14 +1,14 @@
 """
 The observer for Home Security System.
 """
-from core.utils.logger import get_logger
 from core.observers.observer.base_observer import BaseObserver
 from core.observers.subject.base_subject import BaseSubject
-from core.observers.subject.wifi_subject import WiFiSubject
 from core.observers.subject.eye_subject import EyeSubject
-from core.utils.datatypes import EyeStates, WiFiStates
-from core.utils.fileio_adaptor import upload_to_fileio, read_latest_file
+from core.observers.subject.wifi_subject import WiFiSubject
 from core.strategies.notifier.base_notifier_strategy import BaseNotifierStrategy
+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
 
 # Add logging support.
 logger = get_logger(__name__)

core/observers/subject/base_subject.py

@@ -3,8 +3,13 @@
 This observer is used to create subjects.
 """
 from abc import ABCMeta, abstractmethod
-from core.utils.datatypes import ObserverStates
+
 from core.observers.observer.base_observer import BaseObserver
+from core.utils.datatypes import ObserverStates
+from core.utils.logger import get_logger
+
+# Create a logger instance.
+logger = get_logger(__name__)
 
 
 class BaseSubject(metaclass=ABCMeta):
@@ -26,6 +31,8 @@ def detach(self, observer: BaseObserver) -> None:
 
     def notify(self) -> None:
         """This method is called when the observer is updated."""
+        logger.debug("[Subject: %s] Notifying observers...",
+                     self.__class__.__name__)
         for observer in self._observers:
             observer.update(self)
 
@@ -35,6 +42,10 @@ def get_state(self) -> ObserverStates:
 
     def set_state(self, state: ObserverStates) -> None:
         """This method is called when the observer is updated."""
+        logger.debug("[Subject: %s] %s->%s",
+                     self.__class__.__name__,
+                     self._current_state.name,
+                     state.name)
         self._current_state = state
         self.notify()
 

core/observers/subject/eye_subject.py

@@ -3,16 +3,17 @@
 Concretes a subject for Eye/Camera features.
 """
 import os
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
+from threading import Lock
 from time import sleep
-from threading import Thread, Lock
 
 import cv2
 
-from core.utils.logger import get_logger
-from core.utils.datatypes import EyeStates, EyeStrategyResult
 from core.observers.subject.base_subject import BaseSubject
 from core.strategies.eye.base_eye_strategy import BaseEyeStrategy
+from core.utils.datatypes import EyeStates, EyeStrategyResult
+from core.utils.logger import get_logger
 
 # Add logging support.
 logger = get_logger(__name__)
@@ -43,56 +44,72 @@ 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: Lock | None = None
             ) -> None:
         """This method is called when the observer is updated."""
-        thread = Thread(target=self._run_in_loop, args=(self, eye_strategy, wifi_lock))
-        thread.start()
-        logger.debug("EyeSubject is running...")
+        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)
 
     def _run_in_loop(self,
                      eye_strategy: BaseEyeStrategy,
                      wifi_lock: Lock | None = None
                      ) -> None:
         """This method is called when the observer is updated."""
+        logger.debug("[EyeSubject] Thread is started.")
         sleep_interval = EyeSubject.DEFAULT_SLEEP_INTERVAL
 
         # Create a dummy lock instance if not given.
         if wifi_lock is None:
+            logger.debug("[EyeSubject] A lock instance is generated.")
             wifi_lock = Lock()
 
         while True:
             # If WiFi subject would give rights to use camera,
             # Check if any intruders detected.
+            logger.debug("[EyeSubject] WiFi Lock Status: %s", wifi_lock.locked())
             if not wifi_lock.locked():
                 result = eye_strategy.check_if_detected()
-                logger.debug("EyeStrategyResult: %s", str(result.result))
+                logger.debug("[EyeSubject] EyeStrategyResult: %s", str(result.result))
 
                 if result.result:
-                    logger.debug("Changing state to DETECTED...")
+                    logger.debug("[EyeSubject] Changing state to DETECTED...")
                     self._save_image(result)
                     self.set_state(EyeStates.DETECTED)
                     sleep_interval = EyeSubject.SLEEP_INTERVAL_DETECTED
                 else:
-                    logger.debug("Changing state to NOT_DETECTED...")
+                    logger.debug("[EyeSubject] Changing state to NOT_DETECTED...")
                     self.set_state(EyeStates.NOT_DETECTED)
                     sleep_interval = EyeSubject.DEFAULT_SLEEP_INTERVAL
 
             #  If the WiFi subject does not give rights,
             # aka: "There is protectors around the house."
             else:
-                logger.debug("Changing state to UNREACHABLE...")
+                logger.debug("[EyeSubject] Changing state to UNREACHABLE...")
                 self.set_state(EyeStates.UNREACHABLE)
                 sleep_interval = EyeSubject.DEFAULT_SLEEP_INTERVAL
 
             sleep(sleep_interval)
 
     def _save_image(self, result: EyeStrategyResult) -> None:
         """This method is called when the observer is updated."""
-        logger.debug("Saving image to the disk...")
+        logger.debug("[EyeSubject] Saving image to the disk...")
         time_now = datetime.now().strftime("%d-%m-%Y_%H-%M-%S")
         file_location = f"{self._image_path}/intruder_{time_now}.jpg"
         cv2.imwrite(file_location, result.image)
-        logger.debug("Image saved to the disk with name: intruder_%s.jpg", time_now)
+        logger.debug("[EyeSubject] Image saved to the disk with name: intruder_%s.jpg",
+                     time_now)

core/observers/subject/wifi_subject.py

@@ -2,13 +2,14 @@
 This class inherits from IBaseSubject.
 Concretes a subject WiFi features.
 """
-from threading import Thread, Lock
+from concurrent.futures import ThreadPoolExecutor
+from threading import Lock
 from time import sleep
 
-from core.utils.logger import get_logger
-from core.utils.datatypes import WiFiStates
 from core.observers.subject.base_subject import BaseSubject
 from core.strategies.wifi.base_wifi_strategy import BaseWiFiStrategy
+from core.utils.datatypes import WiFiStates
+from core.utils.logger import get_logger
 
 # Add logging support.
 logger = get_logger(__name__)
@@ -27,11 +28,22 @@ 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 = Thread(target=self._run_in_loop, args=(wifi_strategy,))
-        thread.start()
-        logger.debug("WiFiSubject is running...")
+        thread = ThreadPoolExecutor(
+            max_workers=1,
+            thread_name_prefix="wifisubject"
+        ).submit(self._run_in_loop, self, wifi_strategy)
+        thread.add_done_callback(self._cb_save)
 
     @classmethod
     def get_protector_lock(cls) -> Lock:
@@ -43,20 +55,27 @@ def get_protector_lock(cls) -> Lock:
             cls.SINGLETON_LOCK = Lock()
         return cls.SINGLETON_LOCK
 
+    @staticmethod
     def _run_in_loop(self, wifi_strategy: BaseWiFiStrategy) -> None:
         """This method is called when the observer is updated."""
+        logger.debug("[WiFiSubject] Thread is started.")
+
         protector_lock: Lock = self.get_protector_lock()
 
         while True:
             protectors = wifi_strategy.check_protectors()
-            logger.debug("Protectors: %s %s", str(protectors.result), str(protectors.protector))
+            logger.debug("[WiFiSubject] Protectors: %s %s",
+                         str(protectors.result),
+                         str(protectors.protector))
 
             if protectors.result:
                 self.set_state(WiFiStates.CONNECTED)
                 if not protector_lock.locked():
                     protector_lock.acquire()
+                    logger.debug("[WiFiSubject] Protector lock is acquired.")
             else:
                 self.set_state(WiFiStates.DISCONNECTED)
                 if protector_lock.locked():
                     protector_lock.release()
+                    logger.debug("[WiFiSubject] Protector lock is released.")
             sleep(self.CHECK_INTERVAL)

core/strategies/detectors/base_detector_strategy.py

@@ -3,7 +3,7 @@
 This strategy is used to define the interface for all detector strategies.
 """
 
-from abc import abstractmethod, ABCMeta
+from abc import ABCMeta, abstractmethod
 from dataclasses import dataclass
 
 import numpy

core/strategies/detectors/efficientdet_strategy.py

@@ -2,10 +2,11 @@
 An TinyML detection technique using Efficientdet model.
 """
 from typing import Any
+
 import cv2
 import numpy
-from tflite_runtime.interpreter import Interpreter
 from base_detector_strategy import BaseDetectorStrategy, DetectorResult
+from tflite_runtime.interpreter import Interpreter
 
 
 class EfficientdetStrategy(BaseDetectorStrategy):

core/strategies/eye/base_eye_strategy.py

@@ -2,10 +2,15 @@
 The base strategy for eye strategies.
 This strategy is used to define the interface for all eye strategies.
 """
-from abc import abstractmethod, ABCMeta
+from abc import ABCMeta, abstractmethod
+
 from numpy import ndarray
+
+from core.strategies.detectors.base_detector_strategy import (
+    BaseDetectorStrategy,
+    DetectorResult,
+)
 from core.utils.datatypes import EyeStrategyResult
-from core.strategies.detectors.base_detector_strategy import BaseDetectorStrategy, DetectorResult
 
 
 class BaseEyeStrategy(metaclass=ABCMeta):

core/strategies/eye/picamera_strategy.py

@@ -4,9 +4,9 @@
 import numpy
 from picamera2 import Picamera2
 
-from core.utils.logger import get_logger
 from core.strategies.detectors.base_detector_strategy import BaseDetectorStrategy
 from core.strategies.eye.base_eye_strategy import BaseEyeStrategy
+from core.utils.logger import get_logger
 
 # Add logging support.
 logger = get_logger(__name__)

core/strategies/eye/usbcamera_strategy.py

@@ -4,9 +4,9 @@
 import cv2
 import numpy
 
-from core.utils.logger import get_logger
 from core.strategies.detectors.base_detector_strategy import BaseDetectorStrategy
 from core.strategies.eye.base_eye_strategy import BaseEyeStrategy
+from core.utils.logger import get_logger
 
 # Add logging support.
 logger = get_logger(__name__)