yazmolod / pynspd / #10

Committed 30 Apr 2025 09:58AM UTC coverage: 98.461% (+0.4%) from 98.083%

Build # #10

Build Type

push

coveralls-python

Committed by

yazmolod

Commit Message

feat (cli): add coods list support

Run Details

22 of 22 new or added lines in 2 files covered. (100.0%)

18 existing lines in 5 files now uncovered.

2559 of 2599 relevant lines covered (98.46%)

0.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.77

/src/pynspd/_sync/api.py

import json
import re
from functools import wraps
from time import sleep
from typing import Any, Literal, Optional, Type, Union

import mercantile
import numpy as np
from hishel import BaseStorage, CacheTransport
from httpx import (
    BaseTransport,
    Client,
    HTTPStatusError,
    HTTPTransport,
    RemoteProtocolError,
    Response,
    TimeoutException,
)
from httpx._types import ProxyTypes, QueryParamTypes
from shapely import MultiPolygon, Point, Polygon, to_geojson

from pynspd.client import (
    NSPD_CACHE_CONTROLLER,
    SSL_CONTEXT,
    BaseNspdClient,
    get_client_args,
)
from pynspd.logger import logger
from pynspd.map_types.enums import TabTitle, ThemeId
from pynspd.schemas import Layer36048Feature, Layer36049Feature, NspdFeature
from pynspd.schemas.feature import Feat
from pynspd.schemas.responses import (
    NspdTabGroupResponse,
    NspdTabResponse,
    SearchResponse,
)


def retry_on_http_error(func):
    """Декоратор для повторения запроса при ошибках запроса"""

    @wraps(func)
    def wrapper(self: "Nspd", *args, **kwargs):
        attempt = 0
        while attempt <= self.retries:
            logger_suffix = f'Wrapped method "{func.__name__}" -'
            try:
                logger.debug("%s start request", logger_suffix)
                return func(self, *args, **kwargs)
            except (TimeoutException, HTTPStatusError) as e:
                if isinstance(e, HTTPStatusError):
                    if e.response.status_code == 429:
                        logger.debug("%s too many requests", logger_suffix)
                    elif e.response.status_code < 500:
                        logger.debug("%s not server error", logger_suffix)
                        raise e
                attempt += 1
                if attempt > self.retries:
                    logger.debug("%s run out attempts", logger_suffix)
                    raise e
                sleep(1)
                logger.debug("%s attempt %d/%d", logger_suffix, attempt, self.retries)
            except RemoteProtocolError:
                # Запрос иногда рандомно обрывается сервером, проходит при повторном запросе
                logger.debug("%s server disconnect", logger_suffix)

    return wrapper


