This commit is contained in:
2026-01-23 21:19:23 +01:00
parent 7c1b335b1c
commit 1561eb242e
8 changed files with 36417 additions and 35865 deletions

View File

@@ -147,6 +147,15 @@ class Berechnungen:
beobachtsgruppeID_vorher = beobachtsgruppeID_aktuell
return liste_azimut_richtungen, dict_orientierungen
def berechne_zenitwinkel_distanz_bodenbezogen(self, zenitwinkel_messung: float, schraegdistanz_messung: float,
instrumentenhoehe: float, prismenhoehe: float):
HD = np.sin(np.pi - zenitwinkel_messung) * schraegdistanz_messung
delta_h_ihzh = schraegdistanz_messung * np.cos(zenitwinkel_messung)
delta_h_boden = delta_h_ihzh + instrumentenhoehe - prismenhoehe
schraegdistanz_boden = np.sqrt(HD ** 2 + delta_h_boden ** 2)
zw_boden = np.atan2(HD, delta_h_boden)
return schraegdistanz_boden, zw_boden
class Einheitenumrechnung:
def __init__(self) -> None:

File diff suppressed because it is too large Load Diff

View File

@@ -47,10 +47,14 @@ class Datenbank_anlegen:
instrumenteID INTEGER,
tachymeter_richtung NUMERIC(8, 6),
tachymeter_richtung_ausschalten INTEGER DEFAULT 0,
tachymeter_zenitwinkel_roh NUMERIC(8, 6),
tachymeter_zenitwinkel NUMERIC(8, 6),
tachymeter_zenitwinkel_ausschalten INTEGER DEFAULT 0,
tachymeter_distanz_roh NUMERIC(8, 4),
tachymeter_distanz NUMERIC(8, 4),
tachymeter_distanz_auschalten INTEGER DEFAULT 0,
tachymeter_instrumentenhoehe NUMERIC(3, 8),
tachymeter_prismenhoehe NUMERIC(3, 8),
gnss_bx NUMERIC(7, 4),
gnss_bx_ausschalten INTEGER DEFAULT 0,
gnss_by NUMERIC(7, 4),

View File

@@ -5,110 +5,6 @@ from typing import Iterable, List, Sequence, Tuple, Optional
class Datumsfestlegung:
@staticmethod
def datumskomponenten(
auswahl: Iterable[Tuple[str, str]],
liste_punktnummern: Sequence[str],
*,
layout: str = "XYZ"
) -> List[int]:
punkt2pos = {str(p): i for i, p in enumerate(liste_punktnummern)}
layout = layout.upper()
if layout != "XYZ":
raise ValueError("Nur layout='XYZ' unterstützt (wie bei euch).")
comp2off = {"X": 0, "Y": 1, "Z": 2}
aktive: List[int] = []
for pt, comp in auswahl:
spt = str(pt)
c = comp.upper()
if spt not in punkt2pos:
raise KeyError(f"Punkt '{pt}' nicht in liste_punktnummern.")
if c not in comp2off:
raise ValueError(f"Komponente '{comp}' ungültig. Nur X,Y,Z.")
p = punkt2pos[spt]
aktive.append(3 * p + comp2off[c])
# Duplikate entfernen
out, seen = [], set()
for i in aktive:
if i not in seen:
seen.add(i)
out.append(i)
return out
@staticmethod
def auswahlmatrix_E(u: int, aktive_unbekannte_indices: Iterable[int]) -> sp.Matrix:
E = sp.zeros(u, u)
for idx in aktive_unbekannte_indices:
i = int(idx)
if not (0 <= i < u):
raise IndexError(f"Aktiver Index {i} außerhalb [0,{u-1}]")
E[i, i] = 1
return E
@staticmethod
def raenderungsmatrix_G(
x0: sp.Matrix,
liste_punktnummern: Sequence[str],
*,
mit_massstab: bool = True,
layout: str = "XYZ",
) -> sp.Matrix:
if x0.cols != 1:
raise ValueError("x0 muss Spaltenvektor sein.")
layout = layout.upper()
if layout != "XYZ":
raise ValueError("Nur layout='XYZ' unterstützt (wie bei euch).")
nP = len(liste_punktnummern)
u = x0.rows
d = 7 if mit_massstab else 6
G = sp.zeros(u, d)
for p in range(nP):
ix, iy, iz = 3*p, 3*p+1, 3*p+2
xi, yi, zi = x0[ix, 0], x0[iy, 0], x0[iz, 0]
# Translationen
G[ix, 0] = 1
G[iy, 1] = 1
G[iz, 2] = 1
# Rotationen
G[iy, 3] = -zi; G[iz, 3] = yi # Rx
G[ix, 4] = zi; G[iz, 4] = -xi # Ry
G[ix, 5] = -yi; G[iy, 5] = xi # Rz
# Maßstab
if mit_massstab:
G[ix, 6] = xi
G[iy, 6] = yi
G[iz, 6] = zi
return G
@staticmethod
def berechne_dx_geraendert(N: sp.Matrix, n: sp.Matrix, Gi: sp.Matrix) -> sp.Matrix:
if N.rows != N.cols:
raise ValueError("N muss quadratisch sein.")
if n.cols != 1:
raise ValueError("n muss Spaltenvektor sein.")
if Gi.rows != N.rows:
raise ValueError("Gi hat falsche Zeilenzahl.")
u = N.rows
d = Gi.cols
K = N.row_join(Gi)
K = K.col_join(Gi.T.row_join(sp.zeros(d, d)))
rhs = n.col_join(sp.zeros(d, 1))
sol = K.LUsolve(rhs)
return sol[:u, :]
@staticmethod
def weiches_datum(Q_ll: np.ndarray, Q_AA: np.ndarray) -> np.ndarray:
if Q_ll.ndim != 2 or Q_ll.shape[0] != Q_ll.shape[1]:
@@ -117,3 +13,92 @@ class Datumsfestlegung:
raise ValueError("Q_AA muss quadratisch sein.")
Q_ext = np.block([[Q_ll, np.zeros((Q_ll.shape[0], Q_AA.shape[0]))],[np.zeros((Q_AA.shape[0], Q_ll.shape[0])), Q_AA]])
return Q_ext
@staticmethod
def make_index(unbekannten_liste):
names = [str(s).strip() for s in unbekannten_liste]
return names, {n: i for i, n in enumerate(names)}
def build_G_from_names(x0, unbekannten_liste, liste_punktnummern, mit_massstab=True):
x0 = np.asarray(x0, float).reshape(-1)
names, idx = Datumsfestlegung.make_index(unbekannten_liste)
u = len(names)
d = 7 if mit_massstab else 6
G = np.zeros((u, d), dtype=float)
for pid in liste_punktnummern:
sx, sy, sz = f"X{pid}", f"Y{pid}", f"Z{pid}"
if sx not in idx or sy not in idx or sz not in idx:
# Punkt nicht als voller XYZ-Unbekannter vorhanden -> skip
continue
ix, iy, iz = idx[sx], idx[sy], idx[sz]
xi, yi, zi = x0[ix], x0[iy], x0[iz]
# Translationen
G[ix, 0] = 1.0
G[iy, 1] = 1.0
G[iz, 2] = 1.0
# Rotationen (δr = ω × r)
# Rx: δY=-Z, δZ=+Y
G[iy, 3] = -zi; G[iz, 3] = yi
# Ry: δX=+Z, δZ=-X
G[ix, 4] = zi; G[iz, 4] = -xi
# Rz: δX=-Y, δY=+X
G[ix, 5] = -yi; G[iy, 5] = xi
# Maßstab
if mit_massstab:
G[ix, 6] = xi
G[iy, 6] = yi
G[iz, 6] = zi
return G
def aktive_indices_from_selection(auswahl, unbekannten_liste):
names, idx = Datumsfestlegung.make_index(unbekannten_liste)
aktive = []
for pid, comp in auswahl:
key = f"{comp.upper()}{str(pid)}"
if key not in idx:
raise KeyError(f"{key} nicht im Unbekanntenvektor.")
aktive.append(idx[key])
# unique
out = []
seen = set()
for i in aktive:
if i not in seen:
seen.add(i)
out.append(i)
return out
def auswahlmatrix_E(u, aktive_indices):
E = np.zeros((u, u), dtype=float)
for i in aktive_indices:
E[int(i), int(i)] = 1.0
return E
def berechne_dx_geraendert(N, n, G):
N = np.asarray(N, float)
n = np.asarray(n, float).reshape(-1, 1)
G = np.asarray(G, float)
u = N.shape[0]
d = G.shape[1]
K = np.block([
[N, G],
[G.T, np.zeros((d, d))]
])
rhs = np.vstack([n, np.zeros((d, 1))])
sol = np.linalg.solve(K, rhs)
dx = sol[:u]
k = sol[u:]
return dx, k

