Data Preparation for YOLOv5 CV model in Retail Analytics

				
					import cv2
import os
import json
import pandas as pd
from datetime import datetime, timedelta

# ------------- Helper Functions -------------

def categorize_time_block(hour):
    """Classify the hour into a time block."""
    if 6 <= hour < 12:
        return 'morning'
    elif 12 <= hour < 17:
        return 'afternoon'
    elif 17 <= hour < 21:
        return 'evening'
    else:
        return 'night'

def determine_season(month):
    """Map month to a season."""
    if month in [12, 1, 2]:
        return 'winter'
    elif month in [3, 4, 5]:
        return 'spring'
    elif month in [6, 7, 8]:
        return 'summer'
    else:
        return 'fall'

def check_promotional(timestamp, promo_calendar):
    """Return True if the date is in the promotional calendar."""
    date_str = timestamp.strftime('%Y-%m-%d')
    return date_str in promo_calendar

def parse_timestamp_from_filename(filename):
    """
    Extract timestamp from filename format: storeID_YYYY-MM-DD_HH-MM.mp4
    Example: 'store42_2025-05-10_14-00.mp4'
    """
    parts = filename.replace('.mp4', '').split('_')
    if len(parts) < 3:
        raise ValueError("Filename does not contain timestamp in expected format.")
    timestamp_str = f"{parts[1]}_{parts[2]}"
    return datetime.strptime(timestamp_str, '%Y-%m-%d_%H-%M')

# ------------- Main Function -------------

def extract_frames_with_metadata(video_path, output_dir, sampling_rate=1, store_info=None, promo_calendar=None):
    """
    Extract frames from a video and enrich each with temporal metadata.
    
    Args:
        video_path (str): Path to input video file.
        output_dir (str): Directory to save extracted frames.
        sampling_rate (int): Frames per second to extract.
        store_info (dict): Optional store metadata.
        promo_calendar (set): Set of promotional dates (YYYY-MM-DD).
    
    Returns:
        List of dictionaries, each containing metadata for one frame.
    """
    os.makedirs(output_dir, exist_ok=True)
    cap = cv2.VideoCapture(video_path)

    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_interval = int(fps / sampling_rate)
    frame_index = 0
    saved_index = 0
    metadata_records = []

    # Get base timestamp from video filename
    base_timestamp = parse_timestamp_from_filename(os.path.basename(video_path))

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break

        if frame_index % frame_interval == 0:
            # Save frame image
            frame_filename = f"frame_{saved_index:05d}.jpg"
            frame_path = os.path.join(output_dir, frame_filename)
            cv2.imwrite(frame_path, frame)

            # Estimate timestamp for this frame
            frame_timestamp = base_timestamp + timedelta(seconds=(frame_index / fps))

            # Generate metadata
            metadata = {
                'frame_id': f"{saved_index:05d}",
                'timestamp': frame_timestamp.isoformat(),
                'day_of_week': frame_timestamp.weekday(),
                'time_block': categorize_time_block(frame_timestamp.hour),
                'season': determine_season(frame_timestamp.month),
                'is_promotional': check_promotional(frame_timestamp, promo_calendar or set()),
                'frame_path': frame_path
            }

            if store_info:
                metadata.update({
                    'store_id': store_info.get('store_id'),
                    'store_format': store_info.get('format'),
                    'location_type': store_info.get('location_type'),
                    'department': store_info.get('department')
                })

            metadata_records.append(metadata)
            saved_index += 1

        frame_index += 1

    cap.release()
    return metadata_records

# ------------- Run Script -------------

if __name__ == "__main__":
    # Input video (must include timestamp in filename)
    video_file = "store42_2025-05-10_14-00.mp4"
    output_folder = "frames_output"

    # Example store metadata
    store_metadata = {
        'store_id': 'store42',
        'format': 'medium',
        'location_type': 'urban',
        'department': 'snacks'
    }

    # Example promotional calendar
    promotional_dates = {"2025-05-10", "2025-05-11"}

    # Run extraction
    metadata = extract_frames_with_metadata(
        video_path=video_file,
        output_dir=output_folder,
        sampling_rate=1,  # 1 frame per second
        store_info=store_metadata,
        promo_calendar=promotional_dates
    )

    # Save metadata to CSV
    pd.DataFrame(metadata).to_csv("frame_metadata.csv", index=False)
    print(f"Extracted {len(metadata)} frames and saved metadata to frame_metadata.csv.")