-
Notifications
You must be signed in to change notification settings - Fork 24
/
search.py
153 lines (127 loc) · 4.7 KB
/
search.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
from datetime import datetime as dt
from typing import Any, Dict, List, Optional, Tuple, Union, cast
from ciso8601 import parse_rfc3339
from geojson_pydantic.geometries import GeometryCollection # type: ignore
from geojson_pydantic.geometries import (
LineString,
MultiLineString,
MultiPoint,
MultiPolygon,
Point,
Polygon,
)
from pydantic import BaseModel, Field, field_validator, model_validator
from stac_pydantic.api.extensions.fields import FieldsExtension
from stac_pydantic.api.extensions.query import Operator
from stac_pydantic.api.extensions.sort import SortExtension
from stac_pydantic.shared import BBox
Intersection = Union[
Point,
MultiPoint,
LineString,
MultiLineString,
Polygon,
MultiPolygon,
GeometryCollection,
]
class Search(BaseModel):
"""
The base class for STAC API searches.
https://github.com/radiantearth/stac-api-spec/blob/v1.0.0/item-search/README.md#query-parameter-table
"""
collections: Optional[List[str]] = None
ids: Optional[List[str]] = None
bbox: Optional[BBox] = None
intersects: Optional[Intersection] = None
datetime: Optional[str] = None
limit: int = 10
@property
def start_date(self) -> Optional[dt]:
values = (self.datetime or "").split("/")
if len(values) == 1:
return None
if values[0] == ".." or values[0] == "":
return None
return parse_rfc3339(values[0])
@property
def end_date(self) -> Optional[dt]:
values = (self.datetime or "").split("/")
if len(values) == 1:
return parse_rfc3339(values[0])
if values[1] == ".." or values[1] == "":
return None
return parse_rfc3339(values[1])
# Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.
@model_validator(mode="before")
def validate_spatial(cls, values: Dict[str, Any]) -> Dict[str, Any]:
if values.get("intersects") and values.get("bbox") is not None:
raise ValueError("intersects and bbox parameters are mutually exclusive")
return values
@field_validator("bbox")
@classmethod
def validate_bbox(cls, v: BBox) -> BBox:
if v:
# Validate order
if len(v) == 4:
xmin, ymin, xmax, ymax = cast(Tuple[int, int, int, int], v)
else:
xmin, ymin, min_elev, xmax, ymax, max_elev = cast(
Tuple[int, int, int, int, int, int], v
)
if max_elev < min_elev:
raise ValueError(
"Maximum elevation must greater than minimum elevation"
)
if xmax < xmin:
raise ValueError(
"Maximum longitude must be greater than minimum longitude"
)
if ymax < ymin:
raise ValueError(
"Maximum longitude must be greater than minimum longitude"
)
# Validate against WGS84
if xmin < -180 or ymin < -90 or xmax > 180 or ymax > 90:
raise ValueError("Bounding box must be within (-180, -90, 180, 90)")
return v
@field_validator("datetime")
@classmethod
def validate_datetime(cls, v: str) -> str:
if "/" in v:
values = v.split("/")
else:
# Single date is interpreted as end date
values = ["..", v]
dates: List[dt] = []
for value in values:
if value == ".." or value == "":
continue
dates.append(parse_rfc3339(value))
if len(values) > 2:
raise ValueError(
"Invalid datetime range, must match format (begin_date, end_date)"
)
if not {"..", ""}.intersection(set(values)):
if dates[0] > dates[1]:
raise ValueError(
"Invalid datetime range, must match format (begin_date, end_date)"
)
return v
@property
def spatial_filter(self) -> Optional[Intersection]:
"""Return a geojson-pydantic object representing the spatial filter for the search request.
Check for both because the ``bbox`` and ``intersects`` parameters are mutually exclusive.
"""
if self.bbox:
return Polygon.from_bounds(*self.bbox)
if self.intersects:
return self.intersects
else:
return None
class ExtendedSearch(Search):
"""
STAC API search with extensions enabled.
"""
field: Optional[FieldsExtension] = Field(None, alias="fields")
query: Optional[Dict[str, Dict[Operator, Any]]] = None
sortby: Optional[List[SortExtension]] = None