89 lines
2.7 KiB
Python
89 lines
2.7 KiB
Python
"""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.
|
|
"""
|
|
|
|
import socket
|
|
|
|
import lx200
|
|
|
|
|
|
class MountError(Exception):
|
|
"""Verbindungs- oder Kommunikationsfehler mit der Montierung."""
|
|
|
|
|
|
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()
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(self.timeout)
|
|
try:
|
|
sock.connect((self.host, self.port))
|
|
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")
|
|
try:
|
|
self._sock.sendall(framed_cmd.encode("latin-1"))
|
|
except OSError as e:
|
|
raise MountError("Senden gestoert: %s" % e)
|
|
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
|