nasa_epic_images

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Filename: nasa_epic_images.py
# Version: 1.0.0
# Author: Jeoi Reqi

"""
Description:
    This script downloads images from NASA's EPIC (Earth Polychromatic Imaging Camera) satellite.
    EPIC is a camera onboard the NOAA's DSCOVR (Deep Space Climate Observatory) spacecraft,
    positioned at the Lagrange point 1 (L1), approximately one million miles from Earth.
    The camera captures images of the entire sunlit side of Earth, providing a unique perspective on our planet.

    The EPIC images are available in four different types:
        - Natural: True-color images that show Earth as it appears to the human eye.
        - Enhanced: Images that are adjusted to enhance specific features of the atmosphere and surface.
        - Aerosol: Images that highlight the distribution and concentration of aerosols in the atmosphere.
        - Cloud: Images that provide detailed views of cloud formations and their dynamics.

    Users can specify a date range to download images for, and the images will be saved in the specified folder.

Requirements:
    - Python 3.x
    - Required modules:
        - requests
        - datetime
        - concurrent.futures
        - time
        - colorama
Functions:
        - Main:
            - download_images_for_date:
                Checks if images are available for a specific date and option, and downloads them if available.
            - download_image:
                Downloads a single image from the specified source URL and saves it to the destination folder.
            - get_valid_start_date:
                Prompts the user to enter a valid start date after the epoch date for the selected option.
            - get_valid_end_date:
                Prompts the user to enter a valid end date before today or 'today'.

        - Header:
            - generate_nasa_epic_header:
                Generates the NASA EPIC header with color formatting.
            - print_colored_header:
                Prints the colored header with the NASA and EPIC logos.

Classes:
    - NASAEPICManager:
        A class to manage NASA EPIC API requests and rate limits.

Usage:
    1. Ensure there's a file named 'nasa_api.txt' in the current working directory containing your NASA API Key(s).
       The API key file contents must be separated by a new line if you have multiple keys.
    2. Run the script.
    3. Follow the on-screen prompts to select an option, specify the date range, and enter the download location.
    4. The images will be downloaded and saved in folders according to the selected option and date.

Additional Notes:
    - You need to obtain a NASA API key to use this script.
        - You can get it by registering on the [NASA API portal](https://api.nasa.gov/index.html?apply-for-an-api-key).
    - After obtaining your API key, create a text file named 'nasa_api.txt' in the same directory as this script.
    - Save your API key(s) in 'nasa_api.txt', with each key on a new line if you have multiple keys. Example:
        - EXAMPLE:
            - YOUR_FIRST_API_KEY
            - YOUR_SECOND_API_KEY
    - Colorama is initialized to automatically reset colors after each use.
    - To download the metadata, you can go to URL: (https://pastebin.com/zAB34Acu)
        - Save the script as 'nasa_epic_metadata.py' in your current working directory.
        - Follow the detailed prompting from the program
"""

# Get Essential Imports
import os
import requests
from datetime import datetime, timedelta
from concurrent.futures import ThreadPoolExecutor
import time
from colorama import init, Fore, Style

# Initialize colorama
init(autoreset=True)

