Shop-Sense: A Shopping Guide for Visually Impaired People

from flask import Flask, request, jsonify
import firebase_admin
from firebase_admin import credentials, firestore
import logging
import deepspeech
import wave
import numpy as np
import os
import base64
import datetime

logging.basicConfig(level=logging.DEBUG)

app = Flask(__name__)
image_data = None

cred = credentials.Certificate("your certificate json file")
firebase_admin.initialize_app(cred)
db = firestore.client()

model_path = 'FlaskServer\deepspeech-0.9.3-models.pbmm'
scorer_path = 'FlaskServer\deepspeech-0.9.3-models.scorer'
model = deepspeech.Model(model_path)
model.enableExternalScorer(scorer_path)

@app.route('/notehubWebhook', methods=['POST'])
def notehub_webhook():
    logging.debug('Received POST request')
    raw_data = request.data
    logging.debug(f'Raw data: {raw_data}')
    
    try:
        data = request.get_json()
        logging.debug(f'Received data: {data}')
        if data is None:
            raise ValueError("No JSON data received")
        data['creationTimestamp'] = datetime.utcnow()
        db.collection('notehub-data').add(data)
        return jsonify({"status": "success", "message": "Data received and saved"}), 200
    except Exception as e:
        logging.error(f'Error processing request: {e}')
        return jsonify({"status": "error", "message": str(e)}), 500


@app.route('/latestData', methods=['GET'])
def get_latest_data():
    try:
        docs = db.collection('notehub-data').order_by('timestamp', direction=firestore.Query.DESCENDING).limit(1).stream()
        latest_data = None
        for doc in docs:
            latest_data = doc.to_dict()
        
        if latest_data:
            return jsonify(latest_data), 200
        else:
            return jsonify({"error": "No data found"}), 404
    except Exception as e:
        logging.error(f'Error retrieving data: {e}')
        return jsonify({"error": str(e)}), 500

@app.route('/speechtoText', methods=['POST'])
def transcribe_audio():
    if 'audio/wav' in request.headers['Content-Type']:
        audio_data = request.data
        with open("received_audio.wav", "wb") as f:
            f.write(audio_data)
        
        with wave.open("received_audio.wav", "rb") as wf:
            audio = np.frombuffer(wf.readframes(wf.getnframes()), np.int16)
            text = model.stt(audio)
        
        os.remove("received_audio.wav")
        return text
    else:
        return "Unsupported media type", 415

@app.route('/upload', methods=['POST', 'GET'])
def upload_image():
    global image_data
    if request.method == 'POST':
        try:
            data = request.get_json()
            if 'image' in data:
                image_data = base64.b64decode(data['image'])
                with open("received_image.jpg", "wb") as f:
                    f.write(image_data)
                return jsonify({"status": "Image received"}), 200
            else:
                return jsonify({"error": "Invalid data"}), 400
        except Exception as e:
            return jsonify({"error": str(e)}), 500
    else:
        return "<h1>Upload endpoint for POST requests only</h1>"
    
model = tf.keras.models.load_model('path_to_your_model.h5')

def recognize_currency(image):
    image = image.convert('RGB') 
    image = np.array(image)    
    image = np.expand_dims(image, axis=0) 
    image = image / 255.0  

    predictions = model.predict(image)

    denomination = np.argmax(predictions)  
    return denomination

@app.route('/currencyIdentification', methods=['POST'])
def currency_identification():
    data = request.json
    if 'image' not in data:
        return jsonify({'error': 'No image found in request'}), 400

    image_base64 = data['image']
    try:
        image_data = base64.b64decode(image_base64)
        image = Image.open(io.BytesIO(image_data))
        denomination = recognize_currency(image)
        return jsonify({'denomination': denomination}), 200

    except Exception as e:
        print(f'Error: {e}')
        return jsonify({'error': 'Error processing image'}), 500

