grossanzeige/mount_mock.py

#!/usr/bin/env python3
"""Mock der 10Micron GM4000 HPS II -- ein TCP-Server, der LX200-Abfragen beantwortet.

Zum Entwickeln ohne echtes Teleskop. Antwortet auf die Koordinaten- und
Versions-Abfragen; unbekannte Kommandos werden mit "#" quittiert, damit der
Client nicht haengt.

    python3 mount_mock.py                     # feste Koordinaten, Port 3490
    python3 mount_mock.py --ra 18:36:56 --dec +38:47:01
    python3 mount_mock.py --move              # RA laeuft (Sekunden ticken hoch)
    python3 mount_mock.py --lan --move        # auch vom ESP32 aus erreichbar

Vorgabe ist 127.0.0.1, also nur lokal. Damit der ESP32 den Mock uebers WLAN
erreicht, braucht es --lan (bindet auf 0.0.0.0); die Adresse, die dann auf dem
ESP32 einzutragen ist, wird beim Start ausgegeben.

Nur fuer die Entwicklung (CPython). Auf dem ESP32 laeuft nur der Client.
"""

import argparse
import socket
import sys
import threading
import time

import lx200

PRODUCT = "10micron GM4000HPS"
FIRMWARE = "2.15.11"


def _parse_ra(text):
    """RA-String (HH:MM:SS o.ae.) in Sekunden 0..86399."""
    nums = _numbers(text)
    h, m, s = (nums + [0, 0, 0])[:3]
    return (int(h) * 3600 + int(m) * 60 + int(s)) % 86400


def _parse_dec(text):
    """DEC-String (sDD*MM:SS) in vorzeichenbehaftete Bogensekunden."""
    neg = text.strip().startswith("-")
    nums = _numbers(text)
    d, m, s = (nums + [0, 0, 0])[:3]
    total = int(d) * 3600 + int(m) * 60 + int(s)
    return -total if neg else total


def _numbers(text):
    out, cur = [], ""
    for ch in text:
        if ch.isdigit():
            cur += ch
        elif cur:
            out.append(cur)
            cur = ""
    if cur:
        out.append(cur)
    return [int(x) for x in out]


def _fmt_ra(total_s, high=True):
    """Rektaszension wie die GM4000.

    high=True  -> HH:MM:SS   (Hochpraezision, nach ":U#")
    high=False -> HH:MM.T    (Zehntelminuten -- so liefert die Montierung im
                              Auslieferzustand, verifiziert 2026-07-18)
    """
    total_s %= 86400
    h, rest = divmod(total_s, 3600)
    m, s = divmod(rest, 60)
    if high:
        return "%02d:%02d:%02d" % (h, m, s)
    zehntel = (total_s * 10 + 30) // 60         # in Zehntelminuten, gerundet
    minuten, t = divmod(zehntel, 10)
    h, m = divmod(minuten, 60)
    return "%02d:%02d.%d" % (h % 24, m, t)


def _fmt_dec(total_arcsec, high=True):
    """Deklination wie die GM4000.

    high=True  -> sDD*MM:SS  (Hochpraezision)
    high=False -> sDD*MM     (ganze Bogenminuten, Auslieferzustand)

    Gradzeichen als '*', Minuten/Sekunden mit ':' -- der coords-Parser kommt mit
    beiden Trennzeichen klar. (Die echte Montierung sendet 0xDF als Gradzeichen.)
    """
    sign = "-" if total_arcsec < 0 else "+"
    total_arcsec = abs(total_arcsec)
    d, rest = divmod(total_arcsec, 3600)
    m, s = divmod(rest, 60)
    if high:
        return "%s%02d*%02d:%02d" % (sign, d, m, s)
    return "%s%02d*%02d" % (sign, d, m)


