from fastapi import FastAPI, HTTPException, Query, Request, Response from fastapi.responses import HTMLResponse, JSONResponse, FileResponse from fastapi.middleware.cors import CORSMiddleware import pandas as pd import numpy as np from geopy.distance import geodesic import folium from folium import plugins import osmnx as ox import networkx as nx from datetime import datetime import json import matplotlib.pyplot as plt import plotly.express as px import os import time from functools import lru_cache from rtree import index from pydantic import BaseModel, Field from typing import List, Dict, Any, Optional # Create app instance app = FastAPI(title="Store Locator API") # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Create temp directory for files os.makedirs('temp', exist_ok=True) # Load and prepare the store data stores_df = pd.read_csv('dataset of 50 stores.csv') # Custom JSON encoder for numpy types class NumpyJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, np.integer): return int(obj) elif isinstance(obj, np.floating): return float(obj) elif isinstance(obj, np.ndarray): return obj.tolist() return super().default(obj) # Pydantic models for response validation class StoreLocation(BaseModel): lat: float lon: float class Store(BaseModel): store_name: str address: str contact: str distance: float estimated_delivery_time: int product_categories: str location: StoreLocation class StoresResponse(BaseModel): status: str stores: List[Store] class ErrorResponse(BaseModel): status: str message: str class StoreLocator: def __init__(self, stores_dataframe): self.stores_df = stores_dataframe self.network_graph = None self.graph_cache = {} # Cache for network graphs self.spatial_index = self._build_spatial_index() @lru_cache(maxsize=50) def initialize_graph(self, center_point, dist=20000): """Initialize road network graph with caching""" cache_key = f"{center_point[0]}_{center_point[1]}" if cache_key in self.graph_cache: self.network_graph = self.graph_cache[cache_key] return True try: self.network_graph = ox.graph_from_point(center_point, dist=dist, network_type="drive") self.network_graph = ox.add_edge_speeds(self.network_graph) self.network_graph = ox.add_edge_travel_times(self.network_graph) return True except Exception as e: print(f"Error initializing graph: {str(e)}") return False def _build_spatial_index(self): idx = index.Index() for i, row in self.stores_df.iterrows(): idx.insert(i, (row['Latitude'], row['Longitude'], row['Latitude'], row['Longitude'])) return idx def calculate_distance(self, lat1, lon1, lat2, lon2): """Calculate direct distance between two points""" return geodesic((lat1, lon1), (lat2, lon2)).kilometers def estimate_delivery_time(self, distance, current_time=None): """Estimate delivery time based on distance and current time""" if current_time is None: current_time = datetime.now() # Base time: 5 mins base + 2 mins per km base_minutes = 5 + (distance * 2) # Apply traffic multiplier based on time of day hour = current_time.hour if hour in [8, 9, 10, 17, 18, 19]: # Peak hours multiplier = 1.5 elif hour in [23, 0, 1, 2, 3, 4]: # Off-peak hours multiplier = 0.8 else: # Normal hours multiplier = 1.0 return round(base_minutes * multiplier) def find_nearby_stores(self, lat, lon, radius=5): """Find stores within radius using spatial index""" nearby_stores = [] bbox = (lat - radius/111.0, lon - radius/111.0, lat + radius/111.0, lon + radius/111.0) for store_id in self.spatial_index.intersection(bbox): store = self.stores_df.iloc[store_id] distance = self.calculate_distance(lat, lon, store['Latitude'], store['Longitude']) if distance <= radius: delivery_time = self.estimate_delivery_time(distance) nearby_stores.append({ 'store_name': store['Store Name'], 'address': store['Address'], 'contact': str(store['Contact Number']), # Convert to string to avoid int64 issues 'distance': round(distance, 2), 'estimated_delivery_time': int(delivery_time), # Ensure integer type 'product_categories': store['Product Categories'], 'location': { 'lat': float(store['Latitude']), # Ensure float type 'lon': float(store['Longitude']) # Ensure float type } }) return sorted(nearby_stores, key=lambda x: x['distance']) def create_store_map(self, center_lat, center_lon, radius=5): """Create an interactive map with store locations""" # Create base map m = folium.Map(location=[center_lat, center_lon], zoom_start=13, tiles="cartodbpositron", prefer_canvas=True ) # Create marker cluster for better performance with many markers marker_cluster = plugins.MarkerCluster().add_to(m) # Add stores to map nearby_stores = self.find_nearby_stores(center_lat, center_lon, radius) for store in nearby_stores: # Prepare popup content popup_content = f"""
{store['store_name']}
Address: {store['address']}
Distance: {store['distance']} km
Est. Delivery: {store['estimated_delivery_time']} mins
Categories: {store['product_categories']}
""" # Add store marker folium.Marker( location=[store['location']['lat'], store['location']['lon']], popup=folium.Popup(popup_content, max_width=300), icon=folium.Icon(color='red', icon='info-sign') ).add_to(marker_cluster) # Add line to show distance from center folium.PolyLine( locations=[[center_lat, center_lon], [store['location']['lat'], store['location']['lon']]], weight=2, color='blue', opacity=0.3 ).add_to(m) # Add current location marker folium.Marker( location=[center_lat, center_lon], popup='Your Location', icon=folium.Icon(color='green', icon='home') ).add_to(m) # Add fullscreen option plugins.Fullscreen().add_to(m) # Add layer control folium.LayerControl().add_to(m) return m # Initialize store locator store_locator = StoreLocator(stores_df) def create_animated_route(G, path, color, weight=3): """Create an animated route visualization""" features = [] timestamps = [] # Convert path nodes to coordinates route_coords = [ (G.nodes[node]['y'], G.nodes[node]['x']) for node in path ] # Create features for each segment of the route for i in range(len(route_coords) - 1): segment = { 'type': 'Feature', 'geometry': { 'type': 'LineString', 'coordinates': [ [route_coords[i][1], route_coords[i][0]], [route_coords[i+1][1], route_coords[i+1][0]] ] }, 'properties': { 'times': [datetime.now().isoformat()], 'style': { 'color': color, 'weight': weight, 'opacity': 0.8 } } } features.append(segment) timestamps.append(datetime.now().isoformat()) return features def create_route_animation_data(G, path_time, path_length): """Create animation data for route visualization""" lst_start, lst_end = [], [] start_x, start_y = [], [] end_x, end_y = [], [] lst_length, lst_time = [], [] # Process time-based path for a, b in zip(path_time[:-1], path_time[1:]): lst_start.append(a) lst_end.append(b) lst_length.append(round(G.edges[(a,b,0)]['length'])) lst_time.append(round(G.edges[(a,b,0)]['travel_time'])) start_x.append(G.nodes[a]['x']) start_y.append(G.nodes[a]['y']) end_x.append(G.nodes[b]['x']) end_y.append(G.nodes[b]['y']) # Create DataFrame df = pd.DataFrame( list(zip(lst_start, lst_end, start_x, start_y, end_x, end_y, lst_length, lst_time)), columns=["start", "end", "start_x", "start_y", "end_x", "end_y", "length", "travel_time"] ).reset_index().rename(columns={"index": "id"}) return df # Function to clean up temporary files @app.middleware("http") async def cleanup_temp_files(request: Request, call_next): temp_dir = 'temp' if os.path.exists(temp_dir): for file in os.listdir(temp_dir): file_path = os.path.join(temp_dir, file) try: if os.path.isfile(file_path) and file.endswith('.html'): # Delete files older than 1 hour if os.path.getmtime(file_path) < time.time() - 3600: os.remove(file_path) except Exception as e: print(f"Error cleaning up temp files: {e}") response = await call_next(request) return response @app.get("/", response_class=HTMLResponse) async def home(): """API documentation homepage""" html_content = f""" falcao-maps API Documentation

