feat: add complete pipeline and Streamlit code This commit introduces a complete pipeline for both single and real-time inferences using cameras. It includes the implementation of Streamlit code to facilitate the process.

.gitignore

@@ -0,0 +1,163 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+.idea/
+resources
+images
+

README.md

@@ -1,13 +1,126 @@
----
-title: Weapon Detection
-emoji: 😻
-colorFrom: purple
-colorTo: pink
-sdk: streamlit
-sdk_version: 1.32.2
-app_file: app.py
-pinned: false
-license: apache-2.0
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# weapon-detection-alert-location-share
+
+This project focuses on enhancing the security of different places, like public areas and banks, by implementing a
+robust gun detection and location-sharing system. The system is designed to capture images from an IP camera
+strategically placed within the bank premises. When a captured image contains an identifiable gun, the system takes
+appropriate actions, such as sending alerts or sharing the location of the detected firearm.
+
+1. **Image Capture:**
+
+   Utilizes an IP camera to continuously capture images in real-time within the camera range.Images are processed and
+   analyzed for potential gun presence.
+
+2. **Gun Detection:**
+
+   Employs advanced computer vision algorithms for accurate gun detection in captured images.
+   The system recognizes various types of firearms and distinguishes them from other objects.
+3. Alert Systems:
+
+   When a gun is detected, the system triggers immediate alerts.
+   Alerts can be configured to notify security personnel, law enforcement, and relevant authorities.
+4. Location Sharing:
+
+   If a gun is detected, the system captures the location of the incident.
+   The location information is shared in real-time with designated security personnel and law enforcement agencies.
+
+## File Structure
+
+File structure of the project
+
+```commandline
+
+├── config
+│   ├── config.ini
+│   └── config.yaml
+├── examples
+│   ├── start_cap.py
+│   └── stop_cap.py
+├── images
+│   ├── cam_images
+│   └── detected_image
+│       ├── img_1.png
+│       └── img.png
+├── logs
+│   └── gun_det.log
+├── resources
+│   └── models
+│       └── yolov7.pt
+├── services
+│   ├── entity
+│   │   └── __init__.py
+│   ├── gun_det_service
+│   │   ├── detection_example_cam.py
+│   │   └── __init__.py
+│   ├── image_capture
+│   │   ├── capture_main.py
+│   │   ├── __init__.py
+│   │   ├── main_start_stop.py
+│   │   └── yolov7_detection_example.py
+│   ├── image_load
+│   │   ├── __init__.py
+│   │   ├── main_load.py
+│   │   ├── main_start_stop.py
+│   │   ├── start_load.py
+│   │   └── stop_load.py
+│   ├── __init__.py
+│   └── location_share
+│       └── __init__.py
+├── utils
+│   ├── __init__.py
+│   ├── __pycache__
+│   │   ├── __init__.cpython-310.pyc
+│   │   └── settings.cpython-310.pyc
+│   └── settings.py
+├── Dockerfiles
+├── main.py
+├── README.md
+├── requirements.txt
+└── visualization
+
+
+```
+
+## How to Run
+
+1. **Clone the repository:**
+    ```bash
+    git clone https://github.com/ishworrsubedii/gun-detection-alert.git
+    cd cd gun-detection
+    ```
+
+2. **Create and activate the Conda environment:**
+    ```bash
+    conda create -n weapon-detection python=3.10 -y
+    conda activate gun-detection
+    ```
+
+3. **Install required packages:**
+    ```bash
+    pip install -r requirements.txt
+    ```
+
+4. **Perform Inference/Prediction:**
+    ```bash
+    python3 main.py
+    streamlit run streamlit_app.py
+    ```
+
+    - We have to run both programs for inference. py for fastapi post request and streamlit for UI for the prediction.
+
+### Docker
+
+```commandline
+```
+
+# 🔥Features
+
+# ⚠️ Limitations
+
+# Future Work
+
+# Demo
+
+# Recommendations
+
+Your recommendations are highly valuable, and I highly value your insights and suggestions to enhance this project! Feel
+free to propose new features, report bugs, or suggest improvements.

app.py