440
Import.py
View File

@@ -2,6 +2,11 @@ import csv
import sqlite3
from decimal import Decimal
from typing import Any
import re
import xml.etree.ElementTree as ET
from decimal import Decimal, getcontext, ROUND_HALF_UP
from Berechnungen import Berechnungen
import Berechnungen
@@ -65,7 +70,390 @@ class Import:
con.close()
print("Der Import der Näherungskoordinaten wurde erfolgreich abgeschlossen")
def import_beobachtungen_tachymeter(self, pfad_datei: str, instrumentenID: int) -> None:
def ist_rundung_von_jxl(self, wert_csv: str, wert_jxl_voll: str) -> bool:
wert_csv = str(wert_csv).strip()
wert_jxl_voll = str(wert_jxl_voll).strip()
if ":ZH:" in wert_csv:
wert_csv = wert_csv.split(":ZH:", 1)[0].strip()
if ":ZH:" in wert_jxl_voll:
wert_jxl_voll = wert_jxl_voll.split(":ZH:", 1)[0].strip()
def ist_zahl(text: str) -> bool:
text = str(text).strip()
if text == "":
return False
if text[0] == "+" or text[0] == "-":
text = text[1:]
if text == "":
return False
if text.count(",") + text.count(".") > 1:
return False
if "," in text:
teile = text.split(",", 1)
if len(teile) != 2:
return False
if teile[0] == "" or teile[1] == "":
return False
return teile[0].isdigit() and teile[1].isdigit()
if "." in text:
teile = text.split(".", 1)
if len(teile) != 2:
return False
if teile[0] == "" or teile[1] == "":
return False
return teile[0].isdigit() and teile[1].isdigit()
return text.isdigit()
if ist_zahl(wert_csv) == False:
return False
if ist_zahl(wert_jxl_voll) == False:
return False
anzahl_nachkommastellen_csv = 0
if "," in wert_csv:
anzahl_nachkommastellen_csv = len(wert_csv.split(",", 1)[1])
elif "." in wert_csv:
anzahl_nachkommastellen_csv = len(wert_csv.split(".", 1)[1])
anzahl_nachkommastellen_jxl = 0
if "," in wert_jxl_voll:
anzahl_nachkommastellen_jxl = len(wert_jxl_voll.split(",", 1)[1])
elif "." in wert_jxl_voll:
anzahl_nachkommastellen_jxl = len(wert_jxl_voll.split(".", 1)[1])
if anzahl_nachkommastellen_csv >= anzahl_nachkommastellen_jxl:
return False
wert_csv_decimal = self.string_to_decimal(wert_csv.replace(".", ","))
wert_jxl_decimal = self.string_to_decimal(wert_jxl_voll.replace(".", ","))
q = Decimal("1") if anzahl_nachkommastellen_csv == 0 else Decimal("1." + ("0" * anzahl_nachkommastellen_csv))
wert_jxl_gerundet = wert_jxl_decimal.quantize(q, rounding=ROUND_HALF_UP)
wert_csv_gerundet = wert_csv_decimal.quantize(q, rounding=ROUND_HALF_UP)
return wert_jxl_gerundet == wert_csv_gerundet
def ist_zahl_csv(self, text: str) -> bool:
text = str(text).strip()
if text == "":
return False
if text[0] == "+" or text[0] == "-":
text = text[1:]
if text == "":
return False
if text.count(",") + text.count(".") > 1:
return False
if "," in text:
teile = text.split(",", 1)
if len(teile) != 2:
return False
if teile[0] == "" or teile[1] == "":
return False
return teile[0].isdigit() and teile[1].isdigit()
if "." in text:
teile = text.split(".", 1)
if len(teile) != 2:
return False
if teile[0] == "" or teile[1] == "":
return False
return teile[0].isdigit() and teile[1].isdigit()
return text.isdigit()
def korrigiere_beobachtungen_tachymeter_csv_mit_jxl(self,
pfad_datei_csv: str,
pfad_datei_jxl: str,
pfad_datei_csv_out: str) -> dict:
Import_fortsetzen = True
getcontext().prec = 70
dict_ersetzungen = {"Hz": 0, "Z": 0, "SD": 0}
liste_zeilen_ohne_IH = []
liste_zeilen_ohne_ZH = []
liste_zeilen_standpunkt_nicht_in_jxl = []
liste_stationrecords = []
dict_stationname_stationrecords = {}
dict_stationname_zaehler = {}
dict_targetID_zu_ZH = {}
dict_stationID_zu_seq = {}
dict_stationnamen = {}
tree = ET.parse(pfad_datei_jxl)
root = tree.getroot()
if Import_fortsetzen:
# StationRecords einlesen (Standpunkt, StationID, Instrumentenhöhe)
for sr in root.iter("StationRecord"):
stationname = (sr.findtext("StationName") or "").strip()
station_id = (sr.attrib.get("ID") or "").strip()
ih = (sr.findtext("TheodoliteHeight") or "").strip()
if stationname != "" and station_id != "":
liste_stationrecords.append((stationname, station_id, ih))
dict_stationnamen[stationname] = 1
if stationname not in dict_stationname_stationrecords:
dict_stationname_stationrecords[stationname] = []
dict_stationname_stationrecords[stationname].append((station_id, ih))
for stationname in dict_stationname_stationrecords.keys():
dict_stationname_zaehler[stationname] = 0
for tr in root.iter("TargetRecord"):
target_id = (tr.attrib.get("ID") or "").strip()
zh = (tr.findtext("TargetHeight") or "").strip()
if target_id != "":
dict_targetID_zu_ZH[target_id] = zh
for tupel in liste_stationrecords:
station_id = tupel[1]
if station_id not in dict_stationID_zu_seq:
dict_stationID_zu_seq[station_id] = []
for pr in root.iter("PointRecord"):
station_id = (pr.findtext("StationID") or "").strip()
if station_id == "" or station_id not in dict_stationID_zu_seq:
continue
circle = pr.find("Circle")
if circle is None:
continue
zielpunkt_name = (pr.findtext("Name") or "").strip()
target_id = (pr.findtext("TargetID") or "").strip()
hz_deg = (circle.findtext("HorizontalCircle") or "").strip()
z_deg = (circle.findtext("VerticalCircle") or "").strip()
sd_m = (circle.findtext("EDMDistance") or "").strip()
if zielpunkt_name == "" or hz_deg == "" or z_deg == "" or sd_m == "":
continue
stellen_hz = 0
if "." in hz_deg:
stellen_hz = len(hz_deg.split(".", 1)[1])
stellen_z = 0
if "." in z_deg:
stellen_z = len(z_deg.split(".", 1)[1])
stellen_sd = 0
if "." in sd_m:
stellen_sd = len(sd_m.split(".", 1)[1])
# Umrechnung Grad -> gon
hz_gon_decimal = Decimal(hz_deg) * (Decimal(10) / Decimal(9))
z_gon_decimal = Decimal(z_deg) * (Decimal(10) / Decimal(9))
q_hz = Decimal("1") if stellen_hz == 0 else Decimal("1." + ("0" * stellen_hz))
q_z = Decimal("1") if stellen_z == 0 else Decimal("1." + ("0" * stellen_z))
q_sd = Decimal("1") if stellen_sd == 0 else Decimal("1." + ("0" * stellen_sd))
hz_gon_decimal = hz_gon_decimal.quantize(q_hz, rounding=ROUND_HALF_UP)
z_gon_decimal = z_gon_decimal.quantize(q_z, rounding=ROUND_HALF_UP)
sd_decimal = Decimal(sd_m).quantize(q_sd, rounding=ROUND_HALF_UP)
# Ausgabe mit Komma
hz_gon_text = format(hz_gon_decimal, "f").replace(".", ",")
z_gon_text = format(z_gon_decimal, "f").replace(".", ",")
sd_text = format(sd_decimal, "f").replace(".", ",")
zh = dict_targetID_zu_ZH.get(target_id, "")
dict_stationID_zu_seq[station_id].append({
"target": zielpunkt_name,
"hz_gon": hz_gon_text,
"z_gon": z_gon_text,
"sd_m": sd_text,
"zh": zh
})
station_id_aktuell = None
index_seq_aktuell = 0
standpunkt_aktuell = None
# CSV-Datei zeilenweise durchgehen und ggf. Werte ersetzen
if Import_fortsetzen:
with (open(pfad_datei_csv, newline="", encoding="utf-8") as fin,
open(pfad_datei_csv_out, "w", newline="", encoding="utf-8") as fout):
reader = csv.reader(fin, delimiter=";")
writer = csv.writer(fout, delimiter=";", lineterminator="\n")
for i, row in enumerate(reader):
nummer_zeile = i + 1
if len(row) < 4:
row = row + [""] * (4 - len(row))
if row[0].strip() != "" and row[1].strip() == "" and row[2].strip() == "" and row[3].strip() == "":
standpunkt = row[0].strip()
if standpunkt in dict_stationnamen:
zaehler = dict_stationname_zaehler.get(standpunkt, 0)
liste_records = dict_stationname_stationrecords[standpunkt]
if zaehler >= len(liste_records):
Import_fortsetzen = False
print(
f"Der Vorgang wurde abgebrochen: Standpunkt {standpunkt} kommt in der CSV öfter vor als in der JXL.")
break
station_id, ih = liste_records[zaehler]
dict_stationname_zaehler[standpunkt] = zaehler + 1
station_id_aktuell = station_id
index_seq_aktuell = 0
standpunkt_aktuell = standpunkt
if ih is None or str(ih).strip() == "":
liste_zeilen_ohne_IH.append((nummer_zeile, standpunkt))
writer.writerow([standpunkt, f"IH:{ih}", "", "", ""])
continue
if standpunkt.isdigit():
liste_zeilen_standpunkt_nicht_in_jxl.append((nummer_zeile, standpunkt))
writer.writerow(row)
continue
ist_beobachtung = False
if row[0].strip() != "" and row[1].strip() != "" and row[2].strip() != "" and row[3].strip() != "":
wert_hz = row[1].split(":ZH:", 1)[0].strip()
wert_z = row[2].split(":ZH:", 1)[0].strip()
wert_sd = row[3].split(":ZH:", 1)[0].strip()
if self.ist_zahl_csv(wert_hz) and self.ist_zahl_csv(wert_z) and self.ist_zahl_csv(wert_sd):
ist_beobachtung = True
if ist_beobachtung and station_id_aktuell is not None:
zielpunkt = row[0].strip()
hz_csv = row[1].strip()
z_csv = row[2].strip()
sd_csv = row[3].strip()
liste_seq = dict_stationID_zu_seq.get(station_id_aktuell, [])
if liste_seq == []:
writer.writerow(row)
continue
if index_seq_aktuell >= len(liste_seq):
writer.writerow(row)
continue
jxl_eintrag = liste_seq[index_seq_aktuell]
index_neu = index_seq_aktuell + 1
if jxl_eintrag["target"] != zielpunkt:
index_ende = min(len(liste_seq), index_seq_aktuell + 200)
liste_kandidaten = []
for index_kandidat in range(index_seq_aktuell, index_ende):
if liste_seq[index_kandidat]["target"] == zielpunkt:
liste_kandidaten.append((index_kandidat, liste_seq[index_kandidat]))
if liste_kandidaten != []:
if len(liste_kandidaten) == 1:
index_kandidat, kandidat = liste_kandidaten[0]
jxl_eintrag = kandidat
index_neu = index_kandidat + 1
else:
liste_bewertung = []
for index_kandidat, kandidat in liste_kandidaten:
score = 0
if self.ist_rundung_von_jxl(hz_csv, kandidat["hz_gon"]):
score += 1
if self.ist_rundung_von_jxl(z_csv, kandidat["z_gon"]):
score += 1
if self.ist_rundung_von_jxl(sd_csv, kandidat["sd_m"]):
score += 1
liste_bewertung.append((score, index_kandidat, kandidat))
liste_bewertung.sort(key=lambda t: (-t[0], t[1]))
_, index_best, kandidat_best = liste_bewertung[0]
jxl_eintrag = kandidat_best
index_neu = index_best + 1
index_seq_aktuell = index_neu
# Nur ersetzen, wenn die CSV-Werte tatsächlich eine Rundung der JXL-Werte sind
hz_out = hz_csv
z_out = z_csv
sd_out = sd_csv
if self.ist_rundung_von_jxl(hz_csv, jxl_eintrag["hz_gon"]):
hz_out = jxl_eintrag["hz_gon"]
dict_ersetzungen["Hz"] += 1
if self.ist_rundung_von_jxl(z_csv, jxl_eintrag["z_gon"]):
z_out = jxl_eintrag["z_gon"]
dict_ersetzungen["Z"] += 1
if self.ist_rundung_von_jxl(sd_csv, jxl_eintrag["sd_m"]):
sd_out = jxl_eintrag["sd_m"]
dict_ersetzungen["SD"] += 1
zh = jxl_eintrag.get("zh", "")
if zh is None or str(zh).strip() == "":
liste_zeilen_ohne_ZH.append((nummer_zeile, standpunkt_aktuell, zielpunkt))
spalte_letzte = f"{sd_out}:ZH:{zh}" if str(zh).strip() != "" else sd_out
writer.writerow([zielpunkt, hz_out, z_out, spalte_letzte])
continue
writer.writerow(row)
if Import_fortsetzen:
print(f"Korrektur erfolgreich abgeschlossen. Ausgabe: {pfad_datei_csv_out}")
print(f"Ersetzungen (Rundung -> JXL volle Nachkommastellen): {dict_ersetzungen}")
print("\n--- Fehlende IH ---")
print(f"Anzahl: {len(liste_zeilen_ohne_IH)}")
if len(liste_zeilen_ohne_IH) > 0:
print(liste_zeilen_ohne_IH)
print("\n--- Fehlende ZH ---")
print(f"Anzahl: {len(liste_zeilen_ohne_ZH)}")
if len(liste_zeilen_ohne_ZH) > 0:
print(liste_zeilen_ohne_ZH)
print("\n--- Standpunkt in CSV, aber kein StationRecord in JXL ---")
print(f"Anzahl: {len(liste_zeilen_standpunkt_nicht_in_jxl)}")
if len(liste_zeilen_standpunkt_nicht_in_jxl) > 0:
print(liste_zeilen_standpunkt_nicht_in_jxl)
else:
print("Die Korrektur wurde abgebrochen.")
return {
"Import_fortsetzen": Import_fortsetzen,
"dict_ersetzungen": dict_ersetzungen,
"liste_zeilen_ohne_IH": liste_zeilen_ohne_IH,
"liste_zeilen_ohne_ZH": liste_zeilen_ohne_ZH,
"liste_zeilen_standpunkt_nicht_in_jxl": liste_zeilen_standpunkt_nicht_in_jxl,
"pfad_datei_csv_out": pfad_datei_csv_out
}
def import_beobachtungen_tachymeter(self, pfad_datei: str, instrumentenID: int, a: float, b: float) -> None:
berechnungen = Berechnungen.Berechnungen(a, b)
# Prüfen, ob Bereits Daten aus der Datei in der Datenbank vorhanden sind
con = sqlite3.connect(self.pfad_datenbank)
cursor = con.cursor()
@@ -102,10 +490,16 @@ class Import:
if i < 3:
continue
zeile = zeile.strip().split(";")
if zeile[1] == "" and zeile[2] == "" and zeile[3] == "":
if len(zeile) < 5:
zeile = zeile + [""] * (5 - len(zeile))
if zeile[2] == "" and zeile[3] == "" and zeile[4] == "":
nummer_beobachtungsgruppeID += 1
# print("Standpunkt: ",nummer_beobachtungsgruppeID ,zeile[0])
standpunkt = zeile[0]
instrumentenhoehe = zeile[1]
if instrumentenhoehe.startswith("IH:"):
instrumentenhoehe = instrumentenhoehe.split("IH:", 1)[1].strip()
if nummer_zielpunkt % 6 != 0:
liste_fehlerhafte_zeile.append(i)
@@ -116,24 +510,30 @@ class Import:
liste_zielpunkte_vs3 = []
else:
nummer_zielpunkt += 1
if ":ZH:" in zeile[3]:
teil = zeile[3].split(":ZH:", 1)
zeile[3] = teil[0].strip()
zeile[4] = teil[1].strip()
if zeile[0] not in liste_zielpunkte_hs:
liste_zielpunkte_hs.append(zeile[0])
if zeile[0] in liste_zielpunkte_vs3:
# print(f"{nummer_zielpunkt} VS3 HS1 {zeile}")
liste_beobachtungen_vorbereitung.append(
[nummer_beobachtungsgruppeID, "VS3", "HS1", standpunkt, zeile[0], zeile[1],
zeile[2], zeile[3]])
zeile[2], zeile[3], zeile[4], instrumentenhoehe])
elif zeile[0] in liste_zielpunkte_vs2:
# print(f"{nummer_zielpunkt} VS2 HS1 {zeile}")
liste_beobachtungen_vorbereitung.append(
[nummer_beobachtungsgruppeID, "VS2", "HS1", standpunkt, zeile[0], zeile[1],
zeile[2], zeile[3]])
zeile[2], zeile[3], zeile[4], instrumentenhoehe])
else:
# print(f"{nummer_zielpunkt} VS1 HS1 {zeile}")
liste_beobachtungen_vorbereitung.append(
[nummer_beobachtungsgruppeID, "VS1", "HS1", standpunkt, zeile[0], zeile[1],
zeile[2],
zeile[3]])
zeile[3], zeile[4], instrumentenhoehe])
else:
liste_zielpunkte_hs.remove(zeile[0])
@@ -142,7 +542,7 @@ class Import:
liste_beobachtungen_vorbereitung.append(
[nummer_beobachtungsgruppeID, "VS3", "HS2", standpunkt, zeile[0], zeile[1],
zeile[2],
zeile[3]])
zeile[3], zeile[4], instrumentenhoehe])
elif zeile[0] in liste_zielpunkte_vs2:
if zeile[0] not in liste_zielpunkte_vs3:
@@ -151,7 +551,7 @@ class Import:
liste_beobachtungen_vorbereitung.append(
[nummer_beobachtungsgruppeID, "VS2", "HS2", standpunkt, zeile[0], zeile[1],
zeile[2],
zeile[3]])
zeile[3], zeile[4], instrumentenhoehe])
else:
if zeile[0] not in liste_zielpunkte_vs2:
liste_zielpunkte_vs2.append(zeile[0])
@@ -159,7 +559,7 @@ class Import:
liste_beobachtungen_vorbereitung.append(
[nummer_beobachtungsgruppeID, "VS1", "HS2", standpunkt, zeile[0], zeile[1],
zeile[2],
zeile[3]])
zeile[3], zeile[4], instrumentenhoehe])
if liste_fehlerhafte_zeile == []:
# print(f"Einlesen der Datei {pfad_datei} erfolgreich beendet.")
@@ -200,11 +600,28 @@ class Import:
richtung_vollsatz_gon = (richtung1 + richtung2) / 2
richtung_vollsatz_rad = Berechnungen.Einheitenumrechnung.gon_to_rad_Decimal(richtung_vollsatz_gon)
if liste_aktueller_zielpunkt[8] == liste[8]:
prismenhoehe = liste_aktueller_zielpunkt[8]
else:
Import_fortsetzen = False
print(f"Der Import wurde abgebrochen, weil für zwei Halbsätze vom Standpunkt {liste_aktueller_zielpunkt[3]} zum Zielpunkt {aktueller_zielpunkt} unterschiedliche Prismenhöhen vorliegen. Bitte in der Datei {pfad_datei} korrigieren und Import neustarten.")
if liste_aktueller_zielpunkt[9] == liste[9]:
instrumentenhoehe_import = liste_aktueller_zielpunkt[9]
else:
Import_fortsetzen = False
print(f"Der Import wurde abgebrochen, weil für zwei Halbsätze vom Standpunkt {liste_aktueller_zielpunkt[3]} zum Zielpunkt {aktueller_zielpunkt} unterschiedliche Instrumentenhöhen vorliegen. Bitte in der Datei {pfad_datei} korrigieren und Import neustarten.")
# print(richtung_vollsatz)
# print(zenitwinkel_vollsatz)
# print(distanz_vollsatz)
schraegdistanz_bodenbezogen, zenitwinkel_bodenbezogen = berechnungen.berechne_zenitwinkel_distanz_bodenbezogen(
float(zenitwinkel_vollsatz_rad), float(distanz_vollsatz), float(instrumentenhoehe_import), float(prismenhoehe))
liste_beobachtungen_import.append(
[liste[0], liste[3], liste[4], richtung_vollsatz_rad, zenitwinkel_vollsatz_rad, distanz_vollsatz])
[liste[0], liste[3], liste[4], richtung_vollsatz_rad, zenitwinkel_vollsatz_rad, distanz_vollsatz, zenitwinkel_bodenbezogen, schraegdistanz_bodenbezogen, instrumentenhoehe_import, prismenhoehe])
del liste_beobachtungen_vorbereitung[index]
del liste_beobachtungen_vorbereitung[0]
@@ -215,15 +632,16 @@ class Import:
print(
"Der Import wurde abgebrochen. Bitte eine gültige InstrumentenID eingeben. Bei Bedarf ist das Instrument neu anzulegen.")
if Import_fortsetzen:
con = sqlite3.connect(self.pfad_datenbank)
cursor = con.cursor()
for beobachtung_import in liste_beobachtungen_import:
cursor.execute(
"INSERT INTO Beobachtungen (punktnummer_sp, punktnummer_zp, instrumenteID, beobachtungsgruppeID, tachymeter_richtung, tachymeter_zenitwinkel, tachymeter_distanz, dateiname) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
"INSERT INTO Beobachtungen (punktnummer_sp, punktnummer_zp, instrumenteID, beobachtungsgruppeID, tachymeter_richtung, tachymeter_zenitwinkel_roh, tachymeter_distanz_roh, dateiname, tachymeter_instrumentenhoehe, tachymeter_prismenhoehe, tachymeter_zenitwinkel, tachymeter_distanz) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(beobachtung_import[1], beobachtung_import[2], instrumentenID, beobachtung_import[0],
float(beobachtung_import[3]), float(beobachtung_import[4]), float(beobachtung_import[5]),
pfad_datei))
pfad_datei, float(beobachtung_import[8]), float(beobachtung_import[9]), float(beobachtung_import[6]), float(beobachtung_import[7])))
con.commit()
cursor.close()
con.close()

