import requests
from bs4 import BeautifulSoup
from openai import OpenAI
import json

# Initialize OpenAI client
client = OpenAI(api_key="YOUR_API_KEY")

url = "https://example.com/product-page"
response = requests.get(url)
soup = BeautifulSoup(response.text, "html.parser")

# Extract visible text only
page_text = soup.get_text(separator="\n")

prompt = f"""
Extract the following details from the text:
- Product Name
- Price
- Description
- Key Features

Return output in JSON format.

TEXT:
{page_text}
"""

completion = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": prompt}],
)

result = completion.choices[0].message.content

print(result)

  

# yolo_training.py - Complete Custom YOLO Training Pipeline
import ultralytics
from ultralytics import YOLO
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
import cv2
import numpy as np
from pathlib import Path
import yaml
from datetime import datetime
import albumentations as A
from albumentations.pytorch import ToTensorV2
import wandb
from sklearn.model_selection import train_test_split
import json

class CustomYOLOTraining:
    def __init__(self, model_size='yolov8m', project_name='custom-detection'):
        self.model_size = model_size
        self.project_name = project_name
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        
        # Initialize weights and biases for experiment tracking
        wandb.init(project=project_name, config={
            "model_size": model_size,
            "device": str(self.device),
            "timestamp": datetime.now().isoformat()
        })
    
    def prepare_dataset(self, data_dir, annotations_format='yolo', 
                       train_ratio=0.8, val_ratio=0.15, test_ratio=0.05):
        """Prepare custom dataset for YOLO training with proper splits"""
        
        data_dir = Path(data_dir)
        images_dir = data_dir / 'images'
        labels_dir = data_dir / 'labels'
        
        # Get all image files
        image_files = list(images_dir.glob('*.jpg')) + list(images_dir.glob('*.png'))
        image_files = [f for f in image_files if f.exists()]
        
        # Split dataset
        train_files, temp_files = train_test_split(image_files, train_size=train_ratio, random_state=42)
        val_files, test_files = train_test_split(temp_files, train_size=val_ratio/(val_ratio+test_ratio), random_state=42)
        
        # Create dataset YAML configuration
        dataset_config = {
            'path': str(data_dir.absolute()),
            'train': str(images_dir.absolute()),
            'val': str(images_dir.absolute()),
            'test': str(images_dir.absolute()),
            'names': self.get_class_names(data_dir)
        }
        
        # Save dataset YAML
        config_path = data_dir / 'dataset.yaml'
        with open(config_path, 'w') as f:
            yaml.dump(dataset_config, f)
        
        # Create split files
        self._create_split_file(data_dir / 'train.txt', train_files)
        self._create_split_file(data_dir / 'val.txt', val_files)
        self._create_split_file(data_dir / 'test.txt', test_files)
        
        return config_path, len(train_files), len(val_files), len(test_files)
    
    def get_class_names(self, data_dir):
        """Extract class names from dataset"""
        # In practice, you might load this from a classes.txt file
        # or extract from annotation files
        class_files = list((data_dir / 'labels').glob('*.txt'))
        classes = set()
        
        for class_file in class_files[:100]:  # Sample to get classes
            with open(class_file, 'r') as f:
                for line in f:
                    if line.strip():
                        class_id = int(line.split()[0])
                        classes.add(class_id)
        
        # Create class names (you would replace with actual names)
        return {i: f'class_{i}' for i in sorted(classes)}
    
    def _create_split_file(self, split_path, files):
        """Create split file with relative paths"""
        with open(split_path, 'w') as f:
            for file_path in files:
                f.write(f"{file_path.relative_to(file_path.parent.parent)}\n")
    
    def setup_data_augmentation(self):
        """Setup advanced data augmentation pipeline"""
        
        train_transform = A.Compose([
            # Geometric transformations
            A.HorizontalFlip(p=0.5),
            A.RandomRotate90(p=0.3),
            A.ShiftScaleRotate(
                shift_limit=0.1, 
                scale_limit=0.1, 
                rotate_limit=15, 
                p=0.5
            ),
            A.Perspective(scale=(0.05, 0.1), p=0.3),
            
            # Color transformations
            A.RandomBrightnessContrast(
                brightness_limit=0.2, 
                contrast_limit=0.2, 
                p=0.5
            ),
            A.HueSaturationValue(
                hue_shift_limit=10, 
                sat_shift_limit=20, 
                val_shift_limit=10, 
                p=0.5
            ),
            A.CLAHE(clip_limit=2.0, p=0.3),
            A.RandomGamma(gamma_limit=(80, 120), p=0.3),
            
            # Noise and blur
            A.GaussNoise(var_limit=(10.0, 50.0), p=0.3),
            A.MotionBlur(blur_limit=7, p=0.2),
            A.MedianBlur(blur_limit=3, p=0.1),
            
            # Weather effects
            A.RandomFog(fog_coef_lower=0.1, fog_coef_upper=0.3, p=0.1),
            A.RandomShadow(p=0.2),
            
            # Advanced augmentations
            A.Cutout(
                num_holes=8, 
                max_h_size=32, 
                max_w_size=32, 
                fill_value=0, 
                p=0.5
            ),
            A.CoarseDropout(
                max_holes=8, 
                max_height=32, 
                max_width=32, 
                p=0.3
            ),
            
            # Normalization
            A.Normalize(
                mean=[0.485, 0.456, 0.406],
                std=[0.229, 0.224, 0.225]
            ),
            ToTensorV2()
        ], bbox_params=A.BboxParams(
            format='yolo', 
            label_fields=['class_labels']
        ))
        
        val_transform = A.Compose([
            A.Normalize(
                mean=[0.485, 0.456, 0.406],
                std=[0.229, 0.224, 0.225]
            ),
            ToTensorV2()
        ], bbox_params=A.BboxParams(
            format='yolo', 
            label_fields=['class_labels']
        ))
        
        return train_transform, val_transform
    
    def setup_model(self, num_classes, pretrained=True):
        """Initialize YOLO model with custom configuration"""
        
        # Load pre-trained model
        if pretrained:
            model = YOLO(f'{self.model_size}.pt')
        else:
            model = YOLO(f'{self.model_size}.yaml')
        
        # Update model for custom number of classes
        model.model.nc = num_classes
        
        # Freeze backbone for transfer learning (optional)
        if pretrained:
            self._freeze_backbone(model)
        
        return model
    
    def _freeze_backbone(self, model, freeze_ratio=0.5):
        """Freeze portion of backbone for transfer learning"""
        backbone_layers = []
        for name, param in model.model.named_parameters():
            if 'model.0' in name or 'model.1' in name:  # Early layers
                backbone_layers.append(name)
        
        # Freeze first half of backbone layers
        freeze_count = int(len(backbone_layers) * freeze_ratio)
        for name in backbone_layers[:freeze_count]:
            for param_name, param in model.model.named_parameters():
                if name in param_name:
                    param.requires_grad = False
        
        print(f"Froze {freeze_count} backbone layers for transfer learning")
    
    def train_model(self, model, dataset_config, epochs=100, 
                   batch_size=16, learning_rate=0.01, patience=20):
        """Train YOLO model with advanced configuration"""
        
        training_results = model.train(
            data=str(dataset_config),
            epochs=epochs,
            imgsz=640,
            batch=batch_size,
            lr0=learning_rate,
            patience=patience,
            save=True,
            save_period=10,
            cache=False,
            device=self.device,
            workers=8,
            project=self.project_name,
            name=f'train_{datetime.now().strftime("%Y%m%d_%H%M%S")}',
            exist_ok=True,
            
            # Advanced training parameters
            optimizer='AdamW',
            weight_decay=0.0005,
            warmup_epochs=3,
            warmup_momentum=0.8,
            warmup_bias_lr=0.1,
            box=7.5,  # box loss gain
            cls=0.5,  # cls loss gain
            dfl=1.5,  # dfl loss gain
            
            # Augmentation parameters
            hsv_h=0.015,
            hsv_s=0.7,
            hsv_v=0.4,
            degrees=0.0,
            translate=0.1,
            scale=0.5,
            shear=0.0,
            perspective=0.0,
            flipud=0.0,
            fliplr=0.5,
            mosaic=1.0,
            mixup=0.0,
            copy_paste=0.0
        )
        
        return training_results
    
    def evaluate_model(self, model, dataset_config):
        """Comprehensive model evaluation"""
        
        # Validation metrics
        metrics = model.val(
            data=str(dataset_config),
            split='val',
            imgsz=640,
            batch_size=16,
            save_json=True,
            save_hybrid=False,
            conf=0.001,
            iou=0.6,
            max_det=300,
            half=True,
            device=self.device
        )
        
        # Test metrics
        test_metrics = model.val(
            data=str(dataset_config),
            split='test',
            imgsz=640,
            batch_size=16,
            save_json=True,
            conf=0.001,
            iou=0.6,
            device=self.device
        )
        
        return {
            'validation_metrics': metrics,
            'test_metrics': test_metrics
        }
    
    def export_model(self, model, export_formats=['onnx', 'tflite', 'engine']):
        """Export model to various formats for deployment"""
        
        exported_models = {}
        
        for format in export_formats:
            try:
                if format == 'onnx':
                    exported_path = model.export(
                        format='onnx',
                        dynamic=True,
                        simplify=True,
                        opset=17
                    )
                elif format == 'tflite':
                    exported_path = model.export(
                        format='tflite',
                        int8=True,
                        data='path/to/calibration/data'
                    )
                elif format == 'engine':
                    exported_path = model.export(
                        format='engine',
                        half=True,
                        device=0
                    )
                else:
                    continue
                
                exported_models[format] = exported_path
                print(f"Exported model to {format}: {exported_path}")
                
            except Exception as e:
                print(f"Failed to export to {format}: {e}")
        
        return exported_models

