Merge remote-tracking branch 'origin/main'

Stochastisches_Modell.py

@@ -1,6 +1,8 @@
 import sympy as sp
 from dataclasses import dataclass, field
 from typing import Dict, Tuple, Iterable
+from Export import Export
+from Datenbank import Datenbankzugriff
 
 @dataclass
 class StochastischesModell:
@@ -48,6 +50,122 @@ class StochastischesModell:
             Q_ll[i, i] = q_ii                       #Diagonale
         return Q_ll
 
+    def Qll_symbolisch(self, pfad_datenbank, liste_beobachtungen_symbolisch):
+        liste_standardabweichungen_symbole = []
+        liste_beobachtungen_symbolisch = [str(b) for b in liste_beobachtungen_symbolisch]
+        Qll = sp.zeros(len(liste_beobachtungen_symbolisch), len(liste_beobachtungen_symbolisch))
+
+        db_zugriff = Datenbankzugriff(pfad_datenbank)
+        dict_beobachtungenID_instrumenteID = db_zugriff.get_instrumenteID_beobachtungenID_dict()
+
+        for i, beobachtung_symbolisch_i in enumerate(liste_beobachtungen_symbolisch):
+            aufgeteilt_i = beobachtung_symbolisch_i.split("_")
+            beobachtungenID_i = int(aufgeteilt_i[0])
+            instrumenteID_i = dict_beobachtungenID_instrumenteID[beobachtungenID_i]
+            beobachtungsart_i = str(aufgeteilt_i[1])
+
+            if beobachtungsart_i == "SD":
+                stabw_apriori_konstant = sp.Symbol(f"stabw_apriori_konstant_{beobachtungsart_i}_{instrumenteID_i}")
+                stabw_apriori_streckenprop = sp.Symbol(f"stabw_apriori_streckenprop_{beobachtungsart_i}_{instrumenteID_i}")
+                tachymeter_distanz = sp.Symbol(f"SD_{beobachtungenID_i}")
+
+                sigma = sp.sqrt(stabw_apriori_konstant ** 2 + (stabw_apriori_streckenprop * tachymeter_distanz / 1000000) ** 2)
+                liste_standardabweichungen_symbole.append(sigma)
+
+                Qll[i, i] = sigma ** 2
+
+            elif beobachtungsart_i == "R" or beobachtungsart_i == "ZW":
+                stabw_apriori_konstant = sp.Symbol(f"stabw_apriori_konstant_{beobachtungsart_i}_{instrumenteID_i}")
+
+                stabw_apriori_konstant_distanz = sp.Symbol(f"stabw_apriori_konstant_SD_{instrumenteID_i}")
+                tachymeter_distanz = sp.Symbol(f"SD_{beobachtungenID_i}")
+
+                sigma = sp.sqrt(
+                    stabw_apriori_konstant ** 2 + (stabw_apriori_konstant_distanz / tachymeter_distanz) ** 2)
+                liste_standardabweichungen_symbole.append(sigma)
+
+                Qll[i, i] = sigma ** 2
+
+            for j in range(i + 1, len(liste_beobachtungen_symbolisch)):
+                beobachtung_symbolisch_j = liste_beobachtungen_symbolisch[j]
+                aufgeteilt_j = beobachtung_symbolisch_j.split("_")
+                beobachtungsart_j = str(aufgeteilt_j[1])
+
+                if beobachtungsart_i == "SD" and beobachtungsart_j == "SD":
+                    Qll[i, j] = 0
+                    Qll[j, i] = 0
+
+        Export.matrix_to_csv(r"Zwischenergebnisse\Qll_Symbolisch.csv", liste_beobachtungen_symbolisch, liste_beobachtungen_symbolisch, Qll, "Qll")
+        return Qll
+
+    def Qll_numerisch(self, pfad_datenbank, Qll_Matrix_Symbolisch, liste_beobachtungen_symbolisch):
+        db_zugriff = Datenbankzugriff(pfad_datenbank)
+        dict_genauigkeiten = db_zugriff.get_genauigkeiten_dict()
+        dict_beobachtungenID_instrumenteID = db_zugriff.get_instrumenteID_beobachtungenID_dict()
+
+        liste_beobachtungen = db_zugriff.get_beobachtungen_from_beobachtungenid()
+        dict_beobachtungenID_distanz = {}
+        for standpunkt, zielpunkt, beobachtungenID, beobachtungsgruppeID, tachymeter_richtung, tachymeter_zenitwinkel, tachymeter_distanz in liste_beobachtungen:
+            dict_beobachtungenID_distanz[int(beobachtungenID)] = tachymeter_distanz
+
+        dict_genauigkeiten_neu = {}
+        for genauigkeitenID, eintrag in dict_genauigkeiten.items():
+            instrumenteID = int(eintrag[0])
+            beobachtungsart = str(eintrag[1])
+            stabw_apriori_konstant = eintrag[2]
+            stabw_apriori_streckenprop = eintrag[3]
+            dict_genauigkeiten_neu[(instrumenteID, beobachtungsart)] = (stabw_apriori_konstant,
+                                                                           stabw_apriori_streckenprop)
+
+        substitutionen = {}
+
+        dict_konstante_sd = {}
+        for (instrumenteID, beobachtungsart), (stabw_apriori_konstant,
+                                               stabw_apriori_streckenprop) in dict_genauigkeiten_neu.items():
+            if beobachtungsart == "Tachymeter_Strecke":
+                if stabw_apriori_konstant is not None:
+                    dict_konstante_sd[instrumenteID] = float(stabw_apriori_konstant)
+
+        for (instrumenteID, beobachtungsart), (stabw_apriori_konstant,
+                                               stabw_apriori_streckenprop) in dict_genauigkeiten_neu.items():
+
+            if beobachtungsart == "Tachymeter_Strecke":
+                beobachtungsart_kurz = "SD"
+            elif beobachtungsart == "Tachymeter_Richtung":
+                beobachtungsart_kurz = "R"
+            elif beobachtungsart == "Tachymeter_Zenitwinkel":
+                beobachtungsart_kurz = "ZW"
+
+
+            if stabw_apriori_konstant is not None:
+                substitutionen[sp.Symbol(f"stabw_apriori_konstant_{beobachtungsart_kurz}_{instrumenteID}")] = float(stabw_apriori_konstant)
+            if stabw_apriori_streckenprop is not None:
+                substitutionen[sp.Symbol(f"stabw_apriori_streckenprop_{beobachtungsart_kurz}_{instrumenteID}")] = float(stabw_apriori_streckenprop)
+
+        for instrumenteID, wert in dict_konstante_sd.items():
+            substitutionen[sp.Symbol(f"stabw_apriori_konstant_SD_{instrumenteID}")] = float(wert)
+
+        liste_beobachtungen_symbolisch = [str(b) for b in liste_beobachtungen_symbolisch]
+
+        for beobachtung_symbolisch in liste_beobachtungen_symbolisch:
+            aufgeteilt = beobachtung_symbolisch.split("_")
+            beobachtungenID = int(aufgeteilt[0])
+
+            distanz = dict_beobachtungenID_distanz.get(beobachtungenID, None)
+            if distanz is not None:
+                substitutionen[sp.Symbol(f"SD_{beobachtungenID}")] = float(distanz)
+
+        Qll_numerisch = Qll_Matrix_Symbolisch.xreplace(substitutionen)
+
+        Export.matrix_to_csv(
+            r"Zwischenergebnisse\Qll_Numerisch.csv",
+            liste_beobachtungen_symbolisch,
+            liste_beobachtungen_symbolisch,
+            Qll_numerisch,
+            "Qll"
+        )
+
+        return Qll_numerisch
 
     def berechne_P(Q_ll: sp.Matrix) -> sp.Matrix:
         P = Q_ll.inv()