| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395 |
- from django.contrib.gis.db.models.fields import BaseSpatialField
- from django.contrib.gis.measure import Distance
- from django.db import NotSupportedError
- from django.db.models import Expression, Lookup, Transform
- from django.db.models.sql.query import Query
- from django.utils.regex_helper import _lazy_re_compile
- class RasterBandTransform(Transform):
- def as_sql(self, compiler, connection):
- return compiler.compile(self.lhs)
- class GISLookup(Lookup):
- sql_template = None
- transform_func = None
- distance = False
- band_rhs = None
- band_lhs = None
- def __init__(self, lhs, rhs):
- rhs, *self.rhs_params = rhs if isinstance(rhs, (list, tuple)) else [rhs]
- super().__init__(lhs, rhs)
- self.template_params = {}
- self.process_rhs_params()
- def process_rhs_params(self):
- if self.rhs_params:
- # Check if a band index was passed in the query argument.
- if len(self.rhs_params) == (2 if self.lookup_name == "relate" else 1):
- self.process_band_indices()
- elif len(self.rhs_params) > 1:
- raise ValueError("Tuple too long for lookup %s." % self.lookup_name)
- elif isinstance(self.lhs, RasterBandTransform):
- self.process_band_indices(only_lhs=True)
- def process_band_indices(self, only_lhs=False):
- """
- Extract the lhs band index from the band transform class and the rhs
- band index from the input tuple.
- """
- # PostGIS band indices are 1-based, so the band index needs to be
- # increased to be consistent with the GDALRaster band indices.
- if only_lhs:
- self.band_rhs = 1
- self.band_lhs = self.lhs.band_index + 1
- return
- if isinstance(self.lhs, RasterBandTransform):
- self.band_lhs = self.lhs.band_index + 1
- else:
- self.band_lhs = 1
- self.band_rhs, *self.rhs_params = self.rhs_params
- def get_db_prep_lookup(self, value, connection):
- # get_db_prep_lookup is called by process_rhs from super class
- return ("%s", [connection.ops.Adapter(value)])
- def process_rhs(self, compiler, connection):
- if isinstance(self.rhs, Query):
- # If rhs is some Query, don't touch it.
- return super().process_rhs(compiler, connection)
- if isinstance(self.rhs, Expression):
- self.rhs = self.rhs.resolve_expression(compiler.query)
- rhs, rhs_params = super().process_rhs(compiler, connection)
- placeholder = connection.ops.get_geom_placeholder(
- self.lhs.output_field, self.rhs, compiler
- )
- return placeholder % rhs, rhs_params
- def get_rhs_op(self, connection, rhs):
- # Unlike BuiltinLookup, the GIS get_rhs_op() implementation should return
- # an object (SpatialOperator) with an as_sql() method to allow for more
- # complex computations (where the lhs part can be mixed in).
- return connection.ops.gis_operators[self.lookup_name]
- def as_sql(self, compiler, connection):
- lhs_sql, lhs_params = self.process_lhs(compiler, connection)
- rhs_sql, rhs_params = self.process_rhs(compiler, connection)
- sql_params = (*lhs_params, *rhs_params)
- template_params = {
- "lhs": lhs_sql,
- "rhs": rhs_sql,
- "value": "%s",
- **self.template_params,
- }
- rhs_op = self.get_rhs_op(connection, rhs_sql)
- return rhs_op.as_sql(connection, self, template_params, sql_params)
- # ------------------
- # Geometry operators
- # ------------------
- @BaseSpatialField.register_lookup
- class OverlapsLeftLookup(GISLookup):
- """
- The overlaps_left operator returns true if A's bounding box overlaps or is to the
- left of B's bounding box.
- """
- lookup_name = "overlaps_left"
- @BaseSpatialField.register_lookup
- class OverlapsRightLookup(GISLookup):
- """
- The 'overlaps_right' operator returns true if A's bounding box overlaps or is to the
- right of B's bounding box.
- """
- lookup_name = "overlaps_right"
- @BaseSpatialField.register_lookup
- class OverlapsBelowLookup(GISLookup):
- """
- The 'overlaps_below' operator returns true if A's bounding box overlaps or is below
- B's bounding box.
- """
- lookup_name = "overlaps_below"
- @BaseSpatialField.register_lookup
- class OverlapsAboveLookup(GISLookup):
- """
- The 'overlaps_above' operator returns true if A's bounding box overlaps or is above
- B's bounding box.
- """
- lookup_name = "overlaps_above"
- @BaseSpatialField.register_lookup
- class LeftLookup(GISLookup):
- """
- The 'left' operator returns true if A's bounding box is strictly to the left
- of B's bounding box.
- """
- lookup_name = "left"
- @BaseSpatialField.register_lookup
- class RightLookup(GISLookup):
- """
- The 'right' operator returns true if A's bounding box is strictly to the right
- of B's bounding box.
- """
- lookup_name = "right"
- @BaseSpatialField.register_lookup
- class StrictlyBelowLookup(GISLookup):
- """
- The 'strictly_below' operator returns true if A's bounding box is strictly below B's
- bounding box.
- """
- lookup_name = "strictly_below"
- @BaseSpatialField.register_lookup
- class StrictlyAboveLookup(GISLookup):
- """
- The 'strictly_above' operator returns true if A's bounding box is strictly above B's
- bounding box.
- """
- lookup_name = "strictly_above"
- @BaseSpatialField.register_lookup
- class SameAsLookup(GISLookup):
- """
- The "~=" operator is the "same as" operator. It tests actual geometric
- equality of two features. So if A and B are the same feature,
- vertex-by-vertex, the operator returns true.
- """
- lookup_name = "same_as"
- BaseSpatialField.register_lookup(SameAsLookup, "exact")
- @BaseSpatialField.register_lookup
- class BBContainsLookup(GISLookup):
- """
- The 'bbcontains' operator returns true if A's bounding box completely contains
- by B's bounding box.
- """
- lookup_name = "bbcontains"
- @BaseSpatialField.register_lookup
- class BBOverlapsLookup(GISLookup):
- """
- The 'bboverlaps' operator returns true if A's bounding box overlaps B's
- bounding box.
- """
- lookup_name = "bboverlaps"
- @BaseSpatialField.register_lookup
- class ContainedLookup(GISLookup):
- """
- The 'contained' operator returns true if A's bounding box is completely contained
- by B's bounding box.
- """
- lookup_name = "contained"
- # ------------------
- # Geometry functions
- # ------------------
- @BaseSpatialField.register_lookup
- class ContainsLookup(GISLookup):
- lookup_name = "contains"
- @BaseSpatialField.register_lookup
- class ContainsProperlyLookup(GISLookup):
- lookup_name = "contains_properly"
- @BaseSpatialField.register_lookup
- class CoveredByLookup(GISLookup):
- lookup_name = "coveredby"
- @BaseSpatialField.register_lookup
- class CoversLookup(GISLookup):
- lookup_name = "covers"
- @BaseSpatialField.register_lookup
- class CrossesLookup(GISLookup):
- lookup_name = "crosses"
- @BaseSpatialField.register_lookup
- class DisjointLookup(GISLookup):
- lookup_name = "disjoint"
- @BaseSpatialField.register_lookup
- class EqualsLookup(GISLookup):
- lookup_name = "equals"
- @BaseSpatialField.register_lookup
- class IntersectsLookup(GISLookup):
- lookup_name = "intersects"
- @BaseSpatialField.register_lookup
- class OverlapsLookup(GISLookup):
- lookup_name = "overlaps"
- @BaseSpatialField.register_lookup
- class RelateLookup(GISLookup):
- lookup_name = "relate"
- sql_template = "%(func)s(%(lhs)s, %(rhs)s, %%s)"
- pattern_regex = _lazy_re_compile(r"^[012TF*]{9}$")
- def process_rhs(self, compiler, connection):
- # Check the pattern argument
- pattern = self.rhs_params[0]
- backend_op = connection.ops.gis_operators[self.lookup_name]
- if hasattr(backend_op, "check_relate_argument"):
- backend_op.check_relate_argument(pattern)
- elif not isinstance(pattern, str) or not self.pattern_regex.match(pattern):
- raise ValueError('Invalid intersection matrix pattern "%s".' % pattern)
- sql, params = super().process_rhs(compiler, connection)
- return sql, params + [pattern]
- @BaseSpatialField.register_lookup
- class TouchesLookup(GISLookup):
- lookup_name = "touches"
- @BaseSpatialField.register_lookup
- class WithinLookup(GISLookup):
- lookup_name = "within"
- class DistanceLookupBase(GISLookup):
- distance = True
- sql_template = "%(func)s(%(lhs)s, %(rhs)s) %(op)s %(value)s"
- def process_rhs_params(self):
- if not 1 <= len(self.rhs_params) <= 3:
- raise ValueError(
- "2, 3, or 4-element tuple required for '%s' lookup." % self.lookup_name
- )
- elif len(self.rhs_params) == 3 and self.rhs_params[2] != "spheroid":
- raise ValueError(
- "For 4-element tuples the last argument must be the 'spheroid' "
- "directive."
- )
- # Check if the second parameter is a band index.
- if len(self.rhs_params) > 1 and self.rhs_params[1] != "spheroid":
- self.process_band_indices()
- def process_distance(self, compiler, connection):
- dist_param = self.rhs_params[0]
- return (
- compiler.compile(dist_param.resolve_expression(compiler.query))
- if hasattr(dist_param, "resolve_expression")
- else (
- "%s",
- connection.ops.get_distance(
- self.lhs.output_field, self.rhs_params, self.lookup_name
- ),
- )
- )
- @BaseSpatialField.register_lookup
- class DWithinLookup(DistanceLookupBase):
- lookup_name = "dwithin"
- sql_template = "%(func)s(%(lhs)s, %(rhs)s, %(value)s)"
- def process_distance(self, compiler, connection):
- dist_param = self.rhs_params[0]
- if (
- not connection.features.supports_dwithin_distance_expr
- and hasattr(dist_param, "resolve_expression")
- and not isinstance(dist_param, Distance)
- ):
- raise NotSupportedError(
- "This backend does not support expressions for specifying "
- "distance in the dwithin lookup."
- )
- return super().process_distance(compiler, connection)
- def process_rhs(self, compiler, connection):
- dist_sql, dist_params = self.process_distance(compiler, connection)
- self.template_params["value"] = dist_sql
- rhs_sql, params = super().process_rhs(compiler, connection)
- return rhs_sql, params + dist_params
- class DistanceLookupFromFunction(DistanceLookupBase):
- def as_sql(self, compiler, connection):
- spheroid = (
- len(self.rhs_params) == 2 and self.rhs_params[-1] == "spheroid"
- ) or None
- distance_expr = connection.ops.distance_expr_for_lookup(
- self.lhs, self.rhs, spheroid=spheroid
- )
- sql, params = compiler.compile(distance_expr.resolve_expression(compiler.query))
- dist_sql, dist_params = self.process_distance(compiler, connection)
- return (
- "%(func)s %(op)s %(dist)s" % {"func": sql, "op": self.op, "dist": dist_sql},
- params + dist_params,
- )
- @BaseSpatialField.register_lookup
- class DistanceGTLookup(DistanceLookupFromFunction):
- lookup_name = "distance_gt"
- op = ">"
- @BaseSpatialField.register_lookup
- class DistanceGTELookup(DistanceLookupFromFunction):
- lookup_name = "distance_gte"
- op = ">="
- @BaseSpatialField.register_lookup
- class DistanceLTLookup(DistanceLookupFromFunction):
- lookup_name = "distance_lt"
- op = "<"
- @BaseSpatialField.register_lookup
- class DistanceLTELookup(DistanceLookupFromFunction):
- lookup_name = "distance_lte"
- op = "<="
|