# Example usage
def train_custom_detector():
    trainer = CustomYOLOTraining(model_size='yolov8m', project_name='custom-object-detection')
    
    # Prepare dataset
    dataset_config, train_count, val_count, test_count = trainer.prepare_dataset(
        data_dir='path/to/your/dataset',
        train_ratio=0.8,
        val_ratio=0.15,
        test_ratio=0.05
    )
    
    print(f"Dataset prepared: {train_count} train, {val_count} val, {test_count} test images")
    
    # Setup model
    num_classes = 10  # Replace with your actual number of classes
    model = trainer.setup_model(num_classes=num_classes, pretrained=True)
    
    # Train model
    training_results = trainer.train_model(
        model=model,
        dataset_config=dataset_config,
        epochs=100,
        batch_size=16,
        learning_rate=0.01,
        patience=20
    )
    
    # Evaluate model
    evaluation_results = trainer.evaluate_model(model, dataset_config)
    print("Evaluation results:", evaluation_results)
    
    # Export models for deployment
    exported_models = trainer.export_model(model, export_formats=['onnx', 'tflite'])
    
    return model, training_results, evaluation_results, exported_models

if __name__ == "__main__":
    model, training_results, evaluation_results, exported_models = train_custom_detector()

  

# data_preparation.py - Advanced Data Pipeline for Object Detection
import cv2
import numpy as np
from pathlib import Path
import json
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from typing import List, Dict, Tuple
import albumentations as A
from albumentations.pytorch import ToTensorV2
import torch
from torch.utils.data import Dataset, DataLoader

