Teleskop Control Programm für die Seestar Smart Teleskope
#!/usr/bin/env python3
import csv
import math
import os
import platform
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time
import tkinter as tk
import urllib.parse
import urllib.request
import webbrowser
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from tkinter import ttk, filedialog, messagebox, colorchooser
import numpy as np
try:
from PIL import Image, ImageDraw, ImageFont, ImageTk, ImageColor
except ImportError:
sys.exit("Pillow fehlt: pip install pillow")
try:
import cv2
except ImportError:
cv2 = None
try:
from seestarpy import stream as ss_stream
from seestarpy import connection as ss_conn
from seestarpy import auth as ss_auth
except ImportError:
ss_stream = None
ss_conn = None
ss_auth = None
APP_TITLE = "Vlio Control"
VIEW_TITLE = "Vlio Live"
def find_key():
"""
Sucht den RSA-Schluessel in derselben Reihenfolge wie seestarpy:
1. Umgebungsvariable SEESTAR_KEY_PATH
2. seestar.pem im aktuellen Verzeichnis
3. ~/.seestarpy/seestar.pem
Gibt (pfad, quelle) zurueck oder (None, None).
"""
env = os.environ.get("SEESTAR_KEY_PATH")
if env and Path(env).is_file():
return Path(env), "SEESTAR_KEY_PATH"
cwd = Path.cwd() / "seestar.pem"
if cwd.is_file():
return cwd, "Arbeitsverzeichnis"
home = Path.home() / ".seestarpy" / "seestar.pem"
if home.is_file():
return home, "Benutzerordner"
return None, None
def default_key_path():
return Path.home() / ".seestarpy" / "seestar.pem"
def valid_ip(text):
"""Grobe Pruefung: vier Zahlenbloecke 0-255, nichts Halbfertiges."""
parts = text.split(".")
if len(parts) != 4:
return False
try:
return all(0 <= int(p) <= 255 for p in parts)
except ValueError:
return False
def extract_key_from_apk(apk_path, target=None):
"""
Ruft den Extraktor von seestarpy auf. Gibt (erfolg, ausgabe) zurueck.
"""
cmd = [sys.executable, "-m", "seestarpy.extract_pem", str(apk_path)]
if target:
cmd += ["-o", str(target)]
try:
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
except FileNotFoundError:
return False, "Python-Interpreter nicht gefunden."
except subprocess.TimeoutExpired:
return False, "Zeitueberschreitung beim Entpacken der APK."
out = (proc.stdout or "") + (proc.stderr or "")
return proc.returncode == 0, out.strip()
def _hms_to_deg(h, m):
return (h + m / 60.0) * 15.0
def _dms_to_deg(sign, d, m):
return sign * (d + m / 60.0)
def _parse_catalog_line(line):
"""Zeilenformat: NAME|NGC|Typ|hh mm.m|+dd mm|Eigenname"""
parts = [p.strip() for p in line.split("|")]
name, ngc, typ, ra_txt, dec_txt, common = (parts + [""] * 6)[:6]
rh, rm = ra_txt.split()
ra = _hms_to_deg(float(rh), float(rm))
sign = -1.0 if dec_txt.strip().startswith("-") else 1.0
dd, dm = dec_txt.replace("+", "").replace("-", "").split()
dec = _dms_to_deg(sign, float(dd), float(dm))
return {
"name": name,
"ngc": ngc if ngc and ngc != "-" else "",
"type": typ,
"common": common,
"ra": ra,
"dec": dec,
}
_MESSIER_RAW = """
M1|1952|SNR|05 34.5|+22 01|Krebsnebel
M2|7089|GCl|21 33.5|-00 49|
M3|5272|GCl|13 42.2|+28 23|
M4|6121|GCl|16 23.6|-26 32|
M5|5904|GCl|15 18.6|+02 05|
M6|6405|OpCl|17 40.1|-32 13|Schmetterlingshaufen
M7|6475|OpCl|17 53.9|-34 49|
M8|6523|Neb|18 03.8|-24 23|Lagunennebel
M9|6333|GCl|17 19.2|-18 31|
M10|6254|GCl|16 57.1|-04 06|
M11|6705|OpCl|18 51.1|-06 16|Wildentenhaufen
M12|6218|GCl|16 47.2|-01 57|
M13|6205|GCl|16 41.7|+36 28|Herkuleshaufen
M14|6402|GCl|17 37.6|-03 15|
M15|7078|GCl|21 30.0|+12 10|
M16|6611|OpCl|18 18.8|-13 47|Adlernebel
M17|6618|Neb|18 20.8|-16 11|Omega-/Schwanennebel
M18|6613|OpCl|18 19.9|-17 08|
M19|6273|GCl|17 02.6|-26 16|
M20|6514|Neb|18 02.6|-23 02|Trifidnebel
M21|6531|OpCl|18 04.6|-22 30|
M22|6656|GCl|18 36.4|-23 54|
M23|6494|OpCl|17 56.8|-19 01|
M24|6603|MWP|18 16.9|-18 29|Kleine Sagittarius-Sternwolke
M25|4725|OpCl|18 31.6|-19 15|
M26|6694|OpCl|18 45.2|-09 24|
M27|6853|PNe|19 59.6|+22 43|Hantelnebel
M28|6626|GCl|18 24.5|-24 52|
M29|6913|OpCl|20 23.9|+38 32|
M30|7099|GCl|21 40.4|-23 11|
M31|224|SpG|00 42.7|+41 16|Andromedagalaxie
M32|221|ElG|00 42.7|+40 52|
M33|598|SpG|01 33.9|+30 39|Dreiecksgalaxie
M34|1039|OpCl|02 42.0|+42 47|
M35|2168|OpCl|06 08.9|+24 20|
M36|1960|OpCl|05 36.1|+34 08|
M37|2099|OpCl|05 52.4|+32 33|
M38|1922|OpCl|05 28.4|+35 50|
M39|7092|OpCl|21 32.2|+48 26|
M40|9999|Bin*|12 22.4|+58 05|
M41|2287|OpCl|06 46.0|-20 44|
M42|1976|Neb|05 35.4|-05 27|Orionnebel
M43|1982|Neb|05 35.6|-05 16|Orionnebel (Teil B)
M44|2632|OpCl|08 40.1|+19 59|Praesepe / Bienenkorbhaufen
M45|9998|OpCl|03 47.0|+24 07|Plejaden
M46|2437|OpCl|07 41.8|-14 49|
M47|2422|OpCl|07 36.6|-14 30|
M48|2548|OpCl|08 13.8|-05 48|
M49|4472|ElG|12 29.8|+08 00|
M50|2323|OpCl|07 03.2|-08 20|
M51|5194|SpG|13 29.9|+47 12|Whirlpoolgalaxie
M52|7654|OpCl|23 24.2|+61 35|
M53|5024|GCl|13 12.9|+18 10|
M54|6715|GCl|18 55.1|-30 29|
M55|6809|GCl|19 40.0|-30 58|
M56|6779|GCl|19 16.6|+30 11|
M57|6720|PNe|18 53.6|+33 02|Ringnebel
M58|4579|SpG|12 37.7|+11 49|
M59|4621|ElG|12 42.0|+11 39|
M60|4649|ElG|12 43.7|+11 33|
M61|4303|SpG|12 21.9|+04 28|
M62|6266|GCl|17 01.2|-30 07|
M63|5055|SpG|13 15.8|+42 02|Sonnenblumengalaxie
M64|4826|SpG|12 56.7|+21 41|Black-Eye-Galaxie
M65|3623|SpG|11 18.9|+13 05|
M66|3627|SpG|11 20.2|+12 59|
M67|2682|OpCl|08 50.4|+11 49|
M68|4590|GCl|12 39.5|-26 45|
M69|6637|GCl|18 31.4|-32 21|
M70|6681|GCl|18 43.2|-32 18|
M71|6838|GCl|19 53.8|+18 47|
M72|6981|GCl|20 53.5|-12 32|
M73|6994|Ast|20 58.9|-12 38|
M74|628|SpG|01 36.7|+15 47|
M75|6864|GCl|20 06.1|-21 55|
M76|650|PNe|01 42.4|+51 34|Kleiner Hantelnebel
M77|1068|SpG|02 42.7|-00 01|
M78|2068|Neb|05 46.7|+00 03|
M79|1904|GCl|05 24.5|-24 33|
M80|6093|GCl|16 17.0|-22 59|
M81|3031|SpG|09 55.6|+69 04|Bodes Galaxie
M82|3034|IrG|09 55.8|+69 41|Zigarrengalaxie
M83|5236|SpG|13 37.0|-29 52|
M84|4374|S0G|12 25.1|+12 53|
M85|4382|S0G|12 25.4|+18 11|
M86|4406|S0G|12 26.2|+12 57|
M87|4486|ElG|12 30.8|+12 24|Virgo A
M88|4501|SpG|12 32.0|+14 25|
M89|4552|ElG|12 35.7|+12 33|
M90|4569|SpG|12 36.8|+13 10|
M91|4548|SpG|12 35.4|+14 30|
M92|6341|GCl|17 17.1|+43 08|
M93|2447|OpCl|07 44.6|-23 52|
M94|4736|SpG|12 50.9|+41 07|
M95|3351|SpG|10 44.0|+11 42|
M96|3368|SpG|10 46.8|+11 49|
M97|3587|PNe|11 14.8|+55 01|Eulennebel
M98|4192|SpG|12 13.8|+14 54|
M99|4254|SpG|12 18.8|+14 25|
M100|4321|SpG|12 22.9|+15 49|
M101|5457|SpG|14 03.2|+54 21|Feuerradgalaxie
M102|5866|S0G|15 06.5|+55 46|
M103|581|OpCl|01 33.2|+60 42|
M104|4594|SpG|12 40.0|-11 37|Sombrerogalaxie
M105|3379|ElG|10 47.8|+12 35|
M106|4258|SpG|12 19.0|+47 18|
M107|6171|GCl|16 32.5|-13 03|
M108|3556|SpG|11 11.5|+55 40|
M109|3992|SpG|11 57.6|+53 23|
M110|205|ElG|00 40.4|+41 41|
"""
_EXTRA_RAW = """
NGC7000||Neb|20 58.8|+44 20|Nordamerikanebel
IC5070||Neb|20 50.8|+44 21|Pelikannebel
NGC6960||SNR|20 45.7|+30 43|Westlicher Schleiernebel
NGC6992||SNR|20 56.4|+31 43|Oestlicher Schleiernebel
NGC6888||Neb|20 12.0|+38 21|Halbmondnebel
IC1396||Neb|21 39.1|+57 30|Elefantenruesselnebel
NGC7635||Neb|23 20.7|+61 12|Blasennebel
NGC281||Neb|00 52.8|+56 37|Pacman-Nebel
IC1805||Neb|02 33.4|+61 26|Herznebel
IC1848||Neb|02 51.4|+60 25|Seelennebel
NGC7023||Neb|21 01.6|+68 10|Irisnebel
NGC1499||Neb|04 00.7|+36 37|Kaliforniennebel
NGC2237||Neb|06 32.3|+05 03|Rosettennebel
NGC2264||Neb|06 41.1|+09 53|Weihnachtsbaum-/Konusnebel
NGC7822||Neb|00 03.6|+67 09|
NGC6946||SpG|20 34.9|+60 09|Feuerwerksgalaxie
NGC891||SpG|02 22.6|+42 21|
NGC253||SpG|00 47.6|-25 17|Sculptor-Galaxie
NGC2403||SpG|07 36.9|+65 36|
NGC6543||PNe|17 58.6|+66 38|Katzenaugennebel
NGC40||PNe|00 13.0|+72 32|
NGC7009||PNe|21 04.2|-11 22|Saturnnebel
NGC2392||PNe|07 29.2|+20 55|Eskimonebel
NGC3242||PNe|10 24.8|-18 38|Jupiters Geist
NGC6826||PNe|19 44.8|+50 31|Blinkender Planetarischer Nebel
NGC869||OpCl|02 19.0|+57 09|Doppelsternhaufen h Persei
NGC884||OpCl|02 22.4|+57 07|Doppelsternhaufen chi Persei
NGC457||OpCl|01 19.6|+58 17|Eulenhaufen
IC434||Neb|05 41.0|-02 28|Pferdekopfnebel-Region
NGC2244||OpCl|06 31.9|+04 56|Rosettenhaufen
NGC6820||Neb|19 43.1|+23 18|
Sh2-155||Neb|22 56.8|+62 37|Hoehlennebel
"""
def _build_catalog():
entries = []
for raw in (_MESSIER_RAW, _EXTRA_RAW):
for line in raw.strip().splitlines():
line = line.strip()
if not line:
continue
try:
entries.append(_parse_catalog_line(line))
except Exception:
continue
return entries
DSO_CATALOG = _build_catalog()
def angular_separation_deg(ra1, dec1, ra2, dec2):
"""Winkelabstand zweier Himmelspositionen (Grad) via Haversine."""
r1, d1, r2, d2 = map(math.radians, (ra1, dec1, ra2, dec2))
dra = r2 - r1
dd = d2 - d1
a = (math.sin(dd / 2) ** 2
+ math.cos(d1) * math.cos(d2) * math.sin(dra / 2) ** 2)
return math.degrees(2 * math.asin(min(1.0, math.sqrt(max(a, 0.0)))))
def find_catalog_matches(ra_center, dec_center, fov_deg, limit=6):
"""Alle Katalogobjekte, deren Zentrum innerhalb des halben
Bilddiagonalen-Radius liegt, sortiert nach Abstand vom Bildzentrum."""
radius = fov_deg / 2.0 + 0.15
hits = []
for obj in DSO_CATALOG:
sep = angular_separation_deg(ra_center, dec_center, obj["ra"], obj["dec"])
if sep <= radius:
hits.append((sep, obj))
hits.sort(key=lambda t: t[0])
return hits[:limit]
def _split_sexagesimal(value, round_to=1):
"""value in Grad (bzw. Stunden) -> (einheit, minute, sekunde), sauber
gerundet mit Uebertrag (verhindert z.B. '60.0s')."""
whole = int(value)
frac_min = (value - whole) * 60.0
minute = int(frac_min)
sec = round((frac_min - minute) * 60.0, round_to)
if sec >= 60.0:
sec -= 60.0
minute += 1
if minute >= 60:
minute -= 60
whole += 1
return whole, minute, sec
def deg_to_hms(ra_deg):
hh, mm, ss = _split_sexagesimal(ra_deg / 15.0)
return f"{hh:02d}h {mm:02d}m {ss:04.1f}s"
def deg_to_dms(dec_deg):
sign = "+" if dec_deg >= 0 else "-"
dd, mm, ss = _split_sexagesimal(abs(dec_deg))
return f"{sign}{dd:02d}\u00b0 {mm:02d}' {ss:04.1f}\""
def guess_astap_path():
"""Ein paar uebliche Installationsorte fuer astap_cli/astap.exe
(siehe https://www.hnsky.org/astap.htm)."""
candidates = [
r"C:\Program Files\astap\astap_cli.exe",
r"C:\Program Files\astap\astap.exe",
r"C:\astap\astap_cli.exe",
str(Path.home() / ".astap_cli" / "astap_cli.exe"),
str(Path.home() / ".astap_cli" / "astap_cli"),
"/opt/astap/astap_cli",
"/opt/astap/astap",
"/usr/bin/astap_cli",
"/usr/local/bin/astap_cli",
]
for c in candidates:
if Path(c).is_file():
return c
return ""
_ASTAP_CLI_URLS = {
("Windows", "64"): "https://sourceforge.net/projects/astap-program/files/windows_installer/astap_command-line_version_win64.zip/download",
("Windows", "32"): "https://sourceforge.net/projects/astap-program/files/windows_installer/astap_command-line_version_win32.zip/download",
("Linux", "64"): "https://sourceforge.net/projects/astap-program/files/linux_installer/astap_command-line_version_Linux_amd64.zip/download",
("Darwin", "arm64"): "https://sourceforge.net/projects/astap-program/files/macOS%20installer/astap_command-line_version_macOS_M1.zip/download",
("Darwin", "64"): "https://sourceforge.net/projects/astap-program/files/macOS%20installer/astap_command-line_version_macOS_x86_64.zip/download",
}
def _platform_key():
system = platform.system()
machine = platform.machine().lower()
if system == "Darwin" and ("arm" in machine or machine == "arm64"):
return (system, "arm64")
bits = "64" if sys.maxsize > 2**32 else "32"
return (system, bits)
def default_astap_dir():
d = Path.home() / ".astap_cli"
d.mkdir(parents=True, exist_ok=True)
return d
def download_and_extract_astap(progress_cb=None):
"""Laedt NUR das astap_cli-Programm selbst herunter (ein paar MB, ein
normales ZIP-Archiv - laesst sich sicher automatisiert entpacken) und
legt es in ~/.astap_cli ab. Gibt den Pfad zur ausfuehrbaren Datei
zurueck.
WICHTIG: Der eigentliche Sternkatalog (die Datenbank, die ASTAP zum
Loesen braucht, z.B. 'D50', mehrere hundert MB) wird bewusst NICHT
automatisch heruntergeladen und ausgefuehrt. Der wird als Windows-
Installer-.exe bzw. groessere Archive vertrieben, die nicht zuverlaessig
"still" automatisierbar sind - ein Skript, das unbeaufsichtigt ein
heruntergeladenes .exe ausfuehrt, ist ein Sicherheitsrisiko, das ich
bewusst vermeide. Stattdessen wird die passende Downloadseite im
Browser geoeffnet, du installierst den Katalog dort mit einem
Doppelklick - einmalig, danach laeuft alles automatisch.
"""
key = _platform_key()
url = _ASTAP_CLI_URLS.get(key)
if url is None:
raise RuntimeError(
f"Keine bekannte astap_cli-Downloadadresse fuer {key[0]} "
f"({key[1]}-bit). Bitte manuell von www.hnsky.org/astap.htm laden.")
if progress_cb:
progress_cb("Lade astap_cli herunter ...")
target_dir = default_astap_dir()
tmp_zip = target_dir / "_astap_download.zip"
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=60) as resp, open(tmp_zip, "wb") as f:
shutil.copyfileobj(resp, f)
if progress_cb:
progress_cb("Entpacke astap_cli ...")
exe_name = "astap_cli.exe" if key[0] == "Windows" else "astap_cli"
extracted_path = None
with zipfile.ZipFile(tmp_zip) as zf:
for member in zf.namelist():
if os.path.basename(member).lower() in (exe_name.lower(), "astap_cli"):
zf.extract(member, target_dir)
extracted_path = target_dir / member
break
if extracted_path is None:
zf.extractall(target_dir)
for p in target_dir.rglob("astap_cli*"):
if p.is_file() and os.access(p, os.X_OK) or p.suffix == ".exe":
extracted_path = p
break
tmp_zip.unlink(missing_ok=True)
if extracted_path is None or not extracted_path.is_file():
raise RuntimeError("Download war erfolgreich, aber astap_cli wurde "
"im Archiv nicht gefunden. Bitte manuell pruefen: "
+ str(target_dir))
final_path = target_dir / exe_name
if extracted_path != final_path:
shutil.move(str(extracted_path), str(final_path))
if key[0] != "Windows":
os.chmod(final_path, 0o755)
if progress_cb:
progress_cb("astap_cli eingerichtet.")
return str(final_path)
def open_astap_database_page():
"""Oeffnet die offizielle Downloadseite fuer die Sternkatalog-
Datenbanken im Standardbrowser. Das ist der Schritt, den du einmalig
selbst per Doppelklick erledigen musst (siehe Docstring oben)."""
webbrowser.open("https://www.hnsky.org/astap.htm#astap_databases")
_SIMBAD_TAP_URL = "https://simbad.cds.unistra.fr/simbad/sim-tap/sync"
def query_simbad_distances(ra_deg, dec_deg, radius_deg, limit=10, timeout=20):
"""ADQL-Abfrage bei SIMBAD: alle Objekte mit bekannter Parallaxe im
angegebenen Radius um (ra_deg, dec_deg), sortiert nach Abstand vom
Zentrum. Gibt eine Liste von Dicts zurueck (main_id, ra, dec,
parallax_mas, distance_pc, distance_ly, sep_deg)."""
adql = (
"SELECT TOP {limit} main_id, ra, dec, plx_value, "
"DISTANCE(POINT('ICRS', ra, dec), POINT('ICRS', {ra}, {dec})) AS sep "
"FROM basic "
"WHERE plx_value IS NOT NULL AND plx_value > 0 "
"AND CONTAINS(POINT('ICRS', ra, dec), "
"CIRCLE('ICRS', {ra}, {dec}, {radius})) = 1 "
"ORDER BY sep ASC"
).format(limit=int(limit), ra=ra_deg, dec=dec_deg, radius=radius_deg)
params = urllib.parse.urlencode({
"request": "doQuery", "lang": "adql", "format": "csv", "query": adql,
})
req = urllib.request.Request(
_SIMBAD_TAP_URL + "?" + params,
headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=timeout) as resp:
text = resp.read().decode("utf-8", errors="ignore")
rows = list(csv.DictReader(text.splitlines()))
out = []
for row in rows:
try:
plx = float(row["plx_value"])
if plx <= 0:
continue
dist_pc = 1000.0 / plx
out.append({
"name": row["main_id"],
"ra": float(row["ra"]),
"dec": float(row["dec"]),
"parallax_mas": plx,
"distance_pc": dist_pc,
"distance_ly": dist_pc * 3.26156,
"sep_deg": float(row["sep"]),
})
except (KeyError, ValueError):
continue
return out
def recenter_star(luminance, x, y, search=6):
"""Sucht das lokale Helligkeitsmaximum in einem +-search Fenster um
(x, y), um kleine Nachfuehr-/Zentrierfehler zwischen Aufnahmen
auszugleichen. Gibt (x, y) als Float zurueck."""
h, w = luminance.shape
x0, x1 = max(0, int(x - search)), min(w, int(x + search) + 1)
y0, y1 = max(0, int(y - search)), min(h, int(y + search) + 1)
if x1 <= x0 or y1 <= y0:
return float(x), float(y)
patch = luminance[y0:y1, x0:x1]
iy, ix = np.unravel_index(np.argmax(patch), patch.shape)
return float(x0 + ix), float(y0 + iy)
def aperture_photometry(arr, x, y, r_ap=4.0, r_in=7.0, r_out=10.0, search=6):
"""Liefert (flux, cx, cy) - Gesamtfluss in der Apertur nach Abzug des
lokalen Himmelshintergrunds (Median im Ring zwischen r_in und r_out),
sowie die nachzentrierte Sternposition."""
lum = arr.mean(axis=2) if arr.ndim == 3 else arr
cx, cy = recenter_star(lum, x, y, search=search)
h, w = lum.shape
pad = int(r_out) + 2
x0, x1 = max(0, int(cx - pad)), min(w, int(cx + pad) + 1)
y0, y1 = max(0, int(cy - pad)), min(h, int(cy + pad) + 1)
if x1 <= x0 or y1 <= y0:
return 0.0, cx, cy
yy, xx = np.mgrid[y0:y1, x0:x1]
rr = np.sqrt((xx - cx) ** 2 + (yy - cy) ** 2)
patch = lum[y0:y1, x0:x1]
ap_mask = rr <= r_ap
bg_mask = (rr >= r_in) & (rr <= r_out)
if not np.any(ap_mask):
return 0.0, cx, cy
bg_level = float(np.median(patch[bg_mask])) if np.any(bg_mask) else 0.0
flux = float(np.sum(patch[ap_mask] - bg_level))
return max(flux, 1e-9), cx, cy
def format_mpc80_line(designation, dt_utc, ra_deg, dec_deg,
mag=None, band="V", note1=" ", note2="C", obs_code=""):
desig = (designation or "")[:7].ljust(7)
field1_5 = " "
field6_12 = desig
discovery = " "
day_frac = (dt_utc.day + dt_utc.hour / 24.0 + dt_utc.minute / 1440.0
+ dt_utc.second / 86400.0)
date_str = f"{dt_utc.year:04d} {dt_utc.month:02d} {day_frac:08.5f}"
date_field = date_str.ljust(17)[:17]
ra_h, ra_m, ra_s = _split_sexagesimal(ra_deg / 15.0, round_to=2)
ra_str = f"{ra_h:02d} {ra_m:02d} {ra_s:05.2f}"
ra_field = ra_str.ljust(12)[:12]
sign = "+" if dec_deg >= 0 else "-"
dd, dm, ds = _split_sexagesimal(abs(dec_deg), round_to=1)
dec_str = f"{sign}{dd:02d} {dm:02d} {ds:04.1f}"
dec_field = dec_str.ljust(12)[:12]
blank_57_65 = " " * 9
if mag is not None:
mag_field = f"{mag:5.2f}{band[:1]}"
else:
mag_field = " " * 6
blank_72_77 = " " * 6
obs_field = (obs_code or "")[:3].ljust(3)
line = (field1_5 + field6_12 + discovery + note1[:1] + note2[:1]
+ date_field + ra_field + dec_field + blank_57_65
+ mag_field + blank_72_77 + obs_field)
line = line[:80].ljust(80)
assert len(line) == 80
return line
def parse_wcs_result(path):
"""Liest die von ASTAP erzeugte .wcs/.ini Ergebnisdatei (FITS-Header-
aehnliches Textformat: SCHLUESSEL = Wert / Kommentar). Liefert alle
Parameter, die fuer eine echte Pixel<->Himmelskoordinaten-Umrechnung
noetig sind (nicht nur das Bildzentrum)."""
text = Path(path).read_text(errors="ignore")
vals = {}
for line in text.splitlines():
m = re.match(r"\s*([A-Z0-9_-]+)\s*=\s*'?([^/'\n]+)'?", line)
if m:
key = m.group(1).strip()
val = m.group(2).strip()
vals[key] = val
if vals.get("PLTSOLVD", "").upper().startswith("F"):
return None
try:
crval1 = float(vals["CRVAL1"])
crval2 = float(vals["CRVAL2"])
naxis1 = float(vals.get("NAXIS1", 0))
naxis2 = float(vals.get("NAXIS2", 0))
crpix1 = float(vals.get("CRPIX1", naxis1 / 2.0))
crpix2 = float(vals.get("CRPIX2", naxis2 / 2.0))
if all(k in vals for k in ("CD1_1", "CD1_2", "CD2_1", "CD2_2")):
cd = (float(vals["CD1_1"]), float(vals["CD1_2"]),
float(vals["CD2_1"]), float(vals["CD2_2"]))
else:
cdelt1 = float(vals.get("CDELT1", -abs(float(vals.get("CDELT2", 0.0)))))
cdelt2 = float(vals.get("CDELT2", abs(cdelt1)))
crota2 = math.radians(float(vals.get("CROTA2", 0.0)))
cd = (cdelt1 * math.cos(crota2), -cdelt2 * math.sin(crota2),
cdelt1 * math.sin(crota2), cdelt2 * math.cos(crota2))
return {
"ra": crval1, "dec": crval2,
"crpix1": crpix1, "crpix2": crpix2,
"cd": cd,
"cdelt": abs(float(vals.get("CDELT2", vals.get("CDELT1", cd[3])))),
"naxis1": naxis1, "naxis2": naxis2,
}
except (KeyError, ValueError):
return None
def wcs_pixel_to_radec(x, y, wcs):
"""Standard-Tangentialebenen- (TAN-) Deprojektion nach Calabretta &
Greisen (2002). x/y in Pixelkoordinaten (1-indexiert wie FITS/CRPIX),
liefert (ra_deg, dec_deg)."""
cd1_1, cd1_2, cd2_1, cd2_2 = wcs["cd"]
dx = x - wcs["crpix1"]
dy = y - wcs["crpix2"]
xi = math.radians(cd1_1 * dx + cd1_2 * dy)
eta = math.radians(cd2_1 * dx + cd2_2 * dy)
ra0 = math.radians(wcs["ra"])
dec0 = math.radians(wcs["dec"])
ra = ra0 + math.atan2(xi, math.cos(dec0) - eta * math.sin(dec0))
dec = math.atan2(eta * math.cos(dec0) + math.sin(dec0),
math.sqrt((math.cos(dec0) - eta * math.sin(dec0)) ** 2 + xi ** 2))
return math.degrees(ra) % 360.0, math.degrees(dec)
def wcs_radec_to_pixel(ra_deg, dec_deg, wcs):
"""Umkehrung von wcs_pixel_to_radec (gnomonische Vorwaertsprojektion),
liefert Pixelkoordinaten (x, y)."""
ra0 = math.radians(wcs["ra"])
dec0 = math.radians(wcs["dec"])
ra = math.radians(ra_deg)
dec = math.radians(dec_deg)
dra = ra - ra0
denom = (math.sin(dec) * math.sin(dec0)
+ math.cos(dec) * math.cos(dec0) * math.cos(dra))
if abs(denom) < 1e-12:
raise ValueError("Punkt liegt ausserhalb der Tangentialebene (90 Grad entfernt).")
xi = math.cos(dec) * math.sin(dra) / denom
eta = (math.sin(dec) * math.cos(dec0)
- math.cos(dec) * math.sin(dec0) * math.cos(dra)) / denom
xi_deg, eta_deg = math.degrees(xi), math.degrees(eta)
cd1_1, cd1_2, cd2_1, cd2_2 = wcs["cd"]
det = cd1_1 * cd2_2 - cd1_2 * cd2_1
if abs(det) < 1e-15:
raise ValueError("Entartete WCS-Matrix.")
dx = (cd2_2 * xi_deg - cd1_2 * eta_deg) / det
dy = (cd1_1 * eta_deg - cd2_1 * xi_deg) / det
return wcs["crpix1"] + dx, wcs["crpix2"] + dy
def position_angle_and_separation(ra1, dec1, ra2, dec2):
"""Klassischer Doppelstern-Positionswinkel (0 Grad = Nord, 90 Grad =
Ost, wachsend im Uhrzeigersinn) und Winkelabstand in Bogensekunden."""
r1, d1, r2, d2 = map(math.radians, (ra1, dec1, ra2, dec2))
dra = r2 - r1
pa = math.degrees(math.atan2(
math.sin(dra) * math.cos(d2),
math.cos(d1) * math.sin(d2) - math.sin(d1) * math.cos(d2) * math.cos(dra)))
pa = pa % 360.0
sep = angular_separation_deg(ra1, dec1, ra2, dec2) * 3600.0
return pa, sep
def run_astap_solve(exe_path, image_path, search_radius=180, fov_deg=None,
downsample=2, timeout=120):
"""Ruft astap_cli fuer ein Einzelbild auf. Gibt bei Erfolg ein Dict mit
ra/dec (Grad, Bildzentrum) und Pixelskala (Grad/Pixel) zurueck, sonst
hebt eine Exception mit Klartext-Fehlermeldung."""
if not exe_path or not Path(exe_path).is_file():
raise RuntimeError(
"ASTAP-Programm nicht gefunden. Bitte Pfad zu astap_cli(.exe) "
"unter 'Objekterkennung' eintragen (ASTAP muss separat "
"installiert werden, inkl. Sternkatalog-Datenbank: "
"www.hnsky.org/astap.htm).")
out_prefix = str(Path(image_path).with_suffix(""))
cmd = [exe_path, "-f", str(image_path), "-r", str(search_radius),
"-z", str(downsample), "-o", out_prefix, "-wcs"]
if fov_deg and fov_deg > 0:
cmd += ["-fov", str(fov_deg)]
try:
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
except FileNotFoundError:
raise RuntimeError("astap_cli laesst sich nicht starten (Pfad pruefen).")
except subprocess.TimeoutExpired:
raise RuntimeError("Plate-Solving hat zu lange gedauert (Timeout).")
wcs_path = Path(out_prefix + ".wcs")
if not wcs_path.is_file():
wcs_path = Path(out_prefix + ".ini")
if not wcs_path.is_file():
out = (proc.stdout or "") + (proc.stderr or "")
raise RuntimeError("Keine Loesung gefunden.\n" + out.strip()[-400:])
result = parse_wcs_result(wcs_path)
if result is None:
raise RuntimeError("Keine Loesung gefunden (Sternfeld nicht erkannt). "
"Mehr Sterne/kontrastreicheres Bild oder groesseren "
"Suchradius versuchen.")
return result
_COLOR_BUCKETS = [
(-0.30, "blau-weiss (heiss, O/B-aehnlich)"),
(-0.05, "weiss (A-aehnlich)"),
(0.15, "gelb-weiss (F-aehnlich)"),
(0.35, "gelb (G-aehnlich, sonnenaehnlich)"),
(0.60, "orange (K-aehnlich)"),
(999.0, "rot (M-aehnlich, kuehl)"),
]
def _classify_color_index(ci):
for threshold, label in _COLOR_BUCKETS:
if ci <= threshold:
return label
return _COLOR_BUCKETS[-1][1]
def detect_bright_stars(rgb_or_gray, max_stars=15, min_separation=12):
"""Simple, abhaengigkeitsarme Sternerkennung: sucht lokale Helligkeits-
maxima oberhalb eines robusten Schwellwerts. Kein astrometrisches
Sternprofil-Fitting, nur eine praktische Naeherung fuer die Uebersicht."""
if rgb_or_gray.ndim == 3:
lum = rgb_or_gray.mean(axis=2)
else:
lum = rgb_or_gray
h, w = lum.shape
sample = lum[::2, ::2]
med = float(np.median(sample))
mad = float(np.median(np.abs(sample - med))) * 1.4826 or 1e-4
thresh = med + 6.0 * mad
ys, xs = np.where(lum > thresh)
if len(xs) == 0:
return []
order = np.argsort(-lum[ys, xs])
ys, xs = ys[order], xs[order]
picked = []
for y, x in zip(ys, xs):
if len(picked) >= max_stars:
break
too_close = False
for py, px in picked:
if (py - y) ** 2 + (px - x) ** 2 < min_separation ** 2:
too_close = True
break
if not too_close:
picked.append((y, x))
return picked
def analyze_star_colors(rgb_arr, max_stars=15):
"""Liefert pro erkanntem hellen Stern Position + grobe Farbklasse.
Braucht ein Farbbild (RGB); bei Graustufen wird nur die Position
zurueckgegeben (keine Farbschaetzung moeglich)."""
is_color = rgb_arr.ndim == 3 and rgb_arr.shape[2] >= 3
stars = detect_bright_stars(rgb_arr, max_stars=max_stars)
h, w = rgb_arr.shape[:2]
out = []
for (y, x) in stars:
y0, y1 = max(0, y - 2), min(h, y + 3)
x0, x1 = max(0, x - 2), min(w, x + 3)
if is_color:
patch = rgb_arr[y0:y1, x0:x1, :3].reshape(-1, 3)
r, g, b = patch.mean(axis=0)
total = max(r + g + b, 1e-6)
ci = (b - r) / total * -2.5
label = _classify_color_index(ci)
else:
ci = None
label = "keine Farbdaten (Graustufenbild)"
out.append({"x": int(x), "y": int(y), "color_index": ci, "label": label})
return out
def mtf(m, x):
"""Midtone Transfer Function. x und m im Bereich 0..1."""
x = np.clip(x, 0.0, 1.0)
denom = (2.0 * m - 1.0) * x - m
out = np.zeros_like(x)
np.divide((m - 1.0) * x, denom, out=out, where=(denom != 0))
return np.clip(out, 0.0, 1.0)
def autostretch(arr, target_bg=0.25, shadow_sigma=2.8):
"""
Screen-Transfer-Function wie in PixInsight/Siril.
arr: float array 0..1, Form (h, w) oder (h, w, 3).
"""
if arr.ndim == 3:
lum = arr.mean(axis=2)
else:
lum = arr
sample = lum[::4, ::4]
med = float(np.median(sample))
mad = float(np.median(np.abs(sample - med))) * 1.4826
if mad <= 0:
mad = 1e-4
black = float(np.clip(med - shadow_sigma * mad, 0.0, 1.0))
span = max(1.0 - black, 1e-6)
mid = float(np.clip((med - black) / span, 1e-4, 0.9999))
m = mtf(target_bg, np.array([mid]))[0]
m = float(np.clip(m, 1e-4, 0.9999))
shifted = np.clip((arr - black) / span, 0.0, 1.0)
return mtf(m, shifted), black, m
def manual_stretch(arr, black, midtone):
span = max(1.0 - black, 1e-6)
shifted = np.clip((arr - black) / span, 0.0, 1.0)
return mtf(max(min(midtone, 0.9999), 1e-4), shifted)
def to_pil(arr_float01):
"""float 0..1 -> PIL Image (RGB oder L)."""
eight = (np.clip(arr_float01, 0.0, 1.0) * 255.0 + 0.5).astype(np.uint8)
if eight.ndim == 2:
return Image.fromarray(eight, mode="L").convert("RGB")
return Image.fromarray(eight[:, :, :3], mode="RGB")
class SourceBase(threading.Thread):
def __init__(self):
super().__init__(daemon=True)
self._stop_flag = threading.Event()
self._lock = threading.Lock()
self._frame = None
self._meta = {}
self.status = "gestoppt"
self.error = None
def stop(self):
self._stop_flag.set()
def latest(self):
with self._lock:
return self._frame, dict(self._meta)
def _publish(self, frame, meta):
with self._lock:
self._frame = frame
self._meta = meta
class RTSPSource(SourceBase):
"""Echtes Video vom eingebauten RTSP-Server."""
def __init__(self, url):
super().__init__()
self.url = url
def run(self):
if cv2 is None:
self.error = "OpenCV fehlt (pip install opencv-python)"
self.status = "Fehler"
return
os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = (
"rtsp_transport;tcp|stimeout;5000000|max_delay;500000"
)
misses = 0
cap = None
frames = 0
t0 = time.time()
fps = 0.0
while not self._stop_flag.is_set():
if cap is None or not cap.isOpened():
self.status = "verbinde ..."
cap = cv2.VideoCapture(self.url, cv2.CAP_FFMPEG)
try:
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
except Exception:
pass
if not cap.isOpened():
self.status = "keine Verbindung"
time.sleep(2.0)
continue
ok, bgr = cap.read()
if not ok:
misses += 1
if misses > 25:
self.status = "Stream weg, neu verbinden"
cap.release()
cap = None
misses = 0
time.sleep(1.5)
continue
misses = 0
frames += 1
if frames % 20 == 0:
now = time.time()
fps = 20.0 / max(now - t0, 1e-6)
t0 = now
rgb = cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
h, w = rgb.shape[:2]
self.status = "live"
self._publish(rgb, {
"quelle": "RTSP",
"breite": w,
"hoehe": h,
"fps": fps,
})
if cap is not None:
cap.release()
self.status = "gestoppt"
class StackSource(SourceBase):
"""Pollt das aktuelle Stackbild ueber Port 4800."""
def __init__(self, ip, interval=3.0, preview=False, wide=False):
super().__init__()
self.ip = ip
self.interval = interval
self.preview = preview
self.wide = wide
def run(self):
if ss_stream is None:
self.error = "seestarpy fehlt (pip install seestarpy)"
self.status = "Fehler"
return
last_id = None
while not self._stop_flag.is_set():
try:
kwargs = {}
if self.ip:
kwargs["ip"] = self.ip
if self.preview:
kwargs["method"] = "get_current_img"
if self.wide:
wide_port = getattr(ss_stream, "IMAGE_PORT_WIDE", None)
if wide_port:
kwargs["port"] = wide_port
self.status = "hole Bild ..."
header, payload = ss_stream.get_live_image(**kwargs)
pixels = ss_stream.decode_payload(payload, header)
arr = np.asarray(pixels)
if arr.dtype == np.uint16:
arr = arr.astype(np.float32) / 65535.0
elif arr.dtype == np.uint8:
arr = arr.astype(np.float32) / 255.0
else:
arr = arr.astype(np.float32)
if arr.max() > 1.5:
arr = arr / arr.max()
img_type = header.get("img_type")
self.status = "live"
self._publish(arr, {
"quelle": "Stack" if img_type == 5 else "Vorschau",
"breite": header.get("width", arr.shape[1]),
"hoehe": header.get("height", arr.shape[0]),
"frame": header.get("image_id"),
"hfd": header.get("hfd"),
})
last_id = header.get("image_id")
self.error = None
except Exception as exc:
txt = str(exc)
if "No image-bearing frame" in txt:
self.error = ("Seestar liefert gerade kein Bild. "
"Laeuft eine Deep-Sky-Aufnahme (Stacking)? "
"Port 4800 sendet nur waehrend des Stackens.")
elif isinstance(exc, (ConnectionRefusedError, TimeoutError)) \
or "timed out" in txt or "getaddrinfo" in txt:
self.error = (f"Keine Verbindung zu {self.ip or '?'}:4800. "
"IP pruefen; evtl. blockiert die offizielle "
"App den Port oder Firmware verlangt den "
"RSA-Schluessel.")
else:
self.error = txt
self.status = "Fehler"
waited = 0.0
while waited < self.interval and not self._stop_flag.is_set():
time.sleep(0.2)
waited += 0.2
self.status = "gestoppt"
class ViewWindow(tk.Toplevel):
"""Anzeigefenster mit Zoom (Mausrad) und Pan (Ziehen bei gedrueckter
linker Maustaste). Doppelklick oder Taste '0' setzt den Zoom zurueck."""
ZOOM_MIN = 1.0
ZOOM_MAX = 12.0
ZOOM_STEP = 1.25
def __init__(self, master, app):
super().__init__(master)
self.app = app
self.title(VIEW_TITLE)
self.configure(bg="black")
self.geometry("1280x720")
self.minsize(320, 240)
self.canvas = tk.Canvas(self, bg="black", highlightthickness=0, bd=0)
self.canvas.pack(fill="both", expand=True)
self._photo = None
self._fullscreen = False
self._last_pil = None
self.zoom = 1.0
self.center_u = 0.5
self.center_v = 0.5
self._drag_start = None
self._drag_center0 = None
self._pending_click_cb = None
self._draw_x0 = self._draw_y0 = 0.0
self._draw_scale = None
self._draw_nw = self._draw_nh = 0
self._draw_cw = self._draw_ch = 0
self.bind("<F11>", self.toggle_fullscreen)
self.bind("<Escape>", self.leave_fullscreen)
self.bind("0", self.reset_zoom)
self.protocol("WM_DELETE_WINDOW", self.hide)
self.canvas.bind("<MouseWheel>", self._on_wheel)
self.canvas.bind("<Button-4>", self._on_wheel_linux)
self.canvas.bind("<Button-5>", self._on_wheel_linux)
self.canvas.bind("<ButtonPress-1>", self._on_drag_start)
self.canvas.bind("<B1-Motion>", self._on_drag_move)
self.canvas.bind("<Double-Button-1>", self.reset_zoom)
self.canvas.bind("<Configure>", lambda e: self._redraw())
def toggle_fullscreen(self, _=None):
self._fullscreen = not self._fullscreen
self.attributes("-fullscreen", self._fullscreen)
def leave_fullscreen(self, _=None):
self._fullscreen = False
self.attributes("-fullscreen", False)
def hide(self):
self.withdraw()
def show(self):
self.deiconify()
self.lift()
def request_click(self, callback):
"""Naechster Linksklick im Livebild ruft callback(img_x, img_y) mit
Pixelkoordinaten im *angezeigten* (ggf. geflippten) Bild auf, statt
zu schwenken. Fuer Sternmarkierung (Photometrie/Astrometrie)."""
self._pending_click_cb = callback
self.canvas.configure(cursor="crosshair")
self.show()
def cancel_click_request(self):
self._pending_click_cb = None
self.canvas.configure(cursor="")
def canvas_to_image(self, cx, cy):
if self._draw_scale is None:
return None
ox = self._draw_cw // 2 - self._draw_nw // 2
oy = self._draw_ch // 2 - self._draw_nh // 2
ix = (cx - ox) / self._draw_scale + self._draw_x0
iy = (cy - oy) / self._draw_scale + self._draw_y0
return ix, iy
def reset_zoom(self, _=None):
self.zoom = 1.0
self.center_u = 0.5
self.center_v = 0.5
self._redraw()
def _on_wheel(self, event):
self._zoom_at(event.x, event.y, 1 if event.delta > 0 else -1)
def _on_wheel_linux(self, event):
self._zoom_at(event.x, event.y, 1 if event.num == 4 else -1)
def _zoom_at(self, px, py, direction):
if self._last_pil is None:
return
old_zoom = self.zoom
if direction > 0:
self.zoom = min(self.ZOOM_MAX, self.zoom * self.ZOOM_STEP)
else:
self.zoom = max(self.ZOOM_MIN, self.zoom / self.ZOOM_STEP)
if self.zoom == old_zoom:
return
cw = max(self.canvas.winfo_width(), 1)
ch = max(self.canvas.winfo_height(), 1)
u_at_cursor = self.center_u + (px - cw / 2) / (cw * old_zoom)
v_at_cursor = self.center_v + (py - ch / 2) / (ch * old_zoom)
self.center_u = u_at_cursor - (px - cw / 2) / (cw * self.zoom)
self.center_v = v_at_cursor - (py - ch / 2) / (ch * self.zoom)
self._clamp_center()
self._redraw()
def _on_drag_start(self, event):
if self._pending_click_cb is not None:
pos = self.canvas_to_image(event.x, event.y)
cb = self._pending_click_cb
self._pending_click_cb = None
self.canvas.configure(cursor="")
if pos is not None:
cb(pos[0], pos[1])
return
self._drag_start = (event.x, event.y)
self._drag_center0 = (self.center_u, self.center_v)
def _on_drag_move(self, event):
if self._drag_start is None or self.zoom <= 1.0:
return
cw = max(self.canvas.winfo_width(), 1)
ch = max(self.canvas.winfo_height(), 1)
dx = event.x - self._drag_start[0]
dy = event.y - self._drag_start[1]
u0, v0 = self._drag_center0
self.center_u = u0 - dx / (cw * self.zoom)
self.center_v = v0 - dy / (ch * self.zoom)
self._clamp_center()
self._redraw()
def _clamp_center(self):
half = 0.5 / self.zoom
self.center_u = min(1 - half, max(half, self.center_u))
self.center_v = min(1 - half, max(half, self.center_v))
if self.zoom <= 1.0:
self.center_u = 0.5
self.center_v = 0.5
def draw(self, pil_img):
self._last_pil = pil_img
self._redraw()
def _redraw(self):
pil_img = self._last_pil
if pil_img is None:
return
cw = max(self.canvas.winfo_width(), 1)
ch = max(self.canvas.winfo_height(), 1)
if cw < 10 or ch < 10:
return
iw, ih = pil_img.size
if self.zoom > 1.0:
crop_w = iw / self.zoom
crop_h = ih / self.zoom
cx, cy = self.center_u * iw, self.center_v * ih
x0 = max(0, min(iw - crop_w, cx - crop_w / 2))
y0 = max(0, min(ih - crop_h, cy - crop_h / 2))
crop = pil_img.crop((int(x0), int(y0),
int(x0 + crop_w), int(y0 + crop_h)))
src_w, src_h = crop.size
else:
x0, y0 = 0.0, 0.0
crop = pil_img
src_w, src_h = iw, ih
scale = min(cw / src_w, ch / src_h)
nw, nh = max(int(src_w * scale), 1), max(int(src_h * scale), 1)
resample = Image.BILINEAR if scale < 1 else Image.NEAREST
shown = crop.resize((nw, nh), resample)
self._draw_x0, self._draw_y0 = x0, y0
self._draw_scale = scale
self._draw_nw, self._draw_nh = nw, nh
self._draw_cw, self._draw_ch = cw, ch
self._photo = ImageTk.PhotoImage(shown)
self.canvas.delete("all")
self.canvas.create_image(cw // 2, ch // 2, image=self._photo)
if self.zoom > 1.0:
self.canvas.create_text(
cw - 8, 8, anchor="ne", text=f"{self.zoom:.1f}x",
fill="#cccccc", font=("Segoe UI", 11))
def clear(self, text=""):
self._last_pil = None
self.canvas.delete("all")
if text:
cw = max(self.canvas.winfo_width(), 1)
ch = max(self.canvas.winfo_height(), 1)
self.canvas.create_text(cw // 2, ch // 2, text=text,
fill="#555555", font=("Segoe UI", 16))
class LightCurveWindow(tk.Toplevel):
"""Schlanker, abhaengigkeitsfreier Lichtkurven-Plot auf einem Tkinter-
Canvas (keine matplotlib-Abhaengigkeit noetig)."""
def __init__(self, master, log):
super().__init__(master)
self.title("Lichtkurve (Delta-Magnitude)")
self.geometry("640x420")
self.configure(bg="white")
self.canvas = tk.Canvas(self, bg="white", highlightthickness=0)
self.canvas.pack(fill="both", expand=True)
self.log = log
self.canvas.bind("<Configure>", lambda e: self._draw())
self._draw()
def _draw(self):
self.canvas.delete("all")
w = max(self.canvas.winfo_width(), 100)
h = max(self.canvas.winfo_height(), 100)
margin = 50
if len(self.log) < 2:
return
ts = [e["t"] for e in self.log]
mags = [e["delta_mag"] for e in self.log]
t0, t1 = min(ts), max(ts)
m0, m1 = min(mags), max(mags)
if t1 == t0:
t1 = t0 + 1
if m1 == m0:
m0 -= 0.05
m1 += 0.05
pad_m = (m1 - m0) * 0.1
m0 -= pad_m
m1 += pad_m
def px(t):
return margin + (t - t0) / (t1 - t0) * (w - 2 * margin)
def py(m):
return margin + (m - m0) / (m1 - m0) * (h - 2 * margin)
self.canvas.create_line(margin, h - margin, w - margin, h - margin, fill="#888")
self.canvas.create_line(margin, margin, margin, h - margin, fill="#888")
self.canvas.create_text(w // 2, h - margin // 2,
text="Zeit", fill="#333")
self.canvas.create_text(margin // 2, margin, text="Delta-mag",
fill="#333", angle=90)
for i, m in enumerate(np.linspace(m0, m1, 5)):
y = py(m)
self.canvas.create_line(margin - 4, y, margin, y, fill="#888")
self.canvas.create_text(margin - 8, y, text=f"{m:+.2f}",
anchor="e", fill="#333", font=("Segoe UI", 8))
pts = [(px(e["t"]), py(e["delta_mag"])) for e in self.log]
for i in range(len(pts) - 1):
self.canvas.create_line(*pts[i], *pts[i + 1], fill="#2266cc", width=2)
for x, y in pts:
self.canvas.create_oval(x - 3, y - 3, x + 3, y + 3,
fill="#2266cc", outline="")
class ControlApp(tk.Tk):
def __init__(self):
super().__init__()
self.title(APP_TITLE)
self.geometry("500x820")
self.resizable(True, True)
self.source = None
self.view = ViewWindow(self, self)
self._last_pil = None
self._save_dir = None
self._saved_ids = set()
self._solve_lock = threading.Lock()
self._solve_busy = False
self._solve_result = None
self._auto_solve_job = None
self._star_report = []
self._distance_busy = False
self._phot_target = None
self._phot_comp = None
self._phot_log = []
self._phot_auto_job = None
self._ds_pt1 = None
self._ds_pt2 = None
self._ds_log = []
self._ast_log = []
self._ref_frame = None
self._transient_candidates = []
self._build_ui()
self.protocol("WM_DELETE_WINDOW", self.on_close)
self.after(50, self._tick)
def _scroll_tab(self, notebook, title):
"""Erstellt einen neuen Reiter mit scrollbarem Inhalt (falls die
Steuerelemente nicht auf einmal in die Fensterhoehe passen)."""
outer = ttk.Frame(notebook)
notebook.add(outer, text=title)
canvas = tk.Canvas(outer, highlightthickness=0)
vsb = ttk.Scrollbar(outer, orient="vertical", command=canvas.yview)
inner = ttk.Frame(canvas)
inner.bind("<Configure>",
lambda e: canvas.configure(scrollregion=canvas.bbox("all")))
canvas.create_window((0, 0), window=inner, anchor="nw")
canvas.configure(yscrollcommand=vsb.set)
canvas.pack(side="left", fill="both", expand=True)
vsb.pack(side="right", fill="y")
def _wheel(event):
canvas.yview_scroll(int(-1 * (event.delta / 120)), "units")
canvas.bind("<Enter>", lambda e: canvas.bind_all("<MouseWheel>", _wheel))
canvas.bind("<Leave>", lambda e: canvas.unbind_all("<MouseWheel>"))
return inner
def _build_ui(self):
pad = {"padx": 10, "pady": 4}
style = ttk.Style(self)
try:
style.theme_use("vista")
except tk.TclError:
pass
nb = ttk.Notebook(self)
nb.pack(fill="both", expand=True)
t_conn = self._scroll_tab(nb, "Verbindung")
t_view = self._scroll_tab(nb, "Darstellung")
t_obj = self._scroll_tab(nb, "Objekte")
t_dist = self._scroll_tab(nb, "Entfernungen")
t_phot = self._scroll_tab(nb, "Photometrie")
t_astro = self._scroll_tab(nb, "Astrometrie")
t_trans = self._scroll_tab(nb, "Transienten")
t_save = self._scroll_tab(nb, "Speichern")
box = ttk.LabelFrame(t_conn, text="Verbindung")
box.pack(fill="x", **pad)
row = ttk.Frame(box)
row.pack(fill="x", padx=8, pady=6)
ttk.Label(row, text="IP:").pack(side="left")
self.var_ip = tk.StringVar(value="192.168.1.")
ttk.Entry(row, textvariable=self.var_ip, width=18).pack(side="left", padx=6)
ttk.Button(row, text="Suchen", width=9,
command=self.discover).pack(side="left")
box = ttk.LabelFrame(t_conn, text="Schluessel (noetig ab Firmware 7.18)")
box.pack(fill="x", **pad)
self.var_key = tk.StringVar(value="")
ttk.Label(box, textvariable=self.var_key, wraplength=370,
justify="left").pack(anchor="w", padx=8, pady=4)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=4)
ttk.Button(r, text="Aus APK holen", command=self.extract_key
).pack(side="left", expand=True, fill="x", padx=2)
ttk.Button(r, text="PEM waehlen", command=self.pick_key
).pack(side="left", expand=True, fill="x", padx=2)
ttk.Button(r, text="Neu pruefen", width=11, command=self.refresh_key
).pack(side="left", padx=2)
box = ttk.LabelFrame(t_conn, text="Bildquelle")
box.pack(fill="x", **pad)
self.var_mode = tk.StringVar(value="stack")
ttk.Radiobutton(box, text="Live-Stack (Deep Sky, Port 4800)",
variable=self.var_mode, value="stack",
command=self._mode_changed).pack(anchor="w", padx=8, pady=2)
ttk.Radiobutton(box, text="RTSP-Sucher (Mond, Planeten, Landschaft)",
variable=self.var_mode, value="rtsp",
command=self._mode_changed).pack(anchor="w", padx=8, pady=2)
self.frm_stack = ttk.Frame(box)
self.frm_stack.pack(fill="x", padx=20, pady=2)
self.var_preview = tk.BooleanVar(value=False)
ttk.Checkbutton(self.frm_stack, text="Live-Einzelbild (statt wachsendem Summenbild)",
variable=self.var_preview).pack(anchor="w")
ttk.Label(self.frm_stack, foreground="#777777", wraplength=380, justify="left", text=
"Aus: du siehst das staendig waechsende Summenbild (Rauschen "
"sinkt ueber Zeit, aber alte Belichtungen bleiben drin).\n"
"An: du siehst jede einzelne Roh-Belichtung einzeln, so "
"wie sie gerade reinkommt - gut zum Fokussieren/Rahmen, "
"aber verrauschter."
).pack(anchor="w", pady=(0, 4))
self.var_wide = tk.BooleanVar(value=False)
ttk.Checkbutton(self.frm_stack, text="Weitwinkelkamera",
variable=self.var_wide).pack(anchor="w")
r = ttk.Frame(self.frm_stack)
r.pack(fill="x", pady=2)
ttk.Label(r, text="Abfrage alle").pack(side="left")
self.var_interval = tk.DoubleVar(value=3.0)
ttk.Spinbox(r, from_=1.0, to=60.0, increment=1.0, width=5,
textvariable=self.var_interval).pack(side="left", padx=4)
ttk.Label(r, text="Sek.").pack(side="left")
self.frm_rtsp = ttk.Frame(box)
self.frm_rtsp.pack(fill="x", padx=20, pady=2)
ttk.Label(self.frm_rtsp, text="RTSP-URL:").pack(anchor="w")
self.var_url = tk.StringVar(value="")
ttk.Entry(self.frm_rtsp, textvariable=self.var_url,
width=40).pack(fill="x", pady=2)
ttk.Label(self.frm_rtsp, text="leer = rtsp://<IP>:4554/stream",
foreground="#777777").pack(anchor="w")
r = ttk.Frame(t_conn)
r.pack(fill="x", padx=10, pady=6)
self.btn_start = ttk.Button(r, text="Start", command=self.start_source)
self.btn_start.pack(side="left", expand=True, fill="x", padx=2)
self.btn_stop = ttk.Button(r, text="Stop", command=self.stop_source,
state="disabled")
self.btn_stop.pack(side="left", expand=True, fill="x", padx=2)
box = ttk.LabelFrame(t_view, text="Darstellung")
box.pack(fill="x", **pad)
self.var_auto = tk.BooleanVar(value=True)
ttk.Checkbutton(box, text="Auto-Stretch", variable=self.var_auto,
command=self._auto_toggled).pack(anchor="w", padx=8, pady=2)
self.var_black = tk.DoubleVar(value=0.0)
self.var_mid = tk.DoubleVar(value=0.5)
self.sc_black = self._slider(box, "Schwarzpunkt", self.var_black, 0.0, 0.9)
self.sc_mid = self._slider(box, "Mitten", self.var_mid, 0.02, 0.98)
self.var_gray = tk.BooleanVar(value=False)
ttk.Checkbutton(box, text="Graustufen",
variable=self.var_gray).pack(anchor="w", padx=8)
self.var_flip = tk.BooleanVar(value=False)
ttk.Checkbutton(box, text="180 Grad drehen",
variable=self.var_flip).pack(anchor="w", padx=8)
box = ttk.LabelFrame(t_view, text="Einblendung im Livebild")
box.pack(fill="x", **pad)
self.var_overlay = tk.BooleanVar(value=True)
ttk.Checkbutton(box, text="Infozeile anzeigen", variable=self.var_overlay
).pack(anchor="w", padx=8, pady=2)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Label(r, text="Titel:").pack(side="left")
self.var_caption = tk.StringVar(value="")
ttk.Entry(r, textvariable=self.var_caption).pack(
side="left", fill="x", expand=True, padx=4)
grid = ttk.Frame(box)
grid.pack(fill="x", padx=8, pady=(4, 2))
self.var_show_title = tk.BooleanVar(value=True)
self.var_show_time = tk.BooleanVar(value=True)
self.var_show_date = tk.BooleanVar(value=False)
self.var_show_source = tk.BooleanVar(value=True)
self.var_show_res = tk.BooleanVar(value=True)
self.var_show_frame = tk.BooleanVar(value=True)
self.var_show_hfd = tk.BooleanVar(value=True)
self.var_show_fps = tk.BooleanVar(value=True)
self.var_show_object = tk.BooleanVar(value=True)
toggles = [
("Titel", self.var_show_title), ("Uhrzeit", self.var_show_time),
("Datum", self.var_show_date), ("Quelle", self.var_show_source),
("Aufloesung", self.var_show_res), ("Frame-Nr.", self.var_show_frame),
("HFD", self.var_show_hfd), ("FPS", self.var_show_fps),
("Erkanntes Objekt", self.var_show_object),
]
for i, (label, var) in enumerate(toggles):
ttk.Checkbutton(grid, text=label, variable=var).grid(
row=i // 2, column=i % 2, sticky="w", padx=2, pady=1)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=(6, 2))
ttk.Label(r, text="Position:").pack(side="left")
self.var_ov_pos = tk.StringVar(value="unten links")
ttk.Combobox(r, textvariable=self.var_ov_pos, state="readonly", width=13,
values=["oben links", "oben rechts",
"unten links", "unten rechts"]
).pack(side="left", padx=4)
ttk.Label(r, text="Groesse:").pack(side="left", padx=(10, 0))
self.var_ov_size = tk.IntVar(value=0)
ttk.Spinbox(r, from_=0, to=72, increment=1, width=4,
textvariable=self.var_ov_size).pack(side="left", padx=4)
ttk.Label(r, text="(0=auto)", foreground="#777777").pack(side="left")
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=(2, 4))
self.var_ov_color = tk.StringVar(value="#ffffff")
ttk.Button(r, text="Textfarbe waehlen", command=self._pick_overlay_color
).pack(side="left")
self.lbl_ov_color = tk.Label(r, text=" ", bg=self.var_ov_color.get(),
width=3, relief="sunken")
self.lbl_ov_color.pack(side="left", padx=6)
box = ttk.LabelFrame(t_obj, text="Objekterkennung (ASTAP Plate-Solving)")
box.pack(fill="x", **pad)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=4)
ttk.Label(r, text="astap_cli:").pack(side="left")
self.var_astap_path = tk.StringVar(value=guess_astap_path())
ttk.Entry(r, textvariable=self.var_astap_path).pack(
side="left", fill="x", expand=True, padx=4)
ttk.Button(r, text="...", width=3, command=self._pick_astap
).pack(side="left")
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=(0, 4))
ttk.Button(r, text="ASTAP automatisch einrichten",
command=self.setup_astap).pack(fill="x")
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Label(r, text="Bildfeld (Bogenmin, optional):").pack(side="left")
self.var_fov = tk.StringVar(value="")
ttk.Entry(r, textvariable=self.var_fov, width=8).pack(side="left", padx=4)
ttk.Label(r, text="Suchradius:").pack(side="left", padx=(10, 0))
self.var_search_radius = tk.IntVar(value=180)
ttk.Spinbox(r, from_=5, to=180, width=5,
textvariable=self.var_search_radius).pack(side="left", padx=4)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Button(r, text="Jetzt loesen", command=self.solve_now
).pack(side="left", expand=True, fill="x", padx=2)
self.var_auto_solve = tk.BooleanVar(value=False)
ttk.Checkbutton(r, text="automatisch alle", variable=self.var_auto_solve,
command=self._auto_solve_toggled).pack(side="left", padx=(8, 2))
self.var_solve_interval = tk.IntVar(value=60)
ttk.Spinbox(r, from_=15, to=600, width=5,
textvariable=self.var_solve_interval).pack(side="left")
ttk.Label(r, text="Sek.").pack(side="left", padx=2)
self.var_solve_status = tk.StringVar(value="noch nicht geloest")
ttk.Label(box, textvariable=self.var_solve_status, foreground="#333333",
wraplength=390, justify="left").pack(anchor="w", padx=8, pady=(2, 6))
box = ttk.LabelFrame(t_obj, text="Sternanalyse (grobe Naeherung, keine Spektroskopie)")
box.pack(fill="x", **pad)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=4)
ttk.Button(r, text="Sterne analysieren", command=self.analyze_stars_now
).pack(side="left", expand=True, fill="x", padx=2)
self.var_mark_stars = tk.BooleanVar(value=False)
ttk.Checkbutton(r, text="im Bild markieren",
variable=self.var_mark_stars).pack(side="left", padx=6)
self.txt_stars = tk.Text(box, height=6, wrap="word", font=("Consolas", 9))
self.txt_stars.pack(fill="x", padx=8, pady=(0, 6))
self.txt_stars.insert("1.0", "noch keine Analyse durchgefuehrt")
self.txt_stars.configure(state="disabled")
box = ttk.LabelFrame(t_dist, text="Sternentfernungen (SIMBAD/Gaia-Katalog)")
box.pack(fill="x", **pad)
ttk.Label(box, wraplength=390, justify="left", foreground="#555555", text=
"Keine Eigenmessung - fragt bekannte Parallaxen (Gaia) bei "
"SIMBAD ab, im Umkreis um die zuletzt geloeste Bildmitte. "
"Braucht Internet und einen vorherigen Plate-Solve (Tab "
"'Objekte')."
).pack(anchor="w", padx=8, pady=(4, 6))
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Label(r, text="Suchradius (Bogenmin):").pack(side="left")
self.var_dist_radius = tk.DoubleVar(value=15.0)
ttk.Spinbox(r, from_=1, to=120, width=6,
textvariable=self.var_dist_radius).pack(side="left", padx=4)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=4)
ttk.Button(r, text="Entfernungen abrufen", command=self.fetch_distances
).pack(fill="x")
self.txt_dist = tk.Text(box, height=10, wrap="word", font=("Consolas", 9))
self.txt_dist.pack(fill="x", padx=8, pady=(0, 6))
self.txt_dist.insert("1.0", "noch keine Abfrage durchgefuehrt")
self.txt_dist.configure(state="disabled")
box = ttk.LabelFrame(t_phot, text="Helligkeitskurve (differenzielle Photometrie)")
box.pack(fill="x", **pad)
ttk.Label(box, wraplength=390, justify="left", foreground="#555555", text=
"Zielstern gegen einen (moeglichst aehnlich hellen, "
"aehnlich farbigen) Vergleichsstern messen. Funktioniert "
"auf dem ROHEN Bild (vor Stretch), also unabhaengig von "
"der Anzeige-Helligkeit."
).pack(anchor="w", padx=8, pady=(4, 6))
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Button(r, text="Zielstern markieren",
command=lambda: self._start_marking("phot_target")
).pack(side="left", expand=True, fill="x", padx=2)
ttk.Button(r, text="Vergleichsstern markieren",
command=lambda: self._start_marking("phot_comp")
).pack(side="left", expand=True, fill="x", padx=2)
self.var_phot_marks = tk.StringVar(value="Ziel: - | Vergleich: -")
ttk.Label(box, textvariable=self.var_phot_marks,
foreground="#333333").pack(anchor="w", padx=8, pady=2)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Label(r, text="Apertur:").pack(side="left")
self.var_ap_r = tk.DoubleVar(value=4.0)
ttk.Spinbox(r, from_=2, to=15, width=4,
textvariable=self.var_ap_r).pack(side="left", padx=4)
ttk.Label(r, text="Ring innen/aussen:").pack(side="left", padx=(10, 0))
self.var_ap_rin = tk.DoubleVar(value=7.0)
ttk.Spinbox(r, from_=3, to=25, width=4,
textvariable=self.var_ap_rin).pack(side="left", padx=4)
self.var_ap_rout = tk.DoubleVar(value=10.0)
ttk.Spinbox(r, from_=4, to=35, width=4,
textvariable=self.var_ap_rout).pack(side="left", padx=2)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=4)
ttk.Button(r, text="Messpunkt aufnehmen", command=self.take_photometry_point
).pack(side="left", expand=True, fill="x", padx=2)
self.var_phot_auto = tk.BooleanVar(value=False)
ttk.Checkbutton(r, text="automatisch alle", variable=self.var_phot_auto,
command=self._phot_auto_toggled).pack(side="left", padx=(8, 2))
self.var_phot_interval = tk.IntVar(value=60)
ttk.Spinbox(r, from_=10, to=3600, width=6,
textvariable=self.var_phot_interval).pack(side="left")
ttk.Label(r, text="Sek.").pack(side="left", padx=2)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=(0, 4))
ttk.Button(r, text="Lichtkurve anzeigen", command=self.show_light_curve
).pack(side="left", expand=True, fill="x", padx=2)
ttk.Button(r, text="CSV exportieren", command=self.export_photometry_csv
).pack(side="left", expand=True, fill="x", padx=2)
self.txt_phot = tk.Text(box, height=8, wrap="none", font=("Consolas", 9))
self.txt_phot.pack(fill="x", padx=8, pady=(0, 6))
self.txt_phot.insert("1.0", "noch keine Messpunkte")
self.txt_phot.configure(state="disabled")
box = ttk.LabelFrame(t_astro, text="Doppelstern (Positionswinkel & Abstand)")
box.pack(fill="x", **pad)
ttk.Label(box, wraplength=390, justify="left", foreground="#555555", text=
"Braucht einen aktuellen Plate-Solve (Tab 'Objekte'), "
"sonst kann kein echter Himmelswinkel berechnet werden."
).pack(anchor="w", padx=8, pady=(4, 4))
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Button(r, text="Stern A markieren",
command=lambda: self._start_marking("ds_1")
).pack(side="left", expand=True, fill="x", padx=2)
ttk.Button(r, text="Stern B markieren",
command=lambda: self._start_marking("ds_2")
).pack(side="left", expand=True, fill="x", padx=2)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Label(r, text="Paar-Bezeichnung:").pack(side="left")
self.var_ds_name = tk.StringVar(value="")
ttk.Entry(r, textvariable=self.var_ds_name, width=16).pack(side="left", padx=4)
ttk.Button(box, text="PA/Abstand berechnen", command=self.compute_double_star
).pack(fill="x", padx=8, pady=4)
self.var_ds_result = tk.StringVar(value="A: - | B: -")
ttk.Label(box, textvariable=self.var_ds_result, wraplength=390,
justify="left").pack(anchor="w", padx=8, pady=(0, 4))
ttk.Button(box, text="Messreihe als CSV exportieren",
command=self.export_double_star_csv).pack(fill="x", padx=8, pady=(0, 6))
box = ttk.LabelFrame(t_astro, text="Asteroid / Komet (Positions-Astrometrie)")
box.pack(fill="x", **pad)
ttk.Label(box, wraplength=390, justify="left", foreground="#555555", text=
"Erzeugt korrekt spaltenformatierte MPC-80-Zeilen. Fuer "
"eine ECHTE Meldung ans Minor Planet Center brauchst du "
"zusaetzlich einen zugewiesenen Beobachtercode (erst nach "
"Kontakt mit dem MPC) und musst das Format vor dem "
"Absenden selbst gegenchecken - siehe 'Guide to Minor "
"Body Astrometry'."
).pack(anchor="w", padx=8, pady=(4, 4))
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Label(r, text="Bezeichnung (max 7 Zeichen):").pack(side="left")
self.var_ast_desig = tk.StringVar(value="")
ttk.Entry(r, textvariable=self.var_ast_desig, width=10).pack(side="left", padx=4)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Label(r, text="Geschaetzte Helligkeit (mag, optional):").pack(side="left")
self.var_ast_mag = tk.StringVar(value="")
ttk.Entry(r, textvariable=self.var_ast_mag, width=6).pack(side="left", padx=4)
ttk.Button(box, text="Position markieren & Zeile erzeugen",
command=lambda: self._start_marking("asteroid")
).pack(fill="x", padx=8, pady=4)
self.txt_ast = tk.Text(box, height=6, wrap="none", font=("Consolas", 9))
self.txt_ast.pack(fill="x", padx=8, pady=(0, 6))
self.txt_ast.insert("1.0", "noch keine Positionen erfasst")
self.txt_ast.configure(state="disabled")
box = ttk.LabelFrame(t_trans, text="Nova-/Transientensuche (Bildvergleich)")
box.pack(fill="x", **pad)
ttk.Label(box, wraplength=390, justify="left", foreground="#555555", text=
"Vergleicht das aktuelle geloeste Bild mit einem frueher "
"gespeicherten Referenzbild (gleiches Feld). Neue helle "
"Punkte, die vorher nicht da waren, werden als Kandidat "
"gelistet - kann auch Flugzeuge/Satelliten/Rauschen sein, "
"IMMER manuell pruefen, bevor du irgendwo etwas meldest."
).pack(anchor="w", padx=8, pady=(4, 4))
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=4)
ttk.Button(r, text="Aktuelles Bild als Referenz speichern",
command=self.save_reference_frame).pack(fill="x")
self.var_ref_status = tk.StringVar(value="keine Referenz gespeichert")
ttk.Label(box, textvariable=self.var_ref_status,
foreground="#333333").pack(anchor="w", padx=8, pady=2)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=2)
ttk.Label(r, text="Schwelle (Sigma ueber Rauschen):").pack(side="left")
self.var_trans_sigma = tk.DoubleVar(value=8.0)
ttk.Spinbox(r, from_=4, to=30, width=5,
textvariable=self.var_trans_sigma).pack(side="left", padx=4)
ttk.Button(box, text="Mit Referenz vergleichen",
command=self.compare_to_reference).pack(fill="x", padx=8, pady=4)
self.txt_trans = tk.Text(box, height=8, wrap="word", font=("Consolas", 9))
self.txt_trans.pack(fill="x", padx=8, pady=(0, 6))
self.txt_trans.insert("1.0", "noch kein Vergleich durchgefuehrt")
self.txt_trans.configure(state="disabled")
box = ttk.LabelFrame(t_save, text="Speichern")
box.pack(fill="x", **pad)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=4)
ttk.Button(r, text="Screenshot", command=self.snapshot
).pack(side="left", expand=True, fill="x", padx=2)
self.var_autosave = tk.BooleanVar(value=False)
ttk.Checkbutton(r, text="jedes Frame sichern",
variable=self.var_autosave,
command=self.choose_dir).pack(side="left", padx=6)
box = ttk.LabelFrame(t_save, text="Live-Fenster (das nimmt OBS auf)")
box.pack(fill="x", **pad)
r = ttk.Frame(box)
r.pack(fill="x", padx=8, pady=4)
ttk.Button(r, text="Anzeigen", command=self.view.show
).pack(side="left", expand=True, fill="x", padx=2)
ttk.Button(r, text="Vollbild (F11)", command=self.view.toggle_fullscreen
).pack(side="left", expand=True, fill="x", padx=2)
self.var_top = tk.BooleanVar(value=False)
ttk.Checkbutton(box, text="immer im Vordergrund",
variable=self.var_top,
command=self._topmost).pack(anchor="w", padx=8)
self.var_status = tk.StringVar(value="bereit")
ttk.Label(self, textvariable=self.var_status, foreground="#333333",
wraplength=440, justify="left").pack(
anchor="w", padx=12, pady=6)
self._mode_changed()
self._auto_toggled()
self.refresh_key()
def _slider(self, parent, label, var, lo, hi):
r = ttk.Frame(parent)
r.pack(fill="x", padx=8, pady=1)
ttk.Label(r, text=label, width=12).pack(side="left")
sc = ttk.Scale(r, from_=lo, to=hi, variable=var, orient="horizontal")
sc.pack(side="left", fill="x", expand=True)
return sc
def _auto_toggled(self):
state = "disabled" if self.var_auto.get() else "normal"
self.sc_black.state(["disabled"] if state == "disabled" else ["!disabled"])
self.sc_mid.state(["disabled"] if state == "disabled" else ["!disabled"])
def _mode_changed(self):
if self.var_mode.get() == "stack":
self.frm_stack.pack(fill="x", padx=20, pady=2)
self.frm_rtsp.forget()
else:
self.frm_rtsp.pack(fill="x", padx=20, pady=2)
self.frm_stack.forget()
def _topmost(self):
self.view.attributes("-topmost", self.var_top.get())
def refresh_key(self):
path, source = find_key()
if path:
if ss_auth is not None:
try:
ss_auth.set_key_path(path)
except Exception:
pass
self.var_key.set(f"OK - gefunden ueber {source}:\n{path}")
else:
self.var_key.set(
"Kein Schluessel gefunden. Bei Firmware 7.18 und neuer "
"lehnt der Seestar sonst jede Verbindung ab.\n"
f"Erwartet unter: {default_key_path()}")
return path
def pick_key(self):
p = filedialog.askopenfilename(
title="seestar.pem auswaehlen",
filetypes=[("PEM-Schluessel", "*.pem"), ("Alle Dateien", "*.*")])
if not p:
return
os.environ["SEESTAR_KEY_PATH"] = p
if ss_auth is not None:
try:
ss_auth.set_key_path(p)
except Exception as exc:
messagebox.showerror(APP_TITLE, f"Schluessel unbrauchbar: {exc}")
return
self.refresh_key()
self.var_status.set("Schluessel gesetzt und bei seestarpy registriert.")
def extract_key(self):
apk = filedialog.askopenfilename(
title="Seestar-APK auswaehlen",
filetypes=[("Android-App", "*.apk"), ("Alle Dateien", "*.*")])
if not apk:
return
target = default_key_path()
target.parent.mkdir(parents=True, exist_ok=True)
self.var_status.set("Entpacke APK, das kann einen Moment dauern ...")
self.update_idletasks()
ok, out = extract_key_from_apk(apk, target)
if ok and target.is_file():
self.refresh_key()
self.var_status.set(f"Schluessel geschrieben: {target}")
messagebox.showinfo(APP_TITLE, f"Fertig.\n\nSchluessel liegt unter:\n{target}")
else:
self.var_status.set("Extraktion fehlgeschlagen.")
messagebox.showerror(
APP_TITLE,
"Es konnte kein Schluessel aus der APK gelesen werden.\n\n"
"Ausgabe des Extraktors:\n\n" + (out or "(keine)"))
def discover(self):
if ss_conn is None:
messagebox.showerror(APP_TITLE, "seestarpy fehlt.\n\npip install seestarpy")
return
self.var_status.set("suche Seestar im Netzwerk ...")
self.update_idletasks()
try:
ss_conn.find_available_ips(1)
ips = getattr(ss_conn, "AVAILABLE_IPS", {}) or {}
ips = {k: v for k, v in ips.items() if valid_ip(str(v))}
if ips:
first = list(ips.values())[0]
self.var_ip.set(str(first))
self.var_status.set(f"gefunden: {first}")
else:
self.var_status.set("nichts gefunden - IP bitte manuell eintragen")
except Exception as exc:
self.var_status.set(f"Suche fehlgeschlagen: {exc}")
def start_source(self):
self.stop_source()
ip = self.var_ip.get().strip()
if self.var_mode.get() == "rtsp":
url = self.var_url.get().strip()
if not url:
if not ip:
messagebox.showwarning(APP_TITLE, "Bitte IP oder URL angeben.")
return
url = f"rtsp://{ip}:4554/stream"
self.var_url.set(url)
self.source = RTSPSource(url)
else:
if ss_stream is None:
messagebox.showerror(APP_TITLE,
"seestarpy fehlt.\n\npip install seestarpy")
return
if find_key()[0] is None:
go_on = messagebox.askyesno(
APP_TITLE,
"Es ist kein RSA-Schluessel hinterlegt.\n\n"
"Bei Firmware 7.18 und neuer verweigert der Seestar dann "
"die Verbindung. Bei aelterer Firmware ist das egal.\n\n"
"Trotzdem versuchen?")
if not go_on:
return
if not valid_ip(ip):
messagebox.showwarning(
APP_TITLE,
"Fuer den Live-Stack bitte eine vollstaendige IP angeben "
f"(aktuell: '{ip or 'leer'}').\n\n"
"Tipp: Knopf 'Suchen' benutzen oder in der Seestar-App "
"unter Stationsmodus nachsehen.")
return
self.source = StackSource(
ip=ip,
interval=float(self.var_interval.get()),
preview=self.var_preview.get(),
wide=self.var_wide.get(),
)
self.source.start()
self.btn_start.state(["disabled"])
self.btn_stop.state(["!disabled"])
self.view.show()
def stop_source(self):
if self.source is not None:
self.source.stop()
self.source = None
self.btn_start.state(["!disabled"])
self.btn_stop.state(["disabled"])
self.var_status.set("gestoppt")
def choose_dir(self):
if self.var_autosave.get():
d = filedialog.askdirectory(title="Ordner fuer Einzelbilder")
if d:
self._save_dir = d
self._saved_ids.clear()
else:
self.var_autosave.set(False)
def snapshot(self):
if self._last_pil is None:
messagebox.showinfo(APP_TITLE, "Noch kein Bild da.")
return
path = filedialog.asksaveasfilename(
defaultextension=".png",
filetypes=[("PNG", "*.png"), ("JPEG", "*.jpg")],
initialfile=time.strftime("seestar_%Y%m%d_%H%M%S.png"))
if path:
self._last_pil.save(path)
self.var_status.set(f"gespeichert: {os.path.basename(path)}")
def _pick_overlay_color(self):
rgb, hexcode = colorchooser.askcolor(
color=self.var_ov_color.get(), title="Textfarbe der Einblendung")
if hexcode:
self.var_ov_color.set(hexcode)
self.lbl_ov_color.configure(bg=hexcode)
def _pick_astap(self):
p = filedialog.askopenfilename(
title="astap_cli auswaehlen",
filetypes=[("ASTAP", "astap*"), ("Alle Dateien", "*.*")])
if p:
self.var_astap_path.set(p)
def setup_astap(self):
"""Laedt astap_cli automatisch herunter und richtet es ein. Der
Sternkatalog (die eigentliche Datenbank) kann aus Sicherheitsgruenden
nicht automatisch mitinstalliert werden (siehe download_and_extract_
astap-Docstring) - dafuer wird die offizielle Downloadseite geoeffnet."""
self.var_solve_status.set("Richte astap_cli ein ...")
threading.Thread(target=self._setup_astap_worker, daemon=True).start()
def _setup_astap_worker(self):
def progress(msg):
self.after(0, self.var_solve_status.set, msg)
try:
exe_path = download_and_extract_astap(progress_cb=progress)
except Exception as exc:
self.after(0, self._setup_astap_done, None, str(exc))
return
self.after(0, self._setup_astap_done, exe_path, None)
def _setup_astap_done(self, exe_path, error):
if error:
self.var_solve_status.set(f"Einrichtung fehlgeschlagen: {error}")
messagebox.showerror(
"ASTAP-Einrichtung",
f"Konnte astap_cli nicht automatisch einrichten:\n\n{error}\n\n"
"Alternativ manuell von www.hnsky.org/astap.htm laden und "
"den Pfad oben eintragen.")
return
self.var_astap_path.set(exe_path)
self.var_solve_status.set(
"astap_cli eingerichtet. Es fehlt noch der Sternkatalog "
"(die Datenbank) - ich oeffne dafuer gleich die Downloadseite.")
answer = messagebox.askyesno(
"ASTAP-Programm eingerichtet",
"Das astap_cli-Programm wurde automatisch heruntergeladen und "
f"eingerichtet:\n{exe_path}\n\n"
"Zum Loesen braucht ASTAP zusaetzlich einen Sternkatalog "
"(mehrere hundert MB, wird als Installer vertrieben). Diesen "
"einen Schritt kann ich aus Sicherheitsgruenden nicht "
"automatisch ausfuehren.\n\n"
"Soll ich jetzt die offizielle Downloadseite oeffnen? Dort "
"reicht 'D50' (guter Allrounder) oder 'D05' (kleiner, reicht "
"meist fuer das Seestar-Bildfeld) - Installer starten, "
"Standardpfad lassen, fertig.")
if answer:
open_astap_database_page()
def _auto_solve_toggled(self):
if self._auto_solve_job is not None:
self.after_cancel(self._auto_solve_job)
self._auto_solve_job = None
if self.var_auto_solve.get():
self._schedule_auto_solve()
def _schedule_auto_solve(self):
interval_ms = max(15, int(self.var_solve_interval.get())) * 1000
self._auto_solve_job = self.after(interval_ms, self._auto_solve_tick)
def _auto_solve_tick(self):
if self.var_auto_solve.get():
self.solve_now()
self._schedule_auto_solve()
def solve_now(self):
if self._solve_busy:
self.var_solve_status.set("laeuft bereits ...")
return
if self._last_pil is None:
self.var_solve_status.set("noch kein Bild vorhanden.")
return
exe = self.var_astap_path.get().strip()
if not exe or not Path(exe).is_file():
if self.var_auto_solve.get():
self.var_solve_status.set(
"ASTAP nicht gefunden - unten 'automatisch einrichten' klicken.")
return
if messagebox.askyesno(
"ASTAP nicht gefunden",
"Unter dem angegebenen Pfad wurde kein astap_cli gefunden.\n\n"
"Jetzt automatisch herunterladen und einrichten?"):
self.setup_astap()
return
try:
fov_txt = self.var_fov.get().strip()
fov_deg = (float(fov_txt) / 60.0) if fov_txt else None
except ValueError:
fov_deg = None
radius = int(self.var_search_radius.get())
snapshot_pil = self._last_pil.convert("RGB")
self._solve_busy = True
self.var_solve_status.set("loese Bild (ASTAP laeuft) ...")
threading.Thread(target=self._solve_worker,
args=(exe, snapshot_pil, radius, fov_deg),
daemon=True).start()
def _solve_worker(self, exe, pil_img, radius, fov_deg):
tmp_dir = tempfile.mkdtemp(prefix="seestar_solve_")
img_path = os.path.join(tmp_dir, "frame.png")
try:
pil_img.save(img_path)
result = run_astap_solve(exe, img_path, search_radius=radius,
fov_deg=fov_deg)
w, h = pil_img.size
fov_img_deg = max(w, h) * result["cdelt"] if result["cdelt"] else 1.0
matches = find_catalog_matches(result["ra"], result["dec"], fov_img_deg)
result["matches"] = matches
result["ts"] = time.time()
self.after(0, self._apply_solve_result, result, None)
except Exception as exc:
self.after(0, self._apply_solve_result, None, str(exc))
finally:
try:
for f in os.listdir(tmp_dir):
os.remove(os.path.join(tmp_dir, f))
os.rmdir(tmp_dir)
except OSError:
pass
def _apply_solve_result(self, result, error):
self._solve_busy = False
if error:
self.var_solve_status.set(f"Plate-Solving fehlgeschlagen: {error}")
return
ra_txt = deg_to_hms(result["ra"])
dec_txt = deg_to_dms(result["dec"])
matches = result["matches"]
if matches:
names = ", ".join(
(m["common"] or m["name"]) +
(f" ({m['name']})" if m["common"] and m["name"] != m["common"] else "")
for _, m in matches[:3])
self.var_solve_status.set(
f"Geloest: RA {ra_txt} Dec {dec_txt}\nGefunden: {names}")
else:
self.var_solve_status.set(
f"Geloest: RA {ra_txt} Dec {dec_txt}\n"
"Kein Katalogobjekt in der eingebauten Liste in der Naehe "
"(nur Messier + gaengige NGC/IC-Ziele enthalten).")
self._solve_result = result
def analyze_stars_now(self):
if self.source is None:
self.var_status.set("keine Bildquelle aktiv.")
return
arr, _meta = self.source.latest()
if arr is None:
self.var_status.set("noch kein Bild da.")
return
report = analyze_star_colors(arr, max_stars=15)
self._star_report = report
self._update_star_text(report)
def _update_star_text(self, report):
self.txt_stars.configure(state="normal")
self.txt_stars.delete("1.0", "end")
if not report:
self.txt_stars.insert("1.0", "keine hellen Sterne erkannt.")
else:
lines = [f"{i+1:2d}. ({s['x']:4d},{s['y']:4d}) {s['label']}"
for i, s in enumerate(report)]
header = ("Naeherung aus RGB-Farbverhaeltnis, KEINE echte "
"Spektralklassifikation:\n")
self.txt_stars.insert("1.0", header + "\n".join(lines))
self.txt_stars.configure(state="disabled")
def _start_marking(self, role):
"""Bewaffnet das Livefenster fuer einen einzelnen Marker-Klick.
role bestimmt, was mit der geklickten Position passiert."""
if self._last_pil is None:
self.var_status.set("noch kein Bild da zum Markieren.")
return
self.var_status.set("Klicke im Livebild auf den Stern ...")
self.view.request_click(lambda ix, iy: self._on_marked(role, ix, iy))
def _on_marked(self, role, ix, iy):
w, h = self._last_pil.size
if role in ("phot_target", "phot_comp"):
if self.var_flip.get():
raw_x, raw_y = w - 1 - ix, h - 1 - iy
else:
raw_x, raw_y = ix, iy
if role == "phot_target":
self._phot_target = (raw_x, raw_y)
else:
self._phot_comp = (raw_x, raw_y)
t = f"{self._phot_target[0]:.0f},{self._phot_target[1]:.0f}" if self._phot_target else "-"
c = f"{self._phot_comp[0]:.0f},{self._phot_comp[1]:.0f}" if self._phot_comp else "-"
self.var_phot_marks.set(f"Ziel: {t} | Vergleich: {c}")
self.var_status.set("Stern markiert.")
elif role in ("ds_1", "ds_2"):
if role == "ds_1":
self._ds_pt1 = (ix, iy)
else:
self._ds_pt2 = (ix, iy)
a = f"{self._ds_pt1[0]:.0f},{self._ds_pt1[1]:.0f}" if self._ds_pt1 else "-"
b = f"{self._ds_pt2[0]:.0f},{self._ds_pt2[1]:.0f}" if self._ds_pt2 else "-"
self.var_ds_result.set(f"A: {a} | B: {b}")
self.var_status.set("Stern markiert.")
elif role == "asteroid":
self._record_asteroid_position(ix, iy)
def fetch_distances(self):
if self._distance_busy:
return
if not self._solve_result:
messagebox.showinfo(
"Keine Loesung", "Bitte zuerst im Tab 'Objekte' das Bild "
"loesen (Plate-Solving), sonst ist die Himmelsposition "
"nicht bekannt.")
return
ra, dec = self._solve_result["ra"], self._solve_result["dec"]
radius_deg = float(self.var_dist_radius.get()) / 60.0
self._distance_busy = True
self._set_text(self.txt_dist, "frage SIMBAD/Gaia ab ...")
threading.Thread(target=self._fetch_distances_worker,
args=(ra, dec, radius_deg), daemon=True).start()
def _fetch_distances_worker(self, ra, dec, radius_deg):
try:
rows = query_simbad_distances(ra, dec, radius_deg, limit=15)
self.after(0, self._fetch_distances_done, rows, None)
except Exception as exc:
self.after(0, self._fetch_distances_done, None, str(exc))
def _fetch_distances_done(self, rows, error):
self._distance_busy = False
if error:
self._set_text(self.txt_dist, f"Abfrage fehlgeschlagen: {error}")
return
if not rows:
self._set_text(self.txt_dist,
"Keine Objekte mit bekannter Gaia-Parallaxe in "
"diesem Radius gefunden (bei sehr schwachen "
"Sternen hat Gaia oft keine brauchbare Parallaxe).")
return
lines = ["Katalogwerte (SIMBAD/Gaia), KEINE Eigenmessung:\n"]
for r in rows:
lines.append(
f"{r['name']:<18s} {r['distance_ly']:8.1f} Lj "
f"({r['distance_pc']:7.2f} pc, Plx {r['parallax_mas']:.2f} mas)")
self._set_text(self.txt_dist, "\n".join(lines))
def _set_text(self, widget, text):
widget.configure(state="normal")
widget.delete("1.0", "end")
widget.insert("1.0", text)
widget.configure(state="disabled")
def _phot_auto_toggled(self):
if self._phot_auto_job is not None:
self.after_cancel(self._phot_auto_job)
self._phot_auto_job = None
if self.var_phot_auto.get():
self._schedule_phot_auto()
def _schedule_phot_auto(self):
interval_ms = max(10, int(self.var_phot_interval.get())) * 1000
self._phot_auto_job = self.after(interval_ms, self._phot_auto_tick)
def _phot_auto_tick(self):
if self.var_phot_auto.get():
self.take_photometry_point()
self._schedule_phot_auto()
def take_photometry_point(self):
if self.source is None:
self.var_status.set("keine Bildquelle aktiv.")
return
if self._phot_target is None or self._phot_comp is None:
self.var_status.set("bitte erst Ziel- UND Vergleichsstern markieren.")
return
arr, meta = self.source.latest()
if arr is None:
self.var_status.set("noch kein Bild da.")
return
r_ap = float(self.var_ap_r.get())
r_in = float(self.var_ap_rin.get())
r_out = float(self.var_ap_rout.get())
flux_t, cx_t, cy_t = aperture_photometry(
arr, *self._phot_target, r_ap=r_ap, r_in=r_in, r_out=r_out)
flux_c, cx_c, cy_c = aperture_photometry(
arr, *self._phot_comp, r_ap=r_ap, r_in=r_in, r_out=r_out)
self._phot_target = (cx_t, cy_t)
self._phot_comp = (cx_c, cy_c)
if flux_t <= 0 or flux_c <= 0:
self.var_status.set("Photometrie: kein brauchbares Signal in der Apertur.")
return
delta_mag = -2.5 * math.log10(flux_t / flux_c)
entry = {
"t": time.time(), "flux_target": flux_t, "flux_comp": flux_c,
"delta_mag": delta_mag, "frame": meta.get("frame"),
}
self._phot_log.append(entry)
self._update_phot_text()
self.var_status.set(f"Messpunkt aufgenommen: \u0394m = {delta_mag:+.3f}")
def _update_phot_text(self):
if not self._phot_log:
self._set_text(self.txt_phot, "noch keine Messpunkte")
return
lines = ["Zeit Delta-mag (Ziel - Vergleich)"]
for e in self._phot_log[-40:]:
ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(e["t"]))
lines.append(f"{ts} {e['delta_mag']:+.3f}")
self._set_text(self.txt_phot, "\n".join(lines))
def export_photometry_csv(self):
if not self._phot_log:
self.var_status.set("keine Messpunkte zum Exportieren.")
return
path = filedialog.asksaveasfilename(
defaultextension=".csv",
filetypes=[("CSV", "*.csv")],
initialfile="lichtkurve.csv")
if not path:
return
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow(["unix_zeit", "iso_zeit_lokal", "delta_mag",
"fluss_ziel", "fluss_vergleich", "frame"])
for e in self._phot_log:
w.writerow([
f"{e['t']:.3f}",
time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(e["t"])),
f"{e['delta_mag']:.4f}", f"{e['flux_target']:.2f}",
f"{e['flux_comp']:.2f}", e.get("frame", "")])
self.var_status.set(f"Lichtkurve exportiert: {os.path.basename(path)}")
def show_light_curve(self):
if len(self._phot_log) < 2:
self.var_status.set("mindestens 2 Messpunkte fuer eine Kurve noetig.")
return
LightCurveWindow(self, self._phot_log)
def compute_double_star(self):
if not self._solve_result:
messagebox.showinfo(
"Keine Loesung", "Bitte zuerst im Tab 'Objekte' das Bild "
"loesen (Plate-Solving).")
return
if self._ds_pt1 is None or self._ds_pt2 is None:
self.var_status.set("bitte beide Sterne markieren.")
return
wcs = self._solve_result
try:
ra1, dec1 = wcs_pixel_to_radec(self._ds_pt1[0], self._ds_pt1[1], wcs)
ra2, dec2 = wcs_pixel_to_radec(self._ds_pt2[0], self._ds_pt2[1], wcs)
except Exception as exc:
self.var_status.set(f"Umrechnung fehlgeschlagen: {exc}")
return
pa, sep = position_angle_and_separation(ra1, dec1, ra2, dec2)
name = self.var_ds_name.get().strip() or "(unbenannt)"
entry = {"t": time.time(), "name": name, "pa_deg": pa, "sep_arcsec": sep,
"ra1": ra1, "dec1": dec1, "ra2": ra2, "dec2": dec2}
self._ds_log.append(entry)
self.var_ds_result.set(
f"{name}: PA = {pa:.1f}\u00b0 (Nord->Ost), Abstand = {sep:.2f}\"")
def export_double_star_csv(self):
if not self._ds_log:
self.var_status.set("keine Doppelstern-Messungen zum Exportieren.")
return
path = filedialog.asksaveasfilename(
defaultextension=".csv", filetypes=[("CSV", "*.csv")],
initialfile="doppelsterne.csv")
if not path:
return
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow(["iso_zeit_lokal", "paar", "pa_grad", "abstand_bogensek",
"ra1_grad", "dec1_grad", "ra2_grad", "dec2_grad"])
for e in self._ds_log:
w.writerow([
time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(e["t"])),
e["name"], f"{e['pa_deg']:.2f}", f"{e['sep_arcsec']:.3f}",
f"{e['ra1']:.6f}", f"{e['dec1']:.6f}",
f"{e['ra2']:.6f}", f"{e['dec2']:.6f}"])
self.var_status.set(f"Doppelstern-Messreihe exportiert: {os.path.basename(path)}")
def _record_asteroid_position(self, ix, iy):
if not self._solve_result:
messagebox.showinfo(
"Keine Loesung", "Bitte zuerst im Tab 'Objekte' das Bild "
"loesen (Plate-Solving), sonst keine Himmelskoordinaten.")
return
wcs = self._solve_result
try:
ra, dec = wcs_pixel_to_radec(ix, iy, wcs)
except Exception as exc:
self.var_status.set(f"Umrechnung fehlgeschlagen: {exc}")
return
desig = self.var_ast_desig.get().strip() or "TEMP001"
mag = None
try:
mag_txt = self.var_ast_mag.get().strip()
mag = float(mag_txt) if mag_txt else None
except ValueError:
mag = None
now_utc = datetime.now(timezone.utc)
line = format_mpc80_line(desig, now_utc, ra, dec, mag=mag, band="V")
self._ast_log.append(line)
self._update_ast_text()
self.var_status.set("Position erfasst (MPC-Zeile erzeugt).")
def _update_ast_text(self):
header = ("MPC-80-Spalten-Format - Beobachtercode (Spalten 78-80) "
"fehlt bewusst, siehe Hinweis oben:\n")
self._set_text(self.txt_ast, header + "\n".join(self._ast_log))
def save_reference_frame(self):
if self.source is None:
self.var_status.set("keine Bildquelle aktiv.")
return
if not self._solve_result:
messagebox.showinfo(
"Keine Loesung", "Bitte zuerst im Tab 'Objekte' das aktuelle "
"Bild loesen, damit die Referenz eine bekannte Himmelsposition "
"hat (fuer die spaetere Ausrichtung noetig).")
return
arr, meta = self.source.latest()
if arr is None:
self.var_status.set("noch kein Bild da.")
return
self._ref_frame = {
"arr": arr.copy(), "wcs": dict(self._solve_result),
"ts": time.time(),
}
when = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self._ref_frame["ts"]))
self.var_ref_status.set(f"Referenz gespeichert: {when}")
self.var_status.set("Referenzbild gespeichert.")
def compare_to_reference(self):
if self._ref_frame is None:
self.var_status.set("bitte erst eine Referenz speichern.")
return
if not self._solve_result:
messagebox.showinfo(
"Keine Loesung", "Bitte zuerst im Tab 'Objekte' das aktuelle "
"Bild loesen (Plate-Solving).")
return
if self.source is None:
self.var_status.set("keine Bildquelle aktiv.")
return
arr, meta = self.source.latest()
if arr is None:
self.var_status.set("noch kein Bild da.")
return
self._set_text(self.txt_trans, "vergleiche ...")
sigma = float(self.var_trans_sigma.get())
threading.Thread(target=self._compare_worker, args=(arr, sigma),
daemon=True).start()
def _compare_worker(self, cur_arr, sigma):
try:
cur_wcs = self._solve_result
ref = self._ref_frame
ref_arr, ref_wcs = ref["arr"], ref["wcs"]
cur_lum = cur_arr.mean(axis=2) if cur_arr.ndim == 3 else cur_arr
ref_lum = ref_arr.mean(axis=2) if ref_arr.ndim == 3 else ref_arr
h, w = cur_lum.shape
step = max(1, min(h, w) // 120)
gy, gx = np.mgrid[0:h:step, 0:w:step].astype(float)
ref_x = np.full(gx.shape, np.nan)
ref_y = np.full(gx.shape, np.nan)
for j in range(gx.shape[0]):
for i in range(gx.shape[1]):
ra, dec = wcs_pixel_to_radec(gx[j, i], gy[j, i], cur_wcs)
try:
rx, ry = wcs_radec_to_pixel(ra, dec, ref_wcs)
ref_x[j, i] = rx
ref_y[j, i] = ry
except ValueError:
pass
rx_img = Image.fromarray(np.nan_to_num(ref_x, nan=-1).astype(np.float32))
ry_img = Image.fromarray(np.nan_to_num(ref_y, nan=-1).astype(np.float32))
rx_full = np.array(rx_img.resize((w, h), Image.BILINEAR))
ry_full = np.array(ry_img.resize((w, h), Image.BILINEAR))
valid = (rx_full >= 0) & (ry_full >= 0) & \
(rx_full < ref_lum.shape[1]) & (ry_full < ref_lum.shape[0])
ix = np.clip(rx_full.astype(int), 0, ref_lum.shape[1] - 1)
iy = np.clip(ry_full.astype(int), 0, ref_lum.shape[0] - 1)
resampled_ref = np.zeros_like(cur_lum)
resampled_ref[valid] = ref_lum[iy[valid], ix[valid]]
diff = cur_lum.astype(np.float32) - resampled_ref.astype(np.float32)
diff[~valid] = 0.0
med = float(np.median(diff))
mad = float(np.median(np.abs(diff - med))) * 1.4826 or 1e-4
thresh = med + sigma * mad
ys, xs = np.where(diff > thresh)
candidates = []
if len(xs) > 0:
order = np.argsort(-diff[ys, xs])
ys, xs = ys[order], xs[order]
picked = []
for y, x in zip(ys, xs):
if len(picked) >= 15:
break
if any((py - y) ** 2 + (px - x) ** 2 < 10 ** 2 for py, px in picked):
continue
picked.append((y, x))
for y, x in picked:
ra, dec = wcs_pixel_to_radec(float(x), float(y), cur_wcs)
known = find_catalog_matches(ra, dec, 0.05, limit=1)
tag = "bekanntes Katalogobjekt in der Naehe" if known else "kein Katalogtreffer"
candidates.append({
"x": int(x), "y": int(y), "ra": ra, "dec": dec, "tag": tag,
})
self.after(0, self._compare_done, candidates, None)
except Exception as exc:
self.after(0, self._compare_done, None, str(exc))
def _compare_done(self, candidates, error):
if error:
self._set_text(self.txt_trans, f"Vergleich fehlgeschlagen: {error}")
return
self._transient_candidates = candidates or []
if not candidates:
self._set_text(self.txt_trans,
"Keine auffaelligen neuen Punkte gefunden.")
return
lines = [f"{len(candidates)} Kandidat(en) - IMMER manuell pruefen "
"(Flugzeuge/Satelliten/Rauschen moeglich):\n"]
for c in candidates:
lines.append(f"px({c['x']},{c['y']}) RA {deg_to_hms(c['ra'])} "
f"Dec {deg_to_dms(c['dec'])} - {c['tag']}")
self._set_text(self.txt_trans, "\n".join(lines))
def _tick(self):
try:
self._render()
except Exception as exc:
self.var_status.set(f"Anzeigefehler: {exc}")
self.after(40, self._tick)
def _render(self):
if self.source is None:
return
arr, meta = self.source.latest()
if self.source.error:
self.var_status.set(f"{self.source.status} - {self.source.error}")
else:
self.var_status.set(f"Status: {self.source.status}")
if arr is None:
self.view.clear("warte auf Bild ...")
return
work = arr
if self.var_mode.get() == "stack" or not self.var_auto.get():
if self.var_auto.get():
work, blk, mid = autostretch(work)
self.var_black.set(round(blk, 4))
self.var_mid.set(round(mid, 4))
else:
work = manual_stretch(work, float(self.var_black.get()),
float(self.var_mid.get()))
if self.var_gray.get() and work.ndim == 3:
work = work.mean(axis=2)
pil = to_pil(work)
if self.var_flip.get():
pil = pil.rotate(180)
if self.var_mark_stars.get() and self._star_report:
pil = self._mark_stars(pil)
if self.var_overlay.get():
pil = self._overlay(pil, meta)
self._last_pil = pil
self.view.draw(pil)
if self.var_autosave.get() and self._save_dir:
fid = meta.get("frame")
if fid is not None and fid not in self._saved_ids:
self._saved_ids.add(fid)
name = os.path.join(self._save_dir, f"frame_{fid:05d}.png")
try:
pil.save(name)
except Exception:
pass
def _mark_stars(self, pil):
pil = pil.copy()
draw = ImageDraw.Draw(pil)
for s in self._star_report:
x, y = s["x"], s["y"]
draw.ellipse((x - 8, y - 8, x + 8, y + 8), outline=(255, 210, 0), width=1)
return pil
def _overlay(self, pil, meta):
pil = pil.copy()
draw = ImageDraw.Draw(pil)
size = self.var_ov_size.get() or max(14, pil.size[1] // 40)
try:
font = ImageFont.truetype("segoeui.ttf", size)
except Exception:
font = ImageFont.load_default()
bits = []
if self.var_show_title.get():
cap = self.var_caption.get().strip()
if cap:
bits.append(cap)
if self.var_show_source.get() and meta.get("quelle"):
bits.append(meta["quelle"])
if self.var_show_res.get() and meta.get("breite"):
bits.append(f"{meta['breite']}x{meta['hoehe']}")
if self.var_show_frame.get() and meta.get("frame") is not None:
bits.append(f"Frame {meta['frame']}")
if self.var_show_hfd.get() and meta.get("hfd") is not None:
try:
bits.append(f"HFD {float(meta['hfd']):.2f}")
except (TypeError, ValueError):
pass
if self.var_show_fps.get() and meta.get("fps"):
bits.append(f"{meta['fps']:.1f} fps")
if self.var_show_object.get() and self._solve_result:
matches = self._solve_result.get("matches") or []
if matches:
_, obj = matches[0]
bits.append(obj["common"] or obj["name"])
else:
bits.append(f"RA {deg_to_hms(self._solve_result['ra'])} "
f"Dec {deg_to_dms(self._solve_result['dec'])}")
if self.var_show_date.get():
bits.append(time.strftime("%d.%m.%Y"))
if self.var_show_time.get():
bits.append(time.strftime("%H:%M:%S"))
text = " | ".join(bits)
if not text:
return pil
try:
color = ImageColor.getrgb(self.var_ov_color.get())
except ValueError:
color = (255, 255, 255)
bbox = draw.textbbox((0, 0), text, font=font)
tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1]
pw, ph = pil.size
margin = size // 2
pos = self.var_ov_pos.get()
if pos == "oben links":
x, y = margin, margin
elif pos == "oben rechts":
x, y = pw - tw - margin, margin
elif pos == "unten rechts":
x, y = pw - tw - margin, ph - th - margin
else:
x, y = margin, ph - th - margin
for dx in (-2, 0, 2):
for dy in (-2, 0, 2):
draw.text((x + dx, y + dy), text, font=font, fill=(0, 0, 0))
draw.text((x, y), text, font=font, fill=color)
return pil
def on_close(self):
if self._auto_solve_job is not None:
self.after_cancel(self._auto_solve_job)
self._auto_solve_job = None
self.stop_source()
self.destroy()
def main():
missing = []
if cv2 is None:
missing.append("opencv-python (fuer RTSP)")
if ss_stream is None:
missing.append("seestarpy (fuer Live-Stack)")
app = ControlApp()
if missing:
app.var_status.set("Fehlende Pakete: " + ", ".join(missing))
app.mainloop()
if __name__ == "__main__":
main()// hinzugefügt am 12.09.2026