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