yazmolod / pynspd / #3

Committed 03 Feb 2025 11:06PM UTC coverage: 98.285% (+0.02%) from 98.27%

Build # #3

Build Type

push

coveralls-python

Committed by

yazmolod

Commit Message

bump version

Run Details

1891 of 1924 relevant lines covered (98.28%)

0.98 hits per line

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

91.93

/src/pynspd/_sync/api.py

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

import mercantile
import numpy as np
from httpx import HTTPStatusError, RemoteProtocolError, Response
from httpx._types import QueryParamTypes
from shapely import MultiPolygon, Point, Polygon, to_geojson

from pynspd import asyncio_mock
from pynspd.client import BaseNspdClient, ProxyTypes, get_client
from pynspd.errors import TooBigContour
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:
            try:
                return func(self, *args, **kwargs)
            except HTTPStatusError as e:
                if e.response.status_code < 500:
                    raise e
                attempt += 1
                if attempt > self.retries:
                    raise e
            except RemoteProtocolError:
                # Запрос иногда рандомно обрывается сервером, проходит при повторном запросе
                pass

    return wrapper


class Nspd(BaseNspdClient):
    def __init__(
        self,
        timeout: Optional[int] = None,
        retries: int = 10,
        proxy: Optional[ProxyTypes] = None,
    ):
        """Клиент для НСПД

        Usage:
        >>> with pynspd.Nspd() as nspd:
        >>>     feat = nspd.search_zu("77:05:0001005:19")

        Args:
            timeout (Optional[int], optional): Время ожидания ответа. Defaults to None.
            retries (int, optional): Количество попыток при неудачном запросе (таймаут или 5хх ошибки). Defaults to 10.
            proxy (Optional[ProxyTypes], optional): Использовать прокси для запросов. Defaults to None.
        """
        super().__init__(retries=retries)
        self._client = get_client(
            timeout=timeout,
            retries=retries,
            proxy=proxy,
        )

    def __enter__(self):
        return self

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

    def close(self):
        """Окончание сессии"""
        self._client.close()

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

    def _search(self, params: dict[str, Any]) -> Optional[SearchResponse]:
        r = self.request("get", "/api/geoportal/v2/search/geoportal", params=params)
        return self._validate_search_response(r)

    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]

    def search_by_theme(
        self, query: str, theme_id: ThemeId = ThemeId.REAL_ESTATE_OBJECTS
    ) -> Optional[NspdFeature]:
        """Глобальный поисковой запрос

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

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

    def search_by_layers(self, query: str, *layer_ids: int) -> Optional[NspdFeature]:
        """Поисковой запрос по указанным слоям

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

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

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

        Args:
            query (str): поисковой запрос
            layer_def (Type[Feat]): Определение слоя

        Returns:
            Optional[Feat]: валидированная модель слоя, если найдено
        """
        feature = self.search_by_layers(query, layer_def.layer_meta.layer_id)
        return self._cast_feature_to_layer_def(feature, layer_def)

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

    def search_many_zu(self, cns_string: str) -> list[Optional[Layer36048Feature]]:
        """Поиск всех ЗУ, содержащихся в строке"""
        cns = list(self.iter_cn(cns_string))
        features = asyncio_mock.gather(*[self.search_zu(cn) for cn in cns])
        return features

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

    def search_many_oks(self, cns_string: str) -> list[Optional[Layer36049Feature]]:
        """Поиск всех ОКС, содержащихся в строке"""
        cns = list(self.iter_cn(cns_string))
        features = asyncio_mock.gather(*[self.search_oks(cn) for cn in cns])
        return features

    @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 (Union[Polygon, MultiPolygon]): Геометрический объект с контуром
            category_ids (int): id категорий слоев
            epsg (int, optional): Система координат контура. Defaults to 4326.

        Returns:
            Optional[list[Feat]]: Список объектов, пересекающихся с контуром, если найден хоть один
        """
        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": {}}
                ],
            },
        }
        response = self._client.post(
            "/api/geoportal/v1/intersects",
            params={"typeIntersect": "fullObject"},
            json=payload,
        )
        if response.status_code == 500 and response.json()["code"] == 400004:
            raise TooBigContour
        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 (Union[Polygon, MultiPolygon]): Геометрический объект с контуром
            layer_def (Type[Feat]): Модель слоя
            epsg (int, optional): Система координат контура. Defaults to 4326.

        Returns:
            Optional[list[Feat]]: Список объектов, пересекающихся с контуром, если найден хоть один
        """
        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 (Union[Polygon, MultiPolygon]): Геометрический объект с контуром
            epsg (int, optional): Система координат контура. Defaults to 4326.

        Returns:
            Optional[list[Layer36048Feature]]: Список объектов, пересекающихся с контуром, если найден хоть один
        """
        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 (Union[Polygon, MultiPolygon]): Геометрический объект с контуром
            epsg (int, optional): Система координат контура. Defaults to 4326.

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

    def search_at_point(self, pt: Point, layer_id: int) -> Optional[list[NspdFeature]]:
        """Поиск объектов слоя в точке

        Args:
            pt (Point):
            layer_id (int):

        Returns:
            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 (Point):
            layer_def (Type[Feat]): Тип слоя

        Returns:
            Optional[list[Feat]]: Типизированный список объектов, если найдены
        """
        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)

    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
            )
            r.raise_for_status()
            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 typing import Any, Literal, Optional, Type, Union, cast	1✔
