phase1 testing - employees Dataloader

app.py

@@ -31,29 +31,152 @@ from glob import glob
 import shutil
 import torch.nn.functional as F
 
+# Set the device (GPU or CPU)
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
 # Initialse Globle Variables
 MODEL_TRANSFORMER = 'google/vit-base-patch16-224'
 BATCH_SIZE = 8
 
-# Set the device (GPU or CPU)
-device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+# Set Paths
+data_path = 'employees'
+model_path = 'vit_pytorch_GPU_1.pt'
+
+# Set Title
+st.title("Employee Attendance System")
+#pipeline = pipeline(task="image-classification", model="julien-c/hotdog-not-hotdog")
+
+# Define Image Processor
+image_processor_prod = ViTImageProcessor.from_pretrained(MODEL_TRANSFORMER, attn_implementation="sdpa", torch_dtype=torch.float16)
+
+# Define ML Model
+class FaceEmbeddingModel(torch.nn.Module):
+    def __init__(self, model_name, embedding_size):
+        super(FaceEmbeddingModel, self).__init__()
+        self.config = ViTConfig.from_pretrained(model_name, id2label=idx_to_label, label2id=label_to_idx, return_dict=True)
+        self.backbone = ViTModel.from_pretrained(model_name, config=self.config)  # Load ViT model
+        self.fc = torch.nn.Linear(self.backbone.config.hidden_size, embedding_size) # Convert to 512D feature vector
+
+    def forward(self, images):
+        x = self.backbone(images).last_hidden_state[:, 0]  # Extract embeddings
+        x = self.fc(x)  # Convert to 512D embedding
+        return torch.nn.functional.normalize(x)  # Normalize for cosine similarity
+
+# Load the model
+model_pretrained = torch.load(model_path, map_location=device)
+
+# Define the ML model - Evaluation function
+def prod_function(transformer_model, prod_dl, prod_data):
+    # Initialize accelerator
+    accelerator = Accelerator()
+
+    # to INFO for the main process only.
+    if accelerator.is_main_process:
+        datasets.utils.logging.set_verbosity_warning()
+        transformers.utils.logging.set_verbosity_info()
+    else:
+        datasets.utils.logging.set_verbosity_error()
+        transformers.utils.logging.set_verbosity_error()
+
+    # The seed need to be set before we instantiate the model, as it will determine the random head.
+    set_seed(42)
+
+    # There is no specific order to remember, we just need to unpack the objects in the same order we gave them to the prepare method.
+    accelerated_model, acclerated_prod_dl, acclerated_prod_data = accelerator.prepare(transformer_model, prod_dl, prod_data)
+
+    # Evaluate at the end of the epoch
+    accelerated_model.eval()
+
+    # Find Embedding of the image to be evaluated
+    emb_prod = accelerated_model(acclerated_prod_data)
+
+    prod_preds = []
+
+    for batch in tqdm(acclerated_prod_dl):
+        with torch.no_grad():
+            emb = accelerated_model(**batch)
+        distance = F.pairwise_distance(emb, emb_prod)
+
+        prod_preds.append(distance)
+    return prod_preds
+
+# Creation of Dataloader
+class CustomDatasetProd(Dataset):
+    def __init__(self, pixel_values):
+        self.pixel_values = pixel_values
+
+    def __len__(self):
+        return len(self.pixel_values)
+
+    def __getitem__(self, idx):
+        item = {
+            'pixel_values': self.pixel_values[idx].squeeze(0),
+        }
+        return item
+        
+# Creation of Dataset       
+class CreateDatasetProd():
+    def __init__(self, image_processor):
+        super().__init__()
+        self.image_processor = image_processor
+        # Define a transformation pipeline
+        self.transform_prod = transforms.v2.Compose([
+                                    transforms.v2.ToImage(),
+                                    transforms.v2.ToDtype(torch.uint8, scale=False)
+                                ])
+
+    def get_pixels(self, img_paths):
+        pixel_values = []
+        for path in tqdm(img_paths):
+            # Read and process Images
+            img = PIL.Image.open(path)
+            img = self.transform_prod(img)
+
+            # Scaling the video to ML model's desired format
+            img = self.image_processor(img, return_tensors='pt') #, input_data_format='channels_first')
+
+            pixel_values.append(img['pixel_values'].squeeze(0))
 
-pipeline = pipeline(task="image-classification", model="julien-c/hotdog-not-hotdog")
+            # Force garbage collection
+            del img
+            gc.collect()
+        return pixel_values
 
-st.title("Hot Dog? Or Not?")
+    def create_dataset(self, image_paths):
 
+        pixel_values = torch.stack(self.get_pixels(image_paths))
+        return CustomDatasetProd(pixel_values=pixel_values)
+        
 # Read images from directory
+image_paths = []
+image_file = glob(os.path.join(data_path, '*.jpg'))
+#st.write(image_file)
+image_paths.extend(image_file)
+st.write('input path size:', len(image_paths))
+st.write(image_paths)
 
+# Create DataLoader for Employees image
+dataset_prod_obj = CreateDatasetProd(image_processor_prod)
+prod_ds = dataset_prod_obj.create_dataset(image_paths)
+prod_dl = DataLoader(prod_ds, batch_size=BATCH_SIZE)
 
-# Read image from Camera
-enable = st.checkbox("Enable camera")
-picture = st.camera_input("Take a picture", disabled=not enable)
-if picture:
-    col1, col2 = st.columns(2)
-    image = PIL.Image.open(picture)
-    col1.image(image, use_column_width=True)
-    predictions = pipeline(image)
+# Testing the dataloader
+prod_inputs = next(iter(prod_dl))
+print(prod_inputs['pixel_values'].shape) 
 
-    col2.header("Probabilities")
-    for p in predictions:
-        col2.subheader(f"{ p['label'] }: { round(p['score'] * 100, 1)}%")
+## Read image from Camera
+#enable = st.checkbox("Enable camera")
+#picture_path = st.camera_input("Take a picture", disabled=not enable)
+#if picture_path:
+#    # Create DataLoader for Webcam Image
+#    webcam_ds = dataset_prod_obj.create_dataset(picture_path)
+#    webcam_dl = DataLoader(webcam_ds, batch_size=BATCH_SIZE)
+#
+#
+#prediction = prod_function(model_pretrained, prod_dl, webcam_dl)
+#predictions = torch.cat(prediction, 0).to('cpu')
+#match_idx = torch.argmin(predictions)
+#if predictions[match_idx] <= 0.3:
+#  st.write('Welcome: ',image_paths[match_idx].split('/')[-1].split('.')[0])
+#else:
+#  st.write("Match not found")