feat(sf-b6): POI weighted top-7 endpoint GET /parcels/{cad}/poi-score
Formula: weight = (1 / (distance_m + 100)) * category_weight Top-7 POI from osm_poi_ekb ranked by weight DESC. 12 unit tests (pure function, no DB required).
This commit is contained in:
lekss361
parent
e872b1ebe4
commit
f82d8bc870
3 changed files with 378 additions and 0 deletions
|
|
@ -43,6 +43,7 @@ from app.services.site_finder.custom_pois import (
|
|||
get_overlaps_for_scoring as _get_custom_poi_overlaps,
|
||||
)
|
||||
from app.services.site_finder.gate_verdict import compute_gate_verdict
|
||||
from app.services.site_finder.poi_score import PoiScoreResponse, compute_poi_weighted_top7
|
||||
from app.services.site_finder.quarter_dump_lookup import (
|
||||
get_connection_points,
|
||||
get_quarter_dump_data,
|
||||
|
|
@ -2455,6 +2456,56 @@ def get_isochrones(
|
|||
}
|
||||
|
||||
|
||||
@router.get(
|
||||
"/{cad_num}/poi-score",
|
||||
response_model=PoiScoreResponse,
|
||||
summary="POI weighted top-7 (B6)",
|
||||
)
|
||||
async def get_poi_score(
|
||||
cad_num: str,
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
radius_m: Annotated[int, Query(ge=100, le=5000)] = 2000,
|
||||
) -> PoiScoreResponse:
|
||||
"""Вернуть top-7 ближайших POI для участка, взвешенных по формуле:
|
||||
|
||||
weight = (1 / (distance_m + 100)) * category_weight
|
||||
|
||||
POI берутся из osm_poi_ekb в заданном радиусе (default 2000м).
|
||||
Отсортированы по weight DESC — наиболее значимые объекты первыми.
|
||||
"""
|
||||
# Получить координаты центроида участка из геометрических таблиц
|
||||
coord_row = (
|
||||
db.execute(
|
||||
text("""
|
||||
SELECT ST_X(ST_Centroid(g.geom)) AS lon,
|
||||
ST_Y(ST_Centroid(g.geom)) AS lat
|
||||
FROM (
|
||||
SELECT geom FROM cad_quarters_geom WHERE cad_number = :c
|
||||
UNION ALL
|
||||
SELECT geom FROM cad_buildings WHERE cad_num = :c
|
||||
UNION ALL
|
||||
SELECT geom FROM cad_parcels_geom WHERE cad_num = :c
|
||||
) g
|
||||
LIMIT 1
|
||||
"""),
|
||||
{"c": cad_num},
|
||||
)
|
||||
.mappings()
|
||||
.first()
|
||||
)
|
||||
|
||||
if not coord_row:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"Геометрия для {cad_num} не найдена.",
|
||||
)
|
||||
|
||||
lat = float(coord_row["lat"])
|
||||
lon = float(coord_row["lon"])
|
||||
|
||||
return compute_poi_weighted_top7(db, cad_num, lat, lon, radius_m=radius_m)
|
||||
|
||||
|
||||
@router.post("/{cad_num}/competitors", response_model=CompetitorsResponse)
|
||||
async def get_parcel_competitors(
|
||||
cad_num: str,
|
||||
|
|
|
|||
159
backend/app/services/site_finder/poi_score.py
Normal file
159
backend/app/services/site_finder/poi_score.py
Normal file
|
|
@ -0,0 +1,159 @@
|
|||
"""POI weighted score для Site Finder (B6).
|
||||
|
||||
Формула: weight = (1 / (distance_m + 100)) * category_weight
|
||||
|
||||
Возвращает top-7 ближайших POI из osm_poi_ekb, отсортированных по weight DESC.
|
||||
Категории и их веса согласованы с _POI_WEIGHTS в parcels.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy import text
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Веса по категории — согласованы с _POI_WEIGHTS в parcels.py + новые из vault B6.
|
||||
# Задача: "2GIS-style ranking", метро самое приоритетное.
|
||||
CATEGORY_WEIGHTS: dict[str, float] = {
|
||||
"metro_stop": 6.0,
|
||||
"school": 5.0,
|
||||
"kindergarten": 4.5,
|
||||
"hospital": 4.0,
|
||||
"shop_supermarket": 3.5,
|
||||
"shop_mall": 4.0,
|
||||
"park": 3.5,
|
||||
"bus_stop": 4.5,
|
||||
"tram_stop": 2.0,
|
||||
"pharmacy": 2.5,
|
||||
"shop_small": 2.0,
|
||||
"default": 1.0,
|
||||
}
|
||||
|
||||
|
||||
class PoiScoreItem(BaseModel):
|
||||
"""Один POI в ranked-ответе."""
|
||||
|
||||
name: str | None
|
||||
category: str
|
||||
distance_m: float
|
||||
weight: float
|
||||
address: str | None
|
||||
|
||||
|
||||
class PoiScoreResponse(BaseModel):
|
||||
cad_num: str
|
||||
radius_m: int
|
||||
top_poi: list[PoiScoreItem]
|
||||
|
||||
|
||||
def _category_weight(category: str) -> float:
|
||||
"""Вернуть вес категории. Если не знаем — default."""
|
||||
return CATEGORY_WEIGHTS.get(category, CATEGORY_WEIGHTS["default"])
|
||||
|
||||
|
||||
def compute_poi_weighted_top7(
|
||||
db: Any,
|
||||
cad_num: str,
|
||||
lat: float,
|
||||
lon: float,
|
||||
radius_m: int = 2000,
|
||||
top_n: int = 7,
|
||||
) -> PoiScoreResponse:
|
||||
"""Найти top-N POI вокруг (lat, lon) в radius_m, ранжировать по weighted score.
|
||||
|
||||
Запрос к osm_poi_ekb через ST_DWithin + ST_Distance.
|
||||
Формула: weight = (1 / (distance_m + 100)) * category_weight
|
||||
|
||||
Args:
|
||||
db: SQLAlchemy Session
|
||||
cad_num: кадастровый номер (для ответа)
|
||||
lat: широта центроида участка
|
||||
lon: долгота центроида участка
|
||||
radius_m: радиус поиска в метрах (default 2000)
|
||||
top_n: количество POI в ответе (default 7)
|
||||
|
||||
Returns:
|
||||
PoiScoreResponse с отсортированными по weight DESC POI.
|
||||
"""
|
||||
# ST_DWithin с geography=true использует метры напрямую.
|
||||
# ST_Distance тоже в метрах при geography=true.
|
||||
rows = (
|
||||
db.execute(
|
||||
text("""
|
||||
SELECT
|
||||
p.name,
|
||||
p.category,
|
||||
p.tags,
|
||||
CAST(
|
||||
ST_Distance(
|
||||
p.geom::geography,
|
||||
ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography
|
||||
) AS double precision
|
||||
) AS distance_m
|
||||
FROM osm_poi_ekb p
|
||||
WHERE ST_DWithin(
|
||||
p.geom::geography,
|
||||
ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography,
|
||||
:radius_m
|
||||
)
|
||||
ORDER BY distance_m ASC
|
||||
LIMIT :limit
|
||||
"""),
|
||||
{
|
||||
"lat": lat,
|
||||
"lon": lon,
|
||||
"radius_m": radius_m,
|
||||
"limit": top_n * 10, # запрашиваем больше, потом ранжируем
|
||||
},
|
||||
)
|
||||
.mappings()
|
||||
.all()
|
||||
)
|
||||
|
||||
logger.debug(
|
||||
"poi_score: cad=%s lat=%.5f lon=%.5f radius=%dm → %d candidates",
|
||||
cad_num,
|
||||
lat,
|
||||
lon,
|
||||
radius_m,
|
||||
len(rows),
|
||||
)
|
||||
|
||||
items: list[PoiScoreItem] = []
|
||||
for row in rows:
|
||||
distance_m = float(row["distance_m"])
|
||||
category = row["category"] or "default"
|
||||
cat_weight = _category_weight(category)
|
||||
weight = (1.0 / (distance_m + 100.0)) * cat_weight
|
||||
|
||||
# Адрес из tags jsonb если есть
|
||||
tags: dict[str, str] = row["tags"] or {}
|
||||
addr_parts = [
|
||||
tags.get("addr:street"),
|
||||
tags.get("addr:housenumber"),
|
||||
]
|
||||
address = ", ".join(p for p in addr_parts if p) or None
|
||||
|
||||
items.append(
|
||||
PoiScoreItem(
|
||||
name=row["name"],
|
||||
category=category,
|
||||
distance_m=round(distance_m, 1),
|
||||
weight=round(weight, 6),
|
||||
address=address,
|
||||
)
|
||||
)
|
||||
|
||||
# Сортировка по weight DESC, берём top_n
|
||||
items.sort(key=lambda x: x.weight, reverse=True)
|
||||
top_items = items[:top_n]
|
||||
|
||||
return PoiScoreResponse(
|
||||
cad_num=cad_num,
|
||||
radius_m=radius_m,
|
||||
top_poi=top_items,
|
||||
)
|
||||
168
backend/tests/test_poi_score.py
Normal file
168
backend/tests/test_poi_score.py
Normal file
|
|
@ -0,0 +1,168 @@
|
|||
"""Tests for POI weighted score service (B6).
|
||||
|
||||
Юнит-тесты для чистой функции — без DB.
|
||||
"""
|
||||
|
||||
from app.services.site_finder.poi_score import (
|
||||
CATEGORY_WEIGHTS,
|
||||
PoiScoreResponse,
|
||||
_category_weight,
|
||||
compute_poi_weighted_top7,
|
||||
)
|
||||
|
||||
# ── unit: _category_weight ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_category_weight_metro():
|
||||
"""Метро имеет наибольший вес из всех категорий."""
|
||||
metro_w = _category_weight("metro_stop")
|
||||
for cat in CATEGORY_WEIGHTS:
|
||||
if cat != "metro_stop" and cat != "default":
|
||||
assert metro_w >= _category_weight(
|
||||
cat
|
||||
), f"metro_stop weight {metro_w} должен быть >= {cat} weight {_category_weight(cat)}"
|
||||
|
||||
|
||||
def test_category_weight_unknown_returns_default():
|
||||
w = _category_weight("unknown_category_xyz")
|
||||
assert w == CATEGORY_WEIGHTS["default"]
|
||||
|
||||
|
||||
def test_category_weight_all_positive():
|
||||
"""Все веса в CATEGORY_WEIGHTS должны быть положительными (B6 — ranking, не штраф)."""
|
||||
for cat, w in CATEGORY_WEIGHTS.items():
|
||||
assert w > 0, f"Вес {cat}={w} должен быть > 0"
|
||||
|
||||
|
||||
# ── unit: weight formula ratio ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_weight_formula_ratio():
|
||||
"""Ближний объект той же категории должен иметь больший вес."""
|
||||
cat = "school"
|
||||
cw = _category_weight(cat)
|
||||
w_near = (1.0 / (100.0 + 100.0)) * cw # 100м
|
||||
w_far = (1.0 / (1000.0 + 100.0)) * cw # 1000м
|
||||
assert w_near > w_far
|
||||
|
||||
|
||||
def test_weight_formula_category_dominates_at_equal_distance():
|
||||
"""При одинаковом расстоянии метро должно быть впереди автобусной остановки."""
|
||||
dist = 500.0
|
||||
w_metro = (1.0 / (dist + 100.0)) * _category_weight("metro_stop")
|
||||
w_bus = (1.0 / (dist + 100.0)) * _category_weight("bus_stop")
|
||||
assert w_metro > w_bus
|
||||
|
||||
|
||||
# ── unit: compute_poi_weighted_top7 with mock DB ───────────────────────────────
|
||||
|
||||
|
||||
class _MockMappings:
|
||||
def __init__(self, data: list[dict]) -> None:
|
||||
self._data = data
|
||||
|
||||
def all(self) -> list[dict]:
|
||||
return self._data # type: ignore[return-value]
|
||||
|
||||
|
||||
class _MockResult:
|
||||
def __init__(self, data: list[dict]) -> None:
|
||||
self._data = data
|
||||
|
||||
def mappings(self) -> "_MockMappings":
|
||||
return _MockMappings(self._data)
|
||||
|
||||
|
||||
class _MockDb:
|
||||
"""Минимальный мок SQLAlchemy Session для тестирования без БД."""
|
||||
|
||||
def __init__(self, rows: list[dict]) -> None:
|
||||
self._rows = rows
|
||||
|
||||
def execute(self, *_args: object, **_kwargs: object) -> _MockResult:
|
||||
return _MockResult(self._rows)
|
||||
|
||||
|
||||
def _make_row(name: str, category: str, distance_m: float) -> dict:
|
||||
return {
|
||||
"name": name,
|
||||
"category": category,
|
||||
"tags": {},
|
||||
"distance_m": distance_m,
|
||||
}
|
||||
|
||||
|
||||
def test_top7_returns_at_most_7():
|
||||
rows = [_make_row(f"POI {i}", "school", float(i * 50)) for i in range(1, 20)]
|
||||
db = _MockDb(rows)
|
||||
result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605)
|
||||
assert isinstance(result, PoiScoreResponse)
|
||||
assert len(result.top_poi) <= 7
|
||||
|
||||
|
||||
def test_top7_sorted_by_weight_desc():
|
||||
rows = [
|
||||
_make_row("Дальняя школа", "school", 1500.0),
|
||||
_make_row("Метро", "metro_stop", 300.0),
|
||||
_make_row("Близкая школа", "school", 100.0),
|
||||
]
|
||||
db = _MockDb(rows)
|
||||
result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605)
|
||||
weights = [item.weight for item in result.top_poi]
|
||||
assert weights == sorted(weights, reverse=True), "top_poi должны быть по weight DESC"
|
||||
|
||||
|
||||
def test_metro_beats_school_at_equal_distance():
|
||||
"""Метро в 300м должно быть на первом месте перед школой в 300м (равное расстояние)."""
|
||||
rows = [
|
||||
_make_row("Школа №1", "school", 300.0),
|
||||
_make_row("Метро Чкаловская", "metro_stop", 300.0),
|
||||
]
|
||||
db = _MockDb(rows)
|
||||
result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605)
|
||||
assert (
|
||||
result.top_poi[0].category == "metro_stop"
|
||||
), "При равном расстоянии метро (category_weight=6.0) должно быть выше школы (5.0)"
|
||||
|
||||
|
||||
def test_metro_first_when_close():
|
||||
"""Метро в 50м должно быть на первом месте перед школой в 300м."""
|
||||
rows = [
|
||||
_make_row("Школа №1", "school", 300.0),
|
||||
_make_row("Метро Чкаловская", "metro_stop", 50.0),
|
||||
]
|
||||
db = _MockDb(rows)
|
||||
result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605)
|
||||
assert result.top_poi[0].category == "metro_stop", (
|
||||
"Метро (weight=6.0) в 50м должно быть впереди школы (weight=5.0) в 300м — "
|
||||
f"metro_weight={(1/(50+100))*6:.5f} vs school_weight={(1/(300+100))*5:.5f}"
|
||||
)
|
||||
|
||||
|
||||
def test_empty_db_returns_empty_top_poi():
|
||||
db = _MockDb([])
|
||||
result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605)
|
||||
assert result.top_poi == []
|
||||
assert result.cad_num == "66:41:0204016:10"
|
||||
assert result.radius_m == 2000
|
||||
|
||||
|
||||
def test_address_built_from_tags():
|
||||
rows = [
|
||||
{
|
||||
"name": "Магазин",
|
||||
"category": "shop_small",
|
||||
"tags": {"addr:street": "ул. Ленина", "addr:housenumber": "10"},
|
||||
"distance_m": 200.0,
|
||||
}
|
||||
]
|
||||
db = _MockDb(rows)
|
||||
result = compute_poi_weighted_top7(db, "test", 56.838, 60.605)
|
||||
assert result.top_poi[0].address == "ул. Ленина, 10"
|
||||
|
||||
|
||||
def test_address_none_when_no_tags():
|
||||
rows = [_make_row("Парк", "park", 400.0)]
|
||||
db = _MockDb(rows)
|
||||
result = compute_poi_weighted_top7(db, "test", 56.838, 60.605)
|
||||
assert result.top_poi[0].address is None
|
||||
Loading…
Add table
Reference in a new issue