InputDataFetchFunctions.py — Datenbankabfragen¶

Quelldatei

src/InputDataFetchFunctions.py

Stellt alle Funktionen bereit, die Länder- und Zelldaten aus der MySQL-Datenbank (oder CSV-Dateien) laden: Emissionsfaktoren, Bevölkerung, Kläranlagenanschlüsse, Düngerapplikationsraten, Viehexkretionsraten u. v. m.

import os
import mysql.connector
import pandas as pd
from dotenv import load_dotenv
import Paths_and_params as PP

Umgebungsvariablen aus .env-Datei laden (Datenbankzugangsdaten)

load_dotenv()

def LoadDatabase(dbname: str) -> 'mysql.connector.MySQLConnection':

Verbindung zu MySQL-Datenbank herstellen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank

Rückgabe

mysql.connector.MySQLConnection

Datenbankverbindung zu MySQL-Server

    dbase = mysql.connector.connect(
    host = os.getenv("DB_HOST", "134.147.42.33"),
    user = os.getenv("DB_USER"),
    passwd = os.getenv("DB_PASSWORD"),
    database = dbname)

    return dbase

Globale Filterparameter

Basin_grid_cells, IDScen, time und dbname_cell sollten hier sowie im WorldQual_Lite Modul konfiguriert werden

Parameter-ID = 60 würde Phosphor bedeuten (siehe wq_load_general Datenbank) Zeitraum könnte von 1993-2017 iteriert werden

Betrachtetes Referenzjahr

time = 1992

IDFAOReg = 4 würde Nordamerika bedeuten (siehe wq_load_general Datenbank) country_id = 276 würde USA bedeuten (siehe wq_load_general Datenbank)

Szenario-ID

IDScen = 27

cell = 80963 # beispielhafte Rasterzelle Basin_grid_cells = [80963,80964]

Datenbankname mit Zellinputs und Parametern

dbname_cell = "globewq_wq_load_eu"

dbname1 = "globe_wq_load" # Datenbankname mit Länderinputs und Parametern

Eingabedaten auf Zellenebene

def CellInputs(dbname: str, IDScen: int, time: int) -> list:

Alle Eingabeparameter für eine Rasterzelle abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`IDScen`	`int`	Szenario-ID
`time`	`int`	Betrachtetes Jahr

Rückgabe

list

Liste von Tupeln mit Zelleninputs (cell, pop_urb, pop_rur, pop_tot, rtf_man, rtf_dom, rtf_irr, gdp, salinity, humidity, lu)

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT cell, pop_urb, pop_rur, pop_tot, rtf_man, rtf_dom, rtf_irr, gdp, salinity, humidity, lu \
            FROM cell_input \
            WHERE IDScen = %s AND time = %s"

    values = (IDScen, time)
    mycursor.execute(query, values)
    return mycursor.fetchall()


def CellPopulation(dbname: str, IDScen: int, time: int, cell: int) -> tuple:

Bevölkerungsdaten für eine Rasterzelle abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`IDScen`	`int`	Szenario-ID
`time`	`int`	Betrachtetes Jahr
`cell`	`int`	Rasterzellen-ID

Rückgabe

tuple

(pop_urb, pop_rur, pop_tot) - Bevölkerung (urban, rural, gesamt)

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT pop_urb, pop_rur, pop_tot\
           FROM cell_input \
           WHERE IDScen = %s AND time = %s AND cell = %s"

    values = (IDScen, time, cell)
    mycursor.execute(query, values)
    return mycursor.fetchone()

def CellReturnFlows(dbname: str, IDScen: int, time: int, cell: int) -> tuple:

Rückflussraten für eine Rasterzelle abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`IDScen`	`int`	Szenario-ID
`time`	`int`	Betrachtetes Jahr
`cell`	`int`	Rasterzellen-ID

Rückgabe

tuple

(rtf_man, rtf_dom, rtf_irr) - Rückflussraten in m³/a

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT rtf_man, rtf_dom, rtf_irr\
           FROM cell_input \
           WHERE IDScen = %s AND time = %s AND cell = %s"

    values = (IDScen, time, cell)
    mycursor.execute(query, values)
    return mycursor.fetchone()

def CellParameters(dbname: str, IDScen: int, time: int, cell: int) -> tuple:

Zellparameter abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`IDScen`	`int`	Szenario-ID
`time`	`int`	Betrachtetes Jahr
`cell`	`int`	Rasterzellen-ID

