Update app.py

app.py

@@ -3,7 +3,13 @@ import numpy as np
 import cv2
 from PIL import Image
 import matplotlib.pyplot as plt
+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
@@ -12,6 +18,42 @@ 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:
+            # Format URL: https://sdo.gsfc.nasa.gov/assets/img/browse/YEAR/MONTH/DAY/DATE_IDENT_SIZE_TOOL.jpg
+            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")
+            
+            # Fetch image
+            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:
@@ -30,16 +72,10 @@ def extract_frames(gif_path):
 
 def preprocess_frame(frame, lower_bound, upper_bound, morph_iterations):
     """Preprocess a frame: isolate mid-to-light pixels and enhance circular patterns."""
-    # Apply Gaussian blur to reduce noise
     blurred = cv2.GaussianBlur(frame, (9, 9), 0)
-
-    # Isolate mid-to-light pixels using user-defined intensity range
     mask = cv2.inRange(blurred, lower_bound, upper_bound)
-
-    # Apply morphological operation to enhance circular patterns
     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):
@@ -47,17 +83,15 @@ def detect_circles(frame_diff, image_center, center_tolerance, param1, param2, m
     circles = cv2.HoughCircles(
         frame_diff,
         cv2.HOUGH_GRADIENT,
-        dp=1.5,  # Resolution for better detection
-        minDist=100,  # Prevent overlapping circles
-        param1=param1,  # User-defined edge threshold
-        param2=param2,  # User-defined accumulator threshold
+        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")
-        # Filter circles: only keep those centered near the image center
         filtered_circles = []
         for (x, y, r) in circles:
             if (abs(x - image_center[0]) < center_tolerance and 
@@ -66,54 +100,50 @@ def detect_circles(frame_diff, image_center, center_tolerance, param1, param2, m
         return filtered_circles if filtered_circles else None
     return None
 
-def analyze_gif(gif_file, lower_bound, upper_bound, param1, param2, center_tolerance, morph_iterations, min_rad):
-    """Analyze a GIF for concentric circles, highlighting growing series."""
-    try:
-        # Handle Gradio file input
-        gif_path = gif_file.name if hasattr(gif_file, 'name') else gif_file
-
-        # Extract frames
-        frames, error = extract_frames(gif_path)
-        if error:
-            return error, []
+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
 
-        if len(frames) < 2:
-            return "GIF must have at least 2 frames for analysis.", []
+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 the image center (Sun's position)
+        # Determine image center
         height, width = frames[0].shape
-        image_center = (width // 2, height // 2)  # Assume Sun is at the center
-
-        # Initialize results
-        all_circle_data = []  # Store all detected circles
+        image_center = (width // 2, height // 2)
         min_radius = int(min_rad)
-        max_radius = min(height, width) // 2  # Limit max radius to half the image size
+        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)
-
-            # Compute absolute difference between consecutive frames
             frame_diff = cv2.absdiff(frame2, frame1)
-            # Enhance contrast for the difference image
             frame_diff = cv2.convertScaleAbs(frame_diff, alpha=3.0, beta=0)
-
-            # Detect circles centered at the Sun
             circles = detect_circles(frame_diff, image_center, center_tolerance, param1, param2, min_radius, max_radius)
-
+            
             if circles:
-                # Take the largest circle (most prominent CME feature)
-                largest_circle = max(circles, key=lambda c: c[2])  # Sort by radius
+                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]  # Store the frame for visualization
+                    "output_frame": frames[i + 1]
                 })
 
-        # Find the longest series of frames where the circle is growing
+        # Find growing series
         growing_circle_data = []
         current_series = []
         if all_circle_data:
@@ -122,75 +152,141 @@ def analyze_gif(gif_file, lower_bound, upper_bound, param1, param2, center_toler
                 if all_circle_data[i]["radius"] > current_series[-1]["radius"]:
                     current_series.append(all_circle_data[i])
                 else:
-                    # If the radius doesn't increase, check if the current series is the longest
                     if len(current_series) > len(growing_circle_data):
                         growing_circle_data = current_series.copy()
                     current_series = [all_circle_data[i]]
-
-            # Check the last series
             if len(current_series) > len(growing_circle_data):
                 growing_circle_data = current_series.copy()
 
-        # Mark frames that are part of the growing series
         growing_frames = set(c["frame"] for c in growing_circle_data)
-
-        # Generate output frames and report
         results = []
-        report = "Analysis Report (as of 08:07 PM PDT, May 24, 2025):\n"
-        
-        # Report all detected circles
-        if all_circle_data:
-            report += f"\nAll Frames with Detected Circles ({len(all_circle_data)} frames):\n"
+        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":
+            for i, frame in enumerate(frames):
+                output_frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2RGB)
+                if i + 1 in growing_frames:
+                    for c in all_circle_data:
+                        if c["frame"] == i + 1:
+                            cv2.circle(output_frame, c["center"], c["radius"], (0, 255, 0), 2)
+                            cv2.circle(output_frame, c["center"], c["radius"] + 2, (255, 0, 0), 2)
+                results.append(Image.fromarray(output_frame))
+        elif display_mode == "Detected Frames":
             for c in all_circle_data:
-                # Visualize the frame with detected circle (green)
                 output_frame = cv2.cvtColor(c["output_frame"], cv2.COLOR_GRAY2RGB)
-                cv2.circle(output_frame, (c["center"][0], c["center"][1]), c["radius"], (0, 255, 0), 2)
-
-                # If the frame is part of the growing series, add a red circle
+                cv2.circle(output_frame, c["center"], c["radius"], (0, 255, 0), 2)
                 if c["frame"] in growing_frames:
-                    cv2.circle(output_frame, (c["center"][0], c["center"][1]), c["radius"] + 2, (255, 0, 0), 2)
-
-                # Convert to PIL Image for Gradio
-                output_frame = Image.fromarray(output_frame)
-                results.append(output_frame)
+                    cv2.circle(output_frame, c["center"], c["radius"] + 2, (255, 0, 0), 2)
+                results.append(Image.fromarray(output_frame))
+        elif display_mode == "Both (Detected Replaces Original)":
+            for i, frame in enumerate(frames):
+                output_frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2RGB)
+                if i + 1 in growing_frames:
+                    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)
+                            cv2.circle(output_frame, c["center"], c["radius"] + 2, (255, 0, 0), 2)
+                results.append(Image.fromarray(output_frame))
 
+        # 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"
 
-        # Report the growing series
         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 of mid-to-light pixels detected, indicative of a potential Earth-directed CME."
+            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."
 
-        return report, results
+        # 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)}", []
-
-# Gradio interface with user controls
-iface = gr.Interface(
-    fn=analyze_gif,
-    inputs=[
-        gr.File(label="Upload Solar GIF", file_types=[".gif"]),
-        gr.Slider(minimum=0, maximum=255, value=low_int, step=1, label="Lower Intensity Bound (0-255)"),
-        gr.Slider(minimum=0, maximum=255, value=high_int, step=1, label="Upper Intensity Bound (0-255)"),
-        gr.Slider(minimum=10, maximum=200, value=edge_thresh, step=1, label="Hough Param1 (Edge Threshold)"),
-        gr.Slider(minimum=1, maximum=50, value=accum_thresh, step=1, label="Hough Param2 (Accumulator Threshold)"),
-        gr.Slider(minimum=10, maximum=100, value=center_tol, step=1, label="Center Tolerance (Pixels)"),
-        gr.Slider(minimum=1, maximum=5, value=morph_dia, step=1, label="Morphological Dilation Iterations"),
-        gr.Slider(minimum=1, maximum=100, value=min_rad, step=1, label="Minimum Circle Radius")
-    ],
-    outputs=[
-        gr.Textbox(label="Analysis Report"),
-        gr.Gallery(label="Frames with Detected Circles (Green: Detected, Red: Growing Series)")
-    ],
-    title="Solar CME Detection",
-    description="Upload a GIF of solar images to detect concentric circles of mid-to-light pixels. All detected circles are shown in green, and the series of growing circles (indicating an Earth-directed CME) are highlighted in red. Adjust the sliders to fine-tune detection."
-)
+        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):
+    """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, error = fetch_sdo_images(start_date, end_date, ident, size, tool)
+        if error:
+            return error, [], None
+    
+    # Preview first frame if available
+    preview = Image.fromarray(frames[0]) if frames else None
+    
+    # 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 specify a date range to fetch SDO images and detect concentric circles indicative of coronal mass ejections (CMEs). 
+    Green circles mark detected features; red circles highlight growing series (potential Earth-directed CMEs).
+    """)
+    
+    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")
+            
+            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.Image(label="Input Preview (First Frame)")
+            gallery = gr.Gallery(label="Frames with Detected Circles (Green: Detected, Red: Growing Series)")
+            gif_output = gr.File(label="Download Resulting GIF")
+    
+    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
+        ],
+        outputs=[report, gallery, gif_output, preview]
+    )
 
 if __name__ == "__main__":
-    iface.launch()
+    demo.launch()