class NASAEPICManager:
    """
    A class to manage NASA EPIC API requests and rate limits.

    Attributes:
        api_keys (list): List of API keys obtained from a file.
        current_key_index (int): Index of the current API key being used.
        hourly_limit (int): Maximum number of API calls allowed per hour.
        calls_count (int): Number of API calls made in the current hour.
        last_reset_time (float): Timestamp of the last reset time for the hourly limit.
    """

    def __init__(self, api_keys_file):
        """
        Initializes the NASAEPICManager with API keys and rate limit parameters.

        Args:
            api_keys_file (str): Path to the file containing API keys.
        """
        with open(api_keys_file, 'r', encoding='utf-8') as file:
            self.api_keys = file.read().splitlines()
        self.current_key_index = 0
        self.hourly_limit = 1000  # Default value, will be updated based on the headers
        self.calls_count = 0
        self.last_reset_time = time.time()

    def make_api_request(self, url):
        """
        Makes an API request to the specified URL.

        Args:
            url (str): The URL to make the API request to.

        Returns:
            dict or None: JSON response from the API if successful, otherwise None.
        """
        self.check_rate_limit()
        current_api_key = self.api_keys[self.current_key_index]
        params = {'api_key': current_api_key}
        response = requests.get(url, params=params)

        # Update rate limit counters based on response headers
        self.calls_count += 1

        # Handle non-JSON response (e.g., HTML error pages)
        try:
            response_json = response.json()
        except requests.exceptions.JSONDecodeError:
            response_json = None

        remaining_requests = int(response.headers.get('X-RateLimit-Remaining', 0))
        if self.calls_count == 1:
            # Set the initial hourly limit based on the headers
            self.hourly_limit = remaining_requests

        self.last_reset_time = int(response.headers.get('X-RateLimit-Reset', self.last_reset_time))

        print(f"API Key: {current_api_key}, \n\nRemaining Requests: {remaining_requests}")

        return response_json

    def check_rate_limit(self):
        """
        Checks and manages the API call rate limit.
        Resets counters if more than one hour has passed or switches to the next API key if the hourly limit is reached.
        """
        current_time = time.time()
        time_since_reset = current_time - self.last_reset_time

        if time_since_reset >= 3600:
            # Reset counters if more than one hour has passed
            self.calls_count = 0
            self.last_reset_time = current_time

        if self.calls_count >= self.hourly_limit:
            # Switch to the next API key if the hourly limit is reached
            self.current_key_index = (self.current_key_index + 1) % len(self.api_keys)
            self.calls_count = 0

# Set epoch dates for the 4 options
epoch_dates = {
    'natural': '2015-06-13',
    'enhanced': '2015-08-07',
    'aerosol': '2020-09-04',
    'cloud': '2023-01-03'
}

def get_valid_start_date(selected_option):
    """
    Prompt the user to enter a valid start date after the epoch date for the selected option.

    Args:
        selected_option (str): The selected option ('natural', 'enhanced', 'aerosol', or 'cloud').

    Returns:
        str: A valid start date in the format 'YYYY-MM-DD'.
    """
    while True:
        start_date_input = input(f"\nEnter any start date after {epoch_dates[selected_option]} (YYYY-MM-DD) or type 'epoch': ")

        if start_date_input.lower() == 'epoch':
            return epoch_dates[selected_option]

        try:
            start_date = datetime.strptime(start_date_input, "%Y-%m-%d")
            if start_date < datetime.strptime(epoch_dates[selected_option], "%Y-%m-%d"):
                print(f"\nStart date must be after {epoch_dates[selected_option]}! Please try again.\n")
            elif start_date > datetime.utcnow():
                print("\nStart date cannot be in the future! Please try again.")
            else:
                return start_date.strftime("%Y-%m-%d")
        except ValueError:
            print("\nInvalid date format! (Try: 'YYYY-MM-DD) Please try again.")

def get_valid_end_date():
    """
    Prompt the user to enter a valid end date before today or 'today'.

    Returns:
        str: A valid end date in the format 'YYYY-MM-DD'.
    """
    while True:
        end_date_input = input("\nEnter any end date before today (YYYY-MM-DD) or type 'today': ")

        if not end_date_input:
            print("\nEnd date cannot be empty! Please enter a valid end date or type 'today' to use the current date.")
            continue

        if end_date_input.lower() == 'today':
            return datetime.utcnow().strftime("%Y-%m-%d")

        try:
            end_date = datetime.strptime(end_date_input, "%Y-%m-%d")
            if end_date > datetime.utcnow():
                print("\nEnd date cannot be in the future! Please try again.")
            else:
                return end_date.strftime("%Y-%m-%d")
        except ValueError:
            print("\nInvalid date format! (Try: 'YYYY-MM-DD) Please try again.")

def download_images_for_date(date_str, nasa_manager, selected_option, download_location):
    """
    Download images for a specified date and option from the NASA EPIC API.

    Args:
        date_str (str): The date in the format 'YYYY-MM-DD'.
        nasa_manager (NASAEPICManager): An instance of the NASAEPICManager class.
        selected_option (str): The selected option ('natural', 'enhanced', 'aerosol', or 'cloud').
        download_location (str): The folder location where the images will be downloaded.

    Returns:
        bool: True if images are downloaded successfully, False otherwise.
    """
    print(f"\nChecking if images are available for date {date_str} and option {selected_option}...\n")

    # Update the metadata URL based on the selected option
    metadata_url = f"https://api.nasa.gov/EPIC/api/{selected_option}/date/{date_str}"
    metadata = nasa_manager.make_api_request(metadata_url)

    # Check if imagery is available for the date
    if isinstance(metadata, list) and metadata:
        # Check if there are images available in the metadata
        if "\nNo imagery available for the date!\n" in metadata[0].get("reason", ""):
            print(f"No imagery available for the date: {date_str}")
            return False

        # If imagery is available, create and download images to the folder
        folder_name = f"{date_str[:4]}_{selected_option.upper()}"
        folder_path = os.path.join(download_location, folder_name, date_str[5:7])
        print("\nCreating folder and downloading images for date:", date_str)
        os.makedirs(folder_path, exist_ok=True)

        with ThreadPoolExecutor() as executor:
            futures = []

            for item in metadata:
                name = item["image"] + '.png'
                source = f"https://epic.gsfc.nasa.gov/archive/{selected_option}/{date_str[:4]}/{date_str[5:7]}/{date_str[8:10]}/png/{name}"
                destination = os.path.join(folder_path, name)

                futures.append(executor.submit(download_image, source, destination, nasa_manager))

            for future in futures:
                future.result()

        print("\nImages downloaded successfully in folder:\n\n", folder_path)
        return True
    elif metadata is not None:
        # Handle the case where the API request was successful but no metadata is available
        print(f"\nNo metadata available for the date: {date_str}!\n")
        return False
    else:
        # Handle the case where the API request failed (e.g., 503 error)
        print(f"\nError fetching metadata for {date_str}! Please check if the date is valid or try again later.\n")
        return False

def download_image(source, destination, nasa_manager, max_retries=3):
    """
    Download an image from the specified source URL to the destination path.

    Args:
        source (str): The URL of the image to download.
        destination (str): The local file path where the image will be saved.
        nasa_manager (NASAEPICManager): An instance of the NASAEPICManager class for rate limit management.
        max_retries (int): The maximum number of retry attempts in case of download failure. Default is 3.

    Returns:
        bool: True if the image is downloaded successfully, False otherwise.
    """
    for attempt in range(max_retries):
        try:
            response = requests.get(source, timeout=5)  # (Default=5) Adjust the timeout value as needed (in seconds)
            response.raise_for_status()  # Check for request success

            with open(destination, 'wb') as image_file:
                image_file.write(response.content)

            print("Downloaded:", os.path.basename(destination))
            return True  # Image downloaded successfully

        except requests.exceptions.Timeout:
            print(f"\nAttempt {attempt + 1}: Image download timed out! \n\nRetrying...\n")
            continue  # Retry the download

        except requests.exceptions.RequestException as e:
            print(f"\nError downloading image: {e}!\n")
            if 'API Key' in str(e):  # Retry with a new API key if the error is related to the API key
                nasa_manager.check_rate_limit()
                continue
            return False  # Failed to download image

    print(f"\nMax retries reached! Failed to download image: \n\n{os.path.basename(destination)}\n")
    return False

