Script vibe coded mostly with Claude Sonnet 4.5. Creates a mosaic of the year, a bit like a keogram from a directory of images with timestamp file names.

image_of_the_year.py

#!/usr/bin/env python3

'''
MIT License

Copyright (c) 2026 James Wheare

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
'''

import os
import re
from datetime import datetime, timedelta
from PIL import Image
from concurrent.futures import ThreadPoolExecutor, as_completed
import numpy as np
from zoneinfo import ZoneInfo

# ---------------- CONFIG ----------------
# Final output size
DEFAULT_OUTPUT_WIDTH = 1890          # e.g. 10px per day-hour
DEFAULT_ROW_HEIGHT = 6              # height of each day-row in pixels

# Day boundary time (24-hour format)
DAY_START_HOUR = 15          # 3pm = 15:00

# Performance settings
DEFAULT_MAX_WORKERS = 8              # Number of parallel threads for image loading

# prefix-20250103000017.jpg
# blah_2025-09-05_074414.png
TIMESTAMP_REGEX = re.compile(r"(\d{4})-?(\d{2})-?(\d{2})_?(\d{2}\d{2}\d{2})")

# ----------------------------------------

def extract_datetime(filename, tz):
    """Extract datetime from filename."""
    match = TIMESTAMP_REGEX.search(filename)
    if not match:
        return None
    y = match.group(1)
    m = match.group(2)
    d = match.group(3)
    t = match.group(4)
    dt = datetime.strptime(f'{y}{m}{d}{t}', "%Y%m%d%H%M%S")
    
    if tz is not None:
        # Treat as local time and convert to fixed offset (standard time)
        local_dt = dt.replace(tzinfo=ZoneInfo(tz))
        dt = (local_dt - (local_dt.dst() or timedelta())).replace(tzinfo=None)
    
    return dt

def get_custom_day_key(dt, start_hour):
    """
    Return a 'custom day' key based on start_hour.
    If time is before start_hour, it belongs to the previous day's row.
    """
    if dt.hour < start_hour:
        # Before start_hour, so belongs to previous day
        return (dt.date() - timedelta(days=1))
    else:
        return dt.date()

def detect_directory_structure(input_dir, start_year):
    """
    Detect directory structure and return mapping of dates to directories.
    Returns: (format_type, date_to_dir_map) where format_type is 'y-m-d' or 'y/m/d'
    """
    print("Detecting directory structure...")
    
    # Only look at the 3 relevant years
    relevant_years = [start_year - 1, start_year, start_year + 1]
    
    date_to_dir = {}
    format_type = None
    
    for year_num in relevant_years:
        year_name = str(year_num)
        year_path = os.path.join(input_dir, year_name)
        
        if not os.path.isdir(year_path):
            continue
        
        # Check what format this year uses
        subdirs = [d for d in os.listdir(year_path) if os.path.isdir(os.path.join(year_path, d))]
        
        if not subdirs:
            continue
        
        # Detect format from first subdir
        sample_subdir = subdirs[0]
        
        if '-' in sample_subdir:
            # Format: y/y-m-d/
            if format_type is None:
                format_type = 'y-m-d'
                print("  Detected y/y-m-d/ format")
            
            # For prev/next year, only map the specific date we need
            if year_num == start_year - 1:
                target_dates = [f"{year_num}-12-31"]
            elif year_num == start_year + 1:
                target_dates = [f"{year_num}-01-01"]
            else:
                target_dates = subdirs
            
            for date_name in target_dates:
                date_path = os.path.join(year_path, date_name)
                if not os.path.isdir(date_path):
                    continue
                
                try:
                    parts = date_name.split('-')
                    if len(parts) != 3:
                        continue
                    y, m, d = int(parts[0]), int(parts[1]), int(parts[2])
                    date_key = datetime(y, m, d).date()
                    date_to_dir[date_key] = date_path
                except (ValueError, IndexError):
                    continue
        
        else:
            # Format: y/m/d/
            if format_type is None:
                format_type = 'y/m/d'
                print("  Detected y/m/d/ format")
            
            if year_num == start_year - 1:
                target_month_days = [("12", "31")]
            elif year_num == start_year + 1:
                target_month_days = [("01", "01")]
            else:
                target_month_days = []
                for month_name in subdirs:
                    if not month_name.isdigit():
                        continue
                    month_path = os.path.join(year_path, month_name)
                    if not os.path.isdir(month_path):
                        continue
                    day_dirs = [
                        d for d in os.listdir(month_path) 
                        if os.path.isdir(os.path.join(month_path, d)) and d.isdigit()
                    ]
                    for day_name in day_dirs:
                        target_month_days.append((month_name, day_name))
            
            for month_name, day_name in target_month_days:
                day_path = os.path.join(year_path, month_name, day_name)
                if not os.path.isdir(day_path):
                    continue
                
                try:
                    m, d = int(month_name), int(day_name)
                    date_key = datetime(year_num, m, d).date()
                    date_to_dir[date_key] = day_path
                except (ValueError, IndexError):
                    continue
    
    print(f"Found {len(date_to_dir)} date directories")
    return format_type, date_to_dir

