Spaces:
Sleeping
Sleeping
File size: 3,947 Bytes
aa525fd b4fb74f aa525fd b4fb74f aa525fd 3a3ddca aa525fd 3a3ddca aa525fd | 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 | import csv
import json
from shapely.geometry import shape, Point
WARDS_GEOJSON = "wards.geojson"
WARD_CONTACTS = "ward_contacts.csv"
CATEGORY_ROUTING = {
"pothole": ["AE", "AEE", "EE"],
"road_damage": ["AE", "AEE", "EE"],
"road_blocking": ["AE", "AEE", "EE"],
"footpath": ["AE", "AEE", "EE"],
"streetlight": ["AE", "AEE", "EE"],
"garbage": ["Jr. Health Insp", "Sr. Health Insp"],
"drain": ["Jr. Health Insp", "Sr. Health Insp", "AE"],
"waterlogging": ["AE", "AEE", "EE"],
"stray_animal": ["Animal Husbandry"],
"dead_animal": ["Animal Husbandry", "Jr. Health Insp"],
"other": ["AE", "Jr. Health Insp", "JC"],
}
CATEGORY_LABELS = {
"pothole": "Pothole",
"road_damage": "Road damage",
"road_blocking": "Road blocked by debris",
"footpath": "Broken footpath",
"streetlight": "Streetlight not working",
"garbage": "Garbage / illegal dumping",
"drain": "Blocked or open drain",
"waterlogging": "Waterlogging",
"stray_animal": "Stray animal",
"dead_animal": "Dead animal",
"other": "Other civic issue",
}
class WardIndex:
def __init__(self):
self.polygons = []
self.contacts = {}
self._load_geojson()
self._load_contacts()
def _load_geojson(self):
with open(WARDS_GEOJSON, encoding="utf-8") as f:
data = json.load(f)
for feat in data["features"]:
ward_no = feat["properties"]["ward_no"]
geom = shape(feat["geometry"])
self.polygons.append((ward_no, geom))
def _load_contacts(self):
with open(WARD_CONTACTS, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
ward_no = int(row["ward_no"])
self.contacts[ward_no] = {
"ward_name": row["ward_name"],
"zone": row["zone"],
"constituency": row["constituency"],
"officers": json.loads(row["officers_json"]),
}
def find_ward(self, lat, lng):
point = Point(lng, lat)
for ward_no, geom in self.polygons:
if geom.contains(point):
return ward_no
return None
def is_in_bengaluru(self, lat, lng):
point = Point(lng, lat)
for _, geom in self.polygons:
if geom.contains(point):
return True
return False
def get_ward_info(self, ward_no):
return self.contacts.get(ward_no)
def route_officer(self, ward_no, category):
info = self.contacts.get(ward_no)
if info is None:
return None
officers = info["officers"]
chain = CATEGORY_ROUTING.get(category, CATEGORY_ROUTING["other"])
for role in chain:
officer = officers.get(role)
if officer and officer.get("name") and officer.get("phone"):
return {
"role": role,
"name": officer["name"],
"phone": officer["phone"],
}
for role, officer in officers.items():
if officer and officer.get("name") and officer.get("phone"):
return {
"role": role,
"name": officer["name"],
"phone": officer["phone"],
}
return None
def resolve(ward_index, lat, lng, category):
ward_no = ward_index.find_ward(lat, lng)
if ward_no is None:
return None
info = ward_index.get_ward_info(ward_no)
officer = ward_index.route_officer(ward_no, category)
return {
"ward_no": ward_no,
"ward_name": info["ward_name"] if info else "",
"zone": info["zone"] if info else "",
"constituency": info["constituency"] if info else "",
"officer": officer,
"category_label": CATEGORY_LABELS.get(category, CATEGORY_LABELS["other"]),
} |