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) 2023 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 nominatim.errors import UsageError
  20
  21 # pylint: disable=no-member,too-many-boolean-expressions,too-many-instance-attributes
  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
  75     @property
  76     def lat(self) -> float:
  77         """ Return the latitude of the point.
  78         """
  79         return self.y
  80
  81
  82     @property
  83     def lon(self) -> float:
  84         """ Return the longitude of the point.
  85         """
  86         return self.x
  87
  88
  89     def to_geojson(self) -> str:
  90         """ Return the point in GeoJSON format.
  91         """
  92         return f'{{"type": "Point","coordinates": [{self.x}, {self.y}]}}'
  93
  94
  95     @staticmethod
  96     def from_wkb(wkb: Union[str, bytes]) -> 'Point':
  97         """ Create a point from EWKB as returned from the database.
  98         """
  99         if isinstance(wkb, str):
 100             wkb = unhexlify(wkb)
 101         if len(wkb) != 25:
 102             raise ValueError(f"Point wkb has unexpected length {len(wkb)}")
 103         if wkb[0] == 0:
 104             gtype, srid, x, y = unpack('>iidd', wkb[1:])
 105         elif wkb[0] == 1:
 106             gtype, srid, x, y = unpack('<iidd', wkb[1:])
 107         else:
 108             raise ValueError("WKB has unknown endian value.")
 109
 110         if gtype != 0x20000001:
 111             raise ValueError("WKB must be a point geometry.")
 112         if srid != 4326:
 113             raise ValueError("Only WGS84 WKB supported.")
 114
 115         return Point(x, y)
 116
 117
 118     @staticmethod
 119     def from_param(inp: Any) -> 'Point':
 120         """ Create a point from an input parameter. The parameter
 121             may be given as a point, a string or a sequence of
 122             strings or floats. Raises a UsageError if the format is
 123             not correct.
 124         """
 125         if isinstance(inp, Point):
 126             return inp
 127
 128         seq: Sequence[str]
 129         if isinstance(inp, str):
 130             seq = inp.split(',')
 131         elif isinstance(inp, abc.Sequence):
 132             seq = inp
 133
 134         if len(seq) != 2:
 135             raise UsageError('Point parameter needs 2 coordinates.')
 136         try:
 137             x, y = filter(math.isfinite, map(float, seq))
 138         except ValueError as exc:
 139             raise UsageError('Point parameter needs to be numbers.') from exc
 140
 141         if x < -180.0 or x > 180.0 or y < -90.0 or y > 90.0:
 142             raise UsageError('Point coordinates invalid.')
 143
 144         return Point(x, y)
 145
 146
 147     def to_wkt(self) -> str:
 148         """ Return the WKT representation of the point.
 149         """
 150         return f'POINT({self.x} {self.y})'
 151
 152
 153
 154 AnyPoint = Union[Point, Tuple[float, float]]
 155
 156 WKB_BBOX_HEADER_LE = b'\x01\x03\x00\x00\x20\xE6\x10\x00\x00\x01\x00\x00\x00\x05\x00\x00\x00'
 157 WKB_BBOX_HEADER_BE = b'\x00\x20\x00\x00\x03\x00\x00\x10\xe6\x00\x00\x00\x01\x00\x00\x00\x05'
 158
 159 class Bbox:
 160     """ A bounding box in WGS84 projection.
 161
 162         The coordinates are available as an array in the 'coord'
 163         property in the order (minx, miny, maxx, maxy).
 164     """
 165     def __init__(self, minx: float, miny: float, maxx: float, maxy: float) -> None:
 166         """ Create a new bounding box with the given coordinates in WGS84
 167             projection.
 168         """
 169         self.coords = (minx, miny, maxx, maxy)
 170
 171
 172     @property
 173     def minlat(self) -> float:
 174         """ Southern-most latitude, corresponding to the minimum y coordinate.
 175         """
 176         return self.coords[1]
 177
 178
 179     @property
 180     def maxlat(self) -> float:
 181         """ Northern-most latitude, corresponding to the maximum y coordinate.
 182         """
 183         return self.coords[3]
 184
 185
 186     @property
 187     def minlon(self) -> float:
 188         """ Western-most longitude, corresponding to the minimum x coordinate.
 189         """
 190         return self.coords[0]
 191
 192
 193     @property
 194     def maxlon(self) -> float:
 195         """ Eastern-most longitude, corresponding to the maximum x coordinate.
 196         """
 197         return self.coords[2]
 198
 199
 200     @property
 201     def area(self) -> float:
 202         """ Return the area of the box in WGS84.
 203         """
 204         return (self.coords[2] - self.coords[0]) * (self.coords[3] - self.coords[1])
 205
 206
 207     def contains(self, pt: Point) -> bool:
 208         """ Check if the point is inside or on the boundary of the box.
 209         """
 210         return self.coords[0] <= pt[0] and self.coords[1] <= pt[1]\
 211                and self.coords[2] >= pt[0] and self.coords[3] >= pt[1]
 212
 213
 214     def to_wkt(self) -> str:
 215         """ Return the WKT representation of the Bbox. This
 216             is a simple polygon with four points.
 217         """
 218         return 'POLYGON(({0} {1},{0} {3},{2} {3},{2} {1},{0} {1}))'\
 219                   .format(*self.coords) # pylint: disable=consider-using-f-string
 220
 221
 222     @staticmethod
 223     def from_wkb(wkb: Union[None, str, bytes]) -> 'Optional[Bbox]':
 224         """ Create a Bbox from a bounding box polygon as returned by
 225             the database. Returns `None` if the input value is None.
 226         """
 227         if wkb is None:
 228             return None
 229
 230         if isinstance(wkb, str):
 231             wkb = unhexlify(wkb)
 232
 233         if len(wkb) != 97:
 234             raise ValueError("WKB must be a bounding box polygon")
 235         if wkb.startswith(WKB_BBOX_HEADER_LE):
 236             x1, y1, _, _, x2, y2 = unpack('<dddddd', wkb[17:65])
 237         elif wkb.startswith(WKB_BBOX_HEADER_BE):
 238             x1, y1, _, _, x2, y2 = unpack('>dddddd', wkb[17:65])
 239         else:
 240             raise ValueError("WKB has wrong header")
 241
 242         return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2))
 243
 244
 245     @staticmethod
 246     def from_point(pt: Point, buffer: float) -> 'Bbox':
 247         """ Return a Bbox around the point with the buffer added to all sides.
 248         """
 249         return Bbox(pt[0] - buffer, pt[1] - buffer,
 250                     pt[0] + buffer, pt[1] + buffer)
 251
 252
 253     @staticmethod
 254     def from_param(inp: Any) -> 'Bbox':
 255         """ Return a Bbox from an input parameter. The box may be
 256             given as a Bbox, a string or a list or strings or integer.
 257             Raises a UsageError if the format is incorrect.
 258         """
 259         if isinstance(inp, Bbox):
 260             return inp
 261
 262         seq: Sequence[str]
 263         if isinstance(inp, str):
 264             seq = inp.split(',')
 265         elif isinstance(inp, abc.Sequence):
 266             seq = inp
 267
 268         if len(seq) != 4:
 269             raise UsageError('Bounding box parameter needs 4 coordinates.')
 270         try:
 271             x1, y1, x2, y2 = filter(math.isfinite, map(float, seq))
 272         except ValueError as exc:
 273             raise UsageError('Bounding box parameter needs to be numbers.') from exc
 274
 275         x1 = min(180, max(-180, x1))
 276         x2 = min(180, max(-180, x2))
 277         y1 = min(90, max(-90, y1))
 278         y2 = min(90, max(-90, y2))
 279
 280         if x1 == x2 or y1 == y2:
 281             raise UsageError('Bounding box with invalid parameters.')
 282
 283         return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2))
 284
 285
 286 class GeometryFormat(enum.Flag):
 287     """ All search functions support returning the full geometry of a place in
 288         various formats. The internal geometry is converted by PostGIS to
 289         the desired format and then returned as a string. It is possible to
 290         request multiple formats at the same time.
 291     """
 292     NONE = 0
 293     """ No geometry requested. Alias for a empty flag.
 294     """
 295     GEOJSON = enum.auto()
 296     """
 297     [GeoJSON](https://geojson.org/) format
 298     """
 299     KML = enum.auto()
 300     """
 301     [KML](https://en.wikipedia.org/wiki/Keyhole_Markup_Language) format
 302     """
 303     SVG = enum.auto()
 304     """
 305     [SVG](http://www.w3.org/TR/SVG/paths.html) format
 306     """
 307     TEXT = enum.auto()
 308     """
 309     [WKT](https://en.wikipedia.org/wiki/Well-known_text_representation_of_geometry) format
 310     """
 311
 312
 313 class DataLayer(enum.Flag):
 314     """ Layer types that can be selected for reverse and forward search.
 315     """
 316     POI = enum.auto()
 317     ADDRESS = enum.auto()
 318     RAILWAY = enum.auto()
 319     MANMADE = enum.auto()
 320     NATURAL = enum.auto()
 321
 322
 323 def format_country(cc: Any) -> List[str]:
 324     """ Extract a list of country codes from the input which may be either
 325         a string or list of strings. Filters out all values that are not
 326         a two-letter string.
 327     """
 328     clist: Sequence[str]
 329     if isinstance(cc, str):
 330         clist = cc.split(',')
 331     elif isinstance(cc, abc.Sequence):
 332         clist = cc
 333     else:
 334         raise UsageError("Parameter 'country' needs to be a comma-separated list "
 335                          "or a Python list of strings.")
 336
 337     return [cc.lower() for cc in clist if isinstance(cc, str) and len(cc) == 2]
 338
 339
 340 def format_excluded(ids: Any) -> List[int]:
 341     """ Extract a list of place ids from the input which may be either
 342         a string or a list of strings or ints. Ignores empty value but
 343         throws a UserError on anything that cannot be converted to int.
 344     """
 345     plist: Sequence[str]
 346     if isinstance(ids, str):
 347         plist = [s.strip() for s in ids.split(',')]
 348     elif isinstance(ids, abc.Sequence):
 349         plist = ids
 350     else:
 351         raise UsageError("Parameter 'excluded' needs to be a comma-separated list "
 352                          "or a Python list of numbers.")
 353     if not all(isinstance(i, int) or
 354                (isinstance(i, str) and (not i or i.isdigit())) for i in plist):
 355         raise UsageError("Parameter 'excluded' only takes place IDs.")
 356
 357     return [int(id) for id in plist if id] or [0]
 358
 359
 360 def format_categories(categories: List[Tuple[str, str]]) -> List[Tuple[str, str]]:
 361     """ Extract a list of categories. Currently a noop.
 362     """
 363     return categories
 364
 365 TParam = TypeVar('TParam', bound='LookupDetails') # pylint: disable=invalid-name
 366
 367 @dataclasses.dataclass
 368 class LookupDetails:
 369     """ Collection of parameters that define the amount of details
 370         returned with a lookup or details result.
 371     """
 372     geometry_output: GeometryFormat = GeometryFormat.NONE
 373     """ Add the full geometry of the place to the result. Multiple
 374         formats may be selected. Note that geometries can become quite large.
 375     """
 376     address_details: bool = False
 377     """ Get detailed information on the places that make up the address
 378         for the result.
 379     """
 380     linked_places: bool = False
 381     """ Get detailed information on the places that link to the result.
 382     """
 383     parented_places: bool = False
 384     """ Get detailed information on all places that this place is a parent
 385         for, i.e. all places for which it provides the address details.
 386         Only POI places can have parents.
 387     """
 388     keywords: bool = False
 389     """ Add information about the search terms used for this place.
 390     """
 391     geometry_simplification: float = 0.0
 392     """ Simplification factor for a geometry in degrees WGS. A factor of
 393         0.0 means the original geometry is kept. The higher the value, the
 394         more the geometry gets simplified.
 395     """
 396
 397     @classmethod
 398     def from_kwargs(cls: Type[TParam], kwargs: Dict[str, Any]) -> TParam:
 399         """ Load the data fields of the class from a dictionary.
 400             Unknown entries in the dictionary are ignored, missing ones
 401             get the default setting.
 402
 403             The function supports type checking and throws a UsageError
 404             when the value does not fit.
 405         """
 406         def _check_field(v: Any, field: 'dataclasses.Field[Any]') -> Any:
 407             if v is None:
 408                 return field.default_factory() \
 409                        if field.default_factory != dataclasses.MISSING \
 410                        else field.default
 411             if field.metadata and 'transform' in field.metadata:
 412                 return field.metadata['transform'](v)
 413             if not isinstance(v, field.type):
 414                 raise UsageError(f"Parameter '{field.name}' needs to be of {field.type!s}.")
 415             return v
 416
 417         return cls(**{f.name: _check_field(kwargs[f.name], f)
 418                       for f in dataclasses.fields(cls) if f.name in kwargs})
 419
 420
 421 @dataclasses.dataclass
 422 class ReverseDetails(LookupDetails):
 423     """ Collection of parameters for the reverse call.
 424     """
 425     max_rank: int = dataclasses.field(default=30,
 426                                       metadata={'transform': lambda v: max(0, min(v, 30))}
 427                                      )
 428     """ Highest address rank to return.
 429     """
 430     layers: DataLayer = DataLayer.ADDRESS | DataLayer.POI
 431     """ Filter which kind of data to include.
 432     """
 433
 434 @dataclasses.dataclass
 435 class SearchDetails(LookupDetails):
 436     """ Collection of parameters for the search call.
 437     """
 438     max_results: int = 10
 439     """ Maximum number of results to be returned. The actual number of results
 440         may be less.
 441     """
 442     min_rank: int = dataclasses.field(default=0,
 443                                       metadata={'transform': lambda v: max(0, min(v, 30))}
 444                                      )
 445     """ Lowest address rank to return.
 446     """
 447     max_rank: int = dataclasses.field(default=30,
 448                                       metadata={'transform': lambda v: max(0, min(v, 30))}
 449                                      )
 450     """ Highest address rank to return.
 451     """
 452     layers: Optional[DataLayer] = dataclasses.field(default=None,
 453                                                     metadata={'transform': lambda r : r})
 454     """ Filter which kind of data to include. When 'None' (the default) then
 455         filtering by layers is disabled.
 456     """
 457     countries: List[str] = dataclasses.field(default_factory=list,
 458                                              metadata={'transform': format_country})
 459     """ Restrict search results to the given countries. An empty list (the
 460         default) will disable this filter.
 461     """
 462     excluded: List[int] = dataclasses.field(default_factory=list,
 463                                             metadata={'transform': format_excluded})
 464     """ List of OSM objects to exclude from the results. Currently only
 465         works when the internal place ID is given.
 466         An empty list (the default) will disable this filter.
 467     """
 468     viewbox: Optional[Bbox] = dataclasses.field(default=None,
 469                                                 metadata={'transform': Bbox.from_param})
 470     """ Focus the search on a given map area.
 471     """
 472     bounded_viewbox: bool = False
 473     """ Use 'viewbox' as a filter and restrict results to places within the
 474         given area.
 475     """
 476     near: Optional[Point] = dataclasses.field(default=None,
 477                                               metadata={'transform': Point.from_param})
 478     """ Order results by distance to the given point.
 479     """
 480     near_radius: Optional[float] = dataclasses.field(default=None,
 481                                               metadata={'transform': lambda r : r})
 482     """ Use near point as a filter and drop results outside the given
 483         radius. Radius is given in degrees WSG84.
 484     """
 485     categories: List[Tuple[str, str]] = dataclasses.field(default_factory=list,
 486                                                           metadata={'transform': format_categories})
 487     """ Restrict search to places with one of the given class/type categories.
 488         An empty list (the default) will disable this filter.
 489     """
 490     viewbox_x2: Optional[Bbox] = None
 491
 492     def __post_init__(self) -> None:
 493         if self.viewbox is not None:
 494             xext = (self.viewbox.maxlon - self.viewbox.minlon)/2
 495             yext = (self.viewbox.maxlat - self.viewbox.minlat)/2
 496             self.viewbox_x2 = Bbox(self.viewbox.minlon - xext, self.viewbox.minlat - yext,
 497                                    self.viewbox.maxlon + xext, self.viewbox.maxlat + yext)
 498
 499
 500     def restrict_min_max_rank(self, new_min: int, new_max: int) -> None:
 501         """ Change the min_rank and max_rank fields to respect the
 502             given boundaries.
 503         """
 504         assert new_min <= new_max
 505         self.min_rank = max(self.min_rank, new_min)
 506         self.max_rank = min(self.max_rank, new_max)
 507
 508
 509     def is_impossible(self) -> bool:
 510         """ Check if the parameter configuration is contradictionary and
 511             cannot yield any results.
 512         """
 513         return (self.min_rank > self.max_rank
 514                 or (self.bounded_viewbox
 515                     and self.viewbox is not None and self.near is not None
 516                     and self.viewbox.contains(self.near))
 517                 or self.layers is not None and not self.layers)
 518
 519
 520     def layer_enabled(self, layer: DataLayer) -> bool:
 521         """ Check if the given layer has been choosen. Also returns
 522             true when layer restriction has been disabled completely.
 523         """
 524         return self.layers is None or bool(self.layers & layer)