def load_images_for_day(day_dir, start_hour, day_date, tz):
    """
    Load image list for a specific day directory.
    Returns list of (datetime, path) tuples.
    """
    if not os.path.isdir(day_dir):
        return []
    
    images = []
    files = [f for f in os.listdir(day_dir) if f.lower().endswith(".jpg")]
    
    for fname in files:
        dt = extract_datetime(fname, tz)
        if dt is None:
            continue
        day_key = get_custom_day_key(dt, start_hour)
        # Only include if it matches the expected day
        if day_key == day_date:
            full_path = os.path.join(day_dir, fname)
            images.append((dt, full_path))
    
    images.sort(key=lambda x: x[0])
    return images

def load_and_group_images(input_dir, start_hour, start_year):
    """Fast loading for structured directories. Supports both y/m/d/ and y/y-m-d/ formats."""
    # Returns date -> directory mapping for lazy loading
    format_type, date_to_dir = detect_directory_structure(input_dir, start_year)
    
    if not date_to_dir:
        print("No valid date directories found")
        return {}
    
    # Return the mapping - we'll load files on-demand
    return date_to_dir

def load_image_strip(path, row_height, crop_region, crop_pixels_top, crop_pixels_bottom):
    """
    Load and resize a single image to a vertical strip.
    crop_region: tuple of (top_fraction, bottom_fraction) to crop vertically
                 e.g., (0, 0.5) = top 50%, (0.25, 0.75) = middle 50%
    crop_pixels_top: number of pixels to remove from top (e.g., 30 for black bar)
    crop_pixels_bottom: number of pixels to remove from bottom
    """
    try:
        with Image.open(path) as img:
            img = img.convert("RGB")
            width, height = img.size
            
            # Combine both crop operations into a single crop
            top_px = crop_pixels_top
            bottom_px = height - crop_pixels_bottom
            
            # Apply fractional crop if specified
            if crop_region is not None:
                top_fraction, bottom_fraction = crop_region
                # Calculate on the region after pixel crop
                remaining_height = height - crop_pixels_top - crop_pixels_bottom
                top_px = crop_pixels_top + int(remaining_height * top_fraction)
                bottom_px = height - crop_pixels_bottom - int(remaining_height * (1 - bottom_fraction))
            
            # Single crop operation if needed
            if top_px > 0 or bottom_px < height:
                img = img.crop((0, top_px, width, bottom_px))
            
            # Scale to row_height maintaining aspect ratio
            aspect_ratio = img.size[0] / img.size[1]
            new_width = int(row_height * aspect_ratio)
            img = img.resize((new_width, row_height), Image.BILINEAR)
            
            return np.array(img)
    except Exception as e:
        print(f"Error loading {path}: {e}")
        return None

