yazmolod / pynspd / #5

Committed 18 Feb 2025 03:39PM UTC coverage: 98.824% (+1.2%) from 97.584%

Build # #5

Build Type

push

coveralls-python

Committed by Aleksandr Litovchenko

Commit Message

fix!: remove deprecated

Run Details

2184 of 2210 relevant lines covered (98.82%)

0.99 hits per line

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

96.83

/src/pynspd/_sync/api.py

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

import mercantile
import numpy as np
import typing_extensions
from hishel import BaseStorage, CacheTransport, Controller
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 (
    SSL_CONTEXT,
    BaseNspdClient,
    get_client_args,
    get_controller_args,
)
from pynspd.errors import TooBigContour
from pynspd.logger import logger
from pynspd.schemas import Layer36048Feature, Layer36049Feature, NspdFeature
from pynspd.schemas.feature import Feat
from pynspd.schemas.responses import (
    NspdTabGroupResponse,
    NspdTabResponse,
    SearchResponse,
)
from pynspd.types.enums import ThemeId


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)
                        sleep(1)
                    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
                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):
    """Клиент для НСПД

    ```python
    with pynspd.Nspd() as nspd:
        feat = nspd.search_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:
            cache_args = get_controller_args()
            controller = Controller(**cache_args)
            transport = CacheTransport(
                transport=transport, storage=cache_storage, controller=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 _search(self, params: dict[str, Any]) -> Optional[SearchResponse]:
        try:
            r = self.request("get", "/api/geoportal/v2/search/geoportal", params=params)
            return self._validate_search_response(r)
        except HTTPStatusError as e:
            if e.response.status_code == 404:
                return None
            raise e

    def _search_one(self, params: dict[str, Any]) -> Optional[NspdFeature]:
        response = self._search(params)
        if response is None:
            return None
        features = response.data.features
        # иногда поиск багует и дает помимо нужного еще и рандомный результат
        if len(features) > 1:
            features = list(
                filter(
                    lambda x: params["query"] in x.properties.model_dump_json(),
                    features,
                )
            )
        if len(features) == 0:
            return None
        assert len(features) == 1
        return features[0]

    @typing_extensions.deprecated(
        "Will be removed in 0.7.0; use `.search_in_theme(...)` instead`"
    )
    def search_by_theme(
        self, query: str, theme_id: ThemeId = ThemeId.REAL_ESTATE_OBJECTS
    ) -> Optional[NspdFeature]:
        """Поисковой запрос по предустановленной теме

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

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

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

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

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

    @typing_extensions.deprecated(
        "Will be removed in 0.7.0; use `.search_in_layer(...)` instead`"
    )
    def search_by_layers(self, query: str, *layer_ids: int) -> Optional[NspdFeature]:
        """Поисковой запрос по указанным слоям

        Args:
            query: поисковой запрос
            *layer_ids: id слоев, в которых будет производиться поиск

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

    def search_in_layer(self, query: str, layer_id: int) -> Optional[NspdFeature]:
        """Поисковой запрос по указанному слою

        Args:
            query: поисковой запрос
            layer_id: id слоя, в которых будет производиться поиск

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

    @typing_extensions.deprecated(
        "Will be removed in 0.7.0; use `.search_in_layer_by_model(...)` instead`"
    )
    def search_by_model(self, query: str, layer_def: Type[Feat]) -> Optional[Feat]:
        """Поиск одного объекта по определению слоя

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

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

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

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

        Returns:
            Валидированная модель слоя, если найдено
        """
        feature = self.search_in_layer(query, layer_def.layer_meta.layer_id)
        if feature is None:
            return None
        return feature.cast(layer_def)

    def search_zu(self, cn: str) -> Optional[Layer36048Feature]:
        """Поиск ЗУ по кадастровому номеру"""
        layer_def = NspdFeature.by_title("Земельные участки из ЕГРН")
        return self.search_in_layer_by_model(cn, layer_def)

    def search_oks(self, cn: str) -> Optional[Layer36049Feature]:
        """Поиск ОКС по кадастровому номеру"""
        layer_def = NspdFeature.by_title("Здания")
        return self.search_in_layer_by_model(cn, layer_def)

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

        Args:
            countour: Геометрический объект с контуром
            category_ids: ID категорий слоев
            epsg: Система координат контура. По умолчанию 4326.

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

    def search_in_contour_by_model(
        self,
        countour: Union[Polygon, MultiPolygon],
        layer_def: Type[Feat],
        epsg: int = 4326,
    ) -> Optional[list[Feat]]:
        """Поиск объектов в контуре по определению слоя

        Args:
            countour: Геометрический объект с контуром
            layer_def: Модель слоя
            epsg: Система координат контура. По умолчанию 4326.

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

    def search_zu_in_contour(
        self, countour: Union[Polygon, MultiPolygon], epsg: int = 4326
    ) -> Optional[list[Layer36048Feature]]:
        """Поиск ЗУ в контуре

        Args:
            countour: Геометрический объект с контуром
            epsg: Система координат контура. По умолчанию 4326.

        Returns:
            Список объектов, пересекающихся с контуром, если найден хоть один
        """
        return self.search_in_contour_by_model(countour, Layer36048Feature, epsg=epsg)

    def search_oks_in_contour(
        self, countour: Union[Polygon, MultiPolygon], epsg: int = 4326
    ) -> Optional[list[Layer36049Feature]]:
        """Поиск ОКС в контуре

        Args:
            countour: Геометрический объект с контуром
            epsg: Система координат контура. По умолчанию 4326.

        Returns:
            Список объектов, пересекающихся с контуром, если найден хоть один
        """
        return self.search_in_contour_by_model(countour, Layer36049Feature, epsg=epsg)

    @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, mercantile.xy_bounds(tile))
        )  # bbox в 3857, см. комментарий про CRS
        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:3857",  # CRS для bbox
            # можно указать и 4326, но тогда и геометрия будет в 4326
            # Но в других методах мы всегда ждем 3857, поэтому оставляем
            "BBOX": bbox,
            "FEATURE_COUNT": "10",  # Если не указать - вернет только один, даже если попало на границу
        }
        response = self.request("get", f"/api/aeggis/v3/{layer_id}/wms", params=params)
        return self._validate_feature_collection_response(response)

    def search_at_point_by_model(
        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_zu_at_point(self, pt: Point) -> Optional[list[Layer36048Feature]]:
        """Поиск ЗУ в точке"""
        return self.search_at_point_by_model(pt, Layer36048Feature)

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

    @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 = {i.title: i.value for i in item}
        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")

1	import json	1✔
2	from functools import wraps	1✔
3	from time import sleep	1✔
4	from typing import Any, Literal, Optional, Type, Union	1✔
5
6	import mercantile	1✔
7	import numpy as np	1✔
8	import typing_extensions	1✔
9	from hishel import BaseStorage, CacheTransport, Controller	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	SSL_CONTEXT,
24	BaseNspdClient,
25	get_client_args,
26	get_controller_args,
27	)
28	from pynspd.errors import TooBigContour	1✔
29	from pynspd.logger import logger	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	from pynspd.types.enums import ThemeId	1✔
38
39
40	def retry_on_http_error(func):	1✔
41	"""Декоратор для повторения запроса при ошибках запроса"""
42
43	@wraps(func)	1✔
44	def wrapper(self: "Nspd", args, *kwargs):	1✔
45	attempt = 0	1✔
46	while attempt <= self.retries:	1✔
47	logger_suffix = f'Wrapped method "{func.__name__}" -'	1✔
48	try:	1✔
49	logger.debug("%s start request", logger_suffix)	1✔
50	return func(self, args, *kwargs)	1✔
51	except (TimeoutException, HTTPStatusError) as e:	1✔
52	if isinstance(e, HTTPStatusError):	1✔
53	if e.response.status_code == 429:	1✔
54	logger.debug("%s too many requests", logger_suffix)	1✔
55	sleep(1)	1✔
56	elif e.response.status_code < 500:	1✔
57	logger.debug("%s not server error", logger_suffix)	1✔
58	raise e	1✔
59	attempt += 1	1✔
60	if attempt > self.retries:	1✔
61	logger.debug("%s run out attempts", logger_suffix)	1✔
62	raise e	1✔
63	logger.debug("%s attempt %d/%d", logger_suffix, attempt, self.retries)	1✔
64	except RemoteProtocolError:	1✔
65	# Запрос иногда рандомно обрывается сервером, проходит при повторном запросе
66	logger.debug("%s server disconnect", logger_suffix)	1✔
67
68	return wrapper	1✔
69
70
71	class Nspd(BaseNspdClient):	1✔
72	"""Клиент для НСПД
73
74	```python
75	with pynspd.Nspd() as nspd:
76	feat = nspd.search_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	cache_args = get_controller_args()	1✔
124	controller = Controller(**cache_args)	1✔
125	transport = CacheTransport(	1✔
126	transport=transport, storage=cache_storage, controller=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 _search(self, params: dict[str, Any]) -> Optional[SearchResponse]:	1✔
155	try:
156	r = self.request("get", "/api/geoportal/v2/search/geoportal", params=params)
157	return self._validate_search_response(r)
158	except HTTPStatusError as e:
159	if e.response.status_code == 404:
160	return None
161	raise e
162		1✔
163	def _search_one(self, params: dict[str, Any]) -> Optional[NspdFeature]:
164	response = self._search(params)	1✔
165	if response is None:	1✔
166	return None	1✔
167	features = response.data.features	1✔
168	# иногда поиск багует и дает помимо нужного еще и рандомный результат	1✔
169	if len(features) > 1:	1✔
170	features = list(	1✔
171	filter(	1✔
172	lambda x: params["query"] in x.properties.model_dump_json(),	×
173	features,
174	)	1✔
175	)	1✔
176	if len(features) == 0:	1✔
177	return None	1✔
178	assert len(features) == 1	1✔
179	return features[0]
180		1✔
181	@typing_extensions.deprecated(	1✔
182	"Will be removed in 0.7.0; use `.search_in_theme(...)` instead`"
183	)
184	def search_by_theme(
185	self, query: str, theme_id: ThemeId = ThemeId.REAL_ESTATE_OBJECTS
186	) -> Optional[NspdFeature]:
187	"""Поисковой запрос по предустановленной теме	1✔
188		1✔
189	Args:	1✔
190	query: Поисковой запрос	1✔
191	theme_id: Вид объекта (кадастровое деление, объект недвижимости и т.д.)
192		1✔
193	Returns:
194	Положительный ответ от сервиса, либо None, если ничего не найдено
195	"""	1✔
196	return self.search_in_theme(query, theme_id)
197
198	def search_in_theme(
199	self, query: str, theme_id: ThemeId = ThemeId.REAL_ESTATE_OBJECTS
200	) -> Optional[NspdFeature]:
201	"""Поисковой запрос по предустановленной теме
202
203	Args:
204	query: Поисковой запрос
205	theme_id: Вид объекта (кадастровое деление, объект недвижимости и т.д.)
206
207	Returns:	1✔
208	Положительный ответ от сервиса, либо None, если ничего не найдено
209	"""	1✔
210	return self._search_one(
211	params={
212	"query": query,
213	"thematicSearchId": theme_id.value,
214	}
215	)
216
217	@typing_extensions.deprecated(
218	"Will be removed in 0.7.0; use `.search_in_layer(...)` instead`"
219	)
220	def search_by_layers(self, query: str, *layer_ids: int) -> Optional[NspdFeature]:
221	"""Поисковой запрос по указанным слоям	1✔
222
223	Args:
224	query: поисковой запрос
225	*layer_ids: id слоев, в которых будет производиться поиск
226
227	Returns:
228	Положительный ответ от сервиса, либо None, если ничего не найдено	1✔
229	"""
230	return self._search_one(
231	params={	1✔
232	"query": query,
233	"layersId": layer_ids,
234	}
235	)
236
237	def search_in_layer(self, query: str, layer_id: int) -> Optional[NspdFeature]:
238	"""Поисковой запрос по указанному слою
239
240	Args:
241	query: поисковой запрос	×
242	layer_id: id слоя, в которых будет производиться поиск
243
244	Returns:
245	Положительный ответ от сервиса, либо None, если ничего не найдено
246	"""
247	return self._search_one(
248	params={	1✔
249	"query": query,
250	"layersId": layer_id,
251	}
252	)
253
254	@typing_extensions.deprecated(
255	"Will be removed in 0.7.0; use `.search_in_layer_by_model(...)` instead`"
256	)
257	def search_by_model(self, query: str, layer_def: Type[Feat]) -> Optional[Feat]:
258	"""Поиск одного объекта по определению слоя	1✔
259
260	Args:
261	query: Поисковой запрос
262	layer_def: Определение слоя
263
264	Returns:
265	Валидированная модель слоя, если найдено	1✔
266	"""
267	return self.search_in_layer_by_model(query, layer_def)
268		1✔
269	def search_in_layer_by_model(
270	self, query: str, layer_def: Type[Feat]
271	) -> Optional[Feat]:
272	"""Поиск объекта по определению слоя
273
274	Args:
275	query: Поисковой запрос
276	layer_def: Определение слоя
277
278	Returns:	1✔
279	Валидированная модель слоя, если найдено
280	"""	1✔
281	feature = self.search_in_layer(query, layer_def.layer_meta.layer_id)
282	if feature is None:
283	return None
284	return feature.cast(layer_def)
285
286	def search_zu(self, cn: str) -> Optional[Layer36048Feature]:
287	"""Поиск ЗУ по кадастровому номеру"""
288	layer_def = NspdFeature.by_title("Земельные участки из ЕГРН")
289	return self.search_in_layer_by_model(cn, layer_def)
290
291	def search_oks(self, cn: str) -> Optional[Layer36049Feature]:
292	"""Поиск ОКС по кадастровому номеру"""	1✔
293	layer_def = NspdFeature.by_title("Здания")	1✔
294	return self.search_in_layer_by_model(cn, layer_def)	1✔
295		1✔
296	@retry_on_http_error
297	def search_in_contour(	1✔
298	self,
299	countour: Union[Polygon, MultiPolygon],	1✔
300	*category_ids: int,	1✔
301	epsg: int = 4326,
302	) -> Optional[list[NspdFeature]]:	1✔
303	"""Поиск объектов в контуре по ID категорий слоев
304		1✔
305	Args:	1✔
306	countour: Геометрический объект с контуром
307	category_ids: ID категорий слоев	1✔
308	epsg: Система координат контура. По умолчанию 4326.	1✔
309
310	Returns:
311	Список объектов, пересекающихся с контуром, если найден хоть один
312	"""
313	feature_geojson = json.loads(to_geojson(countour))
314	feature_geojson["crs"] = {
315	"type": "name",
316	"properties": {"name": f"EPSG:{epsg}"},
317	}
318	payload = {
319	"categories": [{"id": id_} for id_ in category_ids],
320	"geom": {
321	"type": "FeatureCollection",
322	"features": [
323	{"geometry": feature_geojson, "type": "Feature", "properties": {}}
324	],	1✔
325	},	1✔
326	}
327	try:
328	response = self.request(
329	"post",	1✔
330	"/api/geoportal/v1/intersects",
331	params={"typeIntersect": "fullObject"},
332	json=payload,
333	)
334	except HTTPStatusError as e:
335	if e.response.status_code == 500 and e.response.json()["code"] == 400004:
336	raise TooBigContour from e
337	raise e
338	return self._validate_feature_collection_response(response)	1✔
339		1✔
340	def search_in_contour_by_model(
341	self,
342	countour: Union[Polygon, MultiPolygon],
343	layer_def: Type[Feat],
344	epsg: int = 4326,
345	) -> Optional[list[Feat]]:	1✔
346	"""Поиск объектов в контуре по определению слоя	1✔
347		1✔
348	Args:	×
349	countour: Геометрический объект с контуром	1✔
350	layer_def: Модель слоя
351	epsg: Система координат контура. По умолчанию 4326.	1✔
352
353	Returns:
354	Список объектов, пересекающихся с контуром, если найден хоть один
355	"""
356	raw_features = self.search_in_contour(
357	countour, layer_def.layer_meta.category_id, epsg=epsg
358	)
359	return self._cast_features_to_layer_defs(raw_features, layer_def)
360
361	def search_zu_in_contour(
362	self, countour: Union[Polygon, MultiPolygon], epsg: int = 4326
363	) -> Optional[list[Layer36048Feature]]:
364	"""Поиск ЗУ в контуре
365
366	Args:
367	countour: Геометрический объект с контуром	1✔
368	epsg: Система координат контура. По умолчанию 4326.
369
370	Returns:	1✔
371	Список объектов, пересекающихся с контуром, если найден хоть один
372	"""	1✔
373	return self.search_in_contour_by_model(countour, Layer36048Feature, epsg=epsg)
374
375	def search_oks_in_contour(
376	self, countour: Union[Polygon, MultiPolygon], epsg: int = 4326
377	) -> Optional[list[Layer36049Feature]]:
378	"""Поиск ОКС в контуре
379
380	Args:
381	countour: Геометрический объект с контуром
382	epsg: Система координат контура. По умолчанию 4326.
383
384	Returns:	1✔
385	Список объектов, пересекающихся с контуром, если найден хоть один
386	"""	1✔
387	return self.search_in_contour_by_model(countour, Layer36049Feature, epsg=epsg)
388
389	@retry_on_http_error
390	def search_at_point(self, pt: Point, layer_id: int) -> Optional[list[NspdFeature]]:
391	"""Поиск объектов слоя в точке"""
392	tile_size = 512
393	tile = mercantile.tile(
394	pt.x, pt.y, zoom=24
395	) # zoom=24 должно быть достаточно для самого точного совпадения
396	tile_bounds = mercantile.bounds(tile)
397	i = np.interp(pt.x, [tile_bounds.west, tile_bounds.east], [0, tile_size])
398	j = np.interp(pt.y, [tile_bounds.south, tile_bounds.north], [0, tile_size])	1✔
399	bbox = ",".join(
400	map(str, mercantile.xy_bounds(tile))	1✔
401	) # bbox в 3857, см. комментарий про CRS	1✔
402	params = {
403	"REQUEST": "GetFeatureInfo",	1✔
404	"SERVICE": "WMS",	1✔
405	"VERSION": "1.3.0",
406	"INFO_FORMAT": "application/json",
407	"FORMAT": "image/png",	1✔
408	"STYLES": "",	1✔
409	"TRANSPARENT": "true",	1✔
410	"QUERY_LAYERS": layer_id,	1✔
411	"LAYERS": layer_id,
412	"WIDTH": tile_size,
413	"HEIGHT": tile_size,	1✔
414	"I": int(i),
415	"J": tile_size - int(j), # отсчет координат для пикселей ведется сверху
416	"CRS": "EPSG:3857", # CRS для bbox
417	# можно указать и 4326, но тогда и геометрия будет в 4326
418	# Но в других методах мы всегда ждем 3857, поэтому оставляем
419	"BBOX": bbox,
420	"FEATURE_COUNT": "10", # Если не указать - вернет только один, даже если попало на границу
421	}
422	response = self.request("get", f"/api/aeggis/v3/{layer_id}/wms", params=params)
423	return self._validate_feature_collection_response(response)
424
425	def search_at_point_by_model(
426	self, pt: Point, layer_def: Type[Feat]
427	) -> Optional[list[Feat]]:
428	"""Поиск объектов слоя в точке (с типизацией)
429
430	Args:
431	pt: Точка поиска
432	layer_def: Тип слоя
433		1✔
434	Returns:	1✔
435	Типизированный список объектов, если найдены
436	"""	1✔
437	raw_features = self.search_at_point(pt, layer_def.layer_meta.layer_id)
438	return self._cast_features_to_layer_defs(raw_features, layer_def)
439
440	def search_zu_at_point(self, pt: Point) -> Optional[list[Layer36048Feature]]:
441	"""Поиск ЗУ в точке"""
442	return self.search_at_point_by_model(pt, Layer36048Feature)
443
444	def search_oks_at_point(self, pt: Point) -> Optional[list[Layer36049Feature]]:
445	"""Поиск ОКС в точке"""
446	return self.search_at_point_by_model(pt, Layer36049Feature)
447
448	@retry_on_http_error	1✔
449	def _tab_request(	1✔
450	self, feat: NspdFeature, tab_class: str, type_: Literal["values", "group"]
451	) -> Optional[dict]:	1✔
452	if feat.properties.options.no_coords:
453	params = {	1✔
454	"tabClass": tab_class,
455	"objdocId": feat.properties.options.objdoc_id,	1✔
456	"registersId": feat.properties.options.registers_id,
457	}	1✔
458	else:
459	params = {	1✔
460	"tabClass": tab_class,	1✔
461	"categoryId": feat.properties.category,
462	"geomId": feat.id,
463	}	1✔
464	try:	1✔
465	r = self.request(
466	"get", f"/api/geoportal/v1/tab-{type_}-data", params=params
467	)
468	return r.json()
469	except HTTPStatusError as e:
470	if e.response.status_code == 404:	1✔
471	return None
472	raise e
473
474	def _tab_values_request(
475	self, feat: NspdFeature, tab_class: str	1✔
476	) -> Optional[list[str]]:	1✔
477	resp = self._tab_request(feat, tab_class, "values")
478	if resp is None:
479	return None	1✔
480	return NspdTabResponse.model_validate(resp).value	1✔
481		1✔
482	def _tab_groups_request(	1✔
483	self, feat: NspdFeature, tab_class: str	×
484	) -> Optional[dict[str, Optional[list[str]]]]:
485	resp = self._tab_request(feat, tab_class, "group")	1✔
486	if resp is None:
487	return None
488	item = NspdTabGroupResponse.model_validate(resp).object	1✔
489	data = {i.title: i.value for i in item}	1✔
490	if len(data) == 0:	1✔
491	return None	1✔
492	return data
493		1✔
494	def tab_land_parts(self, feat: NspdFeature) -> Optional[list[str]]:
495	"""Получение данных с вкладки \"Части ЗУ\" """
496	return self._tab_values_request(feat, "landParts")	1✔
497		1✔
498	def tab_land_links(self, feat: NspdFeature) -> Optional[list[str]]:	×
499	"""Получение данных с вкладки \"Связанные ЗУ\" """	1✔
500	return self._tab_values_request(feat, "landLinks")	1✔
501		1✔
502	def tab_permission_type(self, feat: NspdFeature) -> Optional[list[str]]:	×
503	"""Получение данных с вкладки \"Виды разрешенного использования\" """	1✔
504	return self._tab_values_request(feat, "permissionType")
505		1✔
506	def tab_composition_land(self, feat: NspdFeature) -> Optional[list[str]]:
507	"""Получение данных с вкладки \"Состав ЕЗП\" """	1✔
508	return self._tab_values_request(feat, "compositionLand")
509		1✔
510	def tab_build_parts(self, feat: NspdFeature) -> Optional[list[str]]:
511	"""Получение данных с вкладки \"Части ОКС\" """	1✔
512	return self._tab_values_request(feat, "buildParts")
513		1✔
514	def tab_objects_list(
515	self, feat: NspdFeature	1✔
516	) -> Optional[dict[str, Optional[list[str]]]]:
517	"""	1✔
518	Получение данных с вкладки \"Объекты\"
519	"""	1✔
520	return self._tab_groups_request(feat, "objectsList")

yazmolod / pynspd / #5

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