Upload 3 files

services/face_match.py

@@ -0,0 +1,44 @@
+from sklearn.metrics.pairwise import cosine_similarity
+import numpy as np
+from sqlalchemy.orm import Session
+from users.models import UserEmbeddings
+
+class FaceMatch:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def load_embeddings_from_db(self):
+        user_embeddings = self.db.query(UserEmbeddings).all()
+        return {ue.user_id: np.array(ue.embeddings) for ue in user_embeddings}
+
+    def match_faces(self, new_embeddings, saved_embeddings, threshold=0.6):
+        new_embeddings = np.array(new_embeddings)
+        max_similarity = 0
+        identity = None
+
+        for user_id, stored_embeddings in saved_embeddings.items():
+            similarity = cosine_similarity(new_embeddings.reshape(1, -1), stored_embeddings.reshape(1, -1))[0][0]
+            if similarity > max_similarity:
+                max_similarity = similarity
+                identity = user_id
+
+        return identity, max_similarity if max_similarity > threshold else (None, 0)
+
+    def new_face_matching(self, new_embeddings):
+        embeddings_dict = self.load_embeddings_from_db()
+        if not embeddings_dict:
+            return {'status': 'Error', 'message': 'No embeddings available'}
+
+        identity, similarity = self.match_faces(new_embeddings, embeddings_dict)
+        if identity:
+            return {
+                'status': 'Success',
+                'message': 'Match Found',
+                'user_id': identity,
+                'similarity': similarity
+            }
+        return {
+            'status': 'Error',
+            'message': 'No matching face found'
+        }
+

services/facial_processing.py

@@ -0,0 +1,67 @@
+import numpy as np
+import os
+import torch
+from facenet_pytorch import MTCNN, InceptionResnetV1
+import cv2
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class FacialProcessing:
+    def __init__(self):
+        self.face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')
+        self.model = cv2.dnn.readNetFromTorch('openface.nn4.small2.v1.t7')
+        # Set the cache directory to a writable location
+        os.environ['TORCH_HOME'] = '/tmp/.cache/torch'
+
+        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+
+        self.mtcnn = MTCNN(keep_all=True, device=device)
+        self.resnet = InceptionResnetV1(pretrained='vggface2').eval().to(device)
+
+
+    def extract_embeddings(self, image_path):
+        try:
+            image = cv2.imread(image_path)
+            if image is None:
+                logger.error(f"Failed to load image: {image_path}")
+                return None
+            
+            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+            faces = self.face_cascade.detectMultiScale(gray, 1.3, 5)
+            
+            if len(faces) == 0:
+                logger.warning(f"No face detected in image: {image_path}")
+                return None
+            
+            (x, y, w, h) = faces[0]
+            face = image[y:y+h, x:x+w]
+            
+            faceBlob = cv2.dnn.blobFromImage(face, 1.0 / 255, (96, 96), (0, 0, 0), swapRB=True, crop=False)
+            self.model.setInput(faceBlob)
+            vec = self.model.forward()
+            
+            return vec.flatten().tolist()
+        except Exception as e:
+            logger.error(f"An error occurred while extracting embeddings: {e}")
+            return None
+
+
+    def extract_embeddings_vgg(self, image):
+        try:
+            # Preprocess the image
+            preprocessed_image = self.mtcnn(image)
+
+            if preprocessed_image is None:
+                logger.warning(f"No face detected in image")
+                return None
+            
+            # Extract the face embeddings
+            embeddings = self.resnet(preprocessed_image.unsqueeze(0)).detach().cpu().numpy().tolist()
+            if embeddings:
+                return embeddings[0]
+            
+        except Exception as e:
+            logger.error(f"An error occurred while extracting embeddings: {e}")
+            return None

services/recommendation_service.py

@@ -0,0 +1,97 @@
+# recommender_system.py
+import pandas as pd
+from surprise import Dataset, Reader, SVD
+from surprise.model_selection import train_test_split
+import joblib
+from datetime import datetime, timedelta
+import os
+import random
+from sqlalchemy.orm import Session
+from typing import List
+from orders.models import Order, Meal
+from users.models import User
+
+
+class MealRecommender:
+    def __init__(self, db: Session):
+        self.db = db
+        self.model_path = 'recommendation_model.joblib'
+        self.last_train_path = 'last_train_time.txt'
+        self.retrain_interval = timedelta(days=1)
+        self.algo = self.load_or_train_model()
+
+    def fetch_data(self):
+        orders = self.db.query(Order).all()
+        return pd.DataFrame([(order.user_id, order.meal_id, order.quantity) for order in orders], 
+                            columns=['user_id', 'meal_id', 'quantity'])
+
+    def train_model(self):
+        data = self.fetch_data()
+        if data.empty:
+            self.algo = None
+            return None
+        
+        reader = Reader(rating_scale=(1, 5))
+        dataset = Dataset.load_from_df(data[['user_id', 'meal_id', 'quantity']], reader)
+    
+        trainset = dataset.build_full_trainset()
+        algo = SVD()
+        algo.fit(trainset)
+        
+        joblib.dump(algo, self.model_path)
+        self._update_last_train_time()
+        return algo
+
+    def load_or_train_model(self):
+        try:
+            if self._should_retrain():
+                return self.train_model()
+            return joblib.load(self.model_path)
+        except FileNotFoundError:
+            return self.train_model()
+
+    def _should_retrain(self):
+        if not os.path.exists(self.last_train_path):
+            return True
+        with open(self.last_train_path, 'r') as f:
+            last_train_time = datetime.fromisoformat(f.read().strip())
+        return datetime.now() - last_train_time > self.retrain_interval
+
+    def _update_last_train_time(self):
+        with open(self.last_train_path, 'w') as f:
+            f.write(datetime.now().isoformat())
+
+    def get_recommendations(self, user: User):
+        if self._should_retrain():
+            self.algo = self.train_model()
+
+        if self.algo is None:
+            return self.get_random_recommendations()
+
+        all_meals = self.db.query(Meal).all()
+        meal_ids = [meal.id for meal in all_meals]
+
+        predictions = [self.algo.predict(str(user.id), str(meal_id)) for meal_id in meal_ids]
+        sorted_predictions = sorted(predictions, key=lambda x: x.est, reverse=True)
+        top_recommendations = self.db.query(Meal).filter(Meal.id.in_([int(pred.iid) for pred in sorted_predictions[:20]])).all()
+
+        top_recommendations = self.adjust_for_preferences(user, top_recommendations)
+
+        return top_recommendations[:5]  # Return top 5 recommendations
+
+    def adjust_for_preferences(self, user: User, recommendations: List[Meal]) -> List[Meal]:
+        preferences = user.preferences if user.preferences else []
+        preference_scores = {meal.id: 0 for meal in recommendations}
+
+        for meal in recommendations:
+            for preferred in preferences:
+                if preferred.lower() in meal.name.lower() or preferred.lower() in meal.description.lower():
+                    preference_scores[meal.id] += 1
+
+        sorted_recommendations = sorted(recommendations, key=lambda meal: preference_scores[meal.id], reverse=True)
+
+        return sorted_recommendations
+
+    def get_random_recommendations(self):
+        all_meals = self.db.query(Meal).all()
+        return random.sample(all_meals, min(5, len(all_meals)))