Gui

SendToSQSOnS3Upload/lambda_function.py

@@ -68,6 +68,9 @@ def launch_ec2():
 # Install OpenCV dependencies
 apt install python3-opencv -y
 
+# Install MPI lib
+apt install python3-mpi4py -y
+
 # Create server.py file
 cat << EOF > /home/ubuntu/server.py
 import boto3

gui2.py

@@ -0,0 +1,264 @@
+import random
+import threading
+
+from flask import Flask, request, render_template, redirect, url_for
+import os
+from concurrent.futures import ThreadPoolExecutor
+import boto3
+import hashlib
+import time
+import shutil
+import uuid
+import asyncio
+import aioboto3
+from botocore.exceptions import ClientError, NoCredentialsError, PartialCredentialsError
+
+app = Flask(__name__)
+
+# AWS S3 Client setup
+s3_client = boto3.client('s3')
+ec2_client = boto3.client('ec2')
+
+# Ensure the upload folder exists
+UPLOAD_FOLDER = r"D:\UNI\sems\2024 spring\Distributed Computing\Project\static"
+os.makedirs(UPLOAD_FOLDER, exist_ok=True)
+app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
+
+paths_list = []
+
+
+# Function to check if an object exists in S3
+def check_object_exists(bucket_name, object_key):
+    try:
+        s3_client.head_object(Bucket=bucket_name, Key=object_key)
+        print(f"Object '{object_key}' exists in bucket '{bucket_name}'.")
+        return True
+    except ClientError as e:
+        if e.response['Error']['Code'] == '404':
+            return False
+        else:
+            print('Other errors')
+            return False
+    except (NoCredentialsError, PartialCredentialsError):
+        print("Credentials not available or incomplete.")
+        return False
+
+
+# Function to generate a unique key
+def generate_key():
+    # Generate the MAC address
+    mac = ':'.join(['{:02x}'.format((uuid.getnode() >> elements) & 0xff) for elements in range(0, 2 * 6, 2)][::-1])
+    print('MAC is ' + mac)
+
+    # Get the current time
+    current_time = str(time.time()).encode('utf-8')
+
+    # Generate a random value
+    random_value = str(random.randint(0, 1000000)).encode('utf-8')
+
+    # Create a hash object
+    hash_object = hashlib.md5()
+
+    # Update the hash object with MAC address, current time, and random value
+    hash_object.update(mac.encode('utf-8'))
+    hash_object.update(current_time)
+    hash_object.update(random_value)
+
+    # Generate the key
+    key = hash_object.hexdigest()
+
+    return key
+
+
+# Function to check the result in S3
+def check_result(key):
+    print('thread started')
+    start_time = time.time()
+    while True:
+        if check_object_exists('test-n-final-bucket', key):
+            s3_client.download_file('test-n-final-bucket', key, key + '.jpg')
+            image_name = str(key) + '.jpg'
+            current_directory = os.getcwd()
+            source_path = os.path.join(current_directory, image_name)
+            destination_path = "D:\\UNI\sems\\2024 spring\\Distributed Computing\\Project\\static"
+            shutil.move(source_path, destination_path)
+            end_time = time.time()
+            s3_client.delete_object(Bucket='test-n-final-bucket', Key=key)
+            print('Got the object from the bucket directly in', end_time - start_time)
+            break
+
+
+# Photo processing function
+def process_photo(file, op):
+    print(op + ' entered')
+    key = generate_key()
+    s3_client.put_object(
+        Bucket='kmna-juju-bucket',
+        Key=key,
+        Body=file,
+        ContentType='image/jpg',  # Adjust content type if needed
+        Metadata={'operation': op}
+    )
+
+    # Create an event to signal when the thread finishes
+    # check_result(image_key)
+    print('thread started')
+    start_time = time.time()
+    while True:
+        if check_object_exists('test-n-final-bucket', key):
+            s3_client.download_file('test-n-final-bucket', key, key + '.jpg')
+            image_name = str(key) + '.jpg'
+            current_directory = os.getcwd()
+            source_path = os.path.join(current_directory, image_name)
+            destination_path = "D:\\UNI\sems\\2024 spring\\Distributed Computing\\Project\\static"
+            shutil.move(source_path, destination_path)
+            end_time = time.time()
+            s3_client.delete_object(Bucket='test-n-final-bucket', Key=key)
+            print('Got the object from the bucket directly in', end_time - start_time)
+            break
+    return key + '.jpg'
+
+
+# def upload_to_s3(key, op, file):
+#     print('entered put_object thread')
+#     s3_client.put_object(
+#         Bucket='kmna-juju-bucket',
+#         Key=key,
+#         Body=file,
+#         ContentType='image/jpg',  # Adjust content type if needed
+#         Metadata={'operation': op}
+#     )
+#     print('finished putting the object')
+
+
+# Route to handle file uploads
+@app.route('/', methods=['GET', 'POST'])
+def upload_file():
+    if request.method == 'POST':
+        if 'file' not in request.files:
+            return render_template('index3.html', message='No file part')
+
+        files = request.files.getlist('file')
+        option = request.form['options']
+
+        if not files or files[0].filename == '':
+            return render_template('index3.html', message='No selected file')
+
+        global paths_list
+        key_list = []
+        for file in files:
+            key = generate_key()
+            # thread = threading.Thread(target=upload_to_s3, args=(key, option, file))
+            # thread.start()
+            # threads.append(thread)
+            key_list.append(key)
+            paths_list.append((file, key + '.jpg'))
+
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            loop.run_until_complete(big(key_list, files,option))
+        except Exception as e:
+            print('dadada')
+            print(f"RuntimeError: {e}")
+
+        for key in key_list:
+            while True:
+                start_time = time.time()
+                if check_object_exists('test-n-final-bucket', key):
+                    s3_client.download_file('test-n-final-bucket', key, key + '.jpg')
+                    image_name = str(key) + '.jpg'
+                    current_directory = os.getcwd()
+                    source_path = os.path.join(current_directory, image_name)
+                    destination_path = "D:\\UNI\sems\\2024 spring\\Distributed Computing\\Project\\static"
+                    shutil.move(source_path, destination_path)
+                    end_time = time.time()
+                    s3_client.delete_object(Bucket='test-n-final-bucket', Key=key)
+                    print('Got the object from the bucket directly in', end_time - start_time)
+                    break
+
+        return render_template('success.html', processed_paths=paths_list, enumerate=enumerate)
+
+    return render_template('index3.html')
+
+
+# Helper function to handle file processing
+def process_file(file, option):
+    filename = file.filename
+    file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
+    file.save(file_path)
+    with open(file_path, 'rb') as f:
+        image_data = f.read()
+    file_size = len(image_data)
+    print(file_size)
+    processed_path = process_photo(image_data, option)
+    global paths_list
+    paths_list.append((file, processed_path))
+    return processed_path
+
+
+# Route to display result
+@app.route('/result/<filename>/<processed_path>')
+def success(filename, processed_path):
+    return render_template('result.html', filename=filename, processed_path=processed_path)
+
+
+# Function to get EC2 instance information
+def get_ec2_info(ec2_client):
+    response = ec2_client.describe_instances()
+    instances = []
+    for reservation in response['Reservations']:
+        for instance in reservation['Instances']:
+            instance_info = {
+                'Instance ID': instance['InstanceId'],
+                'Type': instance['InstanceType'],
+                'State': instance['State']['Name'],
+                'Public IP': instance.get('PublicIpAddress', 'N/A')
+            }
+            instances.append(instance_info)
+    print(instances)
+    return instances
+
+
+# Route to display backend information
+@app.route('/backend')
+def backend():
+    data = get_ec2_info(ec2_client)
+    return render_template('backend.html', data=data)
+
+
+async def upload_to_s3(client, key, file_data, op):
+    s3_bucket = 'kmna-juju-bucket'
+    await client.put_object(
+        Bucket=s3_bucket,
+        Key=key,
+        Body=file_data,
+        ContentType='image/jpg',  # Adjust content type if needed
+        Metadata={'operation': op}
+    )
+    print(f'Finished uploading {key}')
+
+
+async def big(keys, files, op):
+    print('entered big')
+    l = ['hush.jpg', 'face.jpg', 'square.jpg', 'phone.jpg', 'star.jpg', '1.jpg', '2.jpg', '3.jpg', '4.jpg', '5.jpg']
+    print(len(l))
+    session = aioboto3.Session()
+
+    async with session.client('s3') as s3_client:
+        tasks = []
+        for key, pic in zip(keys, files):
+            try:
+                print('try to open file')
+
+                file_content = pic.read()
+                tasks.append(upload_to_s3(s3_client, key, file_content, op))
+                print('passed file to upload')
+            except FileNotFoundError:
+                print(f"File not found: {key}")
+
+        await asyncio.gather(*tasks)
+
+
+if __name__ == '__main__':
+    app.run(debug=True, port=3000)