@dataclass
class BoundingBox:
    x_center: float
    y_center: float
    width: float
    height: float
    class_id: int
    class_name: str

class ObjectDetectionDataset(Dataset):
    def __init__(self, images_dir: Path, labels_dir: Path, 
                 transform=None, target_size: Tuple[int, int] = (640, 640)):
        self.images_dir = Path(images_dir)
        self.labels_dir = Path(labels_dir)
        self.transform = transform
        self.target_size = target_size
        
        # Get all valid image-label pairs
        self.samples = self._discover_samples()
        
        # Class mapping
        self.classes = self._discover_classes()
        self.class_to_id = {cls: idx for idx, cls in enumerate(self.classes)}
        self.id_to_class = {idx: cls for idx, cls in enumerate(self.classes)}
    
    def _discover_samples(self):
        """Discover all valid image-label pairs"""
        samples = []
        
        for image_path in self.images_dir.glob('*.*'):
            if image_path.suffix.lower() not in ['.jpg', '.jpeg', '.png', '.bmp']:
                continue
            
            # Find corresponding label file
            label_path = self.labels_dir / f"{image_path.stem}.txt"
            
            if label_path.exists():
                samples.append((image_path, label_path))
            else:
                print(f"Warning: No label found for {image_path}")
        
        return samples
    
    def _discover_classes(self):
        """Discover all classes from label files"""
        classes = set()
        
        for _, label_path in self.samples:
            with open(label_path, 'r') as f:
                for line in f:
                    if line.strip():
                        class_id = int(line.split()[0])
                        classes.add(class_id)
        
        return sorted(classes)
    
    def __len__(self):
        return len(self.samples)
    
    def __getitem__(self, idx):
        image_path, label_path = self.samples[idx]
        
        # Load image
        image = cv2.imread(str(image_path))
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        original_height, original_width = image.shape[:2]
        
        # Load bounding boxes
        bboxes = []
        class_labels = []
        
        with open(label_path, 'r') as f:
            for line in f:
                if line.strip():
                    parts = line.strip().split()
                    class_id = int(parts[0])
                    x_center = float(parts[1])
                    y_center = float(parts[2])
                    width = float(parts[3])
                    height = float(parts[4])
                    
                    bboxes.append([x_center, y_center, width, height])
                    class_labels.append(class_id)
        
        # Apply transformations
        if self.transform:
            transformed = self.transform(
                image=image,
                bboxes=bboxes,
                class_labels=class_labels
            )
            image = transformed['image']
            bboxes = transformed['bboxes']
            class_labels = transformed['class_labels']
        
        # Convert to tensor format
        target = {
            'boxes': torch.tensor(bboxes, dtype=torch.float32) if bboxes else torch.zeros((0, 4)),
            'labels': torch.tensor(class_labels, dtype=torch.int64) if class_labels else torch.zeros(0, dtype=torch.int64),
            'image_id': torch.tensor([idx]),
            'area': (torch.tensor(bboxes)[:, 2] * torch.tensor(bboxes)[:, 3]) if bboxes else torch.zeros(0),
            'iscrowd': torch.zeros(len(bboxes) if bboxes else 0, dtype=torch.int64)
        }
        
        return image, target
    
    def visualize_sample(self, idx, save_path=None):
        """Visualize a sample with bounding boxes"""
        image, target = self.__getitem__(idx)
        
        # Convert tensor to numpy for visualization
        if isinstance(image, torch.Tensor):
            image = image.permute(1, 2, 0).numpy()
            image = (image * np.array([0.229, 0.224, 0.225]) + np.array([0.485, 0.456, 0.406])) * 255
            image = image.astype(np.uint8)
        
        image = image.copy()
        boxes = target['boxes'].numpy()
        labels = target['labels'].numpy()
        
        height, width = image.shape[:2]
        
        for box, label in zip(boxes, labels):
            x_center, y_center, w, h = box
            x1 = int((x_center - w/2) * width)
            y1 = int((y_center - h/2) * height)
            x2 = int((x_center + w/2) * width)
            y2 = int((y_center + h/2) * height)
            
            # Draw bounding box
            cv2.rectangle(image, (x1, y1), (x2, y2), (0, 255, 0), 2)
            
            # Draw label
            class_name = self.id_to_class.get(label, f"Class_{label}")
            cv2.putText(image, class_name, (x1, y1-10), 
                       cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)
        
        if save_path:
            cv2.imwrite(str(save_path), cv2.cvtColor(image, cv2.COLOR_RGB2BGR))
        
        return image

