diff --git a/README.md b/README.md index 5228dba..98166cc 100644 --- a/README.md +++ b/README.md @@ -27,22 +27,29 @@ This project underwent three major phases of technical improvement: ![Phase 3 Training History](docs/visualizations/training_history_v3.png) +### Phase 4: Multi-modal Expansion (Image Select ReCAPTCHA) +- **Goal**: Beyond text recognition, expanding to **Object Classification** based challenges. +- **Method**: **EfficientNet-B1** backbone for 3x3 grid image classification. +- **Integration**: Unified both Text and Image CAPTCHA into a single **FastAPI** serving platform. + +![Phase 4 Training History](docs/visualizations/training_history_v4.png) --- -## ๐Ÿ—๏ธ Architecture: CRNN + CTC +## ๐Ÿ—๏ธ Integrated Architecture: FastAPI Unified Serving +The system is designed as a modular **FastAPI** application that serves multiple types of CAPTCHA models through a unified inference pipeline. ```mermaid -graph LR - A[Input Image: 50x200] --> B[CNN: Spatial Features] - B --> C[Bridge: Reshape] - C --> D[BLSTM: Seq Modeling] - D --> E[CTC Loss: Alignment] - E --> F[Output: text] +graph TD + A[Client Request] --> B[FastAPI Gateway] + B --> C{Captcha Type?} + C -- "v1 (Text)" --> D[CRNN + CTC Model] + C -- "v2 (Image)" --> E[EfficientNet-B1 Model] + D --> F[Unified Response] + E --> F ``` -- **Backbone**: CNN Layers for visual feature extraction. -- **Sequence**: Bidirectional LSTM for context modeling between characters. -- **Loss**: CTCLoss to handle unaligned sequences (crucial for CAPTCHAs with varying character widths). +- **Inference Service**: Separate service layers for `v1` and `v2`, sharing a consistent preprocessing and logging logic. +- **Scalable Serving**: Models and preprocessing scripts are encapsulated for seamless deployment across local and cloud environments. ## ๐Ÿ“Š Experimental Results (Latest Run) @@ -91,12 +98,18 @@ pip install -r requirements.txt ``` ### 3. Deploy to AWS (EC2) -The project is configured for automated deployment to **AWS EC2** via GitHub Actions & SSH. -1. Ensure the following are set in GitHub Secrets: - - `EC2_HOST` (IP or DNS), `EC2_USERNAME` (e.g. `ubuntu`), `EC2_SSH_KEY` (Private Key) - - `DOCKERHUB_USERNAME`, `DOCKERHUB_TOKEN` -2. Push or merge to the **`main`** branch to trigger the CD pipeline. -3. The workflow will build/push an image to Docker Hub and then SSH into your EC2 to restart the container on port **80**. +The project is configured for automated deployment to **AWS EC2** via GitHub Actions & SSH. Both Text and Image CAPTCHA models are bundled into a single Docker image for unified serving. + +1. **Integrated Image**: A single Dockerfile handles model weights, metadata, and dependencies for all versions. +2. **Automated CD**: Push to the `main` branch triggers the GitHub Actions pipeline. +3. **Serving Environment**: The workflow pushes the image to Docker Hub and restarts the container on EC2 port **80**. + +### 4. Unified Model Deployment Pipeline +To ensure consistent inference between development and production, we use a unified deployment script for both model versions. + +- **Standardized Inference**: All models use a `RecaptchaPredictor` pattern (as seen in `scripts/inference.py`) to wrap model loading and preprocessing. +- **Metadata-Driven Dashboard**: `generate_v1_metadata.py` and `generate_v2_metadata.py` extract performance metrics, which are then used to dynamically populate the web dashboard at runtime. + ### 3. Run Experiments ```bash diff --git a/app/services/captcha_v2_service.py b/app/services/captcha_v2_service.py index 03ea2b2..bc7aa2f 100644 --- a/app/services/captcha_v2_service.py +++ b/app/services/captcha_v2_service.py @@ -10,11 +10,10 @@ if torch.backends.mps.is_available(): DEVICE = torch.device('mps') -# Paths are relative to project root usually, but let's make them robust +# Paths BASE_DIR = Path(__file__).parent.parent.parent MODEL_PATH = BASE_DIR / "models" / "cnn_best_model.pth" METADATA_PATH = BASE_DIR / "models" / "model_metadata_v2.json" -CLASSES_PATH = BASE_DIR / "models" / "v2_classes.txt" class CaptchaV2Service: _instance = None @@ -26,14 +25,22 @@ def __new__(cls): return cls._instance def __init__(self): - if self._initialized: return + if self._initialized: + return self.model = None - self.classes = [] + self.classes = [ + "Bicycle", "Bridge", "Bus", "Car", "Chimney", + "Crosswalk", "Hydrant", "Motorcycle", "Palm", + "Stair", "Traffic Light" + ] self.transform = transforms.Compose([ transforms.Resize((224, 224)), transforms.ToTensor(), - transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]) + transforms.Normalize( + mean=[0.485, 0.456, 0.406], + std=[0.229, 0.224, 0.225] + ) ]) self.load_model() @@ -41,38 +48,31 @@ def __init__(self): def load_model(self): try: - # Load Classes from metadata + # Load classes from metadata if available if METADATA_PATH.exists(): with open(METADATA_PATH, "r") as f: meta = json.load(f) - self.classes = meta.get("classes", []) + self.classes = meta.get("classes", self.classes) print(f"โœ… [V2 Service] Loaded {len(self.classes)} classes from metadata") - - if not self.classes: - # Fallback classes (11 classes as detected in model weights) - self.classes = [ - "Bicycle", "Bridge", "Bus", "Car", "Chimney", - "Crosswalk", "Hydrant", "Motorcycle", "Other", "Palm", "Stair" - ] - print(f"โš ๏ธ [V2 Service] Using fallback 11 classes") - # Load Model: cnn_best_model.pth is EfficientNet-B1 + # Load Model: EfficientNet-B1 with specific classifier structure self.model = models.efficientnet_b1(weights=None) - - # Match the nested classifier structure found in the state_dict (classifier.1.1) num_ftrs = self.model.classifier[1].in_features + + # Match the exact structure used during training self.model.classifier[1] = nn.Sequential( - nn.Identity(), + nn.Dropout(0.2), nn.Linear(num_ftrs, len(self.classes)) ) if MODEL_PATH.exists(): state_dict = torch.load(MODEL_PATH, map_location=DEVICE) + # Handle 'model.' prefix if present new_state_dict = {} for k, v in state_dict.items(): if k.startswith("model."): - new_state_dict[k[6:]] = v + new_state_dict[k.replace("model.", "", 1)] = v else: new_state_dict[k] = v @@ -94,19 +94,26 @@ def predict(self, img_path: Path): Returns the class name (str) or None if prediction fails. """ if not self.model: - # Try reloading if missing (e.g. usage before startup) self.load_model() - if not self.model: return None + if not self.model: + return None try: - img = Image.open(img_path).convert('RGB') - img_tensor = self.transform(img).unsqueeze(0).to(DEVICE) + # Load and preprocess image + image = Image.open(img_path).convert('RGB') + image_tensor = self.transform(image).unsqueeze(0).to(DEVICE) + # Inference with torch.no_grad(): - outputs = self.model(img_tensor) - _, pred_idx = torch.max(outputs, 1) - - return self.classes[pred_idx.item()] + outputs = self.model(image_tensor) + probabilities = torch.nn.functional.softmax(outputs, dim=1) + confidence, predicted_idx = torch.max(probabilities, 1) + + idx = predicted_idx.item() + conf_score = confidence.item() * 100 + result_class = self.classes[idx] + + return result_class except Exception as e: print(f"โŒ [V2 Service] Prediction error for {img_path}: {e}") return None diff --git a/docs/visualizations/training_history_v4.png b/docs/visualizations/training_history_v4.png new file mode 100644 index 0000000..f0345d8 Binary files /dev/null and b/docs/visualizations/training_history_v4.png differ diff --git a/models/model_metadata_v2.json b/models/model_metadata_v2.json index f84eba4..163f38a 100644 --- a/models/model_metadata_v2.json +++ b/models/model_metadata_v2.json @@ -1,13 +1,15 @@ { - "accuracy": "72.98%", - "char_accuracy": "72.98%", - "loss": "1.1722", - "loss_value": "1.1722", - "precision": "0.67", - "recall": "0.73", - "f1_score": "0.70", - "type": "EfficientNet-B0 (Re-eval)", - "timestamp": "2026-01-28T01:00:10", + "accuracy": "98.02%", + "char_accuracy": "98.02%", + "precision": "0.98", + "recall": "0.98", + "f1_score": "0.98", + "loss_value": "N/A", + "type": "EfficientNet-B1 (Verified)", + "loss": "CrossEntropy", + "features": "Image Classification", + "preprocessing": "Resize (224x224), Normalization", + "timestamp": "2026-01-28T13:21:02", "classes": [ "Bicycle", "Bridge", @@ -17,9 +19,10 @@ "Crosswalk", "Hydrant", "Motorcycle", - "Other", "Palm", - "Stair" + "Stair", + "Traffic Light" ], - "dataset_path": "/Users/parkyoungdu/Downloads/samples_v2/images" + "total_samples": 10390, + "valid_predictions": 10390 } \ No newline at end of file diff --git a/scripts/generate_v2_metadata.py b/scripts/generate_v2_metadata.py index 58f34f0..c999d04 100644 --- a/scripts/generate_v2_metadata.py +++ b/scripts/generate_v2_metadata.py @@ -5,158 +5,127 @@ from pathlib import Path import torch import torch.nn as nn -from torchvision import models, transforms, datasets +from torchvision import transforms, datasets from torch.utils.data import DataLoader from sklearn.metrics import precision_recall_fscore_support from tqdm import tqdm +# Import the verified predictor from inference.py +import sys +BASE_DIR = Path(__file__).resolve().parent.parent +sys.path.append(str(BASE_DIR)) + +from scripts.inference import RecaptchaPredictor +from app.core.config import get_v2_dataset_path + # Constants DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu') if torch.backends.mps.is_available(): DEVICE = torch.device('mps') # Paths -BASE_DIR = Path(__file__).resolve().parent.parent MODELS_DIR = BASE_DIR / "models" MODEL_PATH = MODELS_DIR / "cnn_best_model.pth" -CLASSES_PATH = MODELS_DIR / "v2_classes.txt" OUTPUT_PATH = MODELS_DIR / "model_metadata_v2.json" -# Import app config to get correct dataset path -import sys -sys.path.append(str(BASE_DIR)) -from app.core.config import get_v2_dataset_path - -DATA_DIR = Path(get_v2_dataset_path()) - -def load_classes(): - if not CLASSES_PATH.exists(): - raise FileNotFoundError(f"Classes file not found at {CLASSES_PATH}") - with open(CLASSES_PATH, "r") as f: - return [line.strip() for line in f.readlines()] - -def load_model(num_classes): - if not MODEL_PATH.exists(): - raise FileNotFoundError(f"Model file not found at {MODEL_PATH}") - - print(f"Loading model from {MODEL_PATH}...") - # It's EfficientNet-B1 based on block structure - model = models.efficientnet_b1(weights=None) +def evaluate(predictor): + data_path_str = get_v2_dataset_path() + if not data_path_str: + raise FileNotFoundError("Dataset path not found") - # Match the nested classifier structure found in the state_dict (classifier.1.1) - num_ftrs = model.classifier[1].in_features - model.classifier[1] = nn.Sequential( - nn.Identity(), # 1.0 - nn.Linear(num_ftrs, num_classes) # 1.1 - ) + dataset_root = Path(data_path_str) + if not dataset_root.exists(): + raise FileNotFoundError(f"Dataset not found at {dataset_root}") + + print(f"Evaluating on dataset at {dataset_root}...") - state_dict = torch.load(MODEL_PATH, map_location=DEVICE) + # Get all images with their ground truth categories + all_images = [] + all_labels = [] - # Handle 'model.' prefix if present - new_state_dict = {} - for k, v in state_dict.items(): - if k.startswith("model."): - new_state_dict[k[6:]] = v - else: - new_state_dict[k] = v + for category_dir in dataset_root.iterdir(): + if not category_dir.is_dir() or category_dir.name.startswith('.'): + continue + + category_name = category_dir.name + if category_name not in predictor.classes: + print(f"โš ๏ธ Skipping {category_name} (not in model classes)") + continue - print(f"Sample keys in new_state_dict: {list(new_state_dict.keys())[:5]}") - model.load_state_dict(new_state_dict) - model.to(DEVICE) - model.eval() - return model - -def evaluate(model, classes): - if not DATA_DIR.exists(): - raise FileNotFoundError(f"Dataset directory not found at {DATA_DIR}") - - print(f"Evaluating on dataset at {DATA_DIR}...") - transform = transforms.Compose([ - transforms.Resize((224, 224)), - transforms.ToTensor(), - transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]) - ]) - - dataset = datasets.ImageFolder(str(DATA_DIR), transform=transform) - dataloader = DataLoader(dataset, batch_size=32, shuffle=False, num_workers=0) + category_idx = predictor.classes.index(category_name) + + for img_file in category_dir.glob("*.jpg"): + all_images.append(img_file) + all_labels.append(category_idx) + for img_file in category_dir.glob("*.png"): + all_images.append(img_file) + all_labels.append(category_idx) - criterion = nn.CrossEntropyLoss() + print(f"Total images to evaluate: {len(all_images)}") - running_loss = 0.0 + # Predict all_preds = [] - all_labels = [] + running_loss = 0.0 - with torch.no_grad(): - for inputs, labels in tqdm(dataloader, desc="Evaluating"): - inputs = inputs.to(DEVICE) - labels = labels.to(DEVICE) - - outputs = model(inputs) - loss = criterion(outputs, labels) - - running_loss += loss.item() * inputs.size(0) - _, preds = torch.max(outputs, 1) - - all_preds.extend(preds.cpu().numpy()) - all_labels.extend(labels.cpu().numpy()) + criterion = nn.CrossEntropyLoss() + + for img_path, true_label in tqdm(zip(all_images, all_labels), total=len(all_images), desc="Evaluating"): + result = predictor.predict(img_path) + if result and result["class"] in predictor.classes: + pred_idx = predictor.classes.index(result["class"]) + all_preds.append(pred_idx) - total_loss = running_loss / len(dataset) + # Calculate loss (approximate, since we don't have raw logits) + # Skip loss calculation for simplicity or use dummy + else: + all_preds.append(-1) # Invalid prediction # Calculate Metrics - correct = sum(p == l for p, l in zip(all_preds, all_labels)) - accuracy = correct / len(dataset) + valid_preds = [(p, l) for p, l in zip(all_preds, all_labels) if p != -1] + preds_only = [p for p, _ in valid_preds] + labels_only = [l for _, l in valid_preds] - precision, recall, f1, _ = precision_recall_fscore_support(all_labels, all_preds, average='weighted', zero_division=0) + correct = sum(p == l for p, l in zip(preds_only, labels_only)) + accuracy = correct / len(labels_only) if labels_only else 0 + + precision, recall, f1, _ = precision_recall_fscore_support( + labels_only, preds_only, average='weighted', zero_division=0 + ) return { - "loss": total_loss, "accuracy": accuracy, "precision": precision, "recall": recall, - "f1_score": f1 + "f1_score": f1, + "total_samples": len(all_images), + "valid_predictions": len(valid_preds) } def main(): try: print(f"Using device: {DEVICE}") - # Determine actual num_classes from model state_dict first - state_dict = torch.load(MODEL_PATH, map_location=DEVICE) - actual_num_classes = 0 - for k, v in state_dict.items(): - if 'classifier.1.1.weight' in k or 'classifier.1.weight' in k: - actual_num_classes = v.size(0) - break - - if actual_num_classes == 0: - # Fallback to loading from classes file - classes = load_classes() - actual_num_classes = len(classes) - else: - print(f"Detected {actual_num_classes} classes from model weights.") - # We still need class names for metadata, but we'll handle the mismatch if any - classes = load_classes() - if len(classes) != actual_num_classes: - print(f"โš ๏ธ Warning: Classes file has {len(classes)} classes, but model has {actual_num_classes}.") - classes = classes[:actual_num_classes] # Simple truncation for now - - model = load_model(actual_num_classes) + # Use the verified RecaptchaPredictor from inference.py + predictor = RecaptchaPredictor(str(MODEL_PATH), device=str(DEVICE)) - metrics = evaluate(model, classes) + metrics = evaluate(predictor) # Format Metadata metadata = { "accuracy": f"{metrics['accuracy']*100:.2f}%", - "char_accuracy": f"{metrics['accuracy']*100:.2f}%", # Reuse for consistent UI if needed, or omit - "loss": f"{metrics['loss']:.4f}", # Display string - "loss_value": f"{metrics['loss']:.4f}", # Raw value string + "char_accuracy": f"{metrics['accuracy']*100:.2f}%", "precision": f"{metrics['precision']:.2f}", "recall": f"{metrics['recall']:.2f}", "f1_score": f"{metrics['f1_score']:.2f}", - "type": "EfficientNet-B0 (Re-eval)", + "loss_value": "N/A", # Not calculated in this approach + "type": "EfficientNet-B1 (Verified)", + "loss": "CrossEntropy", + "features": "Image Classification", + "preprocessing": "Resize (224x224), Normalization", "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"), - "classes": classes, - "dataset_path": str(DATA_DIR) + "classes": predictor.classes, + "total_samples": metrics['total_samples'], + "valid_predictions": metrics['valid_predictions'] } # Save diff --git a/scripts/inference.py b/scripts/inference.py new file mode 100644 index 0000000..ac685ad --- /dev/null +++ b/scripts/inference.py @@ -0,0 +1,138 @@ +import torch +import torch.nn as nn +from torchvision import models, transforms +from PIL import Image +from pathlib import Path +import json + +class RecaptchaPredictor: + def __init__(self, model_path, device=None): + """ + reCAPTCHA ๋ชจ๋ธ ์ถ”๋ก ์„ ์œ„ํ•œ ๋ž˜ํผ ํด๋ž˜์Šค. + ๋ชจ๋ธ ๋กœ๋“œ์™€ ์ „์ฒ˜๋ฆฌ๋ฅผ ์ž๋™์œผ๋กœ ์ˆ˜ํ–‰ํ•ฉ๋‹ˆ๋‹ค. + + Args: + model_path (str): ํ•™์Šต๋œ .pth ๋ชจ๋ธ ํŒŒ์ผ ๊ฒฝ๋กœ + device (str): 'cuda', 'mps', 'cpu' ์ค‘ ์„ ํƒ (None์ด๋ฉด ์ž๋™ ์„ ํƒ) + """ + self.classes = [ + "Bicycle", "Bridge", "Bus", "Car", "Chimney", + "Crosswalk", "Hydrant", "Motorcycle", "Palm", + "Stair", "Traffic Light" + ] + + # ํŒŒ์ผ ์‹œ์Šคํ…œ์˜ ํด๋”๋ช…๊ณผ ๋ชจ๋ธ ํด๋ž˜์Šค๋ช… ๊ฐ„์˜ ๋งคํ•‘ ์ •์˜ + self.folder_to_class = { + "TLight": "Traffic Light", + "Traffic Light": "Traffic Light", + # ๋‚˜๋จธ์ง€๋Š” ์ด๋ฆ„์ด ๋™์ผํ•˜๋‹ค๊ณ  ๊ฐ€์ • + } + + # ์žฅ์น˜ ์ž๋™ ์„ค์ • + if device: + self.device = torch.device(device) + elif torch.cuda.is_available(): + self.device = torch.device("cuda") + elif hasattr(torch.backends, 'mps') and torch.backends.mps.is_available(): + self.device = torch.device("mps") + else: + self.device = torch.device("cpu") + + # 1. ๋ชจ๋ธ ์•„ํ‚คํ…์ฒ˜ ์ •์˜ (EfficientNet-B1) + self.model = models.efficientnet_b1(weights=None) + + # ๋งˆ์ง€๋ง‰ ๋ถ„๋ฅ˜ ๋ ˆ์ด์–ด ์ˆ˜์ • (ํด๋ž˜์Šค 11๊ฐœ) + num_ftrs = self.model.classifier[1].in_features + self.model.classifier[1] = nn.Sequential( + nn.Dropout(0.2), + nn.Linear(num_ftrs, len(self.classes)) + ) + + # 2. ๊ฐ€์ค‘์น˜ ๋กœ๋“œ + try: + state_dict = torch.load(model_path, map_location=self.device) + + # 'model.' ์ ‘๋‘์–ด ์ œ๊ฑฐ (ํ•™์Šต ์‹œ RecaptchaCNN ๋ž˜ํผ ์‚ฌ์šฉ์œผ๋กœ ์ธํ•œ ํ‚ค ๋ถˆ์ผ์น˜ ํ•ด๊ฒฐ) + new_state_dict = {} + for k, v in state_dict.items(): + if k.startswith("model."): + new_state_dict[k.replace("model.", "", 1)] = v + else: + new_state_dict[k] = v + + self.model.load_state_dict(new_state_dict) + print(f"โœ… Model loaded successfully from {model_path}") + except Exception as e: + print(f"โŒ Failed to load model: {e}") + raise e + + self.model.to(self.device) + self.model.eval() # ํ‰๊ฐ€ ๋ชจ๋“œ ํ•„์ˆ˜! + + # 3. ์ „์ฒ˜๋ฆฌ ํŒŒ์ดํ”„๋ผ์ธ (ํ•™์Šต ์„ค์ •๊ณผ ๋™์ผํ•˜๊ฒŒ ๊ณ ์ •) + self.transform = transforms.Compose([ + transforms.Resize((224, 224)), # config.yaml์˜ resize ๊ฐ’ (EfficientNet-B1) + transforms.ToTensor(), + transforms.Normalize( + mean=[0.485, 0.456, 0.406], + std=[0.229, 0.224, 0.225] + ) + ]) + + def predict(self, image_source): + """ + ์ด๋ฏธ์ง€ ๊ฒฝ๋กœ ๋˜๋Š” PIL ์ด๋ฏธ์ง€๋ฅผ ๋ฐ›์•„ ์˜ˆ์ธก ๊ฒฐ๊ณผ๋ฅผ ๋ฐ˜ํ™˜ํ•ฉ๋‹ˆ๋‹ค. + + Args: + image_source (str or PIL.Image): ์ด๋ฏธ์ง€ ๊ฒฝ๋กœ ๋˜๋Š” ๊ฐ์ฒด + + Returns: + dict: ์˜ˆ์ธก ํด๋ž˜์Šค, ์‹ ๋ขฐ๋„(Confidence), ํด๋ž˜์Šค ์ธ๋ฑ์Šค + """ + # ์ด๋ฏธ์ง€ ๋กœ๋“œ + if isinstance(image_source, str) or isinstance(image_source, Path): + image = Image.open(image_source).convert('RGB') + else: + image = image_source.convert('RGB') + + # ์ „์ฒ˜๋ฆฌ + image_tensor = self.transform(image).unsqueeze(0).to(self.device) + + # ์ถ”๋ก  + with torch.no_grad(): + outputs = self.model(image_tensor) + probabilities = torch.nn.functional.softmax(outputs, dim=1) + confidence, predicted_idx = torch.max(probabilities, 1) + + idx = predicted_idx.item() + conf_score = confidence.item() * 100 + result_class = self.classes[idx] + + return { + "class": result_class, + "confidence": f"{conf_score:.2f}%", + "index": idx + } + +# --- ์‚ฌ์šฉ ์˜ˆ์‹œ --- +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description='Run inference on a single image') + parser.add_argument('--model', type=str, required=True, help='Path to .pth model file') + parser.add_argument('--image', type=str, required=True, help='Path to image file') + args = parser.parse_args() + + # ์˜ˆ์ธก๊ธฐ ์ƒ์„ฑ + predictor = RecaptchaPredictor(args.model) + + # ์˜ˆ์ธก ์‹คํ–‰ + try: + result = predictor.predict(args.image) + print("\n Prediction Result:") + print(f"------------------------") + print(f" Class: {result['class']}") + print(f" Confidence: {result['confidence']}") + print(f"------------------------") + except Exception as e: + print(f"Error during inference: {e}")