@@ -0,0 +1,56 @@
+"""
+Created By: ishwor subedi
+Date: 2024-04-04
+"""
+import streamlit as st
+from PIL import Image
+
+from demos.single_image_inference import single_image_inference
+
+# Constants
+MAX_FILE_SIZE = 5 * 1024 * 1024  # 5MB
+
+# Configure Streamlit
+st.set_page_config(layout="wide", page_title="Weapon Detection")
+
+# Write to Streamlit
+st.write("## Weapon Detection")
+st.write(
+    "This app uses a custom trained yolov8 model to detect weapons in images. Upload an image to see the detection results."
+)
+st.sidebar.write("## Browse images:")
+
+
+def process_image(upload):
+    """
+    Process the uploaded image and display the original and processed images side by side.
+    """
+    try:
+        image = Image.open(upload)
+        col1, col2 = st.columns(2)
+        col1.write("Original Uploaded Image")
+        col1.image(image)
+
+        processed_image = single_image_inference(image)
+        col2.write("Predicted Image")
+        col2.image(processed_image)
+        st.sidebar.markdown("\n")
+    except Exception as e:
+        st.error(f"Error processing image: {e}")
+
+
+def handle_upload():
+    """
+    Handle the file upload process.
+    """
+    uploaded_file = st.sidebar.file_uploader("Upload an image", type=["png", "jpg", "jpeg"])
+
+    if uploaded_file is not None:
+        if uploaded_file.size > MAX_FILE_SIZE:
+            st.error("The uploaded file is too large. Please upload an image smaller than 5MB.")
+        else:
+            process_image(upload=uploaded_file)
+
+
+# Call the upload handler
+handle_upload()

config/config.ini

@@ -0,0 +1,7 @@
+[default]
+server.host = localhost
+server.port = 8000
+
+[example.model_info]
+yolo.model_path = resources/models/v1/best.pt
+path.cam_image_dir = images/cam_images

demos/__init__.py

@@ -0,0 +1,4 @@
+"""
+Created By: ishwor subedi
+Date: 2023-12-28
+"""

demos/alert_service_example.py

@@ -0,0 +1,9 @@
+"""
+Created By: ishwor subedi
+Date: 2023-12-30
+"""
+from services.alert_service.alert_service import AlertService
+
+if __name__ == '__main__':
+    alert_service = AlertService("resources/alert/alert.mp3")
+    alert_service.play_alert()

demos/cam_service_example/start_frame_capturing.py

@@ -0,0 +1,86 @@
+import os
+import time
+from services.image_capture_service.capture_main import FrameSaver
+
+
+class ImageCaptureService:
+    def __init__(self, flag_path, source, image_path_to_save, image_hash_threshold):
+        self.flag_path = flag_path
+        self.source = source
+        self.image_path_to_save = image_path_to_save
+        self.image_hash_threshold = image_hash_threshold
+        self.running = False
+
+    def create_stop_flag(self):
+        try:
+            with open(self.flag_path, 'w') as flag_file:
+                flag_file.write('False')  # Set initial content as 'False'
+            print("Flag file created and set to 'False'.")
+        except Exception as e:
+            print(f"Error creating flag file: {e}")
+
+    def check_stop_flag(self):
+        try:
+            if os.path.exists(self.flag_path):
+                with open(self.flag_path, 'r') as flag_file:
+                    content = flag_file.read().strip()
+                    return content.lower() == "true"
+            else:
+                return False
+        except Exception as e:
+            print(f"Error checking stop flag: {e}")
+            return False
+
+    def update_stop_flag(self, value):
+        try:
+            with open(self.flag_path, 'w') as flag_file:
+                flag_file.write(str(value))  # Write the provided value to the flag file
+        except Exception as e:
+            print(f"Error updating stop flag: {e}")
+
+    def start_service(self):
+        try:
+            print("Image Capturing service starting.")
+            self.running = True
+            start_capture = FrameSaver(self.source, self.image_path_to_save, self.image_hash_threshold)
+            start_capture.start_stream()
+
+            while self.running:
+                time.sleep(1)
+                stop_flag = self.check_stop_flag()
+
+                if stop_flag:
+                    print("Stop flag detected. Stopping all frame capturing services.")
+                    self.running = False
+                    stop_successful = start_capture.stop_stream()
+                    if stop_successful:
+                        print("Frame capturing services successfully stopped.")
+                    else:
+                        print("Issue encountered while stopping frame capturing services.")
+
+        except Exception as e:
+            print(f"Error in setting up FrameSaver: {e}")
+            print(f"Error in setting up FrameSaver: {e}")
+        finally:
+            if 'start_capture' in locals():
+                stop_successful = start_capture.stop_stream()
+                if stop_successful:
+                    print("Frame capturing services successfully stopped in finally block.")
+                else:
+                    print("Issue encountered while stopping frame capturing services in finally block.")
+
+    def stop_service(self):
+        self.update_stop_flag("True")
+
+
+if __name__ == "__main__":
+    FLAG_PATH = "resources/image_capturing"
+    # SOURCE = 'rtsp://ishwor:subedi@192.168.1.106:5555/h264_opus.sdp'
+    SOURCE = 0
+    IMAGE_PATH_TO_SAVE = "images/cam_images"
+    IMAGE_HASH_THRESHOLD = 5
+
+    image_capture_service = ImageCaptureService(FLAG_PATH, SOURCE, IMAGE_PATH_TO_SAVE, IMAGE_HASH_THRESHOLD)
+    image_capture_service.create_stop_flag()
+    image_capture_service.start_service()
+    time.sleep(1)