class Nspd(BaseNspdClient):
    """Клиент для НСПД

    Example:
    ```python
    with pynspd.Nspd() as nspd:
        feat = nspd.find_zu("77:05:0001005:19")
    ```

    Args:
        timeout:
            Время ожидания ответа.
            Если не установлен - есть вероятность бесконечного ожидания. По умолчанию None.
        retries:
            Количество попыток при неудачном запросе
            (таймаут, неожиданный обрыв соединения, 5хх ошибки). По умолчанию 10.
        proxy:
            Использовать прокси для запросов. По умолчанию None.
        cache_storage:
            Настройка хранения кэша (см. https://hishel.com/advanced/storages/).
            Если установлен, то при повторном запросе результат будет
            извлекаться из хранилища кэша, что сильно увеличивает производительность
            и снижает риск ошибки 429 - Too many requests. По умолчанию None.
    """

    def __init__(
        self,
        *,
        timeout: Optional[int] = None,
        retries: int = 10,
        proxy: Optional[ProxyTypes] = None,
        cache_storage: Optional[BaseStorage] = None,
    ):
        super().__init__(retries=retries)
        self._client = self._build_client(
            timeout=timeout,
            retries=retries,
            proxy=proxy,
            cache_storage=cache_storage,
        )

    @staticmethod
    def _build_client(
        timeout: Optional[int],
        retries: int,
        proxy: Optional[ProxyTypes],
        cache_storage: Optional[BaseStorage],
    ) -> Client:
        client_args = get_client_args(timeout)
        transport: BaseTransport = HTTPTransport(
            verify=SSL_CONTEXT, retries=retries, proxy=proxy
        )
        if cache_storage is not None:
            transport = CacheTransport(
                transport=transport,
                storage=cache_storage,
                controller=NSPD_CACHE_CONTROLLER,
            )
        return Client(**client_args, transport=transport)

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.close()

    def close(self):
        """Завершение сессии"""
        self._client.close()

    def request(
        self,
        method: str,
        url: str,
        params: Optional[QueryParamTypes] = None,
        json: Optional[dict] = None,
    ) -> Response:
        """Базовый запрос к API НСПД"""
        logger.debug("Request %s", url)
        r = self._client.request(method, url, params=params, json=json)
        r.raise_for_status()
        return r

    @retry_on_http_error
    def safe_request(
        self,
        method: str,
        url: str,
        params: Optional[QueryParamTypes] = None,
        json: Optional[dict] = None,
    ) -> Response:
        """Базовый запрос к api НСПД с обработкой ошибок"""
        return self.request(method, url, params, json)

    ####################
    ### QUERY SEARCH ###
    ####################

    @retry_on_http_error
    def _search(self, params: dict[str, Any]) -> Optional[list[NspdFeature]]:
        """Базовый поисковый запрос на НСПД"""
        try:
            r = self.request("get", "/api/geoportal/v2/search/geoportal", params=params)
            return SearchResponse.model_validate(r.json()).data.features
        except HTTPStatusError as e:
            if e.response.status_code == 404:
                return None
            raise e

    def search(
        self, query: str, theme_id: ThemeId = ThemeId.REAL_ESTATE_OBJECTS
    ) -> Optional[list[NspdFeature]]:
        """Поисковой запрос по предустановленной теме

        Args:
            query: Поисковой запрос
            theme_id:
                Вид объекта (кадастровое деление, объект недвижимости и т.д.).
                По умолчанию: объекты недвижимости

        Returns:
            Положительный ответ от сервиса, либо None, если ничего не найдено
        """
        return self._search(
            params={
                "query": query,
                "thematicSearchId": theme_id.value,
            }
        )

    def search_in_layer(
        self, query: str, layer_def: Type[Feat]
    ) -> Optional[list[Feat]]:
        """Поиск объекта по определению слоя

        Args:
            query: Поисковой запрос
            layer_def: Определение слоя

        Returns:
            Валидированная модель слоя, если найдено
        """
        raw_features = self._search(
            params={
                "query": query,
                "layersId": layer_def.layer_meta.layer_id,
            }
        )
        return self._cast_features_to_layer_defs(raw_features, layer_def)

    def find(
        self, query: str, theme_id: ThemeId = ThemeId.REAL_ESTATE_OBJECTS
    ) -> Optional[NspdFeature]:
        """Найти объект по предустановленной теме

        Args:
            query: Поисковой запрос
            theme_id:
                Вид объекта (кадастровое деление, объект недвижимости и т.д.).
                По умолчанию: объекты недвижимости

        Returns:
            Положительный ответ от сервиса, либо None, если ничего не найдено
        """
        return self._filter_search_by_query(self.search(query, theme_id), query)

    def find_in_layer(self, query: str, layer_def: Type[Feat]) -> Optional[Feat]:
        """Найти объект по определению слоя

        Args:
            query: Поисковой запрос
            layer_def: Определение слоя

        Returns:
            Валидированная модель слоя, если найдено
        """
        return self._filter_search_by_query(
            self.search_in_layer(query, layer_def), query
        )

    ######################
    ### POLYGON SEARCH ###
    ######################

    @retry_on_http_error
    def _search_in_contour(
        self,
        countour: Union[Polygon, MultiPolygon],
        *category_ids: int,
    ) -> Optional[list[NspdFeature]]:
        """Поиск объектов в контуре по ID категорий слоев

        Args:
            countour: Геометрический объект с контуром
            category_ids: ID категорий слоев

        Returns:
            Список объектов, пересекающихся с контуром, если найден хоть один
        """
        feature_geojson = json.loads(to_geojson(countour))
        feature_geojson["crs"] = {
            "type": "name",
            "properties": {"name": "EPSG:4326"},
        }
        payload = {
            "categories": [{"id": id_} for id_ in category_ids],
            "geom": {
                "type": "FeatureCollection",
                "features": [
                    {"geometry": feature_geojson, "type": "Feature", "properties": {}}
                ],
            },
        }
        response = self.request(
            "post",
            "/api/geoportal/v1/intersects",
            params={"typeIntersect": "fullObject"},
            json=payload,
        )
        return self._validate_feature_collection_response(response)

    def search_in_contour(
        self,
        countour: Union[Polygon, MultiPolygon],
        layer_def: Type[Feat],
    ) -> Optional[list[Feat]]:
        """Поиск объектов слоя в контуре

        Args:
            countour: Геометрический объект с контуром
            layer_def: Модель слоя

        Returns:
            Список объектов, пересекающихся с контуром, если найден хоть один
        """
        raw_features = self._search_in_contour(
            countour, layer_def.layer_meta.category_id
        )
        return self._cast_features_to_layer_defs(raw_features, layer_def)

    ####################
    ### POINT SEARCH ###
    ####################

    @retry_on_http_error
    def _search_at_point(self, pt: Point, layer_id: int) -> Optional[list[NspdFeature]]:
        """Поиск объектов слоя в точке"""
        tile_size = 512
        tile = mercantile.tile(
            pt.x, pt.y, zoom=24
        )  # zoom=24 должно быть достаточно для самого точного совпадения
        tile_bounds = mercantile.bounds(tile)
        i = np.interp(pt.x, [tile_bounds.west, tile_bounds.east], [0, tile_size])
        j = np.interp(pt.y, [tile_bounds.south, tile_bounds.north], [0, tile_size])
        bbox = ",".join(map(str, tile_bounds))
        params = {
            "REQUEST": "GetFeatureInfo",
            "SERVICE": "WMS",
            "VERSION": "1.3.0",
            "INFO_FORMAT": "application/json",
            "FORMAT": "image/png",
            "STYLES": "",
            "TRANSPARENT": "true",
            "QUERY_LAYERS": layer_id,
            "LAYERS": layer_id,
            "WIDTH": tile_size,
            "HEIGHT": tile_size,
            "I": int(i),
            "J": tile_size - int(j),  # отсчет координат для пикселей ведется сверху
            "CRS": "EPSG:4326",  # CRS для bbox
            "BBOX": bbox,
            "FEATURE_COUNT": "10",  # Иначе вернет только один, даже если попало на границу
        }
        response = self.request(
            "get",
            f"/api/aeggis/v3/{layer_id}/wms",
            params=params,  # type: ignore[arg-type]
        )
        return self._validate_feature_collection_response(response)

    def search_at_point(self, pt: Point, layer_def: Type[Feat]) -> Optional[list[Feat]]:
        """Поиск объектов слоя в точке (с типизацией)

        Args:
            pt: Точка поиска
            layer_def: Тип слоя

        Returns:
            Типизированный список объектов, если найдены
        """
        raw_features = self._search_at_point(pt, layer_def.layer_meta.layer_id)
        return self._cast_features_to_layer_defs(raw_features, layer_def)

    def search_at_coords(
        self, lat: float, lng: float, layer_def: Type[Feat]
    ) -> Optional[list[Feat]]:
        """Поиск объектов слоя в координатах

        Args:
            lat: Широта
            lng: Долгота
            layer_def: Тип слоя

        Returns:
            Типизированный список объектов, если найдены
        """
        return self.search_at_point(Point(lng, lat), layer_def)

    ####################
    ### TAB REQUESTS ###
    ####################

    @retry_on_http_error
    def _tab_request(
        self, feat: NspdFeature, tab_class: str, type_: Literal["values", "group"]
    ) -> Optional[dict]:
        if feat.properties.options.no_coords:
            params = {
                "tabClass": tab_class,
                "objdocId": feat.properties.options.objdoc_id,
                "registersId": feat.properties.options.registers_id,
            }
        else:
            params = {
                "tabClass": tab_class,
                "categoryId": feat.properties.category,
                "geomId": feat.id,
            }
        try:
            r = self.request(
                "get", f"/api/geoportal/v1/tab-{type_}-data", params=params
            )
            return r.json()
        except HTTPStatusError as e:
            if e.response.status_code == 404:
                return None
            raise e

    def _tab_values_request(
        self, feat: NspdFeature, tab_class: str
    ) -> Optional[list[str]]:
        resp = self._tab_request(feat, tab_class, "values")
        if resp is None:
            return None
        return NspdTabResponse.model_validate(resp).value

    def _tab_groups_request(
        self, feat: NspdFeature, tab_class: str
    ) -> Optional[dict[str, Optional[list[str]]]]:
        resp = self._tab_request(feat, tab_class, "group")
        if resp is None:
            return None
        item = NspdTabGroupResponse.model_validate(resp).object
        data = {}
        for i in item:
            title = re.sub(r"[\s:]+$", "", i.title)
            data[title] = i.value
        if len(data) == 0:
            return None
        return data

    def tab_land_parts(self, feat: NspdFeature) -> Optional[list[str]]:
        """Получение данных с вкладки \"Части ЗУ\" """
        return self._tab_values_request(feat, "landParts")

    def tab_land_links(self, feat: NspdFeature) -> Optional[list[str]]:
        """Получение данных с вкладки \"Связанные ЗУ\" """
        return self._tab_values_request(feat, "landLinks")

    def tab_permission_type(self, feat: NspdFeature) -> Optional[list[str]]:
        """Получение данных с вкладки \"Виды разрешенного использования\" """
        return self._tab_values_request(feat, "permissionType")

    def tab_composition_land(self, feat: NspdFeature) -> Optional[list[str]]:
        """Получение данных с вкладки \"Состав ЕЗП\" """
        return self._tab_values_request(feat, "compositionLand")

    def tab_build_parts(self, feat: NspdFeature) -> Optional[list[str]]:
        """Получение данных с вкладки \"Части ОКС\" """
        return self._tab_values_request(feat, "buildParts")

    def tab_objects_list(
        self, feat: NspdFeature
    ) -> Optional[dict[str, Optional[list[str]]]]:
        """
        Получение данных с вкладки \"Объекты\"
        """
        return self._tab_groups_request(feat, "objectsList")

    def get_tab_data(self, feat: NspdFeature, tab_name: TabTitle):
        """Получение данных с указанной вкладки"""
        match tab_name:
            case "Части ЗУ":
                return self.tab_land_parts(feat)
            case "Связанные ЗУ":
                return self.tab_land_links(feat)
            case "Виды разрешенного использования":
                return self.tab_permission_type(feat)
            case "Состав ЕЗП":
                return self.tab_composition_land(feat)
            case "Части ОКС":
                return self.tab_build_parts(feat)
            case "Объекты":
                return self.tab_objects_list(feat)

    #################
    ### SHORTCUTS ###
    #################

    def find_landplot(self, query: str) -> Optional[Layer36048Feature]:
        """Найти ЗУ по кадастровому номеру"""
        return self._filter_search_by_query(self.search_landplots(query), query)

    def find_building(self, query: str) -> Optional[Layer36049Feature]:
        """Найти ОКС по кадастровому номеру"""
        return self._filter_search_by_query(self.search_buildings(query), query)

    def search_landplots_at_point(self, pt: Point) -> Optional[list[Layer36048Feature]]:
        """Поиск ЗУ в точке"""
        return self.search_at_point(pt, Layer36048Feature)

    def search_buildings_at_point(self, pt: Point) -> Optional[list[Layer36049Feature]]:
        """Поиск ОКС в точке"""
        return self.search_at_point(pt, Layer36049Feature)

    def search_landplots(self, cn: str) -> Optional[list[Layer36048Feature]]:
        """Поиск ЗУ по кадастровому номеру"""
        return self.search_in_layer(cn, Layer36048Feature)

    def search_buildings(self, cn: str) -> Optional[list[Layer36049Feature]]:
        """Поиск ОКС по кадастровому номеру"""
        return self.search_in_layer(cn, Layer36049Feature)

    def search_landplots_at_coords(
        self, lat: float, lng: float
    ) -> Optional[list[Layer36048Feature]]:
        """Поиск ЗУ в координатах"""
        return self.search_at_coords(lat, lng, Layer36048Feature)

    def search_buildings_at_coords(
        self, lat: float, lng: float
    ) -> Optional[list[Layer36049Feature]]:
        """Поиск ОКС в координатах"""
        return self.search_at_coords(lat, lng, Layer36049Feature)

    def search_landplots_in_contour(
        self, countour: Union[Polygon, MultiPolygon]
    ) -> Optional[list[Layer36048Feature]]:
        """Поиск ЗУ в контуре"""
        return self.search_in_contour(countour, Layer36048Feature)

    def search_buildings_in_contour(
        self, countour: Union[Polygon, MultiPolygon]
    ) -> Optional[list[Layer36049Feature]]:
        """Поиск ОКС в контуре"""
        return self.search_in_contour(countour, Layer36049Feature)