@app.route('/image', methods=['GET'])
def get_image():
    if image_data:
        return base64.b64encode(image_data).decode('utf-8'), 200
    else:
        return jsonify({"error": "No image available"}), 404

@app.route('/')
def home():
    return "Hello, Flask!"

if __name__ == '__main__':
    app.run(debug=True, port=5000)

import tensorflow as tf
from tensorflow.keras.applications import VGG16
from tensorflow.keras.applications.vgg16 import preprocess_input
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Lambda, Dense, Dropout
from tensorflow.keras import backend as K
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.layers import Layer
from collections import defaultdict
from os.path import join
import numpy as np
from PIL import Image
import os

path_train = 'your dataset path'
batch_size = 2
embedding_dim = 512

def l2_normalize(x):
    return x / np.sqrt(np.sum(x**2, axis=-1, keepdims=True))

def read_and_resize(filepath, retries =3):
    try:
        im = Image.open(filepath).convert('RGB')
        im = im.resize((224, 224))
        return np.array(im, dtype="float32")
    except Exception as e:
        print(f"Error reading {filepath}: {e}")
        if retries > 0 :
            print(f"Error Reading {filepath}: {e}. Retrying {retries} more")
            return read_and_resize(filepath, retries-1)
        print("Giving Up")
        return None

def augment(im_array):
    if np.random.uniform(0, 1) > 0.9:
        im_array = np.fliplr(im_array)
    return im_array

class sample_gen(object):
    def __init__(self, file_class_mapping, other_class="new"):
        self.file_class_mapping = file_class_mapping
        self.class_to_list_files = defaultdict(list)
        self.list_other_class = []
        self.list_all_files = list(file_class_mapping.keys())
        self.range_all_files = list(range(len(self.list_all_files)))

        for file, class_ in file_class_mapping.items():
            if class_ == other_class:
                self.list_other_class.append(file)
            else:
                self.class_to_list_files[class_].append(file)

        self.list_classes = list(set(self.file_class_mapping.values()))
        self.range_list_classes = range(len(self.list_classes))
        self.class_weight = np.array([len(self.class_to_list_files[class_]) for class_ in self.list_classes])
        self.class_weight = self.class_weight / np.sum(self.class_weight)

    def get_sample(self):
        class_idx = np.random.choice(self.range_list_classes, 1, p=self.class_weight)[0]
        examples_class_idx = np.random.choice(range(len(self.class_to_list_files[self.list_classes[class_idx]])), 2)
        positive_example_1, positive_example_2 = \
            self.class_to_list_files[self.list_classes[class_idx]][examples_class_idx[0]], \
            self.class_to_list_files[self.list_classes[class_idx]][examples_class_idx[1]]

        negative_example = None
        while negative_example is None or self.file_class_mapping[negative_example] == \
                self.file_class_mapping[positive_example_1]:
            negative_example_idx = np.random.choice(self.range_all_files, 1)[0]
            negative_example = self.list_all_files[negative_example_idx]
        return positive_example_1, negative_example, positive_example_2

def gen(triplet_gen):
    while True:
        list_positive_examples_1 = []
        list_negative_examples = []
        list_positive_examples_2 = []

        for i in range(batch_size):
            positive_example_1, negative_example, positive_example_2 = triplet_gen.get_sample()
            path_pos1 = join(path_train, positive_example_1)
            path_neg = join(path_train, negative_example)
            path_pos2 = join(path_train, positive_example_2)
            
            positive_example_1_img = read_and_resize(path_pos1)
            negative_example_img = read_and_resize(path_neg)
            positive_example_2_img = read_and_resize(path_pos2)

            if positive_example_1_img is None or negative_example_img is None or positive_example_2_img is None:
                print(f"Skipping batch due to None image: {positive_example_1}, {negative_example}, {positive_example_2}")
                continue

            positive_example_1_img = augment(positive_example_1_img)
            negative_example_img = augment(negative_example_img)
            positive_example_2_img = augment(positive_example_2_img)
            
            list_positive_examples_1.append(positive_example_1_img)
            list_negative_examples.append(negative_example_img)
            list_positive_examples_2.append(positive_example_2_img)

        A = preprocess_input(np.array(list_positive_examples_1))
        B = preprocess_input(np.array(list_positive_examples_2))
        C = preprocess_input(np.array(list_negative_examples))
        
        label = np.zeros((batch_size,))
        
        print(f"Yielding batch - A: {A.shape}, B: {B.shape}, C: {C.shape}")
        yield {'anchor_input': A, 'positive_input': B, 'negative_input': C}, label

