Upload 12 files

LicensePlateAuthorizer.py

@@ -0,0 +1,150 @@
+import queue
+import sqlite3
+import threading
+from typing import Dict
+
+import cv2
+import easyocr
+import torch.nn as nn
+from LicensePlateDetector import LicensePlateDetector
+from LicensePlateReader import LicensePlateReader
+from LicensePlateRecognizer import LicensePlateRecognizer
+from ultralytics import YOLO
+
+
+class LicensePlateAuthorizer(nn.Module):
+    def __init__(
+        self,
+        detector_model: str = "./models/license_plate_192.pt",
+        database_path: str = None,
+        confidence_threshold: int = 5,
+        queue_size: int = 10,
+    ):
+        super(LicensePlateAuthorizer, self).__init__()
+
+        detector = LicensePlateDetector(YOLO(detector_model))
+        reader = LicensePlateReader(easyocr.Reader(["en"], gpu=True))
+        self.recognizer = LicensePlateRecognizer(detector, reader)
+
+        self.database_path = database_path
+        self.confidence_threshold = confidence_threshold
+        self.ocr_result_counts: Dict[str, int] = {}
+        self.current_license_plate_text = None
+
+        self.input_queue = queue.Queue(maxsize=queue_size)
+        self.output_queue = queue.Queue(maxsize=queue_size)
+        self.processing_thread = threading.Thread(
+            target=self._process_queue, daemon=True
+        )
+        self.processing_thread.start()
+
+    def forward(self, frame):
+        self.input_queue.put(frame)
+        result = self.output_queue.get()
+        if result is None:
+            return None
+        return result[0]
+
+    def _process_queue(self):
+        while True:
+            frame = self.input_queue.get()
+            try:
+                result = self._process_single_frame(frame)
+                self.output_queue.put(result)
+            except Exception as e:
+                print(f"Error in LicensePlateAuthorizer: {e}")
+                self.output_queue.put(None)
+            finally:
+                self.input_queue.task_done()
+
+    def _process_single_frame(self, frame):
+        license_plate_text, bbox = self.recognizer.forward(frame)
+
+        if license_plate_text is None:
+            return frame, None, None, False
+
+        authorized = self.verify_license_plate_authorization(license_plate_text)
+        frame = self.annotate_frame(frame, bbox, authorized)
+        return frame, license_plate_text, bbox, authorized
+
+    def verify_license_plate_authorization(self, license_plate_text: str) -> bool:
+        if license_plate_text == "":
+            return False
+
+        if self.current_license_plate_text != license_plate_text:
+            self.current_license_plate_text = license_plate_text
+            self.ocr_result_counts = {}
+
+        self.ocr_result_counts[license_plate_text] = (
+            self.ocr_result_counts.get(license_plate_text, 0) + 1
+        )
+
+        if self.ocr_result_counts[license_plate_text] >= self.confidence_threshold:
+            if self.database_path is not None:
+                authorized = self.check_authorization_from_database(license_plate_text)
+                return authorized
+        return False
+
+    def check_authorization_from_database(self, license_plate: str) -> bool:
+        try:
+            conn = sqlite3.connect(self.database_path)
+            cursor = conn.cursor()
+
+            query = "SELECT EXISTS(SELECT 1 FROM license_plates WHERE plate = ?)"
+            cursor.execute(query, (license_plate,))
+            exists = cursor.fetchone()[0]
+
+            conn.close()
+            return bool(exists)
+        except Exception as e:
+            print(f"Database error: {e}")
+            return False
+
+    def annotate_frame(self, frame, bbox, authorized):
+        if bbox is not None:
+            color = (0, 255, 0) if authorized else (0, 0, 255)
+            cv2.rectangle(frame, (bbox[0], bbox[1]), (bbox[2], bbox[3]), color, 2)
+            status = "Authorized" if authorized else "Unauthorized"
+            cv2.putText(
+                frame,
+                f"Status: {status}",
+                (bbox[0], bbox[1] - 10),  # Top right above the bounding box
+                cv2.FONT_HERSHEY_SIMPLEX,
+                0.9,
+                color,
+                2,
+            )
+        return frame
+
+    def save_license_plate(self, license_plate: str) -> None:
+        conn = sqlite3.connect(self.database_path)
+        cursor = conn.cursor()
+
+        cursor.execute(
+            """
+        CREATE TABLE IF NOT EXISTS license_plates (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            plate TEXT UNIQUE
+        )"""
+        )
+
+        try:
+            cursor.execute(
+                "INSERT INTO license_plates (plate) VALUES (?)", (license_plate,)
+            )
+            conn.commit()
+            print(f"License plate '{license_plate}' added to database.")
+        except sqlite3.IntegrityError:
+            print(
+                f"License plate '{license_plate}' already exists in the database, skipping."
+            )
+
+        conn.close()
+
+    def clear_database(self) -> None:
+        conn = sqlite3.connect(self.database_path)
+        cursor = conn.cursor()
+
+        cursor.execute("DELETE FROM license_plates")
+        conn.commit()
+        conn.close()

