Upload app (5).py

app (5).py

@@ -0,0 +1,340 @@
+"""
+Lanternfly Field Capture Space
+A Gradio app for capturing photos with GPS coordinates and saving to Hugging Face datasets.
+"""
+
+import gradio as gr
+import os
+import json
+import uuid
+from datetime import datetime
+from PIL import Image
+from huggingface_hub import HfApi, hf_hub_download, create_repo, file_exists, upload_file
+import io
+
+# Configuration
+HF_TOKEN = os.getenv("HF_TOKEN") or os.getenv("HF_TOKEN_SPACE")  # Try both token sources
+DATASET_REPO = os.getenv("DATASET_REPO", "rlogh/lanternfly-data")  # Default to your dataset
+
+# Initialize HF API only if credentials are available
+api = None
+if HF_TOKEN and DATASET_REPO:
+    api = HfApi(token=HF_TOKEN)
+    # Ensure dataset repo exists (idempotent)
+    try:
+        create_repo(DATASET_REPO, repo_type="dataset", exist_ok=True, token=HF_TOKEN)
+        print("✅ Hugging Face credentials found - dataset saving enabled")
+    except Exception as e:
+        print(f"⚠️  Error creating dataset repo: {e}")
+        api = None
+else:
+    print("⚠️  Running in test mode - no HF credentials (dataset saving disabled)")
+
+# Constants for file paths
+METADATA_PATH = "metadata/entries.jsonl"
+IMAGES_DIR = "images"
+
+def get_current_time():
+    """Get current timestamp"""
+    return datetime.now().isoformat()
+
+def _append_jsonl_in_repo(new_row: dict) -> None:
+    """
+    Appends a JSON line to metadata/entries.jsonl in the dataset repo.
+    Downloads the existing file (if any), appends, and uploads back.
+    """
+    # Create a temp file in memory to reupload
+    # First, try to download existing entries.jsonl (if present)
+    buf = io.BytesIO()
+    existing_lines = []
+    try:
+        if file_exists(DATASET_REPO, METADATA_PATH, repo_type="dataset", token=HF_TOKEN):
+            local_path = hf_hub_download(
+                repo_id=DATASET_REPO, filename=METADATA_PATH,
+                repo_type="dataset", token=HF_TOKEN
+            )
+            with open(local_path, "r", encoding="utf-8") as f:
+                existing_lines = f.read().splitlines()
+    except Exception:
+        # If download fails for any reason, proceed as if file doesn't exist
+        existing_lines = []
+
+    # Append our new row
+    existing_lines.append(json.dumps(new_row, ensure_ascii=False))
+
+    # Write back to buffer
+    data = "\n".join(existing_lines).encode("utf-8")
+    buf.write(data); buf.seek(0)
+
+    # Upload to the same path (commit creates or updates the file)
+    upload_file(
+        path_or_fileobj=buf,
+        path_in_repo=METADATA_PATH,
+        repo_id=DATASET_REPO,
+        repo_type="dataset",
+        token=HF_TOKEN,
+        commit_message=f"Append 1 entry at {datetime.now().isoformat()}Z",
+    )
+
+def _save_image_to_repo(pil_img: Image.Image, dest_rel_path: str) -> None:
+    """
+    Uploads a PIL image into the dataset repo (e.g., images/<uuid>.jpg).
+    """
+    img_bytes = io.BytesIO()
+    pil_img.save(img_bytes, format="JPEG", quality=90)
+    img_bytes.seek(0)
+
+    upload_file(
+        path_or_fileobj=img_bytes,
+        path_in_repo=dest_rel_path,
+        repo_id=DATASET_REPO,
+        repo_type="dataset",
+        token=HF_TOKEN,
+        commit_message=f"Upload image {dest_rel_path}",
+    )
+
+def handle_time_capture():
+    """Handle time capture and return status message"""
+    timestamp = get_current_time()
+    status_msg = f"🕐 **Time Captured**: {timestamp}"
+    return status_msg, timestamp
+
+def handle_gps_location(json_str):
+    """Handle GPS location data from JavaScript and return values for the textboxes"""
+    try:
+        data = json.loads(json_str)
+        if 'error' in data:
+            status_msg = f"❌ **GPS Error**: {data['error']}"
+            return status_msg, data['error'], "", "", ""
+        
+        lat = str(data.get('latitude', ''))
+        lon = str(data.get('longitude', ''))
+        accuracy = str(data.get('accuracy', ''))
+        timestamp = data.get('timestamp', '')
+        
+        # Convert timestamp to ISO string if it's a number
+        if timestamp and isinstance(timestamp, (int, float)):
+            from datetime import datetime
+            timestamp = datetime.fromtimestamp(timestamp / 1000).isoformat()
+        
+        status_msg = f"✅ **GPS Captured**: {lat[:8]}, {lon[:8]} (accuracy: {accuracy}m)"
+        return status_msg, lat, lon, accuracy, timestamp
+        
+    except Exception as e:
+        status_msg = f"❌ **Error**: {str(e)}"
+        return status_msg, f"Error parsing GPS data: {str(e)}", "", "", ""
+
+def get_gps_js():
+    """JavaScript for GPS capture using hidden textbox approach"""
+    return """
+    () => {
+      // find the textarea element inside Gradio textbox by its elem_id
+      const textarea = document.querySelector('#hidden_gps_input textarea');
+      if (!textarea) {
+        console.log("Hidden GPS textbox not found");
+        return;
+      }
+      if (!navigator.geolocation) {
+        textarea.value = JSON.stringify({error: "Geolocation not supported"});
+        textarea.dispatchEvent(new Event('input', { bubbles: true }));
+        return;
+      }
+      navigator.geolocation.getCurrentPosition(
+        function(position) {
+          const data = {
+            latitude: position.coords.latitude,
+            longitude: position.coords.longitude,
+            accuracy: position.coords.accuracy,
+            timestamp: new Date().toISOString()
+          };
+          textarea.value = JSON.stringify(data);
+          // dispatch 'input' event so Gradio notices the change
+          textarea.dispatchEvent(new Event('input', { bubbles: true }));
+        },
+        function(err) {
+          textarea.value = JSON.stringify({ error: err.message });
+          textarea.dispatchEvent(new Event('input', { bubbles: true }));
+        },
+        { enableHighAccuracy: true, timeout: 10000 }
+      );
+    }
+    """
+
+def save_to_dataset(image, lat, lon, accuracy_m, device_ts):
+    """
+    Save image and metadata to Hugging Face dataset
+    
+    Args:
+        image: PIL Image object
+        lat: latitude as string
+        lon: longitude as string  
+        accuracy_m: accuracy in meters as string
+        device_ts: device timestamp as string
+        
+    Returns:
+        tuple: (status_markdown, preview_json)
+    """
+    try:
+        # Validate inputs
+        if image is None:
+            return "❌ **Error**: No image captured. Please take a photo first.", ""
+        
+        # 🔒 Ensure PIL.Image
+        try:
+            from PIL import Image as _PILImage
+            import numpy as _np
+            if isinstance(image, _np.ndarray):
+                image = _PILImage.fromarray(image)
+        except Exception:
+            pass
+        
+        if not lat or not lon:
+            return "❌ **Error**: GPS coordinates missing. Please click 'Get GPS' first.", ""
+        
+        # Check if running in test mode
+        if not api:
+            # Test mode - just preview the data without saving
+            server_ts = datetime.now().isoformat()
+            img_id = str(uuid.uuid4())
+            
+            timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S")
+            row = {
+                "id": img_id,
+                "image": f"lanternfly_{timestamp_str}_{img_id[:8]}.jpg",
+                "latitude": float(lat) if lat else None,
+                "longitude": float(lon) if lon else None,
+                "accuracy_m": float(accuracy_m) if accuracy_m else None,
+                "device_timestamp": device_ts if device_ts else None,
+                "server_timestamp_utc": server_ts,
+                "notes": ""
+            }
+            
+            status = f"🔍 **Test Mode**: Data validated successfully! Sample {img_id[:8]}"
+            preview = json.dumps(row, indent=2)
+            return status, preview
+        
+        # Build a unique ID and paths with timestamp for better visibility
+        sample_id = str(uuid.uuid4())
+        timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S")
+        image_rel_path = f"lanternfly_{timestamp_str}_{sample_id[:8]}.jpg"
+        
+        # Save image first
+        try:
+            _save_image_to_repo(image, image_rel_path)
+        except Exception as e:
+            return f"❌ **Error**: Failed to upload image: {e}", ""
+        
+        # Server UTC timestamp
+        server_ts_utc = datetime.now().isoformat() + "Z"
+        
+        # Construct metadata row
+        row = {
+            "id": sample_id,
+            "image": image_rel_path,  # This will be visible in dataset viewer
+            "latitude": float(lat) if lat else None,
+            "longitude": float(lon) if lon else None,
+            "accuracy_m": float(accuracy_m) if accuracy_m else None,
+            "device_timestamp": device_ts if device_ts else None,
+            "server_timestamp_utc": server_ts_utc,
+            "location": f"{lat}, {lon}" if lat and lon else None,  # Human-readable location
+            "notes": ""  # placeholder for future labels
+        }
+        
+        # Append metadata row
+        try:
+            _append_jsonl_in_repo(row)
+        except Exception as e:
+            return f"❌ **Error**: Image uploaded, but failed to append metadata: {e}", ""
+        
+        # Return success message and preview
+        status = (
+            "✅ **Success!** Saved to dataset!\n\n"
+            f"- Image: `{image_rel_path}`\n"
+            f"- Lat/Lon: {row['latitude']}, {row['longitude']} (±{row['accuracy_m']} m)\n"
+            f"- Server time (UTC): {server_ts_utc}"
+        )
+        preview = json.dumps(row, indent=2)
+        
+        return status, preview
+        
+    except Exception as e:
+        error_msg = f"❌ **Error**: {str(e)}"
+        return error_msg, ""
+
+# Create Gradio interface
+with gr.Blocks(title="Lanternfly Field Capture") as app:
+    gr.Markdown("# 🦋 Lanternfly Field Capture")
+    gr.Markdown("Capture photos with GPS coordinates for field research data collection.")
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            # Camera input
+            camera = gr.Image(
+                streaming=False,
+                height=380,
+                label="📷 Upload Photo (or use camera)",
+                type="pil",
+                sources=["webcam", "upload"]
+            )
+            
+            # GPS data capture
+            gr.Markdown("### 📍 GPS Coordinates")
+            gr.Markdown("Click the button below to automatically capture your location and timestamp.")
+            
+            # GPS capture button
+            gps_btn = gr.Button("📍 Get GPS", variant="primary")
+            
+            # Hidden input for GPS data
+            hidden_gps_input = gr.Textbox(visible=False, elem_id="hidden_gps_input")
+            
+            with gr.Row():
+                lat_box = gr.Textbox(label="Latitude", interactive=True, elem_id="lat")
+                lon_box = gr.Textbox(label="Longitude", interactive=True, elem_id="lon")
+            
+            with gr.Row():
+                accuracy_box = gr.Textbox(label="Accuracy (meters)", interactive=True, elem_id="accuracy")
+                device_ts_box = gr.Textbox(label="Device Timestamp", interactive=True, elem_id="device_ts")
+            
+            # Time capture button
+            time_btn = gr.Button("🕐 Get Current Time", variant="secondary")
+            
+            # Save button
+            save_btn = gr.Button("💾 Save to Dataset", variant="secondary")
+        
+        with gr.Column(scale=1):
+            # Status display
+            status = gr.Markdown("🔄 **Ready to capture data...** Click 'Get GPS' to start or upload a photo.")
+            
+            # Preview JSON
+            preview = gr.JSON(label="Preview JSON", visible=True)
+    
+    # Event handlers
+    gps_btn.click(
+        fn=None,
+        inputs=[],
+        outputs=[],
+        js=get_gps_js()
+    )
+    
+    # When the hidden GPS input changes, populate the visible fields
+    hidden_gps_input.change(
+        fn=handle_gps_location,
+        inputs=[hidden_gps_input],
+        outputs=[status, lat_box, lon_box, accuracy_box, device_ts_box]
+    )
+    
+    time_btn.click(
+        fn=handle_time_capture,
+        inputs=[],
+        outputs=[status, device_ts_box]
+    )
+    
+    save_btn.click(
+        fn=save_to_dataset,
+        inputs=[camera, lat_box, lon_box, accuracy_box, device_ts_box],
+        outputs=[status, preview]
+    )
+
+# Launch the app
+if __name__ == "__main__":
+    app.launch()