class MountMock:
    """Simuliert die Montierung auf TCP-Ebene."""

    def __init__(self, host="127.0.0.1", port=3490,
                 ra="18:36:56", dec="+38:47:01", move=False, high=True):
        self._ra0 = _parse_ra(ra)
        self._dec = _parse_dec(dec)
        self._move = move
        # high=False bildet den Auslieferzustand der GM4000 nach: keine Sekunden,
        # RA als Zehntelminuten. Vorgabe bleibt True, damit bestehende Tests mit
        # ihren festen Erwartungen unveraendert durchlaufen.
        self._high = high
        self._tick = 0                 # zaehlt bei --move pro RA-Abfrage hoch
        self._dectick = 0              # zaehlt bei --move pro DEC-Abfrage hoch
        self._lock = threading.Lock()
        self._srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self._srv.bind((host, port))
        self._srv.listen(5)
        self.host, self.port = self._srv.getsockname()  # echter Port (bei port=0)
        self._running = False
        self._thread = None

    # -- aktuelle Koordinaten --------------------------------------------
    def _current_ra(self):
        """RA liefern; bei --move pro Abfrage genau 1 Sekunde weiter (kein Zeit-Jitter)."""
        if not self._move:
            return self._ra0
        with self._lock:
            ra = self._ra0 + self._tick
            self._tick += 1
        return ra

    def _current_dec(self):
        """DEC liefern; bei --move pro Abfrage 1 Bogensekunde weiter."""
        if not self._move:
            return self._dec
        with self._lock:
            dec = self._dec + self._dectick
            self._dectick += 1
        return dec

    def _reply_for(self, cmd):
        """Antwort (inkl. '#') fuer ein Kommando ohne ':' und ohne '#'."""
        if cmd == "GR":
            return _fmt_ra(self._current_ra(), self._high) + "#"
        if cmd == "GD":
            return _fmt_dec(self._current_dec(), self._high) + "#"
        if cmd == "GVP":
            return PRODUCT + "#"
        if cmd == "GVN":
            return FIRMWARE + "#"
        # Unbekannt: leere, aber abgeschlossene Antwort, damit der Client weiterlaeuft.
        return "#"

    def _handle(self, conn):
        buf = ""
        try:
            while self._running:
                data = conn.recv(64)
                if not data:
                    break
                buf += data.decode("latin-1")
                # vollstaendige Kommandos (durch '#' getrennt) abarbeiten
                while "#" in buf:
                    raw, buf = buf.split("#", 1)
                    cmd = raw.lstrip(":").strip()
                    if cmd:
                        conn.sendall(self._reply_for(cmd).encode("latin-1"))
        except OSError:
            pass
        finally:
            conn.close()

    def _accept_loop(self):
        while self._running:
            try:
                conn, _ = self._srv.accept()
            except OSError:
                break
            threading.Thread(target=self._handle, args=(conn,), daemon=True).start()

    def start(self):
        """Server im Hintergrund-Thread starten."""
        self._running = True
        self._thread = threading.Thread(target=self._accept_loop, daemon=True)
        self._thread.start()

    def stop(self):
        self._running = False
        try:
            self._srv.close()
        except OSError:
            pass

    def serve_forever(self):
        self.start()
        try:
            while True:
                time.sleep(0.5)
        except KeyboardInterrupt:
            pass
        finally:
            self.stop()


def lan_address():
    """Die eigene LAN-Adresse ermitteln -- die, die der ESP32 ansprechen muss.

    Ueber einen UDP-Socket ans Gateway: der verschickt nichts, verraet aber,
    welche lokale Adresse das Betriebssystem fuer den Weg nach draussen nimmt.
    """
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect(("192.168.1.1", 1))
        return s.getsockname()[0]
    except OSError:
        return None
    finally:
        s.close()


def main(argv=None):
    p = argparse.ArgumentParser(description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--host", default="127.0.0.1")
    p.add_argument("--lan", action="store_true",
                   help="auf allen Schnittstellen lauschen (0.0.0.0), damit der "
                        "ESP32 den Mock uebers WLAN erreicht")
    p.add_argument("--port", type=int, default=3490)
    p.add_argument("--ra", default="18:36:56")
    p.add_argument("--dec", default="+38:47:01")
    p.add_argument("--move", action="store_true", help="RA laeuft mit der Zeit hoch")
    p.add_argument("--low", action="store_true",
                   help="Niedrigpraezision wie im Auslieferzustand der GM4000: "
                        "RA als HH:MM.T, DEC ohne Bogensekunden")
    args = p.parse_args(argv)

    host = "0.0.0.0" if args.lan else args.host
    mock = MountMock(host=host, port=args.port, ra=args.ra, dec=args.dec,
                     move=args.move, high=not args.low)
    print("Mock-Montierung auf %s:%d  (RA=%s DEC=%s%s, %s)"
          % (mock.host, mock.port, args.ra, args.dec,
             ", move" if args.move else "",
             "Niedrigpraezision" if args.low else "Hochpraezision"))
    if host == "0.0.0.0":
        ip = lan_address()
        if ip:
            print("Vom ESP32 aus erreichbar unter:  %s:%d" % (ip, mock.port))
            print("  run_esp32.main(host=\"%s\", fake_display=True)" % ip)
        print("Falls keine Verbindung zustande kommt: macOS fragt beim ersten Mal,")
        print("ob python3 eingehende Verbindungen annehmen darf -- das muss erlaubt sein.")
    print("Beenden mit Ctrl-C.")
    mock.serve_forever()
    return 0


if __name__ == "__main__":
    sys.exit(main())