demos/cam_service_example/stop_frame_capturing.py

@@ -0,0 +1,21 @@
+import os
+
+
+class StopImageCaptureServiceExample:
+    def __init__(self, stop_flag_path):
+        self.stop_flag_path = stop_flag_path
+
+    def stop_example_ipcam_webcam(self):
+        try:
+            with open(self.stop_flag_path, 'w') as flag_file:
+                flag_file.write("True")
+            print("Stop flag set.")
+            os.remove(self.stop_flag_path)
+        except Exception as e:
+            print(f"Error setting stop flag: {e}")
+
+
+if __name__ == "__main__":
+    STOP_FLAG_PATH = "resources/flag"
+    stop_load_image_example = StopImageCaptureServiceExample(STOP_FLAG_PATH)
+    stop_load_image_example.stop_example_ipcam_webcam()

demos/image_load_service_example/start_image_load_example.py

@@ -0,0 +1,84 @@
+import os
+import sys
+import time
+import threading
+from services.image_capture_service.image_load_main import ImageLoad
+
+
+class StartImageLoadExample:
+    def __init__(self, flag_path, image_dir, model_path):
+        self.flag_path = flag_path
+        self.image_dir = image_dir
+        self.model_path = model_path
+
+        self.running = False
+
+    def create_stop_flag(self):
+        try:
+            with open(self.flag_path, 'w') as flag_file:
+                flag_file.write('False')
+        except Exception as e:
+            print(f"Error creating stop flag: {e}")
+
+    def check_stop_flag(self):
+        try:
+            if os.path.exists(self.flag_path):
+                with open(self.flag_path, 'r') as flag_file:
+                    content = flag_file.read().strip()
+                    return content.lower() == "true"
+            else:
+                return False
+        except Exception as e:
+            print(f"Error checking stop flag: {e}")
+            return False
+
+    def update_stop_flag(self, value):
+        try:
+            with open(self.flag_path, 'w') as flag_file:
+                flag_file.write(str(value))
+        except Exception as e:
+            print(f"Error updating stop flag: {e}")
+
+    def start_service(self):
+        try:
+            self.running = True
+            start_load = ImageLoad(self.image_dir, self.model_path)
+            start_load.start_load_image()
+            print("Image loading service successfully started")
+
+            while self.running:
+                time.sleep(1)
+                stop_flag = self.check_stop_flag()
+
+                if stop_flag:
+                    self.running = False
+                    stop_successful = start_load.stop_load_image()
+                    if stop_successful:
+                        print("Image loading services_trinetra successfully stopped.")
+                    else:
+                        print("Issue encountered while stopping image loading services_trinetra.")
+
+        except Exception as e:
+            print(f"Error starting image loading services_trinetra: {e}")
+        finally:
+            if 'start_load' in locals():
+                stop_successful = start_load.stop_load_image()
+                if stop_successful:
+                    print("Image loading services_trinetra successfully stopped in finally block.")
+                else:
+                    print(
+                        "Issue encountered while stopping image loading services_trinetra in finally block.")
+
+    def stop_service(self):
+        self.update_stop_flag("True")
+
+
+if __name__ == "__main__":
+    flag_path = "resources/flag_load_image"
+    image_dir = 'images/cam_images'
+    model_path = 'resources/models/v1/best.pt'
+
+    image_load_service = StartImageLoadExample(flag_path, image_dir, model_path)
+    image_load_service.create_stop_flag()
+    image_load_service.start_service()
+    time.sleep(1)