View File

@@ -1,6 +1,6 @@
import numpy as np
import plotly.graph_objects as go
from scipy.stats import f as f_dist
from scipy.stats import f
import pandas as pd
@@ -57,99 +57,143 @@ class Genauigkeitsmaße:
def standardellipse(Qxx, s0_apost, unbekannten_liste, dim_labels=3):
@staticmethod
def standardellipse(Qxx, s0_apost, unbekannten_liste):
Qxx = np.asarray(Qxx, float)
data = []
daten = []
namen_str = [str(sym) for sym in unbekannten_liste]
n_punkte = len(unbekannten_liste) // dim_labels
punkt_ids = []
for n in namen_str:
if n.upper().startswith('X'):
punkt_ids.append(n[1:])
for i in range(n_punkte):
sym_x = str(unbekannten_liste[dim_labels * i]) # z.B. "X10009"
punkt = sym_x[1:] # -> "10009"
for pid in punkt_ids:
try:
idx_x = next(i for i, n in enumerate(namen_str) if n.upper() == f"X{pid}".upper())
idx_y = next(i for i, n in enumerate(namen_str) if n.upper() == f"Y{pid}".upper())
ix = dim_labels * i
iy = dim_labels * i + 1
qxx = Qxx[idx_x, idx_x]
qyy = Qxx[idx_y, idx_y]
qyx = Qxx[idx_y, idx_x]
# 2x2-Kofaktorblock
Qxx_ = Qxx[ix, ix]
Qyy_ = Qxx[iy, iy]
Qyx_ = Qxx[iy, ix]
# Standardabweichungen
sx = s0_apost * np.sqrt(qxx)
sy = s0_apost * np.sqrt(qyy)
sxy = (s0_apost ** 2) * qyx
# Standardabweichungen der Koordinatenkomponenten
sx = s0_apost * np.sqrt(Qxx_)
sy = s0_apost * np.sqrt(Qyy_)
sxy = (s0_apost ** 2) * Qyx_
k = np.sqrt((qxx - qyy) ** 2 + 4 * (qyx ** 2))
# k und Eigenwerte (Q_dmax, Q_dmin)
k = np.sqrt((Qxx_ - Qyy_) ** 2 + 4 * (Qyx_ ** 2))
Q_dmax = 0.5 * (Qxx_ + Qyy_ + k)
Q_dmin = 0.5 * (Qxx_ + Qyy_ - k)
# Q_dmax/min = 0.5 * (Qyy + Qxx +/- k)
q_dmax = 0.5 * (qyy + qxx + k)
q_dmin = 0.5 * (qyy + qxx - k)
# Halbachsen (Standardabweichungen entlang Hauptachsen)
s_max = s0_apost * np.sqrt(Q_dmax)
s_min = s0_apost * np.sqrt(Q_dmin)
# Halbachsen
s_max = s0_apost * np.sqrt(q_dmax)
s_min = s0_apost * np.sqrt(q_dmin)
# Richtungswinkel theta (Hauptachse) in rad:
theta_rad = 0.5 * np.arctan2(2 * Qyx_, (Qxx_ - Qyy_))
# Richtungswinkel theta in gon:
zaehler = 2 * qyx
nenner = qxx - qyy
t_grund = 0.5 * np.arctan(abs(zaehler) / abs(nenner)) * (200 / np.pi)
# in gon
theta_gon = theta_rad * (200 / np.pi)
if theta_gon < 0:
theta_gon += 200.0
# Quadrantenabfrage
if nenner > 0 and qyx > 0: # Qxx - Qyy > 0 und Qyx > 0
t_gon = t_grund # 0 - 50 gon
elif nenner < 0 and qyx > 0: # Qxx - Qyy < 0 und Qyx > 0
t_gon = 100 - t_grund # 50 - 100 gon
elif nenner < 0 and qyx < 0: # Qxx - Qyy < 0 und Qyx < 0
t_gon = 100 + t_grund # 100 - 150 gon
elif nenner > 0 and qyx < 0: # Qxx - Qyy > 0 und Qyx < 0
t_gon = 200 - t_grund # 150 - 200 gon
else:
t_gon = 0.0
data.append([
punkt,
float(sx), float(sy), float(sxy),
float(s_max), float(s_min),
float(theta_gon)
])
daten.append([
pid,
float(sx), float(sy), float(sxy),
float(s_max), float(s_min),
float(t_gon)
])
standardellipse = pd.DataFrame(data, columns=["Punkt", "σx", "σy", "σxy", "s_max", "s_min", "θ [gon]"])
except:
continue
standardellipse = pd.DataFrame(daten, columns=["Punkt", "σx", "σy", "σxy", "s_max", "s_min", "θ [gon]"])
return standardellipse
def konfidenzellipse(Qxx, s0_apost, unbekannten_liste, R, alpha=0.05):
@staticmethod
def konfidenzellipse(Qxx, s0_apost, unbekannten_liste, R, alpha):
Qxx = np.asarray(Qxx, float)
daten = []
namen_str = [str(sym) for sym in unbekannten_liste]
data = []
n_punkte = len(unbekannten_liste) // 3 # X,Y,Z je Punkt angenommen
punkt_ids = [n[1:] for n in namen_str if n.upper().startswith('X')]
k = float(np.sqrt(2.0 * f_dist.ppf(1.0 - alpha, 2, R)))
# Faktor für Konfidenzellipse (F-Verteilung)
kk = float(np.sqrt(2.0 * f.ppf(1.0 - alpha, 2, R)))
for i in range(n_punkte):
punkt = str(unbekannten_liste[3 * i])[1:] # "X10009" -> "10009"
for pid in punkt_ids:
try:
idx_x = next(i for i, n in enumerate(namen_str) if n.upper() == f"X{pid}".upper())
idx_y = next(i for i, n in enumerate(namen_str) if n.upper() == f"Y{pid}".upper())
ix = 3 * i
iy = 3 * i + 1
qxx = Qxx[idx_x, idx_x]
qyy = Qxx[idx_y, idx_y]
qyx = Qxx[idx_y, idx_x]
Qxx_ = Qxx[ix, ix]
Qyy_ = Qxx[iy, iy]
Qxy_ = Qxx[iy, ix] # = Qyx
# Standardabweichungen
sx = s0_apost * np.sqrt(qxx)
sy = s0_apost * np.sqrt(qyy)
sxy = (s0_apost ** 2) * qyx
# k für Eigenwerte
kk = np.sqrt((Qxx_ - Qyy_) ** 2 + 4 * (Qxy_ ** 2))
Q_dmax = 0.5 * (Qxx_ + Qyy_ + kk)
Q_dmin = 0.5 * (Qxx_ + Qyy_ - kk)
k = np.sqrt((qxx - qyy) ** 2 + 4 * (qyx ** 2))
# Standard-Halbachsen (1-sigma)
s_max = s0_apost * np.sqrt(Q_dmax)
s_min = s0_apost * np.sqrt(Q_dmin)
# Q_dmax/min = 0.5 * (Qyy + Qxx +/- k)
q_dmax = 0.5 * (qyy + qxx + k)
q_dmin = 0.5 * (qyy + qxx - k)
# Orientierung (Hauptachse) in gon
theta_rad = 0.5 * np.arctan2(2 * Qxy_, (Qxx_ - Qyy_))
theta_gon = theta_rad * (200 / np.pi)
if theta_gon < 0:
theta_gon += 200.0
# Halbachsen der Standardellipse
s_max = s0_apost * np.sqrt(q_dmax)
s_min = s0_apost * np.sqrt(q_dmin)
# Konfidenz-Halbachsen
a_K = k * s_max
b_K = k * s_min
# Halbachsen der Konfidenzellipse
A_K = kk * s_max
B_K = kk * s_min
data.append([punkt, float(a_K), float(b_K), float(theta_gon)])
# Richtungswinkel theta in gon:
zaehler = 2 * qyx
nenner = qxx - qyy
t_grund = 0.5 * np.arctan(abs(zaehler) / abs(nenner)) * (200 / np.pi)
# Quadrantenabfrage
if nenner > 0 and qyx > 0:
t_gon = t_grund # 0 - 50 gon
elif nenner < 0 and qyx > 0:
t_gon = 100 - t_grund # 50 - 100 gon
elif nenner < 0 and qyx < 0:
t_gon = 100 + t_grund # 100 - 150 gon
elif nenner > 0 and qyx < 0:
t_gon = 200 - t_grund # 150 - 200 gon
else:
t_gon = 0.0
daten.append([
pid,
float(sx), float(sy), float(sxy),
float(A_K), float(B_K),
float(t_gon)
])
except:
continue
konfidenzellipse = pd.DataFrame(daten, columns= ["Punkt", "σx", "σy", "σxy", "a_K", "b_K","θ [gon]"])
return konfidenzellipse
konfidenzellipsen = pd.DataFrame(data, columns=["Punkt", "a_K", "b_K", "θ [gon]"])
return konfidenzellipsen
def plot_netz_komplett_final(x_vektor, unbekannten_labels, beobachtungs_labels, Qxx, sigma0_apost,

View File

@@ -7,22 +7,26 @@ import pandas as pd
@dataclass
class Zuverlaessigkeit:
@staticmethod
def gesamtredundanz(n, u):
r = n - u
return r
@staticmethod
def berechne_R(Q_vv, P):
R = Q_vv @ P
return R #Redundanzmatrix
@staticmethod
def berechne_ri(R):
ri = np.diag(R)
EVi = 100.0 * ri
return ri, EVi #Redundanzanteile
@staticmethod
def klassifiziere_ri(ri): #Klassifizierung der Redundanzanteile
if ri < 0.01:
return "nicht kontrollierbar"
@@ -35,7 +39,7 @@ class Zuverlaessigkeit:
else:
return "nahezu vollständig redundant"
@staticmethod
def globaltest(r_gesamt, sigma0_apost, sigma0_apriori, alpha):
T_G = (sigma0_apost ** 2) / (sigma0_apriori ** 2)
F_krit = stats.f.ppf(1 - alpha, r_gesamt, 10 ** 9)
@@ -112,7 +116,7 @@ class Zuverlaessigkeit:
return Lokaltest_innere_Zuv
def aeussere_zuverlaessigkeit_EF_EP(Lokaltest, labels, Qxx, A, P, s0_apost, unbekannten_liste, x):
def aeussere_zuverlaessigkeit_EF_EP1(Lokaltest, labels, Qxx, A, P, s0_apost, unbekannten_liste, x):
df = Lokaltest.copy()
labels = list(labels)
Qxx = np.asarray(Qxx, float)
@@ -214,3 +218,97 @@ class Zuverlaessigkeit:
})
return out
def aeussere_zuverlaessigkeit_EF_EP_stabil(Lokaltest, labels, Qxx, A, P, s0_apost, unbekannten_liste, x):
df = Lokaltest.copy()
labels = list(labels)
Qxx = np.asarray(Qxx, float)
A = np.asarray(A, float)
P = np.asarray(P, float)
x = np.asarray(x, float).reshape(-1)
ri = df["r_i"].astype(float).to_numpy()
GF = df["GF_i"].astype(float).to_numpy()
GRZW = df["GRZW_i"].astype(float).to_numpy()
n = A.shape[0]
# Namen als Strings für die Suche
namen_str = [str(sym) for sym in unbekannten_liste]
# 1) Einflussfaktor EF berechnen
EF = np.zeros(n, dtype=float)
for i in range(n):
nabla_l = np.zeros((n, 1))
nabla_l[i, 0] = GRZW[i]
nabla_x = Qxx @ (A.T @ (P @ nabla_l))
Qxx_inv_nabla_x = np.linalg.solve(Qxx, nabla_x)
EF2 = ((nabla_x.T @ Qxx_inv_nabla_x) / (float(s0_apost) ** 2)).item()
EF[i] = np.sqrt(max(0, EF2))
# 2) Koordinaten-Dict
coords = {}
punkt_ids = [n[1:] for n in namen_str if n.upper().startswith("X")]
for pid in punkt_ids:
try:
ix = namen_str.index(f"X{pid}")
iy = namen_str.index(f"Y{pid}")
iz = namen_str.index(f"Z{pid}")
coords[pid] = (x[ix], x[iy], x[iz] if iz is not None else 0.0)
except:
continue
# 3) EP + Standpunkte
EP_m = np.full(len(labels), np.nan, dtype=float)
standpunkte = [""] * len(labels)
for i, lbl in enumerate(labels):
parts = lbl.split("_")
sp, zp = None, None
if any(k in lbl for k in ["_SD_", "_R_", "_ZW_"]):
if len(parts) >= 5: sp, zp = parts[3].strip(), parts[4].strip()
elif "gnss" in lbl.lower():
sp, zp = parts[-2].strip(), parts[-1].strip()
elif "niv" in lbl.lower():
if len(parts) >= 4:
sp = parts[3].strip()
else:
sp = parts[-1].strip()
standpunkte[i] = sp if sp is not None else ""
one_minus_r = (1.0 - ri[i])
# SD, GNSS, Niv: direkt Wegfehler
if "_SD_" in lbl or "gnss" in lbl.lower() or "niv" in lbl.lower():
EP_m[i] = one_minus_r * GF[i]
# Winkel: Streckenäquivalent
elif "_R_" in lbl or "_ZW_" in lbl:
if sp in coords and zp in coords:
X1, Y1, _ = coords[sp]
X2, Y2, _ = coords[zp]
s = np.sqrt((X2 - X1) ** 2 + (Y2 - Y1) ** 2)
EP_m[i] = one_minus_r * (GF[i] * s)
# 4) SP am Standpunkt (2D oder 1D)
diagQ = np.diag(Qxx)
SP_cache_mm = {}
for sp in set([s for s in standpunkte if s]):
try:
ix = namen_str.index(f"X{sp}")
iy = namen_str.index(f"Y{sp}")
SP_cache_mm[sp] = float(s0_apost) * np.sqrt(diagQ[ix] + diagQ[iy]) * 1000.0
except ValueError:
# Falls keine Lage, prüfe Höhe (Nivellement)
try:
iz = namen_str.index(f"Z{sp}")
SP_cache_mm[sp] = float(s0_apost) * np.sqrt(diagQ[iz]) * 1000.0
except ValueError:
SP_cache_mm[sp] = 0.0
SP_mm = np.array([SP_cache_mm.get(sp, np.nan) for sp in standpunkte], dtype=float)
return pd.DataFrame({
"Beobachtung": labels, "Stand-Pkt": standpunkte, "EF": EF,
"EP [mm]": EP_m * 1000.0, "SP [mm]": SP_mm, "EF*SP [mm]": EF * SP_mm
})

View File

@@ -47,6 +47,7 @@ def ausgleichung_global(A, dl, Q_ext):
return dict_ausgleichung, dx
def ausgleichung_lokal(A, dl, Q_ll):
A = np.asarray(A, dtype=float)
dl = np.asarray(dl, dtype=float).reshape(-1, 1)
@@ -60,14 +61,14 @@ def ausgleichung_lokal(A, dl, Q_ll):
n = A.T @ P @ dl
# 3) Datumsfestlegung
G = Datumsfestlegung.raenderungsmatrix_G(x0_sp, liste_punktnummern, mit_massstab=mit_massstab)
aktive = Datumsfestlegung.datumskomponenten(auswahl, liste_punktnummern)
E = Datumsfestlegung.auswahlmatrix_E(u=A.shape[1], aktive_unbekannte_indices=aktive)
Gi_sp = E * G
Gi = np.asarray(Gi_sp, dtype=float)
G = Datumsfestlegung.build_G_from_names(x0, Jacobimatrix_symbolisch_liste_unbekannte, liste_punktnummern, mit_massstab=True)
u = A.shape[1]
aktive = Datumsfestlegung.aktive_indices_from_selection(auswahl, Jacobimatrix_symbolisch_liste_unbekannte)
E = Datumsfestlegung.auswahlmatrix_E(u, aktive)
Gi = E @ G
# 3) Zuschlagsvektor dx
dx = np.linalg.solve(N, n)
dx, k = Datumsfestlegung.berechne_dx_geraendert(N, n, Gi)
# 5) Residuen
v = dl - A @ dx