import gradio as gr import numpy as np import cv2 from PIL import Image import requests from datetime import datetime, timedelta import io import os from urllib.parse import urljoin # Default parameters low_int = 10 high_int = 100 edge_thresh = 50 accum_thresh = 45 center_tol = 30 morph_dia = 5 min_rad = 70 def fetch_sdo_images(start_date, end_date, ident="0171", size="1024", tool="hmiigr"): """Fetch SDO images from NASA URL for a given date range.""" try: start = datetime.strptime(start_date, "%Y-%m-%d %H:%M:%S") end = datetime.strptime(end_date, "%Y-%m-%d %H:%M:%S") if start > end: return None, "Start date must be before end date." base_url = "https://sdo.gsfc.nasa.gov/assets/img/browse/" frames = [] current = start while current <= end: date_str = current.strftime("%Y%m%d_%H%M%S") year, month, day = current.strftime("%Y"), current.strftime("%m"), current.strftime("%d") url = urljoin(base_url, f"{year}/{month}/{day}/{date_str}_{ident}_{size}_{tool}.jpg") try: response = requests.get(url, timeout=5) if response.status_code == 200: img = Image.open(io.BytesIO(response.content)).convert('L') # Convert to grayscale frames.append(np.array(img)) else: print(f"Failed to fetch {url}: Status {response.status_code}") except Exception as e: print(f"Error fetching {url}: {str(e)}") current += timedelta(minutes=12) # SDO images are typically 12 minutes apart if not frames: return None, "No images found in the specified date range." return frames, None except Exception as e: return None, f"Error fetching images: {str(e)}" def extract_frames(gif_path): """Extract frames from a GIF and return as a list of numpy arrays.""" try: img = Image.open(gif_path) frames = [] while True: frame = img.convert('L') # Convert to grayscale frames.append(np.array(frame)) try: img.seek(img.tell() + 1) except EOFError: break return frames, None except Exception as e: return None, f"Error loading GIF: {str(e)}" def preprocess_frame(frame, lower_bound, upper_bound, morph_iterations): """Preprocess a frame: isolate mid-to-light pixels and enhance circular patterns.""" blurred = cv2.GaussianBlur(frame, (9, 9), 0) mask = cv2.inRange(blurred, lower_bound, upper_bound) kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5)) enhanced = cv2.dilate(mask, kernel, iterations=morph_iterations) return enhanced def detect_circles(frame_diff, image_center, center_tolerance, param1, param2, min_radius=20, max_radius=200): """Detect circles in a frame difference image, centered at the Sun.""" circles = cv2.HoughCircles( frame_diff, cv2.HOUGH_GRADIENT, dp=1.5, minDist=100, param1=param1, param2=param2, minRadius=min_radius, maxRadius=max_radius ) if circles is not None: circles = np.round(circles[0, :]).astype("int") filtered_circles = [] for (x, y, r) in circles: if (abs(x - image_center[0]) < center_tolerance and abs(y - image_center[1]) < center_tolerance): filtered_circles.append((x, y, r)) return filtered_circles if filtered_circles else None return None def create_gif(frames, output_path, duration=0.5): """Create a GIF from a list of frames.""" pil_frames = [Image.fromarray(frame) for frame in frames] pil_frames[0].save( output_path, save_all=True, append_images=pil_frames[1:], duration=int(duration * 1000), # Duration in milliseconds loop=0 ) return output_path def handle_fetch(start_date, end_date, ident, size, tool): """Fetch SDO images and return frames for preview and state.""" frames, error = fetch_sdo_images(start_date, end_date, ident, size, tool) if error: return error, [], frames preview_frames = [Image.fromarray(frame) for frame in frames] return f"Fetched {len(frames)} images successfully.", preview_frames, frames def analyze_images(frames, lower_bound, upper_bound, param1, param2, center_tolerance, morph_iterations, min_rad, display_mode): """Analyze frames for concentric circles, highlighting growing series.""" try: if not frames or len(frames) < 2: return "At least 2 frames are required for analysis.", [], None # Determine image center height, width = frames[0].shape image_center = (width // 2, height // 2) min_radius = int(min_rad) max_radius = min(height, width) // 2 # Process frames and detect circles all_circle_data = [] for i in range(len(frames) - 1): frame1 = preprocess_frame(frames[i], lower_bound, upper_bound, morph_iterations) frame2 = preprocess_frame(frames[i + 1], lower_bound, upper_bound, morph_iterations) frame_diff = cv2.absdiff(frame2, frame1) frame_diff = cv2.convertScaleAbs(frame_diff, alpha=3.0, beta=0) circles = detect_circles(frame_diff, image_center, center_tolerance, param1, param2, min_radius, max_radius) if circles: largest_circle = max(circles, key=lambda c: c[2]) x, y, r = largest_circle all_circle_data.append({ "frame": i + 1, "center": (x, y), "radius": r, "output_frame": frames[i + 1] }) # Find growing series growing_circle_data = [] current_series = [] if all_circle_data: current_series.append(all_circle_data[0]) for i in range(1, len(all_circle_data)): if all_circle_data[i]["radius"] > current_series[-1]["radius"]: current_series.append(all_circle_data[i]) else: if len(current_series) > len(growing_circle_data): growing_circle_data = current_series.copy() current_series = [all_circle_data[i]] if len(current_series) > len(growing_circle_data): growing_circle_data = current_series.copy() growing_frames = set(c["frame"] for c in growing_circle_data) results = [] report = f"Analysis Report (as of {datetime.now().strftime('%I:%M %p PDT, %B %d, %Y')}):\n" # Prepare output based on display mode if display_mode == "All Frames": results = [Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_GRAY2RGB)) for frame in frames] elif display_mode == "Detected Frames": for c in all_circle_data: output_frame = cv2.cvtColor(c["output_frame"], cv2.COLOR_GRAY2RGB) cv2.circle(output_frame, c["center"], c["radius"], (0, 255, 0), 2) # Green for detected if c["frame"] in growing_frames: cv2.circle(output_frame, c["center"], c["radius"] + 2, (255, 165, 0), 2) # Orange for growing results.append(Image.fromarray(output_frame)) elif display_mode == "Both (Detected Replaces Original)": for i, frame in enumerate(frames): if i + 1 in [c["frame"] for c in all_circle_data]: for c in all_circle_data: if c["frame"] == i + 1: output_frame = cv2.cvtColor(c["output_frame"], cv2.COLOR_GRAY2RGB) cv2.circle(output_frame, c["center"], c["radius"], (0, 255, 0), 2) # Green if c["frame"] in growing_frames: cv2.circle(output_frame, c["center"], c["radius"] + 2, (255, 165, 0), 2) # Orange results.append(Image.fromarray(output_frame)) break else: results.append(Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_GRAY2RGB))) # Generate report if all_circle_data: report += f"\nAll Frames with Detected Circles ({len(all_circle_data)} frames):\n" for c in all_circle_data: report += f"Frame {c['frame']}: Center at {c['center']}, Radius {c['radius']} pixels\n" else: report += "No circles detected.\n" if growing_circle_data: report += f"\nSeries of Frames with Growing Circles ({len(growing_circle_data)} frames):\n" for c in growing_circle_data: report += f"Frame {c['frame']}: Center at {c['center']}, Radius {c['radius']} pixels\n" report += "\nConclusion: Growing concentric circles detected, indicative of a potential Earth-directed CME." else: report += "\nNo growing concentric circles detected. CME may not be Earth-directed." # Create GIF if results exist gif_path = None if results: gif_frames = [np.array(img) for img in results] gif_path = "output.gif" create_gif(gif_frames, gif_path) return report, results, gif_path except Exception as e: return f"Error during analysis: {str(e)}", [], None def process_input(gif_file, start_date, end_date, ident, size, tool, lower_bound, upper_bound, param1, param2, center_tolerance, morph_iterations, min_rad, display_mode, fetched_frames_state): """Process either uploaded GIF or fetched SDO images.""" if gif_file: frames, error = extract_frames(gif_file.name) if error: return error, [], None, [] else: frames = fetched_frames_state if not frames: return "No fetched frames available. Please fetch images first.", [], None, [] # Preview all frames preview = [Image.fromarray(frame) for frame in frames] if frames else [] # Analyze frames report, results, gif_path = analyze_images( frames, lower_bound, upper_bound, param1, param2, center_tolerance, morph_iterations, min_rad, display_mode ) return report, results, gif_path, preview # Gradio Blocks interface with gr.Blocks(title="Solar CME Detection") as demo: gr.Markdown(""" # Solar CME Detection Upload a GIF or fetch SDO images by date range to detect concentric circles indicative of coronal mass ejections (CMEs). Green circles mark detected features; orange circles highlight growing series (potential Earth-directed CMEs). """) # State to store fetched frames fetched_frames_state = gr.State(value=[]) with gr.Row(): with gr.Column(): gr.Markdown("### Input Options") gif_input = gr.File(label="Upload Solar GIF (optional)", file_types=[".gif"]) start_date = gr.Textbox(label="Start Date (YYYY-MM-DD HH:MM:SS)", value="2025-05-24 00:00:00") end_date = gr.Textbox(label="End Date (YYYY-MM-DD HH:MM:SS)", value="2025-05-24 23:59:59") ident = gr.Textbox(label="Image Identifier", value="0171") size = gr.Textbox(label="Image Size", value="1024") tool = gr.Textbox(label="Instrument", value="hmiigr") fetch_button = gr.Button("Fetch Images from URL") gr.Markdown("### Analysis Parameters") lower_bound = gr.Slider(minimum=0, maximum=255, value=low_int, step=1, label="Lower Intensity Bound (0-255)") upper_bound = gr.Slider(minimum=0, maximum=255, value=high_int, step=1, label="Upper Intensity Bound (0-255)") param1 = gr.Slider(minimum=10, maximum=200, value=edge_thresh, step=1, label="Hough Param1 (Edge Threshold)") param2 = gr.Slider(minimum=1, maximum=50, value=accum_thresh, step=1, label="Hough Param2 (Accumulator Threshold)") center_tolerance = gr.Slider(minimum=10, maximum=100, value=center_tol, step=1, label="Center Tolerance (Pixels)") morph_iterations = gr.Slider(minimum=1, maximum=5, value=morph_dia, step=1, label="Morphological Dilation Iterations") min_rad = gr.Slider(minimum=1, maximum=100, value=min_rad, step=1, label="Minimum Circle Radius") display_mode = gr.Dropdown( choices=["All Frames", "Detected Frames", "Both (Detected Replaces Original)"], value="Detected Frames", label="Display Mode" ) analyze_button = gr.Button("Analyze") with gr.Column(): gr.Markdown("### Outputs") report = gr.Textbox(label="Analysis Report", lines=10) preview = gr.Gallery(label="Input Preview (All Frames)") gallery = gr.Gallery(label="Frames with Detected Circles (Green: Detected, Orange: Growing Series)") gif_output = gr.File(label="Download Resulting GIF") # Fetch button action fetch_button.click( fn=handle_fetch, inputs=[start_date, end_date, ident, size, tool], outputs=[report, preview, fetched_frames_state] ) # Analyze button action analyze_button.click( fn=process_input, inputs=[ gif_input, start_date, end_date, ident, size, tool, lower_bound, upper_bound, param1, param2, center_tolerance, morph_iterations, min_rad, display_mode, fetched_frames_state ], outputs=[report, gallery, gif_output, preview] ) if __name__ == "__main__": demo.launch()