Rückgabe

tuple

(gdp, salinity, humidity, lu) - BIP, Salinität, Luftfeuchte, Landnutzung

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT gdp, salinity, humidity, lu \
           FROM cell_input \
           WHERE IDScen = %s AND time = %s AND cell = %s"

    values = (IDScen, time, cell)
    mycursor.execute(query, values)
    return mycursor.fetchone()

Emissionsfaktoren und Konzentrationen in Rückflüssen

Hinweis: Die kombinierte Abfragefunktion EmmisionFactorAndConcInReturnFlows wurde durch spezialisierte Funktionen ersetzt (CountryConcInReturnFlows, CountryEmmisionFactor)

def CountryConcInReturnFlows(dbname: str, parameter_id: int, time: int, country_id: int) -> tuple:

Phosphor-/Stickstoffkonzentrationen in Rückflüssen abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`parameter_id`	`int`	Nutrient-Parameter-ID
`time`	`int`	Betrachtetes Jahr
`country_id`	`int`	Länder-ID

Rückgabe

tuple

Konzentrationen in mg/l für verschiedene Sektoren (conc_man_f, conc_man_t, conc_man_p, conc_man_c, conc_man_g, conc_man_m, conc_man_nd, c_geogen, conc_urb)

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT conc_man_f, conc_man_t, conc_man_p, conc_man_c, conc_man_g, conc_man_m, conc_man_nd, c_geogen, conc_urb \
           FROM country_parameter_input \
           WHERE parameter_id = %s AND time = %s AND country_id = %s"

    values = (parameter_id, time, country_id)
    mycursor.execute(query, values)
    return mycursor.fetchone()


def CountryEmmisionFactor(dbname: str, parameter_id: int, time: int, country_id: int) -> tuple:

Emissionsfaktor für ein Land und Nährstoff abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`parameter_id`	`int`	Nutrient-Parameter-ID
`time`	`int`	Betrachtetes Jahr
`country_id`	`int`	Länder-ID

Rückgabe

tuple

(ef,) - Emissionsfaktor in kg/Kopf/Jahr

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT ef \
           FROM country_parameter_input \
           WHERE parameter_id = %s AND time = %s AND country_id = %s"

    values = (parameter_id, time, country_id)
    mycursor.execute(query, values)
    return mycursor.fetchone()

Länderinputs: Bevölkerung, Rückflüsse und Kläranlagenanschluss

def CountryPopulation(dbname: str, IDScen: int, time: int, country_id: int) -> tuple:

Bevölkerungsdaten für ein Land abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`IDScen`	`int`	Szenario-ID
`time`	`int`	Betrachtetes Jahr
`country_id`	`int`	Länder-ID

Rückgabe

tuple

(pop_tot, pop_urb, pop_rur) - Bevölkerung (gesamt, urban, rural)

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT pop_tot, pop_urb, pop_rur\
               FROM country_input \
               WHERE IDScen = %s AND time = %s AND country_id = %s"

    values = (IDScen, time, country_id)
    mycursor.execute(query, values)
    return mycursor.fetchone()

def CountryReturnFlows(dbname: str, IDScen: int, time: int, country_id: int) -> tuple:

Rückflussraten und Sektoranteile für ein Land abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`IDScen`	`int`	Szenario-ID
`time`	`int`	Betrachtetes Jahr
`country_id`	`int`	Länder-ID

Rückgabe

tuple

(rtf_man, fra_man_f, fra_man_t, fra_man_p, fra_man_c, fra_man_g, fra_man_m, fra_man_nd, rtf_dom, rtf_irr) - Rückflussraten und Sektoranteile

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT rtf_man, fra_man_f, fra_man_t, fra_man_p, fra_man_c, fra_man_g, fra_man_m, fra_man_nd, rtf_dom, rtf_irr \
               FROM country_input \
               WHERE IDScen = %s AND time = %s AND country_id = %s"

    values = (IDScen, time, country_id)
    mycursor.execute(query, values)
    return mycursor.fetchone()

def CountryConnectionToTreatment(dbname: str, IDScen: int, time: int, country_id: int) -> tuple:

Abwasserbehandlung und -entsorgungspfade für ein Land abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`IDScen`	`int`	Szenario-ID
`time`	`int`	Betrachtetes Jahr
`country_id`	`int`	Länder-ID

Rückgabe

tuple