Welcome to falcao-maps

Based on your uploaded dataset and deployed API, here are example API calls for your client:

1. Find Nearby Stores (JSON Response)

            api/stores/nearby?lat=18.9695&lon=72.8320&radius=0.5
        

Use this to get store details near Market Road area within 1km

2. View Basic Store Map

        
https://lucifer7210-maps.hf.space/api/stores/map?lat=18.9701&lon=72.8330&radius=0.5
        
        

Shows map centered at Main Street with 500m radius

3. View All Store Locations with Color Coding

        
            https://lucifer7210-maps.hf.space/api/stores/locations?lat=18.9685&lon=72.8325&radius=2
        
        

Shows detailed map with color-coded stores within 2km

4. Get Route Between Points

Example routes:

# Route from Park Avenue to Hill Road stores
https://lucifer7210-maps.hf.space/api/stores/route?user_lat=18.9710&user_lon=72.8335&store_lat=18.9705&store_lon=72.8345&viz_type=simple

# Route from Main Street to Market Road stores
https://lucifer7210-maps.hf.space/api/stores/route?user_lat=18.9701&user_lon=72.8330&store_lat=18.9695&store_lon=72.8320&viz_type=simple
        

Key Location Points in Dataset:

""" return HTMLResponse(content=html_content) @app.get("/api/stores/nearby", response_model=StoresResponse, responses={400: {"model": ErrorResponse}}) async def get_nearby_stores( lat: float = Query(..., description="Latitude of user location"), lon: float = Query(..., description="Longitude of user location"), radius: float = Query(5, description="Search radius in kilometers") ): """ Get nearby stores based on user location """ try: nearby_stores = store_locator.find_nearby_stores(lat, lon, radius) return {"status": "success", "stores": nearby_stores} except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.get("/api/stores/map", response_class=HTMLResponse, responses={400: {"model": ErrorResponse}}) async def get_stores_map( lat: float = Query(..., description="Latitude of center point"), lon: float = Query(..., description="Longitude of center point"), radius: float = Query(5, description="Search radius in kilometers") ): """ Get HTML map with store locations """ try: store_map = store_locator.create_store_map(lat, lon, radius) # Create complete HTML content html_content = f""" Stores Map {store_map.get_root().render()} """ # Save the HTML to a file file_path = 'temp/stores_map.html' with open(file_path, 'w', encoding='utf-8') as f: f.write(html_content) # Return the file return FileResponse(file_path, media_type='text/html') except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.get("/api/stores/route", response_class=HTMLResponse, responses={400: {"model": ErrorResponse}, 404: {"model": ErrorResponse}}) async def get_store_route( user_lat: float = Query(..., description="Latitude of user location"), user_lon: float = Query(..., description="Longitude of user location"), store_lat: float = Query(..., description="Latitude of store location"), store_lon: float = Query(..., description="Longitude of store location") ): """ Get route between user location and store """ try: # Initialize graph if not already initialized if store_locator.network_graph is None: store_locator.initialize_graph((user_lat, user_lon)) # Get nearest nodes start_node = ox.distance.nearest_nodes( store_locator.network_graph, user_lon, user_lat) end_node = ox.distance.nearest_nodes( store_locator.network_graph, store_lon, store_lat) try: # Calculate paths path_time = nx.shortest_path( store_locator.network_graph, start_node, end_node, weight='travel_time' ) # Create animation data lst_start, lst_end = [], [] start_x, start_y = [], [] end_x, end_y = [], [] lst_length, lst_time = [], [] for a, b in zip(path_time[:-1], path_time[1:]): lst_start.append(a) lst_end.append(b) lst_length.append(round(store_locator.network_graph.edges[(a,b,0)]['length'])) lst_time.append(round(store_locator.network_graph.edges[(a,b,0)]['travel_time'])) start_x.append(store_locator.network_graph.nodes[a]['x']) start_y.append(store_locator.network_graph.nodes[a]['y']) end_x.append(store_locator.network_graph.nodes[b]['x']) end_y.append(store_locator.network_graph.nodes[b]['y']) df = pd.DataFrame( list(zip(lst_start, lst_end, start_x, start_y, end_x, end_y, lst_length, lst_time)), columns=["start", "end", "start_x", "start_y", "end_x", "end_y", "length", "travel_time"] ).reset_index().rename(columns={"index": "id"}) # Create animation using plotly df_start = df[df["start"] == start_node] df_end = df[df["end"] == end_node] fig = px.scatter_mapbox( data_frame=df, lon="start_x", lat="start_y", zoom=15, width=1000, height=800, animation_frame="id", mapbox_style="carto-positron" ) # Add driver marker fig.data[0].marker = {"size": 12} # Add start point fig.add_trace( px.scatter_mapbox( data_frame=df_start, lon="start_x", lat="start_y" ).data[0] ) fig.data[1].marker = {"size": 15, "color": "red"} # Add end point fig.add_trace( px.scatter_mapbox( data_frame=df_end, lon="start_x", lat="start_y" ).data[0] ) fig.data[2].marker = {"size": 15, "color": "green"} # Add route fig.add_trace( px.line_mapbox( data_frame=df, lon="start_x", lat="start_y" ).data[0] ) # Update layout with slower animation settings fig.update_layout( showlegend=False, margin={"r":0,"t":0,"l":0,"b":0}, autosize=True, height=None, updatemenus=[{ "type": "buttons", "showactive": False, "y": 0, "x": 0, "xanchor": "left", "yanchor": "bottom", "buttons": [ { "label": "Play", "method": "animate", "args": [ None, { "frame": {"duration": 1000, "redraw": True}, "fromcurrent": True, "transition": {"duration": 800} } ] }, { "label": "Pause", "method": "animate", "args": [ [None], { "frame": {"duration": 0, "redraw": False}, "mode": "immediate", "transition": {"duration": 0} } ] } ] }], sliders=[{ "currentvalue": {"prefix": "Step: "}, "pad": {"t": 20}, "len": 0.9, "x": 0.1, "xanchor": "left", "y": 0.02, "yanchor": "bottom", "steps": [ { "args": [ [k], { "frame": {"duration": 1000, "redraw": True}, "transition": {"duration": 500}, "mode": "immediate" } ], "label": str(k), "method": "animate" } for k in range(len(df)) ] }] ) # Create complete HTML content html_content = f""" Route Map
{fig.to_html(include_plotlyjs=True, full_html=False)}
""" # Save the HTML to a file file_path = 'temp/route_map.html' with open(file_path, 'w', encoding='utf-8') as f: f.write(html_content) # Return the file return FileResponse(file_path, media_type='text/html') except nx.NetworkXNoPath: raise HTTPException(status_code=404, detail="No route found") except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.get("/api/stores/locations", response_class=HTMLResponse, responses={400: {"model": ErrorResponse}}) async def get_all_store_locations( lat: float = Query(..., description="Latitude of center point"), lon: float = Query(..., description="Longitude of center point"), radius: float = Query(10, description="Search radius in kilometers") ): """ Get a map showing all stores in the given radius with colors based on distance """ try: # Get nearby stores nearby_stores = store_locator.find_nearby_stores(lat, lon, radius) # Create base map centered on user location m = folium.Map( location=[lat, lon], zoom_start=12, tiles="cartodbpositron" ) # Add user location marker folium.Marker( [lat, lon], popup='Your Location', icon=folium.Icon(color='green', icon='home') ).add_to(m) # Add markers for each store with color coding based on distance for store in nearby_stores: # Color code based on distance if store['distance'] <= 2: color = 'red' # Very close elif store['distance'] <= 5: color = 'orange' # Moderate distance else: color = 'blue' # Further away # Create detailed popup content with mobile-friendly styling popup_content = f"""