demos/image_load_service_example/stop_image_load_example.py

@@ -0,0 +1,28 @@
+"""
+Created By: ishwor subedi
+Date: 2023-12-28
+"""
+
+import os
+import os
+from services import ipcam_logger
+
+
+class StopImageLoadServiceExample:
+    def __init__(self, stop_flag_path):
+        self.stop_flag_path = stop_flag_path
+        self.logger = ipcam_logger()
+
+    def stop_service(self):
+        try:
+            with open(self.stop_flag_path, 'w') as flag_file:
+                flag_file.write("True")
+            self.logger.info("Stop flag set.")
+        except Exception as e:
+            self.logger.error(f"Error setting stop flag: {e}")
+
+
+if __name__ == "__main__":
+    STOP_FLAG_PATH = "resources/flag_load_image"
+    webcam_controller = StopImageLoadServiceExample(STOP_FLAG_PATH)
+    webcam_controller.stop_service()

demos/single_image_inference.py

@@ -0,0 +1,38 @@
+"""
+Created By: ishwor subedi
+Date: 2024-04-04
+"""
+from services.weapon_det_service.weapon_detection_service import DetectionService
+import cv2 as cv
+
+
+def single_image_inference(image_path):
+    detection_service = DetectionService(
+        model_path='resources/models/v1/best.pt',
+
+    )
+    results = detection_service.image_det_save(
+        image_path=image_path,
+        thresh=0.2
+    )
+    for result in results:
+        original_image = result.orig_img
+        bbox = result.boxes.xyxy.int().tolist()
+        for i, bbox in enumerate(bbox):
+            x1, y1, x2, y2 = bbox
+
+            cv.putText(original_image, f'{result.names[result.boxes.cls[i].int().tolist()]}', (x1, y1 - 10),
+                       cv.FONT_HERSHEY_SIMPLEX,
+                       0.9,
+                       (0, 0, 255), 2)
+            cv.rectangle(original_image, (x1, y1), (x2, y2), (0, 0, 255), 2)
+
+        return original_image
+
+
+if __name__ == '__main__':
+    image_path = '/home/ishwor/Desktop/gun-detection/images/cam_images/th-3711382641.jpg'
+    image = single_image_inference(image_path)
+    cv.imshow('Weapon Detection', image)
+    cv.waitKey(0)
+    cv.destroyAllWindows()

logs/ipcam.log

@@ -0,0 +1 @@
+[2024-04-04 00:05:06,251:INFO:stop_image_load_example:Stop flag set.]

main.py

@@ -0,0 +1,10 @@
+"""
+Created By: ishwor subedi
+Date: 2024-01-02
+"""
+import uvicorn
+
+from services.api.fast_api import app
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="localhost", port=8000)

resources/flag_load_image

@@ -0,0 +1 @@
+True

resources/image_capturing

@@ -0,0 +1 @@
+False

resources/models/best.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:21d61ad8068caca3062d33fd8d05da445ef9ae81a2bb817249e085b0f1307cf0
+size 6263257

resources/models/v1/best.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:4937605a869713f16e72981374933daf2ce884ff5603a91c42403ece282c2988
+size 12349682

resources/models/weapon_detector.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:74ee5662c4a57267ac270068aa20a805b221f1ffe6d88c08c9131b9eb873b296
+size 74802191

resources/models/yolov7.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d1ac3c74eb96a3eec77949c0f37a06bc272756606ff011d9353f7abff4e0c71d
+size 75587165

resources/models/yolov7_2.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d1ac3c74eb96a3eec77949c0f37a06bc272756606ff011d9353f7abff4e0c71d
+size 75587165

services/__init__.py