LicensePlateDetector.py

@@ -0,0 +1,127 @@
+from ultralytics import YOLO
+import torch.nn as nn
+import torch
+import numpy as np
+import queue
+import threading
+import cv2
+
+
+class LicensePlateDetector(nn.Module):
+    def __init__(
+        self,
+        model: YOLO | nn.Module | object,
+        img_size: list = [192, 192],
+        queue_size: int = 10,
+    ):
+        super(LicensePlateDetector, self).__init__()
+        self.model = model
+
+        if torch.backends.mps.is_available():
+            mps_device = torch.device("mps")
+            self.model.to(mps_device)
+            
+        self.img_size = img_size
+        self.input_queue = queue.Queue(maxsize=queue_size)
+        self.output_queue = queue.Queue(maxsize=queue_size)
+        self.processing_thread = threading.Thread(
+            target=self._process_queue, daemon=True
+        )
+        self.processing_thread.start()
+
+    def forward(self, input):
+        self.input_queue.put(input)
+        return self.output_queue.get()
+
+    def _process_queue(self):
+        while True:
+            input_frame = self.input_queue.get()
+            try:
+                result = self._process_single_frame(input_frame)
+                self.output_queue.put(result)
+            except Exception as e:
+                self.output_queue.put(None)  # Indicate error
+                print(f"Error in LicensePlateDetector: {e}")
+            finally:
+                self.input_queue.task_done()
+
+    def _process_single_frame(self, input):
+        detected_license_plates: list = self.predict(input)
+        left_license_plate_half, right_license_plate_half, bbox = self.post_process(
+            detected_license_plates
+        )
+        # self.annotate_frame(input, bbox)
+        return left_license_plate_half, right_license_plate_half, bbox
+
+    def predict(self, frame):
+        license_plate_detections: list = self.model.predict(frame, imgsz=self.img_size)
+        return license_plate_detections
+
+    def post_process(self, detections):
+        UPPER_LEFT_Y, UPPER_LEFT_X, BOTTOM_RIGHT_Y, BOTTOM_RIGHT_X = 1, 0, 3, 2
+
+        for detection in detections:
+            if len(detection.boxes.xyxy) == 0:
+                continue
+            bbox = (
+                detection.boxes.xyxy[0].cpu()
+                if detection.boxes.xyxy[0] is not None
+                else np.empty((0, 4))
+            )
+            bbox = bbox.numpy().astype(int)
+
+            DISTANCE_X = np.abs(bbox[UPPER_LEFT_X] - bbox[BOTTOM_RIGHT_X])
+            RIGHT_SHIFT_PERCENTAGE = np.round(DISTANCE_X * 0.15).astype(int)
+            LEFT_SHIFT_PERCENTAGE = np.round(DISTANCE_X * 0.01).astype(int)
+
+            midpoint_y = (bbox[UPPER_LEFT_Y] + bbox[BOTTOM_RIGHT_Y]) // 2
+            midpoint_x = (bbox[UPPER_LEFT_X] + bbox[BOTTOM_RIGHT_X]) // 2
+
+            left_half = detection.orig_img[
+                midpoint_y : bbox[BOTTOM_RIGHT_Y],
+                bbox[UPPER_LEFT_X] : midpoint_x - LEFT_SHIFT_PERCENTAGE,
+            ]
+            right_half = detection.orig_img[
+                midpoint_y : bbox[BOTTOM_RIGHT_Y],
+                midpoint_x : bbox[BOTTOM_RIGHT_X] - RIGHT_SHIFT_PERCENTAGE,
+            ]
+
+            return left_half, right_half, bbox
+
+        raise Exception("No License Plate")
+
+    def annotate_frame(self, frame, bbox):
+        if bbox is not None:
+            cv2.rectangle(
+                frame, (bbox[0], bbox[1]), (bbox[2], bbox[3]), (0, 0, 255), 2
+            )  # Red bounding box
+        return frame
+
+
+## Training, Validation, and Testing may be like this:
+
+# transform = transforms.Compose([
+#     transforms.Resize((192, 192)),
+#     transforms.ToTensor()
+# ])
+
+# train_dataset = LicensePlateDetectorDataset(root_dir, subset='train', transform=transform)
+# valid_dataset = LicensePlateDetectorDataset(root_dir, subset='valid', transform=transform)
+# test_dataset = LicensePlateDetectorDataset(root_dir, subset='test', transform=transform)
+
+# train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
+# valid_loader = DataLoader(valid_dataset, batch_size=32, shuffle=False)
+# test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)
+
+# yolo_model_path = 'path/to/yolo_model.pt'
+# yolo_model = YOLO(yolo_model_path)
+
+# detector = LicensePlateDetector(yolo_model)
+
+# detector.train(data_path=data_path, num_epochs=100)
+
+# # Evaluate on validation set
+# detector.validate(detector, valid_loader)
+
+# # Evaluate on test set
+# detector.test(detector, test_loader)

