Upload 5 files

Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY main.py .
+COPY synthetic_ps_points.geojson .
+COPY synthetic_ps_points.csv .
+
+EXPOSE 7860
+
+CMD ["uvicorn", "main:main", "--host", "0.0.0.0", "--port", "7860"]

main.py

@@ -0,0 +1,174 @@
+from fastapi import FastAPI, HTTPException, Response
+import json
+import random
+import math
+import pandas as pd
+import folium
+from folium.plugins import HeatMap
+from typing import Dict, Any, List
+
+app = FastAPI(title="GeoJSON and Heatmap API", description="API for random coordinates, worker path simulation, and heatmap HTML from PS data")
+
+# Load GeoJSON data from file
+def load_geojson_data(file_path: str = "synthetic_ps_points.geojson") -> Dict[str, Any]:
+    try:
+        with open(file_path, 'r') as file:
+            return json.load(file)
+    except FileNotFoundError:
+        raise HTTPException(status_code=404, detail="GeoJSON file not found")
+    except json.JSONDecodeError:
+        raise HTTPException(status_code=400, detail="Invalid GeoJSON format")
+
+# Load CSV data for heatmap
+def load_csv_data(file_path: str = "synthetic_ps_points.csv") -> pd.DataFrame:
+    try:
+        return pd.read_csv(file_path)
+    except FileNotFoundError:
+        raise HTTPException(status_code=404, detail="CSV file not found")
+    except pd.errors.EmptyDataError:
+        raise HTTPException(status_code=400, detail="Invalid or empty CSV file")
+
+# Calculate Euclidean distance between two coordinates
+def calculate_distance(coord1: List[float], coord2: List[float]) -> float:
+    return math.sqrt((coord2[0] - coord1[0]) ** 2 + (coord2[1] - coord1[1]) ** 2)
+
+# Find the closest feature to a given coordinate
+def find_closest_feature(coord: List[float], features: List[Dict]) -> Dict:
+    min_distance = float('inf')
+    closest_feature = None
+    for feature in features:
+        feature_coord = feature["geometry"]["coordinates"]
+        distance = calculate_distance(coord, feature_coord)
+        if distance < min_distance:
+            min_distance = distance
+            closest_feature = feature
+    return closest_feature
+
+# Generate a linear path between two points
+def generate_path(start_coord: List[float], end_coord: List[float], num_steps: int = 11) -> List[Dict]:
+    path = []
+    for i in range(num_steps):
+        t = i / (num_steps - 1)  # Interpolation factor
+        lon = start_coord[0] + t * (end_coord[0] - start_coord[0])
+        lat = start_coord[1] + t * (end_coord[1] - start_coord[1])
+        path.append({"step": i, "coordinates": [lon, lat]})
+    return path
+
+# Endpoint to get random coordinates
+@app.get("/get-coordinates", response_model=dict)
+async def get_random_coordinates():
+    data = load_geojson_data()
+    features = data.get("features", [])
+    if not features:
+        raise HTTPException(status_code=400, detail="No features found in GeoJSON data")
+    
+    random_feature = random.choice(features)
+    coordinates = random_feature["geometry"]["coordinates"]
+    properties = random_feature["properties"]
+    
+    return {
+        "ps_id": properties["ps_id"],
+        "coordinates": {
+            "longitude": coordinates[0],
+            "latitude": coordinates[1]
+        },
+        "velocity_mm_yr": properties["velocity_mm_yr"],
+        "risk": properties["risk"]
+    }
+
+# Endpoint to simulate a worker's path from normal to high risk
+@app.get("/simulate-worker-path", response_model=dict)
+async def simulate_worker_path():
+    data = load_geojson_data()
+    features = data.get("features", [])
+    if not features:
+        raise HTTPException(status_code=400, detail="No features found in GeoJSON data")
+    
+    normal_risk_features = [f for f in features if f["properties"]["risk"] == "Normal"]
+    high_risk_features = [f for f in features if f["properties"]["risk"] == "High"]
+    
+    if not normal_risk_features or not high_risk_features:
+        raise HTTPException(status_code=400, detail="Insufficient normal or high risk features for path simulation")
+    
+    start_feature = random.choice(normal_risk_features)
+    end_feature = random.choice(high_risk_features)
+    
+    start_coord = start_feature["geometry"]["coordinates"]
+    end_coord = end_feature["geometry"]["coordinates"]
+    
+    path = generate_path(start_coord, end_coord, num_steps=11)
+    
+    path_with_risk = []
+    for point in path:
+        closest_feature = find_closest_feature(point["coordinates"], features)
+        path_with_risk.append({
+            "step": point["step"],
+            "coordinates": {
+                "longitude": point["coordinates"][0],
+                "latitude": point["coordinates"][1]
+            },
+            "risk": closest_feature["properties"]["risk"]
+        })
+    
+    return {
+        "start": {
+            "ps_id": start_feature["properties"]["ps_id"],
+            "coordinates": {"longitude": start_coord[0], "latitude": start_coord[1]},
+            "risk": start_feature["properties"]["risk"]
+        },
+        "end": {
+            "ps_id": end_feature["properties"]["ps_id"],
+            "coordinates": {"longitude": end_coord[0], "latitude": end_coord[1]},
+            "risk": end_feature["properties"]["risk"]
+        },
+        "path": path_with_risk
+    }
+
+# Endpoint to generate and return raw HTML heatmap
+@app.get("/heatmap", response_class=Response)
+async def get_heatmap():
+    # Load CSV data
+    ps_data = load_csv_data()
+    
+    # Polygon bounds for Singrauli
+    polygon_coords = [[(82.5065, 22.3105), (82.628, 22.3105), (82.628, 22.3421), (82.5065, 22.3421), (82.5065, 22.3105)]]
+    
+    # Center for map
+    center_lat = (22.3105 + 22.3421) / 2
+    center_lon = (82.5065 + 82.628) / 2
+    
+    # Create base map
+    m = folium.Map(location=[center_lat, center_lon], zoom_start=12, tiles="OpenStreetMap")
+    
+    # Heatmap using velocity
+    heat_data = [[row['lat'], row['lon'], abs(row['velocity_mm_yr'])] for _, row in ps_data.iterrows()]
+    HeatMap(heat_data, radius=15, gradient={0.2: 'blue', 0.4: 'green', 0.6: 'yellow', 1: 'red'}).add_to(m)
+    
+    # Add polygon boundary
+    folium.Polygon(
+        locations=[(lat, lon) for lon, lat in polygon_coords[0]],
+        color="white",
+        fill=False,
+        weight=2
+    ).add_to(m)
+    
+    # Get HTML content
+    html_content = m.get_root().render()
+    
+    return Response(content=html_content, media_type="text/html")
+
+# Root endpoint for API info
+@app.get("/")
+async def root():
+    return {
+        "message": "Welcome to the GeoJSON and Heatmap API",
+        "endpoints": {
+            "/get-coordinates": "Returns a random coordinate pair with associated properties",
+            "/simulate-worker-path": "Simulates a worker's path from a normal risk to a high risk zone",
+            "/heatmap": "Returns raw HTML for a Folium heatmap of PS data"
+        }
+    }
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)

requirements.txt

@@ -0,0 +1,4 @@
+fastapi==0.115.0
+uvicorn==0.30.6
+pandas==2.2.3
+folium==0.17.0