app.py

import gradio as gr
import cv2
import numpy as np
from scipy.spatial import distance
import requests
from bs4 import BeautifulSoup
from datetime import datetime, timedelta
import urllib.request
from PIL import Image
import io# Function to fetch image filenames from SDO website for today and yesterday
def get_sdo_images():
    base_url = "https://sdo.gsfc.nasa.gov/assets/img/browse/"
    today = datetime.now()
    yesterday = today - timedelta(days=1)
    dates = [today, yesterday]
    image_list = []

    for date in dates:
        date_str = date.strftime("%Y/%m/%d")
        url = f"{base_url}{date_str}/"
        try:
            response = requests.get(url)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, 'html.parser')
            for link in soup.find_all('a'):
                href = link.get('href')
                if href and href.endswith('_1024_HMIBC.jpg'):
                    filename = href.split('/')[-1].replace('_1024_HMIBC.jpg', '')
                    image_list.append((filename, f"{url}{href}"))
        except requests.RequestException as e:
            print(f"Error fetching images for {date_str}: {e}")
    
    image_list.sort(reverse=True, key=lambda x: x[0])
    filenames = [item[0] for item in image_list]
    url_map = {item[0]: item[1] for item in image_list}
    return filenames, url_map

# Function to fetch the selected image
def fetch_image(selected_image, url_map):
    if not selected_image:
        return None
    image_url = url_map.get(selected_image)
    if not image_url:
        return None
    try:
        with urllib.request.urlopen(image_url) as response:
            img_data = response.read()
        img = Image.open(io.BytesIO(img_data))
        return img
    except Exception as e:
        print(f"Error fetching image {selected_image}: {e}")
        return None# Function to analyze the solar image with adjustable parameters