def build_day_row(
    image_list, row_start, row_end, row_height, output_width,
    crop_region, crop_pixels_top, crop_pixels_bottom,
    max_workers, strip_width
):
    """
    Create a single-row image representing one 24-hour period with time-aligned sampling.
    image_list: list of (datetime, path) tuples (already sorted)
    row_start, row_end: datetime objects defining the 24-hour window
    crop_region: tuple of (top_fraction, bottom_fraction) for vertical cropping
    crop_pixels_top: number of pixels to remove from top before other operations
    crop_pixels_bottom: number of pixels to remove from bottom before other operations
    """
    # Filter images within the time window
    valid_images = [(dt, path) for dt, path in image_list if row_start <= dt < row_end]
    
    if not valid_images:
        # Return blank row if no images
        return np.full((row_height, output_width, 3), 128, dtype=np.uint8)
    
    # Process first image to determine strip width
    first_strip = load_image_strip(
        valid_images[0][1],
        row_height,
        crop_region,
        crop_pixels_top,
        crop_pixels_bottom
    )
    
    if first_strip is None:
        return np.full((row_height, output_width, 3), 128, dtype=np.uint8)
    
    # Calculate how many time slots we have
    num_slots = output_width // strip_width
    
    if num_slots == 0:
        num_slots = 1
    
    # Calculate time slot duration
    total_seconds = (row_end - row_start).total_seconds()
    slot_duration = timedelta(seconds=total_seconds / num_slots)
    tolerance = slot_duration / 2  # Accept images within half a slot
    
    # Create time slots
    time_slots = [row_start + i * slot_duration for i in range(num_slots)]
    
    # Find closest image for each time slot using two-pointer technique
    slot_to_path = {}
    img_idx = 0  # Pointer for valid_images
    
    for slot_idx, slot_time in enumerate(time_slots):
        closest_image = None
        closest_diff = None
        
        # Move pointer forward to images that could match this slot
        while img_idx < len(valid_images) and valid_images[img_idx][0] < slot_time - tolerance:
            img_idx += 1
        
        # Check images near this time slot
        check_idx = img_idx
        while check_idx < len(valid_images):
            dt, path = valid_images[check_idx]
            diff = abs((dt - slot_time).total_seconds())
            
            # If we've moved past the tolerance window, stop searching
            if dt > slot_time + tolerance:
                break
            
            # Check if within tolerance
            if diff <= tolerance.total_seconds():
                if closest_diff is None or diff < closest_diff:
                    closest_diff = diff
                    closest_image = path
            
            check_idx += 1
        
        if closest_image:
            slot_to_path[slot_idx] = closest_image
    
    # Load unique images in parallel
    unique_paths = set(slot_to_path.values())
    path_to_strip = {}
    
    with ThreadPoolExecutor(max_workers) as executor:
        future_to_path = {
            executor.submit(
                load_image_strip,
                path,
                row_height,
                crop_region,
                crop_pixels_top,
                crop_pixels_bottom
            ): path 
            for path in unique_paths
        }
        
        for future in as_completed(future_to_path):
            path = future_to_path[future]
            result = future.result()
            if result is not None:
                path_to_strip[path] = result
    
    # Assemble the row from time slots
    row_array = np.full((row_height, output_width, 3), 128, dtype=np.uint8)
    x_offset = 0
    
    for slot_idx in range(num_slots):
        if slot_idx in slot_to_path:
            path = slot_to_path[slot_idx]
            if path in path_to_strip:
                strip = path_to_strip[path]
                strip_w = strip.shape[1]
                
                # Don't overflow the output width
                if x_offset + strip_w > output_width:
                    strip_w = output_width - x_offset
                    strip = strip[:, :strip_w, :]
                
                row_array[:, x_offset:x_offset + strip_w, :] = strip
        # else: leave blank (already zeros)
        
        x_offset += strip_width
        
        if x_offset >= output_width:
            break
    
    return row_array

