yazmolod / pynspd / #12

Committed 28 Jul 2025 06:51PM UTC coverage: 95.698% (-1.9%) from 97.585%

Build # #12

Build Type

push

coveralls-python

Committed by

yazmolod

Commit Message

bump version

Run Details

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

71 existing lines in 5 files now uncovered.

2825 of 2952 relevant lines covered (95.7%)

0.96 hits per line

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

78.43

/src/pynspd/cli.py

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

import geopandas as gpd
import questionary
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.map_types._autogen_layers import LayerTitle
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",
    ),
]

PlainOption = Annotated[
    bool,
    typer.Option(
        "--plain",
        "-p",
        help="Не конвертировать входные данные в список к/н",
        rich_help_panel="General Options",
    ),
]

ChooseLayersOption = Annotated[
    bool,
    typer.Option(
        "--choose-layer",
        "-c",
        help="Выбрать слои из списка вместо слоя по умолчанию",
        rich_help_panel="General Options",
    ),
]


def define_queries(input_: str, plain_query: bool) -> list[str]:
    """Поиск массива к/н в исходной строке или текстовом файле"""
    maybe_file = Path(input_)
    if maybe_file.exists():
        try:
            with open(maybe_file, encoding="utf-8", mode="r") as file:
                content = file.read()
        except UnicodeDecodeError:
            raise typer.BadParameter(f"Невозможно прочитать файл {input_}")
        if plain_query:
            return content.splitlines()
        cns = list(Nspd.iter_cn(content))
        if len(cns) == 0:
            raise typer.BadParameter(
                "В файле не найдены кадастровые номера. Используйте флаг --plain / -p для поиска по обычному тексту"
            )
        return cns
    if plain_query:
        return [input_]
    if not CN_PATTERN.match(input_):
        raise typer.BadParameter(
            "Нужно ввести валидные кадастровые номера или использовать флаг --plain / -p для обычного текста"
        )
    return list(Nspd.iter_cn(input_))


def define_geoms(input_: str) -> list[Point] | list[Polygon] | list[MultiPolygon]:
    """Проверка гео-файла и получение геометрии из него"""
    maybe_file = Path(input_)
    if maybe_file.exists() and maybe_file.suffix == ".txt":
        pts_text = maybe_file.read_text("utf-8")
    else:
        pts_text = input_
    pts_text = pts_text.strip()
    matches = re.findall(r"(\d{2,3}\.\d+),? *(\d{2,3}\.\d+)", pts_text)
    if len(matches) > 0:
        coords = [list(map(float, i)) for i in matches]
        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,
    queries: list[str],
    layer_defs: Optional[list[Type[BaseFeature]]],
) -> Optional[list[NspdFeature]]:
    features = []
    missing = []
    for query in _progress_iter(queries):
        if layer_defs is None:
            feats = client.search(query)
        else:
            feats = client.search_in_layers(query, *layer_defs)
        if feats is None:
            missing.append(query)
            continue
        features += feats
    if missing:
        print(
            f":warning-emoji: [orange3] Не найдены {len(missing)} из {len(queries)} объектов:"
        )
        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 = []
    unknown_layers = Counter()
    for feat in features:
        assert feat.properties.options.model_extra is not None
        props: dict[str, Any] = feat.properties.options.model_extra.pop("tab", {})
        if localize:
            try:
                feat = feat.cast()
            except UnknownLayer:
                unknown_layers[feat.properties.category_name] += 1
                continue
            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)
    for category, count in unknown_layers.items():
        print(
            f'[orange3] Было пропущено объектов с неизвестным слоем "{category}" - {count} (невозможно локализировать данные)'
        )
    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)
    try:
        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)
    except PermissionError:
        _ = typer.prompt(
            f"Файл {output} открыт в другой программе. Закройте его и нажмите любую клавишу",
            default=True,
            hide_input=True,
            prompt_suffix="...",
            show_default=False,
        )
        return process_output(features, output, localize)
    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,
        ),
    ],
    choose_layer: ChooseLayersOption = False,
    output: OutputOption = None,
    localize: LocalizeOption = False,
    add_tab_object: TabObjectsOption = False,
    _test_layer_name: Annotated[Optional[str], typer.Option(hidden=True)] = None,
) -> None:
    """Поиск объектов по геоданным

    Может производить поиск по файлам gpkg, geeojson, координатам точек или WKT-строке.
    По умолчанию ищет в слое "Земельные участки из ЕГРН".
    """
    if choose_layer:
        layer_name = (
            _test_layer_name
            if _test_layer_name is not None
            else questionary.select(
                "Выберите слой: ", choices=get_args(LayerTitle)
            ).ask()
        )
        if layer_name is None:
            raise typer.Abort()
    else:
        layer_name = "Земельные участки из ЕГРН"
    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 search(
    input: Annotated[
        str,
        typer.Argument(
            help="Поисковой запрос. Может быть многострочным текстовым файлом",
            show_default=False,
        ),
    ],
    choose_layer: ChooseLayersOption = False,
    plain_query: PlainOption = False,
    output: OutputOption = None,
    localize: LocalizeOption = False,
    add_tab_object: TabObjectsOption = False,
    _test_layer_names: Annotated[Optional[list[str]], typer.Option(hidden=True)] = None,
) -> None:
    """Поиск объектов по тексту

    По умолчанию разбивает запрос на кадастровые номера и ищет в объектах недвижимости.
    """
    if choose_layer:
        layer_names = (
            _test_layer_names
            if _test_layer_names is not None
            else questionary.checkbox(
                "Выберите слои: ",
                choices=get_args(LayerTitle),
                validate=lambda x: "Не выбрано ни одно значение"
                if len(x) == 0
                else True,
            ).ask()
        )
        if layer_names is None:
            raise typer.Abort()
        layer_defs = [define_layer_def(layer_name) for layer_name in layer_names]
    else:
        layer_defs = None
    queries = define_queries(input, plain_query)
    with Nspd() as client:
        features = _get_features_from_list(client, queries, layer_defs)
        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 collections import Counter	1✔
