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), 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), 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) ); """) 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": 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(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_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