LicensePlateDetectorDataset.py

@@ -0,0 +1,170 @@
+from PIL import Image
+import os
+from torchvision import transforms
+from torch.utils.data import Dataset
+import os
+import cv2
+import numpy as np
+import albumentations as A
+from tqdm import tqdm  # tqdm provides a nice progress bar
+
+# TODO: Load Files From Google Drive
+
+
+
+class LicensePlateDetectorDataset(Dataset):
+    ZOOM_AUGMENTATION = "zoom"
+    GRAYSCALE_AUGMENTATION = "grayscale"
+    FOG_AUGMENTATION = "foggy"
+    RAIN_AUGMENTATION = "rainy"
+    SALT_PEPPER_AUGMENTATION = "saltandpepper"
+    BLUR_AUGMENTATION = "blur"
+
+    def __init__(self, root_dir, subset: str = 'train', transform: transforms.Compose = None, augmentations = {
+            "zoom": [0.5],
+            "grayscale": [],
+            "foggy": [],
+            "rainy": [],
+            "saltandpepper": [0.01, 0.01],
+            "blur": [5],
+        }):
+        self.root_dir = os.path.join(root_dir, subset)
+        self.folder_subset = subset
+        self.transform = transform
+        self.image_paths = []
+        self.labels = []
+
+        self.augmentations_functions = {
+            LicensePlateDetectorDataset.ZOOM_AUGMENTATION: self.apply_zoom,
+            LicensePlateDetectorDataset.GRAYSCALE_AUGMENTATION: self.apply_grayscale,
+            LicensePlateDetectorDataset.FOG_AUGMENTATION: self.apply_fog,
+            LicensePlateDetectorDataset.RAIN_AUGMENTATION: self.apply_rain,
+            LicensePlateDetectorDataset.SALT_PEPPER_AUGMENTATION: self.apply_salt_and_pepper,
+            LicensePlateDetectorDataset.BLUR_AUGMENTATION: self.apply_gaussian_blur,
+        }
+
+        self.augmentations = augmentations
+
+        image_dir = os.path.join(self.root_dir, 'images')
+        label_dir = os.path.join(self.root_dir, 'labels')
+
+        # Count total number of images/files
+        total_files = sum(1 for file in os.listdir(image_dir) if file.endswith(('.png', '.jpg', '.jpeg')))
+
+        progress_bar = tqdm(total=total_files, desc='Initializing Files', unit='image')
+
+        for img_file in os.listdir(image_dir):
+            if img_file.endswith(('.png', '.jpg', '.jpeg')):
+                self.image_paths.append(os.path.join(image_dir, img_file))
+                progress_bar.update(1)  # Update progress bar
+
+                print("Image:", img_file)
+
+                label_file = os.path.join(label_dir, img_file.replace('.jpg', '.txt'))
+                self.labels.append(label_file)
+                print("Label:", label_file)
+
+        progress_bar.close()
+
+    def __len__(self):
+        return len(self.image_paths)
+
+    def __getitem__(self, idx):
+        img_path = self.image_paths[idx]
+        image = Image.open(img_path).convert('RGB')
+        label_path = self.labels[idx]
+
+        if self.transform:
+            image = self.transform(image)
+
+        with open(label_path, 'r') as f:
+            label = f.readline().strip()
+
+        return image, label
+    
+    def process_images(self, output_folder: str, images_limit=-1):
+        """Apply a given processing function to all images in the input folder and save the results to the output folder."""
+        os.makedirs(output_folder, exist_ok=True)
+        count = 0  # Initialize a counter for iterations
+        
+        # Count total number of images/files
+        total_files = sum(1 for file in os.listdir(os.path.join(self.root_dir, "images")) if file.endswith(('.png', '.jpg', '.jpeg'))) if images_limit == -1 else images_limit
+
+        progress_bar = tqdm(total=total_files, desc='Applying Augmentation to Images', unit='image')
+        
+        for filename in os.listdir(os.path.join(self.root_dir, "images")):
+            if filename.endswith((".jpg", ".jpeg", ".png")):
+                image_path = os.path.join(self.root_dir, "images", filename)
+                image = cv2.imread(image_path)
+                for augmentation_name, augmentation_options in self.augmentations.items():
+                    processed_image = self.augmentations_functions[augmentation_name](image, *augmentation_options)
+                    augmentation_folder = os.path.join(output_folder, augmentation_name)
+                    os.makedirs(augmentation_folder, exist_ok=True)  # Ensure augmentation_folder exists
+                    output_path = os.path.join(augmentation_folder, filename)
+                    cv2.imwrite(output_path, processed_image)
+                
+                print("Image Processed with all augmentations: ", os.path.join(output_folder, augmentation_name, filename))
+                progress_bar.update(1)  # Update progress bar
+
+                count += 1  # Increment the counter
+                if count >= images_limit and images_limit != -1: # Dont enter the loop if limit is not specified (-1 means until finish)
+                    print("Reached Limit Specified", images_limit)
+                    break  # Exit the loop if the limit is reached
+
+        progress_bar.close()
+
+    def apply_zoom(self, image: np.ndarray, scale_factor: float) -> np.ndarray:
+        """Apply zoom to the image."""
+        height, width = image.shape[:2]
+        center = (width // 2, height // 2)
+        matrix = cv2.getRotationMatrix2D(center, 0, scale_factor)
+        zoomed_image = cv2.warpAffine(image, matrix, (width, height))
+        return zoomed_image
+
+    def apply_grayscale(self, image: np.ndarray) -> np.ndarray:
+        """Convert image to grayscale."""
+        return cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+
+    def apply_gaussian_blur(self, image: np.ndarray, kernel_size: int) -> np.ndarray:
+        """Apply Gaussian blur to the image."""
+        return cv2.GaussianBlur(image, (kernel_size, kernel_size), 0)
+
+    def apply_salt_and_pepper(self, image: np.ndarray, salt_prob: float, pepper_prob: float) -> np.ndarray:
+        """Apply salt and pepper noise to the image."""
+        noisy_image = np.copy(image)
+        row, col, _ = noisy_image.shape
+        salt_pixels = np.random.rand(row, col) < salt_prob
+        noisy_image[salt_pixels] = 255
+        pepper_pixels = np.random.rand(row, col) < pepper_prob
+        noisy_image[pepper_pixels] = 0
+        return noisy_image
+
+    def apply_fog(self, image: np.ndarray) -> np.ndarray:
+        """Apply fog effect to the image."""
+        transform = A.Compose([A.RandomFog(fog_coef_lower=0.3, fog_coef_upper=0.5, alpha_coef=0.3, p=1)])
+        transformed = transform(image=image)
+        return transformed['image']
+
+    def apply_rain(self, image: np.ndarray) -> np.ndarray:
+        """Apply rain effect to the image."""
+        transform = A.Compose([A.RandomRain(brightness_coefficient=0.9, drop_width=1, blur_value=5, p=1)])
+        transformed = transform(image=image)
+        return transformed['image']
+    
+    def auto_annotate(self):
+        pass
+## Example Usage:
+# transform = transforms.Compose([
+#     transforms.Resize((192, 192)),
+#     transforms.ToTensor()
+# ])
+# augmentations = {
+    # LicensePlateDetectorDataset.ZOOM_AUGMENTATION: [15],
+    # LicensePlateDetectorDataset.GRAYSCALE_AUGMENTATION: [],
+    # LicensePlateDetectorDataset.FOG_AUGMENTATION: [],
+    # LicensePlateDetectorDataset.RAIN_AUGMENTATION: [],
+    # LicensePlateDetectorDataset.SALT_PEPPER_AUGMENTATION: [0,0],
+    # LicensePlateDetectorDataset.BLUR_AUGMENTATION: [15],
+# }
+# lpdd = LicensePlateDetectorDataset("./License Plate YOLOv8/", subset="train", transform=transform, augmentations=augmentations)
+# lpdd.process_images("./License Plate YOLOv8/augmented_images", images_limit=20) # remove images_limit or make it -1 to loop over all images

LicensePlateReader.py

@@ -0,0 +1,185 @@
+import queue
+import re
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from typing import Dict, Optional
+
+import cv2
+import easyocr
+import numpy as np
+import torch
+import torch.nn as nn
+
+
+class LicensePlateReader(nn.Module):
+    def __init__(
+        self,
+        model,
+        char_to_num_mappings: Optional[Dict[str, str]] = None,
+        num_to_char_mappings: Optional[Dict[str, str]] = None,
+        confidence: float = 0.30,
+        queue_size: int = 10,
+    ):
+        """
+        Initialize the LicensePlateReader with the given model and mappings.
+        Args:.
+            model OCR model for reading text.
+            char_to_num_mappings Mappings from characters to numbers.
+            num_to_char_mappings Mappings from numbers to characters.
+            confidence threshold for accepting OCR results.
+        """
+        super(LicensePlateReader, self).__init__()
+
+        # Initializing
+        self.char_to_num_mappings = char_to_num_mappings or {
+            "L": "4",
+            "D": "0",
+            "S": "5",
+            "Z": "2",
+            "B": "8",
+            "C": "0",
+        }
+        self.num_to_char_mappings = num_to_char_mappings or {
+            "2": "Z",
+            "4": "A",
+            "6": "G",
+            "5": "S",
+            "0": "D",
+            "7": "T",
+            "8": "B",
+        }
+        self.model = model
+        self.confidence = confidence
+
+        self.input_queue = queue.Queue(maxsize=queue_size)
+        self.output_queue = queue.Queue(maxsize=queue_size)
+        self.executor = ThreadPoolExecutor(max_workers=2)
+        self.processing_thread = threading.Thread(
+            target=self._process_queue, daemon=True
+        )
+        self.processing_thread.start()
+
+    def forward(self, numbers_side: np.ndarray, letters_side: np.ndarray) -> str:
+        self.input_queue.put((numbers_side, letters_side))
+        return self.output_queue.get()
+
+    def _process_queue(self):
+        while True:
+            numbers_side, letters_side = self.input_queue.get()
+            result = self._process_single_plate(numbers_side, letters_side)
+            self.output_queue.put(result)
+            self.input_queue.task_done()
+
+    def _process_single_plate(
+        self, numbers_side: np.ndarray, letters_side: np.ndarray
+    ) -> str:
+        future_preprocessed_numbers = self.executor.submit(
+            self._pre_process, numbers_side
+        )
+        future_preprocessed_letters = self.executor.submit(
+            self._pre_process, letters_side
+        )
+
+        preprocessed_numbers_side = future_preprocessed_numbers.result()
+        preprocessed_letters_side = future_preprocessed_letters.result()
+
+        future_extracted_numbers = self.executor.submit(
+            self.predict, preprocessed_numbers_side
+        )
+        future_extracted_letters = self.executor.submit(
+            self.predict, preprocessed_letters_side
+        )
+
+        extracted_numbers_side = future_extracted_numbers.result()
+        extracted_letters_side = future_extracted_letters.result()
+
+        future_postprocessed_numbers = self.executor.submit(
+            self._post_process, extracted_numbers_side, True
+        )
+        future_postprocessed_letters = self.executor.submit(
+            self._post_process, extracted_letters_side, False
+        )
+
+        postprocessed_numbers_side = future_postprocessed_numbers.result()
+        postprocessed_letters_side = future_postprocessed_letters.result()
+
+        return postprocessed_numbers_side + "" + postprocessed_letters_side
+
+    def _pre_process(self, frame: np.ndarray) -> np.ndarray:
+        """
+        Preprocess the input frame by blurring, grayscaling, and thresholding.
+        Args:
+            frame Input image frame.
+        Returns:
+            Preprocessed binary image.
+        """
+        # Blurring
+        # blurred_frame = cv2.GaussianBlur(frame, (3, 3), 0)
+
+        # greyscaleing
+        greyscaled_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+
+        # thresholding white and black
+        _, binary_frame = cv2.threshold(
+            greyscaled_frame, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU
+        )
+        return binary_frame
+
+    def predict(self, frame: np.ndarray) -> str:
+        """
+        Predict text from the preprocessed frame using the OCR model.
+        Args:
+            frame Preprocessed image frame.
+        Returns:
+            Extracted text if confidence is above threshold, else an empty string.
+        """
+        # OCR model extraction
+        extraction = self.model.readtext(frame)
+
+        # loop all text extractions that above the confidence
+        for _, text, confidence in extraction:
+            if confidence > self.confidence:
+                return extraction[-1][1]
+
+        # Error handling
+        return ""  # raise Exception No OCR reading
+
+    def _post_process(self, extracted_text: str, is_numbers: bool) -> str:
+        if not extracted_text:
+            return ""
+
+        if is_numbers:
+            result = extracted_text.strip()
+            result = "".join(
+                self.char_to_num_mappings.get(char, char) for char in result
+            )
+
+            result = "".join(re.findall(r"\b([0-9]{1,4})\b", result))
+            # if not result.isdigit() or len(result) > 4:
+            #     return ""
+            return result
+        else:
+            result = extracted_text.strip().upper()
+            result = "".join(
+                self.num_to_char_mappings.get(char, char) for char in result
+            )
+            result = "".join(re.findall(r"[A-Z]{3}", result))
+            if len(result) != 3:
+                return ""
+            return result
+
+    def annotate_frame(self, frame, bbox, extracted_text):
+        if bbox is not None:
+            color = (0, 255, 0) if extracted_text else (0, 0, 255)
+            label = "No Extraction" if not extracted_text else extracted_text
+            cv2.rectangle(frame, (bbox[0], bbox[1]), (bbox[2], bbox[3]), color, 2)
+            cv2.putText(
+                frame,
+                f"{label}",
+                (bbox[0], bbox[1] - 10),  # Top left above the bounding box
+                cv2.FONT_HERSHEY_SIMPLEX,
+                0.9,
+                color,
+                2,
+            )
+        return frame

LicensePlateRecognizer.py

@@ -0,0 +1,64 @@
+import queue
+import threading
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
+
+import torch.nn as nn
+
+if TYPE_CHECKING:
+    from LicensePlateDetector import LicensePlateDetector
+    from LicensePlateReader import LicensePlateReader
+
+
+class LicensePlateRecognizer(nn.Module):
+    def __init__(
+        self,
+        license_detector: "LicensePlateDetector",
+        license_reader: "LicensePlateReader",
+        queue_size: int = 10,
+    ):
+        super(LicensePlateRecognizer, self).__init__()
+        self.license_detector = license_detector
+        self.license_reader = license_reader
+        self.confidence = 0.80
+
+        self.input_queue = queue.Queue(maxsize=queue_size)
+        self.output_queue = queue.Queue(maxsize=queue_size)
+        self.processing_thread = threading.Thread(
+            target=self._process_queue, daemon=True
+        )
+        self.processing_thread.start()
+
+    def forward(self, frame):
+        self.input_queue.put(frame)
+        result = self.output_queue.get()
+        if result == (None, None):
+            return None, None
+        return result
+
+    def _process_queue(self):
+        while True:
+            frame = self.input_queue.get()
+            try:
+                result = self._process_single_frame(frame)
+                self.output_queue.put(result)
+            except Exception as e:
+                print(f"Error in LicensePlateRecognizer: {e}")
+                self.output_queue.put((None, None))
+            finally:
+                self.input_queue.task_done()
+
+    def _process_single_frame(self, frame):
+        detector_result = self.license_detector.forward(frame)
+        if detector_result is None:
+            raise Exception("No license plate detected")
+
+        left_license_plate_half, right_license_plate_half, bbox = detector_result
+
+        extracted_license = self.license_reader.forward(
+            left_license_plate_half, right_license_plate_half
+        )
+
+        if not extracted_license:
+            raise Exception("Failed to read license plate")
+
+        return extracted_license, bbox

client_secrets.json

@@ -0,0 +1 @@
+{"installed":{"client_id":"1072143795553-0fudoqtd2p40nvfdpltvnaf06h03rdnq.apps.googleusercontent.com","project_id":"dunes-aero-backup-service","auth_uri":"https://accounts.google.com/o/oauth2/auth","token_uri":"https://oauth2.googleapis.com/token","auth_provider_x509_cert_url":"https://www.googleapis.com/oauth2/v1/certs","client_secret":"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","redirect_uris":["http://localhost"]}}

hello_world.py

@@ -0,0 +1,40 @@
+# from LicensePlateDetector import LicensePlateDetector
+# from LicensePlateReader import LicensePlateReader
+# from LicensePlateRecognizer import LicensePlateRecognizer
+# from LicensePlateAuthorizer import LicensePlateAuthorizer
+
+# from ultralytics import YOLO
+# import cv2
+# import easyocr
+# import numpy as np
+# import re
+# from utils import *
+# from fastapi import FastAPI, Response
+# from fastapi.responses import StreamingResponse
+# import uvicorn
+# import io
+# from starlette.responses import HTMLResponse
+
+# lpd = LicensePlateDetector(YOLO("license_plate_192.pt")) # 
+# lpr = LicensePlateReader(easyocr.Reader(['en'], gpu=False))
+# lpz = LicensePlateRecognizer(lpd, lpr)
+# lpa = LicensePlateAuthorizer(datebase_path="./database.db")
+
+# cap = cv2.VideoCapture(0)
+# color = (255,0,0)
+# while True:
+#     rat , frame = video.read()
+#     if rat:
+#         # frame = cv2.flip(frame, 1) # if camera reverses image
+#         extracted_plate , bbox = lpz.forward(frame)
+#         # Authoriztion 
+#         if lpa.verify_license_plate_authorization(extracted_plate):
+#             color = (0,255,0)
+#         frame = cv2.rectangle(frame, (bbox[0],bbox[1]),(bbox[2],bbox[3]), color , 2) 
+#         # print(extracted_plate)
+#         # _, frame = cv2.imencode('.jpeg', frame) # for JupyterNotebook
+
+# cap.release()
+
+
+print("Hello World from DALPA, Dunes Aero License Plate Authorizor.")

license_plate_192.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:dcc21d67e4a65dd84aacd5fd4687ecc5b867350afb046835551d6e1d572ab734
+size 22470894

main.py

@@ -0,0 +1,29 @@
+import numpy as np
+import cv2 as cv
+
+from LicensePlateAuthorizer import LicensePlateAuthorizer
+
+cap = cv.VideoCapture(0)
+lpa = LicensePlateAuthorizer()
+
+if not cap.isOpened():
+    print("Cannot open camera")
+    exit()
+
+while True:
+    ret, frame = cap.read()
+    if not ret:
+        print("Can't receive frame (stream end?). Exiting ...")
+        break
+
+    processed_frame = lpa.forward(frame)
+    if processed_frame is None:
+        print("Processing error or no license plate detected. Skipping frame.")
+        continue
+
+    cv.imshow('frame', processed_frame)
+    if cv.waitKey(1) == ord('q'):
+        break
+
+cap.release()
+cv.destroyAllWindows()

requirements.txt

@@ -0,0 +1,5 @@
+streamlit
+torch
+opencv-python-headless
+easyocr
+ultralytics

test.py

@@ -0,0 +1,7 @@
+from pydrive.auth import GoogleAuth
+from pydrive.drive import GoogleDrive
+
+gauth = GoogleAuth()
+gauth.LocalWebserverAuth()
+
+drive = GoogleDrive(gauth)