diff --git a/DataBase/dataBase.py b/DataBase/dataBase.py index 6396b21..f70e778 100644 --- a/DataBase/dataBase.py +++ b/DataBase/dataBase.py @@ -1,17 +1,9 @@ import psycopg2 from configparser import ConfigParser +from contextlib import contextmanager -def add_user(username, password): - cursor, conn = connect() - - query = "insert into users (username, password) values (%s, %s)" - - cursor.execute(query, (username, password)) - conn.commit() - - conn.close() - -def connect(): +@contextmanager +def get_db_connection(): try: conn = psycopg2.connect(host="127.0.0.1", user="postgres", @@ -19,149 +11,89 @@ def connect(): password="root", port="5432") cursor = conn.cursor() - # print connexion settings - print("Connexion settings : ", conn.get_dsn_parameters()) - except (Exception, psycopg2.Error) as error: - print("Error while trying to connect to PostgreSQL ", error) - - return cursor, conn + yield cursor, conn + except Exception as error: + print(f"Error while connecting to PostgreSQL: {error}") + finally: + conn.close() +def add_user(username, password): + with get_db_connection() as (cursor, conn): + query = "INSERT INTO users (username, password) VALUES (%s, %s)" + cursor.execute(query, (username, password)) + conn.commit() + def add_camera(address, nom): - cursor, conn = connect() - - query = "select * from cameras" - cursor.execute(query) - - result = cursor.fetchall() - number_cam = 0 - for rows in result: - number_cam += 1 - - print("Number cameras = ", number_cam) - - if number_cam < 4: - query = "insert into cameras (address, nom) values (%s, %s)" - cursor.execute(query, (address, nom)) - conn.commit() - else: - print("Maximum number of cameras added already") - - conn.close() + with get_db_connection() as (cursor, conn): + try: + query = "SELECT COUNT(*) FROM cameras" + cursor.execute(query) + number_cam = cursor.fetchone()[0] + + if number_cam < 4: + query = "INSERT INTO cameras (address, nom) VALUES (%s, %s)" + cursor.execute(query, (address, nom)) + conn.commit() + else: + print("Maximun number of cameras added already") + except Exception as e: + print(f"Error adding camera: {e}") def remove_camera(id): pass - -def storeFireAlertData(alertTime, videoLink, AlertClass): - - cursor, conn = connect() - # Connection achieved - # Storing alert data - query = "insert into fire_alerts (alert_time, video_link, class) values ( %s, %s, %s)" - - cursor.execute(query, (alertTime, videoLink, AlertClass)) - conn.commit() - - # Close connection - conn.close() - - -def storeFallAlertData(alertTime, videoLink, AlertClass): - # Try connection - cursor, conn = connect() - # Connection achieved - # Storing alert data - try: - query = "insert into fall_alerts (alert_time, video_link, class) values (%s, %s, %s)" - - cursor.execute(query, (alertTime, videoLink, AlertClass)) - conn.commit() - except Exception as e: - print("There is an issue inserting alert information into fall_alerts") - # close connection - conn.close() - - -def storeRobberyAlertData(alertTime, videoLink, AlertClass): - - cursor, conn = connect() - # Connection achieved - # Storing alert data - query = "insert into robbery_alerts (alert_time, video_link, class) values ( %s, %s, %s)" - - cursor.execute(query, (alertTime, videoLink, AlertClass)) - conn.commit() - - # Close connection - conn.close() - - -def retrieve_fire_alerts(): - - cursor, conn = connect() - - query = "select * from fire_alerts" - cursor.execute(query) - print("Fire alerts : \n ------------------------------------ \n") - fire_alerts = cursor.fetchall() - for row in fire_alerts: - print(f"ID : {row[0]} | alert time : {row[1]} | video link : {row[2]} | class : {row[3]}") - - conn.close() - - -def retrieve_fall_alerts(): - - cursor, conn = connect() - query = "select * from fall_alerts" - cursor.execute(query) - print("Fall alerts : \n ------------------------------------ \n") - fall_alerts = cursor.fetchall() - for row in fall_alerts: - print(f"ID : {row[0]} | alert time : {row[1]} | video link : {row[2]} | class : {row[3]}") - - conn.close() - - -def retrieve_robbery_alerts(): - - cursor, conn = connect() - - query = "select * from robbery_alerts" - cursor.execute(query) - print("Robbery alerts : \n ------------------------------------ \n") - robbery_alerts = cursor.fetchall() - for row in robbery_alerts: - print(f"ID : {row[0]} | alert time : {row[1]} | video link : {row[2]} | class : {row[3]}") - - conn.close() - +def store_alert_data(alert_time, video_link, alert_class, alert_type): + + query = f"INSERT INTO {alert_type}_alerts (alert_time, video_link, class) VALUES (%s, %s, %s)" + + with get_db_connection() as (cursor, conn): + try: + cursor.execute(query, (alert_time, video_link, alert_class)) + conn.commit() + except Exception as e: + print(f"Error inserting alert into {alert_type}_alerts: {e}") + +def retrieve_alerts(alert_type): + + query=f"SELECT * FROM {alert_type}_alerts" + + with get_db_connection() as (cursor, conn): + cursor.execute(query) + alerts = cursor.fetchall() + print(f"{alert_type.capitalize()} alerts: \n ----------------------------------------------- \n") + + for row in alerts: + print(f"ID: {row[0]} | Alert Time: {row[1]} | Video Link; {row[2]} | Class: {row[3]}") def retrieve_all_alerts(): - cursor, conn = connect() - - retrieve_fire_alerts() - retrieve_fall_alerts() - retrieve_robbery_alerts() - conn.close() - -# storeFireAlertData("{20:20:20}", "{link}", True) -# storeMouvementAlertData("{20:20:20}", "{link}", True) - + retrieve_alerts("fire") + retrieve_alerts("fall") + retrieve_alerts("robbery") + def retrieve_users(): - cursor, conn = connect() - - query = "select * from users" - cursor.execute(query) - - result = cursor.fetchall() - for row in result: - print(f"id {row[0]} | username : {row[1]} | password : {row[2]}") - - conn.close() - -# add_user("user", "user") -# retrieve_users() \ No newline at end of file + # Establishing Connection to DB + with get_db_connection() as (cursor, conn): + try: + query = "select * from users" + cursor.execute(query) + + result = cursor.fetchall() + for row in result: + print(f"ID: {row[0]} | Username: {row[1]}") + except Exception as e: + print(f"Error retrieving users: {e}") + +def remove_camera(camera_id): + + with get_db_connection() as (cursor, conn): + + try: + query = "DELETE FROM cameras WHERE id = %s" + cursor.execute(query, (camera_id,)) + conn.commit() + print(f"Camera with ID {camera_id} removed successfully.") + except Exception as e: + print(f"Error removing camera: {e}") diff --git a/ImagesScrapper.py b/ImagesScrapper.py index bb5189d..af1a5f1 100644 --- a/ImagesScrapper.py +++ b/ImagesScrapper.py @@ -1,63 +1,94 @@ import hashlib -import io import time +import io import os import requests from bs4 import BeautifulSoup -import pandas as pd -import numpy as np import shutil from tqdm import tqdm from selenium import webdriver from PIL import Image import signal +import platform +import threading + +class TimeoutException(Exception): + pass -driver_path = '/home/iheb/chromedriver' -output_path = 'data/images/robbery_images' -number_of_images = 1000 -GET_IMAGE_TIMEOUT = 2 -SLEEP_BETWEEN_INTERACTIONS = 0.1 -SLEEP_BEFORE_MORE = 5 -IMAGE_QUALITY = 1024 -search_terms = ["armed robbery", - "shop robbery", - "man wearing robber mask", - "man wearing robber mask and knife", - "shop armed looting", - - "persons", - "store customers", - "faces", - "covid mask", - "person portrait", - "full body person portrait", - "person smiling"] -# search_terms = ["armed masked thief"] class timeout: def __init__(self, seconds= 1, error_message="Timeout"): self.seconds = seconds self.error_message = error_message - + self.os_is_windows = platform.system().lower() == 'windows' + def handle_timeout(self, signum, frame): raise TimeoutError(self.error_message) def __enter__(self): - signal.signal(signal.SIGALRM, self.handle_timeout) - signal.alarm(self.seconds) + if self.os_is_windows: + # For better portability a timeout class for windows is needed + self.timer = threading.Timer(self.seconds, self._raise_timeout) + self.timer.start() + else: + # Use signal for Unix-based systems + signal.signal(signal.SIGALRM, self.handle_timeout) + signal.alarm(self.seconds) def __exit__(self, type, value, traceback): - signal.alarm(0) - + if self.os_is_windows: + # Cancel timer for windows + self.timer.cancel() + else: + # Disable Unix alarm + signal.alarm(0) + + def _raise_timeout(self): + raise TimeoutException(self.error_message) + +class ScraperConfig: + + def __init__(self, driver_path, output_path, number_of_images, get_image_timeout, sleep_between_interactions, sleep_before_more + , image_quality, search_terms): + self.driver_path = driver_path + self.output_path = output_path + self.number_of_images = number_of_images + self.get_image_timeout = get_image_timeout + self.sleep_between_interactions = sleep_between_interactions + self.sleep_before_more = sleep_before_more + self.image_quality = image_quality + self.search_terms = search_terms + +config = ScraperConfig( + driver_path = '/home/iheb/chromedriver', + output_path = 'data/images/robbery_images', + number_of_images = 1000, + GET_IMAGE_TIMEOUT = 2, + SLEEP_BETWEEN_INTERACTIONS = 0.1, + SLEEP_BEFORE_MORE = 5, + IMAGE_QUALITY = 1024, + search_terms = ["armed robbery", + "shop robbery", + "man wearing robber mask", + "man wearing robber mask and knife", + "shop armed looting", + "persons", + "store customers", + "faces", + "covid mask", + "person portrait", + "full body person portrait", + "person smiling"] + ) def fetch_image_urls(query: str, max_links_to_fetch: int, wd: webdriver, - sleep_between_interactions: int = 1): + config: ScraperConfig): def scroll_to_end(wd): wd.execute_script("window.scrollTo(0, document.body.scrollHeight);") - time.sleep(sleep_between_interactions) + time.sleep(config.sleep_between_interactions) # building google query search_url = "https://www.google.com/search?safe=off&site=&tbm=isch&source=hp&q={q}&oq={q}&gs_l=img" @@ -83,7 +114,7 @@ def scroll_to_end(wd): # try to click every thumbnail such that we can get the real image behind it try: img.click() - time.sleep(sleep_between_interactions) + time.sleep(config.sleep_between_interactions) except Exception as e: print(f"could not click image - {e}") continue @@ -127,11 +158,11 @@ def scroll_to_end(wd): return image_urls -def persist_image(folder_path:str, url:str): +def persist_image(folder_path:str,url:str, config: ScraperConfig): try: print("getting the image...") # download the image, if timeout is exceeded throw an error - with timeout(GET_IMAGE_TIMEOUT): + with timeout(config.get_image_timeout): image_content = requests.get(url).content except Exception as e: print(f"Error - Could not download {url} - {e}") @@ -142,14 +173,13 @@ def persist_image(folder_path:str, url:str): file_path = os.path.join(folder_path, hashlib.sha1(image_content).hexdigest()[:10] + '.jpg') with open(file_path, 'wb') as f: - image.save(f, "JPEG", quality=IMAGE_QUALITY) + image.save(f, "JPEG", quality=config.image_quality) print(f"Success - Saved {url} - as {file_path} ") except Exception as e: print(f"Error - could not save {url} - {e}") -def search_download(search_term:str, target_path="data/images/robbery_images", number_images=5): - +def search_download(search_term:str, config: ScraperConfig, target_path: str, number_images: int): # create a folder name target_folder = os.path.join(target_path, '_'.join(search_term.lower().split(" "))) @@ -158,8 +188,8 @@ def search_download(search_term:str, target_path="data/images/robbery_images", n os.makedirs(target_folder) # launch chrome - with webdriver.Chrome(executable_path=driver_path) as wd: - res = fetch_image_urls(search_term, number_images, wd= wd, sleep_between_interactions=SLEEP_BETWEEN_INTERACTIONS) + with webdriver.Chrome(executable_path=config.driver_path) as wd: + res = fetch_image_urls(search_term, number_images, wd= wd, sleep_between_interactions=config.sleep_between_interactions) # download images if res is not None: @@ -169,60 +199,9 @@ def search_download(search_term:str, target_path="data/images/robbery_images", n print(f"failed to return links for terms : {search_term}") -for term in search_terms: +for term in config.search_terms: search_download(term, - output_path, - number_of_images) - -# search_term = "dog" -# search_download(search_term=search_term, -# driver_path=driver_path, -# target_path="data/images/robbery_images") -# def ImageScrapper(url): -# -# response = requests.get(url) -# soup = BeautifulSoup(response.content, 'html.parser') -# base_link = '' -# index = 0 -# # for item in soup.find_all('img'): -# # img_link = item.attrs['src'] -# # index += 1 -# # print(f"image number : {index} link : {img_link}") -# # -# # full_url = url + img_link -# # -# # r = requests.get(full_url, -# # stream=True) -# # print(f"code : {r.status_code}") -# # print(r.raw) -# # if r.status_code == 200: -# # print(f"everything is okkay for image number {index}") -# # with open("data/images/robbery_images/img" +str(index)+ ".jpg", 'wb') as f: -# # r.raw.decode_content = True -# # shutil.copyfileobj(r.raw, f) -# -# images = soup.find_all('img') -# print(images[0]) -# img_src = images[0].attrs['src'] -# full_link = url + img_src -# print(full_link) -# -# split_string = img_src.split(".",1) -# print(split_string[1]) -# -# r = requests.get(full_link, stream=True) -# if r.status_code == 200: -# with open("data/images/robbery_images/img"+str(index)+"."+str(split_string[1]) , "wb") as f: -# r.raw.decode_content = True -# shutil.copyfileobj(r.raw, f) -# ImageScrapper("https://www.google.com/search?q=armed+robbery+jpg&tbm=isch&client=opera&hs=Gao&hl=en&sa=X&ved=2ahUKEwjMmPa8s6byAhVaO-wKHXmYAQYQBXoECAEQIw&biw=1865&bih=952") - - -# def getData(url): -# r = requests.get(url) -# return r.text -# -# htmldata = getData("https://www.istockphoto.com/photos/armed-robbery") -# soup = BeautifulSoup(htmldata, 'html.parser') -# for item in soup.find_all('img'): -# print(item['src']) + config, + config.output_path, + config.number_of_images + ) \ No newline at end of file diff --git a/MachineVision/RobberyDetection/base.py b/MachineVision/RobberyDetection/base.py index d284a68..7d728fd 100644 --- a/MachineVision/RobberyDetection/base.py +++ b/MachineVision/RobberyDetection/base.py @@ -1,95 +1,8 @@ -# import cv2 -# -# from multiprocess import QValkkaOpenCVProcess -# from PySide6 import QtCore -# import time -# from keras.models import load_model -# from keras.preprocessing.image import img_to_array -# import tensorflow as tf -# -# # Local imports -# from .PersonDetector import PersonDetector -# -# -# class QValkkaRobberyDetectorProcess(QValkkaOpenCVProcess): -# incoming_signal_defs = { -# "create_client_": [], -# "test_": {"test_int": int, "test_str": str}, -# "ping_": {"message": str} -# } -# -# outgoing_signal_defs = { -# "Robbery_detected": {} -# } -# -# class Signals(QtCore.QObject): -# -# Robbery_detected = QtCore.Signal() -# -# def __init__(self, name, **kwargs): -# super().__init__(name, **kwargs) -# self.signals = self.Signals() -# self.personDetector = PersonDetector() -# self.RobberyDetector = load_model('/home/iheb/PycharmProjects/Vision-Alarm/MachineVision/RobberyDetection/Robbery_Detection_Model3.h5') -# -# def alarm(self): -# print("Robbery detected -- inside alarm") -# self.sendSignal_(name="Robbery_detected") -# -# def cycle_(self): -# # print("inside robbery detection") -# if self.client is None: -# time.sleep(3.0) -# # print('client timedout') -# else: -# index, isize = self.client.pull() -# if (index is None): -# print(self.pre, "Client timed out..") -# pass -# -# else: -# print("Client index, size =", index, isize) -# try: -# data = self.client.shmem_list[index] -# # print(data) -# except BaseException: -# print("There is an issue in getting data from shmem_list") -# try: -# img = data.reshape( -# (self.image_dimensions[1], self.image_dimensions[0], 3)) -# except BaseException: -# print("QValkkaRobberyDetectorProcess: WARNING: could not reshape image") -# -# # if self.personDetector.cycle(img): -# # print(self.personDetector.cycle(img)) -# # print("Now let's detect if there is ongoing robbery ") -# -# img_resized = cv2.resize(img, (224,224)) -# img_array = img_to_array(img=img_resized) -# img_array = tf.expand_dims(img_array, 0) -# -# predictions = self.RobberyDetector.predict(img_array) -# score = predictions[0] -# print("this image is %.2f percent No robbery and %.2f robber" % (100 * (1 - score), 100 * score)) -# -# # else: -# print(self.personDetector.cycle(img)) -# print("Nothing detected yet ! ") -# # pass -# # if self.RobberyDetector(img): -# # print("yeaaaaaaaaaaaaah") -# # else: -# # print("tnekna") -# # ** Frontend methods handling recieved outgoing signals -# -# def Robbery_detected(self): -# print("At frontend: robbery detected ") -# self.signals.Robbery_detected.emit() - import cv2 import time import tensorflow as tf import numpy as np +import logging from PIL import Image from keras.models import load_model from keras.preprocessing.image import img_to_array @@ -97,8 +10,8 @@ from PySide6 import QtCore # local imports -from DataBase import storeFireAlertData -from cloudStorage import uploadBlob +from DataBase import store_alert_data +from cloudStorage import upload_blob from AlertAdmin import send_sms from multiprocess import QValkkaOpenCVProcess from .PersonDetector import PersonDetector @@ -134,33 +47,37 @@ class Signals(QtCore.QObject): def __init__(self, name, **kwargs): super().__init__(name, **kwargs) # does parameterInitCheck self.signals = self.Signals() - - # # parameterInitCheck(QValkkaMovementDetectorProcess.parameter_defs, kwargs, self) - # self.analyzer=MovementDetector(verbose=True) - # self.analyzer = MovementDetector(treshold=0.0001)# To be changed - self.personDetector = PersonDetector() self.RobberyDetector = load_model('/home/iheb/PycharmProjects/Vision-Alarm/MachineVision/RobberyDetection/Robbery_Detection_Model3.h5') def alarm(self): - print('Robbery Robbery') + + logging.debug(f"Robbery detected") self.sendSignal_(name="Robbery_detected") def cycle_(self): - # print('inside Robbery detection') + """ + Cycle function will be automatically called within QValka main process and runs the expected Machine vision analyses + on the passed frames. + + + If robbery detected, The frames of the incident will be stored in the cloud And the admin will be alerted through an SMS using twilio. + + """ + if self.client is None: time.sleep(1.0) - print('client timedout') + logging('client timedout') else: index, isize = self.client.pull() if (index is None): - # print(self.pre, "Client timed out..") + logging(f"{self.pre} Client timed out..") pass else: - print("Client index, size =", index, isize) + logging(f"Client index: {index} size: {isize}") try: data = self.client.shmem_list[index] - # print(data) + except BaseException: print("There is an issue in getting data from shmem_list") try: @@ -173,25 +90,27 @@ def cycle_(self): img_resized = cv2.resize(img, (224, 224)) img = Image.fromarray(img_resized) - print(img_resized.shape) - print(type(img_resized)) - print(type(img)) + logging(img_resized.shape) + img_array = img_to_array(img=img) img_array = tf.expand_dims(img_array, 0) - print("img : ",type(img)) - print("img_array : ",type(img_array)) - # print(img_array) - # try: - # predictions = self.RobberyDetector.predict(img_array) - # print("preds :",predictions) - # score = predictions[0] - # print("this image is %.2f No Robber and %.2f Robbery" %(100 * (1-score), 100 * score)) - # except Exception as e: - # print("Unable to predict image class : "+str(e)) - # ** frontend methods handling received outgoing signals *** + logging("img : ",type(img)) + logging("img_array : ",type(img_array)) + + try: + predictions = self.RobberyDetector.predict(img_array) + logging("preds :",predictions) + score = predictions[0] + logging("This image is %.2f No Robber and %.2f Robbery" %(100 * (1-score), 100 * score)) + except Exception as e: + print(f"Unable to predict image class : {e}") + def Robbery_detected(self): - print("At frontend: Robbery detected ") + """ + Emits the robbery detection signal when a robbery is detected. + """ + logging("At frontend: Robbery detected ") self.signals.Robbery_detected.emit() diff --git a/Streaming/FilterChain.py b/Streaming/FilterChain.py index 254b5d4..ba59f2b 100644 --- a/Streaming/FilterChain.py +++ b/Streaming/FilterChain.py @@ -5,161 +5,6 @@ from valkka.api2 import FragMP4ShmemClient from valkka.api2.logging import setValkkaLogLevel, loglevel_silent - -# class BasicFilterchain: -# """This class implements the following filterchain: -# -# :: -# -# (LiveThread:livethread) -->> (AVThread:avthread) -->> (OpenGLThread:glthread) -# -# i.e. the stream is decoded by an AVThread and sent to the OpenGLThread for presentation -# """ -# setValkkaLogLevel(loglevel_silent) -# parameter_defs = { -# "livethread": LiveThread, -# "openglthread": OpenGLThread, -# "address": str, -# "slot": int, -# -# # these are for the AVThread instance: -# "n_basic": (int, 20), # number of payload frames in the stack -# "n_setup": (int, 20), # number of setup frames in the stack -# "n_signal": (int, 20), # number of signal frames in the stack -# "flush_when_full": (bool, False), # clear fifo at overflow -# -# "affinity": (int, -1), -# "verbose": (bool, False), -# "msreconnect": (int, 0), -# -# # Timestamp correction type: TimeCorrectionType_none, -# # TimeCorrectionType_dummy, or TimeCorrectionType_smart (default) -# "time_correction": None, -# # Operating system socket ringbuffer size in bytes # 0 means default -# "recv_buffer_size": (int, 0), -# # Reordering buffer time for Live555 packets in MILLIseconds # 0 means -# # default -# "reordering_mstime": (int, 0), -# "n_threads": (int, 1) -# } -# -# def __init__(self, **kwargs): -# # auxiliary string for debugging output -# self.pre = self.__class__.__name__ + " : " -# # check for input parameters, attach them to this instance as -# # attributes -# parameterInitCheck(self.parameter_defs, kwargs, self) -# self.init() -# -# def init(self): -# self.idst = str(id(self)) -# self.makeChain() -# self.createContext() -# self.startThreads() -# self.active = True -# -# def __del__(self): -# self.close() -# -# def close(self): -# if (self.active): -# if (self.verbose): -# print(self.pre, "Closing threads and contexes") -# self.decodingOff() -# self.closeContext() -# self.stopThreads() -# self.active = False -# -# def makeChain(self): -# """Create the filter chain -# """ -# self.gl_in_filter = self.openglthread.getInput( -# ) # get input FrameFilter from OpenGLThread -# -# self.framefifo_ctx = core.FrameFifoContext() -# self.framefifo_ctx.n_basic = self.n_basic -# self.framefifo_ctx.n_setup = self.n_setup -# self.framefifo_ctx.n_signal = self.n_signal -# self.framefifo_ctx.flush_when_full = self.flush_when_full -# -# self.avthread = core.AVThread( -# "avthread_" + self.idst, -# self.gl_in_filter, -# self.framefifo_ctx) -# -# if self.affinity > -1 and self.n_threads > 1: -# print("WARNING: can't use affinity with multiple threads") -# -# self.avthread.setAffinity(self.affinity) -# if self.affinity > -1: -# self.avthread.setNumberOfThreads(self.n_threads) -# -# # get input FrameFilter from AVThread -# self.av_in_filter = self.avthread.getFrameFilter() -# -# def createContext(self): -# """Creates a LiveConnectionContext and registers it to LiveThread -# """ -# # define stream source, how the stream is passed on, etc. -# -# self.ctx = core.LiveConnectionContext() -# # slot number identifies the stream source -# self.ctx.slot = self.slot -# -# if (self.address.find("rtsp://") == 0): -# self.ctx.connection_type = core.LiveConnectionType_rtsp -# else: -# self.ctx.connection_type = core.LiveConnectionType_sdp # this is an rtsp connection -# -# self.ctx.address = self.address -# # stream address, i.e. "rtsp://.." -# -# self.ctx.framefilter = self.av_in_filter -# -# self.ctx.msreconnect = self.msreconnect -# -# # some extra parameters -# """ -# // ctx.time_correction =TimeCorrectionType::none; -# // ctx.time_correction =TimeCorrectionType::dummy; -# // default time correction is smart -# // ctx.recv_buffer_size=1024*1024*2; // Operating system ringbuffer size for incoming socket -# // ctx.reordering_time =100000; // Live555 packet reordering treshold time (microsecs) -# """ -# if (self.time_correction is not None): -# self.ctx.time_correction = self.time_correction -# # self.time_correction=core.TimeCorrectionType_smart # default .. -# self.ctx.recv_buffer_size = self.recv_buffer_size -# self.ctx.reordering_time = self.reordering_mstime * \ 1000 # from millisecs to microsecs -# -# # send the information about the stream to LiveThread -# self.livethread.registerStream(self.ctx) -# self.livethread.playStream(self.ctx) -# -# -# def closeContext(self): -# self.livethread.stopStream(self.ctx) -# self.livethread.deregisterStream(self.ctx) -# -# def startThreads(self): -# """Starts thread required by the filter chain -# """ -# self.avthread.startCall() -# -# def stopThreads(self): -# """Stops threads in the filter chain -# """ -# self.avthread.stopCall() -# -# def decodingOff(self): -# self.avthread.decodingOffCall() -# -# def decodingOn(self): -# self.avthread.decodingOnCall() - - - - class VisionAlarmFilterChain: """A filter chain with a shared mem hook and FragMP4ShmemFrameFilter @@ -298,9 +143,7 @@ def makeChain(self): # self.av_in_filter1_1 = self.avthread1_1.getFrameFilter() # Branch 2 : Saving frames to shared memory for openCV/Tensorflow process - # these two lines for debugging bullshit so feel free to comment/uncomment them ya man - print(self.pre, "using shmem name ", self.shmem_name) - print(self.shmem_name) + try: self.shmem_filter = core.RGBShmemFrameFilter( self.shmem_name, @@ -345,9 +188,7 @@ def makeChain(self): self.interval_filter ) - se - - + # Main branch self.framefifo_ctx = core.FrameFifoContext() self.framefifo_ctx.n_basic = self.n_basic diff --git a/cloudStorage.py b/cloudStorage.py index 97a7dc1..d20546a 100644 --- a/cloudStorage.py +++ b/cloudStorage.py @@ -2,78 +2,62 @@ from datetime import datetime import cv2 -import numpy as np +import tempfile from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient, __version__ -def initBlogClient(): +def init_blob_client(): """ - TODO: remove cloud connection string + Initialize the blob service from Azure storage connection string. + Ensures connection string is retrieved securely from env variables. """ + connect_string = os.getenv('AZURE_STORAGE_CONNECTION_STRING') + if not connect_string: + raise EnvironmentError("AZURE_STORAGE_CONNECTION_STRING env variable not set.") + blob_service_client = BlobServiceClient.from_connection_string(connect_string) return blob_service_client -def createAzureContainer(): - - blob_service_client = initBlogClient() - - container_name = "alerts" - - container_client = blob_service_client.create_container(container_name) - - -def uploadBlob(videoArray, videoName, width, height, fps): - - blob_service_client = initBlogClient() - current_time = datetime.now() - current_day = datetime.today() - current_time = current_time.strftime("%H:%M:%S") - videoName = videoName + " " + str(current_day) + " "+ current_time + ".mp4" - blob_client = blob_service_client.get_blob_client(container="alerts", blob=videoName) - - # Converting videoArray ( numpy array ) into video - # fps = 30 # 25 frames per second - - - # print(videoName) - output = cv2.VideoWriter(videoName, cv2.VideoWriter_fourcc(*'mp4v'), fps, (width, height), True) - for i in videoArray: - output.write(i) - output.release() - - path = "./" + videoName - # print("path : ", path) - with open(path, "rb") as data : - # Uploading video to the cloud - blob_client.upload_blob(data) - # Delete file after upload - - os.remove(path) - - return blob_client.url - -# frames = [] -# path = "./data/test.mp4" -# -# cap = cv2.VideoCapture(path) -# -# if cap.isOpened(): -# width = int(cap.get(3)) -# height = int(cap.get(4)) -# print("width : ", width) -# print("height : ", height) -# ret = True -# while ret: -# ret, img = cap.read() -# if ret: -# frames.append(img) -# video = np.stack(frames, axis = 0) -# -# # print(video) -# -# print(uploadBlob(video, "Alert", width, height)) +def create_azure_container(container_name = "alerts"): + + blob_service_client = init_blob_client() + try: + container_client = blob_service_client.create_container(container_name) + except Exception as e: + print(f"Error creating container {container_name}: {e}") + +def upload_blob(video_array, video_name, width, height, fps): + + blob_service_client = init_blob_client() + current_time = datetime.now().strftime("%H:%M:%S") + current_day = datetime.today().strftime("%Y-%m-%d") + + video_name = f"{video_name} {current_day} {current_time}.mp4" + blob_client = blob_service_client.get_blob_client(container="alerts", blob=video_name) + + try: + with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as temp_video_file: + output = cv2.VideoWriter(temp_video_file.name, cv2.VideoWriter_fourcc(*'mp4v'), fps, (width, height), True) + for i in video_array: + output.write(i) + output.release() + + with open(temp_video_file.name, "rb") as data: + # Uploading video to the cloud + blob_client.upload_blob(data) + + os.remove(temp_video_file.name) # removing temporary file after upload + return blob_client.url + + except Exception as e: + print(f"Error uploading video {video_name}: {e}") + return None + + + +