3	from pathlib import Path	1✔
UNCOV 4	from typing import (	×
5	Annotated,
6	Any,
7	Callable,
8	Generator,
9	Optional,
10	Sequence,
11	Type,
12	TypeVar,
13	get_args,
14	)
15
16	import geopandas as gpd	1✔
17	import questionary	1✔
18	import typer	1✔
19	from pyogrio.errors import DataSourceError	1✔
20	from rich import print	1✔
21	from rich.progress import (	1✔
22	BarColumn,
23	Progress,
24	SpinnerColumn,
25	TaskProgressColumn,
26	TextColumn,
27	TimeRemainingColumn,
28	)
UNCOV 29	from shapely import GEOSException, MultiPolygon, Point, Polygon, from_wkt	×
30
31	from pynspd import Nspd, NspdFeature, __version__	1✔
32	from pynspd.errors import UnknownLayer	1✔
33	from pynspd.map_types._autogen_layers import LayerTitle	1✔
UNCOV 34	from pynspd.schemas import BaseFeature	×
35
36	app = typer.Typer(	1✔
37	pretty_exceptions_show_locals=False,
38	no_args_is_help=True,
39	help="Утилита командной строки для поиска на НСПД",
40	)
UNCOV 41	CN_PATTERN = re.compile(r"\d+:\d+:\d+:\d+")	×
42
43
UNCOV 44	T = TypeVar("T")	×
45
46
47	OutputOption = Annotated[	1✔
48	Optional[Path],
49	typer.Option(
50	"--output",
51	"-o",
52	help=(
53	"Файл, в который будет сохранен результат поиска. "
54	"Поддерживаются гео- (.gpkg, .geojson и пр.) и табличные форматы (.xlsx, .csv)"
55	),
56	rich_help_panel="General Options",
57	),
58	]
59
60	LocalizeOption = Annotated[	1✔
61	bool,
62	typer.Option(
63	"--localize",
64	"-l",
65	help="Использовать названия колонок с сайта, а не из оригинальные",
66	rich_help_panel="General Options",
67	),
68	]
69
70	TabObjectsOption = Annotated[	1✔
71	bool,
72	typer.Option(
73	"--tab-objects",
74	help='Получить данные со вкладки "Объекты" для найденных объектов',
75	rich_help_panel="General Options",
76	),
77	]
78
79	PlainOption = Annotated[	1✔
80	bool,
81	typer.Option(
82	"--plain",
83	"-p",
84	help="Не конвертировать входные данные в список к/н",
85	rich_help_panel="General Options",
86	),
87	]
88
89	ChooseLayersOption = Annotated[	1✔
90	bool,
91	typer.Option(
92	"--choose-layer",
93	"-c",
94	help="Выбрать слои из списка вместо слоя по умолчанию",
95	rich_help_panel="General Options",
96	),
97	]
98
99
UNCOV 100	def define_queries(input_: str, plain_query: bool) -> list[str]:	×
101	"""Поиск массива к/н в исходной строке или текстовом файле"""
102	maybe_file = Path(input_)	1✔
103	if maybe_file.exists():	1✔
104	try:	1✔
105	with open(maybe_file, encoding="utf-8", mode="r") as file:	1✔
106	content = file.read()	1✔
107	except UnicodeDecodeError:	1✔
108	raise typer.BadParameter(f"Невозможно прочитать файл {input_}")	1✔
UNCOV 109	if plain_query:	×
110	return content.splitlines()	1✔
111	cns = list(Nspd.iter_cn(content))	1✔
112	if len(cns) == 0:	1✔
113	raise typer.BadParameter(	1✔
114	"В файле не найдены кадастровые номера. Используйте флаг --plain / -p для поиска по обычному тексту"
115	)
116	return cns	1✔
117	if plain_query:	1✔
118	return [input_]	1✔
119	if not CN_PATTERN.match(input_):	1✔
120	raise typer.BadParameter(	1✔
121	"Нужно ввести валидные кадастровые номера или использовать флаг --plain / -p для обычного текста"
122	)
UNCOV 123	return list(Nspd.iter_cn(input_))	×
124
125
UNCOV 126	def define_geoms(input_: str) -> list[Point] \| list[Polygon] \| list[MultiPolygon]:	×
127	"""Проверка гео-файла и получение геометрии из него"""
128	maybe_file = Path(input_)	1✔
129	if maybe_file.exists() and maybe_file.suffix == ".txt":	1✔
130	pts_text = maybe_file.read_text("utf-8")	1✔
131	else:
132	pts_text = input_	1✔
133	pts_text = pts_text.strip()	1✔
134	matches = re.findall(r"(\d{2,3}\.\d+),? *(\d{2,3}\.\d+)", pts_text)	1✔
135	if len(matches) > 0:	1✔
136	coords = [list(map(float, i)) for i in matches]	1✔
137	pts = [Point(*i[::-1]) for i in coords]	1✔
138	geoseries = gpd.GeoSeries(pts, crs=4326)	1✔
139	elif maybe_file.exists():	1✔
140	try:	1✔
141	geoseries = gpd.read_file(input_).geometry	1✔
142	except DataSourceError as e:	1✔
143	raise typer.BadParameter(	1✔
144	f"{input_} не является поддерживаемым файлом"
145	) from e
146	else:
147	try:	1✔
148	wkt_geom = from_wkt(input_)	1✔
149	geoseries = gpd.GeoSeries([wkt_geom], crs=4326)	1✔
150	except GEOSException as e:	1✔
151	raise typer.BadParameter(	1✔
152	f"Файл {input_} не существует или не является валидным WKT"
153	) from e
154	geom_types = geoseries.geom_type.unique().tolist()	1✔
155	geometry = geoseries.tolist()	1✔
UNCOV 156	if len(geom_types) > 1:	×
157	raise typer.BadParameter("Не поддерживаются файла со смешанной геометрией")	1✔
158	if geom_types[0] not in ("Point", "Polygon", "MultiPolygon"):	1✔
159	raise typer.BadParameter(f"Не поддерживаемый тип геометрии - {geom_types[0]}")	1✔
UNCOV 160	return geometry	×
161
162
UNCOV 163	def define_layer_def(layer_name: str) -> Type[BaseFeature]:	×
164	"""Определение типа слоя"""
165	try:	1✔
166	return NspdFeature.by_title(layer_name)	1✔
167	except UnknownLayer as e:	1✔
UNCOV 168	raise typer.BadParameter(f"{layer_name} не является слоем НСПД") from e	×
169
170
171	def _progress_iter(items: Sequence[T]) -> Generator[T, None, None]:	1✔
172	cols = [	1✔
173	SpinnerColumn(),
174	TextColumn("{task.description}"),
175	]
176	if len(items) > 1:	1✔
177	cols += [	1✔
178	BarColumn(),
179	TaskProgressColumn(),
180	TimeRemainingColumn(),
181	]
182	with Progress(	1✔
183	*cols,
184	transient=True,
185	) as progress:
186	task = progress.add_task(description="Поиск...", total=len(items))	1✔
187	for item in items:	1✔
188	yield item	1✔
UNCOV 189	progress.update(task, advance=1)	×
190
191
192	def _get_features_from_list(	1✔
193	client: Nspd,
194	queries: list[str],
195	layer_defs: Optional[list[Type[BaseFeature]]],
196	) -> Optional[list[NspdFeature]]:
197	features = []	1✔
198	missing = []	1✔
199	for query in _progress_iter(queries):	1✔
200	if layer_defs is None:	1✔
UNCOV 201	feats = client.search(query)	×
202	else:
203	feats = client.search_in_layers(query, *layer_defs)	1✔
204	if feats is None:	1✔
205	missing.append(query)	1✔
206	continue	1✔
207	features += feats	1✔
208	if missing:	1✔
209	print(	1✔
210	f":warning-emoji: [orange3] Не найдены {len(missing)} из {len(queries)} объектов:"
211	)
212	for m in missing:	1✔
213	print(f"[orange3] - {m}")	1✔
UNCOV 214	print()	×
215	if len(features) == 0:	1✔
216	return None	1✔
217	return features	1✔
218
219
220	def _get_features_from_geom(	1✔
221	method: Callable[
222	[Point \| Polygon \| MultiPolygon, Type[BaseFeature]], Optional[list[BaseFeature]]
223	],
224	geoms: list[Point \| Polygon \| MultiPolygon],
225	layer_def: Type[BaseFeature],
226	) -> Optional[list[BaseFeature]]:
227	features = []	1✔
228	missing_count = 0	1✔
229	for geom in _progress_iter(geoms):	1✔
230	feats = method(geom, layer_def)	1✔
231	if feats is None:	1✔
232	missing_count += 1	1✔
233	continue	1✔
234	features += feats	1✔
235	if missing_count:	1✔
236	print(	1✔
237	f":warning-emoji: [orange3] Ничего не найдено в {missing_count} из {len(geoms)} локаций"
238	)
239	print()	1✔
240	if len(features) == 0:	1✔
241	return None	1✔
242	return features	1✔
243
244
245	def _get_tab_object(client: Nspd, features: list[NspdFeature]) -> list[NspdFeature]:	1✔
246	"""Получение данных из вкладки"""
247	for feat in _progress_iter(features):	1✔
248	data = client.tab_objects_list(feat)	1✔
249	if data is None:	1✔
UNCOV 250	continue	×
251	assert feat.properties.options.model_extra is not None	1✔
252	feat.properties.options.model_extra["tab"] = data	1✔
253	return features	1✔
254
255
256	def prepare_features(features: list[NspdFeature], localize: bool) -> gpd.GeoDataFrame:	1✔
257	prepared_features: list = []	1✔
258	unknown_layers = Counter()	1✔
259	for feat in features:	1✔
260	assert feat.properties.options.model_extra is not None	1✔
261	props: dict[str, Any] = feat.properties.options.model_extra.pop("tab", {})	1✔
262	if localize:	1✔
263	try:	1✔
264	feat = feat.cast()	1✔
265	except UnknownLayer:	1✔
UNCOV 266	unknown_layers[feat.properties.category_name] += 1	×
267	continue	1✔
268	props.update(feat.properties.options.model_dump_human_readable())	1✔
269	props["Категория"] = feat.properties.category_name	1✔
270	else:
271	props.update(feat.properties.options.model_dump())	1✔
UNCOV 272	props["category"] = feat.properties.category_name	×
UNCOV 273	props["geometry"] = feat.geometry.to_shape()	×
274	prepared_features.append(props)	1✔
275	for category, count in unknown_layers.items():	1✔
276	print(	1✔
277	f'[orange3] Было пропущено объектов с неизвестным слоем "{category}" - {count} (невозможно локализировать данные)'
278	)
279	return gpd.GeoDataFrame(prepared_features, crs=4326).fillna("")	1✔
280
281
282	def process_output(	1✔
283	features: Optional[list[NspdFeature]], output: Optional[Path], localize: bool
284	) -> None:
285	if features is None:	1✔
286	print("[red]Ничего не найдено")	1✔
287	raise typer.Abort()	1✔
UNCOV 288	gdf = prepare_features(features, localize)	×
289	try:	1✔
290	if output is None:	1✔
UNCOV 291	print(gdf.T)	×
UNCOV 292	return	×
293	elif output.suffix == ".xlsx":	1✔
294	gdf.to_excel(output, index=False)	1✔
295	elif output.suffix == ".csv":	1✔
296	gdf.to_csv(output, index=False)	1✔
297	else:
UNCOV 298	gdf.to_file(output)	×
299	except PermissionError:	1✔
300	_ = typer.prompt(	1✔
301	f"Файл {output} открыт в другой программе. Закройте его и нажмите любую клавишу",
302	default=True,
303	hide_input=True,
304	prompt_suffix="...",
305	show_default=False,
306	)
307	return process_output(features, output, localize)	1✔
308	print(f"[green]Найдено {len(gdf)} объектов, сохранено в файл {output.resolve()}[/]")	1✔
309
310
UNCOV 311	def version_callback(version: bool = False) -> None:	×
312	if version:	1✔
UNCOV 313	print(__version__)	×
UNCOV 314	raise typer.Exit()	×
315
316
317	@app.callback()	1✔
318	def common(	1✔
319	version: Annotated[
320	bool,
321	typer.Option(
322	"-v",
323	"--version",
324	help="Show current version",
325	is_eager=True,
326	callback=version_callback,
327	),
328	] = False,
329	):
UNCOV 330	pass	×
331
332
UNCOV 333	@app.command(no_args_is_help=True)	×
334	def geo(	1✔
335	input: Annotated[
336	str,
337	typer.Argument(
338	help="Путь к файлу с геоданными, координаты точек или WKT",
339	show_default=False,
340	),
341	],
342	choose_layer: ChooseLayersOption = False,
343	output: OutputOption = None,
344	localize: LocalizeOption = False,
345	add_tab_object: TabObjectsOption = False,
346	_test_layer_name: Annotated[Optional[str], typer.Option(hidden=True)] = None,
347	) -> None:
348	"""Поиск объектов по геоданным
349
350	Может производить поиск по файлам gpkg, geeojson, координатам точек или WKT-строке.
351	По умолчанию ищет в слое "Земельные участки из ЕГРН".
352	"""
353	if choose_layer:	1✔
354	layer_name = (	1✔
355	_test_layer_name
356	if _test_layer_name is not None
357	else questionary.select(
358	"Выберите слой: ", choices=get_args(LayerTitle)
359	).ask()
360	)
361	if layer_name is None:	1✔
362	raise typer.Abort()	1✔
363	else:
364	layer_name = "Земельные участки из ЕГРН"	1✔
365	geoms = define_geoms(input)	1✔
366	layer_def = define_layer_def(layer_name)	1✔
367	with Nspd() as client:	1✔
UNCOV 368	if isinstance(geoms[0], Point):	×
UNCOV 369	features = _get_features_from_geom(client.search_at_point, geoms, layer_def)	×
370	else:
371	features = _get_features_from_geom(	1✔
372	client.search_in_contour, geoms, layer_def
373	)
374	if features and add_tab_object:	1✔
375	_get_tab_object(client, features)	1✔
376	process_output(features, output, localize)	1✔
377
378
UNCOV 379	@app.command(no_args_is_help=True)	×
380	def search(	1✔
381	input: Annotated[
382	str,
383	typer.Argument(
384	help="Поисковой запрос. Может быть многострочным текстовым файлом",
385	show_default=False,
386	),
387	],
388	choose_layer: ChooseLayersOption = False,
389	plain_query: PlainOption = False,
390	output: OutputOption = None,
391	localize: LocalizeOption = False,
392	add_tab_object: TabObjectsOption = False,
393	_test_layer_names: Annotated[Optional[list[str]], typer.Option(hidden=True)] = None,
394	) -> None:
395	"""Поиск объектов по тексту
396
397	По умолчанию разбивает запрос на кадастровые номера и ищет в объектах недвижимости.
398	"""
399	if choose_layer:	1✔
400	layer_names = (	1✔
401	_test_layer_names
402	if _test_layer_names is not None
403	else questionary.checkbox(
404	"Выберите слои: ",
405	choices=get_args(LayerTitle),
406	validate=lambda x: "Не выбрано ни одно значение"
407	if len(x) == 0
408	else True,
409	).ask()
410	)
UNCOV 411	if layer_names is None:	×
UNCOV 412	raise typer.Abort()	×
UNCOV 413	layer_defs = [define_layer_def(layer_name) for layer_name in layer_names]	×
414	else:
UNCOV 415	layer_defs = None	×
UNCOV 416	queries = define_queries(input, plain_query)	×
UNCOV 417	with Nspd() as client:	×
UNCOV 418	features = _get_features_from_list(client, queries, layer_defs)	×
UNCOV 419	if features and add_tab_object:	×
UNCOV 420	features = _get_tab_object(client, features)	×
UNCOV 421	process_output(features, output, localize)	×
422
423
UNCOV 424	def main() -> Any:	×
UNCOV 425	return app()	×

yazmolod / pynspd / #12

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