{store['store_name']}

Address: {store['address']}
Distance: {store['distance']} km
Est. Delivery: {store['estimated_delivery_time']} mins
Contact: {store['contact']}
Categories: {store['product_categories']}
""" # Add store marker folium.Marker( location=[store['location']['lat'], store['location']['lon']], popup=folium.Popup(popup_content, max_width=300), icon=folium.Icon(color=color, icon='info-sign'), tooltip=f"{store['store_name']} ({store['distance']} km)" ).add_to(m) # Add circle to show distance folium.Circle( location=[store['location']['lat'], store['location']['lon']], radius=store['distance'] * 100, color=color, fill=True, opacity=0.1 ).add_to(m) # Add distance circles from user location for radius_circle, color in [(2000, 'red'), (5000, 'orange'), (radius * 1000, 'blue')]: folium.Circle( location=[lat, lon], radius=radius_circle, color=color, fill=False, weight=1, dash_array='5, 5' ).add_to(m) # Create mobile-friendly HTML content html_content = f""" Nearby Stores {m.get_root().render()}
Distance Zones
< 2 km
2-5 km
> 5 km
Search Radius: {radius} km
Stores Found: {len(nearby_stores)}
""" # Save and return the file file_path = 'temp/locations_map.html' with open(file_path, 'w', encoding='utf-8') as f: f.write(html_content) return FileResponse(file_path, media_type='text/html') except Exception as e: raise HTTPException(status_code=400, detail=str(e)) # Add swagger UI customization @app.on_event("startup") async def startup_event(): app.title = "Store Locator API" app.description = "API for locating nearby stores and generating routes" app.version = "1.0.0" # Entry point for running the application if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)