yazmolod / pynspd / #11

Committed 04 Jun 2025 11:00PM UTC coverage: 97.585% (-0.9%) from 98.461%

Build # #11

Build Type

push

coveralls-python

Committed by yazmolod

Commit Message

chore: update docs

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

48 existing lines in 5 files now uncovered.

2829 of 2899 relevant lines covered (97.59%)

0.98 hits per line

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

98.22

/src/pynspd/cli.py

import re
from pathlib import Path
from typing import (
    Annotated,
    Any,
    Callable,
    Generator,
    Optional,
    Sequence,
    Type,
    TypeVar,
)

import geopandas as gpd
import typer
from pyogrio.errors import DataSourceError
from rich import print
from rich.progress import (
    BarColumn,
    Progress,
    SpinnerColumn,
    TaskProgressColumn,
    TextColumn,
    TimeRemainingColumn,
)
from shapely import GEOSException, MultiPolygon, Point, Polygon, from_wkt

from pynspd import Nspd, NspdFeature, __version__
from pynspd.errors import UnknownLayer
from pynspd.schemas import BaseFeature

app = typer.Typer(
    pretty_exceptions_show_locals=False,
    no_args_is_help=True,
    help="Утилита командной строки для поиска на НСПД",
)
CN_PATTERN = re.compile(r"\d+:\d+:\d+:\d+")


T = TypeVar("T")


OutputOption = Annotated[
    Optional[Path],
    typer.Option(
        "--output",
        "-o",
        help=(
            "Файл, в который будет сохранен результат поиска. "
            "Поддерживаются гео- (.gpkg, .geojson и пр.) и табличные форматы (.xlsx, .csv)"
        ),
        rich_help_panel="General Options",
    ),
]

LocalizeOption = Annotated[
    bool,
    typer.Option(
        "--localize",
        "-l",
        help="Использовать названия колонок с сайта, а не из оригинальные",
        rich_help_panel="General Options",
    ),
]

TabObjectsOption = Annotated[
    bool,
    typer.Option(
        "--tab-objects",
        help='Получить данные со вкладки "Объекты" для найденных объектов',
        rich_help_panel="General Options",
    ),
]


def define_cns(input_: str) -> list[str]:
    """Поиск массива к/н в исходной строке или текстовом файле"""
    if CN_PATTERN.match(input_):
        return list(Nspd.iter_cn(input_))
    maybe_file = Path(input_)
    if not maybe_file.exists():
        raise typer.BadParameter("Файл не найден")
    if maybe_file.suffix in (".txt",):
        cns = list(Nspd.iter_cn(maybe_file.read_text("utf-8")))
        if len(cns) == 0:
            raise typer.BadParameter("В файле не найдены кадастровые номера")
        return cns
    raise typer.BadParameter("Неподдерживаемый формат файла")


def define_geoms(input_: str) -> list[Point] | list[Polygon] | list[MultiPolygon]:
    """Проверка гео-файла и получение геометрии из него"""
    maybe_file = Path(input_)
    coords_pattern = re.compile(r"(\d{2,3}\.\d+),? *(\d{2,3}\.\d+)")
    if maybe_file.suffix == ".txt":
        pts_text = maybe_file.read_text("utf-8")
    else:
        pts_text = input_
    pts_text = pts_text.strip()
    if coords_pattern.match(pts_text):
        coords = [list(map(float, i)) for i in coords_pattern.findall(pts_text)]
        pts = [Point(*i[::-1]) for i in coords]
        geoseries = gpd.GeoSeries(pts, crs=4326)
    elif maybe_file.exists():
        try:
            geoseries = gpd.read_file(input_).geometry
        except DataSourceError as e:
            raise typer.BadParameter(
                f"{input_} не является поддерживаемым файлом"
            ) from e
    else:
        try:
            wkt_geom = from_wkt(input_)
            geoseries = gpd.GeoSeries([wkt_geom], crs=4326)
        except GEOSException as e:
            raise typer.BadParameter(
                f"Файл {input_} не существует или не является валидным WKT"
            ) from e
    geom_types = geoseries.geom_type.unique().tolist()
    geometry = geoseries.tolist()
    if len(geom_types) > 1:
        raise typer.BadParameter("Не поддерживаются файла со смешанной геометрией")
    if geom_types[0] not in ("Point", "Polygon", "MultiPolygon"):
        raise typer.BadParameter(f"Не поддерживаемый тип геометрии - {geom_types[0]}")
    return geometry