class AdvancedDataAugmentation:
    def __init__(self, target_size=(640, 640)):
        self.target_size = target_size
        
        # Training augmentations
        self.train_transform = A.Compose([
            # Geometric transformations
            A.LongestMaxSize(max_size=max(target_size)),
            A.PadIfNeeded(
                min_height=target_size[0],
                min_width=target_size[1],
                border_mode=cv2.BORDER_CONSTANT,
                value=0
            ),
            A.HorizontalFlip(p=0.5),
            A.VerticalFlip(p=0.2),
            A.RandomRotate90(p=0.3),
            A.ShiftScaleRotate(
                shift_limit=0.1,
                scale_limit=0.2,
                rotate_limit=15,
                p=0.5,
                border_mode=cv2.BORDER_CONSTANT,
                value=0
            ),
            
            # Color transformations
            A.RandomBrightnessContrast(
                brightness_limit=0.3,
                contrast_limit=0.3,
                p=0.5
            ),
            A.HueSaturationValue(
                hue_shift_limit=20,
                sat_shift_limit=30,
                val_shift_limit=20,
                p=0.5
            ),
            A.CLAHE(clip_limit=2.0, p=0.3),
            A.RandomGamma(gamma_limit=(80, 120), p=0.3),
            
            # Advanced augmentations
            A.Cutout(
                num_holes=8,
                max_h_size=32,
                max_w_size=32,
                fill_value=0,
                p=0.5
            ),
            A.MixUp(p=0.2),
            A.Mosaic(p=0.2),
            
            # Normalization
            A.Normalize(
                mean=[0.485, 0.456, 0.406],
                std=[0.229, 0.224, 0.225]
            ),
            ToTensorV2()
        ], bbox_params=A.BboxParams(
            format='yolo',
            label_fields=['class_labels']
        ))
        
        # Validation augmentations (minimal)
        self.val_transform = A.Compose([
            A.LongestMaxSize(max_size=max(target_size)),
            A.PadIfNeeded(
                min_height=target_size[0],
                min_width=target_size[1],
                border_mode=cv2.BORDER_CONSTANT,
                value=0
            ),
            A.Normalize(
                mean=[0.485, 0.456, 0.406],
                std=[0.229, 0.224, 0.225]
            ),
            ToTensorV2()
        ], bbox_params=A.BboxParams(
            format='yolo',
            label_fields=['class_labels']
        ))