class MACPooling(Layer):
    def __init__(self, **kwargs):
        super(MACPooling, self).__init__(**kwargs)

    def call(self, inputs):
        return tf.reduce_max(inputs, axis=[1, 2])

class TripletLossLayer(Layer):
    def __init__(self, margin=1.0, **kwargs):
        self.margin = margin
        super(TripletLossLayer, self).__init__(**kwargs)

    def call(self, inputs):
        anchor, positive, negative = inputs
        pos_dist = tf.reduce_sum(tf.square(anchor - positive), axis=-1)
        neg_dist = tf.reduce_sum(tf.square(anchor - negative), axis=-1)
        loss = tf.maximum(pos_dist - neg_dist + self.margin, 0.0)
        self.add_loss(tf.reduce_mean(loss))
        return loss

def GetModel(image_size, embedding_dim):
    base_model = VGG16(weights='imagenet', include_top=False, input_shape=(image_size, image_size, 3))
    base_model_output = base_model.get_layer('block4_pool').output

    x = MACPooling()(base_model_output)
    x = Dropout(0.6)(x)
    x = Dense(embedding_dim)(x)
    x = Lambda(lambda x: tf.math.l2_normalize(x, axis=1))(x)

    embedding_model = Model(base_model.input, x, name="embedding")
    input_shape = (image_size, image_size, 3)
    anchor_input = Input(input_shape, name='anchor_input')
    positive_input = Input(input_shape, name='positive_input')
    negative_input = Input(input_shape, name='negative_input')
    
    anchor_embedding = embedding_model(anchor_input)
    positive_embedding = embedding_model(positive_input)
    negative_embedding = embedding_model(negative_input)

    triplet_loss_layer = TripletLossLayer(name='triplet_loss_layer')([anchor_embedding, positive_embedding, negative_embedding])
    
    triplet_model = Model(inputs=[anchor_input, positive_input, negative_input], outputs=[anchor_embedding, positive_embedding, negative_embedding])

    return embedding_model, triplet_model

# file_class_mapping = {
#     'Image1.jpeg': 'class1',
#     'Image2.jpeg': 'class1',
#     'Image3.jpg': 'class2',
#     'Image4.jpg': 'class2',
# }

triplet_gen = sample_gen(file_class_mapping)
train_generator = gen(triplet_gen)

steps_per_epoch = len(file_class_mapping) // batch_size
epochs = 10
def dummy_loss(y_true, y_pred):
    return tf.reduce_mean(y_pred)


embedding_model, triplet_model = GetModel(image_size=224, embedding_dim=embedding_dim)
triplet_model.compile(loss = dummy_loss , optimizer=Adam(learning_rate=0.0001))

for i in range(1):
    data = next(train_generator)
    print(f"Generator output - {data}")
dummy_target_data = np.zeros(batch_size)
# Train the model
history = triplet_model.fit(train_generator, steps_per_epoch=steps_per_epoch, epochs=epochs, 
                            validation_data=None, 
                            verbose = 1, 
                            callbacks=None, 
                            # Supplying the dummy labels here
                            # y=np.zeros((steps_per_epoch,)),  
                            class_weight=None,
                            max_queue_size=10, 
                            workers=1, 
                            use_multiprocessing=False, 
                            shuffle=True, 
                            initial_epoch=0)