def analyze_solar_image(
    image,
    yellow_hue_min=20, yellow_hue_max=30, yellow_sat_min=100, yellow_val_min=100,
    grey_hue_max=180, grey_sat_max=30, grey_val_min=50, grey_val_max=200,
    blue_hue_min=90, blue_hue_max=150, blue_sat_min=50, blue_val_min=30,
    red_hue_min1=0, red_hue_max1=15, red_hue_min2=155, red_hue_max2=180,
    red_sat_min=50, red_val_min=30,
    flare_dist_threshold=50, min_area_threshold=10, darkness_weight=0.5
):
    if image is None:
        return None, "No image selected or image could not be loaded."

    image = np.array(image)
    image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
    hsv_image = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)
    
    yellow_lower = np.array([yellow_hue_min, yellow_sat_min, yellow_val_min])
    yellow_upper = np.array([yellow_hue_max, 255, 255])
    grey_lower = np.array([0, 0, grey_val_min])
    grey_upper = np.array([grey_hue_max, grey_sat_max, grey_val_max])
    blue_lower = np.array([blue_hue_min, blue_sat_min, blue_val_min])
    blue_upper = np.array([blue_hue_max, 255, 255])
    red_lower1 = np.array([red_hue_min1, red_sat_min, red_val_min])
    red_lower2 = np.array([red_hue_min2, red_sat_min, red_val_min])
    red_upper1 = np.array([red_hue_max1, 255, 255])
    red_upper2 = np.array([red_hue_max2, 255, 255])
    
    mask_yellow = cv2.inRange(hsv_image, yellow_lower, yellow_upper)
    mask_grey = cv2.inRange(hsv_image, grey_lower, grey_upper)
    mask_blue = cv2.inRange(hsv_image, blue_lower, blue_upper)
    mask_red1 = cv2.inRange(hsv_image, red_lower1, red_upper1)
    mask_red2 = cv2.inRange(hsv_image, red_lower2, red_upper2)
    mask_red = cv2.bitwise_or(mask_red1, mask_red2)
    
    blue_contours, _ = cv2.findContours(mask_blue, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    red_contours, _ = cv2.findContours(mask_red, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    
    sunspots = []
    used_blue = set()
    used_red = set()
    
    for i, bc in enumerate(blue_contours):
        if i in used_blue:
            continue
        bc_area = cv2.contourArea(bc)
        if bc_area < min_area_threshold:
            continue
    
        bc_centroid = None
        M = cv2.moments(bc)
        if M["m00"] != 0:
            bcX = int(M["m10"] / M["m00"])
            bcY = int(M["m01"] / M["m00"])
            bc_centroid = (bcX, bcY)
    
        if bc_centroid is None:
            continue
    
        sunspot_contours = [bc]
        used_blue.add(i)
    
        for j, rc in enumerate(red_contours):
            if j in used_red:
                continue
            rc_area = cv2.contourArea(rc)
            if rc_area < min_area_threshold:
                continue
    
            M = cv2.moments(rc)
            if M["m00"] != 0:
                rcX = int(M["m10"] / M["m00"])
                rcY = int(M["m01"] / M["m00"])
                rc_centroid = (rcX, rcY)
                dist = distance.euclidean(bc_centroid, rc_centroid)
                if dist < flare_dist_threshold:
                    sunspot_contours.append(rc)
                    used_red.add(j)
    
        for k, bc2 in enumerate(blue_contours):
            if k in used_blue or k == i:
                continue
            bc2_area = cv2.contourArea(bc2)
            if bc2_area < min_area_threshold:
                continue
    
            M = cv2.moments(bc2)
            if M["m00"] != 0:
                bc2X = int(M["m10"] / M["m00"])
                bc2Y = int(M["m01"] / M["m00"])
                bc2_centroid = (bc2X, bc2Y)
                dist = distance.euclidean(bc_centroid, bc2_centroid)
                if dist < flare_dist_threshold:
                    sunspot_contours.append(bc2)
                    used_blue.add(k)
    
        for j, rc in enumerate(red_contours):
            if j in used_red:
                continue
            rc_area = cv2.contourArea(rc)
            if rc_area < min_area_threshold:
                continue
    
            M = cv2.moments(rc)
            if M["m00"] != 0:
                rcX = int(M["m10"] / M["m00"])
                rcY = int(M["m01"] / M["m00"])
                rc_centroid = (rcX, rcY)
                dist = distance.euclidean(bc_centroid, rc_centroid)
                if dist < flare_dist_threshold:
                    sunspot_contours.append(rc)
                    used_red.add(j)
    
        sunspots.append(sunspot_contours)
    
    for j, rc in enumerate(red_contours):
        if j in used_red:
            continue
        rc_area = cv2.contourArea(rc)
        if rc_area >= min_area_threshold:
            sunspots.append([rc])
            used_red.add(j)
    
    sunspot_image = image.copy()
    report = "Solar Image Analysis Report\n\n"
    overall_flare_risk = "Low"
    overall_flare_percentage = 0
    mixed_sunspot_count = 0
    
    for i, sunspot_contours in enumerate(sunspots):
        all_points = np.concatenate(sunspot_contours)
        x, y, w, h = cv2.boundingRect(all_points)
        cv2.rectangle(sunspot_image, (x, y), (x+w, y+h), (0, 255, 0), 2)
        cX, cY = x + w//2, y + h//2
    
        sunspot_region = np.zeros_like(mask_blue)
        cv2.drawContours(sunspot_region, sunspot_contours, -1, 255, thickness=cv2.FILLED)
    
        blue_in_sunspot = cv2.bitwise_and(mask_blue, sunspot_region)
        red_in_sunspot = cv2.bitwise_and(mask_red, sunspot_region)
        blue_contours_in_sunspot, _ = cv2.findContours(blue_in_sunspot, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        red_contours_in_sunspot, _ = cv2.findContours(red_in_sunspot, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    
        blue_area = sum(cv2.contourArea(bc) for bc in blue_contours_in_sunspot)
        red_area = sum(cv2.contourArea(rc) for rc in red_contours_in_sunspot)
    
        blue_darkness = 0
        red_darkness = 0
        if blue_contours_in_sunspot:
            blue_mask = np.zeros_like(hsv_image[..., 2])
            cv2.drawContours(blue_mask, blue_contours_in_sunspot, -1, 255, thickness=cv2.FILLED)
            blue_values = hsv_image[..., 2][blue_mask == 255]
            if len(blue_values) > 0:
                blue_darkness = 255 - np.mean(blue_values)
        if red_contours_in_sunspot:
            red_mask = np.zeros_like(hsv_image[..., 2])
            cv2.drawContours(red_mask, red_contours_in_sunspot, -1, 255, thickness=cv2.FILLED)
            red_values = hsv_image[..., 2][red_mask == 255]
            if len(red_values) > 0:
                red_darkness = 255 - np.mean(red_values)
    
        blue_centroids = []
        red_centroids = []
        min_dist = float('inf')
        mixing_score = 0
        for bc in blue_contours_in_sunspot:
            M = cv2.moments(bc)
            if M["m00"] != 0:
                bcX = int(M["m10"] / M["m00"])
                bcY = int(M["m01"] / M["m00"])
                blue_centroids.append((bcX, bcY))
    
        for rc in red_contours_in_sunspot:
            M = cv2.moments(rc)
            if M["m00"] != 0:
                rcX = int(M["m10"] / M["m00"])
                rcY = int(M["m01"] / M["m00"])
                red_centroids.append((rcX, rcY))
    
        if blue_centroids and red_centroids:
            for bc in blue_centroids:
                for rc in red_centroids:
                    dist = distance.euclidean(bc, rc)
                    min_dist = min(min_dist, dist)
            mixing_score = 1000 / (min_dist + 1)
    
        darkness_score = (blue_darkness + red_darkness) / 2 if blue_darkness and red_darkness else max(blue_darkness, red_darkness)
        flare_risk_percentage = 0
        risk_level = "Low"
    
        if blue_centroids and red_centroids:
            flare_risk_score = (1 - darkness_weight) * (flare_dist_threshold - min_dist) / flare_dist_threshold + darkness_weight * (darkness_score / 255)
            flare_risk_percentage = flare_risk_score * 100
            mixed_sunspot_count += 1
            overall_flare_percentage = (overall_flare_percentage * (mixed_sunspot_count - 1) + flare_risk_percentage) / mixed_sunspot_count
            if flare_risk_percentage >= 70:
                risk_level = "High"
                overall_flare_risk = "High"
            elif flare_risk_percentage >= 40:
                risk_level = "Medium"
                if overall_flare_risk != "High":
                    overall_flare_risk = "Medium"
            else:
                if overall_flare_risk not in ["High", "Medium"]:
                    overall_flare_risk = "Low"
        else:
            flare_risk_percentage = 0
            risk_level = "Low"
    
        line_height = 30
        y = 0
        lines = [f"Sunspot {i+1}", f"Risk: {risk_level}, {flare_risk_percentage:.1f}%"]
        for line in lines:
            cv2.putText(sunspot_image, line, (cX, cY + y), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 2)
            y += line_height
    
        for bc in blue_contours_in_sunspot:
            cv2.drawContours(sunspot_image, [bc], -1, (255, 0, 0), 2)
        for rc in red_contours_in_sunspot:
            cv2.drawContours(sunspot_image, [rc], -1, (0, 0, 255), 2)
    
        report += f"Sunspot {i+1}:\n"
        report += f"  Blue Area: {blue_area:.2f} pixels\n"
        report += f"  Red Area: {red_area:.2f} pixels\n"
        report += f"  Minimum Distance Between Blue and Red: {min_dist:.2f} pixels\n"
        report += f"  Mixing Score (Higher = Closer): {mixing_score:.2f}\n"
        report += f"  Blue Darkness (Higher = Darker): {blue_darkness:.2f}\n"
        report += f"  Red Darkness (Higher = Darker): {red_darkness:.2f}\n"
        report += f"  Flare Risk: {risk_level}, {flare_risk_percentage:.1f}%\n\n"
    
    if mixed_sunspot_count == 0:
        overall_flare_percentage = 0
        overall_flare_risk = "Low"
    report += f"Overall Flare Risk: {overall_flare_risk}, {overall_flare_percentage:.1f}%\n"
    
    sunspot_image = cv2.cvtColor(sunspot_image, cv2.COLOR_BGR2RGB)
    return sunspot_image, report

# Gradio interface with adjustable controls
def create_interface():
    with gr.Blocks() as interface:
        # State to store image options and URL map
        image_options = gr.State(value=[])
        url_map = gr.State(value={})

        gr.Markdown("# Solar Image Analysis for Sunspots and Flare Prediction")
        gr.Markdown("Select a solar image from the dropdown or upload your own to detect sunspots, identify blue and red regions, and predict solar flare potential based on magnetic mixing. Adjust the sliders to fine-tune detection.")
        
        with gr.Sidebar(open=False):
            yellow_hue_min = gr.Slider(minimum=0, maximum=180, value=20, label="Yellow Hue Min")
            yellow_hue_max = gr.Slider(minimum=0, maximum=180, value=30, label="Yellow Hue Max")
            yellow_sat_min = gr.Slider(minimum=0, maximum=255, value=100, label="Yellow Saturation Min")
            yellow_val_min = gr.Slider(minimum=0, maximum=255, value=100, label="Yellow Value Min")
            grey_hue_max = gr.Slider(minimum=0, maximum=180, value=180, label="Grey Hue Max")
            grey_sat_max = gr.Slider(minimum=0, maximum=255, value=30, label="Grey Saturation Max")
            grey_val_min = gr.Slider(minimum=0, maximum=255, value=50, label="Grey Value Min")
            grey_val_max = gr.Slider(minimum=0, maximum=255, value=200, label="Grey Value Max")
            blue_hue_min = gr.Slider(minimum=0, maximum=180, value=90, label="Blue Hue Min")
            blue_hue_max = gr.Slider(minimum=0, maximum=180, value=150, label="Blue Hue Max")
            blue_sat_min = gr.Slider(minimum=0, maximum=255, value=50, label="Blue Saturation Min")
            blue_val_min = gr.Slider(minimum=0, maximum=255, value=30, label="Blue Value Min")
            red_hue_min1 = gr.Slider(minimum=0, maximum=180, value=0, label="Red Hue Min 1")
            red_hue_max1 = gr.Slider(minimum=0, maximum=180, value=15, label="Red Hue Max 1")
            red_hue_min2 = gr.Slider(minimum=0, maximum=180, value=100, label="Red Hue Min 2")
            red_hue_max2 = gr.Slider(minimum=0, maximum=180, value=180, label="Red Hue Max 2")
            red_sat_min = gr.Slider(minimum=0, maximum=255, value=50, label="Red Saturation Min")
            red_val_min = gr.Slider(minimum=0, maximum=255, value=30, label="Red Value Min")
            flare_dist_threshold = gr.Slider(minimum=0, maximum=200, value=50, label="Flare Distance Threshold (pixels)")
            min_area_threshold = gr.Slider(minimum=0, maximum=100, value=1, label="Minimum Area Threshold (pixels)")
            darkness_weight = gr.Slider(minimum=0, maximum=1, value=0.5, label="Darkness Weight in Flare Risk (0 to 1)")
        
        with gr.Row():
            with gr.Column():
                update_button = gr.Button("Update Image List")
                image_dropdown = gr.Dropdown(label="Select SDO Image (YYYYMMDD_HHMMSS)", choices=[])
                fetch_button = gr.Button("Fetch Selected Image")
                image_input = gr.Image(type="pil", label="Solar Image")
                analyze_button = gr.Button("Analyze Image")
            
            with gr.Column():
                analyzed_image = gr.Image(type="pil", label="Analyzed Image")
                report = gr.Textbox(label="Analysis Report")

        # Function to update dropdown choices
        def update_image_list():
            filenames, new_url_map = get_sdo_images()
            return gr.update(choices=filenames), new_url_map
    
        # Function to handle fetch button
        def fetch_and_update(selected_image, url_map):
            img = fetch_image(selected_image, url_map)
            return img
    
        # Update image list on button click
        update_button.click(
            fn=update_image_list,
            inputs=None,
            outputs=[image_dropdown, url_map]
        )
    
        # Update image list on app load
        interface.load(
            fn=update_image_list,
            inputs=None,
            outputs=[image_dropdown, url_map]
        )
    
        # Fetch button
        fetch_button.click(
            fn=fetch_and_update,
            inputs=[image_dropdown, url_map],
            outputs=image_input
        )
    
        # Analyze button
        analyze_button.click(
            fn=analyze_solar_image,
            inputs=[
                image_input,
                yellow_hue_min, yellow_hue_max, yellow_sat_min, yellow_val_min,
                grey_hue_max, grey_sat_max, grey_val_min, grey_val_max,
                blue_hue_min, blue_hue_max, blue_sat_min, blue_val_min,
                red_hue_min1, red_hue_max1, red_hue_min2, red_hue_max2,
                red_sat_min, red_val_min,
                flare_dist_threshold, min_area_threshold, darkness_weight
            ],
            outputs=[analyzed_image, report]
        )

    return interface

# Launch the interface
interface = create_interface()
interface.launch()