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

95.03

/src/pynspd/_async/api.py

import asyncio
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.client import BaseNspdClient, ProxyTypes, get_async_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)
    async def wrapper(self: "AsyncNspd", *args, **kwargs):
        attempt = 0
        while attempt <= self.retries:
            try:
                return await 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 AsyncNspd(BaseNspdClient):
    def __init__(
        self,
        timeout: Optional[int] = None,
        retries: int = 10,
        proxy: Optional[ProxyTypes] = None,
    ):
        """Асинхронный клиент для НСПД

        Usage:
        >>> async with pynspd.AsyncNspd() as nspd:
        >>>     feat = await 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_async_client(
            timeout=timeout,
            retries=retries,
            proxy=proxy,
        )

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        await self.close()

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

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

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

    async def _search_one(self, params: dict[str, Any]) -> Optional[NspdFeature]:
        response = await 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]

    async 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 await self._search_one(
            params={
                "query": query,
                "thematicSearchId": theme_id.value,
            }
        )

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

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

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

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

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

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

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

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

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

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

    @retry_on_http_error
    async 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 = await 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)

    async 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 = await self.search_in_contour(
            countour, layer_def.layer_meta.category_id, epsg=epsg
        )
        return self._cast_features_to_layer_defs(raw_features, layer_def)

    async 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 await self.search_in_contour_by_model(
            countour, Layer36048Feature, epsg=epsg
        )

    async 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 await self.search_in_contour_by_model(
            countour, Layer36049Feature, epsg=epsg
        )

    async 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 = await self.request(
            "get", f"/api/aeggis/v3/{layer_id}/wms", params=params
        )
        return self._validate_feature_collection_response(response)

    async 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 = await self.search_at_point(pt, layer_def.layer_meta.layer_id)
        return self._cast_features_to_layer_defs(raw_features, layer_def)

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

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

    async 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 = await 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

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

    async def _tab_groups_request(
        self, feat: NspdFeature, tab_class: str
    ) -> Optional[dict[str, Optional[list[str]]]]:
        resp = await 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

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

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

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

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

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

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

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