Merge remote-tracking branch 'upstream/master'

[nominatim.git] / nominatim / api / types.py

diff --git a/nominatim/api/types.py b/nominatim/api/types.py
index c7e15843b551da95579b67f1389488d8b18d396e..a2a265c033e88fec87965b1643d2d42d2a0e1e37 100644 (file)
--- a/nominatim/api/types.py
+++ b/nominatim/api/types.py
@@ -14,9 +14,7 @@ import dataclasses
 import enum
 import math
 from struct import unpack
 import enum
 import math
 from struct import unpack

-
-from geoalchemy2 import WKTElement
-import geoalchemy2.functions
+from binascii import unhexlify

 
 from nominatim.errors import UsageError
 
 
 from nominatim.errors import UsageError
 


@@ -73,11 +71,13 @@ class Point(NamedTuple):
 
 
     @staticmethod
 
 
     @staticmethod

-    def from_wkb(wkb: bytes) -> 'Point':
+    def from_wkb(wkb: Union[str, bytes]) -> 'Point':

         """ Create a point from EWKB as returned from the database.
         """
         """ Create a point from EWKB as returned from the database.
         """

+        if isinstance(wkb, str):
+            wkb = unhexlify(wkb)

         if len(wkb) != 25:
         if len(wkb) != 25:

-            raise ValueError("Point wkb has unexpected length")
+            raise ValueError(f"Point wkb has unexpected length {len(wkb)}")

         if wkb[0] == 0:
             gtype, srid, x, y = unpack('>iidd', wkb[1:])
         elif wkb[0] == 1:
         if wkb[0] == 0:
             gtype, srid, x, y = unpack('>iidd', wkb[1:])
         elif wkb[0] == 1:


@@ -122,10 +122,10 @@ class Point(NamedTuple):
         return Point(x, y)
 
 
         return Point(x, y)
 
 

-    def sql_value(self) -> WKTElement:
-        """ Create an SQL expression for the point.
+    def to_wkt(self) -> str:
+        """ Return the WKT representation of the point.

         """
         """

-        return WKTElement(f'POINT({self.x} {self.y})', srid=4326)
+        return f'POINT({self.x} {self.y})'

 
 
 
 
 
 


@@ -179,12 +179,6 @@ class Bbox:
         return (self.coords[2] - self.coords[0]) * (self.coords[3] - self.coords[1])
 
 
         return (self.coords[2] - self.coords[0]) * (self.coords[3] - self.coords[1])
 
 

-    def sql_value(self) -> Any:
-        """ Create an SQL expression for the box.
-        """
-        return geoalchemy2.functions.ST_MakeEnvelope(*self.coords, 4326)
-
-

     def contains(self, pt: Point) -> bool:
         """ Check if the point is inside or on the boundary of the box.
         """
     def contains(self, pt: Point) -> bool:
         """ Check if the point is inside or on the boundary of the box.
         """


@@ -192,14 +186,25 @@ class Bbox:
                and self.coords[2] >= pt[0] and self.coords[3] >= pt[1]
 
 
                and self.coords[2] >= pt[0] and self.coords[3] >= pt[1]
 
 

+    def to_wkt(self) -> str:
+        """ Return the WKT representation of the Bbox. This
+            is a simple polygon with four points.
+        """
+        return 'POLYGON(({0} {1},{0} {3},{2} {3},{2} {1},{0} {1}))'\
+                  .format(*self.coords) # pylint: disable=consider-using-f-string
+
+

     @staticmethod
     @staticmethod

-    def from_wkb(wkb: Optional[bytes]) -> 'Optional[Bbox]':
+    def from_wkb(wkb: Union[None, str, bytes]) -> 'Optional[Bbox]':

         """ Create a Bbox from a bounding box polygon as returned by
             the database. Return s None if the input value is None.
         """
         if wkb is None:
             return None
 
         """ Create a Bbox from a bounding box polygon as returned by
             the database. Return s None if the input value is None.
         """
         if wkb is None:
             return None
 

+        if isinstance(wkb, str):
+            wkb = unhexlify(wkb)
+

         if len(wkb) != 97:
             raise ValueError("WKB must be a bounding box polygon")
         if wkb.startswith(WKB_BBOX_HEADER_LE):
         if len(wkb) != 97:
             raise ValueError("WKB must be a bounding box polygon")
         if wkb.startswith(WKB_BBOX_HEADER_LE):


@@ -242,9 +247,10 @@ class Bbox:
         except ValueError as exc:
             raise UsageError('Bounding box parameter needs to be numbers.') from exc
 
         except ValueError as exc:
             raise UsageError('Bounding box parameter needs to be numbers.') from exc
 

-        if x1 < -180.0 or x1 > 180.0 or y1 < -90.0 or y1 > 90.0 \
-           or x2 < -180.0 or x2 > 180.0 or y2 < -90.0 or y2 > 90.0:
-            raise UsageError('Bounding box coordinates invalid.')
+        x1 = min(180, max(-180, x1))
+        x2 = min(180, max(-180, x2))
+        y1 = min(90, max(-90, y1))
+        y2 = min(90, max(-90, y2))

 
         if x1 == x2 or y1 == y2:
             raise UsageError('Bounding box with invalid parameters.')
 
         if x1 == x2 or y1 == y2:
             raise UsageError('Bounding box with invalid parameters.')


@@ -302,10 +308,11 @@ def format_excluded(ids: Any) -> List[int]:
     else:
         raise UsageError("Parameter 'excluded' needs to be a comma-separated list "
                          "or a Python list of numbers.")
     else:
         raise UsageError("Parameter 'excluded' needs to be a comma-separated list "
                          "or a Python list of numbers.")

-    if not all(isinstance(i, int) or (isinstance(i, str) and i.isdigit()) for i in plist):
+    if not all(isinstance(i, int) or
+               (isinstance(i, str) and (not i or i.isdigit())) for i in plist):

         raise UsageError("Parameter 'excluded' only takes place IDs.")
 
         raise UsageError("Parameter 'excluded' only takes place IDs.")
 

-    return [int(id) for id in plist if id]
+    return [int(id) for id in plist if id] or [0]

 
 
 def format_categories(categories: List[Tuple[str, str]]) -> List[Tuple[str, str]]:
 
 
 def format_categories(categories: List[Tuple[str, str]]) -> List[Tuple[str, str]]:


@@ -438,6 +445,7 @@ class SearchDetails(LookupDetails):
     """ Restrict search to places with one of the given class/type categories.
         An empty list (the default) will disable this filter.
     """
     """ Restrict search to places with one of the given class/type categories.
         An empty list (the default) will disable this filter.
     """

+    viewbox_x2: Optional[Bbox] = None

 
     def __post_init__(self) -> None:
         if self.viewbox is not None:
 
     def __post_init__(self) -> None:
         if self.viewbox is not None: