feat: add logging support

Author: electricalgorithm

core/observers/observer/hss_observer.py

@@ -1,13 +1,17 @@
 """
 The observer for Home Security System.
 """
+import logging
 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, EyeStrategyResult
+from core.utils.datatypes import EyeStates, WiFiStates
 from core.strategies.notifier.base_notifier_strategy import BaseNotifierStrategy
 
+# Add logging support.
+logger = logging.getLogger(__name__)
+
 
 class HomeSecuritySystemObserver(BaseObserver):
     """
@@ -22,22 +26,16 @@ def update(self, subject: BaseSubject) -> None:
         """This method is called when the observer is updated."""
         if isinstance(subject, WiFiSubject):
             self.wifi_state = subject.get_state()
+            logger.debug("WiFi state: " + str(self.wifi_state.name))
 
         if isinstance(subject, EyeSubject):
             self.eye_state = subject.get_state()
-
-        print("Observer- WiFi state: " + str(self.wifi_state.name))
-        print("Observer- Eye state: " + str(self.eye_state.name))
+            logger.debug("Eye state: " + str(self.eye_state.name))
 
         if self.wifi_state == WiFiStates.DISCONNECTED and self.eye_state == EyeStates.DETECTED:
+            logger.info("There is an intruder!")
             self._notifier.notify_all("There is an intruder!")
-            self._save_image()
 
     def set_notifier(self, notifier: BaseNotifierStrategy) -> None:
         """This method is called when the observer is updated."""
         self._notifier = notifier
-
-    # Internal methods
-    def _save_image(result: EyeStrategyResult) -> None:
-        """This method is called when the observer is updated."""
-        print("Image saved!")

core/observers/subject/eye_subject.py

@@ -2,21 +2,32 @@
 This class inherits from IBaseSubject.
 Concretes a subject for Eye/Camera features.
 """
+import logging
+from datetime import datetime
 from time import sleep
 from threading import Thread
+
+import cv2
+
 from core.utils.datatypes import EyeStates, EyeStrategyResult
 from core.observers.subject.base_subject import BaseSubject
 from core.strategies.eye.base_eye_strategy import BaseEyeStrategy
 
+# Add logging support.
+logger = logging.getLogger(__name__)
+
 
 class EyeSubject(BaseSubject):
     """
     This class inherits from IBaseSubject.
     Concretes a subject for Eye/Camera features.
     """
-    def __init__(self):
+    DEFAULT_SLEEP_INTERVAL = 10
+    SLEEP_INTERVAL_DETECTED = 5
+
+    def __init__(self, image_path: str):
         super().__init__()
-        self._result: EyeStrategyResult = None
+        self._image_path = image_path
 
     @staticmethod
     def get_default_state() -> EyeStates:
@@ -27,22 +38,31 @@ def run(self, eye_strategy: BaseEyeStrategy) -> None:
         """This method is called when the observer is updated."""
         thread = Thread(target=self._run_in_loop, args=(self, eye_strategy,))
         thread.start()
-
-    def save_result(self, result: EyeStrategyResult) -> None:
-        """This method is called when the observer is updated to DETECTED."""
-        self._result = result
+        logger.debug("EyeSubject is running...")
 
     @staticmethod
     def _run_in_loop(self, eye_strategy: BaseEyeStrategy) -> None:
         """This method is called when the observer is updated."""
+        sleep_interval = EyeSubject.DEFAULT_SLEEP_INTERVAL
+
         while True:
             result = eye_strategy.check_if_detected()
-            print("Eye result: " + str(result.result))
-            print("Eye state: " + str(self.get_state().name))
+            logger.debug("EyeStrategyResult: " + str(result.result))
+
             if result.result:
                 self.set_state(EyeStates.DETECTED)
-                self.save_result(result)
+                self._save_image(result)
+                sleep_interval = EyeSubject.SLEEP_INTERVAL_DETECTED
             else:
                 self.set_state(EyeStates.NOT_DETECTED)
-            sleep(5)
+                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."""
+        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: " + f"intruder_{time_now}.jpg")
 

core/observers/subject/wifi_subject.py

@@ -2,12 +2,17 @@
 This class inherits from IBaseSubject.
 Concretes a subject WiFi features.
 """
+import logging
 from threading import Thread
 from time import sleep
+
 from core.utils.datatypes import WiFiStates
 from core.observers.subject.base_subject import BaseSubject
 from core.strategies.wifi.base_wifi_strategy import BaseWiFiStrategy
 
+# Add logging support.
+logger = logging.getLogger(__name__)
+
 
 class WiFiSubject(BaseSubject):
     """
@@ -23,12 +28,15 @@ def run(self, wifi_strategy: BaseWiFiStrategy) -> None:
         """This method is called when the observer is updated."""
         thread = Thread(target=self._run_in_loop, args=(self, wifi_strategy,))
         thread.start()
+        logger.debug("WiFiSubject is running...")
 
     @staticmethod
     def _run_in_loop(self, wifi_strategy: BaseWiFiStrategy) -> None:
         """This method is called when the observer is updated."""
         while True:
             protectors = wifi_strategy.check_protectors()
+            logger.debug("Protectors: " + str(protectors.result) + " " + str(protectors.protector))
+
             if protectors.result:
                 self.set_state(WiFiStates.CONNECTED)
             else:

core/strategies/eye/camera_strategy.py

@@ -1,12 +1,18 @@
 """
 The Camera strategy for eye strategies.
 """
+import logging
 from datetime import datetime
+
 import cv2
 import numpy
+
 from core.strategies.eye.base_eye_strategy import BaseEyeStrategy
 from core.utils.datatypes import EyeStrategyResult
 
+# Add logging support.
+logger = logging.getLogger(__name__)
+
 
 class CameraStrategy(BaseEyeStrategy):
     """