@@ -0,0 +1,64 @@
+import os
+import sys
+import logging
+
+
+def configure_logger(logger_name, log_filename, log_dir="logs"):
+    """
+    Configures a logger with specified parameters.
+
+    :param logger_name: Name of the logger.
+    :param log_filename: Name of the log file.
+    :param log_dir: Directory where logs will be stored. Defaults to "logs".
+    :return: Configured logger.
+    """
+    logging_str = "[%(asctime)s:%(levelname)s:%(module)s:%(message)s]"
+
+    log_path = os.path.join(log_dir, log_filename)
+    os.makedirs(log_dir, exist_ok=True)
+
+    logger = logging.getLogger(logger_name)
+    logger.setLevel(logging.INFO)
+
+    formatter = logging.Formatter(logging_str)
+
+    file_handler = logging.FileHandler(log_path)
+    file_handler.setFormatter(formatter)
+
+    stream_handler = logging.StreamHandler(sys.stdout)
+    stream_handler.setFormatter(formatter)
+
+    logger.addHandler(file_handler)
+    logger.addHandler(stream_handler)
+
+    return logger
+
+
+def main_sys_logger(log_dir="logs"):
+    """
+    Configures and returns the main system logger.
+
+    :param log_dir: Directory where logs will be stored. Defaults to "logs".
+    :return: Main system logger.
+    """
+    return configure_logger("main_sys_logger", "gun_det.log", log_dir)
+
+
+def ipcam_logger(log_dir="logs"):
+    """
+    Configures and returns the IP camera logger.
+
+    :param log_dir: Directory where logs will be stored. Defaults to "logs".
+    :return: IP camera logger.
+    """
+    return configure_logger("ipcam_logger", "ipcam.log", log_dir)
+
+
+def detection_logger(log_dir="logs"):
+    """
+    Configures and returns the detection logger.
+
+    :param log_dir: Directory where logs will be stored. Defaults to "logs".
+    :return: Detection logger.
+    """
+    return configure_logger("detection_logger", "detection.log", log_dir)

services/alert_service/__init__.py

@@ -0,0 +1,4 @@
+"""
+Created By: ishwor subedi
+Date: 2023-12-29
+"""

services/alert_service/alert_service.py

@@ -0,0 +1,30 @@
+"""
+Created By: ishwor subedi
+Date: 2023-12-30
+"""
+
+import pygame
+
+
+class AlertService:
+    def __init__(self, alert_sound_path):
+        self.alert_sound_path = alert_sound_path
+        pygame.init()
+
+    def play_alert(self):
+        """
+        This function help us to play the alert sound
+        :return:
+        """
+        try:
+            pygame.mixer.init()
+            alert_sound = pygame.mixer.Sound(self.alert_sound_path)
+            alert_sound.play()
+            pygame.time.wait(int(alert_sound.get_length() * 1000))  # Wait for the sound to finish playing
+        except pygame.error as e:
+            print(f"Error playing alert sound: {e}")
+
+
+if __name__ == '__main__':
+    alert_service = AlertService("resources/alert/alert.mp3")
+    alert_service.play_alert()

services/api/__init__.py

@@ -0,0 +1,4 @@
+"""
+Created By: ishwor subedi
+Date: 2023-12-28
+"""

services/api/fast_api.py

@@ -0,0 +1,109 @@
+from fastapi import FastAPI
+from fastapi.responses import JSONResponse
+from fastapi import HTTPException
+import uvicorn
+import base64
+import io
+from PIL import Image
+from demos.cam_service_example.start_frame_capturing import ImageCaptureService
+from demos.cam_service_example.stop_frame_capturing import StopImageCaptureServiceExample
+from demos.image_load_service_example.start_image_load_example import StartImageLoadExample
+from demos.image_load_service_example.stop_image_load_example import StopImageLoadServiceExample
+from demos.single_image_inference import single_image_inference
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+# Helper functions
+def start_ipcam():
+    flag_path = "resources/flag"
+
+    source = 'rtsp://ishwor:subedi@192.168.1.106:5555/h264_opus.sdp'
+
+    image_path_to_save = "images/cam_images"
+    image_hash_threshold = 5
+    image_capture_start_example = ImageCaptureService(flag_path, source, image_path_to_save,
+                                                      image_hash_threshold)
+    image_capture_start_example.start_service()
+
+
+def stop_ipcam():
+    flag_path = "resources/flag"
+
+    image_capture_stop_example = StopImageCaptureServiceExample(flag_path)
+    image_capture_stop_example.stop_load_image_example()
+
+
+def start_detection():
+    flag_path = "resources/flag_load_image"
+
+    image_dir_path = "images/cam_images"
+
+    start_load_image_example = StartImageLoadExample(flag_path, image_dir_path, model_path="resources/model/v1/best.pt")
+    start_load_image_example.start_service()
+
+
+def stop_detection():
+    flag_path = "resources/flag_load_image"
+
+    stop_load_image_example = StopImageLoadServiceExample(flag_path)
+    stop_load_image_example.stop_service()
+
+
+# FastAPI Endpoints
+@app.post("/start_ipcam_server")
+def start_ipcam_server_api():
+    try:
+        start_ipcam()
+        return JSONResponse(content={"message": "IP Cam Server started!"}, status_code=200)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/stop_ipcam_server")
+def stop_ipcam_server_api():
+    try:
+        stop_ipcam()
+        return JSONResponse(content={"message": "IP Cam Server stopped!"}, status_code=200)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/start_detection_service")
+def start_detection_service_api():
+    try:
+        start_detection()
+        return JSONResponse(content={"message": "Image loading Server started!"}, status_code=200)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/stop_detection_service")
+def stop_detection_service_api():
+    try:
+        stop_detection()
+        return JSONResponse(content={"message": "Image loading Server stopped!"}, status_code=200)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+class ImageInput(BaseModel):
+    base64_image: str
+
+
+@app.post("/single_image_detection")
+def single_image_detection_api(image_input: ImageInput):
+    try:
+        image_data = base64.b64decode(image_input.base64_image)
+
+        image = Image.open(io.BytesIO(image_data))
+        image = single_image_inference(image)
+        encoded_img = base64.b64encode(image)
+        return encoded_img
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="localhost", port=8001)

