bot-backend
515ac89eaa
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
397 lines
17 KiB
Python
397 lines
17 KiB
Python
"""Celery task: загрузка подзон ПАТ Кольцово из JSON-файла координат (#1065 / #1067).
|
||
|
||
Фаза-2 пайплайна: Фаза-1 (извлечение координат из Приказа 367-П) уже выполнена —
|
||
файл data/pat/koltsovo_367p_subzones.json содержит точки в МСК-66 зона 1.
|
||
|
||
Алгоритм:
|
||
1. Читает JSON из data/pat/koltsovo_367p_subzones.json (путь относительно repo root).
|
||
2. Для каждой подзоны собирает секторы: внешние → отдельные Polygon-ы,
|
||
внутренние (label содержит «внутренн») → дырки в предшествующем внешнем Polygon-е.
|
||
3. Репроецирует МСК-66 → WGS84 через _MSK66_TO_WGS84 (always_xy=True):
|
||
point=[X, Y] где X=northing, Y=easting → transform(easting=Y, northing=X).
|
||
4. shapely.make_valid, union → MultiPolygon, UPSERT в pat_subzones.
|
||
5. Логирует ST_IsValid, bbox-центроид (проверка «в районе Кольцово»),
|
||
сравнение площади union подзон vs aggregate_outline.
|
||
|
||
Beat не нужен — статика, задача запускается вручную: один раз после деплоя.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import pathlib
|
||
from typing import Any
|
||
|
||
from pyproj import CRS, Transformer
|
||
from shapely import make_valid
|
||
from shapely.geometry import MultiPolygon, Polygon
|
||
from shapely.ops import unary_union
|
||
from sqlalchemy import text
|
||
|
||
from app.workers.celery_app import celery_app
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── МСК-66 → WGS84 (те же параметры что в ekburg_permits.py) ────────────────
|
||
_MSK66_PROJ4 = (
|
||
"+proj=tmerc +lat_0=0 +lon_0=60 +k=1"
|
||
" +x_0=1496944 +y_0=-5911077"
|
||
" +ellps=krass +towgs84=23.92,-141.27,-80.9,0,0.35,0.82,-0.12"
|
||
" +units=m +no_defs"
|
||
)
|
||
_MSK66_TO_WGS84: Transformer = Transformer.from_crs(
|
||
CRS.from_proj4(_MSK66_PROJ4),
|
||
CRS.from_epsg(4326),
|
||
always_xy=True,
|
||
)
|
||
|
||
# bbox Кольцово (ЮВ от ЕКБ): lon 60.5..61.2, lat 56.5..57.0
|
||
_KOLTSOVO_LON_MIN = 60.5
|
||
_KOLTSOVO_LON_MAX = 61.2
|
||
_KOLTSOVO_LAT_MIN = 56.5
|
||
_KOLTSOVO_LAT_MAX = 57.0
|
||
|
||
# Путь к JSON — робастный резолв: контейнер (/app/data/...) и dev-среда (repo/data/...)
|
||
# В контейнере: WORKDIR=/app, код в /app/app/..., данные в /app/data/...
|
||
# parents[3] от /app/app/workers/tasks/ = /app → /app/data/pat/ ✓
|
||
# В dev: код в <repo>/backend/app/workers/tasks/, данные в <repo>/data/...
|
||
# parents[4] от <repo>/backend/app/workers/tasks/ = <repo> → <repo>/data/pat/ ✓
|
||
# Опция: PAT_JSON_PATH env-переменная переопределяет всё.
|
||
_PAT_JSON_RELATIVE = pathlib.Path("data") / "pat" / "koltsovo_367p_subzones.json"
|
||
_THIS = pathlib.Path(__file__).resolve()
|
||
_JSON_PATH_CANDIDATES: list[pathlib.Path] = []
|
||
|
||
if _env_path := os.environ.get("PAT_JSON_PATH"):
|
||
_JSON_PATH_CANDIDATES.append(pathlib.Path(_env_path))
|
||
# parents[3] — контейнерный корень /app
|
||
_JSON_PATH_CANDIDATES.append(_THIS.parents[3] / _PAT_JSON_RELATIVE)
|
||
# parents[4] — repo root в dev
|
||
_JSON_PATH_CANDIDATES.append(_THIS.parents[4] / _PAT_JSON_RELATIVE)
|
||
|
||
|
||
def _resolve_json_path() -> pathlib.Path:
|
||
"""Возвращает первый существующий кандидат JSON, иначе первый (для понятного сообщения)."""
|
||
for candidate in _JSON_PATH_CANDIDATES:
|
||
if candidate.exists():
|
||
logger.debug("pat_subzones_load: JSON путь → %s", candidate)
|
||
return candidate
|
||
# Файл не найден ни по одному кандидату — вернём первый (контейнерный) для диагностики
|
||
primary = _JSON_PATH_CANDIDATES[0] if _JSON_PATH_CANDIDATES else _THIS / "MISSING"
|
||
logger.warning(
|
||
"pat_subzones_load: JSON не найден ни по одному кандидату: %s",
|
||
[str(c) for c in _JSON_PATH_CANDIDATES],
|
||
)
|
||
return primary
|
||
|
||
|
||
_SOURCE = "Приказ Росавиации № 367-П от 02.06.2023 (ред. Приказа Росавиации от 27.10.2025 № 804-П)"
|
||
|
||
# Ключевые слова для определения внутренних (hole) секторов
|
||
_HOLE_KEYWORDS = ("внутренн", "internal", "исключ")
|
||
|
||
|
||
def _is_hole_sector(label: str | None) -> bool:
|
||
"""Возвращает True если сектор является внутренней границей (дыркой)."""
|
||
if not label:
|
||
return False
|
||
lbl = label.lower()
|
||
return any(kw in lbl for kw in _HOLE_KEYWORDS)
|
||
|
||
|
||
def _reproject_ring(points: list[list[float]]) -> list[tuple[float, float]]:
|
||
"""МСК-66 [X=northing, Y=easting] → (lon, lat) WGS84 для каждой точки.
|
||
|
||
always_xy=True → transform принимает (x=easting, y=northing).
|
||
"""
|
||
coords: list[tuple[float, float]] = []
|
||
for pt in points:
|
||
northing = pt[0]
|
||
easting = pt[1]
|
||
lon, lat = _MSK66_TO_WGS84.transform(easting, northing)
|
||
coords.append((lon, lat))
|
||
return coords
|
||
|
||
|
||
def _close_ring(coords: list[tuple[float, float]]) -> list[tuple[float, float]]:
|
||
"""Замкнуть кольцо если первая и последняя точки не совпадают."""
|
||
if coords and coords[0] != coords[-1]:
|
||
coords = [*coords, coords[0]]
|
||
return coords
|
||
|
||
|
||
def _build_subzone_geometry(sectors: list[dict[str, Any]]) -> MultiPolygon:
|
||
"""Строит MultiPolygon подзоны из списка секторов.
|
||
|
||
Алгоритм обхода:
|
||
- Встретили внешний сектор → начинаем новый Polygon (exterior).
|
||
- Встретили внутренний сектор → добавляем hole к последнему внешнему Polygon-у.
|
||
- Если нет ни одного внешнего — все секторы трактуются как exterior.
|
||
|
||
Returns MultiPolygon (валидный через make_valid + unary_union).
|
||
"""
|
||
polys: list[Polygon] = []
|
||
current_exterior: list[tuple[float, float]] | None = None
|
||
current_holes: list[list[tuple[float, float]]] = []
|
||
|
||
def flush_current() -> None:
|
||
"""Сохранить накопленный exterior+holes как Polygon."""
|
||
nonlocal current_exterior, current_holes
|
||
if current_exterior is None:
|
||
return
|
||
ext = _close_ring(current_exterior)
|
||
holes = [_close_ring(h) for h in current_holes]
|
||
poly = Polygon(ext, holes)
|
||
poly = make_valid(poly)
|
||
if not poly.is_empty:
|
||
polys.append(poly)
|
||
current_exterior = None
|
||
current_holes = []
|
||
|
||
for sector in sectors:
|
||
raw_pts: list[list[float]] = sector.get("points", [])
|
||
if not raw_pts:
|
||
logger.warning("pat_subzones_load: сектор '%s' без точек, пропуск", sector.get("label"))
|
||
continue
|
||
|
||
coords = _reproject_ring(raw_pts)
|
||
label = sector.get("label")
|
||
|
||
if _is_hole_sector(label):
|
||
# Внутренняя граница — дырка в текущем внешнем контуре
|
||
if current_exterior is None:
|
||
logger.warning(
|
||
"pat_subzones_load: внутренний сектор '%s' без предшествующего внешнего"
|
||
" — трактуем как exterior",
|
||
label,
|
||
)
|
||
current_exterior = coords
|
||
else:
|
||
current_holes.append(coords)
|
||
else:
|
||
# Новый внешний контур → flush предыдущий
|
||
flush_current()
|
||
current_exterior = coords
|
||
|
||
flush_current()
|
||
|
||
if not polys:
|
||
logger.error("pat_subzones_load: нет валидных полигонов в секторах")
|
||
return MultiPolygon()
|
||
|
||
merged = unary_union(polys)
|
||
merged = make_valid(merged)
|
||
|
||
if merged.geom_type == "Polygon":
|
||
merged = MultiPolygon([merged])
|
||
elif merged.geom_type == "MultiPolygon":
|
||
pass
|
||
else:
|
||
# GeometryCollection или иное — пробуем извлечь полигоны
|
||
sub_polys = [g for g in merged.geoms if g.geom_type in ("Polygon", "MultiPolygon")]
|
||
if sub_polys:
|
||
merged = make_valid(unary_union(sub_polys))
|
||
if merged.geom_type == "Polygon":
|
||
merged = MultiPolygon([merged])
|
||
|
||
return merged # type: ignore[return-value]
|
||
|
||
|
||
def _centroid_in_koltsovo(geom: MultiPolygon) -> bool:
|
||
"""Проверяет что центроид геометрии находится в bbox Кольцово."""
|
||
if geom.is_empty:
|
||
return False
|
||
c = geom.centroid
|
||
return (
|
||
_KOLTSOVO_LON_MIN <= c.x <= _KOLTSOVO_LON_MAX
|
||
and _KOLTSOVO_LAT_MIN <= c.y <= _KOLTSOVO_LAT_MAX
|
||
)
|
||
|
||
|
||
@celery_app.task(name="tasks.pat_subzones_load.load_koltsovo_pat")
|
||
def load_koltsovo_pat() -> dict[str, Any]:
|
||
"""Загрузить подзоны ПАТ Кольцово из JSON в таблицу pat_subzones.
|
||
|
||
Читает data/pat/koltsovo_367p_subzones.json, строит MultiPolygon для каждой
|
||
подзоны, UPSERT через SAVEPOINT. Возвращает dict со статистикой.
|
||
"""
|
||
from app.core.db import SessionLocal
|
||
|
||
json_path = _resolve_json_path()
|
||
logger.info("pat_subzones_load: start, json=%s", json_path)
|
||
|
||
if not json_path.exists():
|
||
candidates_str = ", ".join(str(c) for c in _JSON_PATH_CANDIDATES)
|
||
raise FileNotFoundError(f"JSON не найден: {json_path}. Кандидаты: {candidates_str}")
|
||
|
||
with open(json_path, encoding="utf-8") as fh:
|
||
data: dict[str, Any] = json.load(fh)
|
||
|
||
aerodrome: str = data.get("aerodrome", "Екатеринбург (Кольцово)")
|
||
subzones_raw: dict[str, Any] = data.get("subzones", {})
|
||
aggregate_pts: list[list[float]] = data.get("aggregate_outline", {}).get("points", [])
|
||
|
||
# ── Строим aggregate_outline geometry (subzone_no=0) ─────────────────────
|
||
aggregate_geom: MultiPolygon | None = None
|
||
if aggregate_pts:
|
||
coords = _reproject_ring(aggregate_pts)
|
||
coords = _close_ring(coords)
|
||
agg_poly = make_valid(Polygon(coords))
|
||
aggregate_geom = MultiPolygon([agg_poly]) if agg_poly.geom_type == "Polygon" else None
|
||
if aggregate_geom:
|
||
logger.info(
|
||
"pat_subzones_load: aggregate_outline area=%.4f deg² centroid_ok=%s",
|
||
aggregate_geom.area,
|
||
_centroid_in_koltsovo(aggregate_geom),
|
||
)
|
||
|
||
results: list[dict[str, Any]] = []
|
||
total_area = 0.0
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
for sz_no_str, sz_data in subzones_raw.items():
|
||
sz_no = int(sz_no_str)
|
||
name: str = sz_data.get("name", f"Подзона {sz_no}")
|
||
restriction: str = sz_data.get("restriction", "")
|
||
sectors: list[dict[str, Any]] = sz_data.get("sectors", [])
|
||
|
||
geom = _build_subzone_geometry(sectors)
|
||
is_valid = not geom.is_empty and geom.is_valid
|
||
centroid_ok = _centroid_in_koltsovo(geom)
|
||
|
||
logger.info(
|
||
"pat_subzones_load: subzone=%d name='%s' sectors=%d "
|
||
"is_valid=%s centroid_ok=%s area=%.6f",
|
||
sz_no,
|
||
name,
|
||
len(sectors),
|
||
is_valid,
|
||
centroid_ok,
|
||
geom.area,
|
||
)
|
||
|
||
if not centroid_ok:
|
||
logger.warning(
|
||
"pat_subzones_load: subzone=%d centroid ВНЕ Кольцово — "
|
||
"возможна ошибка осей/проекции. centroid=(%.5f, %.5f)",
|
||
sz_no,
|
||
geom.centroid.x if not geom.is_empty else 0,
|
||
geom.centroid.y if not geom.is_empty else 0,
|
||
)
|
||
|
||
total_area += geom.area
|
||
geom_wkt = geom.wkt
|
||
|
||
try:
|
||
db.execute(text("SAVEPOINT sp_subzone"))
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO pat_subzones
|
||
(aerodrome, subzone_no, name, restriction, geom, source)
|
||
VALUES (
|
||
CAST(:aerodrome AS text),
|
||
CAST(:subzone_no AS integer),
|
||
CAST(:name AS text),
|
||
CAST(:restriction AS text),
|
||
ST_Multi(ST_GeomFromText(CAST(:geom_wkt AS text), 4326)),
|
||
CAST(:source AS text)
|
||
)
|
||
ON CONFLICT (aerodrome, subzone_no) DO UPDATE SET
|
||
name = EXCLUDED.name,
|
||
restriction = EXCLUDED.restriction,
|
||
geom = EXCLUDED.geom,
|
||
source = EXCLUDED.source,
|
||
fetched_at = now()
|
||
"""
|
||
),
|
||
{
|
||
"aerodrome": aerodrome,
|
||
"subzone_no": sz_no,
|
||
"name": name,
|
||
"restriction": restriction,
|
||
"geom_wkt": geom_wkt,
|
||
"source": _SOURCE,
|
||
},
|
||
)
|
||
db.execute(text("RELEASE SAVEPOINT sp_subzone"))
|
||
results.append(
|
||
{"subzone_no": sz_no, "is_valid": is_valid, "centroid_ok": centroid_ok}
|
||
)
|
||
except Exception as exc:
|
||
db.execute(text("ROLLBACK TO SAVEPOINT sp_subzone"))
|
||
logger.error("pat_subzones_load: subzone=%d UPSERT failed: %s", sz_no, exc)
|
||
raise
|
||
|
||
# ── UPSERT aggregate_outline (subzone_no=0) ──────────────────────────
|
||
if aggregate_geom is not None:
|
||
try:
|
||
db.execute(text("SAVEPOINT sp_aggregate"))
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO pat_subzones
|
||
(aerodrome, subzone_no, name, restriction, geom, source)
|
||
VALUES (
|
||
CAST(:aerodrome AS text),
|
||
0,
|
||
'Общий контур ПАТ',
|
||
'',
|
||
ST_Multi(ST_GeomFromText(CAST(:geom_wkt AS text), 4326)),
|
||
CAST(:source AS text)
|
||
)
|
||
ON CONFLICT (aerodrome, subzone_no) DO UPDATE SET
|
||
geom = EXCLUDED.geom,
|
||
source = EXCLUDED.source,
|
||
fetched_at = now()
|
||
"""
|
||
),
|
||
{
|
||
"aerodrome": aerodrome,
|
||
"geom_wkt": aggregate_geom.wkt,
|
||
"source": _SOURCE,
|
||
},
|
||
)
|
||
db.execute(text("RELEASE SAVEPOINT sp_aggregate"))
|
||
logger.info("pat_subzones_load: aggregate_outline upserted (subzone_no=0)")
|
||
except Exception as exc:
|
||
db.execute(text("ROLLBACK TO SAVEPOINT sp_aggregate"))
|
||
logger.error("pat_subzones_load: aggregate_outline UPSERT failed: %s", exc)
|
||
raise
|
||
|
||
db.commit()
|
||
except Exception:
|
||
db.rollback()
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
# ── Валидация: суммарная площадь подзон vs aggregate_outline ─────────────
|
||
if aggregate_geom is not None and aggregate_geom.area > 0:
|
||
ratio = total_area / aggregate_geom.area
|
||
if ratio < 0.5 or ratio > 5.0:
|
||
logger.warning(
|
||
"pat_subzones_load: суммарная площадь подзон (%.4f) расходится "
|
||
"с aggregate_outline (%.4f) в %.1fx — возможная ошибка осей/проекции",
|
||
total_area,
|
||
aggregate_geom.area,
|
||
ratio,
|
||
)
|
||
else:
|
||
logger.info(
|
||
"pat_subzones_load: total_area=%.4f aggregate_area=%.4f ratio=%.2fx OK",
|
||
total_area,
|
||
aggregate_geom.area,
|
||
ratio,
|
||
)
|
||
|
||
summary = {
|
||
"aerodrome": aerodrome,
|
||
"subzones_loaded": len(results),
|
||
"results": results,
|
||
"total_area_deg2": round(total_area, 6),
|
||
}
|
||
logger.info("pat_subzones_load: done %s", summary)
|
||
return summary
|