grossanzeige/mount_client.py

"""TCP-Client zur 10Micron-Montierung (LX200 ueber Ethernet).

Die Socket-API (socket.socket/connect/send/recv) ist unter CPython und
MicroPython gleich, laeuft also spaeter auch auf dem ESP32-C3.

Welche Adresse abgefragt wird, klaert resolve_host(): per MQTT umstellbar,
mit config.MOUNT_HOST als Vorgabe.
"""

import socket

import config
import lx200
import settings


class MountError(Exception):
    """Verbindungs- oder Kommunikationsfehler mit der Montierung."""


def resolve_host(vorgabe=None, log=None):
    """Die Adresse liefern, die abgefragt werden soll.

    Reihenfolge, absteigend:

    1. ``vorgabe`` -- von Hand mitgegeben (REPL, Kommandozeile, Tests). Wer die
       Adresse ausdruecklich nennt, meint sie auch.
    2. ``mount_host`` aus settings.json -- per MQTT gesetzt und ueber einen
       Stromausfall hinweg gespeichert.
    3. ``config.MOUNT_HOST`` -- was beim Deployen im Code stand.

    So laesst sich zwischen Mock und echter Montierung umschalten, ohne etwas
    neu aufzuspielen, und ein leerer mount_host holt die Vorgabe aus dem Code
    zurueck.
    """
    if vorgabe:
        return vorgabe
    return settings.load(log=log).get("mount_host") or config.MOUNT_HOST


class MountClient:
    """Fragt Koordinaten von der Montierung ab.

    Verbindung wird nicht automatisch gehalten -- bei einem Fehler wirft die
    Abfrage MountError; der Aufrufer entscheidet ueber Reconnect (siehe
    run_display.py).
    """

    def __init__(self, host, port=3490, timeout=3.0):
        self.host = host
        self.port = port
        self.timeout = timeout
        self._sock = None

    def connect(self):
        self.close()
        # Ueber getaddrinfo statt direkt mit (host, port): MicroPython kann ein
        # Adresstupel mit Hostnamen nicht selbst aufloesen, mit der aufgeloesten
        # Adresse laeuft connect() auf beiden Seiten gleich.
        try:
            addr = socket.getaddrinfo(self.host, self.port)[0][-1]
        except (OSError, IndexError) as e:
            raise MountError("Adresse %s:%d nicht aufloesbar: %s"
                             % (self.host, self.port, e))
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(self.timeout)
        try:
            sock.connect(addr)
        except OSError as e:
            sock.close()
            raise MountError("Verbindung zu %s:%d fehlgeschlagen: %s"
                             % (self.host, self.port, e))
        self._sock = sock

    @property
    def connected(self):
        return self._sock is not None

    def _recv_until_terminator(self):
        buf = bytearray()
        while True:
            try:
                chunk = self._sock.recv(64)
            except OSError as e:
                raise MountError("Empfang gestoert: %s" % e)
            if not chunk:
                raise MountError("Verbindung von der Montierung geschlossen")
            buf.extend(chunk)
            if buf.endswith(lx200.TERMINATOR.encode("latin-1")):
                return bytes(buf)

    def query(self, framed_cmd):
        """Ein gerahmtes Kommando senden und die Antwort ohne '#' zurueckgeben."""
        if not self.connected:
            raise MountError("nicht verbunden")
        # Von Hand schleifen statt sendall(): MicroPython laesst dessen Verhalten
        # auf Sockets mit Timeout ausdruecklich offen. Die Kommandos sind vier
        # Bytes lang, die Schleife laeuft also praktisch immer nur einmal.
        data = framed_cmd.encode("latin-1")
        sent = 0
        while sent < len(data):
            try:
                n = self._sock.send(data[sent:])
            except OSError as e:
                raise MountError("Senden gestoert: %s" % e)
            if not n:
                raise MountError("Verbindung von der Montierung geschlossen")
            sent += n
        raw = self._recv_until_terminator().decode("latin-1")
        return lx200.strip_terminator(raw)

    def get_ra(self):
        """Rektaszension als Rohstring, z.B. '18:36:56'."""
        return self.query(lx200.GET_RA)

    def get_dec(self):
        """Deklination als Rohstring, z.B. '+38*47:01'."""
        return self.query(lx200.GET_DEC)

    def get_coordinates(self):
        """(ra, dec) als Rohstrings in einem Rutsch."""
        return self.get_ra(), self.get_dec()

    def close(self):
        if self._sock is not None:
            try:
                self._sock.close()
            except OSError:
                pass
            self._sock = None