1 # SPDX-License-Identifier: GPL-3.0-or-later
3 # This file is part of Nominatim. (https://nominatim.org)
5 # Copyright (C) 2026 by the Nominatim developer community.
6 # For a full list of authors see the git log.
8 Complex datatypes used by the Nominatim API.
10 from typing import Optional, Union, Tuple, NamedTuple, TypeVar, Type, Dict, \
11 Any, List, Sequence, TYPE_CHECKING
12 from collections import abc
18 from struct import unpack
19 from binascii import unhexlify
22 from .localization import Locales
23 from .errors import UsageError
26 POSTCODE_REF_RE = re.compile(r'^P(?P<cc>[A-Za-z]{2}):(?P<postcode>.+)$')
27 POSTCODE_PARAM_RE = re.compile(r'^(?P<cc>[A-Za-z]{2}):(?P<postcode>.+)$')
30 @dataclasses.dataclass
32 """ Reference a place by Nominatim's internal ID.
34 A PlaceID may reference place from the main table placex, from
35 the interpolation tables or the postcode tables. Place IDs are not
36 stable between installations. You may use this type theefore only
37 with place IDs obtained from the same database.
41 The internal ID of the place to reference.
44 def __str__(self) -> str:
45 return str(self.place_id)
48 @dataclasses.dataclass
50 """ Reference a place by its OSM ID and potentially the basic category.
52 The OSM ID may refer to places in the main table placex and OSM
56 """ OSM type of the object. Must be one of `N`(node), `W`(way) or
60 """ The OSM ID of the object.
62 osm_class: Optional[str] = None
63 """ Optional restriction on the main tag of the object. It corresponds to
64 the key part of that tag. When it is given, then it must be equal to
65 the class of the object or the lookup returns nothing. Databases that
66 were migrated from a version before 5.4 may still hold one entry per
67 main tag. There the parameter selects between them and, when it is
68 left out, one of the entries is returned at random.
71 def __str__(self) -> str:
72 return f"{self.osm_type}{self.osm_id}"
74 def __post_init__(self) -> None:
75 if self.osm_type not in ('N', 'W', 'R'):
76 raise ValueError(f"Illegal OSM type '{self.osm_type}'. Must be one of N, W, R.")
78 def class_as_housenumber(self) -> Optional[int]:
79 """ Interpret the class property as a housenumber and return it.
81 If the OSM ID points to an interpolation, then the class may be
82 a number pointing to the exact number requested. This function
83 returns the housenumber as an int, if class is set and is a number.
85 if self.osm_class and self.osm_class.isdecimal():
86 return int(self.osm_class)
90 @dataclasses.dataclass
92 """ Reference an artificial postcode by country code and postcode.
97 def __post_init__(self) -> None:
98 if len(self.country_code) != 2 or not self.country_code.isalpha():
99 raise ValueError('Country code must be two letters.')
100 if not self.postcode:
101 raise ValueError('Postcode must not be empty.')
102 self.country_code = self.country_code.lower()
103 self.postcode = self.postcode.replace('_', ' ')
105 def __str__(self) -> str:
106 return f"P{self.country_code}:{self.place_id_part()}"
108 def place_id_part(self) -> str:
109 """Return a URL-safe identifier part for APIs.
111 return self.postcode.replace(' ', '_')
114 PlaceRef = Union[PlaceID, OsmID, PostcodeRef]
117 def parse_place_ref(ref: Any) -> PlaceRef:
118 """ Parse a stable place reference string.
120 if isinstance(ref, (PlaceID, OsmID, PostcodeRef)):
123 if not isinstance(ref, str):
124 raise UsageError("Parameter 'place_ref' must be a string.")
126 if match := POSTCODE_REF_RE.fullmatch(ref):
127 return PostcodeRef(match.group('cc'), match.group('postcode'))
129 raise UsageError(f"Invalid place_ref: {ref}")
132 def parse_postcode_param(ref: Any) -> PostcodeRef:
133 """Parse the /details postcode parameter.
135 if isinstance(ref, PostcodeRef):
138 if not isinstance(ref, str):
139 raise UsageError("Parameter 'postcode' must be a string.")
141 if match := POSTCODE_REF_RE.fullmatch(ref):
142 return PostcodeRef(match.group('cc'), match.group('postcode'))
144 if match := POSTCODE_PARAM_RE.fullmatch(ref):
145 return PostcodeRef(match.group('cc'), match.group('postcode'))
147 raise UsageError(f"Invalid postcode ID: {ref}")
150 class Point(NamedTuple):
151 """ A geographic point in WGS84 projection.
157 def lat(self) -> float:
158 """ Return the latitude of the point.
163 def lon(self) -> float:
164 """ Return the longitude of the point.
168 def to_geojson(self) -> str:
169 """ Return the point in GeoJSON format.
171 return f'{{"type": "Point","coordinates": [{self.x}, {self.y}]}}'
174 def from_wkb(wkb: Union[str, bytes]) -> 'Point':
175 """ Create a point from EWKB as returned from the database.
177 if isinstance(wkb, str):
180 raise ValueError(f"Point wkb has unexpected length {len(wkb)}")
182 gtype, srid, x, y = unpack('>iidd', wkb[1:])
184 gtype, srid, x, y = unpack('<iidd', wkb[1:])
186 raise ValueError("WKB has unknown endian value.")
188 if gtype != 0x20000001:
189 raise ValueError("WKB must be a point geometry.")
191 raise ValueError("Only WGS84 WKB supported.")
196 def from_param(inp: Any) -> 'Point':
197 """ Create a point from an input parameter. The parameter
198 may be given as a point, a string or a sequence of
199 strings or floats. Raises a UsageError if the format is
202 if isinstance(inp, Point):
206 if isinstance(inp, str):
208 elif isinstance(inp, abc.Sequence):
212 raise UsageError('Point parameter needs 2 coordinates.')
214 x, y = filter(math.isfinite, map(float, seq))
215 except ValueError as exc:
216 raise UsageError('Point parameter needs to be numbers.') from exc
218 if not -180 <= x <= 180 or not -90 <= y <= 90.0:
219 raise UsageError('Point coordinates invalid.')
223 def to_wkt(self) -> str:
224 """ Return the WKT representation of the point.
226 return f'POINT({self.x} {self.y})'
229 AnyPoint = Union[Point, Tuple[float, float]]
231 WKB_BBOX_HEADER_LE = b'\x01\x03\x00\x00\x20\xE6\x10\x00\x00\x01\x00\x00\x00\x05\x00\x00\x00'
232 WKB_BBOX_HEADER_BE = b'\x00\x20\x00\x00\x03\x00\x00\x10\xe6\x00\x00\x00\x01\x00\x00\x00\x05'
236 """ A bounding box in WGS84 projection.
238 The coordinates are available as an array in the 'coord'
239 property in the order (minx, miny, maxx, maxy).
241 def __init__(self, minx: float, miny: float, maxx: float, maxy: float) -> None:
242 """ Create a new bounding box with the given coordinates in WGS84
245 self.coords = (minx, miny, maxx, maxy)
248 def minlat(self) -> float:
249 """ Southern-most latitude, corresponding to the minimum y coordinate.
251 return self.coords[1]
254 def maxlat(self) -> float:
255 """ Northern-most latitude, corresponding to the maximum y coordinate.
257 return self.coords[3]
260 def minlon(self) -> float:
261 """ Western-most longitude, corresponding to the minimum x coordinate.
263 return self.coords[0]
266 def maxlon(self) -> float:
267 """ Eastern-most longitude, corresponding to the maximum x coordinate.
269 return self.coords[2]
272 def area(self) -> float:
273 """ Return the area of the box in WGS84.
275 return (self.coords[2] - self.coords[0]) * (self.coords[3] - self.coords[1])
277 def contains(self, pt: Point) -> bool:
278 """ Check if the point is inside or on the boundary of the box.
280 return self.coords[0] <= pt[0] and self.coords[1] <= pt[1]\
281 and self.coords[2] >= pt[0] and self.coords[3] >= pt[1]
283 def to_wkt(self) -> str:
284 """ Return the WKT representation of the Bbox. This
285 is a simple polygon with four points.
287 return 'POLYGON(({0} {1},{0} {3},{2} {3},{2} {1},{0} {1}))'\
288 .format(*self.coords)
291 def from_wkb(wkb: Union[None, str, bytes]) -> 'Optional[Bbox]':
292 """ Create a Bbox from a bounding box polygon as returned by
293 the database. Returns `None` if the input value is None.
298 if isinstance(wkb, str):
302 raise ValueError("WKB must be a bounding box polygon")
303 if wkb.startswith(WKB_BBOX_HEADER_LE):
304 x1, y1, _, _, x2, y2 = unpack('<dddddd', wkb[17:65])
305 elif wkb.startswith(WKB_BBOX_HEADER_BE):
306 x1, y1, _, _, x2, y2 = unpack('>dddddd', wkb[17:65])
308 raise ValueError("WKB has wrong header")
310 return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2))
313 def from_point(pt: Point, buffer: float) -> 'Bbox':
314 """ Return a Bbox around the point with the buffer added to all sides.
316 return Bbox(pt[0] - buffer, pt[1] - buffer,
317 pt[0] + buffer, pt[1] + buffer)
320 def from_param(inp: Any) -> 'Bbox':
321 """ Return a Bbox from an input parameter. The box may be
322 given as a Bbox, a string or a list or strings or integer.
323 Raises a UsageError if the format is incorrect.
325 if isinstance(inp, Bbox):
329 if isinstance(inp, str):
331 elif isinstance(inp, abc.Sequence):
335 raise UsageError('Bounding box parameter needs 4 coordinates.')
337 x1, y1, x2, y2 = filter(math.isfinite, map(float, seq))
338 except ValueError as exc:
339 raise UsageError('Bounding box parameter needs to be numbers.') from exc
341 x1 = min(180, max(-180, x1))
342 x2 = min(180, max(-180, x2))
343 y1 = min(90, max(-90, y1))
344 y2 = min(90, max(-90, y2))
346 if x1 == x2 or y1 == y2:
347 raise UsageError('Bounding box with invalid parameters.')
349 return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2))
352 class GeometryFormat(enum.Flag):
353 """ All search functions support returning the full geometry of a place in
354 various formats. The internal geometry is converted by PostGIS to
355 the desired format and then returned as a string. It is possible to
356 request multiple formats at the same time.
359 """ No geometry requested. Alias for a empty flag.
361 GEOJSON = enum.auto()
363 [GeoJSON](https://geojson.org/) format
367 [KML](https://en.wikipedia.org/wiki/Keyhole_Markup_Language) format
371 [SVG](http://www.w3.org/TR/SVG/paths.html) format
375 [WKT](https://en.wikipedia.org/wiki/Well-known_text_representation_of_geometry) format
379 class DataLayer(enum.Flag):
380 """ The `DataLayer` flag type defines the layers that can be selected
381 for reverse and forward search.
383 ADDRESS = enum.auto()
384 """ The address layer contains all places relevant for addresses:
385 fully qualified addresses with a house number (or a house name equivalent,
386 for some addresses) and places that can be part of an address like
387 roads, cities, states.
390 """ Layer for points of interest like shops, restaurants but also
391 recycling bins or postboxes.
393 RAILWAY = enum.auto()
394 """ Layer with railway features including tracks and other infrastructure.
395 Note that in Nominatim's standard configuration, only very few railway
396 features are imported into the database. Thus a custom configuration
397 is required to make full use of this layer.
399 NATURAL = enum.auto()
400 """ Layer with natural features like rivers, lakes and mountains.
402 MANMADE = enum.auto()
403 """ Layer with other human-made features and boundaries. This layer is
404 the catch-all and includes all features not covered by the other
405 layers. A typical example for this layer are national park boundaries.
409 class QueryStatistics(dict[str, Any]):
410 """ A specialised dictionary for collecting query statistics.
413 def __enter__(self) -> 'QueryStatistics':
414 self.log_time('start')
417 def __exit__(self, *_: Any) -> None:
419 self['total_time'] = (self['end'] - self['start']).total_seconds()
420 if 'start_query' in self:
421 self['wait_time'] = (self['start_query'] - self['start']).total_seconds()
423 self['wait_time'] = self['total_time']
424 self['start_query'] = self['end']
425 self['query_time'] = self['total_time'] - self['wait_time']
427 def __missing__(self, key: str) -> str:
430 def log_time(self, key: str) -> None:
431 self[key] = dt.datetime.now(tz=dt.timezone.utc)
435 """ Null object to use, when no query statistics are requested.
438 def __enter__(self) -> 'NoQueryStats':
441 def __exit__(self, *_: Any) -> None:
444 def __setitem__(self, key: str, value: Any) -> None:
447 def __getitem__(self, key: str) -> Any:
450 def __contains__(self, key: str, default: Any = None) -> bool:
453 def log_time(self, key: str) -> None:
457 def format_country(cc: Any) -> List[str]:
458 """ Extract a list of country codes from the input which may be either
459 a string or list of strings. Filters out all values that are not
463 if isinstance(cc, str):
464 clist = cc.split(',')
465 elif isinstance(cc, abc.Sequence):
468 raise UsageError("Parameter 'country' needs to be a comma-separated list "
469 "or a Python list of strings.")
471 return [cc.lower() for cc in clist if isinstance(cc, str) and len(cc) == 2]
474 def format_excluded(ids: Any) -> List[PlaceRef]:
475 """ Extract a list of place IDs and OSM IDs from the input.
481 if isinstance(ids, str):
482 plist = [s.strip() for s in ids.split(',')]
483 elif isinstance(ids, abc.Sequence):
486 raise UsageError("Parameter 'excluded' needs to be a comma-separated list "
487 "or a Python list of place IDs or OSM IDs.")
489 result: List[PlaceRef] = []
493 if isinstance(i, (PlaceID, OsmID, PostcodeRef)):
495 elif isinstance(i, int):
497 result.append(PlaceID(i))
498 elif isinstance(i, str):
501 result.append(PlaceID(int(i)))
502 elif len(i) > 1 and i[0].upper() in ('N', 'W', 'R') and i[1:].isdecimal():
504 result.append(OsmID(i[0].upper(), int(i[1:])))
505 elif match := POSTCODE_REF_RE.fullmatch(i):
506 result.append(PostcodeRef(match.group('cc'), match.group('postcode')))
508 raise UsageError(f"Invalid exclude ID: {i}")
510 raise UsageError("Parameter 'excluded' contains invalid types.")
515 def format_categories(categories: List[Tuple[str, str]]) -> List[Tuple[str, str]]:
516 """ Extract a list of categories. Currently a noop.
521 CATEGORY_RE = re.compile(r'[A-Za-z0-9_]{1,255}(\.[A-Za-z0-9_]{1,255})+')
524 def format_category_filters(filters: Any) -> list[list[str]]:
525 """ Parse category filters into a list of groups of categories.
527 Each entry of 'filters' is a comma-separated list of categories which
528 together form one group. A category must consist of at least two
529 dot-separated labels made up of letters, digits and underscores.
531 if isinstance(filters, str):
535 for group in filters:
536 if not isinstance(group, str):
537 raise UsageError("Category filters must be strings.")
539 for category in group.split(','):
540 category = category.strip()
541 if not CATEGORY_RE.fullmatch(category):
542 raise UsageError(f"Invalid category '{category}'. A category must consist"
543 " of at least two labels separated by dots.")
544 categories.append(category)
545 result.append(categories)
550 TParam = TypeVar('TParam', bound='LookupDetails')
553 @dataclasses.dataclass
555 """ Collection of parameters that define which kind of details are
556 returned with a lookup or details result.
558 geometry_output: GeometryFormat = GeometryFormat.NONE
559 """ Add the full geometry of the place to the result. Multiple
560 formats may be selected. Note that geometries can become quite large.
562 address_details: bool = False
563 """ Get detailed information on the places that make up the address
566 linked_places: bool = False
567 """ Get detailed information on the places that link to the result.
569 parented_places: bool = False
570 """ Get detailed information on all places that this place is a parent
571 for, i.e. all places for which it provides the address details.
572 Only POI places can have parents.
574 entrances: bool = False
575 """ Get detailed information about the tagged entrances for the result.
577 keywords: bool = False
578 """ Add information about the search terms used for this place.
580 geometry_simplification: float = 0.0
581 """ Simplification factor for a geometry in degrees WGS. A factor of
582 0.0 means the original geometry is kept. The higher the value, the
583 more the geometry gets simplified.
585 query_stats: Union[QueryStatistics, NoQueryStats] = \
586 dataclasses.field(default_factory=NoQueryStats)
587 """ Optional QueryStatistics object collecting information about
588 runtime behaviour of the call.
592 def from_kwargs(cls: Type[TParam], kwargs: Dict[str, Any]) -> TParam:
593 """ Load the data fields of the class from a dictionary.
594 Unknown entries in the dictionary are ignored, missing ones
595 get the default setting.
597 The function supports type checking and throws a UsageError
598 when the value does not fit.
600 def _check_field(v: Any, field: 'dataclasses.Field[Any]') -> Any:
602 return field.default_factory() \
603 if field.default_factory != dataclasses.MISSING \
605 if field.metadata and 'transform' in field.metadata:
606 return field.metadata['transform'](v)
607 if not isinstance(v, field.type): # type: ignore[arg-type]
608 raise UsageError(f"Parameter '{field.name}' needs to be of {field.type!s}.")
611 return cls(**{f.name: _check_field(kwargs[f.name], f)
612 for f in dataclasses.fields(cls) if f.name in kwargs})
615 @dataclasses.dataclass
616 class ReverseDetails(LookupDetails):
617 """ Collection of parameters for the reverse call.
620 max_rank: int = dataclasses.field(default=30,
621 metadata={'transform': lambda v: max(0, min(v, 30))})
622 """ Highest address rank to return.
625 layers: DataLayer = DataLayer.ADDRESS | DataLayer.POI
626 """ Filter which kind of data to include.
630 @dataclasses.dataclass
631 class SearchDetails(LookupDetails):
632 """ Collection of parameters for the search call.
634 max_results: int = 10
635 """ Maximum number of results to be returned. The actual number of results
639 min_rank: int = dataclasses.field(default=0,
640 metadata={'transform': lambda v: max(0, min(v, 30))})
641 """ Lowest address rank to return.
644 max_rank: int = dataclasses.field(default=30,
645 metadata={'transform': lambda v: max(0, min(v, 30))})
646 """ Highest address rank to return.
649 layers: Optional[DataLayer] = dataclasses.field(default=None,
650 metadata={'transform': lambda r: r})
651 """ Filter which kind of data to include. When 'None' (the default) then
652 filtering by layers is disabled.
655 countries: List[str] = dataclasses.field(default_factory=list,
656 metadata={'transform': format_country})
657 """ Restrict search results to the given countries. An empty list (the
658 default) will disable this filter.
661 excluded: List[PlaceRef] = dataclasses.field(default_factory=list,
662 metadata={'transform': format_excluded})
663 """ List of OSM objects to exclude from the results,
664 provided as either internal Place IDs or OSM IDs.
667 viewbox: Optional[Bbox] = dataclasses.field(default=None,
668 metadata={'transform': Bbox.from_param})
669 """ Focus the search on a given map area.
672 bounded_viewbox: bool = False
673 """ Use 'viewbox' as a filter and restrict results to places within the
677 near: Optional[Point] = dataclasses.field(default=None,
678 metadata={'transform': Point.from_param})
679 """ Order results by distance to the given point.
682 near_radius: Optional[float] = dataclasses.field(default=None,
683 metadata={'transform': lambda r: r})
684 """ Use near point as a filter and drop results outside the given
685 radius. Radius is given in degrees WSG84.
688 categories: List[Tuple[str, str]] = dataclasses.field(default_factory=list,
689 metadata={'transform': format_categories})
690 """ Restrict search to places with one of the given main tags, given as
691 class/type pairs. An empty list (the default) will disable this filter.
693 Deprecated. Use `include` instead, which restricts on the same
694 information in its hierarchical form.
697 include: list[list[str]] = \
698 dataclasses.field(default_factory=list,
699 metadata={'transform': format_category_filters})
700 """ Restrict search to places matching the given categories. A place must
701 match at least one category of every group and matches a category when
702 it is assigned the category itself or one of its descendants. An empty
703 list (the default) will disable this filter.
706 exclude: list[list[str]] = \
707 dataclasses.field(default_factory=list,
708 metadata={'transform': format_category_filters})
709 """ Drop places matching the given categories from the results. A place is
710 dropped when it matches all categories of any of the groups. An empty
711 list (the default) will disable this filter.
714 viewbox_x2: Optional[Bbox] = None
716 locales: Optional['Locales'] = dataclasses.field(
717 default=None, metadata={'transform': lambda v: v})
718 """ Locale preferences of the caller.
719 Used during result re-ranking to prefer results that match the
720 caller's locale over results that only match in an alternate language.
723 def __post_init__(self) -> None:
724 if self.viewbox is not None:
725 xext = (self.viewbox.maxlon - self.viewbox.minlon)/2
726 yext = (self.viewbox.maxlat - self.viewbox.minlat)/2
727 self.viewbox_x2 = Bbox(self.viewbox.minlon - xext, self.viewbox.minlat - yext,
728 self.viewbox.maxlon + xext, self.viewbox.maxlat + yext)
731 def excluded_place_ids(self) -> List[int]:
732 """ Return excluded entries as a plain list of place ID integers.
733 Only includes PlaceID entries. Returns [0] if empty to
734 ensure SQL NOT IN clauses work correctly.
736 return [e.place_id for e in self.excluded if isinstance(e, PlaceID)] or [0]
738 def restrict_min_max_rank(self, new_min: int, new_max: int) -> None:
739 """ Change the min_rank and max_rank fields to respect the
742 assert new_min <= new_max
743 self.min_rank = max(self.min_rank, new_min)
744 self.max_rank = min(self.max_rank, new_max)
746 def is_impossible(self) -> bool:
747 """ Check if the parameter configuration is contradictionary and
748 cannot yield any results.
750 return (self.min_rank > self.max_rank
751 or (self.bounded_viewbox
752 and self.viewbox is not None and self.near is not None
753 and self.viewbox.contains(self.near))
754 or (self.layers is not None and not self.layers)
755 or (self.max_rank <= 4 and
756 self.layers is not None and not self.layers & DataLayer.ADDRESS))
758 def layer_enabled(self, layer: DataLayer) -> bool:
759 """ Check if the given layer has been chosen. Also returns
760 true when layer restriction has been disabled completely.
762 return self.layers is None or bool(self.layers & layer)
765 @dataclasses.dataclass
766 class EntranceDetails:
767 """ Reference a place by its OSM ID and potentially the basic category.
769 The OSM ID may refer to places in the main table placex and OSM
773 """ The OSM ID of the object.
776 """ The value of the OSM entrance tag (i.e. yes, main, secondary, etc.).
779 """ The location of the entrance node.
781 extratags: Dict[str, str]
782 """ The other tags associated with the entrance node.