297 lines
14 KiB
Python
297 lines
14 KiB
Python
import os
|
|
import sqlite3
|
|
import sympy as sp
|
|
from Berechnungen import Einheitenumrechnung
|
|
from decimal import Decimal
|
|
|
|
|
|
|
|
class Datenbank_anlegen:
|
|
def __init__(self, pfad_datenbank):
|
|
self.pfad_datenbank = pfad_datenbank
|
|
self.db_anlegen()
|
|
|
|
def db_anlegen(self):
|
|
# pfad = r"C:\Users\fabia\OneDrive\Jade HS\Master\MGW2\Masterprojekt_allgemein\Masterprojekt\Programmierung\Campusnetz\Campusnetz.db"
|
|
if not os.path.exists(self.pfad_datenbank):
|
|
con = sqlite3.connect(self.pfad_datenbank)
|
|
cursor = con.cursor()
|
|
cursor.executescript("""CREATE TABLE Netzpunkte (
|
|
punktnummer TEXT(10),
|
|
naeherungx_lh NUMERIC(9,3),
|
|
naeherungy_lh NUMERIC(7,3),
|
|
naeherungz_lh NUMERIC(8,3),
|
|
naeherungx_us NUMERIC(9,3),
|
|
naeherungy_us NUMERIC(7,3),
|
|
naeherungz_us NUMERIC(8,3),
|
|
datumskoordinate_x INTEGER,
|
|
datumskoordinate_y INTEGER,
|
|
datumskoordinate_z INTEGER,
|
|
stabw_vorinfo_x NUMERIC(3, 8),
|
|
stabw_vorinfo_y NUMERIC(3, 8),
|
|
stabw_vorinfo_z NUMERIC(3, 8),
|
|
CONSTRAINT pk_Netzpunkte PRIMARY KEY (punktnummer)
|
|
);
|
|
""");
|
|
cursor.executescript("""CREATE TABLE Beobachtungen(
|
|
beobachtungenID INTEGER,
|
|
beobachtungsgruppeID INTEGER,
|
|
punktnummer_sp TEXT(10),
|
|
punktnummer_zp TEXT(10),
|
|
instrumenteID INTEGER,
|
|
tachymeter_richtung NUMERIC(8, 6),
|
|
tachymeter_zenitwinkel NUMERIC(8, 6),
|
|
tachymeter_distanz NUMERIC(8, 4),
|
|
gnss_bx NUMERIC(7, 4),
|
|
gnss_by NUMERIC(7, 4),
|
|
gnss_bz NUMERIC(7, 4),
|
|
dateiname TEXT(200),
|
|
CONSTRAINT pk_Beobachtungen PRIMARY KEY (beobachtungenID),
|
|
CONSTRAINT fk_Beobachtungen_Netzpunktesp FOREIGN KEY (punktnummer_sp) REFERENCES Netzpunkte(punktnummer),
|
|
CONSTRAINT fk_Beobachtungen_Netzpunktezp FOREIGN KEY (punktnummer_zp) REFERENCES Netzpunkte(punktnummer),
|
|
CONSTRAINT fk_Beobachtungen_Instrumente FOREIGN KEY (instrumenteID) REFERENCES Instrumente(instrumenteID)
|
|
);
|
|
""");
|
|
cursor.executescript("""CREATE TABLE Instrumente(
|
|
instrumenteID INTEGER,
|
|
typ TEXT(200),
|
|
name TEXT(200),
|
|
CONSTRAINT pk_Instrumente PRIMARY KEY (instrumenteID)
|
|
);
|
|
""")
|
|
cursor.executescript("""CREATE TABLE Genauigkeiten(
|
|
genauigkeitenID INTEGER,
|
|
instrumenteID INTEGER,
|
|
beobachtungsart TEXT(25),
|
|
stabw_apriori_konstant NUMERIC(3, 8),
|
|
stabw_apriori_streckenprop NUMERIC(3, 8),
|
|
CONSTRAINT pk_Genauigkeiten PRIMARY KEY (genauigkeitenID),
|
|
CONSTRAINT fk_Genauigkeiten_Instrumente FOREIGN KEY (instrumenteID) REFERENCES Instrumente(instrumenteID)
|
|
);
|
|
""")
|
|
cursor.executescript("""CREATE TABLE GenauigkeitenGNSS
|
|
(
|
|
genauigkeitengnssID INTEGER,
|
|
beobachtungenID INTEGER,
|
|
m0 NUMERIC(1, 8),
|
|
q11 NUMERIC(1, 8),
|
|
q12 NUMERIC(1, 8),
|
|
q13 NUMERIC(1, 8),
|
|
q21 NUMERIC(1, 8),
|
|
q22 NUMERIC(1, 8),
|
|
q23 NUMERIC(1, 8),
|
|
q31 NUMERIC(1, 8),
|
|
q32 NUMERIC(1, 8),
|
|
q33 NUMERIC(1, 8),
|
|
CONSTRAINT pk_GenauigkeitenGNSS PRIMARY KEY (genauigkeitengnssID),
|
|
CONSTRAINT fk_GenauigkeitenGNSS_Beobachtungen FOREIGN KEY (beobachtungenID) REFERENCES Beobachtungen (beobachtungenID)
|
|
);
|
|
""")
|
|
con.commit()
|
|
cursor.close()
|
|
con.close()
|
|
|
|
class Datenbankzugriff:
|
|
def __init__(self, pfad_datenbank):
|
|
self.pfad_datenbank = pfad_datenbank
|
|
|
|
def get_koordinaten(self, koordinatenart, ausgabeart = "Dict"):
|
|
con = sqlite3.connect(self.pfad_datenbank)
|
|
cursor = con.cursor()
|
|
if koordinatenart == "naeherung_lh":
|
|
values = "punktnummer, naeherungx_lh, naeherungy_lh, naeherungz_lh"
|
|
elif koordinatenart == "naeherung_us":
|
|
values = "punktnummer, naeherungx_us, naeherungy_us, naeherungz_us"
|
|
|
|
liste_koordinaten = cursor.execute(f"""
|
|
SELECT {values} FROM Netzpunkte;
|
|
""").fetchall()
|
|
cursor.close()
|
|
con.close()
|
|
|
|
liste_koordinaten = [
|
|
koordinate for koordinate in liste_koordinaten
|
|
if koordinate[1] is not None and koordinate[2] is not None and koordinate[3] is not None
|
|
]
|
|
if ausgabeart == "Dict":
|
|
return {
|
|
koordinate[0]: sp.Matrix([
|
|
sp.Float(str(koordinate[1])),
|
|
sp.Float(str(koordinate[2])),
|
|
sp.Float(str(koordinate[3]))
|
|
])
|
|
for koordinate in liste_koordinaten
|
|
}
|
|
|
|
def set_koordinaten(self, dict_koordinaten, koordinatenart):
|
|
con = sqlite3.connect(self.pfad_datenbank)
|
|
cursor = con.cursor()
|
|
|
|
daten = []
|
|
|
|
for punktnummer, wert in dict_koordinaten.items():
|
|
daten.append((
|
|
float(wert[0]),
|
|
float(wert[1]),
|
|
float(wert[2]),
|
|
str(punktnummer)
|
|
))
|
|
|
|
if koordinatenart == "naeherung_lh":
|
|
pass
|
|
elif koordinatenart == "naeherung_us":
|
|
cursor.executemany(f"""UPDATE Netzpunkte SET naeherungx_us = ?, naeherungy_us = ?, naeherungz_us = ? WHERE punktnummer = ? AND naeherungx_us IS NULL AND naeherungy_us IS NULL AND naeherungz_us IS NULL""", daten)
|
|
|
|
|
|
con.commit()
|
|
cursor.close()
|
|
con.close()
|
|
|
|
def set_instrument(self, typ, name):
|
|
con = sqlite3.connect(self.pfad_datenbank)
|
|
cursor = con.cursor()
|
|
liste_instrumente = cursor.execute("SELECT * FROM Instrumente WHERE typ = ? AND name =?", (typ, name)).fetchall()
|
|
if liste_instrumente == []:
|
|
cursor.execute(
|
|
"INSERT INTO Instrumente (typ, name) VALUES (?, ?)", (typ, name)
|
|
)
|
|
print(f"Das Instrument {name} wurde erfolgreich hinzugefügt.")
|
|
else:
|
|
id_instrument = cursor.execute(
|
|
"SELECT instrumenteID FROM Instrumente WHERE typ = ? AND name =?", (typ, name))
|
|
print(f"Das Instrument {name} ist bereits in der Datenbank vorhanden.\nEs hat die ID {id_instrument.fetchone()[0]}")
|
|
con.commit()
|
|
cursor.close()
|
|
con.close()
|
|
|
|
def set_genauigkeiten(self, instrumenteID, beobachtungsart, stabw_apriori_konstant =None, stabw_apriori_streckenprop =None):
|
|
con = sqlite3.connect(self.pfad_datenbank)
|
|
cursor = con.cursor()
|
|
|
|
instrumentenname = cursor.execute("SELECT name FROM Instrumente WHERE instrumenteID = ?",
|
|
(instrumenteID, )).fetchone()
|
|
if instrumentenname is None:
|
|
print(
|
|
f"Die InstumentenID {instrumenteID} ist in der Datenbank nicht vorhanden. Bitte zuerst das Instrument hinzufügen.")
|
|
cursor.close()
|
|
con.close()
|
|
return
|
|
instrumentenname = instrumentenname[0]
|
|
|
|
if stabw_apriori_konstant is None and stabw_apriori_streckenprop is None:
|
|
print(
|
|
"Es wurden keine Genauigkeiten importiert. Bitte stabw_apriori_konstant und / oder stabw_apriori_streckenprop angeben.")
|
|
cursor.close()
|
|
con.close()
|
|
return
|
|
|
|
if beobachtungsart == "Tachymeter_Richtung" or beobachtungsart == "Tachymeter_Zenitwinkel" :
|
|
stabw_apriori_konstant = Einheitenumrechnung.mgon_to_rad_Decimal(stabw_apriori_konstant)
|
|
|
|
if beobachtungsart == "Tachymeter_Strecke":
|
|
stabw_apriori_konstant = Einheitenumrechnung.mm_to_m(stabw_apriori_konstant)
|
|
|
|
if isinstance(stabw_apriori_konstant, Decimal):
|
|
stabw_apriori_konstant = float(stabw_apriori_konstant)
|
|
|
|
if isinstance(stabw_apriori_streckenprop, Decimal):
|
|
stabw_apriori_streckenprop = float(stabw_apriori_streckenprop)
|
|
|
|
sql = "SELECT 1 FROM Genauigkeiten WHERE instrumenteID = ? AND beobachtungsart = ?"
|
|
params = [instrumenteID, beobachtungsart]
|
|
|
|
if stabw_apriori_konstant is None:
|
|
sql += " AND stabw_apriori_konstant IS NULL"
|
|
else:
|
|
sql += " AND stabw_apriori_konstant = ?"
|
|
params.append(stabw_apriori_konstant)
|
|
|
|
if stabw_apriori_streckenprop is None:
|
|
sql += " AND stabw_apriori_streckenprop IS NULL"
|
|
else:
|
|
sql += " AND stabw_apriori_streckenprop = ?"
|
|
params.append(stabw_apriori_streckenprop)
|
|
|
|
liste_genauigkeiten = cursor.execute(sql, tuple(params)).fetchall()
|
|
|
|
if liste_genauigkeiten == []:
|
|
if stabw_apriori_konstant is not None and stabw_apriori_streckenprop is not None:
|
|
cursor.execute(
|
|
"INSERT INTO Genauigkeiten (instrumenteID, beobachtungsart, stabw_apriori_konstant, stabw_apriori_streckenprop) VALUES (?, ?, ?, ?)",
|
|
(instrumenteID, beobachtungsart, stabw_apriori_konstant, stabw_apriori_streckenprop)
|
|
)
|
|
print(
|
|
f"Die Genauigkeitsangabe für die Beobachtungsart {beobachtungsart} des Instrumentes {instrumentenname} wurde erfolgreich hinzugefügt.")
|
|
|
|
elif stabw_apriori_konstant is None and stabw_apriori_streckenprop is not None:
|
|
cursor.execute(
|
|
"INSERT INTO Genauigkeiten (instrumenteID, beobachtungsart, stabw_apriori_streckenprop) VALUES (?, ?, ?)",
|
|
(instrumenteID, beobachtungsart, stabw_apriori_streckenprop)
|
|
)
|
|
print(
|
|
f"Die Genauigkeitsangabe für die Beobachtungsart {beobachtungsart} des Instrumentes {instrumentenname} wurde erfolgreich hinzugefügt.")
|
|
|
|
elif stabw_apriori_streckenprop is None and stabw_apriori_konstant is not None:
|
|
cursor.execute(
|
|
"INSERT INTO Genauigkeiten (instrumenteID, beobachtungsart, stabw_apriori_konstant) VALUES (?, ?, ?)",
|
|
(instrumenteID, beobachtungsart, stabw_apriori_konstant)
|
|
)
|
|
print(
|
|
f"Die Genauigkeitsangabe für die Beobachtungsart {beobachtungsart} des Instrumentes {instrumentenname} wurde erfolgreich hinzugefügt.")
|
|
else:
|
|
print("Die Genauigkeitsangabe ist bereits in der Datenbank vorhanden.")
|
|
|
|
con.commit()
|
|
cursor.close()
|
|
con.close()
|
|
|
|
def get_instrument_liste(self, typ):
|
|
con = sqlite3.connect(self.pfad_datenbank)
|
|
cursor = con.cursor()
|
|
liste_instrumente = cursor.execute("SELECT * FROM Instrumente WHERE typ = ?", (typ,)).fetchall()
|
|
liste_typen = cursor.execute("SELECT DISTINCT typ FROM Instrumente").fetchall()
|
|
cursor.close()
|
|
con.close()
|
|
if liste_instrumente == []:
|
|
liste_instrumente = f"Kein Instrument vom Typ {typ} gefunden. Folgende Typen stehen aktuell zur Auswahl: {liste_typen}"
|
|
return liste_instrumente
|
|
|
|
def get_genauigkeiten_dict(self):
|
|
dict = {}
|
|
con = sqlite3.connect(self.pfad_datenbank)
|
|
cursor = con.cursor()
|
|
liste_genauigkeiten = cursor.execute("SELECT * FROM Genauigkeiten").fetchall()
|
|
for genauigkeit in liste_genauigkeiten:
|
|
dict[genauigkeit[0]] = genauigkeit[1:]
|
|
cursor.close()
|
|
con.close()
|
|
return dict
|
|
|
|
def get_instrumenteID_beobachtungenID_dict(self):
|
|
dict = {}
|
|
con = sqlite3.connect(self.pfad_datenbank)
|
|
cursor = con.cursor()
|
|
liste_beobachtungen_instrumente = cursor.execute("SELECT beobachtungenID, instrumenteID FROM Beobachtungen").fetchall()
|
|
for i in liste_beobachtungen_instrumente:
|
|
dict[i[0]] = i[1]
|
|
cursor.close()
|
|
con.close()
|
|
return dict
|
|
|
|
def get_beobachtungen_id_beobachtungsgruppe_standpunkt_zielpunkt(self, beobachtungsart):
|
|
con = sqlite3.connect(self.pfad_datenbank)
|
|
cursor = con.cursor()
|
|
liste_beobachtungen = cursor.execute(f"SELECT beobachtungenID, beobachtungsgruppeID, punktnummer_sp, punktnummer_zp FROM Beobachtungen WHERE {beobachtungsart} IS NOT NULL").fetchall()
|
|
cursor.close()
|
|
con.close()
|
|
return liste_beobachtungen
|
|
|
|
def get_beobachtungen_from_beobachtungenid(self):
|
|
con = sqlite3.connect(self.pfad_datenbank)
|
|
cursor = con.cursor()
|
|
liste_beobachtungen = cursor.execute(f"SELECT punktnummer_sp, punktnummer_zp, beobachtungenID, beobachtungsgruppeID, tachymeter_richtung, tachymeter_zenitwinkel, tachymeter_distanz FROM Beobachtungen").fetchall()
|
|
cursor.close()
|
|
con.close()
|
|
return liste_beobachtungen
|
|
|