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) 2025 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 datetime as dt
  15 import enum
  16 import math
  17 from struct import unpack
  18 from binascii import unhexlify
  19
  20 from .errors import UsageError
  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     def class_as_housenumber(self) -> Optional[int]:
  65         """ Interpret the class property as a housenumber and return it.
  66
  67             If the OSM ID points to an interpolation, then the class may be
  68             a number pointing to the exact number requested. This function
  69             returns the housenumber as an int, if class is set and is a number.
  70         """
  71         if self.osm_class and self.osm_class.isdigit():
  72             return int(self.osm_class)
  73         return None
  74
  75
  76 PlaceRef = Union[PlaceID, OsmID]
  77
  78
  79 class Point(NamedTuple):
  80     """ A geographic point in WGS84 projection.
  81     """
  82     x: float
  83     y: float
  84
  85     @property
  86     def lat(self) -> float:
  87         """ Return the latitude of the point.
  88         """
  89         return self.y
  90
  91     @property
  92     def lon(self) -> float:
  93         """ Return the longitude of the point.
  94         """
  95         return self.x
  96
  97     def to_geojson(self) -> str:
  98         """ Return the point in GeoJSON format.
  99         """
 100         return f'{{"type": "Point","coordinates": [{self.x}, {self.y}]}}'
 101
 102     @staticmethod
 103     def from_wkb(wkb: Union[str, bytes]) -> 'Point':
 104         """ Create a point from EWKB as returned from the database.
 105         """
 106         if isinstance(wkb, str):
 107             wkb = unhexlify(wkb)
 108         if len(wkb) != 25:
 109             raise ValueError(f"Point wkb has unexpected length {len(wkb)}")
 110         if wkb[0] == 0:
 111             gtype, srid, x, y = unpack('>iidd', wkb[1:])
 112         elif wkb[0] == 1:
 113             gtype, srid, x, y = unpack('<iidd', wkb[1:])
 114         else:
 115             raise ValueError("WKB has unknown endian value.")
 116
 117         if gtype != 0x20000001:
 118             raise ValueError("WKB must be a point geometry.")
 119         if srid != 4326:
 120             raise ValueError("Only WGS84 WKB supported.")
 121
 122         return Point(x, y)
 123
 124     @staticmethod
 125     def from_param(inp: Any) -> 'Point':
 126         """ Create a point from an input parameter. The parameter
 127             may be given as a point, a string or a sequence of
 128             strings or floats. Raises a UsageError if the format is
 129             not correct.
 130         """
 131         if isinstance(inp, Point):
 132             return inp
 133
 134         seq: Sequence[str]
 135         if isinstance(inp, str):
 136             seq = inp.split(',')
 137         elif isinstance(inp, abc.Sequence):
 138             seq = inp
 139
 140         if len(seq) != 2:
 141             raise UsageError('Point parameter needs 2 coordinates.')
 142         try:
 143             x, y = filter(math.isfinite, map(float, seq))
 144         except ValueError as exc:
 145             raise UsageError('Point parameter needs to be numbers.') from exc
 146
 147         if not -180 <= x <= 180 or not -90 <= y <= 90.0:
 148             raise UsageError('Point coordinates invalid.')
 149
 150         return Point(x, y)
 151
 152     def to_wkt(self) -> str:
 153         """ Return the WKT representation of the point.
 154         """
 155         return f'POINT({self.x} {self.y})'
 156
 157
 158 AnyPoint = Union[Point, Tuple[float, float]]
 159
 160 WKB_BBOX_HEADER_LE = b'\x01\x03\x00\x00\x20\xE6\x10\x00\x00\x01\x00\x00\x00\x05\x00\x00\x00'
 161 WKB_BBOX_HEADER_BE = b'\x00\x20\x00\x00\x03\x00\x00\x10\xe6\x00\x00\x00\x01\x00\x00\x00\x05'
 162
 163
 164 class Bbox:
 165     """ A bounding box in WGS84 projection.
 166
 167         The coordinates are available as an array in the 'coord'
 168         property in the order (minx, miny, maxx, maxy).
 169     """
 170     def __init__(self, minx: float, miny: float, maxx: float, maxy: float) -> None:
 171         """ Create a new bounding box with the given coordinates in WGS84
 172             projection.
 173         """
 174         self.coords = (minx, miny, maxx, maxy)
 175
 176     @property
 177     def minlat(self) -> float:
 178         """ Southern-most latitude, corresponding to the minimum y coordinate.
 179         """
 180         return self.coords[1]
 181
 182     @property
 183     def maxlat(self) -> float:
 184         """ Northern-most latitude, corresponding to the maximum y coordinate.
 185         """
 186         return self.coords[3]
 187
 188     @property
 189     def minlon(self) -> float:
 190         """ Western-most longitude, corresponding to the minimum x coordinate.
 191         """
 192         return self.coords[0]
 193
 194     @property
 195     def maxlon(self) -> float:
 196         """ Eastern-most longitude, corresponding to the maximum x coordinate.
 197         """
 198         return self.coords[2]
 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     def contains(self, pt: Point) -> bool:
 207         """ Check if the point is inside or on the boundary of the box.
 208         """
 209         return self.coords[0] <= pt[0] and self.coords[1] <= pt[1]\
 210             and self.coords[2] >= pt[0] and self.coords[3] >= pt[1]
 211
 212     def to_wkt(self) -> str:
 213         """ Return the WKT representation of the Bbox. This
 214             is a simple polygon with four points.
 215         """
 216         return 'POLYGON(({0} {1},{0} {3},{2} {3},{2} {1},{0} {1}))'\
 217             .format(*self.coords)
 218
 219     @staticmethod
 220     def from_wkb(wkb: Union[None, str, bytes]) -> 'Optional[Bbox]':
 221         """ Create a Bbox from a bounding box polygon as returned by
 222             the database. Returns `None` if the input value is None.
 223         """
 224         if wkb is None:
 225             return None
 226
 227         if isinstance(wkb, str):
 228             wkb = unhexlify(wkb)
 229
 230         if len(wkb) != 97:
 231             raise ValueError("WKB must be a bounding box polygon")
 232         if wkb.startswith(WKB_BBOX_HEADER_LE):
 233             x1, y1, _, _, x2, y2 = unpack('<dddddd', wkb[17:65])
 234         elif wkb.startswith(WKB_BBOX_HEADER_BE):
 235             x1, y1, _, _, x2, y2 = unpack('>dddddd', wkb[17:65])
 236         else:
 237             raise ValueError("WKB has wrong header")
 238
 239         return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2))
 240
 241     @staticmethod
 242     def from_point(pt: Point, buffer: float) -> 'Bbox':
 243         """ Return a Bbox around the point with the buffer added to all sides.
 244         """
 245         return Bbox(pt[0] - buffer, pt[1] - buffer,
 246                     pt[0] + buffer, pt[1] + buffer)
 247
 248     @staticmethod
 249     def from_param(inp: Any) -> 'Bbox':
 250         """ Return a Bbox from an input parameter. The box may be
 251             given as a Bbox, a string or a list or strings or integer.
 252             Raises a UsageError if the format is incorrect.
 253         """
 254         if isinstance(inp, Bbox):
 255             return inp
 256
 257         seq: Sequence[str]
 258         if isinstance(inp, str):
 259             seq = inp.split(',')
 260         elif isinstance(inp, abc.Sequence):
 261             seq = inp
 262
 263         if len(seq) != 4:
 264             raise UsageError('Bounding box parameter needs 4 coordinates.')
 265         try:
 266             x1, y1, x2, y2 = filter(math.isfinite, map(float, seq))
 267         except ValueError as exc:
 268             raise UsageError('Bounding box parameter needs to be numbers.') from exc
 269
 270         x1 = min(180, max(-180, x1))
 271         x2 = min(180, max(-180, x2))
 272         y1 = min(90, max(-90, y1))
 273         y2 = min(90, max(-90, y2))
 274
 275         if x1 == x2 or y1 == y2:
 276             raise UsageError('Bounding box with invalid parameters.')
 277
 278         return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2))
 279
 280
 281 class GeometryFormat(enum.Flag):
 282     """ All search functions support returning the full geometry of a place in
 283         various formats. The internal geometry is converted by PostGIS to
 284         the desired format and then returned as a string. It is possible to
 285         request multiple formats at the same time.
 286     """
 287     NONE = 0
 288     """ No geometry requested. Alias for a empty flag.
 289     """
 290     GEOJSON = enum.auto()
 291     """
 292     [GeoJSON](https://geojson.org/) format
 293     """
 294     KML = enum.auto()
 295     """
 296     [KML](https://en.wikipedia.org/wiki/Keyhole_Markup_Language) format
 297     """
 298     SVG = enum.auto()
 299     """
 300     [SVG](http://www.w3.org/TR/SVG/paths.html) format
 301     """
 302     TEXT = enum.auto()
 303     """
 304     [WKT](https://en.wikipedia.org/wiki/Well-known_text_representation_of_geometry) format
 305     """
 306
 307
 308 class DataLayer(enum.Flag):
 309     """ The `DataLayer` flag type defines the layers that can be selected
 310         for reverse and forward search.
 311     """
 312     ADDRESS = enum.auto()
 313     """ The address layer contains all places relevant for addresses:
 314         fully qualified addresses with a house number (or a house name equivalent,
 315         for some addresses) and places that can be part of an address like
 316         roads, cities, states.
 317     """
 318     POI = enum.auto()
 319     """ Layer for points of interest like shops, restaurants but also
 320         recycling bins or postboxes.
 321     """
 322     RAILWAY = enum.auto()
 323     """ Layer with railway features including tracks and other infrastructure.
 324         Note that in Nominatim's standard configuration, only very few railway
 325         features are imported into the database. Thus a custom configuration
 326         is required to make full use of this layer.
 327     """
 328     NATURAL = enum.auto()
 329     """ Layer with natural features like rivers, lakes and mountains.
 330     """
 331     MANMADE = enum.auto()
 332     """ Layer with other human-made features and boundaries. This layer is
 333         the catch-all and includes all features not covered by the other
 334         layers. A typical example for this layer are national park boundaries.
 335     """
 336
 337
 338 class QueryStatistics(dict[str, Any]):
 339     """ A specialised dictionary for collecting query statistics.
 340     """
 341
 342     def __enter__(self) -> 'QueryStatistics':
 343         self.log_time('start')
 344         return self
 345
 346     def __exit__(self, *_: Any) -> None:
 347         self.log_time('end')
 348         self['total_time'] = (self['end'] - self['start']).total_seconds()
 349         if 'start_query' in self:
 350             self['wait_time'] = (self['start_query'] - self['start']).total_seconds()
 351         else:
 352             self['wait_time'] = self['total_time']
 353             self['start_query'] = self['end']
 354         self['query_time'] = self['total_time'] - self['wait_time']
 355
 356     def __missing__(self, key: str) -> str:
 357         return ''
 358
 359     def log_time(self, key: str) -> None:
 360         self[key] = dt.datetime.now(tz=dt.timezone.utc)
 361
 362
 363 class NoQueryStats:
 364     """ Null object to use, when no query statistics are requested.
 365     """
 366
 367     def __enter__(self) -> 'NoQueryStats':
 368         return self
 369
 370     def __exit__(self, *_: Any) -> None:
 371         pass
 372
 373     def __setitem__(self, key: str, value: Any) -> None:
 374         pass
 375
 376     def __getitem__(self, key: str) -> Any:
 377         return None
 378
 379     def __contains__(self, key: str, default: Any = None) -> bool:
 380         return False
 381
 382     def log_time(self, key: str) -> None:
 383         pass
 384
 385
 386 def format_country(cc: Any) -> List[str]:
 387     """ Extract a list of country codes from the input which may be either
 388         a string or list of strings. Filters out all values that are not
 389         a two-letter string.
 390     """
 391     clist: Sequence[str]
 392     if isinstance(cc, str):
 393         clist = cc.split(',')
 394     elif isinstance(cc, abc.Sequence):
 395         clist = cc
 396     else:
 397         raise UsageError("Parameter 'country' needs to be a comma-separated list "
 398                          "or a Python list of strings.")
 399
 400     return [cc.lower() for cc in clist if isinstance(cc, str) and len(cc) == 2]
 401
 402
 403 def format_excluded(ids: Any) -> List[int]:
 404     """ Extract a list of place ids from the input which may be either
 405         a string or a list of strings or ints. Ignores empty value but
 406         throws a UserError on anything that cannot be converted to int.
 407     """
 408     plist: Sequence[str]
 409     if isinstance(ids, str):
 410         plist = [s.strip() for s in ids.split(',')]
 411     elif isinstance(ids, abc.Sequence):
 412         plist = ids
 413     else:
 414         raise UsageError("Parameter 'excluded' needs to be a comma-separated list "
 415                          "or a Python list of numbers.")
 416     if not all(isinstance(i, int) or
 417                (isinstance(i, str) and (not i or i.isdigit())) for i in plist):
 418         raise UsageError("Parameter 'excluded' only takes place IDs.")
 419
 420     return [int(id) for id in plist if id] or [0]
 421
 422
 423 def format_categories(categories: List[Tuple[str, str]]) -> List[Tuple[str, str]]:
 424     """ Extract a list of categories. Currently a noop.
 425     """
 426     return categories
 427
 428
 429 TParam = TypeVar('TParam', bound='LookupDetails')
 430
 431
 432 @dataclasses.dataclass
 433 class LookupDetails:
 434     """ Collection of parameters that define which kind of details are
 435         returned with a lookup or details result.
 436     """
 437     geometry_output: GeometryFormat = GeometryFormat.NONE
 438     """ Add the full geometry of the place to the result. Multiple
 439         formats may be selected. Note that geometries can become quite large.
 440     """
 441     address_details: bool = False
 442     """ Get detailed information on the places that make up the address
 443         for the result.
 444     """
 445     linked_places: bool = False
 446     """ Get detailed information on the places that link to the result.
 447     """
 448     parented_places: bool = False
 449     """ Get detailed information on all places that this place is a parent
 450         for, i.e. all places for which it provides the address details.
 451         Only POI places can have parents.
 452     """
 453     entrances: bool = False
 454     """ Get detailed information about the tagged entrances for the result.
 455     """
 456     keywords: bool = False
 457     """ Add information about the search terms used for this place.
 458     """
 459     geometry_simplification: float = 0.0
 460     """ Simplification factor for a geometry in degrees WGS. A factor of
 461         0.0 means the original geometry is kept. The higher the value, the
 462         more the geometry gets simplified.
 463     """
 464     query_stats: Union[QueryStatistics, NoQueryStats] = \
 465         dataclasses.field(default_factory=NoQueryStats)
 466     """ Optional QueryStatistics object collecting information about
 467         runtime behaviour of the call.
 468     """
 469
 470     @classmethod
 471     def from_kwargs(cls: Type[TParam], kwargs: Dict[str, Any]) -> TParam:
 472         """ Load the data fields of the class from a dictionary.
 473             Unknown entries in the dictionary are ignored, missing ones
 474             get the default setting.
 475
 476             The function supports type checking and throws a UsageError
 477             when the value does not fit.
 478         """
 479         def _check_field(v: Any, field: 'dataclasses.Field[Any]') -> Any:
 480             if v is None:
 481                 return field.default_factory() \
 482                        if field.default_factory != dataclasses.MISSING \
 483                        else field.default
 484             if field.metadata and 'transform' in field.metadata:
 485                 return field.metadata['transform'](v)
 486             if not isinstance(v, field.type):  # type: ignore[arg-type]
 487                 raise UsageError(f"Parameter '{field.name}' needs to be of {field.type!s}.")
 488             return v
 489
 490         return cls(**{f.name: _check_field(kwargs[f.name], f)
 491                       for f in dataclasses.fields(cls) if f.name in kwargs})
 492
 493
 494 @dataclasses.dataclass
 495 class ReverseDetails(LookupDetails):
 496     """ Collection of parameters for the reverse call.
 497     """
 498
 499     max_rank: int = dataclasses.field(default=30,
 500                                       metadata={'transform': lambda v: max(0, min(v, 30))})
 501     """ Highest address rank to return.
 502     """
 503
 504     layers: DataLayer = DataLayer.ADDRESS | DataLayer.POI
 505     """ Filter which kind of data to include.
 506     """
 507
 508
 509 @dataclasses.dataclass
 510 class SearchDetails(LookupDetails):
 511     """ Collection of parameters for the search call.
 512     """
 513     max_results: int = 10
 514     """ Maximum number of results to be returned. The actual number of results
 515         may be less.
 516     """
 517
 518     min_rank: int = dataclasses.field(default=0,
 519                                       metadata={'transform': lambda v: max(0, min(v, 30))})
 520     """ Lowest address rank to return.
 521     """
 522
 523     max_rank: int = dataclasses.field(default=30,
 524                                       metadata={'transform': lambda v: max(0, min(v, 30))})
 525     """ Highest address rank to return.
 526     """
 527
 528     layers: Optional[DataLayer] = dataclasses.field(default=None,
 529                                                     metadata={'transform': lambda r: r})
 530     """ Filter which kind of data to include. When 'None' (the default) then
 531         filtering by layers is disabled.
 532     """
 533
 534     countries: List[str] = dataclasses.field(default_factory=list,
 535                                              metadata={'transform': format_country})
 536     """ Restrict search results to the given countries. An empty list (the
 537         default) will disable this filter.
 538     """
 539
 540     excluded: List[int] = dataclasses.field(default_factory=list,
 541                                             metadata={'transform': format_excluded})
 542     """ List of OSM objects to exclude from the results. Currently only
 543         works when the internal place ID is given.
 544         An empty list (the default) will disable this filter.
 545     """
 546
 547     viewbox: Optional[Bbox] = dataclasses.field(default=None,
 548                                                 metadata={'transform': Bbox.from_param})
 549     """ Focus the search on a given map area.
 550     """
 551
 552     bounded_viewbox: bool = False
 553     """ Use 'viewbox' as a filter and restrict results to places within the
 554         given area.
 555     """
 556
 557     near: Optional[Point] = dataclasses.field(default=None,
 558                                               metadata={'transform': Point.from_param})
 559     """ Order results by distance to the given point.
 560     """
 561
 562     near_radius: Optional[float] = dataclasses.field(default=None,
 563                                                      metadata={'transform': lambda r: r})
 564     """ Use near point as a filter and drop results outside the given
 565         radius. Radius is given in degrees WSG84.
 566     """
 567
 568     categories: List[Tuple[str, str]] = dataclasses.field(default_factory=list,
 569                                                           metadata={'transform': format_categories})
 570     """ Restrict search to places with one of the given class/type categories.
 571         An empty list (the default) will disable this filter.
 572     """
 573
 574     viewbox_x2: Optional[Bbox] = None
 575
 576     def __post_init__(self) -> None:
 577         if self.viewbox is not None:
 578             xext = (self.viewbox.maxlon - self.viewbox.minlon)/2
 579             yext = (self.viewbox.maxlat - self.viewbox.minlat)/2
 580             self.viewbox_x2 = Bbox(self.viewbox.minlon - xext, self.viewbox.minlat - yext,
 581                                    self.viewbox.maxlon + xext, self.viewbox.maxlat + yext)
 582
 583     def restrict_min_max_rank(self, new_min: int, new_max: int) -> None:
 584         """ Change the min_rank and max_rank fields to respect the
 585             given boundaries.
 586         """
 587         assert new_min <= new_max
 588         self.min_rank = max(self.min_rank, new_min)
 589         self.max_rank = min(self.max_rank, new_max)
 590
 591     def is_impossible(self) -> bool:
 592         """ Check if the parameter configuration is contradictionary and
 593             cannot yield any results.
 594         """
 595         return (self.min_rank > self.max_rank
 596                 or (self.bounded_viewbox
 597                     and self.viewbox is not None and self.near is not None
 598                     and self.viewbox.contains(self.near))
 599                 or (self.layers is not None and not self.layers)
 600                 or (self.max_rank <= 4 and
 601                     self.layers is not None and not self.layers & DataLayer.ADDRESS))
 602
 603     def layer_enabled(self, layer: DataLayer) -> bool:
 604         """ Check if the given layer has been chosen. Also returns
 605             true when layer restriction has been disabled completely.
 606         """
 607         return self.layers is None or bool(self.layers & layer)
 608
 609
 610 @dataclasses.dataclass
 611 class EntranceDetails:
 612     """ Reference a place by its OSM ID and potentially the basic category.
 613
 614         The OSM ID may refer to places in the main table placex and OSM
 615         interpolation lines.
 616     """
 617     osm_id: int
 618     """ The OSM ID of the object.
 619     """
 620     type: str
 621     """ The value of the OSM entrance tag (i.e. yes, main, secondary, etc.).
 622     """
 623     location: Point
 624     """ The location of the entrance node.
 625     """
 626     extratags: Dict[str, str]
 627     """ The other tags associated with the entrance node.
 628     """