Spaces:
Sleeping
Sleeping
File size: 4,919 Bytes
dda93ab |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
from flask import session
from app.utils.helper import is_ajax, get_response, validate_latlon, coordinates_match
import pandas as pd
class CoordinateController:
MAX_LIMIT = 30
@staticmethod
def add_coordinate(lat, lon, coordinates, request):
"""Add a coordinate to the session."""
try:
lat, lon = validate_latlon(lat, lon)
except ValueError:
return get_response("Invalid coordinates. Please enter valid numbers.", "error", 400, is_ajax(request))
if len(coordinates) >= CoordinateController.MAX_LIMIT:
return get_response(f"Maximum {CoordinateController.MAX_LIMIT} coordinates allowed.", "error", 400, is_ajax(request))
if any(coordinates_match(c, lat, lon) for c in coordinates):
return get_response("Coordinate already exists.", "warning", 400, is_ajax(request))
coordinates.append({"lat": lat, "lon": lon})
session["coordinates"] = coordinates
session["map_center"] = {"lat": lat, "lon": lon}
session.modified = True
return get_response(
"Coordinate added successfully!",
"success",
200,
is_ajax(request),
{"lat": lat, "lon": lon}
)
@staticmethod
def delete_coordinate(lat, lon):
"""Delete coordinate by lat/lon"""
try:
lat = float(lat)
lon = float(lon)
except ValueError:
return get_response("Invalid coordinates.", "error", 400)
coords = session.get("coordinates", [])
new_list = [c for c in coords if not coordinates_match(c, lat, lon)]
session["coordinates"] = new_list
session.modified = True
return get_response("Coordinate deleted successfully!", "success", 200, extra={"coordinates": new_list})
@staticmethod
def clear_all(request):
"""Clear all coordinates"""
session["coordinates"] = []
session.modified = True
return get_response(
"All coordinates cleared!",
"success",
200,
is_ajax(request),
extra={"coordinates": []}
)
@staticmethod
def upload_coordinates(file_obj, request):
"""Upload coordinates from a CSV file, supporting flexible column naming."""
if not file_obj or file_obj.filename == "":
return get_response("No file selected.", "error", 400, is_ajax(request))
if not file_obj.filename.endswith(".csv"):
return get_response("Invalid file format. Please upload a CSV file.", "error", 400, is_ajax(request))
# 1. Read CSV
try:
df = pd.read_csv(file_obj)
except Exception as e:
return get_response(f"Error reading CSV: {str(e)}", "error", 500, is_ajax(request))
# 2. Normalize column names
df.columns = df.columns.str.strip().str.lower()
# 3. Possible matching names
lat_candidates = ["latitude", "lat", "x", "y_lat", "latitude (deg)"]
lon_candidates = ["longitude", "lon", "lng", "long", "y", "x_lon", "longitude (deg)"]
# 4. Detect actual columns
lat_col = next((c for c in df.columns if any(key in c for key in lat_candidates)), None)
lon_col = next((c for c in df.columns if any(key in c for key in lon_candidates)), None)
if not lat_col or not lon_col:
return get_response(
"CSV must contain recognizable latitude/longitude columns. "
"Examples: latitude, lat, Latitude, LAT, lon, longitude",
"error",
400,
is_ajax(request)
)
# 5. Process coordinates
coordinates = []
new_count = 0
invalid_count = 0
duplicates_count = 0
for _, row in df.iterrows():
try:
lat, lon = validate_latlon(row[lat_col], row[lon_col])
except ValueError:
invalid_count += 1
continue
if len(coordinates) >= CoordinateController.MAX_LIMIT:
break
if any(coordinates_match(c, lat, lon) for c in coordinates):
duplicates_count += 1
continue
coordinates.append({"lat": lat, "lon": lon})
new_count += 1
# 6. Save to session
session["coordinates"] = coordinates
session.modified = True
msg = (
f"Successfully added {new_count} new coordinates. "
f"{invalid_count} invalid coordinates ignored. "
f"{duplicates_count} duplicates ignored."
)
return get_response(msg, "success", 200, is_ajax(request), extra={"coordinates": coordinates})
|