(con_prim, con_sec, con_tert, con_untr, stp_failure, to_treat_and_unknown, to_hanging_t, to_open_def, UrbSewerConn, RurSewerConn, SPO_treat, con_quat) Behandlungs- und Entsorgungspfade in Prozent

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT con_prim, con_sec, con_tert, con_untr, stp_failure, to_treat_and_unknown, to_hanging_t, to_open_def, UrbSewerConn, RurSewerConn, SPO_treat, con_quat\
               FROM country_input \
               WHERE IDScen = %s AND time = %s AND country_id = %s"

    values = (IDScen, time, country_id)
    mycursor.execute(query, values)
    return mycursor.fetchone()

Nährstoff-Eliminationsraten und Bodenparameter

Hinweis: Die kombinierte Funktion RemovalRateSoilParametersAndTemperatureParameters wurde durch spezialisierte Funktionen ersetzt

def RemovalRate(dbname: str, IDScen: int, time: int, parameter_id: int) -> tuple:

Nährstoff-Eliminationsraten in Kläranlagen abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`IDScen`	`int`	Szenario-ID
`time`	`int`	Betrachtetes Jahr
`parameter_id`	`int`	Parameter-ID (Nährstoff: z.B. Phosphor=60)

Rückgabe

tuple

(rem_prim, rem_sec, rem_tert, rem_untr, treat_failure, rem_soil, red_fac_org, red_fac_inor, rem_quat) - Eliminationsraten und Reduktionsfaktoren

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT rem_prim, rem_sec, rem_tert, rem_untr, treat_failure, rem_soil, red_fac_org, red_fac_inor, rem_quat\
           FROM parameter_input \
           WHERE IDScen = %s AND time = %s AND parameter_id = %s "

    values = (IDScen, time, parameter_id)
    mycursor.execute(query, values)
    return mycursor.fetchone()

def SoilParametersForFC(dbname: str, IDScen: int, time: int, parameter_id: int) -> tuple:

Bodenparameter für Feldkapazität und Sedimentationsprozesse abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`IDScen`	`int`	Szenario-ID
`time`	`int`	Betrachtetes Jahr
`parameter_id`	`int`	Parameter-ID

Rückgabe

tuple

(k_storage, k_soil, ks, sed_veloc, ke_tss_reg_alpha, ke_tss_reg_beta, teta, teta_lake) - Bodenparameter

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT k_storage, k_soil, ks, sed_veloc, ke_tss_reg_alpha, ke_tss_reg_beta, teta, teta_lake\
           FROM parameter_input \
           WHERE IDScen = %s AND time = %s AND parameter_id = %s "

    values = (IDScen, time, parameter_id)
    mycursor.execute(query, values)
    return mycursor.fetchone()

def TemperatureParameters(dbname: str, IDScen: int, time: int, parameter_id: int) -> tuple:

Temperaturkoeffizenten für Abbauraten abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`IDScen`	`int`	Szenario-ID
`time`	`int`	Betrachtetes Jahr
`parameter_id`	`int`	Parameter-ID

Rückgabe

tuple

(teta, teta_lake) - Temperaturempfindlichkeitskoeffizenten

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT teta, teta_lake\
           FROM parameter_input \
           WHERE IDScen = %s AND time = %s AND parameter_id = %s "

    values = (IDScen, time, parameter_id)
    mycursor.execute(query, values)
    return mycursor.fetchone()

Düngemittelanwendungsraten (Stickstoff und Phosphor)

def Fertilizer_P_ApplicationRate(dbname: str, country_id: int, time: int) -> list:

Phosphor-Düngemittelanwendungsraten abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`country_id`	`int`	Länder-ID
`time`	`int`	Betrachtetes Jahr

Rückgabe

list

Anwendungsraten (t/km²) für jeden der 21 Fruchttypen

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT rate_P\
           FROM ind_fert_use_input \
           WHERE country_id = %s AND YearFrom = %s AND crop_type_id = %s"
    rate_P = []
    for crop_type_id in range(PP.crop_type_count):
        values = (country_id, time, crop_type_id)
        mycursor.execute(query, values)
        v = mycursor.fetchone()[0]
        rate_P.append(v)
    return rate_P

def Fertilizer_N_ApplicationRate(dbname: str, country_id: int, time: int) -> list:

Stickstoff-Düngemittelanwendungsraten abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`country_id`	`int`	Länder-ID
`time`	`int`	Betrachtetes Jahr