services/entity/entity_config.py

@@ -0,0 +1,7 @@
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class StreamLitConfig:
+    pass

services/image_capture_service/capture_main.py

@@ -0,0 +1,81 @@
+"""
+Author: ishwor subedi
+Date: 2023-12-28
+Project Name:gun-detection
+
+"""
+import os
+from PIL import Image
+from datetime import datetime
+import threading
+import imagehash
+import cv2 as cv
+from utils.settings import get_frame_save_dir
+
+
+class FrameSaver:
+    def __init__(self, source, image_path_to_save, image_hash_threshold):
+        self.source = source
+        self.image_path_to_save = image_path_to_save
+        if self.image_path_to_save is None or not os.path.exists(self.image_path_to_save):
+            self.image_path_to_save = get_frame_save_dir()
+        self.thread_running = False
+        self.frame_save_thread = None
+        self.cap = cv.VideoCapture(self.source)
+        self.image_hash_threshold = image_hash_threshold
+
+    def start_stream(self):
+        self.frame_save_thread = threading.Thread(target=self.video_webcam_frame_extraction)
+        self.frame_save_thread.start()
+        self.thread_running = True
+
+    def stop_stream(self):
+        if self.thread_running:
+            self.thread_running = False
+            self.frame_save_thread.join()
+        self.cap.release()
+
+    def hashing_diff(self, prev_frame, current_frame):
+        if prev_frame is None:
+            return None
+        else:
+            hash1 = imagehash.average_hash(Image.fromarray(prev_frame))
+            hash2 = imagehash.average_hash(Image.fromarray(current_frame))
+            return hash2 - hash1
+
+    def video_webcam_frame_extraction(self):
+        """
+        
+        :return:
+        """
+
+        try:
+
+            previous_frame = None
+            while self.cap.isOpened():
+                success, frame = self.cap.read()
+                if success:
+                    hash_diff = self.hashing_diff(previous_frame, frame)
+                    if hash_diff is None or hash_diff > self.image_hash_threshold:
+                        current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+                        name = os.path.join(self.image_path_to_save, f"{current_time}.jpg")
+                        success = cv.imwrite(name, frame)
+                        if success:
+                            print(f"Saved {name}...")
+                        else:
+                            print(f"Failed to save image: {name}")
+                    previous_frame = frame
+        except Exception as e:
+            print(f"Exception occurred: {e}")
+            try:
+                self.stop_stream()
+            except Exception as e:
+                print(f"Failed to stop stream: {e}")
+
+
+if __name__ == '__main__':
+    source = 'rtsp://ishwor:subedi@192.168.1.106:5555/h264_opus.sdp'
+    image_path_to_save = "images/cam_images"
+    image_hash_threshold = 5
+    image_capture_service = FrameSaver(source, image_path_to_save, image_hash_threshold)
+    image_capture_service.video_webcam_frame_extraction()