import os
import gradio as gr
import geopandas as gpd
import folium
import requests
# osmnx opsiyonel (fallback için), yoksa sorun değil
try:
import osmnx as ox
except ImportError:
ox = None
from huggingface_hub import InferenceClient, login
# ==========================================
# HF TOKEN & MODEL
# ==========================================
HF_TOKEN = (os.environ.get("HUGGINGFACE_HUB_TOKEN", "") or "").strip()
if HF_TOKEN:
login(token=HF_TOKEN)
else:
print("UYARI: HF_TOKEN / HUGGINGFACE_HUB_TOKEN bulunamadı, gated modellere erişilemeyebilir.")
#client = InferenceClient(
# model="abacusai/Dracarys-72B-Instruct",
# token=HF_TOKEN if HF_TOKEN else None,
#)
# ==========================================
# ÖNCEDEN HAZIRLANMIŞ OSM VERİSİ
# ==========================================
NEIGH_PATH = "data/neighborhoods.geojson"
POIS_PATH = "data/pois.geojson"
if not (os.path.exists(NEIGH_PATH) and os.path.exists(POIS_PATH)):
print("UYARI: data/ klasöründe neighborhoods.geojson / pois.geojson bulunamadı!")
neighborhoods_gdf = None
pois_gdf = None
else:
neighborhoods_gdf = gpd.read_file(NEIGH_PATH)
pois_gdf = gpd.read_file(POIS_PATH)
print(f"{len(neighborhoods_gdf)} mahalle/semt, {len(pois_gdf)} POI yüklendi.")
DEFAULT_TAGS = {
"amenity": ["school", "pharmacy", "hospital", "restaurant", "cafe", "bank"],
"leisure": ["park", "playground"],
"shop": True,
# >>> ULAŞIM ETİKETLERİ
"highway": ["bus_stop"], # otobüs durakları
"railway": ["station", "halt", "tram_stop"], # tren / metro / tramvay istasyonları
"public_transport": ["stop_position", "platform"], # toplu taşıma durak/istasyonları
}
# ==========================================
# OSM / CBS FONKSİYONLARI
# ==========================================
from shapely.geometry import Point
import math
def get_neighborhood_center(city, district, neighborhood):
"""
Verilen şehir/ilçe/mahalle için poligonu alır ve merkez (centroid) koordinatını döndürür.
(lat, lon) şeklinde WGS84 (EPSG:4326) döner.
"""
gdf = get_neighborhood_gdf(city, district, neighborhood)
if gdf is None or len(gdf) == 0:
return None
# Emin olmak için WGS84'e çevir
try:
gdf_ll = gdf.to_crs(epsg=4326)
except Exception:
gdf_ll = gdf # zaten 4326 ise
poly = gdf_ll.geometry.iloc[0]
c = poly.centroid
return c.y, c.x # (lat, lon)
def _build_nearest_poi_query(lat, lon, key, values, radius):
"""
Overpass için: verilen key + values (örn: amenity=university, shop=supermarket)
etrafında en yakın POI'yi arayan sorguyu üretir.
"""
if isinstance(values, str):
values = [values]
filters = []
for v in values:
filters.append(f' node["{key}"="{v}"](around:{radius},{lat},{lon});')
filters.append(f' way["{key}"="{v}"](around:{radius},{lat},{lon});')
filters.append(f' rel["{key}"="{v}"](around:{radius},{lat},{lon});')
filters_str = "\n".join(filters)
query = f"""
[out:json][timeout:25];
(
{filters_str}
);
out center 1;
"""
return query
def _haversine_m(lat1, lon1, lat2, lon2):
"""
İki nokta arasındaki yaklaşık mesafeyi (metre) hesaplar.
"""
R = 6371000 # Dünya yarıçapı (m)
phi1 = math.radians(lat1)
phi2 = math.radians(lat2)
dphi = math.radians(lat2 - lat1)
dlambda = math.radians(lon2 - lon1)
a = math.sin(dphi/2)**2 + math.cos(phi1)*math.cos(phi2)*math.sin(dlambda/2)**2
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
return R * c
def find_nearest_poi_overpass(city, district, neighborhood, key, values, radius=5000):
"""
Mahalle merkezinden belirli bir yarıçap içinde (m cinsinden)
verilen OSM tag'ine göre (örn: amenity, shop) en yakın POI'yi arar.
Örnek:
key="amenity", values=["university", "college"]
key="shop", values=["supermarket", "convenience"]
Dönüş:
None -> bulunamadı
dict -> {"name", "lat", "lon", "tag_key", "tag_value", "distance_m", "distance_km"}
"""
center = get_neighborhood_center(city, district, neighborhood)
if center is None:
return None
lat_c, lon_c = center
query = _build_nearest_poi_query(lat_c, lon_c, key, values, radius)
url = "https://overpass-api.de/api/interpreter"
try:
resp = requests.post(url, data={"data": query}, timeout=30)
resp.raise_for_status()
data = resp.json()
except Exception as e:
print("Overpass nearest POI isteği hatası:", e)
return None
elements = data.get("elements", [])
if not elements:
return None
# İlk element en yakın kabul ediliyor
e = elements[0]
tags = e.get("tags", {}) or {}
name = tags.get("name", "(isimsiz POI)")
# node ise lat/lon direk var; way/rel ise center altında
lat_p = e.get("lat") or (e.get("center") or {}).get("lat")
lon_p = e.get("lon") or (e.get("center") or {}).get("lon")
tag_value = None
if isinstance(values, list):
for v in values:
if tags.get(key) == v:
tag_value = v
break
if tag_value is None:
tag_value = tags.get(key)
else:
tag_value = tags.get(key)
dist_m = None
dist_km = None
if lat_p is not None and lon_p is not None:
dist_m = _haversine_m(lat_c, lon_c, lat_p, lon_p)
dist_km = dist_m / 1000.0
return {
"name": name,
"lat": lat_p,
"lon": lon_p,
"tag_key": key,
"tag_value": tag_value,
"distance_m": dist_m,
"distance_km": dist_km,
}
def answer_nearest_poi(city, district, neighborhood, key, values, human_label=None, radius=5000):
"""
Kullanıcıya gösterilecek generic cevap metni.
human_label: 'üniversite', 'market', 'hastane' gibi daha okunaklı isim.
"""
info = find_nearest_poi_overpass(city, district, neighborhood, key, values, radius=radius)
if info is None:
label = human_label or f"{key}={values}"
return f"{city} / {district} / {neighborhood} çevresinde belirli bir yarıçapta {label} bulunamadı."
name = info["name"]
lat = info["lat"]
lon = info["lon"]
dist_km = info["distance_km"]
label = human_label or info.get("tag_value") or f"{key}={values}"
if lat is not None and lon is not None and dist_km is not None:
return (
f"{city} / {district} / {neighborhood} mahallesi merkezine en yakın {label}: "
f"{name} (yaklaşık {dist_km:.2f} km, konum: {lat:.5f}, {lon:.5f})."
)
else:
return (
f"{city} / {district} / {neighborhood} mahallesi merkezine en yakın {label}: {name}."
)
def nearest_poi_wrapper(city, district, neighborhood, poi_type):
"""
Gradio dropdown'dan seçilen poi_type'ı gerçek tag'lere çeviren yardımcı.
"""
if poi_type == "Üniversite":
key = "amenity"
values = ["university", "college"]
label = "üniversite"
elif poi_type == "Market":
key = "shop"
values = ["supermarket", "convenience", "mall", "department_store"]
label = "market"
elif poi_type == "Hastane":
key = "amenity"
values = "hospital"
label = "hastane"
elif poi_type == "Eczane":
key = "amenity"
values = "pharmacy"
label = "eczane"
elif poi_type == "Park":
key = "leisure"
values = "park"
label = "park"
else:
# varsayılan: üniversite
key = "amenity"
values = ["university", "college"]
label = "üniversite"
return answer_nearest_poi(city, district, neighborhood, key, values, human_label=label)
def get_neighborhood_gdf(city: str, district: str, neighborhood: str):
"""
1) Önce önceden kaydedilmiş GeoJSON'dan arar (city + district + neighborhood).
2) Eğer orada yoksa ve osmnx mevcutsa, OSM'den canlı çeker (fallback).
"""
city = (city or "").strip()
district = (district or "").strip()
neighborhood = (neighborhood or "").strip()
# 1) Precomputed veri
if neighborhoods_gdf is not None:
mask = (
(neighborhoods_gdf["city"] == city)
& (neighborhoods_gdf["district"] == district)
& (neighborhoods_gdf["neighborhood"] == neighborhood)
)
gdf_local = neighborhoods_gdf[mask]
if gdf_local is not None and len(gdf_local) > 0:
return gdf_local
# 2) Fallback: canlı OSM çağrısı
if ox is None:
print("OSM fallback kullanılamıyor: osmnx yüklü değil.")
return None
# İlçe bilgisini de geocode sorgusuna ekliyoruz
query = f"{neighborhood}, {district}, {city}, Türkiye"
print(f"OSM fallback: {query}")
try:
gdf_osm = ox.geocode_to_gdf(query)
except Exception as e:
print("Mahalle geocode hatası (OSM fallback):", e)
return None
if gdf_osm is None or len(gdf_osm) == 0:
return None
gdf_osm = gdf_osm.copy()
gdf_osm["city"] = city
gdf_osm["district"] = district
gdf_osm["neighborhood"] = neighborhood
return gdf_osm
def get_pois_within(gdf, tags=None):
"""
1) Eğer GeoJSON'da bu mahalle için POI varsa, oradan döner.
2) Yoksa ve osmnx mevcutsa, poligon üzerinden OSM'den canlı çeker (fallback).
"""
if gdf is None:
return None
if tags is None:
tags = DEFAULT_TAGS
# 1) Precomputed POI verisi
if (
pois_gdf is not None
and all(col in gdf.columns for col in ["city", "district", "neighborhood"])
):
row = gdf.iloc[0]
city = row["city"]
district = row["district"]
neighborhood = row["neighborhood"]
mask = (
(pois_gdf["city"] == city)
& (pois_gdf["district"] == district)
& (pois_gdf["neighborhood"] == neighborhood)
)
pois_local = pois_gdf[mask]
if pois_local is not None and len(pois_local) > 0:
return pois_local
# 2) Fallback: OSM'den canlı POI çek
if ox is None:
print("OSM POI fallback kullanılamıyor: osmnx yüklü değil.")
return None
try:
polygon = gdf.geometry.iloc[0]
pois_osm = ox.features_from_polygon(polygon, tags)
return pois_osm
except Exception as e:
print("POI hatası (OSM fallback):", e)
return None
def summarize_pois(gdf, pois):
summary = {}
try:
area_m2 = gdf.to_crs(epsg=32636).geometry.iloc[0].area
summary["alan_m2"] = float(area_m2)
summary["alan_km2"] = float(area_m2 / 1_000_000)
except Exception as e:
print("Alan hesaplama hatası:", e)
summary["alan_m2"] = None
summary["alan_km2"] = None
if pois is None or len(pois) == 0:
summary["toplam_poi"] = 0
return summary
summary["toplam_poi"] = int(len(pois))
if "amenity" in pois.columns:
amenity_counts = pois["amenity"].value_counts().to_dict()
for k, v in amenity_counts.items():
summary[f"amenity_{k}"] = int(v)
if "leisure" in pois.columns:
leisure_counts = pois["leisure"].value_counts().to_dict()
for k, v in leisure_counts.items():
summary[f"leisure_{k}"] = int(v)
if "shop" in pois.columns:
shop_counts = pois["shop"].value_counts().to_dict()
for k, v in shop_counts.items():
summary[f"shop_{k}"] = int(v)
# >>> ULAŞIM: highway / railway / public_transport
if "highway" in pois.columns:
hw_counts = pois["highway"].value_counts().to_dict()
for k, v in hw_counts.items():
summary[f"highway_{k}"] = int(v)
if "railway" in pois.columns:
rw_counts = pois["railway"].value_counts().to_dict()
for k, v in rw_counts.items():
summary[f"railway_{k}"] = int(v)
if "public_transport" in pois.columns:
pt_counts = pois["public_transport"].value_counts().to_dict()
for k, v in pt_counts.items():
summary[f"public_transport_{k}"] = int(v)
return summary
def build_poi_names_text(pois, max_per_category=15) -> str:
"""
POI GeoDataFrame'inden okul, eczane, kafe, restoran, market vb.
için isim listeleri çıkarır. LLM bağlamında kullanılacak metni döndürür.
"""
if pois is None or len(pois) == 0:
return "Bu mahalle için isim verisi olan POI bulunamadı.\n"
if "name" not in pois.columns:
return "Bu mahallede POI'ler için 'name' alanı bulunamadı.\n"
lines = []
def add_category(title, mask):
sub = pois[mask]
if sub is None or len(sub) == 0:
return
names = (
sub["name"]
.dropna()
.astype(str)
.str.strip()
)
names = [n for n in names if n]
if not names:
return
unique_names = sorted(set(names))[:max_per_category]
lines.append(f"{title}:")
for n in unique_names:
lines.append(f" - {n}")
lines.append("") # kategori arası boş satır
# Kategoriler
if "amenity" in pois.columns:
amenity = pois["amenity"]
add_category("Okullar", amenity == "school")
add_category("Eczaneler", amenity == "pharmacy")
add_category("Kafeler", amenity == "cafe")
add_category("Restoranlar", amenity == "restaurant")
if "shop" in pois.columns:
shop = pois["shop"]
# İstersen buraya başka shop türleri de ekleyebilirsin
add_category("Marketler", shop.isin(["supermarket", "convenience", "mall", "department_store"]))
if not lines:
return "Bu mahallede adı bilinen POI listesi çıkarılamadı.\n"
return "\n".join(lines)
def build_stats_text(summary: dict) -> str:
if not summary:
return "Veri bulunamadı."
alan = summary.get("alan_km2", 0) or 0.0
toplam_poi = summary.get("toplam_poi", 0)
okul = summary.get("amenity_school", 0)
park = summary.get("leisure_park", 0)
eczane = summary.get("amenity_pharmacy", 0)
cafe = summary.get("amenity_cafe", 0)
restoran = summary.get("amenity_restaurant", 0)
# >>> ULAŞIM SAYILARI
otobus_duragi = summary.get("highway_bus_stop", 0)
tren_istasyonu = summary.get("railway_station", 0) + summary.get("railway_halt", 0)
tramvay_duragi = summary.get("railway_tram_stop", 0)
pt_platform = summary.get("public_transport_platform", 0)
lines = [
f"- Tahmini alan: {alan:.2f} km²",
f"- Toplam POI (ilgi noktası): {toplam_poi}",
f"- Okul sayısı: {okul}",
f"- Park sayısı: {park}",
f"- Eczane sayısı: {eczane}",
f"- Kafe sayısı: {cafe}",
f"- Restoran sayısı: {restoran}",
# >>> ULAŞIM SATIRLARI
f"- Otobüs durağı sayısı: {otobus_duragi}",
f"- Tren/metro istasyonu sayısı: {tren_istasyonu}",
f"- Tramvay durağı sayısı: {tramvay_duragi}",
f"- Toplu taşıma platform/istasyon öğesi: {pt_platform}",
]
return "\n".join(lines)
import requests
import json
import geopandas as gpd
from shapely.geometry import shape
from pathlib import Path
def is_osm_query(message: str) -> bool:
return isinstance(message, str) and message.strip().lower().startswith("/osm")
OSM_PROMPT_PATH = Path("osm_query_system.txt") # veya Path("data/osm_query_system.txt")
try:
with OSM_PROMPT_PATH.open("r", encoding="utf-8") as f:
OSM_QUERY_SYSTEM = f.read()
print(f"OSM system prompt '{OSM_PROMPT_PATH}' dosyasından yüklendi.")
except FileNotFoundError:
print(f"UYARI: {OSM_PROMPT_PATH} bulunamadı, kısa yedek prompt kullanılacak.")
OSM_QUERY_SYSTEM = """
You are an expert in OpenStreetMap and Overpass API.
Produce only valid Overpass QL.
No explanations. No markdown.
Use [out:json][timeout:25]; at top.
End with: out center;
"""
def generate_overpass_query_from_llm(prompt: str, model_name: str) -> str:
"""
Doğal dilde verilen prompt'u kullanarak Overpass QL sorgusu üretir.
Sadece geçerli Overpass QL döndürmeye çalışır, markdown vs. temizler.
"""
prompt = (prompt or "").strip()
if not prompt:
prompt = "Generate an Overpass QL query for my request."
client = InferenceClient(model=model_name, token=HF_TOKEN if HF_TOKEN else None)
messages = [
{"role": "system", "content": OSM_QUERY_SYSTEM},
{"role": "user", "content": prompt},
]
# Streaming'e gerek yok, tek seferde alalım
result = client.chat_completion(
messages=messages,
max_tokens=400,
temperature=0.2,
top_p=0.9,
stream=False,
)
# HF InferenceClient sonucu
query_text = result.choices[0].message.content
# Olası ```ql ``` bloklarını temizle
query_text = query_text.replace("```ql", "").replace("```QL", "").replace("```", "")
return query_text.strip()
def normalize_overpass_query(raw_query: str) -> str:
if not raw_query:
return ""
q = raw_query.strip()
if "{{geocodeArea:" in q:
# Kullanıcıyı uyarmak için
raise ValueError("Overpass API, {{geocodeArea:...}} makrosunu desteklemez. "
"Lütfen bunun yerine area[...] filtresi veya numeric area id kullanın.")
# (senin mevcut temizleme kodların)
for token in ("```ql", "```QL", "```", "", "", "", "
"):
q = q.replace(token, "")
return q.strip()
# ==========================================
# MAHALLE KARŞILAŞTIRMA BAĞLAMI
# ==========================================
def prepare_comparison(city1, district1, neigh1,city2, district2, neigh2):
"""
Butona basıldığında:
- Her iki mahalle için OSM özetini hazırlar
- İki metin döndürür
- LLM için karşılaştırma bağlamını üretir
"""
city1 = (city1 or "").strip()
district1 = (district1 or "").strip()
neigh1 = (neigh1 or "").strip()
city2 = (city2 or "").strip()
district2 = (district2 or "").strip()
neigh2 = (neigh2 or "").strip()
if not city1 or not district1 or not city1 or not neigh1 or not district2 or not neigh2:
msg = "Şehir, ilçe ve iki mahalle de girilmelidir."
return (msg, msg, "", "Harita için yeterli veri yok.")
# Varsayılan olarak boş metinler
labels1 = "Bu mahalle için isim verisi çıkarılamadı."
labels2 = "Bu mahalle için isim verisi çıkarılamadı."
# Mahalle 1
gdf1 = get_neighborhood_gdf(city1, district1, neigh1)
if gdf1 is None or len(gdf1) == 0:
stats1 = f"{city1} / {district1} / {neigh1} için veri bulunamadı."
summary1 = None
else:
pois1 = get_pois_within(gdf1)
summary1 = summarize_pois(gdf1, pois1)
stats1 = build_stats_text(summary1)
# >>> POI isim metni
labels1 = build_poi_names_text(pois1)
# Mahalle 2
gdf2 = get_neighborhood_gdf(city2, district2, neigh2)
if gdf2 is None or len(gdf2) == 0:
stats2 = f"{city2} / {district2} / {neigh2} için veri bulunamadı."
summary2 = None
else:
pois2 = get_pois_within(gdf2)
summary2 = summarize_pois(gdf2, pois2)
stats2 = build_stats_text(summary2)
# >>> POI isim metni
labels2 = build_poi_names_text(pois2)
# LLM bağlamı
compare_context_parts = [
f"Şehir: {city1}",
"",
f"1. Mahalle: {neigh1} (İlçe: {district1})",
stats1,
"",
"1. mahalledeki önemli POI isimleri (okullar, eczaneler, marketler, kafeler vb.):",
labels1,
"",
f"Şehir: {city2}",
"",
f"2. Mahalle: {neigh2} (İlçe: {district2})",
stats2,
"",
"2. mahalledeki önemli POI isimleri (okullar, eczaneler, marketler, kafeler vb.):",
labels2,
"",
"Bu iki mahalleyi alan, toplam POI sayısı, park, okul, kafe, restoran, eczane sayıları"
" ve verilen POI isimleri açısından karşılaştır."
" Kullanıcı soru sorarsa, hem sayısal verilere hem de POI isimlerine dayanarak"
" açıklayıcı ve dengeli bir karşılaştırma yap."
]
compare_context = "\n".join(compare_context_parts)
# Harita HTML'i
map_html = create_comparison_map(gdf1, gdf2)
return stats1, stats2, compare_context, map_html,gdf1, gdf2
def add_poi_markers_to_map(pois, m, layer_prefix="POI"):
"""
POI GeoDataFrame'ini alır, amenity/leisure/shop/railway/highway/public_transport
sütunlarına göre kategorik katmanlar oluşturup haritaya ekler.
"""
if pois is None or len(pois) == 0:
return
# Gerekirse WGS84'e (lat/lon) çevir
try:
if pois.crs is not None and pois.crs.to_epsg() != 4326:
pois = pois.to_crs(epsg=4326)
except Exception:
# CRS yoksa veya hata olursa direkt devam
pass
# >>> ÖNEMLİ: layer_groups burada tanımlanmalı
layer_groups = {} # {kategori_ismi: folium.FeatureGroup}
for _, row in pois.iterrows():
geom = row.geometry
if geom is None or geom.is_empty:
continue
# Nokta değilse centroid al
try:
if geom.geom_type == "Point":
lat, lon = geom.y, geom.x
else:
c = geom.centroid
lat, lon = c.y, c.x
except Exception:
continue
amenity = row.get("amenity")
leisure = row.get("leisure")
shop = row.get("shop")
highway = row.get("highway")
railway = row.get("railway")
public_transport = row.get("public_transport")
# Kategori belirle (öncelik: amenity > leisure > shop > railway > highway > public_transport)
if isinstance(amenity, str):
cat = f"Amenity: {amenity}"
elif isinstance(leisure, str):
cat = f"Leisure: {leisure}"
elif isinstance(shop, str):
cat = f"Shop: {shop}"
elif isinstance(railway, str):
cat = f"Railway: {railway}"
elif isinstance(highway, str):
cat = f"Highway: {highway}"
elif isinstance(public_transport, str):
cat = f"PT: {public_transport}"
else:
cat = "Diğer"
layer_name = f"{layer_prefix} - {cat}"
if layer_name not in layer_groups:
fg = folium.FeatureGroup(name=layer_name, show=True)
fg.add_to(m)
layer_groups[layer_name] = fg
name = row.get("name")
popup_items = []
# Önce isim
if isinstance(name, str) and name.strip():
popup_items.append(name.strip())
if isinstance(amenity, str):
popup_items.append(f"amenity={amenity}")
if isinstance(leisure, str):
popup_items.append(f"leisure={leisure}")
if isinstance(shop, str):
popup_items.append(f"shop={shop}")
if isinstance(railway, str):
popup_items.append(f"railway={railway}")
if isinstance(highway, str):
popup_items.append(f"highway={highway}")
if isinstance(public_transport, str):
popup_items.append(f"public_transport={public_transport}")
popup_text = ", ".join(popup_items) if popup_items else "POI"
folium.CircleMarker(
location=[lat, lon],
radius=4,
popup=popup_text,
tooltip=name.strip() if isinstance(name, str) and name.strip() else None,
weight=1,
fill=True,
fill_opacity=0.7,
).add_to(layer_groups[layer_name])
def create_comparison_map(gdf1, gdf2):
"""
İki mahalle poligonunu tek bir Folium haritasında gösterir.
+ Her mahalle için POI'leri kategorik katmanlar hâlinde ekler.
"""
# Hiç veri yoksa
if (gdf1 is None or len(gdf1) == 0) and (gdf2 is None or len(gdf2) == 0):
return "Harita için yeterli veri yok."
# Merkez olarak mevcut bir poligonun centroid'ini al
centroid = None
if gdf1 is not None and len(gdf1) > 0:
centroid = gdf1.geometry.iloc[0].centroid
elif gdf2 is not None and len(gdf2) > 0:
centroid = gdf2.geometry.iloc[0].centroid
if centroid is None:
return "Geometri bulunamadı."
m = folium.Map(location=[centroid.y, centroid.x], zoom_start=14)
# Mahalle 1
pois1 = None
if gdf1 is not None and len(gdf1) > 0:
name1 = str(gdf1.get("neighborhood", ["Mahalle 1"]).iloc[0])
folium.GeoJson(
gdf1.geometry.__geo_interface__,
name=name1,
style_function=lambda feat: {
"color": "red",
"fill": False,
"weight": 3,
},
).add_to(m)
# >>> Mahalle 1 için POI'ler
pois1 = get_pois_within(gdf1)
add_poi_markers_to_map(pois1, m, layer_prefix=f"{name1} POI")
# Mahalle 2
pois2 = None
if gdf2 is not None and len(gdf2) > 0:
name2 = str(gdf2.get("neighborhood", ["Mahalle 2"]).iloc[0])
folium.GeoJson(
gdf2.geometry.__geo_interface__,
name=name2,
style_function=lambda feat: {
"color": "blue",
"fill": False,
"weight": 3,
},
).add_to(m)
# >>> Mahalle 2 için POI'ler
pois2 = get_pois_within(gdf2)
add_poi_markers_to_map(pois2, m, layer_prefix=f"{name2} POI")
folium.LayerControl().add_to(m)
return m._repr_html_()
def run_overpass_to_map(query: str,
previous_elements: list | None,
previous_spatial: str | None,
layer_color: str,
base_gdf1=None,
base_gdf2=None):
layer_name = "Spatial Query"
query = normalize_overpass_query(query) if 'normalize_overpass_query' in globals() else query
if not query or not query.strip():
return (
"Overpass sorgusu boş.",
previous_spatial or "Geçerli bir Overpass sorgusu sağlanmadı.",
previous_elements,
)
url = "https://overpass-api.de/api/interpreter"
try:
resp = requests.post(url, data={"data": query}, timeout=30)
resp.raise_for_status()
data = resp.json()
except Exception as e:
print("Overpass isteği hatası:", e)
try:
print("Overpass response text:", resp.text[:500])
except Exception:
pass
return (
f"Overpass isteği hatası: {e}",
previous_spatial or "Overpass isteğinde hata oluştu, veri yok.",
previous_elements,
)
new_elements = data.get("elements", [])
if not new_elements:
return (
"Overpass sonucu: veri bulunamadı.",
previous_spatial or "Overpass sonucu: hiç element bulunamadı.",
previous_elements,
)
# ---- Eski + yeni elementleri birleştir ----
if not previous_elements:
all_elements = new_elements
else:
all_elements = previous_elements + new_elements
# ---- Merkez bul (tüm elementler üzerinden) ----
center_lat, center_lon = None, None
for el in all_elements:
if "lat" in el and "lon" in el:
center_lat, center_lon = el["lat"], el["lon"]
break
if center_lat is None:
return (
"Overpass sonucu: nokta verisi yok.",
previous_spatial or "Overpass sonucu: nokta verisi bulunamadı.",
all_elements,
)
# ---- Haritayı çiz ----
# Önce merkez: mümkünse mahallelerden, yoksa Overpass noktalardan
if base_gdf1 is not None and len(base_gdf1) > 0:
c = base_gdf1.geometry.iloc[0].centroid
center_lat, center_lon = c.y, c.x
elif base_gdf2 is not None and len(base_gdf2) > 0:
c = base_gdf2.geometry.iloc[0].centroid
center_lat, center_lon = c.y, c.x
# (aksi halde yukarıda all_elements'tan zaten bulduk)
m = folium.Map(location=[center_lat, center_lon], zoom_start=14)
# 1. mahalleyi tekrar çiz
if base_gdf1 is not None and len(base_gdf1) > 0:
name1 = str(base_gdf1.get("neighborhood", ["Mahalle 1"]).iloc[0])
folium.GeoJson(
base_gdf1.geometry.__geo_interface__,
name=name1,
style_function=lambda feat: {
"color": "red",
"fill": False,
"weight": 3,
},
).add_to(m)
pois1 = get_pois_within(base_gdf1)
add_poi_markers_to_map(pois1, m, layer_prefix=f"{name1} POI")
# 2. mahalleyi tekrar çiz
if base_gdf2 is not None and len(base_gdf2) > 0:
name2 = str(base_gdf2.get("neighborhood", ["Mahalle 2"]).iloc[0])
folium.GeoJson(
base_gdf2.geometry.__geo_interface__,
name=name2,
style_function=lambda feat: {
"color": "blue",
"fill": False,
"weight": 3,
},
).add_to(m)
pois2 = get_pois_within(base_gdf2)
add_poi_markers_to_map(pois2, m, layer_prefix=f"{name2} POI")
# Burada tek bir "spatial" katman oluşturuyoruz
fg_nodes = folium.FeatureGroup(
name=f"{layer_name} - Noktalar", # katman adı
show=True
)
fg_nodes.add_to(m)
fg_ways = folium.FeatureGroup(
name=f"{layer_name} - Yollar",
show=True
)
fg_ways.add_to(m)
for el in all_elements:
etype = el.get("type")
tags = el.get("tags", {}) or {}
name = tags.get("name", "")
popup_items = []
if name:
popup_items.append(name)
for k, v in tags.items():
if k != "name":
popup_items.append(f"{k}={v}")
popup_text = "
".join(popup_items) if popup_items else etype
if etype == "node" and "lat" in el and "lon" in el:
folium.CircleMarker(
location=[el["lat"], el["lon"]],
radius=4,
popup=popup_text,
tooltip=name or None,
weight=1,
fill=True,
fill_opacity=0.7,
color=layer_color, # 👈 stroke rengi
fill_color=layer_color, # 👈 doldurma rengi
).add_to(fg_nodes)
elif etype == "way" and "geometry" in el:
coords = [(p["lat"], p["lon"]) for p in el["geometry"]]
if len(coords) >= 2:
folium.PolyLine(
locations=coords,
popup=popup_text,
weight=3,
color=layer_color, # 👈 yol rengi
).add_to(fg_ways)
folium.LayerControl().add_to(m)
map_html = m._repr_html_()
# ---- Yeni spatial özet ----
new_summary = summarize_overpass_data({"elements": new_elements})
if previous_spatial:
combined_spatial = previous_spatial + "\n\n--- Yeni Sorgu ---\n" + new_summary
else:
combined_spatial = new_summary
return map_html, combined_spatial, all_elements
def llm_overpass_to_map(natural_prompt: str,
model_name: str,
previous_elements: list | None,
previous_spatial: str | None,
layer_color: str,
base_gdf1=None,
base_gdf2=None):
if not natural_prompt or not natural_prompt.strip():
return (
"Doğal dil sorgu boş.",
"Overpass sonucu: sorgu üretilemedi.",
previous_spatial or "Overpass sonucu: sorgu üretilemedi.",
previous_elements,
)
try:
query = generate_overpass_query_from_llm(natural_prompt, model_name)
except Exception as e:
print("LLM Overpass üretim hatası:", e)
return (
f"LLM Overpass üretim hatası: {e}",
"Overpass sonucu: LLM hatası.",
previous_spatial or "Overpass sonucu: LLM hatası.",
previous_elements,
)
map_html, combined_spatial, all_elements = run_overpass_to_map(
query,
previous_elements,
previous_spatial,
layer_color,
base_gdf1,
base_gdf2,
)
# 4 output: üretilen query, harita, birikmiş spatial metin, birikmiş element listesi
return query, map_html, combined_spatial, all_elements
def summarize_overpass_data(data: dict, max_examples: int = 30) -> str:
"""
Overpass JSON sonucundan LLM'e verilecek metinsel bir özet üretir.
Çok büyük verilerde token patlamaması için sınırlı örnek verir.
"""
if not data:
return "Overpass sonucu boş veya geçersiz.\n"
elements = data.get("elements", [])
if not elements:
return "Overpass sonucu: hiç element bulunamadı.\n"
total_nodes = sum(1 for e in elements if e.get("type") == "node")
total_ways = sum(1 for e in elements if e.get("type") == "way")
total_rel = sum(1 for e in elements if e.get("type") == "relation")
# Basit tag istatistikleri
amenity_counts = {}
leisure_counts = {}
shop_counts = {}
highway_counts = {}
railway_counts = {}
pt_counts = {}
examples = []
for e in elements[:max_examples]:
etype = e.get("type")
tags = e.get("tags", {})
name = tags.get("name", "(isimsiz)")
amenity = tags.get("amenity")
leisure = tags.get("leisure")
shop = tags.get("shop")
highway = tags.get("highway")
railway = tags.get("railway")
pt = tags.get("public_transport")
opening_hours = tags.get("opening_hours")
wheelchair = tags.get("wheelchair")
if amenity:
amenity_counts[amenity] = amenity_counts.get(amenity, 0) + 1
if leisure:
leisure_counts[leisure] = leisure_counts.get(leisure, 0) + 1
if shop:
shop_counts[shop] = shop_counts.get(shop, 0) + 1
if highway:
highway_counts[highway] = highway_counts.get(highway, 0) + 1
if railway:
railway_counts[railway] = railway_counts.get(railway, 0) + 1
if pt:
pt_counts[pt] = pt_counts.get(pt, 0) + 1
# Örnek satır
tag_parts = []
for k in ["amenity", "leisure", "shop", "highway", "railway", "public_transport",
"opening_hours", "wheelchair"]:
v = tags.get(k)
if v:
tag_parts.append(f"{k}={v}")
tag_text = ", ".join(tag_parts) if tag_parts else "etiket yok"
examples.append(f"- {etype} | {name} | {tag_text}")
def dict_to_lines(title, d):
if not d:
return []
items = sorted(d.items(), key=lambda x: -x[1])
lines = [title]
for k, v in items:
lines.append(f" - {k}: {v}")
return lines
lines = [
f"Toplam node sayısı: {total_nodes}",
f"Toplam way sayısı: {total_ways}",
f"Toplam relation sayısı: {total_rel}",
"",
]
lines += dict_to_lines("Amenity türleri:", amenity_counts)
lines += dict_to_lines("Leisure türleri:", leisure_counts)
lines += dict_to_lines("Shop türleri:", shop_counts)
lines += dict_to_lines("Highway türleri:", highway_counts)
lines += dict_to_lines("Railway türleri:", railway_counts)
lines += dict_to_lines("Public transport türleri:", pt_counts)
if examples:
lines.append("")
lines.append(f"İlk {len(examples)} elementten bazı örnekler:")
lines.extend(examples)
return "\n".join(lines)
# ==========================================
# LLM SOHBET FONKSİYONU
# ==========================================
def respond(
message,
history,
model_name,
system_message,
max_tokens,
temperature,
top_p,
compare_context, # Mahalle karşılaştırma
spatial_context, # Overpass sonuçlar
):
# --------------- OSM SPATIAL QUERY MODU ---------------
# --- /osm ile başlayan mesajlar: sadece Overpass sorgusu üret ---
if is_osm_query(message):
user_text = message.lstrip()[4:].strip()
if not user_text:
user_text = "Generate an Overpass QL query for my request."
client = InferenceClient(model=model_name, token=HF_TOKEN if HF_TOKEN else None)
messages = [
{"role": "system", "content": OSM_QUERY_SYSTEM},
{"role": "user", "content": user_text},
]
query_text = ""
for chunk in client.chat_completion(
messages=messages,
max_tokens=400,
stream=True,
temperature=0.2,
top_p=0.9,
):
choices = chunk.choices
token_text = ""
if len(choices) and choices[0].delta.content:
token_text = choices[0].delta.content
query_text += token_text
# Kullanıcıya sadece sorguyu göster
yield f"Üretilen Overpass Sorgusu:\n```ql\n{query_text}\n```"
return
# --------------- NORMAL CHAT AKIŞI ---------------
"""
Streaming chat using Hugging Face Inference API.
history: list of {"role": "...", "content": "..."}
"""
# Temperature güvenli aralık (model max 2)
temperature = max(0.0, min(2.0, float(temperature)))
top_p = max(0.0, min(1.0, float(top_p)))
client = InferenceClient(model=model_name, token=HF_TOKEN if HF_TOKEN else None)
# System mesajına mahalle karşılaştırma bağlamını ekle
full_system = system_message
if compare_context:
full_system += (
"\n\nAşağıda aynı şehirdeki iki mahalleye ait sayısal özetler var.\n"
"Kullanıcı bu mahalleler hakkında soru sorarsa bu bağlama göre cevap ver:\n"
f"{compare_context}"
)
if spatial_context:
full_system += (
"\n\nAyrıca kullanıcı tarafından en son çalıştırılan bir Overpass (spatial) sorgusunun"
" özet sonuçları var. Kullanıcı bu sorgudan gelen veriler hakkında soru sorarsa,"
" bu özet bağlamına dayanarak cevap ver:\n"
f"{spatial_context}"
)
messages = [{"role": "system", "content": full_system}]
messages.extend(history)
messages.append({"role": "user", "content": message})
response = ""
for chunk in client.chat_completion(
messages=messages,
max_tokens=max_tokens,
stream=True,
temperature=temperature,
top_p=top_p,
):
choices = chunk.choices
token_text = ""
if len(choices) and choices[0].delta.content:
token_text = choices[0].delta.content
response += token_text
yield response + f"\n\n---\n**Model:** {model_name}"
# ==========================================
# GRADIO ARAYÜZÜ (SOL CHAT, SAĞ KARŞILAŞTIRMA PANELİ)
# ==========================================
with gr.Blocks() as demo:
gr.Markdown("## Mahalle Karşılaştırmalı Chat Botu")
compare_state = gr.State("")
spatial_state = gr.State("") # son Overpass özetleri
overpass_elements_state = gr.State([]) # 👈 tüm Overpass sonuçlarını biriktireceğimiz liste
gdf1_state = gr.State(None) # 1. mahalle geometri
gdf2_state = gr.State(None) # 2. mahalle geometri
with gr.Row():
# SOL SÜTUN: CHAT
with gr.Column(scale=2):
chatbox = gr.Chatbot(height=800, scale=1)
model_dropdown = gr.Dropdown(
choices=[
# Küçük modeller
"google/gemma-2-2b-it", # 2B
"meta-llama/Meta-Llama-3.1-8B-Instruct", # 8B
# Büyük modeller
"abacusai/Dracarys-72B-Instruct", # 72B
"Qwen/Qwen2.5-72B-Instruct", # 72B
# Çok büyük model
"openai/gpt-oss-120b", # 120Bdı
],
label="Model Seç (Bu listedeki modeller Hugging Face Inference API chat_completion ile uyumludur)",
value="google/gemma-2-2b-it"
)
system_box = gr.Textbox(
value="Sen şehir planlama ve mahalleler hakkında bilgi veren yardımsever bir asistansın.",
label="System message",
)
max_tokens_slider = gr.Slider(
minimum=1,
maximum=2048,
value=512,
step=1,
label="Max new tokens",
)
temperature_slider = gr.Slider(
minimum=0.0,
maximum=2.0, # model max 2
value=0.5,
step=0.1,
label="Temperature",
)
top_p_slider = gr.Slider(
minimum=0.1,
maximum=1.0,
value=0.5,
step=0.05,
label="Top-p (nucleus sampling)",
)
chatbot = gr.ChatInterface(
respond,
chatbot=chatbox,
type="messages",
title="Basit Chat Botu",
description="Küçük bir sohbet botu, HF Inference API ve OSM verisi ile çalışıyor.",
additional_inputs=[
model_dropdown,
system_box,
max_tokens_slider,
temperature_slider,
top_p_slider,
compare_state, # >>> mahalle karşılaştırma bağlamı
spatial_state, # >>> son Overpass sonucu özeti
],
)
# SAĞ SÜTUN: MAHALLE KARŞILAŞTIRMA PANELİ
with gr.Column(scale=1):
gr.Markdown("### Mahalle Karşılaştırma")
# 1. mahalle: ilçe + mahalle aynı satırda
with gr.Row():
city_in1 = gr.Textbox(
label="1. Şehir",
value="Ankara",
placeholder="Örn: Ankara",
)
district1_in = gr.Textbox(
label="1. İlçe",
value="Gölbaşı",
scale=1,
placeholder="Örn: Gölbaşı",
)
neigh1_in = gr.Textbox(
label="1. Mahalle",
value="İncek",
scale=2,
placeholder="Örn: İncek Mahallesi",
)
# 2. mahalle: ilçe + mahalle aynı satırda
with gr.Row():
city_in2 = gr.Textbox(
label="2. Şehir",
value="Ankara",
placeholder="Örn: Ankara",
)
district2_in = gr.Textbox(
label="2. İlçe",
value="Gölbaşı",
scale=1,
placeholder="Örn: Gölbaşı",
)
neigh2_in = gr.Textbox(
label="2. Mahalle",
value="Kızılcaşar",
scale=2,
placeholder="Örn: Kızılcaşar Mahallesi",
)
compare_btn = gr.Button("Karşılaştırmayı Hazırla")
with gr.Row():
stats1_box = gr.Textbox(
label="1. Mahalle Özeti",
lines=4,
)
stats2_box = gr.Textbox(
label="2. Mahalle Özeti",
lines=4,
)
map_html = gr.HTML(label="Mahalle Haritası")
gr.Markdown("### Spatial Query (Overpass)")
# 1) LLM'e doğal dil prompt'u
osm_nl_prompt = gr.Textbox(
label="LLM ile Overpass Sorgusu (Doğal Dil)",
lines=3,
placeholder="Örn: İncek ve Kızılcaşar çevresindeki tüm park ve okulları getir"
)
gen_and_run_btn = gr.Button("LLM ile Sorguyu Üret ve Çalıştır")
# 2) Üretilen veya manuel Overpass sorgusu
overpass_box = gr.Textbox(
label="Overpass Sorgusu",
lines=6,
placeholder="Buraya LLM'in ürettiği Overpass QL sorgusunu yapıştırın..."
)
layer_color_dd = gr.Dropdown(
choices=["red", "blue", "green", "orange", "purple"],
value="red",
label="Spatial katman rengi"
)
run_overpass_btn = gr.Button("Sorguyu Çalıştır ve Haritayı Güncelle")
# En yakın POI (generic)
poi_type_dropdown = gr.Dropdown(
choices=["Üniversite", "Market", "Hastane", "Eczane", "Park"],
value="Üniversite",
label="En yakın neyi bulmak istiyorsun?"
)
nearest_poi_btn = gr.Button("1. Mahalle için en yakın noktayı bul")
nearest_poi_box = gr.Textbox(
label="En Yakın POI",
lines=2,
)
nearest_poi_btn.click(
fn=nearest_poi_wrapper,
inputs=[city_in1, district1_in, neigh1_in, poi_type_dropdown],
outputs=[nearest_poi_box],
)
# LLM ile üret + çalıştır (birikimli)
gen_and_run_btn.click(
fn=llm_overpass_to_map,
inputs=[osm_nl_prompt, model_dropdown, overpass_elements_state,
spatial_state, layer_color_dd, gdf1_state, gdf2_state],
outputs=[overpass_box, map_html, spatial_state, overpass_elements_state],
)
# Manuel Overpass çalıştırma (birikimli)
run_overpass_btn.click(
fn=run_overpass_to_map,
inputs=[overpass_box, overpass_elements_state, spatial_state,
layer_color_dd, gdf1_state, gdf2_state],
outputs=[map_html, spatial_state, overpass_elements_state],
)
compare_btn.click(
fn=prepare_comparison,
inputs=[city_in1, district1_in, neigh1_in, city_in1, district2_in, neigh2_in],
outputs=[stats1_box, stats2_box, compare_state, map_html, gdf1_state, gdf2_state],
)
if __name__ == "__main__":
demo.launch()