diff --git a/.gitignore b/.gitignore index ebf54f5..de46001 100644 --- a/.gitignore +++ b/.gitignore @@ -38,4 +38,27 @@ data/ outputs/ # OS files -.DS_Store \ No newline at end of file +.DS_Store + +# Python +__pycache__/ +*.pyc +*.pyo +*.pyd +.Python +env/ +venv/ +*.egg-info/ +.pytest_cache/ + +# Data +data/processed/ +data/raw/images/ # If images are very large and you manage them separately +data/*.csv # If you don't want to commit generated CSVs +data/*.json + +# Outputs +outputs/ + +# Logs +*.log \ No newline at end of file diff --git a/src/main.py b/src/main.py index e69de29..3826056 100644 --- a/src/main.py +++ b/src/main.py @@ -0,0 +1,93 @@ +import logging +import os +from pathlib import Path +from model.smart_farm_tagger import SmartFarmTagger +from data.data_loader import load_human_keywords, get_image_paths +from utils.helpers import setup_logging + +# Setup logging for the main script +setup_logging(level=logging.INFO) +logger = logging.getLogger(__name__) + +def main(): + """ + Main function to run the Smart Farm Photo Keyword Tagging AI system. + It initializes the tagger, loads images and optional human keywords, + processes the images in a batch, and saves the results. + """ + logger.info("Starting Smart Farm Tagger application...") + + # Initialize the tagger + try: + tagger = SmartFarmTagger() + except Exception as e: + logger.critical(f"Failed to initialize SmartFarmTagger: {e}. Exiting.") + return + + # Setup directories + data_raw_images_dir = Path("data/raw/images") + output_dir = Path("outputs") + + # Ensure output directory exists + output_dir.mkdir(parents=True, exist_ok=True) + + # Get image files + image_paths = get_image_paths(data_raw_images_dir) + + if not image_paths: + logger.warning(f"No images found in {data_raw_images_dir}. Please add images to process.") + # Optionally, create the raw images directory if it doesn't exist + data_raw_images_dir.mkdir(parents=True, exist_ok=True) + print(f"Created directory structure. Please add images to {data_raw_images_dir}") + return + + logger.info(f"Found {len(image_paths)} images to process.") + + # Load human keywords if available for comparison or augmentation + human_keywords_path = "data/human_keywords.csv" + human_keywords_map = load_human_keywords(human_keywords_path) + if human_keywords_map: + logger.info(f"Loaded human keywords for {len(human_keywords_map)} images.") + + # Define output CSV path + results_csv_path = output_dir / "farm_photo_keywords.csv" + + # Process batch of images + results_df = tagger.process_batch( + image_paths=[str(p) for p in image_paths], # Convert Path objects to strings for compatibility + output_path=str(results_csv_path), + human_keywords_map=human_keywords_map + ) + + # Print summary + print("\n" + "="*30) + print("=== Processing Summary ===") + print(f"Total images processed: {len(results_df)}") + if not results_df.empty and 'ai_keywords' in results_df.columns: + # Filter out rows where 'ai_keywords' might be empty lists due to processing errors + valid_keyword_rows = results_df[results_df['ai_keywords'].apply(lambda x: isinstance(x, list) and len(x) > 0)] + if not valid_keyword_rows.empty: + print(f"Average AI keywords per image (for successful tags): {valid_keyword_rows['ai_keywords'].apply(len).mean():.1f}") + else: + print("No AI keywords generated successfully.") + print(f"Results saved to: {results_csv_path}") + print("="*30) + + # Show results + if not results_df.empty: + print("\n=== Sample Results (First 3) ===") + for idx, row in results_df.head(3).iterrows(): + print(f"\nImage: {row['filename']}") + print(f"AI Title: {row['ai_title']}") + print(f"AI Keywords: {', '.join(row['ai_keywords'])}") + if row['location'] and row['location'] != "None": # Check if location is not None or "None" string + print(f"Location: {row['location']}") + if row['human_keywords']: + print(f"Human Keywords: {', '.join(row['human_keywords'])}") + print("="*30) + else: + print("\nNo results to display.") + + +if __name__ == "__main__": + main() diff --git a/src/model/smart_farm_tagger.py b/src/model/smart_farm_tagger.py new file mode 100644 index 0000000..2bea920 --- /dev/null +++ b/src/model/smart_farm_tagger.py @@ -0,0 +1,287 @@ +import logging +from pathlib import Path +from typing import List, Dict, Optional +from datetime import datetime +import pandas as pd +from PIL import Image, ExifTags +import torch +from transformers import BlipProcessor, BlipForConditionalGeneration +import os + +logger = logging.getLogger(__name__) + + +class SmartFarmTagger: + """ + AI system for generating agricultural keywords and titles for farm photos. + It leverages a vision-language model (BLIP) for image captioning and + incorporates an agricultural vocabulary for enhanced keyword generation. + """ + + def __init__(self, model_name: str = "Salesforce/blip-image-captioning-base"): + """ + Initialize the tagger with a vision-language model and an agricultural vocabulary. + + Args: + model_name (str): The name of the pre-trained BLIP model to use. + """ + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + logger.info(f"Using device: {self.device}") + + # Load vision-language model for image understanding + try: + self.processor = BlipProcessor.from_pretrained(model_name) + self.model = BlipForConditionalGeneration.from_pretrained(model_name).to(self.device) + logger.info(f"Successfully loaded model: {model_name}") + except Exception as e: + logger.error(f"Failed to load BLIP model {model_name}: {e}") + raise + + # Agricultural vocabulary for enhanced keyword generation + self.ag_vocabulary = { + 'crops': ['corn', 'wheat', 'soybeans', 'cotton', 'rice', 'barley', 'oats', 'hay', 'alfalfa', 'sunflower', 'vegetables', 'fruit'], + 'livestock': ['cattle', 'cows', 'pigs', 'chickens', 'sheep', 'goats', 'horses', 'dairy', 'poultry'], + 'people': ['farmer', 'rancher', 'agricultural worker', 'farm family', 'male farmer', 'female farmer'], + 'equipment': ['tractor', 'combine', 'harvester', 'plow', 'planting equipment', 'irrigation', 'sprayer', 'truck'], + 'activities': ['planting', 'harvesting', 'feeding', 'milking', 'irrigation', 'cultivation', 'tilling', 'sowing'], + 'locations': ['farm', 'ranch', 'field', 'pasture', 'barn', 'silo', 'greenhouse', 'feedlot', 'orchard', 'vineyard'], + 'seasons': ['spring planting', 'summer growth', 'fall harvest', 'winter preparation'], + 'concepts': ['sustainable farming', 'organic agriculture', 'precision farming', 'livestock management', 'agribusiness', 'rural life'] + } + + def extract_location_from_exif(self, image_path: str) -> Optional[str]: + """ + Extract GPS location from image EXIF data. + Note: This is a simplified implementation. In a production environment, + you'd typically integrate with a geocoding service to convert coordinates + into human-readable addresses. + + Args: + image_path (str): The path to the image file. + + Returns: + Optional[str]: A string indicating GPS data is available, or None if not found/error. + """ + try: + with Image.open(image_path) as img: + exif = img.getexif() + if not exif: + return None + + for tag_id in exif: + tag = ExifTags.TAGS.get(tag_id, tag_id) + if tag == "GPSInfo": + return "GPS_Location_Available" + + except Exception as e: + logger.warning(f"Could not extract EXIF from {image_path}: {e}") + return None + + def generate_caption(self, image: Image.Image) -> str: + """ + Generate a descriptive caption for the given image using the BLIP model. + + Args: + image (Image.Image): The PIL Image object to caption. + + Returns: + str: The generated image caption. Defaults to "Agricultural scene" on failure. + """ + try: + inputs = self.processor(images=image, return_tensors="pt").to(self.device) + + with torch.no_grad(): + out = self.model.generate(**inputs, max_length=50, num_beams=5) + + caption = self.processor.decode(out[0], skip_special_tokens=True) + return caption.strip() + + except Exception as e: + logger.error(f"Caption generation failed: {e}") + return "Agricultural scene" + + def extract_agricultural_keywords(self, caption: str, image_context: str = "") -> List[str]: + """ + Extract agricultural keywords from the generated caption and optional image context. + This method uses a predefined agricultural vocabulary and common patterns. + + Args: + caption (str): The generated caption for the image. + image_context (str): Additional context about the image (e.g., from metadata). + + Returns: + List[str]: A list of relevant agricultural keywords, typically 5-10. + """ + keywords = set() + caption_lower = caption.lower() + context_lower = image_context.lower() + full_text = f"{caption_lower} {context_lower}" + + # Extract keywords from agricultural vocabulary + for category, terms in self.ag_vocabulary.items(): + for term in terms: + if term in full_text: + keywords.add(term) + + # Additional keyword extraction based on common patterns + agricultural_terms = [ + 'agriculture', 'farming', 'farm', 'rural', 'countryside', + 'field', 'crop', 'harvest', 'plant', 'grow', 'livestock', + 'barn', 'tractor', 'equipment', 'organic', 'sustainable', + 'produce', 'cultivate', 'acre', 'soil', 'irrigation', 'greenhouse' + ] + + for term in agricultural_terms: + if term in full_text: + keywords.add(term) + + # Gender-specific farmer detection + if ('man' in full_text or 'male' in full_text) and any(x in full_text for x in ['farmer', 'rancher', 'agricultural worker']): + keywords.add('male farmer') + + if ('woman' in full_text or 'female' in full_text) and any(x in full_text for x in ['farmer', 'rancher', 'agricultural worker']): + keywords.add('female farmer') + + keyword_list = list(keywords) + + # Ensure at least 5 keywords + if len(keyword_list) < 5: + generic_keywords = ['agriculture', 'farming', 'rural', 'countryside', 'agricultural scene'] + for kw in generic_keywords: + if kw not in keyword_list: + keyword_list.append(kw) + if len(keyword_list) >= 5: + break + + return sorted(list(set(keyword_list)))[:10] # Limit to 10 unique keywords and sort + + def generate_title(self, caption: str, keywords: List[str]) -> str: + """ + Generate a descriptive product title based on the caption and keywords. + + Args: + caption (str): The generated image caption. + keywords (List[str]): A list of extracted keywords. + + Returns: + str: A concise title for the image. + """ + if not caption: + return "Agricultural Scene Photo" + + # Capitalize first letter and clean up + title = caption.strip() + if title: + title = title[0].upper() + title[1:] + + # Add location context if available in keywords + location_keywords = [kw for kw in keywords if any(loc in kw.lower() for loc in ['field', 'farm', 'barn', 'pasture', 'greenhouse'])] + if location_keywords and len(title) < 70: + title += f" in {location_keywords[0]}" + + # Ensure it ends with a descriptive term if short + if len(title) < 20 and not any(term in title.lower() for term in ['photo', 'image', 'scene']): + title += " Photo" + + return title[:100].strip() # Limit title length and remove trailing spaces + + def process_single_image(self, image_path: str, human_keywords: Optional[List[str]] = None) -> Dict: + """ + Process a single image to generate AI-driven tags (caption, keywords, title) + and extract EXIF location data. + + Args: + image_path (str): The file path to the image. + human_keywords (Optional[List[str]]): Optional list of human-generated keywords for comparison/inclusion. + + Returns: + Dict: A dictionary containing all generated information for the image. + Includes filename, human and AI keywords, AI title, location, caption, and processing timestamp. + """ + human_keywords = human_keywords or [] + try: + # Load image + with Image.open(image_path) as img: + # Convert to RGB + if img.mode != 'RGB': + img = img.convert('RGB') + + # Generate caption + caption = self.generate_caption(img) + logger.info(f"Generated caption for {Path(image_path).name}: {caption}") + + # Extract location from EXIF + location = self.extract_location_from_exif(image_path) + + # Generate keywords + ai_keywords = self.extract_agricultural_keywords(caption) + + # Generate title + ai_title = self.generate_title(caption, ai_keywords) + + result = { + 'filename': Path(image_path).name, + 'human_keywords': human_keywords, + 'ai_keywords': ai_keywords, + 'ai_title': ai_title, + 'location': location, + 'caption': caption, + 'processed_at': datetime.now().isoformat() + } + + logger.info(f"Processed {result['filename']}: Generated {len(ai_keywords)} keywords and title '{ai_title}'") + return result + + except Exception as e: + logger.error(f"Error processing {image_path}: {e}") + return { + 'filename': Path(image_path).name, + 'human_keywords': human_keywords, + 'ai_keywords': [], + 'ai_title': 'Processing Error', + 'location': None, + 'caption': 'Error during processing', + 'processed_at': datetime.now().isoformat(), + 'error': str(e) + } + + def process_batch(self, image_paths: List[str], output_path: str, + human_keywords_map: Optional[Dict[str, List[str]]] = None) -> pd.DataFrame: + """ + Process a batch of images and save the results to a CSV file. + + Args: + image_paths (List[str]): A list of file paths to the images to process. + output_path (str): The file path where the results CSV will be saved. + human_keywords_map (Optional[Dict[str, List[str]]]): A dictionary mapping filenames + to human-generated keywords. + + Returns: + pd.DataFrame: A DataFrame containing the processing results for all images. + """ + results = [] + human_keywords_map = human_keywords_map or {} + + logger.info(f"Starting batch processing for {len(image_paths)} images...") + + for i, image_path in enumerate(image_paths): + filename = Path(image_path).name + human_keywords = human_keywords_map.get(filename, []) + + result = self.process_single_image(image_path, human_keywords) + results.append(result) + + if (i + 1) % 10 == 0: + logger.info(f"Processed {i + 1}/{len(image_paths)} images so far.") + + # Convert to DataFrame for easy saving and analysis + df = pd.DataFrame(results) + + # Create output directory if it doesn't exist + os.makedirs(Path(output_path).parent, exist_ok=True) + + # Save to CSV + df.to_csv(output_path, index=False) + logger.info(f"Batch processing complete. Results saved to {output_path}") + + return df diff --git a/src/utils/helpers.py b/src/utils/helpers.py new file mode 100644 index 0000000..b18e399 --- /dev/null +++ b/src/utils/helpers.py @@ -0,0 +1,15 @@ +import logging + + +def setup_logging(level=logging.INFO): + """ + Logging configuration. + """ + logging.basicConfig(level=level, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') + logger = logging.getLogger(__name__) + logger.info("Logging configured.") + + +def clean_string(text: str) -> str: + """Removes extra spaces and makes string lowercase.""" + return ' '.join(text.split()).lower() \ No newline at end of file