@@ -50,6 +56,8 @@ def _detect_humans(self, frame: numpy.ndarray) -> tuple[list[tuple[int, int, int
             padding=(4, 4),
             scale=1.05
         )
+        logger.debug("Number of detections: " + str(num_detections)
+                    + " Regions: " + str(regions))
         return regions, num_detections
 
     def _outline_the_regions(self, frame: numpy.ndarray, regions: list[tuple[int, int, int, int]]) -> None:

core/strategies/notifier/whatsapp_strategy.py

@@ -2,7 +2,9 @@
 This module contains the Whatsapp notifier strategy.
 """
 import logging
+
 import requests
+
 from core.utils.datatypes import WhatsappReciever
 from core.strategies.notifier.base_notifier_strategy import BaseNotifierStrategy
 
@@ -24,6 +26,7 @@ def notify_all(self, message: str) -> None:
 
     def _send_message(self, reciever: WhatsappReciever, message: str) -> bool:
         """Send a WhatsApp message to the user."""
+        logger.debug("Sending WhatsApp message (%s) to %s", message, reciever.telephone_number)
         # Send the request.
         request_url = f"{self.API_URL}?phone={reciever.telephone_number}&text={message}&apikey={reciever.api_key}"
         response = requests.get(request_url, timeout=10)

core/strategies/wifi/ipaddress_strategy.py

@@ -1,10 +1,14 @@
 """
 The strategy which searches for IP addresses.
 """
+import logging
 from core.utils.program_launcher import run_program, PingCommands
 from core.strategies.wifi.base_wifi_strategy import BaseWiFiStrategy
 from core.utils.datatypes import WiFiStrategyResult
 
+# Add logging support.
+logger = logging.getLogger(__name__)
+
 
 class IpAddressStrategy(BaseWiFiStrategy):
     """
@@ -16,17 +20,20 @@ def __init__(self):
         # Check if the program is installed.
         try:
             run_program(PingCommands.GET_VERSION_INFO)
+            logger.debug("Program is already installed.")
         except RuntimeError:
+            logger.debug("Program is not installed. Installing...")
             run_program(PingCommands.INSTALL_PROGRAM)
 
     def check_protectors(self) -> WiFiStrategyResult:
         """This method checks if there are any protectors around."""
         for protector in self.protectors:
             # Send a ping to the protector.
+            logger.debug("Sending ping to " + protector.name)
             output = run_program(PingCommands.PING_TO + " " + protector.address)
             # Check if the ping was successful.
             if "Destination Host Unreachable" not in output:
-                print("Protector found: " + protector.name)
+                logger.debug("Protector found: " + protector.name)
                 return WiFiStrategyResult(protector, True)
-        print("Protector not found.")
+        logger.debug("No protectors found.")
         return WiFiStrategyResult(None, False)

core/strategies/wifi/macaddress_strategy.py

@@ -1,10 +1,15 @@
 """
 The strategy which searches for MAC addresses.
 """
+import logging
+
 from core.utils.program_launcher import run_program, ArpScanCommands
 from core.strategies.wifi.base_wifi_strategy import BaseWiFiStrategy
 from core.utils.datatypes import WiFiStrategyResult, ConnectedDeviceResult
 
+# Add logging support.
+logger = logging.getLogger(__name__)
+
 
 class MacAddressStrategy(BaseWiFiStrategy):
     """
@@ -14,15 +19,20 @@ def __init__(self):
         super().__init__()
         # Check if the program is installed.
         try:
+            logger.debug("Checking if the program is installed...")
             run_program(ArpScanCommands.GET_VERSION_INFO)
         except RuntimeError:
+            logger.debug("Program is not installed. Installing...")
             run_program(ArpScanCommands.INSTALL_PROGRAM)
 
     def check_protectors(self) -> WiFiStrategyResult:
         """This method checks if there are any protectors around."""
         for protector in self.protectors:
+            # Check if the protector is connected to the network.
             if protector.address in [device.address for device in self._get_all_connected()]:
+                logger.debug("Protector found: " + protector.name)
                 return WiFiStrategyResult(protector, True)
+        logger.debug("No protectors found.")
         return WiFiStrategyResult(None, False)
 
     # Internal methods
@@ -31,4 +41,5 @@ def _get_all_connected(self) -> list[ConnectedDeviceResult]:
         output = run_program(ArpScanCommands.GET_CONNECTED_MAC_ADDRESSES)
         mac_addrs = output.split('\n')
         mac_addrs = mac_addrs[:-1]
+        logger.debug("Connected devices: " + str(mac_addrs))
         return [ConnectedDeviceResult(mac_addr.upper()) for mac_addr in mac_addrs]

main.py

@@ -1,6 +1,7 @@
 """
 Main application layer for Home Security System.
 """
+import logging
 from core.observers.subject.eye_subject import EyeSubject
 from core.observers.subject.wifi_subject import WiFiSubject
 from core.observers.observer.hss_observer import HomeSecuritySystemObserver
@@ -10,6 +11,16 @@
 from core.utils.datatypes import WhatsappReciever, Protector
 
 
+# Add logging support.
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='[%(asctime)s] -- [%(levelname)s] -- %(name)s (%(funcName)s): %(message)s',
+    datefmt='%Y-%m-%d %H:%M:%S',
+    filename='hss.log',
+    filemode='a',
+)
+
+
 def main():
     """
     This method is the entry point of the application.
@@ -29,7 +40,7 @@ def main():
     # Create subjects to observe.
     wifi_subject = WiFiSubject()
     wifi_subject.attach(hss_observer)
-    eye_subject = EyeSubject()
+    eye_subject = EyeSubject("images/")
     eye_subject.attach(hss_observer)
 
     # Run subjects.