def main(
    input_dir, output_path,
    output_width, row_height,
    start_year, start_hour, tz,
    max_workers,
    crop_region, crop_pixels_top, crop_pixels_bottom, dry_run
):
    """
    Generate year visualization from start_year Jan 1 to following year Jan 1.
    Each row runs from start_hour on previous day to start_hour on current day.
    """
    if dry_run:
        print("=== DRY RUN MODE - No images will be processed or written ===")
    
    if crop_pixels_top > 0:
        print(f"Removing {crop_pixels_top}px from top of each image")
    if crop_pixels_bottom > 0:
        print(f"Removing {crop_pixels_bottom}px from bottom of each image")
    if crop_region:
        print(f"Using crop region: {crop_region[0]*100:.0f}% to {crop_region[1]*100:.0f}% of image height (after pixel crop)")
    
    images_by_day = load_and_group_images(input_dir, start_hour, start_year)
    
    if not images_by_day:
        raise RuntimeError("No valid images found.")
    
    # Generate all days from Dec 31 (prev year) to Jan 1 (next year)
    start_date = datetime(start_year - 1, 12, 31).date()
    end_date = datetime(start_year + 1, 1, 1).date()
    
    all_days = []
    current = start_date
    while current <= end_date:
        all_days.append(current)
        current += timedelta(days=1)
    
    print(f"\nPlanned output: {output_width}x{row_height * len(all_days)} pixels")
    print(f"Total rows: {len(all_days)} (from {all_days[0]} to {all_days[-1]})")
    
    # Show statistics
    days_with_images = len(images_by_day)
    print(f"Days with directories: {days_with_images}")
    
    days_without_images = len(all_days) - days_with_images
    print(f"Days without images (will be blank): {days_without_images}")
    
    if dry_run:
        print("\n=== DRY RUN SUMMARY ===")
        print(f"Would process {len(all_days)} days")
        print(f"Would save to: {output_path}")
        
        print("\nLazy loading mode - image counts not available in dry-run")
        print(f"Available date directories: {len(images_by_day)}")
        
        print("\n=== DRY RUN COMPLETE - No files written ===")
        return
    
    # Pre-allocate the full image array
    print(f"\nCreating output image: {output_width}x{row_height * len(all_days)}")
    year_array = np.zeros((row_height * len(all_days), output_width, 3), dtype=np.uint8)
    
    # Determine strip width from first available image
    strip_width = None
    for day in all_days:
        day_dir = images_by_day.get(day)
        if day_dir:
            test_images = load_images_for_day(day_dir, start_hour, day, tz)
            if test_images:
                test_strip = load_image_strip(
                    test_images[0][1],
                    row_height,
                    crop_region,
                    crop_pixels_top,
                    crop_pixels_bottom
                )
                if test_strip is not None:
                    strip_width = test_strip.shape[1]
                    print(f"Detected strip width: {strip_width}px (from {day})")
                    break
    
    if strip_width is None:
        raise RuntimeError("Could not determine strip width - no valid images found")
    
    num_slots = output_width // strip_width
    print(f"Using {num_slots} time slots per day")
    
    # Process each day
    for idx, day in enumerate(all_days):
        # Each row represents start_hour on 'day' to start_hour on 'day+1'
        row_start = datetime.combine(day, datetime.min.time()) + timedelta(hours=start_hour)
        row_end = row_start + timedelta(hours=24)
        
        # Load images on-demand from directory
        images = []

        # Current day directory
        day_dir = images_by_day.get(day)
        if day_dir:
            images.extend(load_images_for_day(day_dir, start_hour, day, tz))

        # Next day directory (for early hours)
        next_day = day + timedelta(days=1)
        next_day_dir = images_by_day.get(next_day)
        if next_day_dir:
            images.extend(load_images_for_day(next_day_dir, start_hour, day, tz))

        if (idx + 1) % 10 == 0 or idx == 0:
            print(f"\nProcessing day {idx + 1}/{len(all_days)}: {day} - {len(images)} images")
        else:
            print('.', end='', flush=True)
        
        day_row = build_day_row(
            images, row_start, row_end, row_height, output_width,
            crop_region, crop_pixels_top, crop_pixels_bottom,
            max_workers, strip_width
        )
        
        # Insert into the output array
        y_start = idx * row_height
        y_end = y_start + row_height
        year_array[y_start:y_end, :, :] = day_row
    
    # Convert to PIL and save
    print("\nConverting to image and saving...")
    year_image = Image.fromarray(year_array, mode='RGB')
    
    # Auto-detect output format based on file extension
    if output_path.lower().endswith('.png'):
        year_image.save(output_path, format='PNG', compress_level=0)
        print(f"Saved {output_path} (PNG, uncompressed)")
    else:
        year_image.save(output_path, quality=95)
        print(f"Saved {output_path} (JPEG, quality=95)")
    
    print(f"Total rows: {len(all_days)} (from {all_days[0]} to {all_days[-1]})")