1	import json	1✔
2	import re	1✔
3	from functools import wraps	1✔
4	from time import sleep	1✔
5	from typing import Any, Literal, Optional, Type, Union	1✔
6
7	import mercantile	1✔
8	import numpy as np	1✔
9	from hishel import BaseStorage, CacheTransport	1✔
10	from httpx import (	1✔
11	BaseTransport,
12	Client,
13	HTTPStatusError,
14	HTTPTransport,
15	RemoteProtocolError,
16	Response,
17	TimeoutException,
18	)
19	from httpx._types import ProxyTypes, QueryParamTypes	1✔
20	from shapely import MultiPolygon, Point, Polygon, to_geojson	1✔
21
22	from pynspd.client import (	1✔
23	NSPD_CACHE_CONTROLLER,
24	SSL_CONTEXT,
25	BaseNspdClient,
26	get_client_args,
27	)
28	from pynspd.logger import logger	1✔
29	from pynspd.map_types.enums import TabTitle, ThemeId	1✔
30	from pynspd.schemas import Layer36048Feature, Layer36049Feature, NspdFeature	1✔
31	from pynspd.schemas.feature import Feat	1✔
32	from pynspd.schemas.responses import (	1✔
33	NspdTabGroupResponse,
34	NspdTabResponse,
35	SearchResponse,
36	)
37
38
39	def retry_on_http_error(func):	1✔
40	"""Декоратор для повторения запроса при ошибках запроса"""
41
42	@wraps(func)	1✔
43	def wrapper(self: "Nspd", args, *kwargs):	1✔
44	attempt = 0	1✔
45	while attempt <= self.retries:	1✔
46	logger_suffix = f'Wrapped method "{func.__name__}" -'	1✔
47	try:	1✔
48	logger.debug("%s start request", logger_suffix)	1✔
49	return func(self, args, *kwargs)	1✔
50	except (TimeoutException, HTTPStatusError) as e:	1✔
51	if isinstance(e, HTTPStatusError):	1✔
52	if e.response.status_code == 429:	1✔
53	logger.debug("%s too many requests", logger_suffix)	1✔
54	elif e.response.status_code < 500:	1✔
55	logger.debug("%s not server error", logger_suffix)	1✔
56	raise e	1✔
57	attempt += 1	1✔
58	if attempt > self.retries:	1✔
59	logger.debug("%s run out attempts", logger_suffix)	1✔
60	raise e	1✔
61	sleep(1)	1✔
62	logger.debug("%s attempt %d/%d", logger_suffix, attempt, self.retries)	1✔
63	except RemoteProtocolError:	1✔
64	# Запрос иногда рандомно обрывается сервером, проходит при повторном запросе
65	logger.debug("%s server disconnect", logger_suffix)	1✔
66
67	return wrapper	1✔
68
69
70	class Nspd(BaseNspdClient):	1✔
71	"""Клиент для НСПД
72
73	Example:
74	```python
75	with pynspd.Nspd() as nspd:
76	feat = nspd.find_zu("77:05:0001005:19")
77	```
78
79	Args:
80	timeout:
81	Время ожидания ответа.
82	Если не установлен - есть вероятность бесконечного ожидания. По умолчанию None.
83	retries:
84	Количество попыток при неудачном запросе
85	(таймаут, неожиданный обрыв соединения, 5хх ошибки). По умолчанию 10.
86	proxy:
87	Использовать прокси для запросов. По умолчанию None.
88	cache_storage:
89	Настройка хранения кэша (см. https://hishel.com/advanced/storages/).
90	Если установлен, то при повторном запросе результат будет
91	извлекаться из хранилища кэша, что сильно увеличивает производительность
92	и снижает риск ошибки 429 - Too many requests. По умолчанию None.
93	"""
94
95	def __init__(	1✔
96	self,
97	*,
98	timeout: Optional[int] = None,
99	retries: int = 10,
100	proxy: Optional[ProxyTypes] = None,
101	cache_storage: Optional[BaseStorage] = None,
102	):
103	super().__init__(retries=retries)	1✔
104	self._client = self._build_client(	1✔
105	timeout=timeout,
106	retries=retries,
107	proxy=proxy,
108	cache_storage=cache_storage,
109	)
110
111	@staticmethod	1✔
112	def _build_client(	1✔
113	timeout: Optional[int],
114	retries: int,
115	proxy: Optional[ProxyTypes],
116	cache_storage: Optional[BaseStorage],
117	) -> Client:
118	client_args = get_client_args(timeout)	1✔
119	transport: BaseTransport = HTTPTransport(	1✔
120	verify=SSL_CONTEXT, retries=retries, proxy=proxy
121	)
122	if cache_storage is not None:	1✔
123	transport = CacheTransport(	1✔
124	transport=transport,
125	storage=cache_storage,
126	controller=NSPD_CACHE_CONTROLLER,
127	)
128	return Client(**client_args, transport=transport)	1✔
129
130	def __enter__(self):	1✔
131	return self	1✔
132
133	def __exit__(self, *exc):	1✔
134	self.close()	1✔
135
136	def close(self):	1✔
137	"""Завершение сессии"""
138	self._client.close()	1✔
139
140	def request(	1✔
141	self,
142	method: str,
143	url: str,
144	params: Optional[QueryParamTypes] = None,
145	json: Optional[dict] = None,
146	) -> Response:
147	"""Базовый запрос к API НСПД"""
148	logger.debug("Request %s", url)	1✔
149	r = self._client.request(method, url, params=params, json=json)	1✔
150	r.raise_for_status()	1✔
151	return r	1✔
152
153	@retry_on_http_error	1✔
154	def safe_request(	1✔
155	self,
156	method: str,
157	url: str,
158	params: Optional[QueryParamTypes] = None,
159	json: Optional[dict] = None,
160	) -> Response:
161	"""Базовый запрос к api НСПД с обработкой ошибок"""
162	return self.request(method, url, params, json)	1✔
163
164	####################
165	### QUERY SEARCH ###
166	####################
167
168	@retry_on_http_error	1✔
169	def _search(self, params: dict[str, Any]) -> Optional[list[NspdFeature]]:	1✔
170	"""Базовый поисковый запрос на НСПД"""
171	try:	1✔
172	r = self.request("get", "/api/geoportal/v2/search/geoportal", params=params)	1✔
173	return SearchResponse.model_validate(r.json()).data.features	1✔
174	except HTTPStatusError as e:	1✔
175	if e.response.status_code == 404:	1✔
176	return None	1✔
UNCOV 177	raise e	×
178
179	def search(	1✔
180	self, query: str, theme_id: ThemeId = ThemeId.REAL_ESTATE_OBJECTS
181	) -> Optional[list[NspdFeature]]:
182	"""Поисковой запрос по предустановленной теме
183
184	Args:
185	query: Поисковой запрос
186	theme_id:
187	Вид объекта (кадастровое деление, объект недвижимости и т.д.).
188	По умолчанию: объекты недвижимости
189
190	Returns:
191	Положительный ответ от сервиса, либо None, если ничего не найдено
192	"""
193	return self._search(	1✔
194	params={
195	"query": query,
196	"thematicSearchId": theme_id.value,
197	}
198	)
199
200	def search_in_layer(	1✔
201	self, query: str, layer_def: Type[Feat]
202	) -> Optional[list[Feat]]:
203	"""Поиск объекта по определению слоя
204
205	Args:
206	query: Поисковой запрос
207	layer_def: Определение слоя
208
209	Returns:
210	Валидированная модель слоя, если найдено
211	"""
212	raw_features = self._search(	1✔
213	params={
214	"query": query,
215	"layersId": layer_def.layer_meta.layer_id,
216	}
217	)
218	return self._cast_features_to_layer_defs(raw_features, layer_def)	1✔
219
220	def find(	1✔
221	self, query: str, theme_id: ThemeId = ThemeId.REAL_ESTATE_OBJECTS
222	) -> Optional[NspdFeature]:
223	"""Найти объект по предустановленной теме
224
225	Args:
226	query: Поисковой запрос
227	theme_id:
228	Вид объекта (кадастровое деление, объект недвижимости и т.д.).
229	По умолчанию: объекты недвижимости
230
231	Returns:
232	Положительный ответ от сервиса, либо None, если ничего не найдено
233	"""
234	return self._filter_search_by_query(self.search(query, theme_id), query)	1✔
235
236	def find_in_layer(self, query: str, layer_def: Type[Feat]) -> Optional[Feat]:	1✔
237	"""Найти объект по определению слоя
238
239	Args:
240	query: Поисковой запрос
241	layer_def: Определение слоя
242
243	Returns:
244	Валидированная модель слоя, если найдено
245	"""
246	return self._filter_search_by_query(	1✔
247	self.search_in_layer(query, layer_def), query
248	)
249
250	######################
251	### POLYGON SEARCH ###
252	######################
253
254	@retry_on_http_error	1✔
255	def _search_in_contour(	1✔
256	self,
257	countour: Union[Polygon, MultiPolygon],
258	*category_ids: int,
259	) -> Optional[list[NspdFeature]]:
260	"""Поиск объектов в контуре по ID категорий слоев
261
262	Args:
263	countour: Геометрический объект с контуром
264	category_ids: ID категорий слоев
265
266	Returns:
267	Список объектов, пересекающихся с контуром, если найден хоть один
268	"""
269	feature_geojson = json.loads(to_geojson(countour))	1✔
270	feature_geojson["crs"] = {	1✔
271	"type": "name",
272	"properties": {"name": "EPSG:4326"},
273	}
274	payload = {	1✔
275	"categories": [{"id": id_} for id_ in category_ids],
276	"geom": {
277	"type": "FeatureCollection",
278	"features": [
279	{"geometry": feature_geojson, "type": "Feature", "properties": {}}
280	],
281	},
282	}
283	response = self.request(	1✔
284	"post",
285	"/api/geoportal/v1/intersects",
286	params={"typeIntersect": "fullObject"},
287	json=payload,
288	)
289	return self._validate_feature_collection_response(response)	1✔
290
291	def search_in_contour(	1✔
292	self,
293	countour: Union[Polygon, MultiPolygon],
294	layer_def: Type[Feat],
295	) -> Optional[list[Feat]]:
296	"""Поиск объектов слоя в контуре
297
298	Args:
299	countour: Геометрический объект с контуром
300	layer_def: Модель слоя
301
302	Returns:
303	Список объектов, пересекающихся с контуром, если найден хоть один
304	"""
305	raw_features = self._search_in_contour(	1✔
306	countour, layer_def.layer_meta.category_id
307	)
308	return self._cast_features_to_layer_defs(raw_features, layer_def)	1✔
309
310	####################
311	### POINT SEARCH ###
312	####################
313
314	@retry_on_http_error	1✔
315	def _search_at_point(self, pt: Point, layer_id: int) -> Optional[list[NspdFeature]]:	1✔
316	"""Поиск объектов слоя в точке"""
317	tile_size = 512	1✔
318	tile = mercantile.tile(	1✔
319	pt.x, pt.y, zoom=24
320	) # zoom=24 должно быть достаточно для самого точного совпадения
321	tile_bounds = mercantile.bounds(tile)	1✔
322	i = np.interp(pt.x, [tile_bounds.west, tile_bounds.east], [0, tile_size])	1✔
323	j = np.interp(pt.y, [tile_bounds.south, tile_bounds.north], [0, tile_size])	1✔
324	bbox = ",".join(map(str, tile_bounds))	1✔
325	params = {	1✔
326	"REQUEST": "GetFeatureInfo",
327	"SERVICE": "WMS",
328	"VERSION": "1.3.0",
329	"INFO_FORMAT": "application/json",
330	"FORMAT": "image/png",
331	"STYLES": "",
332	"TRANSPARENT": "true",
333	"QUERY_LAYERS": layer_id,
334	"LAYERS": layer_id,
335	"WIDTH": tile_size,
336	"HEIGHT": tile_size,
337	"I": int(i),
338	"J": tile_size - int(j), # отсчет координат для пикселей ведется сверху
339	"CRS": "EPSG:4326", # CRS для bbox
340	"BBOX": bbox,
341	"FEATURE_COUNT": "10", # Иначе вернет только один, даже если попало на границу
342	}
343	response = self.request(	1✔
344	"get",
345	f"/api/aeggis/v3/{layer_id}/wms",
346	params=params, # type: ignore[arg-type]
347	)
348	return self._validate_feature_collection_response(response)	1✔
349
350	def search_at_point(self, pt: Point, layer_def: Type[Feat]) -> Optional[list[Feat]]:	1✔
351	"""Поиск объектов слоя в точке (с типизацией)
352
353	Args:
354	pt: Точка поиска
355	layer_def: Тип слоя
356
357	Returns:
358	Типизированный список объектов, если найдены
359	"""
360	raw_features = self._search_at_point(pt, layer_def.layer_meta.layer_id)	1✔
361	return self._cast_features_to_layer_defs(raw_features, layer_def)	1✔
362
363	def search_at_coords(	1✔
364	self, lat: float, lng: float, layer_def: Type[Feat]
365	) -> Optional[list[Feat]]:
366	"""Поиск объектов слоя в координатах
367
368	Args:
369	lat: Широта
370	lng: Долгота
371	layer_def: Тип слоя
372
373	Returns:
374	Типизированный список объектов, если найдены
375	"""
376	return self.search_at_point(Point(lng, lat), layer_def)	1✔
377
378	####################
379	### TAB REQUESTS ###
380	####################
381
382	@retry_on_http_error	1✔
383	def _tab_request(	1✔
384	self, feat: NspdFeature, tab_class: str, type_: Literal["values", "group"]
385	) -> Optional[dict]:
386	if feat.properties.options.no_coords:	1✔
387	params = {	1✔
388	"tabClass": tab_class,
389	"objdocId": feat.properties.options.objdoc_id,
390	"registersId": feat.properties.options.registers_id,
391	}
392	else:
393	params = {	1✔
394	"tabClass": tab_class,
395	"categoryId": feat.properties.category,
396	"geomId": feat.id,
397	}
398	try:	1✔
399	r = self.request(	1✔
400	"get", f"/api/geoportal/v1/tab-{type_}-data", params=params
401	)
402	return r.json()	1✔
403	except HTTPStatusError as e:	1✔
404	if e.response.status_code == 404:	1✔
405	return None	1✔
UNCOV 406	raise e	×
407
408	def _tab_values_request(	1✔
409	self, feat: NspdFeature, tab_class: str
410	) -> Optional[list[str]]:
411	resp = self._tab_request(feat, tab_class, "values")	1✔
412	if resp is None:	1✔
413	return None	1✔
414	return NspdTabResponse.model_validate(resp).value	1✔
415
416	def _tab_groups_request(	1✔
417	self, feat: NspdFeature, tab_class: str
418	) -> Optional[dict[str, Optional[list[str]]]]:
419	resp = self._tab_request(feat, tab_class, "group")	1✔
420	if resp is None:	1✔
UNCOV 421	return None	×
422	item = NspdTabGroupResponse.model_validate(resp).object	1✔
423	data = {}	1✔
424	for i in item:	1✔
425	title = re.sub(r"[\s:]+$", "", i.title)	1✔
426	data[title] = i.value	1✔
427	if len(data) == 0:	1✔
UNCOV 428	return None	×
429	return data	1✔
430
431	def tab_land_parts(self, feat: NspdFeature) -> Optional[list[str]]:	1✔
432	"""Получение данных с вкладки \"Части ЗУ\" """
433	return self._tab_values_request(feat, "landParts")	1✔
434
435	def tab_land_links(self, feat: NspdFeature) -> Optional[list[str]]:	1✔
436	"""Получение данных с вкладки \"Связанные ЗУ\" """
437	return self._tab_values_request(feat, "landLinks")	1✔
438
439	def tab_permission_type(self, feat: NspdFeature) -> Optional[list[str]]:	1✔
440	"""Получение данных с вкладки \"Виды разрешенного использования\" """
441	return self._tab_values_request(feat, "permissionType")	1✔
442
443	def tab_composition_land(self, feat: NspdFeature) -> Optional[list[str]]:	1✔
444	"""Получение данных с вкладки \"Состав ЕЗП\" """
445	return self._tab_values_request(feat, "compositionLand")	1✔
446
447	def tab_build_parts(self, feat: NspdFeature) -> Optional[list[str]]:	1✔
448	"""Получение данных с вкладки \"Части ОКС\" """
449	return self._tab_values_request(feat, "buildParts")	1✔
450
451	def tab_objects_list(	1✔
452	self, feat: NspdFeature
453	) -> Optional[dict[str, Optional[list[str]]]]:
454	"""
455	Получение данных с вкладки \"Объекты\"
456	"""
457	return self._tab_groups_request(feat, "objectsList")	1✔
458
459	def get_tab_data(self, feat: NspdFeature, tab_name: TabTitle):	1✔
460	"""Получение данных с указанной вкладки"""
461	match tab_name:	1✔
462	case "Части ЗУ":	1✔
463	return self.tab_land_parts(feat)	1✔
464	case "Связанные ЗУ":	1✔
465	return self.tab_land_links(feat)	1✔
466	case "Виды разрешенного использования":	1✔
467	return self.tab_permission_type(feat)	1✔
468	case "Состав ЕЗП":	1✔
469	return self.tab_composition_land(feat)	1✔
470	case "Части ОКС":	1✔
471	return self.tab_build_parts(feat)	1✔
472	case "Объекты":	1✔
473	return self.tab_objects_list(feat)	1✔
474
475	#################
476	### SHORTCUTS ###
477	#################
478
479	def find_landplot(self, query: str) -> Optional[Layer36048Feature]:	1✔
480	"""Найти ЗУ по кадастровому номеру"""
481	return self._filter_search_by_query(self.search_landplots(query), query)	1✔
482
483	def find_building(self, query: str) -> Optional[Layer36049Feature]:	1✔
484	"""Найти ОКС по кадастровому номеру"""
485	return self._filter_search_by_query(self.search_buildings(query), query)	1✔
486
487	def search_landplots_at_point(self, pt: Point) -> Optional[list[Layer36048Feature]]:	1✔
488	"""Поиск ЗУ в точке"""
UNCOV 489	return self.search_at_point(pt, Layer36048Feature)	×
490
491	def search_buildings_at_point(self, pt: Point) -> Optional[list[Layer36049Feature]]:	1✔
492	"""Поиск ОКС в точке"""
UNCOV 493	return self.search_at_point(pt, Layer36049Feature)	×
494
495	def search_landplots(self, cn: str) -> Optional[list[Layer36048Feature]]:	1✔
496	"""Поиск ЗУ по кадастровому номеру"""
497	return self.search_in_layer(cn, Layer36048Feature)	1✔
498
499	def search_buildings(self, cn: str) -> Optional[list[Layer36049Feature]]:	1✔
500	"""Поиск ОКС по кадастровому номеру"""
501	return self.search_in_layer(cn, Layer36049Feature)	1✔
502
503	def search_landplots_at_coords(	1✔
504	self, lat: float, lng: float
505	) -> Optional[list[Layer36048Feature]]:
506	"""Поиск ЗУ в координатах"""
507	return self.search_at_coords(lat, lng, Layer36048Feature)	1✔
508
509	def search_buildings_at_coords(	1✔
510	self, lat: float, lng: float
511	) -> Optional[list[Layer36049Feature]]:
512	"""Поиск ОКС в координатах"""
513	return self.search_at_coords(lat, lng, Layer36049Feature)	1✔
514
515	def search_landplots_in_contour(	1✔
516	self, countour: Union[Polygon, MultiPolygon]
517	) -> Optional[list[Layer36048Feature]]:
518	"""Поиск ЗУ в контуре"""
519	return self.search_in_contour(countour, Layer36048Feature)	1✔
520
521	def search_buildings_in_contour(	1✔
522	self, countour: Union[Polygon, MultiPolygon]
523	) -> Optional[list[Layer36049Feature]]:
524	"""Поиск ОКС в контуре"""
525	return self.search_in_contour(countour, Layer36049Feature)	1✔

yazmolod / pynspd / #10

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous