<s> import geojson <EOL> import colspecs <EOL> import geo <EOL> import geofeatures <EOL> import log <EOL> class DatabaseInterface ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , db_name , srid = None , srs = None , verbose = True ) : <EOL> self . db_name = db_name <EOL> self . srid = srid <EOL> self . srs = srs <EOL> self . verbose = verbose <EOL> if self . verbose : log . info ( '<STR_LIT>' , db_name ) <EOL> ( self . con , self . cur ) = self . _connect ( ) <EOL> self . _add_spatial_extension ( ) <EOL> self . add_spatial_ref_sys ( ) <EOL> def _connect ( self ) : <EOL> log . error ( '<STR_LIT>' ) <EOL> def _add_spatial_extension ( self ) : <EOL> log . error ( '<STR_LIT>' ) <EOL> _interpolation_ref = '<STR_LIT>' <EOL> def code_point ( self , point , srid = None ) : <EOL> if not srid : srid = self . srid <EOL> return "<STR_LIT>" % ( geo . wkt_point ( point ) , srid ) <EOL> def code_linestring ( self , points , srid = None ) : <EOL> if not srid : srid = self . srid <EOL> return "<STR_LIT>" % ( geo . wkt_linestring ( points ) , srid ) <EOL> def commit ( self ) : <EOL> self . con . commit ( ) <EOL> def close ( self ) : <EOL> self . commit ( ) <EOL> self . con . close ( ) <EOL> def exec_sql ( self , s , values = None ) : <EOL> if values : <EOL> self . cur . execute ( s , values ) <EOL> else : <EOL> self . cur . execute ( s ) <EOL> def exec_sql_r ( self , s , values = None ) : <EOL> '''<STR_LIT>''' <EOL> self . exec_sql ( s , values ) <EOL> return self . cur . fetchall ( ) <EOL> def select ( self , cols , table , where = None ) : <EOL> '''<STR_LIT>''' <EOL> parms = { '<STR_LIT>' : cols , '<STR_LIT>' : table , '<STR_LIT>' : where } <EOL> if where : <EOL> return self . exec_sql_r ( "<STR_LIT>" % parms ) <EOL> else : <EOL> return self . exec_sql_r ( "<STR_LIT>" % parms ) <EOL> def get_table ( self , table ) : <EOL> rows = self . select ( '<STR_LIT:*>' , table ) ; <EOL> return rows <EOL> def get_table_names ( self ) : <EOL> rows = self . select ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> return [ row [ <NUM_LIT:0> ] for row in rows ] <EOL> def get_num_rows ( self , table ) : <EOL> rows = self . exec_sql_r ( '<STR_LIT>' % table ) <EOL> assert len ( rows ) == <NUM_LIT:1> <EOL> return rows [ <NUM_LIT:0> ] [ '<STR_LIT:count>' ] <EOL> def add_column ( self , table , name , type = '<STR_LIT:text>' ) : <EOL> self . exec_sql ( '<STR_LIT>' % <EOL> { '<STR_LIT>' : table , '<STR_LIT:name>' : name , '<STR_LIT:type>' : type } ) <EOL> def add_geometry_column ( self , table , name , type , srid = None ) : <EOL> if not srid : srid = self . srid <EOL> self . exec_sql ( '''<STR_LIT>''' % { '<STR_LIT>' : table , '<STR_LIT:name>' : name , '<STR_LIT>' : srid , '<STR_LIT:type>' : type } ) <EOL> def drop_table ( self , table ) : <EOL> self . exec_sql ( '<STR_LIT>' % table ) <EOL> def create_table ( self , table , col_specs ) : <EOL> col_specs = colspecs . fix ( col_specs ) <EOL> self . drop_table ( table ) <EOL> data_specs = [ ] <EOL> geom_specs = [ ] <EOL> for col , typ , prop in col_specs : <EOL> if typ == '<STR_LIT>' or typ == '<STR_LIT>' : <EOL> geom_specs . append ( ( col , typ , prop ) ) <EOL> else : <EOL> data_specs . append ( ( col , typ , prop ) ) <EOL> spec_str = [ '<STR_LIT>' % spec [ : <NUM_LIT:2> ] for spec in data_specs ] <EOL> spec_str = '<STR_LIT:U+002CU+0020>' . join ( spec_str ) <EOL> self . exec_sql ( '<STR_LIT>' % ( table , spec_str ) ) <EOL> for col_name , col_type , parm_name in geom_specs : <EOL> self . add_geometry_column ( table , col_name , col_type ) <EOL> def write_row ( self , table , col_specs , row , srid = None ) : <EOL> if not srid : srid = self . srid <EOL> col_specs = colspecs . fix ( col_specs ) <EOL> def code_value ( col_name , col_type , value , ref ) : <EOL> if col_type == '<STR_LIT>' : <EOL> return ( self . code_point ( value , srid ) , None ) <EOL> if col_type == '<STR_LIT>' : <EOL> return ( self . code_linestring ( value , srid ) , None ) <EOL> if col_type == '<STR_LIT>' : <EOL> if not isinstance ( value , basestring ) : value = str ( value ) <EOL> return ( ref , value ) <EOL> return ( ref , value ) <EOL> if isinstance ( row , dict ) : <EOL> rowd = row . copy ( ) <EOL> else : <EOL> assert len ( row ) == len ( col_specs ) <EOL> rowd = { } <EOL> for i in range ( len ( col_specs ) ) : rowd [ col_specs [ i ] [ <NUM_LIT:2> ] ] = row [ i ] <EOL> columns = [ ] <EOL> values = [ ] <EOL> parms = [ ] <EOL> for col_name , col_type , prop_name in col_specs : <EOL> if not prop_name in rowd . keys ( ) : continue <EOL> value = rowd [ prop_name ] <EOL> if value == None : continue <EOL> columns . append ( '<STR_LIT:">' + col_name + '<STR_LIT:">' ) <EOL> ( value , parm ) = code_value ( col_name , col_type , value , self . _interpolation_ref ) <EOL> values . append ( value ) <EOL> if parm != None : parms . append ( parm ) <EOL> columns = '<STR_LIT:U+002C>' . join ( columns ) <EOL> values = '<STR_LIT:U+002C>' . join ( values ) <EOL> sql_template = '''<STR_LIT>''' % { <EOL> '<STR_LIT>' : table , <EOL> '<STR_LIT>' : columns , <EOL> '<STR_LIT>' : values <EOL> } <EOL> self . exec_sql ( sql_template , parms ) <EOL> def write_table ( self , table , col_specs , rows ) : <EOL> self . create_table ( table , col_specs ) <EOL> for row in rows : <EOL> self . write_row ( table , col_specs , row ) <EOL> def write_geo ( self , geo , table_name , srid = None , geometry_type = None , col_specs = None ) : <EOL> features = geofeatures . geo_features ( geo ) <EOL> if len ( features ) == <NUM_LIT:0> and not col_specs : <EOL> log . warning ( '<STR_LIT>' , table_name ) <EOL> return <EOL> if not geometry_type : geometry_type = features [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] <EOL> if not col_specs : <EOL> col_specs = colspecs . gen ( geo , geometry_type ) <EOL> rows = [ ] <EOL> for feature in features : <EOL> if feature [ '<STR_LIT>' ] [ '<STR_LIT:type>' ] != geometry_type : continue <EOL> row = { } <EOL> row . update ( feature [ '<STR_LIT>' ] ) <EOL> row [ '<STR_LIT>' ] = feature [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> rows . append ( row ) <EOL> self . write_table ( table_name , col_specs , rows ) <EOL> self . commit ( ) <EOL> def write_pnwk ( self , pnwk , name = None ) : <EOL> if not name : name = pnwk . name <EOL> self . write_geo ( pnwk , name + '<STR_LIT>' , geometry_type = '<STR_LIT>' ) <EOL> self . write_geo ( pnwk , name + '<STR_LIT>' , geometry_type = '<STR_LIT>' ) <EOL> def add_spatial_ref_sys ( self , srid = None , srs = None ) : <EOL> if not srid : srid = self . srid <EOL> if not srs : srs = self . srs <EOL> if srs : <EOL> srs_info = dict ( <EOL> srid = srid , <EOL> auth_name = srs . auth_name , <EOL> auth_srid = srs . auth_srid , <EOL> proj4text = srs . proj4text , <EOL> srtext = srs . srtext <EOL> ) <EOL> else : <EOL> srs_info = dict ( srid = srid ) <EOL> existing = self . select ( <EOL> '<STR_LIT:*>' , <EOL> '<STR_LIT>' , <EOL> where = '<STR_LIT>' % srs_info <EOL> ) <EOL> assert len ( existing ) < <NUM_LIT:2> <EOL> if len ( existing ) == <NUM_LIT:1> : <EOL> '''<STR_LIT>''' <EOL> else : <EOL> assert len ( existing ) == <NUM_LIT:0> <EOL> have_info = srs and srs . auth_name and srs . auth_srid and srs . proj4text and srs . srtext <EOL> if not have_info : <EOL> log . error ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> srs_info [ '<STR_LIT>' ] , <EOL> self . db_name , <EOL> str ( srs_info ) ) <EOL> if have_info : <EOL> log . info ( '<STR_LIT>' , <EOL> srs_info [ '<STR_LIT>' ] , <EOL> self . db_name ) <EOL> colspecs = [ ( col , '<STR_LIT>' ) for col in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> self . write_row ( '<STR_LIT>' , colspecs , srs_info ) <EOL> self . commit ( ) <EOL> def test ( self , verbose = True ) : <EOL> PREFIX = '<STR_LIT>' <EOL> TABLE1 = PREFIX + '<STR_LIT>' <EOL> TABLE2 = PREFIX + '<STR_LIT>' <EOL> TABLE_SEGS = PREFIX + '<STR_LIT>' <EOL> TABLE_JCTS = PREFIX + '<STR_LIT>' <EOL> ref = self . _interpolation_ref <EOL> self . exec_sql ( '<STR_LIT>' % TABLE1 ) <EOL> self . exec_sql ( '<STR_LIT>' % TABLE1 ) <EOL> self . exec_sql ( '<STR_LIT>' % ( TABLE1 , ref , ref ) , <EOL> ( '''<STR_LIT>''' , <NUM_LIT:1> ) ) <EOL> self . exec_sql ( '''<STR_LIT>''' % TABLE1 ) <EOL> self . exec_sql ( '''<STR_LIT>''' % TABLE1 ) <EOL> table = self . get_table ( TABLE1 ) <EOL> assert len ( table ) == <NUM_LIT:3> <EOL> assert table [ <NUM_LIT:1> ] [ '<STR_LIT>' ] == <NUM_LIT:5> <EOL> assert table [ <NUM_LIT:2> ] [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> assert len ( table ) == self . get_num_rows ( TABLE1 ) <EOL> col_names = self . get_column_names ( TABLE1 ) <EOL> assert col_names == [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> assert len ( table [ <NUM_LIT:0> ] . keys ( ) ) == <NUM_LIT:3> <EOL> self . add_geometry_column ( TABLE1 , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert self . get_column_names ( TABLE1 ) == [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> col_specs = [ ( '<STR_LIT:id>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> rows = [ ] <EOL> rows . append ( { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } ) <EOL> rows . append ( { '<STR_LIT:id>' : <NUM_LIT:2> , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> rows . append ( [ <NUM_LIT:3> , '<STR_LIT>' , <NUM_LIT> , ( - <NUM_LIT> , <NUM_LIT> ) ] ) <EOL> rows . append ( { '<STR_LIT:id>' : <NUM_LIT:4> , '<STR_LIT>' : None , '<STR_LIT>' : <NUM_LIT:100> } ) <EOL> self . write_table ( TABLE2 , col_specs , rows ) <EOL> assert self . get_column_names ( TABLE2 ) == [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:id>' , '<STR_LIT>' ] <EOL> rowsr = self . get_table ( TABLE2 ) <EOL> assert len ( rowsr ) == <NUM_LIT:4> <EOL> assert rowsr [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> assert rowsr [ <NUM_LIT:2> ] [ '<STR_LIT:id>' ] == <NUM_LIT:3> <EOL> assert rowsr [ <NUM_LIT:1> ] [ '<STR_LIT>' ] == None <EOL> geometry = geojson . Point ( ( - <NUM_LIT> , <NUM_LIT> ) ) <EOL> ls = geojson . LineString ( [ ( - <NUM_LIT> , <NUM_LIT> ) , ( - <NUM_LIT> , <NUM_LIT> ) ] ) <EOL> features = [ <EOL> geojson . Feature ( geometry = geometry , id = <NUM_LIT:1> , properties = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> } ) , <EOL> geojson . Feature ( geometry = geometry , id = <NUM_LIT:2> , properties = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:3> , '<STR_LIT:foo>' : None } ) , <EOL> geojson . Feature ( geometry = ls , id = <NUM_LIT:10> , properties = { '<STR_LIT>' : <NUM_LIT:10> } ) , <EOL> geojson . Feature ( geometry = geometry , id = <NUM_LIT:3> , properties = { '<STR_LIT>' : <NUM_LIT:2> ** <NUM_LIT> } ) , <EOL> geojson . Feature ( geometry = geometry , id = <NUM_LIT:4> , properties = { '<STR_LIT>' : <NUM_LIT:2> } ) , <EOL> geojson . Feature ( geometry = geometry , id = <NUM_LIT:5> , properties = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> ] <EOL> geo = geojson . FeatureCollection ( features ) <EOL> self . write_geo ( geo , TABLE_SEGS , geometry_type = '<STR_LIT>' ) <EOL> self . write_geo ( geo , TABLE_JCTS , geometry_type = '<STR_LIT>' ) <EOL> table_names = self . get_table_names ( ) <EOL> assert TABLE1 in table_names <EOL> assert '<STR_LIT>' in self . get_table_names ( ) <EOL> rows = self . select ( '<STR_LIT:*>' , '<STR_LIT>' , '<STR_LIT>' % self . srid ) <EOL> assert len ( rows ) == <NUM_LIT:1> <EOL> self . drop_table ( TABLE1 ) <EOL> self . drop_table ( TABLE2 ) <EOL> self . drop_table ( TABLE_SEGS ) <EOL> self . drop_table ( TABLE_JCTS ) <EOL> self . commit ( ) <EOL> def test ( ) : <EOL> print '<STR_LIT>' <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> test ( ) </s>
