SolarImageDownloader / gradio_app.py
AK51's picture
Upload 13308 files
b610d23 verified
#!/usr/bin/env python3
"""
NASA Solar Image Downloader - Complete Gradio Web Interface
Web-based interface with all features from the original GUI application.
"""
import sys
import os
from pathlib import Path
from datetime import datetime, timedelta
import subprocess
import shutil
import threading
import time
# Add src to Python path
sys.path.insert(0, str(Path(__file__).parent / "src"))
try:
import gradio as gr
from PIL import Image
import cv2
import numpy as np
except ImportError as e:
print(f"❌ Required libraries not available: {e}")
print("💡 Install with: pip install gradio pillow opencv-python")
sys.exit(1)
from src.downloader.directory_scraper import DirectoryScraper
from src.storage.storage_organizer import StorageOrganizer
from src.downloader.image_fetcher import ImageFetcher, DownloadManager
class NASADownloaderGradio:
"""Complete Gradio web interface for NASA Solar Image Downloader."""
def __init__(self):
"""Initialize the Gradio application."""
self.resolution = "1024"
self.solar_filter = "0211"
# Initialize components
self.storage = StorageOrganizer("data",
resolution=self.resolution,
solar_filter=self.solar_filter)
self.scraper = DirectoryScraper(rate_limit_delay=1.0,
resolution=self.resolution,
solar_filter=self.solar_filter)
self.fetcher = ImageFetcher(rate_limit_delay=1.0)
self.download_manager = DownloadManager(self.fetcher, self.storage)
# Filter data with full information and thumbnail paths
self.filter_data = {
"0193": {"name": "193 Å", "desc": "Coronal loops", "color": "#ff6b6b", "image": "src/ui_img/20251220_000753_1024_0193.jpg"},
"0304": {"name": "304 Å", "desc": "Chromosphere", "color": "#4ecdc4", "image": "src/ui_img/20251220_000854_1024_0304.jpg"},
"0171": {"name": "171 Å", "desc": "Quiet corona", "color": "#45b7d1", "image": "src/ui_img/20251220_000658_1024_0171.jpg"},
"0211": {"name": "211 Å", "desc": "Active regions", "color": "#f9ca24", "image": "src/ui_img/20251220_000035_1024_0211.jpg"},
"0131": {"name": "131 Å", "desc": "Flaring regions", "color": "#f0932b", "image": "src/ui_img/20251220_000644_1024_0131.jpg"},
"0335": {"name": "335 Å", "desc": "Active cores", "color": "#eb4d4b", "image": "src/ui_img/20251220_000114_1024_0335.jpg"},
"0094": {"name": "94 Å", "desc": "Hot plasma", "color": "#6c5ce7", "image": "src/ui_img/20251220_000600_1024_0094.jpg"},
"1600": {"name": "1600 Å", "desc": "Transition region", "color": "#a29bfe", "image": "src/ui_img/20251220_000151_1024_1600.jpg"},
"1700": {"name": "1700 Å", "desc": "Temperature min", "color": "#fd79a8", "image": "src/ui_img/20251220_000317_1024_1700.jpg"},
"094335193": {"name": "094+335+193", "desc": "Composite: Hot plasma + Active cores + Coronal loops", "color": "#8e44ad", "image": "src/ui_img/20251219_000311_1024_094335193.jpg"},
"304211171": {"name": "304+211+171", "desc": "Composite: Chromosphere + Active regions + Quiet corona", "color": "#e67e22", "image": "src/ui_img/20251219_000311_1024_304211171.jpg"},
"211193171": {"name": "211+193+171", "desc": "Composite: Active regions + Coronal loops + Quiet corona", "color": "#27ae60", "image": "src/ui_img/20251219_001633_1024_211193171.jpg"}
}
# Custom keywords for advanced users
self.custom_keywords = {filter_num: filter_num for filter_num in self.filter_data.keys()}
# Image viewer state
self.current_images = []
self.current_image_index = 0
self.is_playing = False
self.play_speed = 120.0 # FPS for playback
self.last_update_time = 0 # Track last update time for playback
def set_date_range(self, days_back):
"""Set date range for quick selection."""
end_date = datetime.now()
start_date = end_date - timedelta(days=days_back)
return start_date.strftime("%Y-%m-%d"), end_date.strftime("%Y-%m-%d")
def download_images(self, start_date, end_date, resolution, solar_filter, progress=gr.Progress()):
"""Download images for the specified date range."""
try:
# Update settings
self.resolution = resolution
self.solar_filter = solar_filter
# Use custom keyword if available
search_keyword = self.custom_keywords.get(solar_filter, solar_filter)
self.scraper.update_filters(resolution, search_keyword)
self.storage.update_file_pattern(resolution, search_keyword)
# Parse dates
start = datetime.strptime(start_date, "%Y-%m-%d")
end = datetime.strptime(end_date, "%Y-%m-%d")
progress(0, desc="Scanning directories...")
# Get available images
available_images = self.scraper.get_available_images_for_date_range(start, end)
if not available_images:
return f"❌ No images found for date range {start_date} to {end_date}", self.get_available_dates()
progress(0.2, desc=f"Found {len(available_images)} images")
# Filter new images
new_images = self.scraper.filter_new_images(available_images, self.storage)
if not new_images:
return f"✅ All {len(available_images)} images already downloaded!", self.get_available_dates()
progress(0.3, desc=f"Downloading {len(new_images)} new images...")
# Create download tasks
tasks = self.scraper.create_download_tasks(new_images, self.storage)
# Download images
successful = 0
failed = 0
for i, task in enumerate(tasks):
progress((0.3 + (i / len(tasks)) * 0.6), desc=f"Downloading {i+1}/{len(tasks)}: {task.target_path.name}")
success = self.download_manager.download_and_save(task)
if success:
successful += 1
else:
failed += 1
progress(1.0, desc="Complete!")
result = f"✅ Download complete!\n"
result += f"📥 Downloaded: {successful} images\n"
result += f"❌ Failed: {failed} images\n"
result += f"📊 Total available: {len(available_images)} images\n"
result += f"� DFilter: {self.filter_data[solar_filter]['name']} - {self.filter_data[solar_filter]['desc']}"
return result, self.get_available_dates()
except Exception as e:
return f"❌ Error: {str(e)}", self.get_available_dates()
def get_latest_image(self):
"""Get the most recent downloaded image."""
try:
data_dir = self.storage.base_data_dir
# Find the most recent image
all_images = []
for year_dir in sorted(data_dir.iterdir(), reverse=True):
if not year_dir.is_dir():
continue
for month_dir in sorted(year_dir.iterdir(), reverse=True):
if not month_dir.is_dir():
continue
for day_dir in sorted(month_dir.iterdir(), reverse=True):
if not day_dir.is_dir():
continue
images = list(day_dir.glob(f"*_{self.resolution}_*.jpg"))
if images:
all_images.extend(images)
if all_images:
# Sort by filename (which includes timestamp) and get the latest
latest = sorted(all_images, reverse=True)[0]
return str(latest)
return None
except Exception as e:
print(f"Error getting latest image: {e}")
return None
def get_available_dates(self, resolution=None, solar_filter=None):
"""Get list of available dates with images for specific resolution and filter."""
dates = []
data_dir = self.storage.base_data_dir
# Use current settings if not specified
if resolution is None:
resolution = self.resolution
if solar_filter is None:
solar_filter = self.solar_filter
# Use custom keyword if available
search_keyword = self.custom_keywords.get(solar_filter, solar_filter)
if data_dir.exists():
for year_dir in data_dir.iterdir():
if not year_dir.is_dir() or not year_dir.name.isdigit():
continue
for month_dir in year_dir.iterdir():
if not month_dir.is_dir() or not month_dir.name.isdigit():
continue
for day_dir in month_dir.iterdir():
if not day_dir.is_dir() or not day_dir.name.isdigit():
continue
# Filter images by resolution and solar filter
all_images = list(day_dir.glob("*.jpg"))
filtered_images = [img for img in all_images
if f"_{resolution}_" in img.name and search_keyword in img.name]
if filtered_images:
try:
date = datetime(int(year_dir.name), int(month_dir.name), int(day_dir.name))
date_str = f"{date.strftime('%Y-%m-%d')} ({len(filtered_images)} images)"
dates.append(date_str)
except ValueError:
continue
return sorted(dates, reverse=True)
def load_images_for_date_range(self, from_date, to_date, resolution, solar_filter):
"""Load images for a date range with specific resolution and filter."""
try:
# Update settings
self.resolution = resolution
self.solar_filter = solar_filter
# Use custom keyword if available
search_keyword = self.custom_keywords.get(solar_filter, solar_filter)
# Update storage pattern to match selected filter and resolution
self.storage.update_file_pattern(resolution, search_keyword)
start_date = datetime.strptime(from_date.split(' ')[0], "%Y-%m-%d")
end_date = datetime.strptime(to_date.split(' ')[0], "%Y-%m-%d")
# Ensure from_date is not after to_date
if start_date > end_date:
start_date, end_date = end_date, start_date
# Load images from all dates in the range
self.current_images = []
total_images = 0
# Get all dates in range
current_date = start_date
while current_date <= end_date:
images = self.storage.list_local_images(current_date)
if images:
date_path = self.storage.get_date_path(current_date)
# Filter images by resolution and solar filter
for filename in sorted(images):
# Check if filename matches the selected resolution and filter
if f"_{resolution}_" in filename and search_keyword in filename:
image_path = date_path / filename
self.current_images.append((str(image_path), filename, current_date))
total_images += 1
current_date += timedelta(days=1)
if not self.current_images:
filter_name = self.filter_data[solar_filter]['name']
return None, f"❌ No images found for date range {start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}\nResolution: {resolution}px, Filter: {filter_name}", "0 / 0"
# Sort all images by filename (which includes timestamp)
self.current_images.sort(key=lambda x: x[1])
self.current_image_index = 0
date_range_text = f"{start_date.strftime('%Y-%m-%d')}" if start_date == end_date else f"{start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}"
filter_name = self.filter_data[solar_filter]['name']
return self.current_images[0][0], f"✅ Loaded {total_images} images for {date_range_text}\nResolution: {resolution}px, Filter: {filter_name}", f"1 / {len(self.current_images)}"
except Exception as e:
return None, f"❌ Error: {str(e)}", "0 / 0"
def navigate_image(self, direction):
"""Navigate through images."""
if not self.current_images:
return None, "No images loaded", "0 / 0", "▶ Play"
if direction == "first":
self.current_image_index = 0
elif direction == "prev":
self.current_image_index = max(0, self.current_image_index - 1)
elif direction == "next":
self.current_image_index = min(len(self.current_images) - 1, self.current_image_index + 1)
elif direction == "last":
self.current_image_index = len(self.current_images) - 1
current_image = self.current_images[self.current_image_index]
image_path, filename, image_date = current_image
# Extract timestamp
timestamp = filename.split('_')[1] if '_' in filename else "Unknown"
if len(timestamp) == 6:
formatted_time = f"{timestamp[:2]}:{timestamp[2:4]}:{timestamp[4:6]}"
else:
formatted_time = timestamp
info_text = f"📅 {image_date.strftime('%Y-%m-%d')}{formatted_time}"
position_text = f"{self.current_image_index + 1} / {len(self.current_images)}"
play_button_text = "⏸ Pause" if self.is_playing else "▶ Play"
return image_path, info_text, position_text, play_button_text
def toggle_play(self):
"""Toggle play/pause for image sequence."""
if not self.current_images:
return None, "No images loaded", "0 / 0", "▶ Play", f"{self.play_speed:.1f} FPS"
if self.is_playing:
self.is_playing = False
play_button_text = "▶ Play"
else:
self.is_playing = True
play_button_text = "⏸ Pause"
self.last_update_time = time.time() # Reset timer
current_image = self.current_images[self.current_image_index]
image_path, filename, image_date = current_image
# Extract timestamp
timestamp = filename.split('_')[1] if '_' in filename else "Unknown"
if len(timestamp) == 6:
formatted_time = f"{timestamp[:2]}:{timestamp[2:4]}:{timestamp[4:6]}"
else:
formatted_time = timestamp
info_text = f"📅 {image_date.strftime('%Y-%m-%d')}{formatted_time}"
position_text = f"{self.current_image_index + 1} / {len(self.current_images)}"
return image_path, info_text, position_text, play_button_text, f"{self.play_speed:.1f} FPS"
def update_playback(self):
"""Update playback - called by timer."""
if not self.is_playing or not self.current_images:
# Return current state without changes
if not self.current_images:
return None, "No images loaded", "0 / 0", "▶ Play", f"{self.play_speed:.1f} FPS"
current_image = self.current_images[self.current_image_index]
image_path, filename, image_date = current_image
# Extract timestamp
timestamp = filename.split('_')[1] if '_' in filename else "Unknown"
if len(timestamp) == 6:
formatted_time = f"{timestamp[:2]}:{timestamp[2:4]}:{timestamp[4:6]}"
else:
formatted_time = timestamp
info_text = f"📅 {image_date.strftime('%Y-%m-%d')}{formatted_time}"
position_text = f"{self.current_image_index + 1} / {len(self.current_images)}"
play_button_text = "⏸ Pause" if self.is_playing else "▶ Play"
return image_path, info_text, position_text, play_button_text, f"{self.play_speed:.1f} FPS"
# Check if enough time has passed for next frame
current_time = time.time()
frame_interval = 1.0 / self.play_speed
if current_time - self.last_update_time >= frame_interval:
# Advance to next image
if self.current_image_index >= len(self.current_images) - 1:
self.current_image_index = 0 # Loop back to start
else:
self.current_image_index += 1
self.last_update_time = current_time
# Return current image
current_image = self.current_images[self.current_image_index]
image_path, filename, image_date = current_image
# Extract timestamp
timestamp = filename.split('_')[1] if '_' in filename else "Unknown"
if len(timestamp) == 6:
formatted_time = f"{timestamp[:2]}:{timestamp[2:4]}:{timestamp[4:6]}"
else:
formatted_time = timestamp
info_text = f"📅 {image_date.strftime('%Y-%m-%d')}{formatted_time}"
position_text = f"{self.current_image_index + 1} / {len(self.current_images)}"
play_button_text = "⏸ Pause"
return image_path, info_text, position_text, play_button_text, f"{self.play_speed:.1f} FPS"
def update_play_speed(self, speed):
"""Update playback speed."""
self.play_speed = float(speed)
return f"{self.play_speed:.1f} FPS"
def select_video_file(self):
"""Select and preview video file."""
# This would be handled by Gradio's file upload component
# Return placeholder for now
return "Please use the file upload component to select an MP4 video"
def get_video_list(self):
"""Get list of available video files."""
video_dir = Path("video")
if not video_dir.exists():
return []
videos = []
for video_file in video_dir.glob("*.mp4"):
try:
size_mb = video_file.stat().st_size / (1024 * 1024)
videos.append(f"{video_file.name} ({size_mb:.1f} MB)")
except:
videos.append(video_file.name)
return sorted(videos, reverse=True)
def open_data_folder(self):
"""Open data folder (returns path for web interface)."""
data_dir = self.storage.base_data_dir
return f"📁 Data folder location: {data_dir.absolute()}"
def cleanup_corrupted_files(self):
"""Clean up corrupted files."""
total_removed = 0
data_dir = self.storage.base_data_dir
if data_dir.exists():
for year_dir in data_dir.iterdir():
if not year_dir.is_dir():
continue
for month_dir in year_dir.iterdir():
if not month_dir.is_dir():
continue
for day_dir in month_dir.iterdir():
if not day_dir.is_dir():
continue
try:
date = datetime(int(year_dir.name), int(month_dir.name), int(day_dir.name))
removed = self.storage.cleanup_corrupted_files(date)
total_removed += removed
except:
continue
return f"🧹 Cleanup complete! Removed {total_removed} corrupted files."
def create_video(self, start_date, end_date, fps, resolution, solar_filter, progress=gr.Progress()):
"""Create MP4 video from images."""
try:
# Update settings
self.resolution = resolution
self.solar_filter = solar_filter
# Use custom keyword if available
search_keyword = self.custom_keywords.get(solar_filter, solar_filter)
self.scraper.update_filters(resolution, search_keyword)
self.storage.update_file_pattern(resolution, search_keyword)
# Parse dates
start = datetime.strptime(start_date, "%Y-%m-%d")
end = datetime.strptime(end_date, "%Y-%m-%d")
progress(0.1, desc="Collecting images...")
# Collect all images from the date range
all_image_paths = []
current_date = start
while current_date <= end:
images = self.storage.list_local_images(current_date)
if images:
date_path = self.storage.get_date_path(current_date)
for filename in sorted(images):
image_path = date_path / filename
if image_path.exists():
all_image_paths.append(image_path)
current_date += timedelta(days=1)
if not all_image_paths:
return None, f"❌ No images found for date range {start_date} to {end_date}"
progress(0.2, desc=f"Found {len(all_image_paths)} images. Creating video...")
# Create video directory
video_dir = Path("video")
video_dir.mkdir(exist_ok=True)
# Generate output filename
if start == end:
output_file = f"nasa_solar_{start.strftime('%Y%m%d')}.mp4"
else:
output_file = f"nasa_solar_{start.strftime('%Y%m%d')}_to_{end.strftime('%Y%m%d')}.mp4"
output_path = video_dir / output_file
# Create temporary directory for ffmpeg
temp_dir = Path("temp_video_frames")
temp_dir.mkdir(exist_ok=True)
try:
# Create sequential frame files
for i, src_path in enumerate(all_image_paths):
progress(0.2 + (i / len(all_image_paths)) * 0.5,
desc=f"Preparing frame {i+1}/{len(all_image_paths)}")
temp_path = temp_dir / f"frame_{i:06d}.jpg"
if temp_path.exists():
temp_path.unlink()
try:
temp_path.symlink_to(src_path.absolute())
except OSError:
shutil.copy2(src_path, temp_path)
progress(0.7, desc="Running FFmpeg to create video...")
# Run ffmpeg
input_pattern = str(temp_dir / "frame_%06d.jpg")
ffmpeg_cmd = [
'ffmpeg', '-y', '-framerate', str(fps),
'-i', input_pattern, '-c:v', 'libx264',
'-pix_fmt', 'yuv420p', '-crf', '18',
str(output_path)
]
result = subprocess.run(ffmpeg_cmd, capture_output=True, text=True)
progress(1.0, desc="Video creation complete!")
if result.returncode == 0:
size_mb = output_path.stat().st_size / (1024 * 1024)
duration = len(all_image_paths) / fps
message = f"✅ Video created successfully!\n"
message += f"📁 File: {output_path.name}\n"
message += f"💾 Size: {size_mb:.1f} MB\n"
message += f"🎞️ Frames: {len(all_image_paths)}\n"
message += f"⏱️ Duration: {duration:.1f} seconds\n"
message += f"🔍 Filter: {self.filter_data[solar_filter]['name']}"
return str(output_path), message
else:
return None, f"❌ FFmpeg error: {result.stderr}"
finally:
if temp_dir.exists():
shutil.rmtree(temp_dir)
except Exception as e:
return None, f"❌ Error: {str(e)}"
def get_system_info(self):
"""Get system information."""
info = "🖥️ **System Information**\n\n"
# Check FFmpeg
try:
result = subprocess.run(['ffmpeg', '-version'], capture_output=True, timeout=5)
ffmpeg_status = "✅ Available" if result.returncode == 0 else "❌ Not found"
except:
ffmpeg_status = "❌ Not found"
info += f"**FFmpeg**: {ffmpeg_status}\n"
# Check OpenCV
try:
import cv2
opencv_status = f"✅ Available (v{cv2.__version__})"
except:
opencv_status = "❌ Not found"
info += f"**OpenCV**: {opencv_status}\n"
# Check PIL
try:
from PIL import Image
pil_status = f"✅ Available (v{Image.__version__})"
except:
pil_status = "❌ Not found"
info += f"**Pillow**: {pil_status}\n\n"
# Data directory info
data_dir = self.storage.base_data_dir
if data_dir.exists():
info += f"**Data Directory**: {data_dir.absolute()}\n"
# Count total images
total_images = 0
for year_dir in data_dir.iterdir():
if year_dir.is_dir():
for month_dir in year_dir.iterdir():
if month_dir.is_dir():
for day_dir in month_dir.iterdir():
if day_dir.is_dir():
images = list(day_dir.glob("*.jpg"))
total_images += len(images)
info += f"**Total Images**: {total_images}\n"
else:
info += f"**Data Directory**: Not created yet\n"
info += f"\n**Created by Andy Kong**"
return info
def update_custom_keyword(self, filter_name, keyword):
"""Update custom keyword for a filter."""
if filter_name in self.custom_keywords:
self.custom_keywords[filter_name] = keyword.strip() if keyword.strip() else filter_name
return f"✅ Updated {filter_name} keyword to: {self.custom_keywords[filter_name]}"
return f"❌ Invalid filter: {filter_name}"
def get_filter_gallery_data(self):
"""Get gallery data for filter selection with thumbnails."""
gallery_data = []
for filter_key, data in self.filter_data.items():
# Create gallery item with image path and caption
caption = f"{data['name']}\n{data['desc']}"
gallery_data.append((data['image'], caption))
return gallery_data
def get_filter_key_from_gallery_index(self, index):
"""Get filter key from gallery selection index."""
filter_keys = list(self.filter_data.keys())
if 0 <= index < len(filter_keys):
return filter_keys[index]
return "0211" # Default
def get_gallery_index_from_filter_key(self, filter_key):
"""Get gallery index from filter key."""
filter_keys = list(self.filter_data.keys())
try:
return filter_keys.index(filter_key)
except ValueError:
return 3 # Default to 0211 (index 3)
def on_filter_gallery_select(self, evt: gr.SelectData):
"""Handle filter gallery selection."""
if evt.index is not None:
filter_key = self.get_filter_key_from_gallery_index(evt.index)
filter_data = self.filter_data[filter_key]
info_text = f"**Selected:** {filter_data['name']} - {filter_data['desc']}"
return filter_key, info_text
return "0211", "**Selected:** 211 Å - Active regions"
def reset_custom_keywords(self):
"""Reset all custom keywords to defaults."""
self.custom_keywords = {filter_num: filter_num for filter_num in self.filter_data.keys()}
return "✅ All keywords reset to defaults"
def create_interface(self):
"""Create the complete Gradio interface with all features."""
with gr.Blocks(title="🌞 NASA Solar Image Downloader") as app:
gr.Markdown("# 🌞 NASA Solar Image Downloader")
gr.Markdown("**Complete web interface** - Download, view, and create videos from NASA Solar Dynamics Observatory images")
with gr.Tabs():
# Download Tab
with gr.Tab("📥 Download Images"):
gr.Markdown("### Download NASA Solar Images")
with gr.Row():
with gr.Column():
gr.Markdown("#### 🗓️ Quick Date Selection")
with gr.Row():
today_btn = gr.Button("Today", size="sm")
last3_btn = gr.Button("Last 3 Days", size="sm")
lastweek_btn = gr.Button("Last Week", size="sm")
gr.Markdown("#### 📅 Custom Date Range")
download_start_date = gr.Textbox(
label="Start Date (YYYY-MM-DD)",
value=datetime.now().strftime("%Y-%m-%d")
)
download_end_date = gr.Textbox(
label="End Date (YYYY-MM-DD)",
value=datetime.now().strftime("%Y-%m-%d")
)
gr.Markdown("#### 🔧 Image Settings")
download_resolution = gr.Dropdown(
choices=["1024", "2048", "4096"],
value="1024",
label="Resolution (pixels)"
)
download_btn = gr.Button("🔍 Find & Download Images", variant="primary", size="lg")
with gr.Column():
gr.Markdown("#### 🌞 Solar Filter Selection")
gr.Markdown("*Click on a thumbnail to select a solar filter*")
download_filter_gallery = gr.Gallery(
value=self.get_filter_gallery_data(),
label="Solar Filters",
columns=4,
rows=3,
height="auto",
object_fit="contain",
show_label=False,
selected_index=3 # Default to 0211
)
# Hidden state to store the selected filter key
download_filter = gr.State(value="0211")
# Display selected filter info
download_filter_info = gr.Markdown("**Selected:** 211 Å - Active regions")
download_output = gr.Textbox(label="Download Status", lines=8)
# Quick date button actions
today_btn.click(
fn=lambda: self.set_date_range(0),
outputs=[download_start_date, download_end_date]
)
last3_btn.click(
fn=lambda: self.set_date_range(2),
outputs=[download_start_date, download_end_date]
)
lastweek_btn.click(
fn=lambda: self.set_date_range(6),
outputs=[download_start_date, download_end_date]
)
# Store available dates for refresh
available_dates_state = gr.State(value=self.get_available_dates())
# Gallery selection event for download tab
download_filter_gallery.select(
fn=self.on_filter_gallery_select,
outputs=[download_filter, download_filter_info]
)
download_btn.click(
fn=self.download_images,
inputs=[download_start_date, download_end_date, download_resolution, download_filter],
outputs=[download_output, available_dates_state]
)
# View Images Tab
with gr.Tab("👁️ View Images"):
gr.Markdown("### View Downloaded Images with Full Playback Controls")
with gr.Row():
with gr.Column():
gr.Markdown("#### 🔧 Image Settings")
with gr.Row():
view_resolution = gr.Dropdown(
choices=["1024", "2048", "4096"],
value="1024",
label="Resolution (pixels)",
scale=1
)
gr.Markdown("#### 🌞 Solar Filter Selection")
gr.Markdown("*Click on a thumbnail to select a solar filter*")
view_filter_gallery = gr.Gallery(
value=self.get_filter_gallery_data(),
label="Solar Filters",
columns=4,
rows=3,
height="auto",
object_fit="contain",
show_label=False,
selected_index=3 # Default to 0211
)
# Hidden state to store the selected filter key
view_filter = gr.State(value="0211")
# Display selected filter info
view_filter_info = gr.Markdown("**Selected:** 211 Å - Active regions")
gr.Markdown("#### 📅 Date Range Selection")
with gr.Row():
view_from_date = gr.Dropdown(
choices=self.get_available_dates(),
label="From Date",
info="Select starting date"
)
view_to_date = gr.Dropdown(
choices=self.get_available_dates(),
label="To Date",
info="Select ending date"
)
with gr.Row():
refresh_dates_btn = gr.Button("🔄 Refresh Dates", size="sm")
load_images_btn = gr.Button("📂 Load Images", variant="primary")
view_status = gr.Textbox(label="Status", lines=3)
image_position = gr.Textbox(label="Position", value="0 / 0", interactive=False)
with gr.Column():
gr.Markdown("#### 🎮 Playback Controls")
with gr.Row():
first_btn = gr.Button("⏮ First", size="sm")
prev_btn = gr.Button("⏪ Prev", size="sm")
play_btn = gr.Button("▶ Play", variant="primary")
next_btn = gr.Button("Next ⏩", size="sm")
last_btn = gr.Button("Last ⏭", size="sm")
view_image = gr.Image(label="Solar Image", type="filepath", height=600)
image_info = gr.Textbox(label="Image Information", interactive=False)
gr.Markdown("#### ⚡ Speed Control")
with gr.Row():
speed_slider = gr.Slider(
minimum=0.5,
maximum=240.0,
value=120.0,
step=0.1,
label="Playback Speed (FPS)",
info="Frames per second during playback"
)
speed_display = gr.Textbox(
value="120.0 FPS",
label="Current Speed",
interactive=False,
scale=0
)
# Gallery selection event
view_filter_gallery.select(
fn=self.on_filter_gallery_select,
outputs=[view_filter, view_filter_info]
)
refresh_dates_btn.click(
fn=lambda res, filt: [gr.Dropdown(choices=self.get_available_dates(res, filt))] * 2,
inputs=[view_resolution, view_filter],
outputs=[view_from_date, view_to_date]
)
# Auto-refresh dates when resolution or filter changes
view_resolution.change(
fn=lambda res, filt: [gr.Dropdown(choices=self.get_available_dates(res, filt))] * 2,
inputs=[view_resolution, view_filter],
outputs=[view_from_date, view_to_date]
)
load_images_btn.click(
fn=self.load_images_for_date_range,
inputs=[view_from_date, view_to_date, view_resolution, view_filter],
outputs=[view_image, view_status, image_position]
)
# Navigation buttons
first_btn.click(
fn=lambda: self.navigate_image("first"),
outputs=[view_image, image_info, image_position, play_btn]
)
prev_btn.click(
fn=lambda: self.navigate_image("prev"),
outputs=[view_image, image_info, image_position, play_btn]
)
next_btn.click(
fn=lambda: self.navigate_image("next"),
outputs=[view_image, image_info, image_position, play_btn]
)
last_btn.click(
fn=lambda: self.navigate_image("last"),
outputs=[view_image, image_info, image_position, play_btn]
)
# Play/Pause button
play_btn.click(
fn=self.toggle_play,
outputs=[view_image, image_info, image_position, play_btn, speed_display]
)
# Speed control
speed_slider.change(
fn=self.update_play_speed,
inputs=[speed_slider],
outputs=[speed_display]
)
# Auto-update timer for playback (every 100ms)
play_timer = gr.Timer(0.1) # 100ms interval
play_timer.tick(
fn=self.update_playback,
outputs=[view_image, image_info, image_position, play_btn, speed_display]
)
# Create Video Tab
with gr.Tab("🎬 Create Videos"):
gr.Markdown("### Create MP4 Time-lapse Videos")
with gr.Row():
with gr.Column():
gr.Markdown("#### 🗓️ Quick Date Selection")
with gr.Row():
video_today_btn = gr.Button("Today", size="sm")
video_last3_btn = gr.Button("Last 3 Days", size="sm")
video_lastweek_btn = gr.Button("Last Week", size="sm")
gr.Markdown("#### 📅 Video Date Range")
video_start_date = gr.Textbox(
label="Start Date (YYYY-MM-DD)",
value=datetime.now().strftime("%Y-%m-%d")
)
video_end_date = gr.Textbox(
label="End Date (YYYY-MM-DD)",
value=datetime.now().strftime("%Y-%m-%d")
)
gr.Markdown("#### 🎬 Video Settings")
video_fps = gr.Slider(
minimum=1,
maximum=120,
value=10,
step=1,
label="FPS (Frames Per Second)",
info="Higher FPS = smoother but faster playback"
)
video_resolution = gr.Dropdown(
choices=["1024", "2048", "4096"],
value="1024",
label="Resolution"
)
gr.Markdown("#### 🌞 Solar Filter Selection")
gr.Markdown("*Click on a thumbnail to select a solar filter*")
video_filter_gallery = gr.Gallery(
value=self.get_filter_gallery_data(),
label="Solar Filters",
columns=4,
rows=3,
height="auto",
object_fit="contain",
show_label=False,
selected_index=3 # Default to 0211
)
# Hidden state to store the selected filter key
video_filter = gr.State(value="0211")
# Display selected filter info
video_filter_info = gr.Markdown("**Selected:** 211 Å - Active regions")
with gr.Row():
create_video_btn = gr.Button("🎬 Create Video for Date Range", variant="primary")
create_all_btn = gr.Button("🎬 Create Combined Video (All Available)")
with gr.Column():
video_output = gr.Textbox(label="Video Creation Status", lines=8)
video_player = gr.Video(label="Created Video", height=400)
# Video Playback Section
gr.Markdown("### 🎥 Play MP4 Videos")
with gr.Row():
with gr.Column():
gr.Markdown("#### 📁 Video Selection")
video_file_upload = gr.File(
label="Select MP4 File",
file_types=[".mp4"],
type="filepath"
)
# Or select from created videos
available_videos = gr.Dropdown(
choices=self.get_video_list(),
label="Or Select from Created Videos",
info="Choose from previously created videos"
)
refresh_videos_btn = gr.Button("🔄 Refresh Video List", size="sm")
gr.Markdown("#### 🎮 Video Controls")
with gr.Row():
video_play_btn = gr.Button("▶ Play Video", variant="primary")
video_stop_btn = gr.Button("⏹ Stop")
video_fullscreen_btn = gr.Button("🔳 Fullscreen")
video_info = gr.Textbox(label="Video Information", lines=3)
# Quick date button actions for video
video_today_btn.click(
fn=lambda: self.set_date_range(0),
outputs=[video_start_date, video_end_date]
)
video_last3_btn.click(
fn=lambda: self.set_date_range(2),
outputs=[video_start_date, video_end_date]
)
video_lastweek_btn.click(
fn=lambda: self.set_date_range(6),
outputs=[video_start_date, video_end_date]
)
# Gallery selection event for video tab
video_filter_gallery.select(
fn=self.on_filter_gallery_select,
outputs=[video_filter, video_filter_info]
)
create_video_btn.click(
fn=self.create_video,
inputs=[video_start_date, video_end_date, video_fps, video_resolution, video_filter],
outputs=[video_player, video_output]
)
# Settings Tab
with gr.Tab("⚙️ Settings"):
gr.Markdown("### Application Settings & System Information")
with gr.Row():
with gr.Column():
gr.Markdown("#### 🔧 Download Settings")
rate_limit = gr.Slider(
minimum=0.5,
maximum=5.0,
value=1.0,
step=0.1,
label="Rate Limit Delay (seconds)",
info="Delay between downloads to be respectful to NASA servers"
)
gr.Markdown("#### 🔍 Custom Keyword Search")
gr.Markdown("*Customize search keywords for each solar filter. Leave empty to use defaults.*")
# Create custom keyword inputs for each filter
keyword_inputs = {}
for filter_num, data in list(self.filter_data.items())[:6]: # First 6 filters
with gr.Row():
gr.Markdown(f"**{data['name']}** ({filter_num})")
keyword_inputs[filter_num] = gr.Textbox(
value=filter_num,
placeholder=f"Default: {filter_num}",
scale=2
)
with gr.Row():
reset_keywords_btn = gr.Button("🔄 Reset to Defaults", size="sm")
apply_keywords_btn = gr.Button("✅ Apply Keywords", variant="primary")
keyword_status = gr.Textbox(label="Keyword Status", lines=2)
with gr.Column():
gr.Markdown("#### 📁 Data Management")
with gr.Row():
open_data_btn = gr.Button("📁 Open Data Folder")
cleanup_btn = gr.Button("🧹 Clean Up Corrupted Files")
data_management_output = gr.Textbox(label="Data Management Status", lines=2)
gr.Markdown("#### 🖥️ System Information")
system_info = gr.Markdown(self.get_system_info())
refresh_info_btn = gr.Button("🔄 Refresh System Info")
# Keyword management
reset_keywords_btn.click(
fn=self.reset_custom_keywords,
outputs=[keyword_status]
)
# Data management
open_data_btn.click(
fn=self.open_data_folder,
outputs=[data_management_output]
)
cleanup_btn.click(
fn=self.cleanup_corrupted_files,
outputs=[data_management_output]
)
refresh_info_btn.click(
fn=self.get_system_info,
outputs=[system_info]
)
# Video controls
refresh_videos_btn.click(
fn=self.get_video_list,
outputs=[available_videos]
)
# About Tab
with gr.Tab("ℹ️ About"):
gr.Markdown("""
## About NASA Solar Image Downloader
This comprehensive web application downloads and processes images from NASA's Solar Dynamics Observatory (SDO).
### 🌟 Features
- **Download Management**: Bulk download solar images for any date range
- **Image Viewer**: Browse images with full playback controls (First, Previous, Next, Last)
- **Video Creation**: Create time-lapse MP4 videos with customizable FPS
- **Multiple Filters**: 12 different wavelengths and composite filters
- **High Resolution**: Support for 1024, 2048, and 4096 pixel images
- **Custom Keywords**: Advanced search customization
- **Progress Tracking**: Real-time progress for all operations
### 🔬 Solar Filters Explained
**Individual Wavelengths:**
- **193 Å**: Shows coronal loops and hot active regions
- **304 Å**: Reveals the chromosphere and filament channels
- **171 Å**: Displays quiet corona and coronal holes
- **211 Å**: Highlights active regions and hot plasma
- **131 Å**: Shows flaring regions and very hot plasma
- **335 Å**: Reveals active region cores
- **94 Å**: Shows extremely hot plasma and flare ribbons
- **1600 Å**: Displays transition region and upper photosphere
- **1700 Å**: Shows temperature minimum and photosphere
**Composite Filters:**
- **094+335+193**: Multi-wavelength view of hot plasma structures
- **304+211+171**: Comprehensive view from chromosphere to corona
- **211+193+171**: Active regions with coronal context
### 🚀 Quick Start Guide
1. **Download Images**: Select date range, choose filter, click download
2. **View Images**: Load images and use playback controls to browse
3. **Create Videos**: Set date range and FPS, create time-lapse videos
4. **Customize**: Use Settings tab for advanced configuration
### 📊 Data Source
All images are sourced from NASA's Solar Dynamics Observatory (SDO), which provides continuous observations of the Sun in multiple wavelengths.
**🏆 Created by Andy Kong**
---
*This application provides the same functionality as the desktop version but accessible from any web browser.*
""")
return app
def launch(self, share=False, server_port=7860):
"""Launch the Gradio interface."""
app = self.create_interface()
app.launch(share=share, server_port=server_port, theme=gr.themes.Soft())
def main():
"""Main application entry point."""
try:
print("🚀 Starting NASA Solar Image Downloader (Gradio Web Interface)")
print("=" * 60)
# Create and launch the Gradio app
gradio_app = NASADownloaderGradio()
# Launch with share=True to create a public link
# Set share=False for local-only access
gradio_app.launch(share=False, server_port=None) # Auto-find available port
except Exception as e:
print(f"❌ Error starting application: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()