def create_data_loaders(images_dir, labels_dir, batch_size=16, num_workers=8):
    """Create training and validation data loaders"""
    
    aug = AdvancedDataAugmentation(target_size=(640, 640))
    
    # Split dataset
    dataset = ObjectDetectionDataset(images_dir, labels_dir)
    train_size = int(0.8 * len(dataset))
    val_size = len(dataset) - train_size
    train_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size])
    
    # Apply transforms
    train_dataset.dataset.transform = aug.train_transform
    val_dataset.dataset.transform = aug.val_transform
    
    # Create data loaders
    train_loader = DataLoader(
        train_dataset,
        batch_size=batch_size,
        shuffle=True,
        num_workers=num_workers,
        pin_memory=True,
        collate_fn=collate_fn
    )
    
    val_loader = DataLoader(
        val_dataset,
        batch_size=batch_size,
        shuffle=False,
        num_workers=num_workers,
        pin_memory=True,
        collate_fn=collate_fn
    )
    
    return train_loader, val_loader, dataset.classes

def collate_fn(batch):
    """Custom collate function for object detection"""
    images = []
    targets = []
    
    for image, target in batch:
        images.append(image)
        targets.append(target)
    
    return images, targets

# Example usage
def prepare_custom_dataset():
    images_dir = Path('path/to/your/images')
    labels_dir = Path('path/to/your/labels')
    
    train_loader, val_loader, classes = create_data_loaders(
        images_dir, labels_dir, batch_size=16, num_workers=8
    )
    
    print(f"Created data loaders with {len(classes)} classes: {classes}")
    print(f"Training samples: {len(train_loader.dataset)}")
    print(f"Validation samples: {len(val_loader.dataset)}")
    
    return train_loader, val_loader, classes

