2c1e3771e9
- libs/ai-services: new POST /neighborhood/score router computing weighted 6-axis livability score from per-category POI counts; algorithm versioned for future iteration (sigmoid curves, percentile thresholds). - apps/api: HttpNeighborhoodScoreService proxies to Python first, falls back to PrismaNeighborhoodScoreService when AI service unavailable. Mirrors the HttpAVMService pattern. Existing GET /analytics/neighborhoods/:district/score endpoint and CQRS handler now flow through the proxy. - AnalyticsModule binds Http variant by default, retains Prisma variant as injectable fallback. - Tests: 5 pytest cases for Python heuristic, 4 vitest cases for HTTP proxy fallback behaviour. Co-Authored-By: Paperclip <noreply@paperclip.ing>
72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
import logging
|
|
|
|
from app.models.neighborhood import (
|
|
NeighborhoodPOICounts,
|
|
NeighborhoodScoreRequest,
|
|
NeighborhoodScoreResponse,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ALGORITHM_VERSION = "neighborhood-heuristic-v1"
|
|
|
|
# Sum = 100. Mirrors NestJS PrismaNeighborhoodScoreServiceImpl for fallback parity.
|
|
CATEGORY_WEIGHTS: dict[str, int] = {
|
|
"education": 20,
|
|
"healthcare": 20,
|
|
"transport": 20,
|
|
"shopping": 15,
|
|
"greenery": 15,
|
|
"safety": 10,
|
|
}
|
|
|
|
# Count yielding a 10/10 sub-score. Calibrated against HCMC/HN audit benchmarks.
|
|
MAX_COUNTS: dict[str, int] = {
|
|
"education": 15,
|
|
"healthcare": 8,
|
|
"transport": 12,
|
|
"shopping": 10,
|
|
"greenery": 6,
|
|
"safety": 4,
|
|
}
|
|
|
|
|
|
class NeighborhoodScoreService:
|
|
"""Stateless scoring algorithm.
|
|
|
|
NestJS owns the PostGIS radius query and passes per-category counts.
|
|
This service applies the weighting + capping curve so the algorithm
|
|
can evolve independently of the persistence layer.
|
|
"""
|
|
|
|
def score(self, req: NeighborhoodScoreRequest) -> NeighborhoodScoreResponse:
|
|
counts = req.poi_counts
|
|
sub_scores = self._sub_scores(counts)
|
|
total = sum(
|
|
CATEGORY_WEIGHTS[cat] * sub_scores[cat] / 10.0 for cat in CATEGORY_WEIGHTS
|
|
)
|
|
|
|
return NeighborhoodScoreResponse(
|
|
district=req.district,
|
|
city=req.city,
|
|
education_score=sub_scores["education"],
|
|
healthcare_score=sub_scores["healthcare"],
|
|
transport_score=sub_scores["transport"],
|
|
shopping_score=sub_scores["shopping"],
|
|
greenery_score=sub_scores["greenery"],
|
|
safety_score=sub_scores["safety"],
|
|
total_score=round(total, 1),
|
|
poi_counts=counts.model_dump(),
|
|
algorithm_version=ALGORITHM_VERSION,
|
|
)
|
|
|
|
def _sub_scores(self, counts: NeighborhoodPOICounts) -> dict[str, float]:
|
|
raw = counts.model_dump()
|
|
return {
|
|
cat: round(min(10.0, raw[cat] / MAX_COUNTS[cat] * 10.0), 2)
|
|
for cat in CATEGORY_WEIGHTS
|
|
}
|
|
|
|
|
|
neighborhood_score_service = NeighborhoodScoreService()
|