Push
This commit is contained in:
98182
Campusnetz.ipynb
98182
Campusnetz.ipynb
File diff suppressed because it is too large
Load Diff
@@ -15,37 +15,43 @@ class Genauigkeitsmaße:
|
||||
return float(s0apost)
|
||||
|
||||
|
||||
|
||||
@staticmethod
|
||||
def helmert_punktfehler(Qxx, s0_apost, unbekannten_liste, dim=3):
|
||||
diagQ = np.diag(Qxx)
|
||||
daten = []
|
||||
namen_str = [str(sym) for sym in unbekannten_liste]
|
||||
|
||||
n_punkte = len(unbekannten_liste) // 3
|
||||
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[3 * 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())
|
||||
|
||||
qx = diagQ[3 * i]
|
||||
qy = diagQ[3 * i + 1]
|
||||
qz = diagQ[3 * i + 2]
|
||||
qx = diagQ[idx_x]
|
||||
qy = diagQ[idx_y]
|
||||
qz = 0.0
|
||||
|
||||
if dim == 3:
|
||||
try:
|
||||
idx_z = next(i for i, n in enumerate(namen_str) if n.upper() == f"Z{pid}".upper())
|
||||
qz = diagQ[idx_z]
|
||||
except StopIteration:
|
||||
qz = 0.0
|
||||
|
||||
sx = s0_apost * np.sqrt(qx)
|
||||
sy = s0_apost * np.sqrt(qy)
|
||||
sz = s0_apost * np.sqrt(qz)
|
||||
sz = s0_apost * np.sqrt(qz) if dim == 3 else 0
|
||||
|
||||
if dim == 2:
|
||||
sP = s0_apost * np.sqrt(qx + qy)
|
||||
else:
|
||||
sP = s0_apost * np.sqrt(qx + qy + qz)
|
||||
|
||||
daten.append([
|
||||
punkt,
|
||||
float(sx),
|
||||
float(sy),
|
||||
float(sz),
|
||||
float(sP)
|
||||
])
|
||||
daten.append([pid, float(sx), float(sy), float(sz), float(sP)])
|
||||
except:
|
||||
continue
|
||||
|
||||
helmert_punktfehler = pd.DataFrame(daten, columns=["Punkt", "σx", "σy", "σz", f"σP_{dim}D"])
|
||||
return helmert_punktfehler
|
||||
|
||||
@@ -257,3 +263,172 @@ def plot_netz_komplett_final(x_vektor, unbekannten_labels, beobachtungs_labels,
|
||||
bgcolor="white", bordercolor="black", borderwidth=1)
|
||||
|
||||
fig.show(config={'scrollZoom': True})
|
||||
|
||||
|
||||
|
||||
def plot_netz_final_mit_df_ellipsen(x_vektor, unbekannten_labels, beobachtungs_labels, df_ellipsen, v_faktor=1000):
|
||||
# 1. Punkte extrahieren
|
||||
coords = {}
|
||||
# Wir nehmen an, dass die Reihenfolge im x_vektor X, Y, Z pro Punkt ist
|
||||
punkt_ids = sorted(set(str(l)[1:] for l in unbekannten_labels if str(l).startswith(('X', 'Y', 'Z'))))
|
||||
|
||||
for pid in punkt_ids:
|
||||
try:
|
||||
ix = next(i for i, s in enumerate(unbekannten_labels) if str(s) == f"X{pid}")
|
||||
iy = next(i for i, s in enumerate(unbekannten_labels) if str(s) == f"Y{pid}")
|
||||
coords[pid] = (float(x_vektor[ix]), float(x_vektor[iy]))
|
||||
except:
|
||||
continue
|
||||
|
||||
fig = go.Figure()
|
||||
|
||||
# 2. Beobachtungslinien (Gruppiert)
|
||||
beob_typen = {
|
||||
'GNSS-Basislinien': {'pattern': 'gnss', 'color': 'rgba(255, 100, 0, 0.4)'},
|
||||
'Nivellement': {'pattern': 'niv', 'color': 'rgba(0, 200, 100, 0.4)'},
|
||||
'Tachymeter': {'pattern': '', 'color': 'rgba(100, 100, 100, 0.3)'}
|
||||
}
|
||||
|
||||
for typ, info in beob_typen.items():
|
||||
x_l, y_l = [], []
|
||||
for bl in beobachtungs_labels:
|
||||
bl_str = str(bl).lower()
|
||||
# Einfache Logik zur Typtrennung
|
||||
if (info['pattern'] in bl_str and info['pattern'] != '') or \
|
||||
(info['pattern'] == '' and 'gnss' not in bl_str and 'niv' not in bl_str):
|
||||
pts = [pid for pid in coords if f"_{pid}" in str(bl) or str(bl).startswith(f"{pid}_")]
|
||||
if len(pts) >= 2:
|
||||
x_l.extend([coords[pts[0]][0], coords[pts[1]][0], None])
|
||||
y_l.extend([coords[pts[0]][1], coords[pts[1]][1], None])
|
||||
|
||||
if x_l:
|
||||
fig.add_trace(go.Scatter(x=x_l, y=y_l, mode='lines', name=typ, line=dict(color=info['color'], width=1)))
|
||||
|
||||
# 3. Ellipsen aus dem DataFrame zeichnen
|
||||
for _, row in df_ellipsen.iterrows():
|
||||
pid = str(row['Punkt'])
|
||||
if pid in coords:
|
||||
x0, y0 = coords[pid]
|
||||
|
||||
# Werte aus DF (mit v_faktor skalieren)
|
||||
a = row['a_K'] * v_faktor
|
||||
b = row['b_K'] * v_faktor
|
||||
theta_gon = row['θ [gon]']
|
||||
|
||||
# Umrechnung: gon -> rad für die Rotation
|
||||
# Da im Plot X horizontal und Y vertikal ist, entspricht theta_gon dem Winkel zur X-Achse
|
||||
theta_rad = theta_gon * (np.pi / 200.0)
|
||||
|
||||
# Ellipsen berechnen
|
||||
t = np.linspace(0, 2 * np.pi, 50)
|
||||
e_x = a * np.cos(t)
|
||||
e_y = b * np.sin(t)
|
||||
|
||||
# Ausrichtung der Ellipsen
|
||||
R = np.array([[np.cos(theta_rad), -np.sin(theta_rad)],
|
||||
[np.sin(theta_rad), np.cos(theta_rad)]])
|
||||
|
||||
rot = np.dot(R, np.array([e_x, e_y]))
|
||||
|
||||
fig.add_trace(go.Scatter(
|
||||
x=rot[0, :] + x0, y=rot[1, :] + y0,
|
||||
mode='lines', line=dict(color='red', width=1.5),
|
||||
name='Konfidenzellipsen',
|
||||
legendgroup='Ellipsen',
|
||||
showlegend=(pid == df_ellipsen.iloc[0]['Punkt']),
|
||||
hoverinfo='text',
|
||||
text=f"Punkt {pid}<br>a_K: {row['a_K']:.4f}m<br>b_K: {row['b_K']:.4f}m"
|
||||
))
|
||||
|
||||
# Punkte plotten
|
||||
df_pts = pd.DataFrame([(pid, c[0], c[1]) for pid, c in coords.items()], columns=['ID', 'X', 'Y'])
|
||||
fig.add_trace(go.Scatter(
|
||||
x=df_pts['X'], y=df_pts['Y'], mode='markers+text',
|
||||
text=df_pts['ID'], textposition="top center",
|
||||
marker=dict(size=8, color='black'), name="Netzpunkte"))
|
||||
|
||||
# Layout
|
||||
fig.update_layout(
|
||||
title=f"Netzplot (Ellipsen {v_faktor}x überhöht)",
|
||||
xaxis=dict(title="X [m]", tickformat="f", separatethousands=True, scaleanchor="y", scaleratio=1,
|
||||
showgrid=True, gridcolor='lightgrey'),
|
||||
yaxis=dict(title="Y [m]", tickformat="f", separatethousands=True, showgrid=True, gridcolor='lightgrey'),
|
||||
width=1100, height=900,
|
||||
plot_bgcolor='white')
|
||||
|
||||
# Maßstabsangabe
|
||||
fig.add_annotation(
|
||||
text=f"<b>Skalierung:</b><br>Ellipsengröße im Plot = {v_faktor} × Realität",
|
||||
align='left', showarrow=False, xref='paper', yref='paper', x=0.02, y=0.02,
|
||||
bgcolor="rgba(255,255,255,0.8)", bordercolor="black", borderwidth=1)
|
||||
|
||||
fig.show(config={'scrollZoom': True})
|
||||
|
||||
import plotly.graph_objects as go
|
||||
import numpy as np
|
||||
|
||||
|
||||
|
||||
def plot_netz_3D(x_vektor, unbekannten_labels, beobachtungs_labels, df_ellipsen, v_faktor=1000):
|
||||
"""
|
||||
Erzeugt einen interaktiven 3D-Plot des Netzes.
|
||||
- v_faktor: Vergrößerung der Genauigkeits-Achsen (z.B. 1000 für mm -> m)
|
||||
"""
|
||||
# 1. Punkte extrahieren
|
||||
pts = {}
|
||||
punkt_ids = sorted(set(str(l)[1:] for l in unbekannten_labels if str(l).startswith(('X', 'Y', 'Z'))))
|
||||
|
||||
for pid in punkt_ids:
|
||||
try:
|
||||
ix = next(i for i, s in enumerate(unbekannten_labels) if str(s) == f"X{pid}")
|
||||
iy = next(i for i, s in enumerate(unbekannten_labels) if str(s) == f"Y{pid}")
|
||||
iz = next(i for i, s in enumerate(unbekannten_labels) if str(s) == f"Z{pid}")
|
||||
pts[pid] = (float(x_vektor[ix]), float(x_vektor[iy]), float(x_vektor[iz]))
|
||||
except:
|
||||
continue
|
||||
|
||||
fig = go.Figure()
|
||||
|
||||
# 2. Beobachtungen (Linien im Raum)
|
||||
# Wir zeichnen hier einfach alle Verbindungen
|
||||
x_line, y_line, z_line = [], [], []
|
||||
for bl in beobachtungs_labels:
|
||||
p_in_l = [pid for pid in pts if f"_{pid}" in str(bl) or str(bl).startswith(f"{pid}_")]
|
||||
if len(p_in_l) >= 2:
|
||||
p1, p2 = pts[p_in_l[0]], pts[p_in_l[1]]
|
||||
x_line.extend([p1[0], p2[0], None])
|
||||
y_line.extend([p1[1], p2[1], None])
|
||||
z_line.extend([p1[2], p2[2], None])
|
||||
|
||||
fig.add_trace(go.Scatter3d(
|
||||
x=x_line, y=y_line, z=z_line,
|
||||
mode='lines', line=dict(color='gray', width=2),
|
||||
name='Beobachtungen'
|
||||
))
|
||||
|
||||
# 3. Punkte & "Fehler-Kreuze" (als Ersatz für Ellipsoide)
|
||||
# Ein echtes 3D-Ellipsoid ist grafisch schwer, daher zeichnen wir 3 Achsen
|
||||
for pid, coord in pts.items():
|
||||
# Hier könnten wir die echten Halbachsen aus der 3D-Eigenwertanalyse nutzen
|
||||
# Für den Anfang plotten wir die Standardabweichungen sX, sY, sZ als Kreuz
|
||||
fig.add_trace(go.Scatter3d(
|
||||
x=[coord[0]], y=[coord[1]], z=[coord[2]],
|
||||
mode='markers+text', text=[pid],
|
||||
marker=dict(size=4, color='black'), name=f'Punkt {pid}'
|
||||
))
|
||||
|
||||
# 4. Layout
|
||||
fig.update_layout(
|
||||
scene=dict(
|
||||
xaxis_title='X [m]',
|
||||
yaxis_title='Y [m]',
|
||||
zaxis_title='Z [m]',
|
||||
aspectmode='data' # WICHTIG: Verhältnisse 1:1:1 bewahren
|
||||
),
|
||||
width=1000, height=800,
|
||||
title="Geozentrisches Netz in 3D"
|
||||
)
|
||||
|
||||
fig.show()
|
||||
|
||||
# Aufruf
|
||||
|
||||
@@ -18,7 +18,7 @@ def ausgleichung_global(A, dl, Q_ext):
|
||||
N = A.T @ P @ A
|
||||
n = A.T @ P @ dl
|
||||
|
||||
# 3) Zuschlagsvektor dx und Unbekanntenvektor x
|
||||
# 3) Zuschlagsvektor dx
|
||||
dx = np.linalg.solve(N, n)
|
||||
|
||||
# 4) Residuenvektor v
|
||||
@@ -47,20 +47,10 @@ def ausgleichung_global(A, dl, Q_ext):
|
||||
return dict_ausgleichung, dx
|
||||
|
||||
|
||||
def ausgleichung_lokal(
|
||||
A,
|
||||
dl,
|
||||
Q_ll,
|
||||
x0,
|
||||
liste_punktnummern,
|
||||
auswahl,
|
||||
mit_massstab: bool = True,
|
||||
):
|
||||
|
||||
def ausgleichung_lokal(A, dl, Q_ll):
|
||||
A = np.asarray(A, dtype=float)
|
||||
dl = np.asarray(dl, dtype=float).reshape(-1, 1)
|
||||
Q_ll = np.asarray(Q_ll, dtype=float)
|
||||
x0 = np.asarray(x0, dtype=float).reshape(-1, 1)
|
||||
|
||||
# 1) Gewichtsmatrix
|
||||
P = np.linalg.inv(Q_ll)
|
||||
@@ -69,73 +59,37 @@ def ausgleichung_lokal(
|
||||
N = A.T @ P @ A
|
||||
n = A.T @ P @ dl
|
||||
|
||||
# 3) Datum: G, E, Gi
|
||||
# -> Datumsfestlegung ist sympy-basiert, daher nur dafür kurz Sympy verwenden
|
||||
x0_sp = sp.Matrix(x0)
|
||||
# 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)
|
||||
|
||||
Gi = np.asarray(Gi_sp, dtype=float) # zurück nach numpy
|
||||
|
||||
# 4) gerändertes System lösen:
|
||||
# [N Gi] [dx] = [n]
|
||||
# [GiT 0] [k ] [0]
|
||||
u = N.shape[0]
|
||||
d = Gi.shape[1]
|
||||
|
||||
K = np.block([
|
||||
[N, Gi],
|
||||
[Gi.T, np.zeros((d, d))]
|
||||
])
|
||||
rhs = np.vstack([n, np.zeros((d, 1))])
|
||||
|
||||
sol = np.linalg.solve(K, rhs)
|
||||
dx = sol[:u, :]
|
||||
# 3) Zuschlagsvektor dx
|
||||
dx = np.linalg.solve(N, n)
|
||||
|
||||
# 5) Residuen
|
||||
v = dl - A @ dx
|
||||
|
||||
# 6) Qxx (innere Lösung)
|
||||
N_inv = np.linalg.inv(N)
|
||||
N_inv_G = N_inv @ Gi
|
||||
S = Gi.T @ N_inv_G
|
||||
print("rank(Gi) =", np.linalg.matrix_rank(Gi))
|
||||
print("Gi shape =", Gi.shape)
|
||||
print("rank(S) =", np.linalg.matrix_rank(S))
|
||||
print("S shape =", S.shape)
|
||||
S_inv = np.linalg.inv(S)
|
||||
Q_xx = N_inv - N_inv_G @ S_inv @ N_inv_G.T
|
||||
# 5) Kofaktormatrix der Unbekannten Q_xx
|
||||
Q_xx = StochastischesModell.berechne_Q_xx(N)
|
||||
|
||||
# 7) Q_lhat_lhat, Q_vv
|
||||
Q_lhat_lhat = A @ Q_xx @ A.T
|
||||
Q_vv = np.linalg.inv(P) - Q_lhat_lhat
|
||||
# 6) Kofaktormatrix der Beobachtungen Q_ll_dach
|
||||
Q_ll_dach = StochastischesModell.berechne_Q_ll_dach(A, Q_xx)
|
||||
|
||||
# 8) Redundanz
|
||||
R = Q_vv @ P
|
||||
r_vec = np.diag(R).reshape(-1, 1)
|
||||
n_beob = A.shape[0]
|
||||
u = A.shape[1]
|
||||
d = Gi.shape[1]
|
||||
r_gesamt = n_beob - u + d
|
||||
# 7) Kofaktormatrix der Verbesserungen Q_vv
|
||||
Q_vv = StochastischesModell.berechne_Qvv(Q_ll, Q_ll_dach)
|
||||
|
||||
# 9) sigma0
|
||||
vv = float(v.T @ P @ v)
|
||||
sigma0_apost = float(np.sqrt(vv / r_gesamt))
|
||||
|
||||
return {
|
||||
# 8) Ausgabe
|
||||
dict_ausgleichung = {
|
||||
"dx": dx,
|
||||
"v": v,
|
||||
"P": P,
|
||||
"N": N,
|
||||
"Q_xx": Q_xx,
|
||||
"Q_lhat_lhat": Q_lhat_lhat,
|
||||
"Q_ll_dach": Q_ll_dach,
|
||||
"Q_vv": Q_vv,
|
||||
"R": R,
|
||||
"r": r_vec,
|
||||
"r_gesamt": r_gesamt,
|
||||
"sigma0_apost": sigma0_apost,
|
||||
"G": np.asarray(G, dtype=float),
|
||||
"Gi": Gi,
|
||||
}, dx
|
||||
"Q_ll": Q_ll,
|
||||
}
|
||||
return dict_ausgleichung, dx
|
||||
28
Proben.py
Normal file
28
Proben.py
Normal file
@@ -0,0 +1,28 @@
|
||||
import numpy as np
|
||||
# d
|
||||
def atpv_probe(A, P, v, tol=1e-7):
|
||||
A = np.asarray(A, float)
|
||||
P = np.asarray(P, float)
|
||||
v = np.asarray(v, float).reshape(-1, 1)
|
||||
|
||||
ATPV = A.T @ P @ v
|
||||
|
||||
if np.allclose(ATPV, 0, atol=tol):
|
||||
print("ATPv-Probe erfolgreich")
|
||||
else:
|
||||
print("ATPv-Probe nicht erfolgreich. Fehler bei der Lösung des Normalgleichungssystems")
|
||||
|
||||
|
||||
def hauptprobe(A, x, l, v, tol=1e-7):
|
||||
A = np.asarray(A, float)
|
||||
x = np.asarray(x, float).reshape(-1, 1)
|
||||
l = np.asarray(l, float).reshape(-1, 1)
|
||||
v = np.asarray(v, float).reshape(-1, 1)
|
||||
|
||||
v_test = A @ x - l
|
||||
|
||||
if np.allclose(v, v_test, atol=tol):
|
||||
print("Hauptprobe erfolgreich")
|
||||
else:
|
||||
diff = v - v_test
|
||||
print("Hauptprobe nicht erfolgreich. Abweichung zu v: ", diff)
|
||||
@@ -1,6 +1,361 @@
|
||||
i = 5
|
||||
import csv
|
||||
import xml.etree.ElementTree as ET
|
||||
from decimal import Decimal, getcontext, ROUND_HALF_UP
|
||||
|
||||
if i % 6 != 0:
|
||||
print("nein")
|
||||
else:
|
||||
print("ja")
|
||||
# ========= Einstellungen =========
|
||||
JXL_IN = r"C:\Users\fabia\Desktop\Masterprojekt_V3\Daten\campusnetz_bereinigt_plus_nachmessung.jxl"
|
||||
CSV_IN = r"C:\Users\fabia\Desktop\Masterprojekt_V3\Daten\campsnetz_beobachtungen_plus_nachmessungen.csv"
|
||||
CSV_OUT = r"C:\Users\fabia\Desktop\Masterprojekt_V3\Daten\campsnetz_beobachtungen_plus_nachmessungen_korrigiert.csv"
|
||||
|
||||
getcontext().prec = 70
|
||||
|
||||
|
||||
# ========= Hilfsfunktionen =========
|
||||
def count_decimals(s: str, sep: str) -> int:
|
||||
if s is None:
|
||||
return 0
|
||||
s = s.strip()
|
||||
if s == "":
|
||||
return 0
|
||||
if ":ZH:" in s:
|
||||
s = s.split(":ZH:", 1)[0].strip()
|
||||
if sep in s:
|
||||
return len(s.split(sep, 1)[1])
|
||||
return 0
|
||||
|
||||
|
||||
def fmt_decimal_fixed(x: Decimal, decimals: int, sep: str) -> str:
|
||||
q = Decimal("1") if decimals == 0 else Decimal("1." + ("0" * decimals))
|
||||
y = x.quantize(q, rounding=ROUND_HALF_UP)
|
||||
txt = format(y, "f")
|
||||
if sep != ".":
|
||||
txt = txt.replace(".", sep)
|
||||
if decimals == 0:
|
||||
txt = txt.split(sep)[0]
|
||||
return txt
|
||||
|
||||
|
||||
def parse_decimal_csv(s: str) -> Decimal:
|
||||
"""
|
||||
CSV-Zahlen mit Komma, evtl. mit ":ZH:..." im letzten Feld.
|
||||
"""
|
||||
s = (s or "").strip()
|
||||
if ":ZH:" in s:
|
||||
s = s.split(":ZH:", 1)[0].strip()
|
||||
s = s.replace(",", ".")
|
||||
return Decimal(s)
|
||||
|
||||
|
||||
def parse_decimal_comma(s: str) -> Decimal:
|
||||
"""
|
||||
Komma-String nach Decimal.
|
||||
"""
|
||||
return Decimal((s or "").strip().replace(",", "."))
|
||||
|
||||
|
||||
def deg_to_gon_str(deg_str: str) -> str:
|
||||
"""
|
||||
JXL: Winkel in Grad (Dezimalpunkt).
|
||||
gon = deg * (10/9)
|
||||
Ausgabe mit exakt so vielen Nachkommastellen wie im JXL-Gradwert enthalten.
|
||||
Dezimaltrennzeichen: Komma.
|
||||
"""
|
||||
deg_str = (deg_str or "").strip()
|
||||
d = count_decimals(deg_str, ".")
|
||||
deg = Decimal(deg_str)
|
||||
gon = deg * (Decimal(10) / Decimal(9))
|
||||
return fmt_decimal_fixed(gon, d, ",")
|
||||
|
||||
|
||||
def meter_str_from_jxl(m_str: str) -> str:
|
||||
"""
|
||||
JXL: Distanz in Meter (Dezimalpunkt).
|
||||
Ausgabe mit exakt so vielen Nachkommastellen wie in der JXL enthalten.
|
||||
Dezimaltrennzeichen: Komma.
|
||||
"""
|
||||
m_str = (m_str or "").strip()
|
||||
d = count_decimals(m_str, ".")
|
||||
return fmt_decimal_fixed(Decimal(m_str), d, ",")
|
||||
|
||||
|
||||
def is_obs_line(row: list[str]) -> bool:
|
||||
"""
|
||||
Beobachtungszeile: Zielpunkt nicht leer, Hz/Z/SD numerisch parsebar.
|
||||
Zielpunkt darf alphanumerisch sein (FH3 etc.).
|
||||
"""
|
||||
if len(row) < 4:
|
||||
return False
|
||||
if row[0].strip() == "" or row[1].strip() == "" or row[2].strip() == "" or row[3].strip() == "":
|
||||
return False
|
||||
try:
|
||||
_ = parse_decimal_csv(row[1])
|
||||
_ = parse_decimal_csv(row[2])
|
||||
_ = parse_decimal_csv(row[3])
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def is_station_candidate(row: list[str]) -> bool:
|
||||
"""
|
||||
Kandidat für Standpunkt: erste Spalte nicht leer, Messspalten leer.
|
||||
Ob es wirklich ein Standpunkt ist, entscheiden wir später über StationName-Menge.
|
||||
"""
|
||||
if len(row) < 4:
|
||||
return False
|
||||
return (
|
||||
row[0].strip() != ""
|
||||
and row[1].strip() == ""
|
||||
and row[2].strip() == ""
|
||||
and row[3].strip() == ""
|
||||
)
|
||||
|
||||
|
||||
def csv_is_rounding_of_jxl(csv_str: str, jxl_full_str: str) -> bool:
|
||||
"""
|
||||
Prüft: CSV ist gerundete Darstellung des JXL-Wertes.
|
||||
Kriterium:
|
||||
- CSV hat weniger Nachkommastellen als JXL
|
||||
- und: JXL auf CSV-Dezimalstellen gerundet == CSV-Wert (numerisch)
|
||||
"""
|
||||
dc = count_decimals(csv_str, ",")
|
||||
dj = count_decimals(jxl_full_str, ",")
|
||||
if dc >= dj:
|
||||
return False
|
||||
|
||||
try:
|
||||
csv_val = parse_decimal_csv(csv_str)
|
||||
jxl_val = parse_decimal_comma(jxl_full_str)
|
||||
|
||||
q = Decimal("1") if dc == 0 else Decimal("1." + ("0" * dc))
|
||||
jxl_rounded = jxl_val.quantize(q, rounding=ROUND_HALF_UP)
|
||||
csv_q = csv_val.quantize(q, rounding=ROUND_HALF_UP)
|
||||
|
||||
return jxl_rounded == csv_q
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
# ========= JXL einlesen =========
|
||||
tree = ET.parse(JXL_IN)
|
||||
root = tree.getroot()
|
||||
|
||||
# StationRecords: (StationName, StationID, IH)
|
||||
station_records = []
|
||||
for sr in root.iter("StationRecord"):
|
||||
sname = (sr.findtext("StationName") or "").strip()
|
||||
sid = (sr.attrib.get("ID") or "").strip()
|
||||
ih = (sr.findtext("TheodoliteHeight") or "").strip()
|
||||
if sname != "" and sid != "":
|
||||
station_records.append((sname, sid, ih))
|
||||
|
||||
station_names_set = {sname for sname, _, _ in station_records}
|
||||
|
||||
# pro StationName ggf. mehrere Aufbauten -> "nächsten unbenutzten" nehmen
|
||||
stationname_to_records = {}
|
||||
for sname, sid, ih in station_records:
|
||||
stationname_to_records.setdefault(sname, []).append((sid, ih))
|
||||
stationname_usecount = {k: 0 for k in stationname_to_records.keys()}
|
||||
|
||||
# TargetHeight je TargetRecord-ID
|
||||
target_height_by_id = {}
|
||||
for tr in root.iter("TargetRecord"):
|
||||
tid = (tr.attrib.get("ID") or "").strip()
|
||||
zh = (tr.findtext("TargetHeight") or "").strip()
|
||||
if tid != "":
|
||||
target_height_by_id[tid] = zh
|
||||
|
||||
# Pro StationID: Sequenz der PointRecords
|
||||
station_seq = {sid: [] for _, sid, _ in station_records}
|
||||
|
||||
for pr in root.iter("PointRecord"):
|
||||
stid = (pr.findtext("StationID") or "").strip()
|
||||
if stid == "" or stid not in station_seq:
|
||||
continue
|
||||
|
||||
circle = pr.find("Circle")
|
||||
if circle is None:
|
||||
continue
|
||||
|
||||
target_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 target_name == "" or hz_deg == "" or z_deg == "" or sd_m == "":
|
||||
continue
|
||||
|
||||
station_seq[stid].append({
|
||||
"target": target_name,
|
||||
"hz_gon": deg_to_gon_str(hz_deg),
|
||||
"z_gon": deg_to_gon_str(z_deg),
|
||||
"sd_m": meter_str_from_jxl(sd_m),
|
||||
"zh": target_height_by_id.get(target_id, ""),
|
||||
})
|
||||
|
||||
|
||||
# ========= Matching-Funktion =========
|
||||
def pick_jxl_entry_for_obs(seq, start_ptr, zp, hz_csv, z_csv, sd_csv, search_window=200):
|
||||
"""
|
||||
Standard: nimmt seq[start_ptr]
|
||||
Wenn target nicht passt: sucht im Fenster nach passendem zp.
|
||||
Bei Mehrfachtreffern wird bevorzugt, wo gerundete Werte passen.
|
||||
"""
|
||||
if start_ptr >= len(seq):
|
||||
return None, start_ptr
|
||||
|
||||
first = seq[start_ptr]
|
||||
if first["target"] == zp:
|
||||
return first, start_ptr + 1
|
||||
|
||||
end = min(len(seq), start_ptr + search_window)
|
||||
candidates = []
|
||||
for i in range(start_ptr, end):
|
||||
if seq[i]["target"] == zp:
|
||||
candidates.append((i, seq[i]))
|
||||
|
||||
if not candidates:
|
||||
return first, start_ptr + 1
|
||||
|
||||
if len(candidates) == 1:
|
||||
i, entry = candidates[0]
|
||||
return entry, i + 1
|
||||
|
||||
good = []
|
||||
for i, entry in candidates:
|
||||
ok_hz = csv_is_rounding_of_jxl(hz_csv, entry["hz_gon"])
|
||||
ok_z = csv_is_rounding_of_jxl(z_csv, entry["z_gon"])
|
||||
ok_sd = csv_is_rounding_of_jxl(sd_csv, entry["sd_m"])
|
||||
score = int(ok_hz) + int(ok_z) + int(ok_sd)
|
||||
good.append((score, i, entry))
|
||||
|
||||
good.sort(key=lambda t: (-t[0], t[1]))
|
||||
_, i_best, entry_best = good[0]
|
||||
return entry_best, i_best + 1
|
||||
|
||||
|
||||
# ========= CSV verarbeiten =========
|
||||
repl_counts = {"Hz": 0, "Z": 0, "SD": 0}
|
||||
current_station_id = None
|
||||
current_station_ptr = 0
|
||||
|
||||
line_no = 0
|
||||
|
||||
fehlende_IH = [] # (zeilennummer, standpunkt)
|
||||
fehlende_ZH = [] # (zeilennummer, standpunkt, zielpunkt)
|
||||
fehlender_StationRecord = [] # (zeilennummer, standpunkt_text)
|
||||
|
||||
current_station_name = None
|
||||
|
||||
with open(CSV_IN, newline="", encoding="utf-8") as fin, open(CSV_OUT, "w", newline="", encoding="utf-8") as fout:
|
||||
reader = csv.reader(fin, delimiter=";")
|
||||
writer = csv.writer(fout, delimiter=";", lineterminator="\n")
|
||||
|
||||
for row in reader:
|
||||
line_no += 1
|
||||
|
||||
if len(row) < 4:
|
||||
row = row + [""] * (4 - len(row))
|
||||
|
||||
# ---- Standpunkt-Kandidat? ----
|
||||
if is_station_candidate(row):
|
||||
sp = row[0].strip()
|
||||
|
||||
# Nur als Standpunkt behandeln, wenn er wirklich in der JXL als StationName existiert:
|
||||
if sp in station_names_set:
|
||||
use = stationname_usecount.get(sp, 0)
|
||||
recs = stationname_to_records[sp]
|
||||
if use >= len(recs):
|
||||
raise RuntimeError(f"Standpunkt {sp} kommt in CSV öfter vor als in der JXL (StationRecords).")
|
||||
|
||||
sid, ih = recs[use]
|
||||
stationname_usecount[sp] = use + 1
|
||||
|
||||
current_station_name = sp
|
||||
current_station_id = sid
|
||||
current_station_ptr = 0
|
||||
|
||||
# fehlende IH loggen
|
||||
if ih is None or str(ih).strip() == "":
|
||||
fehlende_IH.append((line_no, sp))
|
||||
|
||||
writer.writerow([sp, f"IH:{ih}", "", "", ""])
|
||||
continue
|
||||
|
||||
# NICHT in JXL: wenn es wie ein Standpunkt aussieht -> loggen
|
||||
if sp.isdigit():
|
||||
fehlender_StationRecord.append((line_no, sp))
|
||||
|
||||
writer.writerow(row)
|
||||
continue
|
||||
|
||||
# ---- Beobachtung? ----
|
||||
if is_obs_line(row) and current_station_id is not None:
|
||||
zp = row[0].strip()
|
||||
hz_csv = row[1].strip()
|
||||
z_csv = row[2].strip()
|
||||
sd_csv = row[3].strip()
|
||||
|
||||
seq = station_seq.get(current_station_id, [])
|
||||
jxl_entry, new_ptr = pick_jxl_entry_for_obs(seq, current_station_ptr, zp, hz_csv, z_csv, sd_csv)
|
||||
|
||||
if jxl_entry is None:
|
||||
writer.writerow(row)
|
||||
continue
|
||||
|
||||
current_station_ptr = new_ptr
|
||||
|
||||
hz_out = hz_csv
|
||||
z_out = z_csv
|
||||
sd_out = sd_csv
|
||||
|
||||
if csv_is_rounding_of_jxl(hz_csv, jxl_entry["hz_gon"]):
|
||||
hz_out = jxl_entry["hz_gon"]
|
||||
repl_counts["Hz"] += 1
|
||||
|
||||
if csv_is_rounding_of_jxl(z_csv, jxl_entry["z_gon"]):
|
||||
z_out = jxl_entry["z_gon"]
|
||||
repl_counts["Z"] += 1
|
||||
|
||||
if csv_is_rounding_of_jxl(sd_csv, jxl_entry["sd_m"]):
|
||||
sd_out = jxl_entry["sd_m"]
|
||||
repl_counts["SD"] += 1
|
||||
|
||||
# fehlende ZH loggen
|
||||
zh_val = jxl_entry.get("zh", "")
|
||||
if zh_val is None or str(zh_val).strip() == "":
|
||||
fehlende_ZH.append((line_no, current_station_name, zp))
|
||||
|
||||
last_col = f"{sd_out}:ZH:{zh_val}" if str(zh_val).strip() != "" else sd_out
|
||||
writer.writerow([zp, hz_out, z_out, last_col])
|
||||
continue
|
||||
|
||||
# ---- alles andere unverändert ----
|
||||
writer.writerow(row)
|
||||
|
||||
print("Fertig.")
|
||||
print("Ausgabe:", CSV_OUT)
|
||||
print("Ersetzungen (Rundung -> JXL volle Nachkommastellen):", repl_counts)
|
||||
|
||||
print("\n--- Fehlende IH ---")
|
||||
print("Anzahl:", len(fehlende_IH))
|
||||
for z, sp in fehlende_IH[:50]:
|
||||
print(f"Zeile {z}: Standpunkt {sp} (IH leer in JXL)")
|
||||
if len(fehlende_IH) > 50:
|
||||
print("... (weitere gekürzt)")
|
||||
|
||||
print("\n--- Fehlende ZH ---")
|
||||
print("Anzahl:", len(fehlende_ZH))
|
||||
for z, sp, zp in fehlende_ZH[:50]:
|
||||
print(f"Zeile {z}: Standpunkt {sp}, Ziel {zp} (ZH nicht ermittelt)")
|
||||
if len(fehlende_ZH) > 50:
|
||||
print("... (weitere gekürzt)")
|
||||
|
||||
print("\n--- Standpunkt in CSV, aber kein StationRecord in JXL ---")
|
||||
print("Anzahl:", len(fehlender_StationRecord))
|
||||
for z, sp in fehlender_StationRecord[:50]:
|
||||
print(f"Zeile {z}: Standpunkt {sp} (nicht in JXL als StationName gefunden)")
|
||||
if len(fehlender_StationRecord) > 50:
|
||||
print("... (weitere gekürzt)")
|
||||
|
||||
Reference in New Issue
Block a user