Rückgabe

list

Anwendungsraten (t/km²) für jeden der 21 Fruchttypen

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT rate_N\
           FROM ind_fert_use_input \
           WHERE country_id = %s AND YearFrom = %s AND crop_type_id = %s"
    rate_N = []
    for crop_type_id in range(PP.crop_type_count):
        values = (country_id, time, crop_type_id)
        mycursor.execute(query, values)
        v = mycursor.fetchone()[0]
        rate_N.append(v)
    return rate_N

Viehbestand-Exkretionsraten

def LivestockExcretionRate(dbname: str, parameter_id: int, IDFAOReg: int) -> list:

Exkretionsraten von Viehbeständen abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`parameter_id`	`int`	Parameter-ID (Nährstoff)
`IDFAOReg`	`int`	FAO-Regionen-ID

Rückgabe

list

Exkretionsraten für jeden Tiertyp

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT ls_exr_rate\
           FROM ls_exr_input \
           WHERE parameter_id = %s AND IDFAOReg = %s AND LS = %s"

    ls_exr_rate = []
    for LS in range(PP.livestock_type_count):
        values = (parameter_id, IDFAOReg, LS)
        mycursor.execute(query, values)
        ls_exr_rate.append(mycursor.fetchone()[0])
    return ls_exr_rate

Geogene Hintergrundbelastung

def GeogenicBackgroundDeposition(dbname: str, parameter_id: int, lu: int) -> float:

Geogene Hintergrundbelastung abrufen.

Parameter

Parameter	Typ	Beschreibung
`dbname`	`str`	Name der Datenbank
`parameter_id`	`int`	Parameter-ID (Nährstoff)
`lu`	`int`	Landnutzungs-Typ-ID

Rückgabe

float

Geogene Hintergrundbelastung

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT geo_back\
               FROM geogenic_background_input \
               WHERE parameter_id = %s AND LU = %s"

    values = (parameter_id, lu)
    mycursor.execute(query, values)
    return mycursor.fetchone()[0]

Datenkonvertierungen und Nachschlageabfragen

def Cell_ID_To_GCRC(cell: int, IDReg: int, dbname: str = "watergap_unf") -> int:

Rasterzellen-ID in GCRC-Format konvertieren.

Parameter

Parameter	Typ	Beschreibung
`cell`	`int`	Rasterzellen-ID
`IDReg`	`int`	Regionen-ID
`dbname`	`str, optional`	Name der Datenbank (Standard: "watergap_unf")

Rückgabe

int

GCRC-Rasterzellen-ID (Land/Wasser-Indikator)

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT cell_land_water\
           FROM gcrc \
           WHERE IDVersion = {} AND cell = %s AND IDReg = %s ".format(PP.db_gcrc_version)

    values = (cell, IDReg)
    mycursor.execute(query, values)
    return mycursor.fetchone()[0]

def Crop_ID_From_GLCC(GLCC: int, dbname: str = "wq_general") -> int:

Fruchttyp-ID aus GLCC-Landnutzungscode abrufen.

Parameter

Parameter	Typ	Beschreibung
`GLCC`	`int`	GLCC-Landnutzungscode (101-117, nicht 108, oder 999)
`dbname`	`str, optional`	Name der Datenbank (Standard: "wq_general")

Rückgabe

int

Fruchttyp-ID oder -9999 falls GLCC ungültig

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT crop_type_id\
           FROM _crop_type \
           WHERE GLCC = %s"
    Valid_GLCC_list = range(101,118)
    if ((GLCC in Valid_GLCC_list) and (GLCC != 108)) or GLCC==999:
        values = (GLCC,)
        mycursor.execute(query, values)
        crop_type_id = mycursor.fetchone()
    else:
        crop_type_id = -9999
    return crop_type_id

def IDFaoReg_from_Country_Id(country_id: int, dbname: str = "wq_general") -> int:

FAO-Regionen-ID aus Länder-ID abrufen.

Parameter

Parameter	Typ	Beschreibung
`country_id`	`int`	Länder-ID
`dbname`	`str, optional`	Name der Datenbank (Standard: "wq_general")

Rückgabe

int

FAO-Regionen-ID

    db = LoadDatabase(dbname)
    mycursor = db.cursor()

    query = "SELECT IDFAOReg \
           FROM _country \
           WHERE country_id = %s "
    values = (country_id,)
    mycursor.execute(query, values)
    return mycursor.fetchone()[0]