def define_layer_def(layer_name: str) -> Type[BaseFeature]:
    """Определение типа слоя"""
    try:
        return NspdFeature.by_title(layer_name)
    except UnknownLayer as e:
        raise typer.BadParameter(f"{layer_name} не является слоем НСПД") from e


def _progress_iter(items: Sequence[T]) -> Generator[T, None, None]:
    cols = [
        SpinnerColumn(),
        TextColumn("{task.description}"),
    ]
    if len(items) > 1:
        cols += [
            BarColumn(),
            TaskProgressColumn(),
            TimeRemainingColumn(),
        ]
    with Progress(
        *cols,
        transient=True,
    ) as progress:
        task = progress.add_task(description="Поиск...", total=len(items))
        for item in items:
            yield item
            progress.update(task, advance=1)


def _get_features_from_list(
    client: Nspd, cns: list[str]
) -> Optional[list[NspdFeature]]:
    features = []
    missing = []
    for cn in _progress_iter(cns):
        feat = client.find(cn)
        if feat is None:
            missing.append(cn)
            continue
        features.append(feat)
    if missing:
        print(
            f":warning-emoji: [orange3] Не найдены {len(missing)} из {len(cns)} объектов:"
        )
        for m in missing:
            print(f"[orange3]   - {m}")
        print()
    if len(features) == 0:
        return None
    return features


def _get_features_from_geom(
    method: Callable[
        [Point | Polygon | MultiPolygon, Type[BaseFeature]], Optional[list[BaseFeature]]
    ],
    geoms: list[Point | Polygon | MultiPolygon],
    layer_def: Type[BaseFeature],
) -> Optional[list[BaseFeature]]:
    features = []
    missing_count = 0
    for geom in _progress_iter(geoms):
        feats = method(geom, layer_def)
        if feats is None:
            missing_count += 1
            continue
        features += feats
    if missing_count:
        print(
            f":warning-emoji: [orange3] Ничего не найдено в {missing_count} из {len(geoms)} локаций"
        )
        print()
    if len(features) == 0:
        return None
    return features


def _get_tab_object(client: Nspd, features: list[NspdFeature]) -> list[NspdFeature]:
    """Получение данных из вкладки"""
    for feat in _progress_iter(features):
        data = client.tab_objects_list(feat)
        if data is None:
            continue
        assert feat.properties.options.model_extra is not None
        feat.properties.options.model_extra["tab"] = data
    return features


def prepare_features(features: list[NspdFeature], localize: bool) -> gpd.GeoDataFrame:
    prepared_features: list = []
    for feat in features:
        if isinstance(feat, NspdFeature):
            feat = feat.cast()
        assert feat.properties.options.model_extra is not None
        props: dict[str, Any] = feat.properties.options.model_extra.pop("tab", {})
        if localize:
            props.update(feat.properties.options.model_dump_human_readable())
            props["Категория"] = feat.properties.category_name
        else:
            props.update(feat.properties.options.model_dump())
            props["category"] = feat.properties.category_name
        props["geometry"] = feat.geometry.to_shape()
        prepared_features.append(props)
    return gpd.GeoDataFrame(prepared_features, crs=4326).fillna("")


def process_output(
    features: Optional[list[NspdFeature]], output: Optional[Path], localize: bool
) -> None:
    if features is None:
        print("[red]Ничего не найдено")
        raise typer.Abort()
    gdf = prepare_features(features, localize)
    if output is None:
        print(gdf.T)
        return
    elif output.suffix == ".xlsx":
        gdf.to_excel(output, index=False)
    elif output.suffix == ".csv":
        gdf.to_csv(output, index=False)
    else:
        gdf.to_file(output)
    print(f"[green]Найдено {len(gdf)} объектов, сохранено в файл {output.resolve()}[/]")


def version_callback(version: bool = False) -> None:
    if version:
        print(__version__)
        raise typer.Exit()


@app.callback()
def common(
    version: Annotated[
        bool,
        typer.Option(
            "-v",
            "--version",
            help="Show current version",
            is_eager=True,
            callback=version_callback,
        ),
    ] = False,
):
    pass


@app.command(no_args_is_help=True)
def geo(
    input: Annotated[
        str,
        typer.Argument(
            help="Путь к файлу с геоданными, координаты точек или WKT",
            show_default=False,
        ),
    ],
    layer_name: Annotated[
        str,
        typer.Argument(help="Имя слоя с НСПД, в котором нужно производить поиск"),
    ] = "Земельные участки из ЕГРН",
    output: OutputOption = None,
    localize: LocalizeOption = False,
    add_tab_object: TabObjectsOption = False,
) -> None:
    """Поиск объектов по геоданным"""
    geoms = define_geoms(input)
    layer_def = define_layer_def(layer_name)
    with Nspd() as client:
        if isinstance(geoms[0], Point):
            features = _get_features_from_geom(client.search_at_point, geoms, layer_def)
        else:
            features = _get_features_from_geom(
                client.search_in_contour, geoms, layer_def
            )
        if features and add_tab_object:
            _get_tab_object(client, features)
    process_output(features, output, localize)