main.py

@@ -4,6 +4,7 @@
 import hashlib
 import threading
 import logging
+from botocore.exceptions import NoCredentialsError, PartialCredentialsError, ClientError
 
 # Configure logging
 logging.basicConfig(filename='system_logs.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
@@ -23,8 +24,43 @@ def generate_key():
     key = hash_object.hexdigest()
     return key
 
+def check_object_exists(bucket_name, object_key):
+    # Create a session using your credentials or rely on default session
+
+    try:
+        # Try to get the object metadata
+        s3_client.head_object(Bucket=bucket_name, Key=object_key)
+        print(f"Object '{object_key}' exists in bucket '{bucket_name}'.")
+        return True
+    except ClientError as e:
+        # If a client error is raised, the object does not exist
+        if e.response['Error']['Code'] == '404':
+            #print(f"Object '{object_key}' does not exist in bucket '{bucket_name}'.")
+            return False
+        else:
+            print('other errors')
+            # For other errors, raise the exception
+            return False
+    except (NoCredentialsError, PartialCredentialsError):
+        print("Credentials not available or incomplete.")
+        return False
+
+def check_result_from_bucket(key):
+    print('entered check from bucket')
+    # Record the start time
+    start_time = time.time()
+    while True:
+        if check_object_exists('test-n-final-bucket',key):
+            s3_client.download_file('test-n-final-bucket', key, 'Bucket'+key + '.jpg')
+            # Record the end time
+            end_time = time.time()
+            print('got the object from the bucket directly in ', end_time - start_time)
+
+            break
+
 
 def check_result(key):
+    start_time = time.time()
     queue_url = 'https://sqs.eu-north-1.amazonaws.com/992382542532/results-queue'
     while True:
         logging.info('waiting for the image')
@@ -39,7 +75,7 @@ def check_result(key):
             logging.info('message body is ' + message['Body'] + 'and key is ' + key)
             if message['Body'] != key:
                 # it's not the message you are waiting for so sleep to allow others to get the message
-                time.sleep(2)
+                #time.sleep(2)
                 continue
             sqs.delete_message(
                 QueueUrl=queue_url,
@@ -50,7 +86,8 @@ def check_result(key):
                 Bucket='test-n-final-bucket',
                 Key=key,
             )
-            print('we got your image')
+            end_time = time.time()
+            print('we got your image in ', end_time-start_time)
             break
         else:
             logging.info('waiting for the image')
@@ -62,7 +99,7 @@ def check_result(key):
     if image_path == ':q':
         break
     operation = input('Choose a number Operation to do \n1-Blur \n2-Convert to Grayscale \n3-Dilate \n4-Erode'
-                      '\n5-open \n6-close \n7-edge-detection \n8-threshold \n9-contour-detection\n')
+                      '\n5-open \n6-close \n7-edge-detection \n8-threshold \n9-contour-detection \n10-face detection\n')
 
     if operation == '1':
         op = 'blur'
