<s> import re <EOL> from decimal import Decimal <EOL> from django . conf import settings <EOL> from django . contrib . gis . db . backends . base import BaseSpatialOperations <EOL> from django . contrib . gis . db . backends . util import SpatialOperation , SpatialFunction <EOL> from django . contrib . gis . db . backends . postgis . adapter import PostGISAdapter <EOL> from django . contrib . gis . geometry . backend import Geometry <EOL> from django . contrib . gis . measure import Distance <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> from django . db . backends . postgresql_psycopg2 . base import DatabaseOperations <EOL> from django . db . utils import DatabaseError <EOL> from django . utils import six <EOL> from django . utils . functional import cached_property <EOL> from . models import GeometryColumns , SpatialRefSys <EOL> class PostGISOperator ( SpatialOperation ) : <EOL> "<STR_LIT>" <EOL> def __init__ ( self , operator ) : <EOL> super ( PostGISOperator , self ) . __init__ ( operator = operator ) <EOL> class PostGISFunction ( SpatialFunction ) : <EOL> "<STR_LIT>" <EOL> def __init__ ( self , prefix , function , ** kwargs ) : <EOL> super ( PostGISFunction , self ) . __init__ ( prefix + function , ** kwargs ) <EOL> class PostGISFunctionParam ( PostGISFunction ) : <EOL> "<STR_LIT>" <EOL> sql_template = '<STR_LIT>' <EOL> class PostGISDistance ( PostGISFunction ) : <EOL> "<STR_LIT>" <EOL> dist_func = '<STR_LIT>' <EOL> sql_template = '<STR_LIT>' <EOL> def __init__ ( self , prefix , operator ) : <EOL> super ( PostGISDistance , self ) . __init__ ( prefix , self . dist_func , <EOL> operator = operator ) <EOL> class PostGISSpheroidDistance ( PostGISFunction ) : <EOL> "<STR_LIT>" <EOL> dist_func = '<STR_LIT>' <EOL> sql_template = '<STR_LIT>' <EOL> def __init__ ( self , prefix , operator ) : <EOL> super ( PostGISSpheroidDistance , self ) . __init__ ( prefix , self . dist_func , <EOL> operator = operator ) <EOL> class PostGISSphereDistance ( PostGISDistance ) : <EOL> "<STR_LIT>" <EOL> dist_func = '<STR_LIT>' <EOL> class PostGISRelate ( PostGISFunctionParam ) : <EOL> "<STR_LIT>" <EOL> pattern_regex = re . compile ( r'<STR_LIT>' ) <EOL> def __init__ ( self , prefix , pattern ) : <EOL> if not self . pattern_regex . match ( pattern ) : <EOL> raise ValueError ( '<STR_LIT>' % pattern ) <EOL> super ( PostGISRelate , self ) . __init__ ( prefix , '<STR_LIT>' ) <EOL> class PostGISOperations ( DatabaseOperations , BaseSpatialOperations ) : <EOL> compiler_module = '<STR_LIT>' <EOL> name = '<STR_LIT>' <EOL> postgis = True <EOL> geom_func_prefix = '<STR_LIT>' <EOL> version_regex = re . compile ( r'<STR_LIT>' ) <EOL> valid_aggregates = dict ( [ ( k , None ) for k in <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> Adapter = PostGISAdapter <EOL> Adaptor = Adapter <EOL> def __init__ ( self , connection ) : <EOL> super ( PostGISOperations , self ) . __init__ ( connection ) <EOL> prefix = self . geom_func_prefix <EOL> self . geometry_operators = { <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT:left>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT:right>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT:@>' ) , <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> } <EOL> self . geometry_functions = { <EOL> '<STR_LIT>' : PostGISFunction ( prefix , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISFunction ( prefix , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISFunction ( prefix , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISFunction ( prefix , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISFunction ( prefix , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISFunction ( prefix , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISFunction ( prefix , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISFunction ( prefix , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( PostGISRelate , six . string_types ) , <EOL> '<STR_LIT>' : PostGISFunction ( prefix , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : PostGISFunction ( prefix , '<STR_LIT>' ) , <EOL> } <EOL> dtypes = ( Decimal , Distance , float ) + six . integer_types <EOL> def get_dist_ops ( operator ) : <EOL> "<STR_LIT>" <EOL> return { '<STR_LIT>' : PostGISDistance ( prefix , operator ) , <EOL> '<STR_LIT>' : PostGISSphereDistance ( prefix , operator ) , <EOL> '<STR_LIT>' : PostGISSpheroidDistance ( prefix , operator ) , <EOL> } <EOL> self . distance_functions = { <EOL> '<STR_LIT>' : ( get_dist_ops ( '<STR_LIT:>>' ) , dtypes ) , <EOL> '<STR_LIT>' : ( get_dist_ops ( '<STR_LIT>' ) , dtypes ) , <EOL> '<STR_LIT>' : ( get_dist_ops ( '<STR_LIT:<>' ) , dtypes ) , <EOL> '<STR_LIT>' : ( get_dist_ops ( '<STR_LIT>' ) , dtypes ) , <EOL> '<STR_LIT>' : ( PostGISFunctionParam ( prefix , '<STR_LIT>' ) , dtypes ) <EOL> } <EOL> self . geometry_functions . update ( self . distance_functions ) <EOL> if self . spatial_version < ( <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:4> ) : <EOL> GEOJSON = False <EOL> else : <EOL> GEOJSON = prefix + '<STR_LIT>' <EOL> if self . spatial_version >= ( <NUM_LIT:1> , <NUM_LIT:4> , <NUM_LIT:0> ) : <EOL> GEOHASH = '<STR_LIT>' <EOL> BOUNDINGCIRCLE = '<STR_LIT>' <EOL> self . geometry_functions [ '<STR_LIT>' ] = PostGISFunction ( prefix , '<STR_LIT>' ) <EOL> else : <EOL> GEOHASH , BOUNDINGCIRCLE = False , False <EOL> if self . spatial_version >= ( <NUM_LIT:1> , <NUM_LIT:5> , <NUM_LIT:0> ) : <EOL> self . geography = True <EOL> self . geography_functions = self . distance_functions . copy ( ) <EOL> self . geography_functions . update ( { <EOL> '<STR_LIT>' : self . geometry_functions [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : self . geometry_functions [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : self . geometry_functions [ '<STR_LIT>' ] , <EOL> } ) <EOL> self . geography_operators = { <EOL> '<STR_LIT>' : PostGISOperator ( '<STR_LIT>' ) , <EOL> } <EOL> if self . spatial_version >= ( <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:0> ) : <EOL> self . geometry = True <EOL> self . gis_terms = set ( [ '<STR_LIT>' ] ) <EOL> self . gis_terms . update ( self . geometry_operators ) <EOL> self . gis_terms . update ( self . geometry_functions ) <EOL> self . area = prefix + '<STR_LIT>' <EOL> self . bounding_circle = BOUNDINGCIRCLE <EOL> self . centroid = prefix + '<STR_LIT>' <EOL> self . collect = prefix + '<STR_LIT>' <EOL> self . difference = prefix + '<STR_LIT>' <EOL> self . distance = prefix + '<STR_LIT>' <EOL> self . distance_sphere = prefix + '<STR_LIT>' <EOL> self . distance_spheroid = prefix + '<STR_LIT>' <EOL> self . envelope = prefix + '<STR_LIT>' <EOL> self . extent = prefix + '<STR_LIT>' <EOL> self . force_rhr = prefix + '<STR_LIT>' <EOL> self . geohash = GEOHASH <EOL> self . geojson = GEOJSON <EOL> self . gml = prefix + '<STR_LIT>' <EOL> self . intersection = prefix + '<STR_LIT>' <EOL> self . kml = prefix + '<STR_LIT>' <EOL> self . length = prefix + '<STR_LIT>' <EOL> self . length_spheroid = prefix + '<STR_LIT>' <EOL> self . makeline = prefix + '<STR_LIT>' <EOL> self . mem_size = prefix + '<STR_LIT>' <EOL> self . num_geom = prefix + '<STR_LIT>' <EOL> self . num_points = prefix + '<STR_LIT>' <EOL> self . perimeter = prefix + '<STR_LIT>' <EOL> self . point_on_surface = prefix + '<STR_LIT>' <EOL> self . polygonize = prefix + '<STR_LIT>' <EOL> self . reverse = prefix + '<STR_LIT>' <EOL> self . scale = prefix + '<STR_LIT>' <EOL> self . snap_to_grid = prefix + '<STR_LIT>' <EOL> self . svg = prefix + '<STR_LIT>' <EOL> self . sym_difference = prefix + '<STR_LIT>' <EOL> self . transform = prefix + '<STR_LIT>' <EOL> self . translate = prefix + '<STR_LIT>' <EOL> self . union = prefix + '<STR_LIT>' <EOL> self . unionagg = prefix + '<STR_LIT>' <EOL> if self . spatial_version >= ( <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:0> ) : <EOL> self . extent3d = prefix + '<STR_LIT>' <EOL> self . length3d = prefix + '<STR_LIT>' <EOL> self . perimeter3d = prefix + '<STR_LIT>' <EOL> else : <EOL> self . extent3d = prefix + '<STR_LIT>' <EOL> self . length3d = prefix + '<STR_LIT>' <EOL> self . perimeter3d = prefix + '<STR_LIT>' <EOL> @ cached_property <EOL> def spatial_version ( self ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( settings , '<STR_LIT>' ) : <EOL> version = settings . POSTGIS_VERSION <EOL> else : <EOL> try : <EOL> vtup = self . postgis_version_tuple ( ) <EOL> except DatabaseError : <EOL> raise ImproperlyConfigured ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % self . connection . settings_dict [ '<STR_LIT>' ] <EOL> ) <EOL> version = vtup [ <NUM_LIT:1> : ] <EOL> return version <EOL> def check_aggregate_support ( self , aggregate ) : <EOL> """<STR_LIT>""" <EOL> agg_name = aggregate . __class__ . __name__ <EOL> return agg_name in self . valid_aggregates <EOL> def convert_extent ( self , box ) : <EOL> """<STR_LIT>""" <EOL> ll , ur = box [ <NUM_LIT:4> : - <NUM_LIT:1> ] . split ( '<STR_LIT:U+002C>' ) <EOL> xmin , ymin = map ( float , ll . split ( ) ) <EOL> xmax , ymax = map ( float , ur . split ( ) ) <EOL> return ( xmin , ymin , xmax , ymax ) <EOL> def convert_extent3d ( self , box3d ) : <EOL> """<STR_LIT>""" <EOL> ll , ur = box3d [ <NUM_LIT:6> : - <NUM_LIT:1> ] . split ( '<STR_LIT:U+002C>' ) <EOL> xmin , ymin , zmin = map ( float , ll . split ( ) ) <EOL> xmax , ymax , zmax = map ( float , ur . split ( ) ) <EOL> return ( xmin , ymin , zmin , xmax , ymax , zmax ) <EOL> def convert_geom ( self , hex , geo_field ) : <EOL> """<STR_LIT>""" <EOL> if hex : <EOL> return Geometry ( hex ) <EOL> else : <EOL> return None <EOL> def geo_db_type ( self , f ) : <EOL> """<STR_LIT>""" <EOL> if f . geography : <EOL> if not self . geography : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> if f . srid != <NUM_LIT> : <EOL> raise NotImplementedError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return '<STR_LIT>' % ( f . geom_type , f . srid ) <EOL> elif self . geometry : <EOL> if f . dim == <NUM_LIT:3> : <EOL> geom_type = f . geom_type + '<STR_LIT>' <EOL> else : <EOL> geom_type = f . geom_type <EOL> return '<STR_LIT>' % ( geom_type , f . srid ) <EOL> else : <EOL> return None <EOL> def get_distance ( self , f , dist_val , lookup_type ) : <EOL> """<STR_LIT>""" <EOL> if len ( dist_val ) == <NUM_LIT:1> : <EOL> value , option = dist_val [ <NUM_LIT:0> ] , None <EOL> else : <EOL> value , option = dist_val <EOL> geodetic = f . geodetic ( self . connection ) <EOL> geography = f . geography and self . geography <EOL> if isinstance ( value , Distance ) : <EOL> if geography : <EOL> dist_param = value . m <EOL> elif geodetic : <EOL> if lookup_type == '<STR_LIT>' : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> dist_param = value . m <EOL> else : <EOL> dist_param = getattr ( value , Distance . unit_attname ( f . units_name ( self . connection ) ) ) <EOL> else : <EOL> dist_param = value <EOL> if ( not geography and geodetic and lookup_type != '<STR_LIT>' <EOL> and option == '<STR_LIT>' ) : <EOL> return [ f . _spheroid , dist_param ] <EOL> else : <EOL> return [ dist_param ] <EOL> def get_geom_placeholder ( self , f , value ) : <EOL> """<STR_LIT>""" <EOL> if value is None or value . srid == f . srid : <EOL> placeholder = '<STR_LIT:%s>' <EOL> else : <EOL> placeholder = '<STR_LIT>' % ( self . transform , f . srid ) <EOL> if hasattr ( value , '<STR_LIT>' ) : <EOL> placeholder = placeholder % self . get_expression_column ( value ) <EOL> return placeholder <EOL> def _get_postgis_func ( self , func ) : <EOL> """<STR_LIT>""" <EOL> with self . connection . temporary_connection ( ) as cursor : <EOL> cursor . execute ( '<STR_LIT>' % func ) <EOL> return cursor . fetchone ( ) [ <NUM_LIT:0> ] <EOL> def postgis_geos_version ( self ) : <EOL> "<STR_LIT>" <EOL> return self . _get_postgis_func ( '<STR_LIT>' ) <EOL> def postgis_lib_version ( self ) : <EOL> "<STR_LIT>" <EOL> return self . _get_postgis_func ( '<STR_LIT>' ) <EOL> def postgis_proj_version ( self ) : <EOL> "<STR_LIT>" <EOL> return self . _get_postgis_func ( '<STR_LIT>' ) <EOL> def postgis_version ( self ) : <EOL> "<STR_LIT>" <EOL> return self . _get_postgis_func ( '<STR_LIT>' ) <EOL> def postgis_full_version ( self ) : <EOL> "<STR_LIT>" <EOL> return self . _get_postgis_func ( '<STR_LIT>' ) <EOL> def postgis_version_tuple ( self ) : <EOL> """<STR_LIT>""" <EOL> version = self . postgis_lib_version ( ) <EOL> m = self . version_regex . match ( version ) <EOL> if m : <EOL> major = int ( m . group ( '<STR_LIT>' ) ) <EOL> minor1 = int ( m . group ( '<STR_LIT>' ) ) <EOL> minor2 = int ( m . group ( '<STR_LIT>' ) ) <EOL> else : <EOL> raise Exception ( '<STR_LIT>' % version ) <EOL> return ( version , major , minor1 , minor2 ) <EOL> def proj_version_tuple ( self ) : <EOL> """<STR_LIT>""" <EOL> proj_regex = re . compile ( r'<STR_LIT>' ) <EOL> proj_ver_str = self . postgis_proj_version ( ) <EOL> m = proj_regex . search ( proj_ver_str ) <EOL> if m : <EOL> return tuple ( map ( int , [ m . group ( <NUM_LIT:1> ) , m . group ( <NUM_LIT:2> ) , m . group ( <NUM_LIT:3> ) ] ) ) <EOL> else : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> def num_params ( self , lookup_type , num_param ) : <EOL> """<STR_LIT>""" <EOL> def exactly_two ( np ) : return np == <NUM_LIT:2> <EOL> def two_to_three ( np ) : return np >= <NUM_LIT:2> and np <= <NUM_LIT:3> <EOL> if ( lookup_type in self . distance_functions and <EOL> lookup_type != '<STR_LIT>' ) : <EOL> return two_to_three ( num_param ) <EOL> else : <EOL> return exactly_two ( num_param ) <EOL> def spatial_lookup_sql ( self , lvalue , lookup_type , value , field , qn ) : <EOL> """<STR_LIT>""" <EOL> alias , col , db_type = lvalue <EOL> geo_col = '<STR_LIT>' % ( qn ( alias ) , qn ( col ) ) <EOL> if lookup_type in self . geometry_operators : <EOL> if field . geography and not lookup_type in self . geography_operators : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' % lookup_type ) <EOL> op = self . geometry_operators [ lookup_type ] <EOL> return op . as_sql ( geo_col , self . get_geom_placeholder ( field , value ) ) <EOL> elif lookup_type in self . geometry_functions : <EOL> if field . geography and not lookup_type in self . geography_functions : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' % lookup_type ) <EOL> tmp = self . geometry_functions [ lookup_type ] <EOL> if isinstance ( tmp , tuple ) : <EOL> op , arg_type = tmp <EOL> if not isinstance ( value , ( tuple , list ) ) : <EOL> raise ValueError ( '<STR_LIT>' % lookup_type ) <EOL> geom = value [ <NUM_LIT:0> ] <EOL> nparams = len ( value ) <EOL> if not self . num_params ( lookup_type , nparams ) : <EOL> raise ValueError ( '<STR_LIT>' % lookup_type ) <EOL> if not isinstance ( value [ <NUM_LIT:1> ] , arg_type ) : <EOL> raise ValueError ( '<STR_LIT>' % ( arg_type , type ( value [ <NUM_LIT:1> ] ) ) ) <EOL> if lookup_type == '<STR_LIT>' : <EOL> op = op ( self . geom_func_prefix , value [ <NUM_LIT:1> ] ) <EOL> elif lookup_type in self . distance_functions and lookup_type != '<STR_LIT>' : <EOL> if not field . geography and field . geodetic ( self . connection ) : <EOL> if not self . connection . ops . geography : <EOL> if field . geom_type != '<STR_LIT>' : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if str ( geom . geom_type ) != '<STR_LIT>' : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if nparams == <NUM_LIT:3> and value [ <NUM_LIT:2> ] == '<STR_LIT>' : <EOL> op = op [ '<STR_LIT>' ] <EOL> else : <EOL> op = op [ '<STR_LIT>' ] <EOL> else : <EOL> op = op [ '<STR_LIT>' ] <EOL> else : <EOL> op = tmp <EOL> geom = value <EOL> return op . as_sql ( geo_col , self . get_geom_placeholder ( field , geom ) ) <EOL> elif lookup_type == '<STR_LIT>' : <EOL> return "<STR_LIT>" % ( geo_col , ( '<STR_LIT>' if value else '<STR_LIT>' ) ) , [ ] <EOL> raise TypeError ( "<STR_LIT>" % repr ( lookup_type ) ) <EOL> def spatial_aggregate_sql ( self , agg ) : <EOL> """<STR_LIT>""" <EOL> agg_name = agg . __class__ . __name__ <EOL> if not self . check_aggregate_support ( agg ) : <EOL> raise NotImplementedError ( '<STR_LIT>' % agg_name ) <EOL> agg_name = agg_name . lower ( ) <EOL> if agg_name == '<STR_LIT>' : <EOL> agg_name += '<STR_LIT>' <EOL> sql_template = '<STR_LIT>' <EOL> sql_function = getattr ( self , agg_name ) <EOL> return sql_template , sql_function <EOL> def geometry_columns ( self ) : <EOL> return GeometryColumns <EOL> def spatial_ref_sys ( self ) : <EOL> return SpatialRefSys </s>
