feat(site-finder): OSM-геокод городских ТП/РП ЕКБ (#2119 B2, финальный хвост) #2252
4 changed files with 641 additions and 0 deletions
319
backend/app/services/site_finder/osm_tp_geocoder.py
Normal file
319
backend/app/services/site_finder/osm_tp_geocoder.py
Normal file
|
|
@ -0,0 +1,319 @@
|
|||
"""OSM-геокод городских ТП/РП ЕКБ → geom для power_tp_rp_reserves (#2119 B2, финал).
|
||||
|
||||
Хвост §3 «свободная мощность»: у ~2012 ЕКБ-строк power_tp_rp_reserves (ЕЭСК,
|
||||
«ниже 35» кВ) name_norm = чистый номер («10003»), но geom NULL. OSM в bbox ЕКБ
|
||||
несёт ~821 substation; часть подписана номером ТП/РП в теге ``ref`` или ``name``
|
||||
(«РП-101», «ТП-180», «ТП 1591») — по нему матчим наши строки и заливаем координату.
|
||||
|
||||
Разведка 2026-07-03: матч по номеру даёт ~119 наших ТП, 0 неоднозначностей в срезе.
|
||||
Консервативно: берём число только если это похоже на номер ТП/РП (ТП/РП/RP/TP-токен
|
||||
рядом ИЛИ голый ref-число) — «110 кВ» из voltage мимо. Дубль-номера (>1 OSM-фичи на
|
||||
один номер) честно скипаем в ambiguous.
|
||||
|
||||
RUN-ON-PROD-friendly, но источник (Overpass) без гео-блока — можно и локально.
|
||||
httpx с явным таймаутом, per-row SAVEPOINT (backend.md), idempotent UPDATE
|
||||
(geom IS NULL + geocode_status='osm'). Dry-run по умолчанию; CLI ``--apply``.
|
||||
|
||||
Overpass endpoint: overpass.kumi.systems (overpass-api.de даёт 406 на наш POST;
|
||||
area-запрос роняет соединение — используем ТОЛЬКО bbox).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import re
|
||||
|
||||
import httpx
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.db import SessionLocal
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Overpass-инстанс: overpass-api.de отдаёт 406 на наш POST, kumi стабилен (разведка).
|
||||
OVERPASS_URL = "https://overpass.kumi.systems/api/interpreter"
|
||||
|
||||
# bbox ЕКБ (south, west, north, east) для Overpass. area-запрос роняет соединение —
|
||||
# ТОЛЬКО bbox. Тот же прямоугольник используем и для sanity-гейта координаты.
|
||||
_EKB_BBOX = (56.70, 60.40, 56.97, 60.80)
|
||||
|
||||
_HTTP_TIMEOUT = 60
|
||||
|
||||
# Overpass QL: node/way/relation с power=substation в bbox, вывод с центроидом (center)
|
||||
# и тегами. `out center tags;` — для way/relation отдаёт center-точку, для node — саму.
|
||||
_OVERPASS_QL = (
|
||||
"[out:json][timeout:{timeout}];"
|
||||
"("
|
||||
'node["power"="substation"]({s},{w},{n},{e});'
|
||||
'way["power"="substation"]({s},{w},{n},{e});'
|
||||
'relation["power"="substation"]({s},{w},{n},{e});'
|
||||
");"
|
||||
"out center tags;"
|
||||
)
|
||||
|
||||
# Токен «ТП/РП» (кириллица и латиница-двойники) + число 2-5 цифр после него.
|
||||
# Примеры: «РП-101» → 101, «ТП-180» → 180, «ТП 1591» → 1591, «TP-42» → 42.
|
||||
# Разделитель между токеном и числом — дефис/пробел/точка/двоеточие (или отсутствует).
|
||||
_TP_TOKEN_NUM_RE = re.compile(
|
||||
r"(?:тп|рп|tp|rp)\s*[-–—.:№]?\s*(\d{2,5})\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
# Голый ref-номер целиком: только цифры 2-5 знаков (напр. ref='101' без ТП-токена).
|
||||
_BARE_NUM_RE = re.compile(r"^\s*(\d{2,5})\s*$")
|
||||
|
||||
|
||||
def fetch_osm_substations() -> list[dict]:
|
||||
"""Тянет OSM-подстанции ЕКБ (power=substation) в bbox через Overpass → element-list.
|
||||
|
||||
POST data= (GET area роняет соединение). Скип ``substation=traction`` (тяговые
|
||||
ж/д подстанции — не городские ТП/РП). httpx с явным таймаутом. Возвращает
|
||||
сырые Overpass-элементы ``{"type","id","lat"/"center","tags"}``.
|
||||
"""
|
||||
ql = _OVERPASS_QL.format(
|
||||
timeout=_HTTP_TIMEOUT,
|
||||
s=_EKB_BBOX[0],
|
||||
w=_EKB_BBOX[1],
|
||||
n=_EKB_BBOX[2],
|
||||
e=_EKB_BBOX[3],
|
||||
)
|
||||
resp = httpx.post(OVERPASS_URL, data={"data": ql}, timeout=_HTTP_TIMEOUT)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
elements: list[dict] = (data or {}).get("elements") or []
|
||||
|
||||
kept: list[dict] = []
|
||||
for el in elements:
|
||||
tags = el.get("tags") or {}
|
||||
# Тяговые ж/д подстанции — не городские ТП/РП, выкидываем.
|
||||
if tags.get("substation") == "traction":
|
||||
continue
|
||||
kept.append(el)
|
||||
|
||||
logger.info(
|
||||
"osm_tp_geocoder: Overpass отдал %d substation, после traction-скипа %d",
|
||||
len(elements),
|
||||
len(kept),
|
||||
)
|
||||
return kept
|
||||
|
||||
|
||||
def extract_tp_number(tags: dict) -> str | None:
|
||||
"""Извлекает номер ТП/РП из тегов OSM-фичи (``ref`` приоритетнее ``name``).
|
||||
|
||||
Консервативно — только когда число выглядит номером ТП/РП, не вольтажем:
|
||||
- ``ref`` = голое число 2-5 цифр («101») → берём;
|
||||
- в ``ref``/``name`` есть ТП/РП/TP/RP-токен + число («РП-101», «ТП 1591») → берём число;
|
||||
- «110 кВ» / «110 kV» БЕЗ ТП-токена → None (это класс напряжения, не номер).
|
||||
|
||||
Приоритет источника: ref, затем name. Возвращает строку-номер или None.
|
||||
"""
|
||||
ref = (tags.get("ref") or "").strip()
|
||||
name = (tags.get("name") or "").strip()
|
||||
|
||||
# 1. Голый ref-номер (наиболее надёжный сигнал: оператор проставил ref явно).
|
||||
if ref:
|
||||
m_bare = _BARE_NUM_RE.match(ref)
|
||||
if m_bare:
|
||||
return m_bare.group(1)
|
||||
|
||||
# 2. ТП/РП-токен + число — сперва в ref, потом в name.
|
||||
for source in (ref, name):
|
||||
if not source:
|
||||
continue
|
||||
m = _TP_TOKEN_NUM_RE.search(source)
|
||||
if m:
|
||||
return m.group(1)
|
||||
|
||||
# «110 кВ» и прочий вольтаж без ТП-токена и без голого ref — намеренно None.
|
||||
return None
|
||||
|
||||
|
||||
def _center_coords(el: dict) -> tuple[float, float] | None:
|
||||
"""(lat, lon) из Overpass-элемента: node → lat/lon, way/relation → center. None — нет."""
|
||||
if el.get("lat") is not None and el.get("lon") is not None:
|
||||
return float(el["lat"]), float(el["lon"])
|
||||
center = el.get("center") or {}
|
||||
if center.get("lat") is not None and center.get("lon") is not None:
|
||||
return float(center["lat"]), float(center["lon"])
|
||||
return None
|
||||
|
||||
|
||||
def _in_ekb_bbox(lat: float, lon: float) -> bool:
|
||||
"""Координата в bbox ЕКБ? Overpass и так ограничен bbox'ом — проверка дёшева/страховка."""
|
||||
s, w, n, e = _EKB_BBOX
|
||||
return s <= lat <= n and w <= lon <= e
|
||||
|
||||
|
||||
def _index_osm_by_number(elements: list[dict]) -> tuple[dict[str, tuple[float, float]], set[str]]:
|
||||
"""Строит {номер → (lat,lon)} по OSM-фичам; номера с >1 фичей → ambiguous-set.
|
||||
|
||||
Номер извлекается ``extract_tp_number``; берётся только фича с валидной координатой
|
||||
в bbox ЕКБ. Если один номер несут ≥2 разных фичи (разные координаты) — он попадает
|
||||
в ambiguous и НЕ матчится (честность: не угадываем, какая из ТП «та самая»).
|
||||
"""
|
||||
by_number: dict[str, tuple[float, float]] = {}
|
||||
seen_coords: dict[str, set[tuple[float, float]]] = {}
|
||||
ambiguous: set[str] = set()
|
||||
|
||||
for el in elements:
|
||||
tags = el.get("tags") or {}
|
||||
number = extract_tp_number(tags)
|
||||
if number is None:
|
||||
continue
|
||||
coords = _center_coords(el)
|
||||
if coords is None:
|
||||
continue
|
||||
lat, lon = coords
|
||||
if not _in_ekb_bbox(lat, lon):
|
||||
logger.info(
|
||||
"osm_tp_geocoder: номер %s вне bbox ЕКБ (lat=%.4f lon=%.4f) → скип",
|
||||
number,
|
||||
lat,
|
||||
lon,
|
||||
)
|
||||
continue
|
||||
|
||||
prev = seen_coords.setdefault(number, set())
|
||||
# Округляем до ~11 м, чтобы дубли-теги одной ТП (way+node) не считались коллизией.
|
||||
coord_key = (round(lat, 4), round(lon, 4))
|
||||
prev.add(coord_key)
|
||||
if len(prev) > 1:
|
||||
ambiguous.add(number)
|
||||
by_number.pop(number, None)
|
||||
continue
|
||||
if number not in ambiguous:
|
||||
by_number[number] = (lat, lon)
|
||||
|
||||
return by_number, ambiguous
|
||||
|
||||
|
||||
# UPDATE geom + geocode_status='osm'. CAST(:x AS ...) — psycopg3-канон (backend.md,
|
||||
# НИКОГДА :x::float). Пишем ТОЛЬКО ЕКБ + geom всё ещё NULL → idempotent при повторе.
|
||||
_UPDATE_SQL = text("""
|
||||
UPDATE power_tp_rp_reserves
|
||||
SET geom = ST_SetSRID(
|
||||
ST_MakePoint(CAST(:lon AS float), CAST(:lat AS float)),
|
||||
4326
|
||||
),
|
||||
geocode_status = 'osm'
|
||||
WHERE name_norm = CAST(:name_norm AS text)
|
||||
AND municipality = 'Екатеринбург'
|
||||
AND geom IS NULL
|
||||
""")
|
||||
|
||||
# Кандидаты на геокод: ЕКБ-строки без координаты. name_norm = чистый номер.
|
||||
_CANDIDATES_SQL = text("""
|
||||
SELECT name_norm
|
||||
FROM power_tp_rp_reserves
|
||||
WHERE municipality = 'Екатеринбург'
|
||||
AND geom IS NULL
|
||||
AND name_norm IS NOT NULL
|
||||
ORDER BY name_norm
|
||||
""")
|
||||
|
||||
|
||||
def geocode_tp_from_osm(db: Session | None = None, *, apply: bool = False) -> dict:
|
||||
"""Матч name_norm ЕКБ-ТП/РП ↔ OSM-номер → UPDATE geom + geocode_status='osm'.
|
||||
|
||||
Дефолт dry-run (apply=False): считает счётчики + логирует первые 20 пар
|
||||
номер→(lat,lon), НО В БД НЕ ПИШЕТ. apply=True — реально пишет.
|
||||
|
||||
Алгоритм:
|
||||
1. fetch OSM-substation ЕКБ (traction уже отсеян).
|
||||
2. индекс {номер → координата}; дубль-номер → ambiguous (не матчим).
|
||||
3. SELECT ЕКБ-строк с geom IS NULL; matched, если номер есть в индексе.
|
||||
4. UPDATE geom WHERE name_norm=... AND municipality='Екатеринбург' AND geom IS NULL
|
||||
(idempotent). Per-row SAVEPOINT — битый UPDATE не валит батч.
|
||||
|
||||
Returns счётчики {osm_features, numbered, matched, updated, ambiguous, applied}.
|
||||
"""
|
||||
owns_session = db is None
|
||||
if db is None:
|
||||
db = SessionLocal()
|
||||
|
||||
counts = {
|
||||
"osm_features": 0,
|
||||
"numbered": 0,
|
||||
"matched": 0,
|
||||
"updated": 0,
|
||||
"ambiguous": 0,
|
||||
"applied": apply,
|
||||
}
|
||||
sample: list[tuple[str, tuple[float, float]]] = []
|
||||
|
||||
try:
|
||||
elements = fetch_osm_substations()
|
||||
counts["osm_features"] = len(elements)
|
||||
|
||||
by_number, ambiguous = _index_osm_by_number(elements)
|
||||
counts["numbered"] = len(by_number)
|
||||
counts["ambiguous"] = len(ambiguous)
|
||||
|
||||
candidates: list[str] = [str(row[0]) for row in db.execute(_CANDIDATES_SQL).all()]
|
||||
|
||||
for name_norm in candidates:
|
||||
coords = by_number.get(name_norm)
|
||||
if coords is None:
|
||||
continue
|
||||
counts["matched"] += 1
|
||||
lat, lon = coords
|
||||
if len(sample) < 20:
|
||||
sample.append((name_norm, (lat, lon)))
|
||||
|
||||
if not apply:
|
||||
continue
|
||||
|
||||
try:
|
||||
with db.begin_nested(): # SAVEPOINT — откат только этой строки
|
||||
res = db.execute(_UPDATE_SQL, {"name_norm": name_norm, "lat": lat, "lon": lon})
|
||||
if res.rowcount and res.rowcount > 0:
|
||||
counts["updated"] += res.rowcount
|
||||
except Exception as e:
|
||||
logger.warning("osm_tp_geocoder UPDATE failed for %r: %s", name_norm, e)
|
||||
|
||||
if apply:
|
||||
db.commit()
|
||||
except Exception as e:
|
||||
if apply:
|
||||
db.rollback()
|
||||
logger.exception("geocode_tp_from_osm: rolled back: %s", e)
|
||||
raise
|
||||
finally:
|
||||
if owns_session:
|
||||
db.close()
|
||||
|
||||
logger.info("geocode_tp_from_osm done (apply=%s): %s", apply, counts)
|
||||
if not apply:
|
||||
for number, (lat, lon) in sample:
|
||||
logger.info(" sample: ТП/РП %s → (%.6f, %.6f)", number, lat, lon)
|
||||
return counts
|
||||
|
||||
|
||||
def _main(argv: list[str] | None = None) -> int:
|
||||
"""CLI: OSM-геокод городских ТП/РП ЕКБ → geom. Dry-run по умолчанию.
|
||||
|
||||
Примеры:
|
||||
python -m app.services.site_finder.osm_tp_geocoder # dry-run
|
||||
python -m app.services.site_finder.osm_tp_geocoder --apply # пишет в БД
|
||||
"""
|
||||
parser = argparse.ArgumentParser(
|
||||
description="OSM-геокод городских ТП/РП ЕКБ (#2119 B2): матч name_norm ↔ "
|
||||
"OSM substation ref/name → geom + geocode_status='osm'."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--apply",
|
||||
action="store_true",
|
||||
help="Реально писать в БД. Без флага — dry-run (только счётчики + сэмпл).",
|
||||
)
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
|
||||
logger.info("osm_tp_geocoder result: %s", geocode_tp_from_osm(apply=args.apply))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
raise SystemExit(_main())
|
||||
|
|
@ -333,6 +333,16 @@ def build_beat_schedule() -> dict:
|
|||
"schedule": _parse_cron("30 5 * * tue"), # 05:30 МСК, вторник
|
||||
"options": {"queue": "celery"},
|
||||
}
|
||||
# OSM-геокод городских ТП/РП ЕКБ → geom (Forgejo #2119 Фаза B2, финал). Матч
|
||||
# name_norm (чистый номер) ↔ OSM substation ref/name через Overpass →
|
||||
# power_tp_rp_reserves.geom. Источник (Overpass) БЕЗ гео-блока, но держим в том же
|
||||
# вторник-кластере рядом с ЕЭСК. 05:45 МСК — после газа/ЕЭСК (05:30), до тепла
|
||||
# (06:00); идемпотентно (geom IS NULL-гейт), вне monday-Overpass-нагрузки.
|
||||
schedule["osm-tp-geocode-weekly"] = {
|
||||
"task": "tasks.connection_capacity_sync.sync_osm_tp_geocode",
|
||||
"schedule": _parse_cron("45 5 * * tue"), # 05:45 МСК, вторник
|
||||
"options": {"queue": "celery"},
|
||||
}
|
||||
# Тепло-резервы (Forgejo #2119 Фаза B2): резерв мощности источников тепла из
|
||||
# реестра раскрытия ФАС (ri.eias.ru, ЕТК + Екатеринбургэнерго). Источник ГЕО-
|
||||
# БЛОКИРУЕТ non-RU IP + российский УЦ → RUN-ON-PROD. 06:00 МСК — через 30 мин
|
||||
|
|
|
|||
|
|
@ -145,6 +145,32 @@ def sync_eesk_reserves() -> dict:
|
|||
raise
|
||||
|
||||
|
||||
@celery_app.task(
|
||||
name="tasks.connection_capacity_sync.sync_osm_tp_geocode",
|
||||
queue="celery",
|
||||
)
|
||||
def sync_osm_tp_geocode() -> dict:
|
||||
"""OSM-геокод городских ТП/РП ЕКБ → geom для power_tp_rp_reserves. #2119 B2 (финал).
|
||||
|
||||
RUN-ON-PROD рядом с ЕЭСК (тот же вторник-кластер). Матч name_norm (чистый номер)
|
||||
↔ OSM substation ref/name → UPDATE geom + geocode_status='osm'. Идемпотентно
|
||||
(geom IS NULL-гейт): не-новые точки просто пере-скипаются без эффекта. apply=True.
|
||||
Источник (Overpass) без гео-блока, но держим на проде в едином weekly-кластере.
|
||||
"""
|
||||
_log_breadcrumb("osm_tp_geocode_sync", "info", "task started")
|
||||
try:
|
||||
from app.services.site_finder.osm_tp_geocoder import geocode_tp_from_osm
|
||||
|
||||
result = geocode_tp_from_osm(apply=True)
|
||||
logger.info("sync_osm_tp_geocode done: %s", result)
|
||||
_log_breadcrumb("osm_tp_geocode_sync", "info", f"done: {result}")
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.exception("sync_osm_tp_geocode failed: %s", e)
|
||||
_log_breadcrumb("osm_tp_geocode_sync", "error", f"{type(e).__name__}: {e}")
|
||||
raise
|
||||
|
||||
|
||||
@celery_app.task(
|
||||
name="tasks.connection_capacity_sync.sync_gas_outlet_points",
|
||||
queue="celery",
|
||||
|
|
|
|||
286
backend/tests/test_osm_tp_geocoder.py
Normal file
286
backend/tests/test_osm_tp_geocoder.py
Normal file
|
|
@ -0,0 +1,286 @@
|
|||
"""Unit-тесты OSM-геокода городских ТП/РП ЕКБ (#2119 B2, финал).
|
||||
|
||||
Pure / mock-based — без реальной сети и БД. Покрывает:
|
||||
- extract_tp_number: «РП-101»→101, «ТП 1591»→1591, голый ref='101'→101,
|
||||
«110 кВ» БЕЗ ТП-токена/ref→None, traction-скип на уровне fetch.
|
||||
- fetch_osm_substations: скип substation=traction (mock httpx.post).
|
||||
- _index_osm_by_number: дубль-номер (разные координаты) → ambiguous, не матчится;
|
||||
дубль-теги одной точки (way+node, те же координаты) → НЕ ambiguous.
|
||||
- geocode_tp_from_osm: dry-run НЕ пишет; apply=True шлёт UPDATE с CAST-формами
|
||||
и geom IS NULL-гейтом; sanity-bbox отсекает координату вне ЕКБ.
|
||||
- Celery-регистрация таска sync_osm_tp_geocode + beat-строка.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import contextmanager
|
||||
from typing import Any
|
||||
|
||||
from app.services.site_finder import osm_tp_geocoder as og
|
||||
|
||||
# ── extract_tp_number ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_extract_tp_number_rp_token() -> None:
|
||||
"""«РП-101» в name → 101 (ТП-токен + число)."""
|
||||
assert og.extract_tp_number({"name": "РП-101"}) == "101"
|
||||
|
||||
|
||||
def test_extract_tp_number_tp_hyphen() -> None:
|
||||
"""«ТП-180» → 180."""
|
||||
assert og.extract_tp_number({"name": "ТП-180"}) == "180"
|
||||
|
||||
|
||||
def test_extract_tp_number_tp_space() -> None:
|
||||
"""«ТП 1591» (пробел-разделитель) → 1591."""
|
||||
assert og.extract_tp_number({"name": "ТП 1591"}) == "1591"
|
||||
|
||||
|
||||
def test_extract_tp_number_bare_ref() -> None:
|
||||
"""Голый ref='101' (без ТП-токена) → 101 — оператор проставил ref явно."""
|
||||
assert og.extract_tp_number({"ref": "101"}) == "101"
|
||||
|
||||
|
||||
def test_extract_tp_number_ref_priority_over_name() -> None:
|
||||
"""ref приоритетнее name: ref='42', name='ТП-999' → 42."""
|
||||
assert og.extract_tp_number({"ref": "42", "name": "ТП-999"}) == "42"
|
||||
|
||||
|
||||
def test_extract_tp_number_voltage_without_token_is_none() -> None:
|
||||
"""«110 кВ» БЕЗ ТП-токена и без ref → None (это класс напряжения, не номер)."""
|
||||
assert og.extract_tp_number({"name": "Подстанция 110 кВ"}) is None
|
||||
assert og.extract_tp_number({"name": "110 кВ", "voltage": "110000"}) is None
|
||||
|
||||
|
||||
def test_extract_tp_number_empty_tags_is_none() -> None:
|
||||
"""Нет ни ref, ни name → None."""
|
||||
assert og.extract_tp_number({}) is None
|
||||
|
||||
|
||||
def test_extract_tp_number_latin_tp_token() -> None:
|
||||
"""Латинский токен «TP-42» → 42."""
|
||||
assert og.extract_tp_number({"name": "TP-42"}) == "42"
|
||||
|
||||
|
||||
# ── fetch_osm_substations: traction-скип ─────────────────────────────────────
|
||||
|
||||
|
||||
class _FakeResp:
|
||||
def __init__(self, payload: dict) -> None:
|
||||
self._payload = payload
|
||||
|
||||
def raise_for_status(self) -> None:
|
||||
pass
|
||||
|
||||
def json(self) -> dict:
|
||||
return self._payload
|
||||
|
||||
|
||||
def test_fetch_skips_traction(monkeypatch: Any) -> None:
|
||||
"""substation=traction (тяговые ж/д) отсеиваются; обычные ТП остаются."""
|
||||
payload = {
|
||||
"elements": [
|
||||
{"type": "node", "id": 1, "lat": 56.8, "lon": 60.6, "tags": {"name": "ТП-1"}},
|
||||
{
|
||||
"type": "node",
|
||||
"id": 2,
|
||||
"lat": 56.8,
|
||||
"lon": 60.6,
|
||||
"tags": {"substation": "traction", "name": "Тяговая-2"},
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
def _fake_post(url: str, data: Any = None, timeout: Any = None) -> _FakeResp:
|
||||
return _FakeResp(payload)
|
||||
|
||||
monkeypatch.setattr(og.httpx, "post", _fake_post)
|
||||
kept = og.fetch_osm_substations()
|
||||
assert len(kept) == 1
|
||||
assert kept[0]["id"] == 1
|
||||
|
||||
|
||||
# ── _index_osm_by_number: ambiguous ──────────────────────────────────────────
|
||||
|
||||
|
||||
def _node(el_id: int, lat: float, lon: float, tags: dict) -> dict:
|
||||
return {"type": "node", "id": el_id, "lat": lat, "lon": lon, "tags": tags}
|
||||
|
||||
|
||||
def test_index_ambiguous_duplicate_number() -> None:
|
||||
"""Один номер несут 2 фичи с РАЗНЫМИ координатами → ambiguous, из матча выпадает."""
|
||||
elements = [
|
||||
_node(1, 56.80, 60.60, {"name": "ТП-100"}),
|
||||
_node(2, 56.85, 60.65, {"name": "ТП-100"}), # тот же номер, другая точка
|
||||
_node(3, 56.82, 60.62, {"name": "ТП-200"}), # уникальный
|
||||
]
|
||||
by_number, ambiguous = og._index_osm_by_number(elements)
|
||||
assert "100" in ambiguous
|
||||
assert "100" not in by_number
|
||||
assert "200" in by_number
|
||||
|
||||
|
||||
def test_index_same_point_not_ambiguous() -> None:
|
||||
"""Дубль-теги ОДНОЙ точки (way+node на тех же координатах) → НЕ ambiguous."""
|
||||
elements = [
|
||||
_node(1, 56.80, 60.60, {"name": "ТП-100"}),
|
||||
{"type": "way", "id": 2, "center": {"lat": 56.80, "lon": 60.60}, "tags": {"ref": "100"}},
|
||||
]
|
||||
by_number, ambiguous = og._index_osm_by_number(elements)
|
||||
assert "100" not in ambiguous
|
||||
assert by_number["100"] == (56.80, 60.60)
|
||||
|
||||
|
||||
def test_index_out_of_bbox_skipped() -> None:
|
||||
"""Координата вне bbox ЕКБ (sanity-гейт) → не индексируется."""
|
||||
elements = [_node(1, 55.75, 37.61, {"name": "ТП-300"})] # Москва
|
||||
by_number, ambiguous = og._index_osm_by_number(elements)
|
||||
assert by_number == {}
|
||||
assert ambiguous == set()
|
||||
|
||||
|
||||
def test_index_way_center_coords() -> None:
|
||||
"""way без lat/lon, но с center → берём center."""
|
||||
elements = [
|
||||
{"type": "way", "id": 9, "center": {"lat": 56.83, "lon": 60.63}, "tags": {"ref": "555"}}
|
||||
]
|
||||
by_number, _ = og._index_osm_by_number(elements)
|
||||
assert by_number["555"] == (56.83, 60.63)
|
||||
|
||||
|
||||
# ── Fake DB session (execute + begin_nested), как в test_eesk_reserve_loader ──
|
||||
|
||||
|
||||
class _FakeResult:
|
||||
def __init__(self, rows: list[Any] | None = None, rowcount: int = 1) -> None:
|
||||
self._rows = rows or []
|
||||
self.rowcount = rowcount
|
||||
|
||||
def all(self) -> list[Any]:
|
||||
return self._rows
|
||||
|
||||
|
||||
class _FakeSession:
|
||||
"""Мини-заглушка Session: собирает execute-вызовы, поддерживает begin_nested."""
|
||||
|
||||
def __init__(self, candidate_norms: list[str], rowcount: int = 1) -> None:
|
||||
self.calls: list[tuple[str, Any]] = []
|
||||
self._candidates = [(n,) for n in candidate_norms]
|
||||
self._rowcount = rowcount
|
||||
self.committed = False
|
||||
|
||||
@contextmanager
|
||||
def begin_nested(self): # type: ignore[no-untyped-def]
|
||||
yield
|
||||
|
||||
def execute(self, stmt: Any, params: Any = None) -> _FakeResult:
|
||||
sql = str(stmt)
|
||||
self.calls.append((sql, params))
|
||||
if "SELECT" in sql and "power_tp_rp_reserves" in sql:
|
||||
return _FakeResult(rows=self._candidates)
|
||||
return _FakeResult(rowcount=self._rowcount)
|
||||
|
||||
def commit(self) -> None:
|
||||
self.committed = True
|
||||
|
||||
def rollback(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
def _patch_fetch(monkeypatch: Any, elements: list[dict]) -> None:
|
||||
monkeypatch.setattr(og, "fetch_osm_substations", lambda: elements)
|
||||
|
||||
|
||||
# ── geocode_tp_from_osm ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_geocode_dry_run_does_not_write(monkeypatch: Any) -> None:
|
||||
"""Dry-run (apply=False): matched считается, но UPDATE не шлётся и commit не зовётся."""
|
||||
_patch_fetch(monkeypatch, [_node(1, 56.80, 60.60, {"name": "ТП-101"})])
|
||||
db = _FakeSession(candidate_norms=["101"])
|
||||
result = og.geocode_tp_from_osm(db=db, apply=False)
|
||||
|
||||
assert result["matched"] == 1
|
||||
assert result["updated"] == 0
|
||||
assert result["applied"] is False
|
||||
assert not db.committed
|
||||
# Только SELECT-кандидатов, никаких UPDATE.
|
||||
assert all("UPDATE" not in sql for sql, _ in db.calls)
|
||||
|
||||
|
||||
def test_geocode_apply_updates_with_cast(monkeypatch: Any) -> None:
|
||||
"""apply=True: UPDATE шлётся с CAST-формами, geom IS NULL-гейтом, commit зовётся."""
|
||||
_patch_fetch(monkeypatch, [_node(1, 56.80, 60.60, {"name": "ТП-101"})])
|
||||
db = _FakeSession(candidate_norms=["101"], rowcount=1)
|
||||
result = og.geocode_tp_from_osm(db=db, apply=True)
|
||||
|
||||
assert result["matched"] == 1
|
||||
assert result["updated"] == 1
|
||||
assert result["applied"] is True
|
||||
assert db.committed
|
||||
|
||||
update_calls = [(sql, p) for sql, p in db.calls if "UPDATE" in sql]
|
||||
assert len(update_calls) == 1
|
||||
sql, params = update_calls[0]
|
||||
# CAST-формы (psycopg3-канон), НИКОГДА :x::type.
|
||||
assert "CAST(:lon AS float)" in sql
|
||||
assert "CAST(:lat AS float)" in sql
|
||||
assert "CAST(:name_norm AS text)" in sql
|
||||
assert "::float" not in sql and "::text" not in sql
|
||||
# Идемпотентность + скоуп ЕКБ.
|
||||
assert "geom IS NULL" in sql
|
||||
assert "geocode_status = 'osm'" in sql
|
||||
assert "municipality = 'Екатеринбург'" in sql
|
||||
assert params == {"name_norm": "101", "lat": 56.80, "lon": 60.60}
|
||||
|
||||
|
||||
def test_geocode_ambiguous_not_matched(monkeypatch: Any) -> None:
|
||||
"""Дубль-номер (ambiguous) не матчится: candidate '100' есть, но geom не пишется."""
|
||||
_patch_fetch(
|
||||
monkeypatch,
|
||||
[
|
||||
_node(1, 56.80, 60.60, {"name": "ТП-100"}),
|
||||
_node(2, 56.85, 60.65, {"name": "ТП-100"}), # коллизия
|
||||
],
|
||||
)
|
||||
db = _FakeSession(candidate_norms=["100"], rowcount=1)
|
||||
result = og.geocode_tp_from_osm(db=db, apply=True)
|
||||
|
||||
assert result["ambiguous"] == 1
|
||||
assert result["matched"] == 0
|
||||
assert result["updated"] == 0
|
||||
assert all("UPDATE" not in sql for sql, _ in db.calls)
|
||||
|
||||
|
||||
def test_geocode_unmatched_candidate_skipped(monkeypatch: Any) -> None:
|
||||
"""Кандидат без OSM-номера просто скипается (matched=0), без падения."""
|
||||
_patch_fetch(monkeypatch, [_node(1, 56.80, 60.60, {"name": "ТП-777"})])
|
||||
db = _FakeSession(candidate_norms=["101"], rowcount=1) # 101 нет в OSM
|
||||
result = og.geocode_tp_from_osm(db=db, apply=True)
|
||||
assert result["matched"] == 0
|
||||
assert result["updated"] == 0
|
||||
|
||||
|
||||
# ── Celery / beat регистрация ────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_celery_task_registered() -> None:
|
||||
"""Таск sync_osm_tp_geocode регистрируется при импорте модуля тасков.
|
||||
|
||||
Модуль в celery_app.include — на worker-старте импортируется и таск попадает в
|
||||
celery_app.tasks. В тесте импорт делаем явно (Celery не бутит воркер).
|
||||
"""
|
||||
import app.workers.tasks.connection_capacity_sync # noqa: F401 — импорт регистрирует таск
|
||||
from app.workers.celery_app import celery_app
|
||||
|
||||
assert "tasks.connection_capacity_sync.sync_osm_tp_geocode" in celery_app.tasks
|
||||
|
||||
|
||||
def test_beat_schedule_entry() -> None:
|
||||
"""Beat содержит еженедельную вторник-05:45 запись osm-tp-geocode."""
|
||||
from app.workers.beat_schedule import build_beat_schedule
|
||||
|
||||
schedule = build_beat_schedule()
|
||||
assert "osm-tp-geocode-weekly" in schedule
|
||||
entry = schedule["osm-tp-geocode-weekly"]
|
||||
assert entry["task"] == "tasks.connection_capacity_sync.sync_osm_tp_geocode"
|
||||
Loading…
Add table
Reference in a new issue