ayomide/smart-farm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
Ayomide
2025-07-22 15:02:17 +01:00
parent d0668af517
commit ee678811b0
4 changed files with 419 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+24 -1
View File
@@ -38,4 +38,27 @@ data/
outputs/
# OS files
.DS_Store
.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
src/main.py
+93
View File
@@ -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()
src/model/smart_farm_tagger.py
+287
View File
@@ -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
src/utils/helpers.py
+15
View File
@@ -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()