@app.command(no_args_is_help=True)
def find(
    input: Annotated[
        str,
        typer.Argument(
            help="Список искомых к/н. Может быть текстовым файлом", show_default=False
        ),
    ],
    output: OutputOption = None,
    localize: LocalizeOption = False,
    add_tab_object: TabObjectsOption = False,
) -> None:
    """Поиск объектов по списку к/н"""
    cns = define_cns(input)
    with Nspd() as client:
        features = _get_features_from_list(client, cns)
        if features and add_tab_object:
            features = _get_tab_object(client, features)
    process_output(features, output, localize)


def main() -> Any:
    return app()

1	import re	1✔
2	from pathlib import Path	1✔
3	from typing import (	1✔
4	Annotated,
5	Any,
6	Callable,
7	Generator,
8	Optional,
9	Sequence,
10	Type,
11	TypeVar,
12	)
13
14	import geopandas as gpd	1✔
15	import typer	1✔
16	from pyogrio.errors import DataSourceError	1✔
17	from rich import print	1✔
18	from rich.progress import (	1✔
19	BarColumn,
20	Progress,
21	SpinnerColumn,
22	TaskProgressColumn,
23	TextColumn,
24	TimeRemainingColumn,
25	)
26	from shapely import GEOSException, MultiPolygon, Point, Polygon, from_wkt	1✔
27
28	from pynspd import Nspd, NspdFeature, __version__	1✔
29	from pynspd.errors import UnknownLayer	1✔
30	from pynspd.schemas import BaseFeature	1✔
31
32	app = typer.Typer(	1✔
33	pretty_exceptions_show_locals=False,
34	no_args_is_help=True,
35	help="Утилита командной строки для поиска на НСПД",
36	)
37	CN_PATTERN = re.compile(r"\d+:\d+:\d+:\d+")	1✔
38
39
40	T = TypeVar("T")	1✔
41
42
43	OutputOption = Annotated[	1✔
44	Optional[Path],
45	typer.Option(
46	"--output",
47	"-o",
48	help=(
49	"Файл, в который будет сохранен результат поиска. "
50	"Поддерживаются гео- (.gpkg, .geojson и пр.) и табличные форматы (.xlsx, .csv)"
51	),
52	rich_help_panel="General Options",
53	),
54	]
55
56	LocalizeOption = Annotated[	1✔
57	bool,
58	typer.Option(
59	"--localize",
60	"-l",
61	help="Использовать названия колонок с сайта, а не из оригинальные",
62	rich_help_panel="General Options",
63	),
64	]
65
66	TabObjectsOption = Annotated[	1✔
67	bool,
68	typer.Option(
69	"--tab-objects",
70	help='Получить данные со вкладки "Объекты" для найденных объектов',
71	rich_help_panel="General Options",
72	),
73	]
74
75
76	def define_cns(input_: str) -> list[str]:	1✔
77	"""Поиск массива к/н в исходной строке или текстовом файле"""
78	if CN_PATTERN.match(input_):	1✔
79	return list(Nspd.iter_cn(input_))	1✔
80	maybe_file = Path(input_)	1✔
81	if not maybe_file.exists():	1✔
82	raise typer.BadParameter("Файл не найден")	1✔
83	if maybe_file.suffix in (".txt",):	1✔
84	cns = list(Nspd.iter_cn(maybe_file.read_text("utf-8")))	1✔
85	if len(cns) == 0:	1✔
86	raise typer.BadParameter("В файле не найдены кадастровые номера")	1✔
87	return cns	1✔
88	raise typer.BadParameter("Неподдерживаемый формат файла")	1✔
89
90
91	def define_geoms(input_: str) -> list[Point] \| list[Polygon] \| list[MultiPolygon]:	1✔
92	"""Проверка гео-файла и получение геометрии из него"""
93	maybe_file = Path(input_)	1✔
94	coords_pattern = re.compile(r"(\d{2,3}\.\d+),? *(\d{2,3}\.\d+)")	1✔
95	if maybe_file.suffix == ".txt":	1✔
96	pts_text = maybe_file.read_text("utf-8")	1✔
97	else:
98	pts_text = input_	1✔
99	pts_text = pts_text.strip()	1✔
100	if coords_pattern.match(pts_text):	1✔
101	coords = [list(map(float, i)) for i in coords_pattern.findall(pts_text)]	1✔
102	pts = [Point(*i[::-1]) for i in coords]	1✔
103	geoseries = gpd.GeoSeries(pts, crs=4326)	1✔
104	elif maybe_file.exists():	1✔
105	try:	1✔
106	geoseries = gpd.read_file(input_).geometry	1✔
107	except DataSourceError as e:	1✔
108	raise typer.BadParameter(	1✔
109	f"{input_} не является поддерживаемым файлом"
110	) from e
111	else:
112	try:	1✔
113	wkt_geom = from_wkt(input_)	1✔
114	geoseries = gpd.GeoSeries([wkt_geom], crs=4326)	1✔
115	except GEOSException as e:	1✔
116	raise typer.BadParameter(	1✔
117	f"Файл {input_} не существует или не является валидным WKT"
118	) from e
119	geom_types = geoseries.geom_type.unique().tolist()	1✔
120	geometry = geoseries.tolist()	1✔
121	if len(geom_types) > 1:	1✔
UNCOV 122	raise typer.BadParameter("Не поддерживаются файла со смешанной геометрией")	×
123	if geom_types[0] not in ("Point", "Polygon", "MultiPolygon"):	1✔
124	raise typer.BadParameter(f"Не поддерживаемый тип геометрии - {geom_types[0]}")	1✔
125	return geometry	1✔
126
127
128	def define_layer_def(layer_name: str) -> Type[BaseFeature]:	1✔
129	"""Определение типа слоя"""
130	try:	1✔
131	return NspdFeature.by_title(layer_name)	1✔
132	except UnknownLayer as e:	1✔
133	raise typer.BadParameter(f"{layer_name} не является слоем НСПД") from e	1✔
134
135
136	def _progress_iter(items: Sequence[T]) -> Generator[T, None, None]:	1✔
137	cols = [	1✔
138	SpinnerColumn(),
139	TextColumn("{task.description}"),
140	]
141	if len(items) > 1:	1✔
142	cols += [	1✔
143	BarColumn(),
144	TaskProgressColumn(),
145	TimeRemainingColumn(),
146	]
147	with Progress(	1✔
148	*cols,
149	transient=True,
150	) as progress:
151	task = progress.add_task(description="Поиск...", total=len(items))	1✔
152	for item in items:	1✔
153	yield item	1✔
154	progress.update(task, advance=1)	1✔
155
156
157	def _get_features_from_list(	1✔
158	client: Nspd, cns: list[str]
159	) -> Optional[list[NspdFeature]]:
160	features = []	1✔
161	missing = []	1✔
162	for cn in _progress_iter(cns):	1✔
163	feat = client.find(cn)	1✔
164	if feat is None:	1✔
165	missing.append(cn)	1✔
166	continue	1✔
167	features.append(feat)	1✔
168	if missing:	1✔
169	print(	1✔
170	f":warning-emoji: [orange3] Не найдены {len(missing)} из {len(cns)} объектов:"
171	)
172	for m in missing:	1✔
173	print(f"[orange3] - {m}")	1✔
174	print()	1✔
175	if len(features) == 0:	1✔
176	return None	1✔
177	return features	1✔
178
179
180	def _get_features_from_geom(	1✔
181	method: Callable[
182	[Point \| Polygon \| MultiPolygon, Type[BaseFeature]], Optional[list[BaseFeature]]
183	],
184	geoms: list[Point \| Polygon \| MultiPolygon],
185	layer_def: Type[BaseFeature],
186	) -> Optional[list[BaseFeature]]:
187	features = []	1✔
188	missing_count = 0	1✔
189	for geom in _progress_iter(geoms):	1✔
190	feats = method(geom, layer_def)	1✔
191	if feats is None:	1✔
192	missing_count += 1	1✔
193	continue	1✔
194	features += feats	1✔
195	if missing_count:	1✔
196	print(	1✔
197	f":warning-emoji: [orange3] Ничего не найдено в {missing_count} из {len(geoms)} локаций"
198	)
199	print()	1✔
200	if len(features) == 0:	1✔
201	return None	1✔
202	return features	1✔
203
204
205	def _get_tab_object(client: Nspd, features: list[NspdFeature]) -> list[NspdFeature]:	1✔
206	"""Получение данных из вкладки"""
207	for feat in _progress_iter(features):	1✔
208	data = client.tab_objects_list(feat)	1✔
209	if data is None:	1✔
UNCOV 210	continue	×
211	assert feat.properties.options.model_extra is not None	1✔
212	feat.properties.options.model_extra["tab"] = data	1✔
213	return features	1✔
214
215
216	def prepare_features(features: list[NspdFeature], localize: bool) -> gpd.GeoDataFrame:	1✔
217	prepared_features: list = []	1✔
218	for feat in features:	1✔
219	if isinstance(feat, NspdFeature):	1✔
220	feat = feat.cast()	1✔
221	assert feat.properties.options.model_extra is not None	1✔
222	props: dict[str, Any] = feat.properties.options.model_extra.pop("tab", {})	1✔
223	if localize:	1✔
224	props.update(feat.properties.options.model_dump_human_readable())	1✔
225	props["Категория"] = feat.properties.category_name	1✔
226	else:
227	props.update(feat.properties.options.model_dump())	1✔
228	props["category"] = feat.properties.category_name	1✔
229	props["geometry"] = feat.geometry.to_shape()	1✔
230	prepared_features.append(props)	1✔
231	return gpd.GeoDataFrame(prepared_features, crs=4326).fillna("")	1✔
232
233
234	def process_output(	1✔
235	features: Optional[list[NspdFeature]], output: Optional[Path], localize: bool
236	) -> None:
237	if features is None:	1✔
238	print("[red]Ничего не найдено")	1✔
239	raise typer.Abort()	1✔
240	gdf = prepare_features(features, localize)	1✔
241	if output is None:	1✔
242	print(gdf.T)	1✔
243	return	1✔
244	elif output.suffix == ".xlsx":	1✔
245	gdf.to_excel(output, index=False)	1✔
246	elif output.suffix == ".csv":	1✔
247	gdf.to_csv(output, index=False)	1✔
248	else:
249	gdf.to_file(output)	1✔
250	print(f"[green]Найдено {len(gdf)} объектов, сохранено в файл {output.resolve()}[/]")	1✔
251
252
253	def version_callback(version: bool = False) -> None:	1✔
254	if version:	1✔
255	print(__version__)	1✔
256	raise typer.Exit()	1✔
257
258
259	@app.callback()	1✔
260	def common(	1✔
261	version: Annotated[
262	bool,
263	typer.Option(
264	"-v",
265	"--version",
266	help="Show current version",
267	is_eager=True,
268	callback=version_callback,
269	),
270	] = False,
271	):
272	pass	1✔
273
274
275	@app.command(no_args_is_help=True)	1✔
276	def geo(	1✔
277	input: Annotated[
278	str,
279	typer.Argument(
280	help="Путь к файлу с геоданными, координаты точек или WKT",
281	show_default=False,
282	),
283	],
284	layer_name: Annotated[
285	str,
286	typer.Argument(help="Имя слоя с НСПД, в котором нужно производить поиск"),
287	] = "Земельные участки из ЕГРН",
288	output: OutputOption = None,
289	localize: LocalizeOption = False,
290	add_tab_object: TabObjectsOption = False,
291	) -> None:
292	"""Поиск объектов по геоданным"""
293	geoms = define_geoms(input)	1✔
294	layer_def = define_layer_def(layer_name)	1✔
295	with Nspd() as client:	1✔
296	if isinstance(geoms[0], Point):	1✔
297	features = _get_features_from_geom(client.search_at_point, geoms, layer_def)	1✔
298	else:
299	features = _get_features_from_geom(	1✔
300	client.search_in_contour, geoms, layer_def
301	)
302	if features and add_tab_object:	1✔
303	_get_tab_object(client, features)	1✔
304	process_output(features, output, localize)	1✔
305
306
307	@app.command(no_args_is_help=True)	1✔
308	def find(	1✔
309	input: Annotated[
310	str,
311	typer.Argument(
312	help="Список искомых к/н. Может быть текстовым файлом", show_default=False
313	),
314	],
315	output: OutputOption = None,
316	localize: LocalizeOption = False,
317	add_tab_object: TabObjectsOption = False,
318	) -> None:
319	"""Поиск объектов по списку к/н"""
320	cns = define_cns(input)	1✔
321	with Nspd() as client:	1✔
322	features = _get_features_from_list(client, cns)	1✔
323	if features and add_tab_object:	1✔
324	features = _get_tab_object(client, features)	1✔
325	process_output(features, output, localize)	1✔
326
327
328	def main() -> Any:	1✔
UNCOV 329	return app()	×

yazmolod / pynspd / #11

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