# Define the header as a list of strings for easier manipulation
header_lines = [
    "                        ███    ██  █████  ███████  █████                            ",
    "                        ████   ██ ██   ██ ██      ██   ██                           ",
    "                        ██ ██  ██ ███████ ███████ ███████                           ",
    "                        ██  ██ ██ ██   ██      ██ ██   ██                           ",
    "                        ██   ████ ██   ██ ███████ ██   ██                           ",
    "███████  █████         ██████  ███████        █████  ██████         ██████  ███████ ",
    "     ██ ██   ██       ██       ██            ██   ██      ██       ██       ██      ",
    "    ██   █████  █████ ███████  ███████ █████  █████   █████  █████ ███████  ███████ ",
    "   ██   ██   ██       ██    ██      ██       ██   ██      ██       ██    ██      ██ ",
    "   ██    █████         ██████  ███████        █████  ██████         ██████  ███████ ",
    "                            ███████ ██████  ██  ██████                              ",
    "                            ██      ██   ██ ██ ██                                   ",
    "                            █████   ██████  ██ ██                                   ",
    "                            ██      ██      ██ ██                                   ",
    "                            ███████ ██      ██  ██████                     (IMAGES) "
]

# Apply colors to the specified lines
colored_header = []
for i, line in enumerate(header_lines):
    if 0 <= i < 5:  # NASA
        colored_header.append(Fore.RED + line + Style.RESET_ALL)
    elif 5 <= i < 10:  # 78-65-83-65 (ASCII CODE)
        colored_header.append(Fore.WHITE + line + Style.RESET_ALL)
    elif 10 <= i < 15:  # EPIC
        colored_header.append(Fore.BLUE + line + Style.RESET_ALL)
    else:
        colored_header.append(line)

# Print the colored header
for line in colored_header:
    print(line)

# Start menu
while True:
    """
    Main loop to interact with the user, allowing them to select options for downloading EPIC images.

    The loop continuously prompts the user to select an option for EPIC imagery download, input start and end dates,
    and specify the download location. It then iterates through the specified date range, attempting to download
    EPIC images for each date within the range.

    The loop continues until the user decides to exit the program.

    """
    nasa_manager_loop = NASAEPICManager('nasa_api.txt')

    # Display options for the user
    print("_" * 83)
    print("\nOptions:")
    print("1. Natural")
    print("2. Enhanced")
    print("3. Cloud")
    print("4. Aerosol")

    # Get user's choice
    option_loop = input("\nSelect an option (1-4): ")

    # Map the user's choice to the corresponding URL component
    options_mapping = {
        '1': 'natural',
        '2': 'enhanced',
        '3': 'cloud',
        '4': 'aerosol'
    }

    selected_option_loop = options_mapping.get(option_loop)

    # Check if the selected option is valid
    if not selected_option_loop:
        # If the option is invalid, print an error message
        print("\nInvalid option! Please enter a valid option number.")
        # Since the option is invalid, continue to the next iteration of the loop
        # to prompt the user again for a valid option
        continue

    # Get the start date from the user
    start_date_str_loop = get_valid_start_date(selected_option_loop)

    print(f"\nSelected Option: {selected_option_loop}, Start Date: {start_date_str_loop}, Epoch Date: {epoch_dates[selected_option_loop]}")

    # Get the end date from the user
    end_date_str_loop = get_valid_end_date()

    # Get the download location from the user
    download_location_loop = input("\nEnter the download location (e.g., 'EPIC_ARCHIVE'): ")
    if not download_location_loop:
        print("\nDownload location cannot be empty!\n")
        continue

    # Convert the start and end dates to datetime objects
    start_date_loop = datetime.strptime(start_date_str_loop, "%Y-%m-%d")
    end_date_loop = datetime.strptime(end_date_str_loop, "%Y-%m-%d")

    # Iterate through the date range and download images for each date
    current_date_loop = start_date_loop
    while current_date_loop <= end_date_loop:
        current_date_str_loop = current_date_loop.strftime("%Y-%m-%d")
        if not download_images_for_date(current_date_str_loop, nasa_manager_loop, selected_option_loop, download_location_loop):
            # If no images are available for the current date, move to the next date
            current_date_loop += timedelta(days=1)
            continue

        # Move to the next date
        current_date_loop += timedelta(days=1)

    # Exit the loop if all dates have been processed
    break

# Once the loop exits, print a farewell message
print("\nExiting Program...   GoodBye!\n")