]> git.openstreetmap.org Git - nominatim.git/blob - src/nominatim_api/types.py
move server route creation into async function
[nominatim.git] / src / nominatim_api / types.py
1 # SPDX-License-Identifier: GPL-3.0-or-later
2 #
3 # This file is part of Nominatim. (https://nominatim.org)
4 #
5 # Copyright (C) 2024 by the Nominatim developer community.
6 # For a full list of authors see the git log.
7 """
8 Complex datatypes used by the Nominatim API.
9 """
10 from typing import Optional, Union, Tuple, NamedTuple, TypeVar, Type, Dict, \
11                    Any, List, Sequence
12 from collections import abc
13 import dataclasses
14 import enum
15 import math
16 from struct import unpack
17 from binascii import unhexlify
18
19 from .errors import UsageError
20 from .localization import Locales
21
22
23 @dataclasses.dataclass
24 class PlaceID:
25     """ Reference a place by Nominatim's internal ID.
26
27         A PlaceID may reference place from the main table placex, from
28         the interpolation tables or the postcode tables. Place IDs are not
29         stable between installations. You may use this type theefore only
30         with place IDs obtained from the same database.
31     """
32     place_id: int
33     """
34     The internal ID of the place to reference.
35     """
36
37
38 @dataclasses.dataclass
39 class OsmID:
40     """ Reference a place by its OSM ID and potentially the basic category.
41
42         The OSM ID may refer to places in the main table placex and OSM
43         interpolation lines.
44     """
45     osm_type: str
46     """ OSM type of the object. Must be one of `N`(node), `W`(way) or
47         `R`(relation).
48     """
49     osm_id: int
50     """ The OSM ID of the object.
51     """
52     osm_class: Optional[str] = None
53     """ The same OSM object may appear multiple times in the database under
54         different categories. The optional class parameter allows to distinguish
55         the different categories and corresponds to the key part of the category.
56         If there are multiple objects in the database and `osm_class` is
57         left out, then one of the objects is returned at random.
58     """
59
60     def __post_init__(self) -> None:
61         if self.osm_type not in ('N', 'W', 'R'):
62             raise ValueError(f"Illegal OSM type '{self.osm_type}'. Must be one of N, W, R.")
63
64
65 PlaceRef = Union[PlaceID, OsmID]
66
67
68 class Point(NamedTuple):
69     """ A geographic point in WGS84 projection.
70     """
71     x: float
72     y: float
73
74     @property
75     def lat(self) -> float:
76         """ Return the latitude of the point.
77         """
78         return self.y
79
80     @property
81     def lon(self) -> float:
82         """ Return the longitude of the point.
83         """
84         return self.x
85
86     def to_geojson(self) -> str:
87         """ Return the point in GeoJSON format.
88         """
89         return f'{{"type": "Point","coordinates": [{self.x}, {self.y}]}}'
90
91     @staticmethod
92     def from_wkb(wkb: Union[str, bytes]) -> 'Point':
93         """ Create a point from EWKB as returned from the database.
94         """
95         if isinstance(wkb, str):
96             wkb = unhexlify(wkb)
97         if len(wkb) != 25:
98             raise ValueError(f"Point wkb has unexpected length {len(wkb)}")
99         if wkb[0] == 0:
100             gtype, srid, x, y = unpack('>iidd', wkb[1:])
101         elif wkb[0] == 1:
102             gtype, srid, x, y = unpack('<iidd', wkb[1:])
103         else:
104             raise ValueError("WKB has unknown endian value.")
105
106         if gtype != 0x20000001:
107             raise ValueError("WKB must be a point geometry.")
108         if srid != 4326:
109             raise ValueError("Only WGS84 WKB supported.")
110
111         return Point(x, y)
112
113     @staticmethod
114     def from_param(inp: Any) -> 'Point':
115         """ Create a point from an input parameter. The parameter
116             may be given as a point, a string or a sequence of
117             strings or floats. Raises a UsageError if the format is
118             not correct.
119         """
120         if isinstance(inp, Point):
121             return inp
122
123         seq: Sequence[str]
124         if isinstance(inp, str):
125             seq = inp.split(',')
126         elif isinstance(inp, abc.Sequence):
127             seq = inp
128
129         if len(seq) != 2:
130             raise UsageError('Point parameter needs 2 coordinates.')
131         try:
132             x, y = filter(math.isfinite, map(float, seq))
133         except ValueError as exc:
134             raise UsageError('Point parameter needs to be numbers.') from exc
135
136         if x < -180.0 or x > 180.0 or y < -90.0 or y > 90.0:
137             raise UsageError('Point coordinates invalid.')
138
139         return Point(x, y)
140
141     def to_wkt(self) -> str:
142         """ Return the WKT representation of the point.
143         """
144         return f'POINT({self.x} {self.y})'
145
146
147 AnyPoint = Union[Point, Tuple[float, float]]
148
149 WKB_BBOX_HEADER_LE = b'\x01\x03\x00\x00\x20\xE6\x10\x00\x00\x01\x00\x00\x00\x05\x00\x00\x00'
150 WKB_BBOX_HEADER_BE = b'\x00\x20\x00\x00\x03\x00\x00\x10\xe6\x00\x00\x00\x01\x00\x00\x00\x05'
151
152
153 class Bbox:
154     """ A bounding box in WGS84 projection.
155
156         The coordinates are available as an array in the 'coord'
157         property in the order (minx, miny, maxx, maxy).
158     """
159     def __init__(self, minx: float, miny: float, maxx: float, maxy: float) -> None:
160         """ Create a new bounding box with the given coordinates in WGS84
161             projection.
162         """
163         self.coords = (minx, miny, maxx, maxy)
164
165     @property
166     def minlat(self) -> float:
167         """ Southern-most latitude, corresponding to the minimum y coordinate.
168         """
169         return self.coords[1]
170
171     @property
172     def maxlat(self) -> float:
173         """ Northern-most latitude, corresponding to the maximum y coordinate.
174         """
175         return self.coords[3]
176
177     @property
178     def minlon(self) -> float:
179         """ Western-most longitude, corresponding to the minimum x coordinate.
180         """
181         return self.coords[0]
182
183     @property
184     def maxlon(self) -> float:
185         """ Eastern-most longitude, corresponding to the maximum x coordinate.
186         """
187         return self.coords[2]
188
189     @property
190     def area(self) -> float:
191         """ Return the area of the box in WGS84.
192         """
193         return (self.coords[2] - self.coords[0]) * (self.coords[3] - self.coords[1])
194
195     def contains(self, pt: Point) -> bool:
196         """ Check if the point is inside or on the boundary of the box.
197         """
198         return self.coords[0] <= pt[0] and self.coords[1] <= pt[1]\
199             and self.coords[2] >= pt[0] and self.coords[3] >= pt[1]
200
201     def to_wkt(self) -> str:
202         """ Return the WKT representation of the Bbox. This
203             is a simple polygon with four points.
204         """
205         return 'POLYGON(({0} {1},{0} {3},{2} {3},{2} {1},{0} {1}))'\
206             .format(*self.coords)
207
208     @staticmethod
209     def from_wkb(wkb: Union[None, str, bytes]) -> 'Optional[Bbox]':
210         """ Create a Bbox from a bounding box polygon as returned by
211             the database. Returns `None` if the input value is None.
212         """
213         if wkb is None:
214             return None
215
216         if isinstance(wkb, str):
217             wkb = unhexlify(wkb)
218
219         if len(wkb) != 97:
220             raise ValueError("WKB must be a bounding box polygon")
221         if wkb.startswith(WKB_BBOX_HEADER_LE):
222             x1, y1, _, _, x2, y2 = unpack('<dddddd', wkb[17:65])
223         elif wkb.startswith(WKB_BBOX_HEADER_BE):
224             x1, y1, _, _, x2, y2 = unpack('>dddddd', wkb[17:65])
225         else:
226             raise ValueError("WKB has wrong header")
227
228         return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2))
229
230     @staticmethod
231     def from_point(pt: Point, buffer: float) -> 'Bbox':
232         """ Return a Bbox around the point with the buffer added to all sides.
233         """
234         return Bbox(pt[0] - buffer, pt[1] - buffer,
235                     pt[0] + buffer, pt[1] + buffer)
236
237     @staticmethod
238     def from_param(inp: Any) -> 'Bbox':
239         """ Return a Bbox from an input parameter. The box may be
240             given as a Bbox, a string or a list or strings or integer.
241             Raises a UsageError if the format is incorrect.
242         """
243         if isinstance(inp, Bbox):
244             return inp
245
246         seq: Sequence[str]
247         if isinstance(inp, str):
248             seq = inp.split(',')
249         elif isinstance(inp, abc.Sequence):
250             seq = inp
251
252         if len(seq) != 4:
253             raise UsageError('Bounding box parameter needs 4 coordinates.')
254         try:
255             x1, y1, x2, y2 = filter(math.isfinite, map(float, seq))
256         except ValueError as exc:
257             raise UsageError('Bounding box parameter needs to be numbers.') from exc
258
259         x1 = min(180, max(-180, x1))
260         x2 = min(180, max(-180, x2))
261         y1 = min(90, max(-90, y1))
262         y2 = min(90, max(-90, y2))
263
264         if x1 == x2 or y1 == y2:
265             raise UsageError('Bounding box with invalid parameters.')
266
267         return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2))
268
269
270 class GeometryFormat(enum.Flag):
271     """ All search functions support returning the full geometry of a place in
272         various formats. The internal geometry is converted by PostGIS to
273         the desired format and then returned as a string. It is possible to
274         request multiple formats at the same time.
275     """
276     NONE = 0
277     """ No geometry requested. Alias for a empty flag.
278     """
279     GEOJSON = enum.auto()
280     """
281     [GeoJSON](https://geojson.org/) format
282     """
283     KML = enum.auto()
284     """
285     [KML](https://en.wikipedia.org/wiki/Keyhole_Markup_Language) format
286     """
287     SVG = enum.auto()
288     """
289     [SVG](http://www.w3.org/TR/SVG/paths.html) format
290     """
291     TEXT = enum.auto()
292     """
293     [WKT](https://en.wikipedia.org/wiki/Well-known_text_representation_of_geometry) format
294     """
295
296
297 class DataLayer(enum.Flag):
298     """ The `DataLayer` flag type defines the layers that can be selected
299         for reverse and forward search.
300     """
301     ADDRESS = enum.auto()
302     """ The address layer contains all places relevant for addresses:
303         fully qualified addresses with a house number (or a house name equivalent,
304         for some addresses) and places that can be part of an address like
305         roads, cities, states.
306     """
307     POI = enum.auto()
308     """ Layer for points of interest like shops, restaurants but also
309         recycling bins or postboxes.
310     """
311     RAILWAY = enum.auto()
312     """ Layer with railway features including tracks and other infrastructure.
313         Note that in Nominatim's standard configuration, only very few railway
314         features are imported into the database. Thus a custom configuration
315         is required to make full use of this layer.
316     """
317     NATURAL = enum.auto()
318     """ Layer with natural features like rivers, lakes and mountains.
319     """
320     MANMADE = enum.auto()
321     """ Layer with other human-made features and boundaries. This layer is
322         the catch-all and includes all features not covered by the other
323         layers. A typical example for this layer are national park boundaries.
324     """
325
326
327 def format_country(cc: Any) -> List[str]:
328     """ Extract a list of country codes from the input which may be either
329         a string or list of strings. Filters out all values that are not
330         a two-letter string.
331     """
332     clist: Sequence[str]
333     if isinstance(cc, str):
334         clist = cc.split(',')
335     elif isinstance(cc, abc.Sequence):
336         clist = cc
337     else:
338         raise UsageError("Parameter 'country' needs to be a comma-separated list "
339                          "or a Python list of strings.")
340
341     return [cc.lower() for cc in clist if isinstance(cc, str) and len(cc) == 2]
342
343
344 def format_excluded(ids: Any) -> List[int]:
345     """ Extract a list of place ids from the input which may be either
346         a string or a list of strings or ints. Ignores empty value but
347         throws a UserError on anything that cannot be converted to int.
348     """
349     plist: Sequence[str]
350     if isinstance(ids, str):
351         plist = [s.strip() for s in ids.split(',')]
352     elif isinstance(ids, abc.Sequence):
353         plist = ids
354     else:
355         raise UsageError("Parameter 'excluded' needs to be a comma-separated list "
356                          "or a Python list of numbers.")
357     if not all(isinstance(i, int) or
358                (isinstance(i, str) and (not i or i.isdigit())) for i in plist):
359         raise UsageError("Parameter 'excluded' only takes place IDs.")
360
361     return [int(id) for id in plist if id] or [0]
362
363
364 def format_categories(categories: List[Tuple[str, str]]) -> List[Tuple[str, str]]:
365     """ Extract a list of categories. Currently a noop.
366     """
367     return categories
368
369
370 TParam = TypeVar('TParam', bound='LookupDetails')
371
372
373 @dataclasses.dataclass
374 class LookupDetails:
375     """ Collection of parameters that define which kind of details are
376         returned with a lookup or details result.
377     """
378     geometry_output: GeometryFormat = GeometryFormat.NONE
379     """ Add the full geometry of the place to the result. Multiple
380         formats may be selected. Note that geometries can become quite large.
381     """
382     address_details: bool = False
383     """ Get detailed information on the places that make up the address
384         for the result.
385     """
386     linked_places: bool = False
387     """ Get detailed information on the places that link to the result.
388     """
389     parented_places: bool = False
390     """ Get detailed information on all places that this place is a parent
391         for, i.e. all places for which it provides the address details.
392         Only POI places can have parents.
393     """
394     keywords: bool = False
395     """ Add information about the search terms used for this place.
396     """
397     geometry_simplification: float = 0.0
398     """ Simplification factor for a geometry in degrees WGS. A factor of
399         0.0 means the original geometry is kept. The higher the value, the
400         more the geometry gets simplified.
401     """
402     locales: Locales = Locales()
403     """ Preferred languages for localization of results.
404     """
405
406     @classmethod
407     def from_kwargs(cls: Type[TParam], kwargs: Dict[str, Any]) -> TParam:
408         """ Load the data fields of the class from a dictionary.
409             Unknown entries in the dictionary are ignored, missing ones
410             get the default setting.
411
412             The function supports type checking and throws a UsageError
413             when the value does not fit.
414         """
415         def _check_field(v: Any, field: 'dataclasses.Field[Any]') -> Any:
416             if v is None:
417                 return field.default_factory() \
418                        if field.default_factory != dataclasses.MISSING \
419                        else field.default
420             if field.metadata and 'transform' in field.metadata:
421                 return field.metadata['transform'](v)
422             if not isinstance(v, field.type):  # type: ignore[arg-type]
423                 raise UsageError(f"Parameter '{field.name}' needs to be of {field.type!s}.")
424             return v
425
426         return cls(**{f.name: _check_field(kwargs[f.name], f)
427                       for f in dataclasses.fields(cls) if f.name in kwargs})
428
429
430 @dataclasses.dataclass
431 class ReverseDetails(LookupDetails):
432     """ Collection of parameters for the reverse call.
433     """
434
435     max_rank: int = dataclasses.field(default=30,
436                                       metadata={'transform': lambda v: max(0, min(v, 30))})
437     """ Highest address rank to return.
438     """
439
440     layers: DataLayer = DataLayer.ADDRESS | DataLayer.POI
441     """ Filter which kind of data to include.
442     """
443
444
445 @dataclasses.dataclass
446 class SearchDetails(LookupDetails):
447     """ Collection of parameters for the search call.
448     """
449     max_results: int = 10
450     """ Maximum number of results to be returned. The actual number of results
451         may be less.
452     """
453
454     min_rank: int = dataclasses.field(default=0,
455                                       metadata={'transform': lambda v: max(0, min(v, 30))})
456     """ Lowest address rank to return.
457     """
458
459     max_rank: int = dataclasses.field(default=30,
460                                       metadata={'transform': lambda v: max(0, min(v, 30))})
461     """ Highest address rank to return.
462     """
463
464     layers: Optional[DataLayer] = dataclasses.field(default=None,
465                                                     metadata={'transform': lambda r: r})
466     """ Filter which kind of data to include. When 'None' (the default) then
467         filtering by layers is disabled.
468     """
469
470     countries: List[str] = dataclasses.field(default_factory=list,
471                                              metadata={'transform': format_country})
472     """ Restrict search results to the given countries. An empty list (the
473         default) will disable this filter.
474     """
475
476     excluded: List[int] = dataclasses.field(default_factory=list,
477                                             metadata={'transform': format_excluded})
478     """ List of OSM objects to exclude from the results. Currently only
479         works when the internal place ID is given.
480         An empty list (the default) will disable this filter.
481     """
482
483     viewbox: Optional[Bbox] = dataclasses.field(default=None,
484                                                 metadata={'transform': Bbox.from_param})
485     """ Focus the search on a given map area.
486     """
487
488     bounded_viewbox: bool = False
489     """ Use 'viewbox' as a filter and restrict results to places within the
490         given area.
491     """
492
493     near: Optional[Point] = dataclasses.field(default=None,
494                                               metadata={'transform': Point.from_param})
495     """ Order results by distance to the given point.
496     """
497
498     near_radius: Optional[float] = dataclasses.field(default=None,
499                                                      metadata={'transform': lambda r: r})
500     """ Use near point as a filter and drop results outside the given
501         radius. Radius is given in degrees WSG84.
502     """
503
504     categories: List[Tuple[str, str]] = dataclasses.field(default_factory=list,
505                                                           metadata={'transform': format_categories})
506     """ Restrict search to places with one of the given class/type categories.
507         An empty list (the default) will disable this filter.
508     """
509
510     viewbox_x2: Optional[Bbox] = None
511
512     def __post_init__(self) -> None:
513         if self.viewbox is not None:
514             xext = (self.viewbox.maxlon - self.viewbox.minlon)/2
515             yext = (self.viewbox.maxlat - self.viewbox.minlat)/2
516             self.viewbox_x2 = Bbox(self.viewbox.minlon - xext, self.viewbox.minlat - yext,
517                                    self.viewbox.maxlon + xext, self.viewbox.maxlat + yext)
518
519     def restrict_min_max_rank(self, new_min: int, new_max: int) -> None:
520         """ Change the min_rank and max_rank fields to respect the
521             given boundaries.
522         """
523         assert new_min <= new_max
524         self.min_rank = max(self.min_rank, new_min)
525         self.max_rank = min(self.max_rank, new_max)
526
527     def is_impossible(self) -> bool:
528         """ Check if the parameter configuration is contradictionary and
529             cannot yield any results.
530         """
531         return (self.min_rank > self.max_rank
532                 or (self.bounded_viewbox
533                     and self.viewbox is not None and self.near is not None
534                     and self.viewbox.contains(self.near))
535                 or (self.layers is not None and not self.layers)
536                 or (self.max_rank <= 4 and
537                     self.layers is not None and not self.layers & DataLayer.ADDRESS))
538
539     def layer_enabled(self, layer: DataLayer) -> bool:
540         """ Check if the given layer has been chosen. Also returns
541             true when layer restriction has been disabled completely.
542         """
543         return self.layers is None or bool(self.layers & layer)