- 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>
34 lines
1.4 KiB
Python
34 lines
1.4 KiB
Python
from pydantic import BaseModel, Field
|
|
|
|
|
|
class NeighborhoodPOICounts(BaseModel):
|
|
education: int = Field(0, ge=0, description="SCHOOL + UNIVERSITY within 2km")
|
|
healthcare: int = Field(0, ge=0, description="HOSPITAL + CLINIC within 3km")
|
|
transport: int = Field(0, ge=0, description="METRO_STATION + BUS_STOP within 1km")
|
|
shopping: int = Field(0, ge=0, description="MALL + MARKET + SUPERMARKET within 2km")
|
|
greenery: int = Field(0, ge=0, description="PARK within 1km")
|
|
safety: int = Field(0, ge=0, description="POLICE_STATION + FIRE_STATION within 3km")
|
|
|
|
|
|
class NeighborhoodScoreRequest(BaseModel):
|
|
district: str = Field(..., min_length=1, description="District name (e.g. Quận 1)")
|
|
city: str = Field(..., min_length=1, description="City name (e.g. Hồ Chí Minh)")
|
|
poi_counts: NeighborhoodPOICounts = Field(
|
|
...,
|
|
description="Per-category POI counts already filtered by radius in NestJS",
|
|
)
|
|
|
|
|
|
class NeighborhoodScoreResponse(BaseModel):
|
|
district: str
|
|
city: str
|
|
education_score: float = Field(..., ge=0, le=10)
|
|
healthcare_score: float = Field(..., ge=0, le=10)
|
|
transport_score: float = Field(..., ge=0, le=10)
|
|
shopping_score: float = Field(..., ge=0, le=10)
|
|
greenery_score: float = Field(..., ge=0, le=10)
|
|
safety_score: float = Field(..., ge=0, le=10)
|
|
total_score: float = Field(..., ge=0, le=100)
|
|
poi_counts: dict[str, int]
|
|
algorithm_version: str
|