if __name__ == "__main__":
    train_loader, val_loader, classes = prepare_custom_dataset()

  

# tensorflow_serving.py - Production Model Serving
import tensorflow as tf
import grpc
import numpy as np
from typing import Dict, List, Any
import cv2
import json
from datetime import datetime
import requests
from concurrent import futures
import threading
from prometheus_client import start_http_server, Counter, Histogram, Gauge
import logging

class YOLOTensorFlowServing:
    def __init__(self, model_path: str, serving_url: str = "localhost:8501"):
        self.serving_url = serving_url
        self.model_path = Path(model_path)
        self.logger = self._setup_logging()
        
        # Prometheus metrics
        self.setup_metrics()
        
        # Load model signature (if available)
        self.signature = self._load_model_signature()
    
    def _setup_logging(self):
        """Setup structured logging"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        return logging.getLogger(__name__)
    
    def setup_metrics(self):
        """Setup Prometheus metrics for monitoring"""
        self.request_counter = Counter(
            'yolo_inference_requests_total',
            'Total number of inference requests',
            ['model_version', 'status']
        )
        
        self.inference_latency = Histogram(
            'yolo_inference_latency_seconds',
            'Inference latency in seconds',
            ['model_version']
        )
        
        self.batch_size_gauge = Gauge(
            'yolo_batch_size',
            'Current batch size being processed'
        )
        
        # Start metrics server
        start_http_server(8000)
    
    def preprocess_image(self, image: np.ndarray, target_size: tuple = (640, 640)) -> np.ndarray:
        """Preprocess image for YOLO inference"""
        
        # Resize image
        original_shape = image.shape[:2]
        image_resized = cv2.resize(image, target_size)
        
        # Normalize
        image_normalized = image_resized.astype(np.float32) / 255.0
        
        # Convert to RGB if needed
        if len(image_normalized.shape) == 3 and image_normalized.shape[2] == 3:
            image_normalized = cv2.cvtColor(image_normalized, cv2.COLOR_BGR2RGB)
        
        # Add batch dimension
        image_batch = np.expand_dims(image_normalized, axis=0)
        
        return image_batch, original_shape
    
    def postprocess_detections(self, predictions: np.ndarray, 
                             original_shape: tuple, 
                             confidence_threshold: float = 0.25,
                             iou_threshold: float = 0.45) -> List[Dict[str, Any]]:
        """Postprocess YOLO model predictions"""
        
        detections = []
        
        # YOLO output format: [batch, num_detections, 6] 
        # where 6 = [x1, y1, x2, y2, confidence, class_id]
        if len(predictions.shape) == 3 and predictions.shape[2] == 6:
            batch_detections = predictions[0]  # Take first batch
            
            for detection in batch_detections:
                x1, y1, x2, y2, confidence, class_id = detection
                
                if confidence < confidence_threshold:
                    continue
                
                # Scale coordinates to original image size
                scale_x = original_shape[1] / 640  # Assuming model input was 640x640
                scale_y = original_shape[0] / 640
                
                x1_scaled = int(x1 * scale_x)
                y1_scaled = int(y1 * scale_y)
                x2_scaled = int(x2 * scale_x)
                y2_scaled = int(y2 * scale_y)
                
                detection_dict = {
                    'bbox': [x1_scaled, y1_scaled, x2_scaled, y2_scaled],
                    'confidence': float(confidence),
                    'class_id': int(class_id),
                    'class_name': f'class_{int(class_id)}'  # Replace with actual class names
                }
                
                detections.append(detection_dict)
        
        # Apply Non-Maximum Suppression
        detections = self._apply_nms(detections, iou_threshold)
        
        return detections
    
    def _apply_nms(self, detections: List[Dict], iou_threshold: float) -> List[Dict]:
        """Apply Non-Maximum Suppression to remove overlapping boxes"""
        
        if not detections:
            return []
        
        # Sort by confidence
        detections.sort(key=lambda x: x['confidence'], reverse=True)
        
        filtered_detections = []
        
        while detections:
            # Take the detection with highest confidence
            best_detection = detections.pop(0)
            filtered_detections.append(best_detection)
            
            # Remove overlapping detections
            detections = [
                det for det in detections 
                if self._calculate_iou(best_detection['bbox'], det['bbox']) < iou_threshold
            ]
        
        return filtered_detections
    
    def _calculate_iou(self, box1: List[float], box2: List[float]) -> float:
        """Calculate Intersection over Union between two boxes"""
        
        x11, y1_1, x2_1, y2_1 = box1
        x1_2, y1_2, x2_2, y2_2 = box2
        
        # Calculate intersection area
        xi1 = max(x1_1, x1_2)
        yi1 = max(y1_1, y1_2)
        xi2 = min(x2_1, x2_2)
        yi2 = min(y2_1, y2_2)
        
        intersection_area = max(0, xi2 - xi1) * max(0, yi2 - yi1)
        
        # Calculate union area
        box1_area = (x2_1 - x1_1) * (y2_1 - y1_1)
        box2_area = (x2_2 - x1_2) * (y2_2 - y1_2)
        union_area = box1_area + box2_area - intersection_area
        
        return intersection_area / union_area if union_area > 0 else 0
    
    def predict_single(self, image: np.ndarray, model_version: str = "1") -> Dict[str, Any]:
        """Perform single image inference"""
        
        start_time = datetime.now()
        
        try:
            # Preprocess image
            processed_image, original_shape = self.preprocess_image(image)
            
            # Prepare request data
            request_data = {
                "signature_name": "serving_default",
                "instances": processed_image.tolist()
            }
            
            # Make REST API request to TensorFlow Serving
            response = requests.post(
                f"http://{self.serving_url}/v1/models/yolo_model/versions/{model_version}:predict",
                json=request_data,
                timeout=30
            )
            
            if response.status_code == 200:
                predictions = np.array(response.json()['predictions'])
                
                # Postprocess detections
                detections = self.postprocess_detections(predictions, original_shape)
                
                # Record successful inference
                self.request_counter.labels(model_version=model_version, status='success').inc()
                
                result = {
                    'success': True,
                    'detections': detections,
                    'inference_time': (datetime.now() - start_time).total_seconds(),
                    'model_version': model_version
                }
                
            else:
                self.request_counter.labels(model_version=model_version, status='error').inc()
                result = {
                    'success': False,
                    'error': f"HTTP {response.status_code}: {response.text}",
                    'model_version': model_version
                }
            
            # Record latency
            inference_time = (datetime.now() - start_time).total_seconds()
            self.inference_latency.labels(model_version=model_version).observe(inference_time)
            
            return result
            
        except Exception as e:
            self.request_counter.labels(model_version=model_version, status='error').inc()
            self.logger.error(f"Inference error: {e}")
            
            return {
                'success': False,
                'error': str(e),
                'model_version': model_version
            }
    
    def predict_batch(self, images: List[np.ndarray], model_version: str = "1") -> List[Dict[str, Any]]:
        """Perform batch inference"""
        
        self.batch_size_gauge.set(len(images))
        results = []
        
        for image in images:
            result = self.predict_single(image, model_version)
            results.append(result)
        
        return results
    
    def get_model_status(self) -> Dict[str, Any]:
        """Get TensorFlow Serving model status"""
        
        try:
            response = requests.get(f"http://{self.serving_url}/v1/models/yolo_model")
            
            if response.status_code == 200:
                return response.json()
            else:
                return {'error': f"HTTP {response.status_code}: {response.text}"}
                
        except Exception as e:
            return {'error': str(e)}
    
    def load_new_model_version(self, new_model_path: str, version: str):
        """Load a new model version for A/B testing"""
        
        # This would typically be done through TensorFlow Serving's model management API
        # or by updating the model directory structure
        
        self.logger.info(f"Loading new model version {version} from {new_model_path}")
        
        # In production, you might use:
        # - Model version directories
        # - TensorFlow Serving's model config API
        # - Custom model management system
        
        return True

class ModelVersionManager:
    def __init__(self, model_base_path: str):
        self.model_base_path = Path(model_base_path)
        self.available_versions = self._discover_versions()
    
    def _discover_versions(self) -> Dict[str, Path]:
        """Discover available model versions"""
        versions = {}
        
        for version_dir in self.model_base_path.glob("*/"):
            if version_dir.is_dir() and version_dir.name.isdigit():
                model_files = list(version_dir.glob("saved_model.pb"))
                if model_files:
                    versions[version_dir.name] = version_dir
        
        return versions
    
    def get_latest_version(self) -> str:
        """Get the latest model version"""
        if not self.available_versions:
            return None
        
        return max(self.available_versions.keys(), key=int)
    
    def route_request(self, request_data: Dict, version: str = None) -> str:
        """Route request to appropriate model version"""
        
        if version and version in self.available_versions:
            return version
        
        # Default to latest version
        return self.get_latest_version()

# Example usage
def serve_yolo_model():
    # Initialize serving client
    serving_client = YOLOTensorFlowServing(
        model_path="path/to/your/saved_model",
        serving_url="localhost:8501"
    )
    
    # Load test image
    test_image = cv2.imread("test_image.jpg")
    
    # Perform inference
    result = serving_client.predict_single(test_image, model_version="1")
    
    if result['success']:
        print(f"Found {len(result['detections'])} detections")
        for detection in result['detections']:
            print(f"Class: {detection['class_name']}, Confidence: {detection['confidence']:.3f}")
    else:
        print(f"Inference failed: {result['error']}")
    
    return result

if __name__ == "__main__":
    result = serve_yolo_model()

  

# Simple AI-based phishing email detector using Python & Scikit-learn
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.naive_bayes import MultinomialNB

# Sample dataset (for demonstration)
emails = [
    "Urgent! Verify your account now to avoid suspension",
    "Meeting scheduled for tomorrow at 3PM",
    "You won a $10,000 lottery prize. Claim now!",
    "Project report attached for your review"
]
labels = [1, 0, 1, 0]  # 1 = phishing, 0 = safe

# Vectorize emails
vectorizer = CountVectorizer()
X = vectorizer.fit_transform(emails)

# Train classifier
model = MultinomialNB()
model.fit(X, labels)

# Test prediction
test_email = ["Please update your bank details immediately"]
prediction = model.predict(vectorizer.transform(test_email))
print("Phishing Detected" if prediction[0] == 1 else "Safe Email")

  

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

# Initialize the model
chat = ChatOpenAI(model="gpt-4.5-turbo", temperature=0.7)

# Create a prompt
prompt = ChatPromptTemplate.from_template("Translate the following English text to French: {text}")

# Format and run
response = chat.invoke(prompt.format_messages(text="How are you today?"))
print(response.content)

  

from haystack.nodes import PromptNode
from haystack.pipelines import Pipeline

# Initialize a pipeline with an LLM node
prompt_node = PromptNode("gpt-4.5-turbo", api_key="YOUR_API_KEY")
pipe = Pipeline()
pipe.add_node(component=prompt_node, name="prompt_node", inputs=["Query"])

# Ask a question
result = pipe.run(query="Summarize the document about climate change policies.")
print(result["results"][0])

  

import openai
import pyttsx3

# Initialize text-to-speech
engine = pyttsx3.init()

# Function to get AI response
def get_avatar_response(user_input):
    response = openai.ChatCompletion.create(
        model="gpt-5",
        messages=[{"role": "user", "content": user_input}]
    )
    return response["choices"][0]["message"]["content"]

# Main loop
while True:
    user_input = input("You: ")
    if user_input.lower() in ["exit", "quit"]:
        break
    reply = get_avatar_response(user_input)
    print("Avatar:", reply)
    engine.say(reply)
    engine.runAndWait()