@@ -102,4 +139,7 @@ def check_result(key):
     )
 
     thread = threading.Thread(target=check_result, args=(image_key,))
+    thread2 =  threading.Thread(target=check_result_from_bucket, args=(image_key,))
+    thread2.start()
     thread.start()
+

server.py

@@ -2,7 +2,7 @@
 import cv2
 import numpy as np
 import logging
-
+import requests
 # Configure logging
 logging.basicConfig(filename='system_logs.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
@@ -47,6 +47,17 @@ def process_image(image, op):
         _, thresh = cv2.threshold(gray_image, 127, 255, cv2.THRESH_BINARY)
         contours, _ = cv2.findContours(thresh, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)
         return cv2.drawContours(img_np, contours, -1, (0, 255, 0), 3)
+    elif op == 'face-detection':
+        xml_url = 'https://github.com/opencv/opencv/raw/master/data/haarcascades/haarcascade_frontalface_default.xml'
+        req = requests.get(xml_url)
+        with open('haarcascade_frontalface_default.xml', 'wb') as file:
+            file.write(req.content)
+        face_cascade = cv2.CascadeClassifier('haarcascade_frontalface_default.xml')
+        gray_image = cv2.cvtColor(img_np, cv2.COLOR_BGR2GRAY)
+        faces = face_cascade.detectMultiScale(gray_image, 1.3, 5)
+        for (x, y, w, h) in faces:
+            cv2.rectangle(img_np, (x, y), (x + w, y + h), (255, 0, 0), 2)
+        return img_np
     else:
         return img_np