Commit 7294f5ce authored by Danijel Schorlemmer's avatar Danijel Schorlemmer
Browse files

Added District table and function to import boundaries and district information

parent 579cbccd
Pipeline #20908 passed with stage
in 1 minute and 15 seconds
#!/usr/bin/env python3
# Copyright (C) 2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.
import logging
import sqlite3
# Initialize log
logger = logging.getLogger(__name__)
class Database:
"""The Database class represents a Spatialite database. It manages the database
connection and cursor. Upon connection, the Spatialite extension is loaded.
Args:
database_filepath (str):
File path for the Spatialite database file.
spatialite_filepath (str):
File path of the Spatialite extension.
Attributes:
database_filepath (str): Spatialite database file path.
spatialite_filepath (str): File path to the Spatialite extension.
"""
def __init__(self, database_filepath, spatialite_filepath="mod_spatialite"):
self.database_filepath = database_filepath
self.spatialite_filepath = spatialite_filepath
self.connection = None
self.cursor = None
def create_connection_and_cursor(self, init_spatial_metadata=True):
"""Create a database connection, loads the Spatialite extension and initializes it.
Attributes:
init_spatial_metadata (bool, optional):
If set, the InitSpatialMetadata function of Spatialite is invoked
"""
# Create SQLite database for the exposure data
logger.debug("Connecting to database at %s" % self.database_filepath)
self.connection = sqlite3.connect(self.database_filepath)
logger.debug("Connection to database established")
self.connection.enable_load_extension(True)
sql_statement = "SELECT load_extension('%s');" % self.spatialite_filepath
self.connection.execute(sql_statement)
if init_spatial_metadata:
self.connection.execute("SELECT InitSpatialMetaData(1);")
logger.debug("Spatialite extension loaded and initialized")
versions = self.connection.execute("SELECT sqlite_version(), spatialite_version()")
for row in versions:
logger.debug("SQLite version: %s" % row[0])
logger.debug("Spatialite version: %s" % row[1])
self.cursor = self.connection.cursor()
class JapanDatabase(Database):
"""The JapanDatabase class represents a Spatialite database for the Japan
exposure model. It is derived from the generic Database class.
Args:
database_filepath (str):
File path for the Spatialite database file.
spatialite_filepath (str):
File path of the Spatialite extension.
Attributes:
database_filepath (str): Spatialite database file path.
spatialite_filepath (str): File path to the Spatialite extension.
"""
def __init__(self, database_filepath, spatialite_filepath="mod_spatialite"):
Database.__init__(self, database_filepath, spatialite_filepath)
def create_tables(self):
"""
Creates all necessary tables in the database. These are:
District : Stores the districts with their IDs, names and geometries
"""
# Create table District
sql_statement = "CREATE TABLE District ("
sql_statement += "id INTEGER PRIMARY KEY AUTOINCREMENT, "
sql_statement += "admin_id TEXT, "
sql_statement += "admin_name TEXT)"
self.connection.execute(sql_statement)
sql_statement = "SELECT AddGeometryColumn('District', 'geom', 4326, "
sql_statement += "'MULTIPOLYGON', 'XY')"
self.connection.execute(sql_statement)
logger.debug("Table District created")
def insert_district(self, admin_id, admin_name, geom):
"""
Inserts the ID, name, and geometry of a district to the District table.
Args:
admin_id (str):
ID of the administrative district
admin_name (str):
Name of the administrative district
geom (multipolygon):
Geometry (Spatialite) of the district boundary
"""
sql_statement = "INSERT INTO District "
sql_statement += "(admin_id, admin_name, geom) "
sql_statement += "VALUES ('%s', '%s', %s)" % (admin_id, admin_name, geom)
self.cursor.execute(sql_statement)
def read_districts_and_boundaries(self, district_boundary_filepath):
"""
Imports all districts and boundaries from a prepared geopackage file
('estat_bound_municipal`). The function opens this file as Spatialite database
and copies all necessary district and boundary information into the `District` table.
Args:
district_boundary_filepath (str):
File path to the boundary file
"""
boundary_db = Database(district_boundary_filepath, self.spatialite_filepath)
boundary_db.create_connection_and_cursor()
sql_statement = "SELECT key_code_ward, "
sql_statement += "CITY_NAME, "
sql_statement += "AsWKT(CastAutomagic(geom)) "
sql_statement += "FROM estat_bound_municipal"
boundary_db.cursor.execute(sql_statement)
for row in boundary_db.cursor:
self.insert_district(
row[0],
row[1],
"ST_Transform(CastToMultiPolygon(GeomFromText('%s', 4612)), 4326)" % row[2],
)
logger.debug("District %d with boundary added" % row[0])
self.connection.commit()
logger.info("Districts and boundaries added")
...@@ -19,6 +19,9 @@ ...@@ -19,6 +19,9 @@
import logging import logging
import sys import sys
import sqlite3
from database import JapanDatabase # pylint: disable=E0611,E0401
# Add a logger printing error, warning, info and debug messages to the screen # Add a logger printing error, warning, info and debug messages to the screen
logger = logging.getLogger() logger = logging.getLogger()
...@@ -31,6 +34,16 @@ def main(): ...@@ -31,6 +34,16 @@ def main():
# Example logging output # Example logging output
logger.info("exposurejapan started") logger.info("exposurejapan started")
db = JapanDatabase("test.sqlite", "/usr/lib64/mod_spatialite.so.7")
try:
db.create_connection_and_cursor()
except sqlite3.OperationalError:
logger.warning("Spatialite extension cannot be loaded. Exiting ...")
exit()
db.create_tables()
db.read_districts_and_boundaries("../data/estat_bound_municipal.gpkg")
# Leave the program # Leave the program
sys.exit() sys.exit()
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment