|
| 1 | +from __future__ import annotations |
| 2 | + |
1 | 3 | from json import JSONDecodeError |
2 | | -from typing import Literal, Optional, Union |
| 4 | +from typing import Literal |
3 | 5 |
|
4 | 6 | from dataretrieval.utils import query |
5 | 7 |
|
@@ -32,13 +34,13 @@ def _query_nldi(url, query_params, error_message): |
32 | 34 | def get_flowlines( |
33 | 35 | navigation_mode: str, |
34 | 36 | distance: int = 5, |
35 | | - feature_source: Optional[str] = None, |
36 | | - feature_id: Optional[str] = None, |
37 | | - comid: Optional[int] = None, |
38 | | - stop_comid: Optional[int] = None, |
| 37 | + feature_source: str | None = None, |
| 38 | + feature_id: str | None = None, |
| 39 | + comid: int | None = None, |
| 40 | + stop_comid: int | None = None, |
39 | 41 | trim_start: bool = False, |
40 | 42 | as_json: bool = False, |
41 | | -) -> Union[gpd.GeoDataFrame, dict]: |
| 43 | +) -> gpd.GeoDataFrame | dict: |
42 | 44 | """Gets the flowlines for the specified navigation either by comid or feature |
43 | 45 | source in WGS84 lat/long coordinates as GeoDataFrame containing a polyline geometry. |
44 | 46 |
|
@@ -116,7 +118,7 @@ def get_basin( |
116 | 118 | simplified: bool = True, |
117 | 119 | split_catchment: bool = False, |
118 | 120 | as_json: bool = False, |
119 | | -) -> Union[gpd.GeoDataFrame, dict]: |
| 121 | +) -> gpd.GeoDataFrame | dict: |
120 | 122 | """Gets the aggregated basin for the specified feature in WGS84 lat/lon |
121 | 123 | as GeoDataFrame or as JSON conatining a polygon geometry. |
122 | 124 |
|
@@ -164,17 +166,17 @@ def get_basin( |
164 | 166 |
|
165 | 167 |
|
166 | 168 | def get_features( |
167 | | - data_source: Optional[str] = None, |
168 | | - navigation_mode: Optional[str] = None, |
| 169 | + data_source: str | None = None, |
| 170 | + navigation_mode: str | None = None, |
169 | 171 | distance: int = 50, |
170 | | - feature_source: Optional[str] = None, |
171 | | - feature_id: Optional[str] = None, |
172 | | - comid: Optional[int] = None, |
173 | | - lat: Optional[float] = None, |
174 | | - long: Optional[float] = None, |
175 | | - stop_comid: Optional[int] = None, |
| 172 | + feature_source: str | None = None, |
| 173 | + feature_id: str | None = None, |
| 174 | + comid: int | None = None, |
| 175 | + lat: float | None = None, |
| 176 | + long: float | None = None, |
| 177 | + stop_comid: int | None = None, |
176 | 178 | as_json: bool = False, |
177 | | -) -> Union[gpd.GeoDataFrame, dict]: |
| 179 | +) -> gpd.GeoDataFrame | dict: |
178 | 180 | """Gets all features found along the specified navigation either by |
179 | 181 | comid or feature source as points in WGS84 lat/long coordinates - a GeoDataFrame |
180 | 182 | containing a point geometry. |
@@ -247,11 +249,10 @@ def get_features( |
247 | 249 | ) |
248 | 250 |
|
249 | 251 | if not lat: |
250 | | - if comid or data_source: |
251 | | - if navigation_mode is None: |
252 | | - raise ValueError( |
253 | | - "navigation_mode is required if comid or data_source is provided" |
254 | | - ) |
| 252 | + if (comid or data_source) and navigation_mode is None: |
| 253 | + raise ValueError( |
| 254 | + "navigation_mode is required if comid or data_source is provided" |
| 255 | + ) |
255 | 256 | # validate the feature source and comid |
256 | 257 | _validate_feature_source_comid(feature_source, feature_id, comid) |
257 | 258 | # validate the data source |
@@ -334,14 +335,14 @@ def get_features_by_data_source(data_source: str) -> gpd.GeoDataFrame: |
334 | 335 |
|
335 | 336 |
|
336 | 337 | def search( |
337 | | - feature_source: Optional[str] = None, |
338 | | - feature_id: Optional[str] = None, |
339 | | - navigation_mode: Optional[str] = None, |
340 | | - data_source: Optional[str] = None, |
| 338 | + feature_source: str | None = None, |
| 339 | + feature_id: str | None = None, |
| 340 | + navigation_mode: str | None = None, |
| 341 | + data_source: str | None = None, |
341 | 342 | find: Literal["basin", "flowlines", "features"] = "features", |
342 | | - comid: Optional[int] = None, |
343 | | - lat: Optional[float] = None, |
344 | | - long: Optional[float] = None, |
| 343 | + comid: int | None = None, |
| 344 | + lat: float | None = None, |
| 345 | + long: float | None = None, |
345 | 346 | distance: int = 50, |
346 | 347 | ) -> dict: |
347 | 348 | """Searches for the specified feature in NLDI and returns the results |
@@ -489,7 +490,7 @@ def _validate_navigation_mode(navigation_mode: str): |
489 | 490 |
|
490 | 491 |
|
491 | 492 | def _validate_feature_source_comid( |
492 | | - feature_source: Optional[str], feature_id: Optional[str], comid: Optional[int] |
| 493 | + feature_source: str | None, feature_id: str | None, comid: int | None |
493 | 494 | ): |
494 | 495 | if feature_source is not None and feature_id is None: |
495 | 496 | raise ValueError("feature_id is required if feature_source is provided") |
|
0 commit comments