if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(
        description='Generate a year visualization from timelapse images with custom day boundaries.',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --input ./images --output year.jpg --year 2025 --dry-run
  %(prog)s --input ./images --output year.jpg --year 2025 --start-hour 18
  %(prog)s -i ./images -o year.jpg -y 2025 --crop-pixels-top 30
  %(prog)s -i ./images -o year.jpg -y 2025 --crop-pixels-top 30 --crop-top 0 --crop-bottom 0.5
        """
    )
    
    parser.add_argument(
        '-i', '--input',
        required=True,
        help='Input directory containing timelapse images (will scan nested folders)'
    )
    
    parser.add_argument(
        '-o', '--output',
        required=True,
        help='Output path for the generated image (e.g., year.jpg)'
    )
    
    parser.add_argument(
        '-y', '--year',
        type=int,
        required=True,
        help='Starting year for the visualization'
    )
    
    parser.add_argument(
        '-s', '--start-hour',
        type=int,
        default=DAY_START_HOUR,
        metavar='HOUR',
        help=f'Hour (0-23) when each day starts (default: {DAY_START_HOUR} for 3pm)'
    )
    
    parser.add_argument(
        '--width',
        type=int,
        default=DEFAULT_OUTPUT_WIDTH,
        metavar='PIXELS',
        help=f'Output image width in pixels (default: {DEFAULT_OUTPUT_WIDTH})'
    )
    
    parser.add_argument(
        '--row-height',
        type=int,
        default=DEFAULT_ROW_HEIGHT,
        metavar='PIXELS',
        help=f'Height of each day row in pixels (default: {DEFAULT_ROW_HEIGHT})'
    )
    
    parser.add_argument(
        '--workers',
        type=int,
        default=DEFAULT_MAX_WORKERS,
        metavar='N',
        help=f'Number of parallel threads for image loading (default: {DEFAULT_MAX_WORKERS})'
    )
    
    parser.add_argument(
        '--crop-pixels-top',
        type=int,
        default=0,
        metavar='PIXELS',
        help='Remove this many pixels from top of each image (e.g., 30 for black bar)'
    )
    
    parser.add_argument(
        '--crop-pixels-bottom',
        type=int,
        default=0,
        metavar='PIXELS',
        help='Remove this many pixels from bottom of each image'
    )
    
    parser.add_argument(
        '--crop-top',
        type=float,
        metavar='FRACTION',
        help='Top of crop region as fraction (0.0-1.0), applied after pixel crop. Use with --crop-bottom'
    )
    
    parser.add_argument(
        '--crop-bottom',
        type=float,
        metavar='FRACTION',
        help='Bottom of crop region as fraction (0.0-1.0). E.g., --crop-top 0 --crop-bottom 0.5 = top 50%% (after pixel crop)'
    )
    parser.add_argument(
        '--dry-run',
        action='store_true',
        help='Scan directories and show statistics without processing images or writing output'
    )
    parser.add_argument(
        '--timezone',
        type=str,
        default=None,
        help='Optional IANA timezone (e.g., Europe/London) to interpret filenames as local time and normalize to standard time'
    )
    
    args = parser.parse_args()
    
    # Validate start hour
    if not 0 <= args.start_hour <= 23:
        parser.error("start-hour must be between 0 and 23")
    
    # Validate and parse crop region
    crop_region = None
    if args.crop_top is not None or args.crop_bottom is not None:
        if args.crop_top is None or args.crop_bottom is None:
            parser.error("Both --crop-top and --crop-bottom must be specified together")
        if not (0 <= args.crop_top < args.crop_bottom <= 1):
            parser.error("Crop region must satisfy: 0 <= crop-top < crop-bottom <= 1")
        crop_region = (args.crop_top, args.crop_bottom)
    
    main(
        args.input,
        args.output,
        args.width,
        args.row_height,
        args.year,
        args.start_hour,
        args.timezone,
        args.workers,
        crop_region,
        args.crop_pixels_top,
        args.crop_pixels_bottom,
        args.dry_run
    )