admin/wiki
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

auto-sync: 2026-05-04 11:20:01

Browse Source
This commit is contained in:
Stream
2026-05-04 11:20:01 +03:00
parent ff7d301fc4
commit d1e0267184
1 changed files with 59 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
tasks/enduro-trails/prototype/app.py
View File

@@ -227,15 +227,15 @@ def haversine_m(lon1, lat1, lon2, lat2) -> float:
def calc_route_stats(geometry: dict, conn) -> dict | None: def calc_route_stats(geometry: dict, conn) -> dict | None:
""" """
Считает статистику покрытия маршрута по типам дорог. Считает статистику покрытия маршрута по типам дорог.
Оптимизированная версия: сэмплирует каждые ~500м, один запрос на сэмпл.
geometry — GeoJSON LineString {"type":"LineString","coordinates":[[lon,lat],...]} geometry — GeoJSON LineString {"type":"LineString","coordinates":[[lon,lat],...]}
Возвращает словарь с метрами и процентами по категориям.
""" """
try: try:
coords = geometry.get("coordinates", []) coords = geometry.get("coordinates", [])
if len(coords) < 2: if len(coords) < 2:
return None return None
# Считаем общую длину маршрута и шаг между точками # Считаем длины сегментов и общую длину
total_len = 0.0 total_len = 0.0
seg_lengths = [] seg_lengths = []
for i in range(len(coords) - 1): for i in range(len(coords) - 1):
@@ -246,12 +246,11 @@ def calc_route_stats(geometry: dict, conn) -> dict | None:
if total_len < 1: if total_len < 1:
return None return None
# Средний шаг между точками # Сэмплируем каждые ~500м (не чаще чем каждые 5 точек)
avg_step = total_len / len(seg_lengths) # Для маршрута 100км → ~200 сэмплов, для 500км → ~1000 сэмплов
avg_step = total_len / max(len(seg_lengths), 1)
# Сколько точек пропускать чтобы получить ~100м сегменты # Шаг в точках для ~500м интервала
# Минимум 1 точка, максимум 50 pts_per_500m = max(5, int(round(500.0 / avg_step))) if avg_step > 0 else 20
step = max(1, min(50, int(round(100.0 / avg_step)))) if avg_step > 0 else 5
cur = conn.cursor() cur = conn.cursor()
@@ -262,10 +261,14 @@ def calc_route_stats(geometry: dict, conn) -> dict | None:
"asphalt_m": 0.0, "asphalt_m": 0.0,
} }
# Проходим по маршруту с шагом step, берём среднюю точку сегмента # Кэш результатов по ячейкам сетки ~0.01° (~1км)
# Чтобы не делать повторные запросы для близких точек
grid_cache: dict = {}
i = 0 i = 0
while i < len(coords) - 1: while i < len(coords) - 1:
end_i = min(i + step, len(coords) - 1) end_i = min(i + pts_per_500m, len(coords) - 1)
# Средняя точка сегмента # Средняя точка сегмента
mid_lon = (coords[i][0] + coords[end_i][0]) / 2 mid_lon = (coords[i][0] + coords[end_i][0]) / 2
mid_lat = (coords[i][1] + coords[end_i][1]) / 2 mid_lat = (coords[i][1] + coords[end_i][1]) / 2
@@ -273,45 +276,50 @@ def calc_route_stats(geometry: dict, conn) -> dict | None:
# Длина этого сегмента # Длина этого сегмента
seg_len = sum(seg_lengths[i:end_i]) seg_len = sum(seg_lengths[i:end_i])
# Bbox для поиска ближайшего трека (~500м вокруг точки) # Ключ кэша: ячейка 0.01° (~1км)
delta = 0.005 # ~500м grid_key = (round(mid_lon * 100), round(mid_lat * 100))
try:
cur.execute("""
SELECT highway_type, track_type, length_m
FROM trails
WHERE min_lon <= ? AND max_lon >= ?
AND min_lat <= ? AND max_lat >= ?
ORDER BY ABS(min_lon - ?) + ABS(min_lat - ?)
LIMIT 1
""", (
mid_lon + delta, mid_lon - delta,
mid_lat + delta, mid_lat - delta,
mid_lon, mid_lat
))
row = cur.fetchone()
except Exception:
row = None
if row: if grid_key in grid_cache:
hw = (row["highway_type"] or "").lower() hw, tt = grid_cache[grid_key]
tt = (row["track_type"] or "").lower() else:
if hw == "track": # Bbox ~300м вокруг точки
if tt in ("grade1", "grade2"): delta = 0.003
stats["track_lev12_m"] += seg_len try:
else: cur.execute("""
# grade3/4/5 или NULL SELECT highway_type, track_type
stats["track_lev345_m"] += seg_len FROM trails
elif hw in ("path", "bridleway", "footway"): WHERE min_lon <= ? AND max_lon >= ?
stats["path_m"] += seg_len AND min_lat <= ? AND max_lat >= ?
else: ORDER BY ABS((min_lon + max_lon) / 2.0 - ?) +
stats["asphalt_m"] += seg_len ABS((min_lat + max_lat) / 2.0 - ?)
LIMIT 1
""", (
mid_lon + delta, mid_lon - delta,
mid_lat + delta, mid_lat - delta,
mid_lon, mid_lat
))
row = cur.fetchone()
if row:
hw = (row["highway_type"] or "").lower()
tt = (row["track_type"] or "").lower()
else:
hw, tt = "asphalt", ""
except Exception:
hw, tt = "asphalt", ""
grid_cache[grid_key] = (hw, tt)
if hw == "track":
if tt in ("grade1", "grade2"):
stats["track_lev12_m"] += seg_len
else:
stats["track_lev345_m"] += seg_len
elif hw in ("path", "bridleway", "footway"):
stats["path_m"] += seg_len
else: else:
# Нет данных — считаем асфальтом
stats["asphalt_m"] += seg_len stats["asphalt_m"] += seg_len
i = end_i i = end_i
# Считаем итоговую длину из статистики
computed_total = ( computed_total = (
stats["track_lev12_m"] + stats["track_lev345_m"] + stats["track_lev12_m"] + stats["track_lev345_m"] +
stats["path_m"] + stats["asphalt_m"] stats["path_m"] + stats["asphalt_m"]
@@ -325,15 +333,15 @@ def calc_route_stats(geometry: dict, conn) -> dict | None:
dirt_total = stats["track_lev12_m"] + stats["track_lev345_m"] + stats["path_m"] dirt_total = stats["track_lev12_m"] + stats["track_lev345_m"] + stats["path_m"]
return { return {
"track_lev12_m": round(stats["track_lev12_m"]), "track_lev12_m": round(stats["track_lev12_m"]),
"track_lev345_m": round(stats["track_lev345_m"]), "track_lev345_m": round(stats["track_lev345_m"]),
"path_m": round(stats["path_m"]), "path_m": round(stats["path_m"]),
"asphalt_m": round(stats["asphalt_m"]), "asphalt_m": round(stats["asphalt_m"]),
"track_lev12_pct": pct(stats["track_lev12_m"]), "track_lev12_pct": pct(stats["track_lev12_m"]),
"track_lev345_pct": pct(stats["track_lev345_m"]), "track_lev345_pct": pct(stats["track_lev345_m"]),
"path_pct": pct(stats["path_m"]), "path_pct": pct(stats["path_m"]),
"asphalt_pct": pct(stats["asphalt_m"]), "asphalt_pct": pct(stats["asphalt_m"]),
"dirt_total_pct": pct(dirt_total), "dirt_total_pct": pct(dirt_total),
} }
except Exception: except Exception:
return None return None