4
5	import mercantile	1✔
6	import numpy as np	1✔
7	from httpx import HTTPStatusError, RemoteProtocolError, Response	1✔
8	from httpx._types import QueryParamTypes	1✔
9	from shapely import MultiPolygon, Point, Polygon, to_geojson	1✔
10
11	from pynspd import asyncio_mock	1✔
12	from pynspd.client import BaseNspdClient, ProxyTypes, get_client	1✔
13	from pynspd.errors import TooBigContour	1✔
14	from pynspd.schemas import Layer36048Feature, Layer36049Feature, NspdFeature	1✔
15	from pynspd.schemas.feature import Feat	1✔
16	from pynspd.schemas.responses import (	1✔
17	NspdTabGroupResponse,
18	NspdTabResponse,
19	SearchResponse,
20	)
21	from pynspd.types.enums import ThemeId	1✔
22
23
24	def retry_on_http_error(func):	1✔
25	"""Декоратор для повторения запроса при ошибках запроса"""
26
27	@wraps(func)	1✔
28	def wrapper(self: "Nspd", args, *kwargs):	1✔
29	attempt = 0	1✔
30	while attempt <= self.retries:	1✔
31	try:	1✔
32	return func(self, args, *kwargs)	1✔
33	except HTTPStatusError as e:	1✔
34	if e.response.status_code < 500:	×
35	raise e	×
36	attempt += 1	×
37	if attempt > self.retries:	×
38	raise e	×
39	except RemoteProtocolError:	1✔
40	# Запрос иногда рандомно обрывается сервером, проходит при повторном запросе
41	pass	1✔
42
43	return wrapper	1✔
44
45
46	class Nspd(BaseNspdClient):	1✔
47	def __init__(	1✔
48	self,
49	timeout: Optional[int] = None,
50	retries: int = 10,
51	proxy: Optional[ProxyTypes] = None,
52	):
53	"""Клиент для НСПД
54
55	Usage:
56	>>> with pynspd.Nspd() as nspd:
57	>>> feat = nspd.search_zu("77:05:0001005:19")
58
59	Args:
60	timeout (Optional[int], optional): Время ожидания ответа. Defaults to None.
61	retries (int, optional): Количество попыток при неудачном запросе (таймаут или 5хх ошибки). Defaults to 10.
62	proxy (Optional[ProxyTypes], optional): Использовать прокси для запросов. Defaults to None.
63	"""
64	super().__init__(retries=retries)	1✔
65	self._client = get_client(	1✔
66	timeout=timeout,
67	retries=retries,
68	proxy=proxy,
69	)
70
71	def __enter__(self):	1✔
72	return self	1✔
73
74	def __exit__(self, *exc):	1✔
75	self.close()	1✔
76
77	def close(self):	1✔
78	"""Окончание сессии"""
79	self._client.close()	1✔
80
81	@retry_on_http_error	1✔
82	def request(
83	self, method: str, url: str, params: Optional[QueryParamTypes] = None
84	) -> Response:
85	"""Базовый запрос к api с обработкой стандартных ошибок от НСПД"""
86	r = self._client.request(method, url, params=params)
87	return r
88
89	def _search(self, params: dict[str, Any]) -> Optional[SearchResponse]:	1✔
90	r = self.request("get", "/api/geoportal/v2/search/geoportal", params=params)	1✔
91	return self._validate_search_response(r)	1✔
92
93	def _search_one(self, params: dict[str, Any]) -> Optional[NspdFeature]:	1✔
94	response = self._search(params)	1✔
95	if response is None:	1✔
96	return None	1✔
97	features = response.data.features	1✔
98	# иногда поиск багует и дает помимо нужного еще и рандомный результат	1✔
99	if len(features) > 1:	×
100	features = list(	×
101	filter(	×
102	lambda x: params["query"] in x.properties.model_dump_json(),
103	features,	1✔
104	)	1✔
105	)	1✔
106	if len(features) == 0:	×
107	return None	1✔
108	assert len(features) == 1
109	return features[0]	1✔
110		1✔
111	def search_by_theme(
112	self, query: str, theme_id: ThemeId = ThemeId.REAL_ESTATE_OBJECTS
113	) -> Optional[NspdFeature]:
114	"""Глобальный поисковой запрос
115
116	Args:	1✔
117	query (str): поисковой запрос	1✔
118	theme_id (int): вид объекта (кадастровое деление, объект недвижимости и т.д.)	1✔
119		1✔
120	Returns:
121	Optional[SearchResponse]: положительный ответ от сервиса, либо None, если ничего не найдено	1✔
122	"""
123	return self._search_one(
124	params={
125	"query": query,
126	"thematicSearchId": theme_id.value,
127	}
128	)
129
130	def search_by_layers(self, query: str, *layer_ids: int) -> Optional[NspdFeature]:
131	"""Поисковой запрос по указанным слоям
132
133	Args:	1✔
134	query (str): поисковой запрос
135	*layer_ids (int): id слоев, в которых будет производиться поиск
136
137	Returns:
138	Optional[SearchResponse]: положительный ответ от сервиса, либо None, если ничего не найдено
139	"""
140	return self._search_one(	1✔
141	params={
142	"query": query,
143	"layersId": layer_ids,
144	}
145	)
146
147	def search_by_model(self, query: str, layer_def: Type[Feat]) -> Optional[Feat]:
148	"""Поиск одного объекта по определению слоя
149
150	Args:	1✔
151	query (str): поисковой запрос
152	layer_def (Type[Feat]): Определение слоя
153
154	Returns:
155	Optional[Feat]: валидированная модель слоя, если найдено
156	"""
157	feature = self.search_by_layers(query, layer_def.layer_meta.layer_id)	1✔
158	return self._cast_feature_to_layer_def(feature, layer_def)
159
160	def search_zu(self, cn: str) -> Optional[Layer36048Feature]:
161	"""Поиск ЗУ по кадастровому номеру"""
162	layer_def = cast(
163	Type[Layer36048Feature], NspdFeature.by_title("Земельные участки из ЕГРН")
164	)
165	return self.search_by_model(cn, layer_def)
166
167	def search_many_zu(self, cns_string: str) -> list[Optional[Layer36048Feature]]:	1✔
168	"""Поиск всех ЗУ, содержащихся в строке"""	1✔
169	cns = list(self.iter_cn(cns_string))
170	features = asyncio_mock.gather(*[self.search_zu(cn) for cn in cns])	1✔
171	return features
172		1✔
173	def search_oks(self, cn: str) -> Optional[Layer36049Feature]:
174	"""Поиск ОКС по кадастровому номеру"""
175	layer_def = cast(Type[Layer36049Feature], NspdFeature.by_title("Здания"))	1✔
176	return self.search_by_model(cn, layer_def)
177		1✔
178	def search_many_oks(self, cns_string: str) -> list[Optional[Layer36049Feature]]:
179	"""Поиск всех ОКС, содержащихся в строке"""	1✔
180	cns = list(self.iter_cn(cns_string))	1✔
181	features = asyncio_mock.gather(*[self.search_oks(cn) for cn in cns])	1✔
182	return features
183		1✔
184	@retry_on_http_error
185	def search_in_contour(	1✔
186	self,	1✔
187	countour: Union[Polygon, MultiPolygon],
188	*category_ids: int,	1✔
189	epsg: int = 4326,
190	) -> Optional[list[NspdFeature]]:	1✔
191	"""Поиск объектов в контуре по id категорий слоев	1✔
192		1✔
193	Args:
194	countour (Union[Polygon, MultiPolygon]): Геометрический объект с контуром	1✔
195	category_ids (int): id категорий слоев	1✔
196	epsg (int, optional): Система координат контура. Defaults to 4326.
197
198	Returns:
199	Optional[list[Feat]]: Список объектов, пересекающихся с контуром, если найден хоть один
200	"""
201	feature_geojson = json.loads(to_geojson(countour))
202	feature_geojson["crs"] = {
203	"type": "name",
204	"properties": {"name": f"EPSG:{epsg}"},
205	}
206	payload = {
207	"categories": [{"id": id_} for id_ in category_ids],
208	"geom": {
209	"type": "FeatureCollection",
210	"features": [
211	{"geometry": feature_geojson, "type": "Feature", "properties": {}}	1✔
212	],	1✔
213	},
214	}
215	response = self._client.post(
216	"/api/geoportal/v1/intersects",	1✔
217	params={"typeIntersect": "fullObject"},
218	json=payload,
219	)
220	if response.status_code == 500 and response.json()["code"] == 400004:
221	raise TooBigContour
222	return self._validate_feature_collection_response(response)
223
224	def search_in_contour_by_model(
225	self,	1✔
226	countour: Union[Polygon, MultiPolygon],	1✔
227	layer_def: Type[Feat],
228	epsg: int = 4326,
229	) -> Optional[list[Feat]]:
230	"""Поиск объектов в контуре по определению слоя
231
232	Args:	1✔
233	countour (Union[Polygon, MultiPolygon]): Геометрический объект с контуром	1✔
234	layer_def (Type[Feat]): Модель слоя	1✔
235	epsg (int, optional): Система координат контура. Defaults to 4326.	×
236		1✔
237	Returns:
238	Optional[list[Feat]]: Список объектов, пересекающихся с контуром, если найден хоть один	1✔
239	"""
240	raw_features = self.search_in_contour(
241	countour, layer_def.layer_meta.category_id, epsg=epsg
242	)
243	return self._cast_features_to_layer_defs(raw_features, layer_def)
244
245	def search_zu_in_contour(
246	self, countour: Union[Polygon, MultiPolygon], epsg: int = 4326
247	) -> Optional[list[Layer36048Feature]]:
248	"""Поиск ЗУ в контуре
249
250	Args:
251	countour (Union[Polygon, MultiPolygon]): Геометрический объект с контуром
252	epsg (int, optional): Система координат контура. Defaults to 4326.
253
254	Returns:	1✔
255	Optional[list[Layer36048Feature]]: Список объектов, пересекающихся с контуром, если найден хоть один
256	"""
257	return self.search_in_contour_by_model(countour, Layer36048Feature, epsg=epsg)	1✔
258
259	def search_oks_in_contour(	1✔
260	self, countour: Union[Polygon, MultiPolygon], epsg: int = 4326
261	) -> Optional[list[Layer36049Feature]]:
262	"""Поиск ОКС в контуре
263
264	Args:
265	countour (Union[Polygon, MultiPolygon]): Геометрический объект с контуром
266	epsg (int, optional): Система координат контура. Defaults to 4326.
267
268	Returns:
269	Optional[list[Layer36048Feature]]: Список объектов, пересекающихся с контуром, если найден хоть один
270	"""
271	return self.search_in_contour_by_model(countour, Layer36049Feature, epsg=epsg)	1✔
272
273	def search_at_point(self, pt: Point, layer_id: int) -> Optional[list[NspdFeature]]:	1✔
274	"""Поиск объектов слоя в точке
275
276	Args:
277	pt (Point):
278	layer_id (int):
279
280	Returns:
281	Optional[list[NspdFeature]]: Список объектов, если найдены
282	"""
283	TILE_SIZE = 512
284	tile = mercantile.tile(
285	pt.x, pt.y, zoom=24	1✔
286	) # zoom=24 должно быть достаточно для самого точного совпадения
287	tile_bounds = mercantile.bounds(tile)	1✔
288	i = np.interp(pt.x, [tile_bounds.west, tile_bounds.east], [0, TILE_SIZE])	1✔
289	j = np.interp(pt.y, [tile_bounds.south, tile_bounds.north], [0, TILE_SIZE])
290	bbox = ",".join(
291	map(str, mercantile.xy_bounds(tile))
292	) # bbox в 3857, см. комментарий про CRS
293	params = {
294	"REQUEST": "GetFeatureInfo",
295	"SERVICE": "WMS",
296	"VERSION": "1.3.0",
297	"INFO_FORMAT": "application/json",
298	"FORMAT": "image/png",	1✔
299	"STYLES": "",	1✔
300	"TRANSPARENT": "true",
301	"QUERY_LAYERS": layer_id,
302	"LAYERS": layer_id,	1✔
303	"WIDTH": TILE_SIZE,	1✔
304	"HEIGHT": TILE_SIZE,	1✔
305	"I": int(i),	1✔
306	"J": TILE_SIZE - int(j), # отсчет координат для пикселей ведется сверху
307	"CRS": "EPSG:3857", # CRS для bbox
308	# можно указать и 4326, но тогда и геометрия будет в 4326	1✔
309	# Но в других методах мы всегда ждем 3857, поэтому оставляем
310	"BBOX": bbox,
311	"FEATURE_COUNT": "10", # Если не указать - вернет только один, даже если попало на границу
312	}
313	response = self.request("get", f"/api/aeggis/v3/{layer_id}/wms", params=params)
314	return self._validate_feature_collection_response(response)
315
316	def search_at_point_by_model(
317	self, pt: Point, layer_def: Type[Feat]
318	) -> Optional[list[Feat]]:
319	"""Поиск объектов слоя в точке (с типизацией)
320
321	Args:
322	pt (Point):
323	layer_def (Type[Feat]): Тип слоя
324
325	Returns:
326	Optional[list[Feat]]: Типизированный список объектов, если найдены
327	"""
328	raw_features = self.search_at_point(pt, layer_def.layer_meta.layer_id)	1✔
329	return self._cast_features_to_layer_defs(raw_features, layer_def)	1✔
330
331	def search_zu_at_point(self, pt: Point) -> Optional[list[Layer36048Feature]]:	1✔
332	"""Поиск ЗУ в точке"""
333	return self.search_at_point_by_model(pt, Layer36048Feature)
334
335	def search_oks_at_point(self, pt: Point) -> Optional[list[Layer36049Feature]]:
336	"""Поиск ОКС в точке"""
337	return self.search_at_point_by_model(pt, Layer36049Feature)
338
339	def _tab_request(
340	self, feat: NspdFeature, tab_class: str, type_: Literal["values", "group"]
341	) -> Optional[dict]:
342	if feat.properties.options.no_coords:
343	params = {	1✔
344	"tabClass": tab_class,	1✔
345	"objdocId": feat.properties.options.objdoc_id,
346	"registersId": feat.properties.options.registers_id,	1✔
347	}
348	else:	1✔
349	params = {
350	"tabClass": tab_class,	1✔
351	"categoryId": feat.properties.category,
352	"geomId": feat.id,	1✔
353	}
354	try:	1✔
355	r = self.request(	1✔
356	"get", f"/api/geoportal/v1/tab-{type_}-data", params=params
357	)
358	r.raise_for_status()	1✔
359	return r.json()	1✔
360	except HTTPStatusError as e:
361	if e.response.status_code == 404:
362	return None
363	raise e
364
365	def _tab_values_request(	1✔
366	self, feat: NspdFeature, tab_class: str
367	) -> Optional[list[str]]:
368	resp = self._tab_request(feat, tab_class, "values")
369	if resp is None:
370	return None	1✔
371	return NspdTabResponse.model_validate(resp).value	1✔
372
373	def _tab_groups_request(
374	self, feat: NspdFeature, tab_class: str	1✔
375	) -> Optional[dict[str, Optional[list[str]]]]:	1✔
376	resp = self._tab_request(feat, tab_class, "group")	1✔
377	if resp is None:	1✔
378	return None	×
379	item = NspdTabGroupResponse.model_validate(resp).object
380	data = {i.title: i.value for i in item}	1✔
381	if len(data) == 0:
382	return None
383	return data	1✔
384		1✔
385	def tab_land_parts(self, feat: NspdFeature) -> Optional[list[str]]:	1✔
386	"""Получение данных с вкладки \"Части ЗУ\" """	1✔
387	return self._tab_values_request(feat, "landParts")
388		1✔
389	def tab_land_links(self, feat: NspdFeature) -> Optional[list[str]]:
390	"""Получение данных с вкладки \"Связанные ЗУ\" """
391	return self._tab_values_request(feat, "landLinks")	1✔
392		1✔
393	def tab_permission_type(self, feat: NspdFeature) -> Optional[list[str]]:	×
394	"""Получение данных с вкладки \"Виды разрешенного использования\" """	1✔
395	return self._tab_values_request(feat, "permissionType")	1✔
396		1✔
397	def tab_composition_land(self, feat: NspdFeature) -> Optional[list[str]]:	×
398	"""Получение данных с вкладки \"Состав ЕЗП\" """	1✔
399	return self._tab_values_request(feat, "compositionLand")
400		1✔
401	def tab_build_parts(self, feat: NspdFeature) -> Optional[list[str]]:
402	"""Получение данных с вкладки \"Части ОКС\" """	1✔
403	return self._tab_values_request(feat, "buildParts")
404		1✔
405	def tab_objects_list(
406	self, feat: NspdFeature	1✔
407	) -> Optional[dict[str, Optional[list[str]]]]:
408	"""Получение данных с вкладки \"Объекты\" """	1✔
409	return self._tab_groups_request(feat, "objectsList")

yazmolod / pynspd / #3

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