Extract insights from beneficiary photos

Analyse photos with Cloud Vision API

This extracts activity insights from photo libraries. Requires Google Cloud Vision API credentials.

import os
import pandas as pd
from google.cloud import vision
from pathlib import Path
import json
from datetime import datetime

# IMPORTANT: Set up authentication first
# export GOOGLE_APPLICATION_CREDENTIALS="path/to/credentials.json"

client = vision.ImageAnnotatorClient()

def analyse_image(image_path):
    """
    Analyse a single image using Cloud Vision API
    Returns detected labels, objects, and text
    """
    with open(image_path, 'rb') as image_file:
        content = image_file.read()

    image = vision.Image(content=content)

    # Get label detection (what's in the image)
    label_response = client.label_detection(image=image)
    labels = [(label.description, label.score) for label in label_response.label_annotations]

    # Get object detection (specific items)
    object_response = client.object_localization(image=image)
    objects = [(obj.name, obj.score) for obj in object_response.localized_object_annotations]

    # Detect text if any (for context like activity names on boards)
    text_response = client.text_detection(image=image)
    texts = [text.description for text in text_response.text_annotations[:5]] if text_response.text_annotations else []

    return {
        'labels': labels[:10],  # Top 10 labels
        'objects': objects,
        'text': texts
    }

def classify_activity(labels, objects):
    """
    Map detected labels/objects to activity categories
    Adapt these mappings to your organisation's activities
    """
    label_text = ' '.join([l[0].lower() for l in labels])
    object_text = ' '.join([o[0].lower() for o in objects])
    combined = label_text + ' ' + object_text

    # Activity type detection
    activity_type = None
    if any(word in combined for word in ['football', 'basketball', 'sport', 'game', 'team sport']):
        activity_type = 'Sports'
    elif any(word in combined for word in ['art', 'paint', 'drawing', 'craft', 'creative']):
        activity_type = 'Arts & Crafts'
    elif any(word in combined for word in ['music', 'musical instrument', 'singing']):
        activity_type = 'Music'
    elif any(word in combined for word in ['cooking', 'food', 'kitchen']):
        activity_type = 'Cooking'
    elif any(word in combined for word in ['computer', 'laptop', 'technology']):
        activity_type = 'Digital Skills'
    elif any(word in combined for word in ['reading', 'book', 'education']):
        activity_type = 'Learning'
    else:
        activity_type = 'General Activity'

    # Setting detection
    setting = 'Indoor'  # Default
    if any(word in combined for word in ['outdoor', 'park', 'field', 'grass', 'sky']):
        setting = 'Outdoor'

    # Engagement indicators
    group_activity = any(word in combined for word in ['group', 'team', 'crowd', 'people'])

    # Estimate participant count from objects
    person_objects = [o for o in objects if 'person' in o[0].lower()]
    estimated_participants = len(person_objects) if person_objects else None

    return {
        'activity_type': activity_type,
        'setting': setting,
        'group_activity': group_activity,
        'estimated_participants': estimated_participants
    }

# Process photo library
photo_directory = 'path/to/photos'  # Change this
results = []

print(f"Scanning photos in {photo_directory}...")

# Walk through all images in directory
image_extensions = {'.jpg', '.jpeg', '.png'}
image_paths = []

for root, dirs, files in os.walk(photo_directory):
    for file in files:
        if Path(file).suffix.lower() in image_extensions:
            image_paths.append(os.path.join(root, file))

print(f"Found {len(image_paths)} images")
print(f"Cost estimate: £{len(image_paths) * 0.0015:.2f} (after free tier)")
print("\nProcessing (this may take a while)...")

for idx, image_path in enumerate(image_paths):
    try:
        # Extract metadata from path
        rel_path = os.path.relpath(image_path, photo_directory)
        path_parts = Path(rel_path).parts

        # Try to extract date and activity from folder structure
        # Adapt this to your folder naming convention
        folder_date = None
        folder_activity = None

        for part in path_parts:
            # Try to find date (e.g., "2024" or "01_January")
            if '2023' in part or '2024' in part:
                folder_date = part
            # Other parts might be activity names
            elif part not in ['.', '..'] and not part.endswith(('.jpg', '.jpeg', '.png')):
                folder_activity = part.replace('_', ' ')

        # Analyse image
        analysis = analyse_image(image_path)

        # Classify activity
        classification = classify_activity(analysis['labels'], analysis['objects'])

        # Combine results
        result = {
            'file_path': image_path,
            'folder_date': folder_date,
            'folder_activity': folder_activity,
            'activity_type': classification['activity_type'],
            'setting': classification['setting'],
            'group_activity': classification['group_activity'],
            'estimated_participants': classification['estimated_participants'],
            'top_labels': ', '.join([f"{l[0]} ({l[1]:.2f})" for l in analysis['labels'][:5]]),
            'detected_objects': ', '.join([o[0] for o in analysis['objects']]),
            'detected_text': ', '.join(analysis['text'][:3]) if analysis['text'] else ''
        }

        results.append(result)

        # Progress update
        if (idx + 1) % 50 == 0:
            print(f"  Processed {idx + 1}/{len(image_paths)} images...")

        # Save periodically in case of interruption
        if (idx + 1) % 100 == 0:
            temp_df = pd.DataFrame(results)
            temp_df.to_csv('photo_analysis_temp.csv', index=False)

    except Exception as e:
        print(f"  Error processing {image_path}: {e}")
        continue

# Create final dataframe
df = pd.DataFrame(results)
df.to_csv('photo_analysis_results.csv', index=False)

print(f"\nAnalysis complete! Processed {len(df)} images")
print("\nResults Summary:")
print("="*60)

# Activity type breakdown
print("\nActivity Types:")
activity_counts = df['activity_type'].value_counts()
for activity, count in activity_counts.items():
    print(f"  {activity}: {count} photos ({count/len(df)*100:.1f}%)")

# Setting breakdown
print("\nSettings:")
setting_counts = df['setting'].value_counts()
for setting, count in setting_counts.items():
    print(f"  {setting}: {count} photos ({count/len(df)*100:.1f}%)")

# Group activity prevalence
group_pct = df['group_activity'].sum() / len(df) * 100
print(f"\nGroup Activities: {group_pct:.1f}% of photos")

# Average participants (where detected)
participants_data = df[df['estimated_participants'].notna()]
if len(participants_data) > 0:
    avg_participants = participants_data['estimated_participants'].mean()
    print(f"\nAverage Participants (when detected): {avg_participants:.1f}")

# Activity type by setting
print("\nActivity Types by Setting:")
activity_setting = df.groupby(['activity_type', 'setting']).size().unstack(fill_value=0)
print(activity_setting)

print("\n" + "="*60)
print("Results saved to photo_analysis_results.csv")
print("\nUse this data to:")
print("  - Search photos by activity type or setting")
print("  - Analyse participation patterns over time")
print("  - Find examples for reports and communications")
print("  - Understand which activities are most common")
print("\nREMEMBER:")
print("  - This data contains information about beneficiaries")
print("  - Implement appropriate access controls")
print("  - Delete when no longer needed")
print("  - Use only for purposes covered by consent")