updated files

app.py

@@ -0,0 +1,222 @@
+import gradio as gr
+import json
+from mistralai import Mistral
+from pydantic import BaseModel, Field
+from datetime import datetime
+import base64
+from io import BytesIO
+from PIL import Image
+
+class ClimateData(BaseModel):
+    document_type: str = Field(..., description="Type of document: report, policy, research, assessment, etc.")
+    title: str = Field(..., description="Document title or main heading")
+    organization: str = Field(..., description="Publishing organization or agency")
+    publication_date: str = Field(..., description="Publication or release date")
+    temperature_data: list[str] = Field(default=[], description="Temperature readings, anomalies, projections")
+    precipitation_data: list[str] = Field(default=[], description="Precipitation measurements and forecasts")
+    co2_levels: list[str] = Field(default=[], description="CO2 concentration data and emissions")
+    sea_level_data: list[str] = Field(default=[], description="Sea level rise measurements")
+    extreme_events: list[str] = Field(default=[], description="Extreme weather events and frequencies")
+    year_ranges: list[str] = Field(default=[], description="Time periods and date ranges covered")
+    baseline_periods: list[str] = Field(default=[], description="Reference or baseline periods used")
+    projection_periods: list[str] = Field(default=[], description="Future projection timeframes")
+    policy_recommendations: list[str] = Field(default=[], description="Policy recommendations and actions")
+    targets_goals: list[str] = Field(default=[], description="Climate targets, goals, and commitments")
+    mitigation_strategies: list[str] = Field(default=[], description="Mitigation approaches and strategies")
+    adaptation_measures: list[str] = Field(default=[], description="Adaptation measures and plans")
+    regions_covered: list[str] = Field(default=[], description="Geographical regions or countries covered")
+    sectors_affected: list[str] = Field(default=[], description="Economic sectors or systems affected")
+    main_conclusions: list[str] = Field(default=[], description="Primary conclusions and findings")
+    risk_assessments: list[str] = Field(default=[], description="Risk levels and assessments")
+    uncertainty_levels: list[str] = Field(default=[], description="Uncertainty ranges and confidence levels")
+
+class ChartDescription(BaseModel):
+    chart_type: str = Field(..., description="Type of visualization: line chart, bar chart, map, table, etc.")
+    data_type: str = Field(..., description="Type of data shown: temperature, emissions, policy timeline, etc.")
+    trend_description: str = Field(..., description="Description of trends, patterns, and changes")
+    key_insights: str = Field(..., description="Important findings and takeaways from the visualization")
+    time_period: str = Field(..., description="Time period or range covered in the chart")
+    geographical_scope: str = Field(..., description="Geographical area or regions shown")
+
+def initialize_client(api_key):
+    if not api_key:
+        raise ValueError("Please provide a valid Mistral API key")
+    return Mistral(api_key=api_key)
+
+def extract_climate_data(api_key, file_path=None, url=None):
+    try:
+        client = initialize_client(api_key)
+        from mistralai.extra import response_format_from_pydantic_model
+        if file_path:
+            uploaded_file = client.files.upload(
+                file={"file_name": os.path.basename(file_path), "content": open(file_path, "rb")},
+                purpose="ocr"
+            )
+            signed_url = client.files.get_signed_url(file_id=uploaded_file.id)
+            document_url = signed_url.url
+        elif url:
+            document_url = url
+        else:
+            return {"error": "No file or URL provided"}
+        response = client.ocr.process(
+            model="mistral-ocr-latest",
+            document={"type": "document_url", "document_url": document_url},
+            bbox_annotation_format=response_format_from_pydantic_model(ChartDescription),
+            document_annotation_format=response_format_from_pydantic_model(ClimateData),
+            include_image_base64=True
+        )
+        extracted_text = response.text if hasattr(response, 'text') else ""
+        bbox_annotations = response.bbox_annotations if hasattr(response, 'bbox_annotations') else []
+        doc_annotations = response.document_annotation if hasattr(response, 'document_annotation') else {}
+        return {
+            "success": True,
+            "extracted_text": extracted_text,
+            "climate_data": doc_annotations,
+            "chart_descriptions": bbox_annotations,
+            "raw_response": str(response)
+        }
+    except Exception as e:
+        return {"error": f"OCR processing failed: {str(e)}"}
+
+def process_climate_document(api_key, file, url_input):
+    if file:
+        result = extract_climate_data(api_key, file_path=file.name)
+    elif url_input.strip():
+        result = extract_climate_data(api_key, url=url_input.strip())
+    else:
+        return "Please provide either a file or URL"
+    if "error" in result:
+        return f"Error: {result['error']}"
+    output = "# Climate Document Analysis Results\n\n"
+    if result.get("climate_data"):
+        data = result['climate_data']
+        if isinstance(data, dict):
+            output += f"## Document Overview:\n"
+            output += f"**Type:** {data.get('document_type', 'N/A')}\n"
+            output += f"**Title:** {data.get('title', 'N/A')}\n"
+            output += f"**Organization:** {data.get('organization', 'N/A')}\n"
+            output += f"**Date:** {data.get('publication_date', 'N/A')}\n\n"
+            output += "## Complete Structured Data:\n"
+            output += f"```json\n{json.dumps(data, indent=2)}\n```\n\n"
+        else:
+            output += "## Extracted Climate Data:\n"
+            output += f"```\n{str(data)}\n```\n\n"
+    if result.get("chart_descriptions"):
+        output += "## Chart Analysis:\n"
+        charts = result['chart_descriptions']
+        if isinstance(charts, list):
+            for i, chart in enumerate(charts, 1):
+                if isinstance(chart, dict):
+                    output += f"### Chart {i}:\n{json.dumps(chart, indent=2)}\n\n"
+                else:
+                    output += f"### Chart {i}:\n{str(chart)}\n\n"
+        else:
+            output += f"```\n{str(charts)}\n```\n\n"
+    if result.get("extracted_text"):
+        output += "## Extracted Text:\n"
+        output += f"{result['extracted_text']}...\n\n"
+    return output
+
+def analyze_image(api_key, image):
+    try:
+        client = initialize_client(api_key)
+        buffered = BytesIO()
+        image.save(buffered, format="PNG")
+        img_str = base64.b64encode(buffered.getvalue()).decode()
+        prompt = """Analyze this image and provide a JSON output with the following fields:
+        - image_type: Type of image (e.g., satellite, ground, aerial)
+        - climate_feature: Primary climate feature observed (e.g., cloud_cover, precipitation)
+        - location: Estimated or general location (e.g., Pacific Ocean, Sahara Desert)
+        - date_captured: Current date in YYYY-MM-DD format
+        - cloud_density: Estimated cloud density (0.0 to 1.0) if applicable
+        - temperature_anomaly: Estimated temperature anomaly in Celsius (e.g., 1.2)
+        - description: Brief description of the image content
+        """
+        response = client.chat.complete(
+            model="pixtral-large-latest",
+            messages=[
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": prompt},
+                        {"type": "image_url", "image_url": f"data:image/png;base64,{img_str}"}
+                    ]
+                }
+            ]
+        )
+        response_text = response.choices[0].message.content
+        try:
+            response_text = response_text.replace("```json", "").replace("```", "").strip()
+            result = json.loads(response_text)
+        except json.JSONDecodeError:
+            result = {
+                "image_type": "unknown",
+                "climate_feature": "unknown",
+                "location": "unknown",
+                "date_captured": datetime.now().strftime("%Y-%m-DD"),
+                "cloud_density": 0.0,
+                "temperature_anomaly": 0.0,
+                "description": "Error parsing model output."
+            }
+        return json.dumps(result, indent=2)
+    except Exception as e:
+        error_result = {
+            "image_type": "error",
+            "climate_feature": "none",
+            "location": "none",
+            "date_captured": datetime.now().strftime("%Y-%m-DD"),
+            "cloud_density": 0.0,
+            "temperature_anomaly": 0.0,
+            "description": f"Error processing image: {str(e)}"
+        }
+        return json.dumps(error_result, indent=2)
+
+with gr.Blocks(title="Climate Data and Image Analyzer") as demo:
+    gr.Markdown("# 🌍 Climate Data and Image Analysis Tool\nAnalyze climate documents or images using Mistral OCR and Pixtral models")
+    api_key_input = gr.Textbox(
+        label="Mistral API Key",
+        placeholder="Enter your Mistral API key here",
+        type="password"
+    )
+    with gr.Tabs():
+        with gr.Tab(label="Document Analysis"):
+            gr.Markdown("## Document Analysis\nExtract data from climate reports, policies, or research papers")
+            with gr.Row():
+                with gr.Column():
+                    file_input = gr.File(
+                        label="Upload Climate Document",
+                        file_types=[".pdf", ".png", ".jpg", ".jpeg", ".docx", ".pptx"]
+                    )
+                    url_input = gr.Textbox(
+                        label="Or Enter Document URL",
+                        placeholder="https://example.com/climate-policy.pdf"
+                    )
+                    process_btn = gr.Button("Analyze Document", variant="primary")
+                with gr.Column():
+                    doc_output = gr.Markdown(label="Document Analysis Results")
+            process_btn.click(
+                fn=process_climate_document,
+                inputs=[api_key_input, file_input, url_input],
+                outputs=doc_output
+            )
+            gr.Examples(
+                examples=[
+                    [None, "https://static.pib.gov.in/WriteReadData/specificdocs/documents/2021/dec/doc202112101.pdf"],
+                    [None, "https://www.ipcc.ch/site/assets/uploads/2018/02/WG1AR5_Chapter02_FINAL.pdf"],
+                    [None, "https://unfccc.int/sites/default/files/resource/parisagreement_publication.pdf"]
+                ],
+                inputs=[file_input, url_input]
+            )
+        with gr.Tab(label="Image Analysis"):
+            gr.Markdown("## Image Analysis\nAnalyze climate-related images for features like cloud cover or temperature anomalies")
+            image_input = gr.Image(type="pil", label="Upload Image")
+            image_btn = gr.Button("Analyze Image", variant="primary")
+            image_output = gr.JSON(label="Image Analysis Result")
+            image_btn.click(
+                fn=analyze_image,
+                inputs=[api_key_input, image_input],
+                outputs=image_output
+            )
+
+if __name__ == "__main__":
+    demo.launch(mcp_server=True)