gendesign/backend/app/services/site_finder/saturation.py

"""POI saturation per capita — обеспеченность района социнфраструктурой (#42).

Идея (Issue #42 / Limit_POI_Factor_Tiny): «10 школ vs 1 школа на 5000 детей» в
straight-count POI-scoring дают одинаковый сигнал — это неверно. Насыщенность
*на душу целевой когорты* важнее абсолютного числа объектов.

Что считаем для района (одного из 8 админ-районов ЕКБ):
  • фактическое число объектов категории (школа / детсад / поликлиника),
    привязанных к району spatial-join'ом `ST_Contains(district.geom, poi.geom)`
    по таблице `osm_poi_ekb` (нормализованные категории);
  • население целевой когорты = population × age_share (детсад → 0-7,
    школа → 7-18, поликлиника → всё население);
  • норматив СП 42.13330 (мест/учреждений на 1000 чел. когорты);
  • provision_ratio = факт / норматив → severity (дефицит / норма / профицит).

ВАЖНО про данные:
  • population/area района — ФАКТ (Росстат 01.01.2025, миграция 154).
  • Возрастные когорты — ОЦЕНКА (региональные доли, age_cohorts_estimated=TRUE).
    Поэтому в ответе всегда `cohort_estimated`-флаг: число «детей школьного
    возраста» — производная оценка, НЕ район-факт переписи.

Влияние на POI-score (acceptance #42): saturation_multiplier(category, district):
  дефицитный район → ×1.2 (incentive строить school-adjacent), перенасыщенный
  → ×0.5. Чистая функция — drop-in для poi_score без изменения его сигнатур.

psycopg v3: все binds через CAST(:name AS type), а не postfix double-colon. Read-only SELECT.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

# ── Нормативы СП 42.13330.2016 (Свод правил «Градостроительство») ──────────────
# Источник значений — issue #42 + vault A3. Знаменатель когорты различается:
#   - школа/детсад нормируются на 1000 ДЕТЕЙ соответствующего возраста,
#   - поликлиника — на 10000 ВСЕХ жителей.
# capacity_per_1000 — мест (школа/сад) либо учреждений (поликлиника) на 1000 чел.
# cohort — какую возрастную долю населения брать как знаменатель.


@dataclass(frozen=True)
class _Norm:
    """Норматив обеспеченности одной POI-категории по СП 42.13330."""

    cohort: str  # 'preschool' | 'school' | 'all'
    capacity_per_1000: float  # мест/учреждений на 1000 чел. когорты
    unit: str  # человекочитаемая единица норматива


# poi_loader-категории (osm_poi_ekb.category) → норматив.
_NORMS: dict[str, _Norm] = {
    # Школа: 92 места на 1000 детей школьного возраста (7-18).
    "school": _Norm(cohort="school", capacity_per_1000=92.0, unit="мест/1000 школьников"),
    # Детсад: 35 мест на 1000 детей дошкольного возраста (0-7).
    "kindergarten": _Norm(
        cohort="preschool", capacity_per_1000=35.0, unit="мест/1000 дошкольников"
    ),
    # Поликлиника: 1 учреждение на 10000 жителей = 0.1 на 1000 всех жителей.
    "hospital": _Norm(cohort="all", capacity_per_1000=0.1, unit="учр./1000 жителей"),
}

# Сколько «мест» условно даёт один объект OSM. У OSM нет вместимости (только точка),
# поэтому считаем число ОБЪЕКТОВ и переводим в условные «места» множителем-капасити.
# Для поликлиник 1 объект = 1 учреждение. Для школ/садов — усреднённая вместимость
# типового объекта (оценка, vault A3); меняем здесь, если появится OSM `capacity`.
_OBJECT_CAPACITY: dict[str, float] = {
    "school": 600.0,  # мест в типовой школе (оценка)
    "kindergarten": 200.0,  # мест в типовом детсаду (оценка)
    "hospital": 1.0,  # 1 объект = 1 учреждение
}

# ── Severity-пороги по provision_ratio (факт/норматив) ───────────────────────
# < 0.5  — острый дефицит | 0.5–0.85 — дефицит | 0.85–1.3 — норма | > 1.3 — профицит.
_SEVERITY_ACUTE_DEFICIT = 0.5
_SEVERITY_DEFICIT = 0.85
_SEVERITY_SURPLUS = 1.3

# ── POI-score множитель (acceptance #42): дефицит → boost, профицит → штраф ───
_MULT_DEFICIT = 1.2  # дефицитный район: incentive строить такую соцобъект-adjacent
_MULT_SURPLUS = 0.5  # перенасыщенный район: сигнал «ещё одна школа мало что даёт»
_MULT_NEUTRAL = 1.0


def classify_severity(provision_ratio: float | None) -> str | None:
    """provision_ratio (факт/норматив) → ярлык severity. PURE.

    None → None (нет данных: нет когорты / нет норматива — НЕ «острый дефицит»).
    """
    if provision_ratio is None:
        return None
    if provision_ratio < _SEVERITY_ACUTE_DEFICIT:
        return "острый дефицит"
    if provision_ratio < _SEVERITY_DEFICIT:
        return "дефицит"
    if provision_ratio <= _SEVERITY_SURPLUS:
        return "норма"
    return "профицит"


def saturation_multiplier(category: str, provision_ratio: float | None) -> float:
    """Множитель для POI-score по обеспеченности района категорией. PURE.

    Acceptance #42: дефицитный район → ×1.2 (стимул строить школо-adjacent),
    перенасыщенный → ×0.5. Норма / нет данных / категория без норматива → ×1.0
    (нейтрально: не наказываем за отсутствие сигнала).
    """
    if category not in _NORMS or provision_ratio is None:
        return _MULT_NEUTRAL
    if provision_ratio < _SEVERITY_DEFICIT:
        return _MULT_DEFICIT
    if provision_ratio > _SEVERITY_SURPLUS:
        return _MULT_SURPLUS
    return _MULT_NEUTRAL


def compute_provision_ratio(
    *,
    poi_count: int,
    population: int,
    age_share: float | None,
    category: str,
) -> float | None:
    """Коэффициент обеспеченности района категорией = факт / норматив. PURE.

    факт_мест_на_1000   = poi_count × object_capacity / (cohort_pop / 1000)
    норматив_мест_на_1000 = _NORMS[category].capacity_per_1000
    provision_ratio       = факт / норматив

    cohort_pop = population × age_share (для 'all' — age_share=1.0).
    None если: неизвестная категория / нет населения / нулевая когорта.
    """
    norm = _NORMS.get(category)
    if norm is None or population <= 0:
        return None

    share = 1.0 if norm.cohort == "all" else age_share
    if share is None or share <= 0:
        return None

    cohort_pop = population * share
    if cohort_pop <= 0:
        return None

    object_capacity = _OBJECT_CAPACITY.get(category, 1.0)
    actual_per_1000 = (poi_count * object_capacity) / (cohort_pop / 1000.0)
    if norm.capacity_per_1000 <= 0:
        return None
    return actual_per_1000 / norm.capacity_per_1000


# ──────────────────────────────────────────────────────────────────────────────
# DB-зависимый слой: spatial POI-count по району + join демографии.
# ──────────────────────────────────────────────────────────────────────────────

# POI-count категорий в полигоне района + демография района. ekb_districts.geom —
# полигон (миграция 56/74), osm_poi_ekb.geom — Point. ST_Contains(poly, point).
# LEFT JOIN osm_poi_ekb, чтобы район без POI давал 0, а не отсутствие строки.
_DISTRICT_SATURATION_SQL = text(
    """
    SELECT
        d.population,
        d.area_km2,
        d.age_share_preschool,
        d.age_share_school,
        d.age_share_elderly,
        d.age_cohorts_estimated,
        d.source,
        d.as_of_date,
        COUNT(*) FILTER (WHERE p.category = 'school')       AS n_school,
        COUNT(*) FILTER (WHERE p.category = 'kindergarten') AS n_kindergarten,
        COUNT(*) FILTER (WHERE p.category = 'hospital')     AS n_hospital
      FROM ekb_district_demographics d
      JOIN ekb_districts g
        ON g.district_name = d.district_name
       AND g.geom IS NOT NULL
      LEFT JOIN osm_poi_ekb p
        ON p.category IN ('school', 'kindergarten', 'hospital')
       AND ST_Contains(g.geom, p.geom)
     WHERE d.district_name = CAST(:dn AS text)
     GROUP BY
        d.population, d.area_km2, d.age_share_preschool, d.age_share_school,
        d.age_share_elderly, d.age_cohorts_estimated, d.source, d.as_of_date
    """
)

# Какую возрастную долю брать для категории (для отчёта по когорте).
_CATEGORY_SHARE_COLUMN: dict[str, str] = {
    "school": "age_share_school",
    "kindergarten": "age_share_preschool",
    "hospital": "age_share_elderly",  # информативно; норматив hospital — на всё население
}
_CATEGORY_COUNT_COLUMN: dict[str, str] = {
    "school": "n_school",
    "kindergarten": "n_kindergarten",
    "hospital": "n_hospital",
}


def _round_or_none(value: float | None, digits: int = 3) -> float | None:
    return round(value, digits) if value is not None else None


def compute_district_saturation(db: Session, district_name: str) -> dict[str, Any] | None:
    """Saturation per capita района — блок для analyze (`infra.saturation`).

    Возвращает dict (JSON-готовый) либо None, если у района нет демографии
    (нет строки в ekb_district_demographics / нет geom-полигона).

    Структура:
        {
          "district": "Чкаловский",
          "population": 286277,
          "as_of_date": "2025-01-01",
          "source": "...",
          "cohorts_estimated": true,        # возрастные доли — оценка, НЕ факт
          "categories": {
            "school": {
              "poi_count": 12,
              "cohort_population": 32922,    # оценка (population × age_share)
              "cohort_estimated": true,
              "provision_ratio": 0.68,       # факт/норматив
              "severity": "дефицит",
              "score_multiplier": 1.2,       # влияние на POI-score (#42)
              "norm": {"capacity_per_1000": 92.0, "unit": "мест/1000 школьников"}
            }, ...
          }
        }

    Graceful: на ошибке БД логирует и возвращает None (не роняет analyze).
    psycopg v3: bind через CAST(:dn AS text). Read-only.
    """
    try:
        row = db.execute(_DISTRICT_SATURATION_SQL, {"dn": district_name}).mappings().first()
    except Exception:
        logger.exception("saturation: query failed (district=%s)", district_name)
        return None

    if row is None or row["population"] is None:
        return None

    population = int(row["population"])
    cohorts_estimated = bool(row["age_cohorts_estimated"])

    categories: dict[str, Any] = {}
    for category, norm in _NORMS.items():
        poi_count = int(row[_CATEGORY_COUNT_COLUMN[category]] or 0)
        share_col = _CATEGORY_SHARE_COLUMN[category]
        raw_share = row[share_col]
        age_share = float(raw_share) if raw_share is not None else None

        provision = compute_provision_ratio(
            poi_count=poi_count,
            population=population,
            age_share=age_share,
            category=category,
        )

        # Когорта-знаменатель для отчёта: для hospital норматив на всё население.
        if norm.cohort == "all":
            cohort_pop: float | None = float(population)
            cohort_is_estimate = False  # всё население — факт
        elif age_share is not None:
            cohort_pop = population * age_share
            cohort_is_estimate = cohorts_estimated
        else:
            cohort_pop = None
            cohort_is_estimate = cohorts_estimated

        categories[category] = {
            "poi_count": poi_count,
            "cohort_population": int(cohort_pop) if cohort_pop is not None else None,
            "cohort_estimated": cohort_is_estimate,
            "provision_ratio": _round_or_none(provision),
            "severity": classify_severity(provision),
            "score_multiplier": saturation_multiplier(category, provision),
            "norm": {
                "capacity_per_1000": norm.capacity_per_1000,
                "unit": norm.unit,
            },
        }

    as_of = row["as_of_date"]
    return {
        "district": district_name,
        "population": population,
        "as_of_date": as_of.isoformat() if as_of is not None else None,
        "source": row["source"],
        "cohorts_estimated": cohorts_estimated,
        "categories": categories,
    }