Skip to content

GitLab

Commit 77df4636 authored by Tammo Jan Dijkema's avatar Tammo Jan Dijkema
Browse files

Move code to telescope.py

parent 440f64c3
Hide whitespace changes
Inline Side-by-side
__init__.py
View file @ 77df4636
from __future__ import print_function
import socket
import logging
from astropy.coordinates import SkyCoord, AltAz, RangeError
from astropy import units as u
from configparser import ConfigParser
import os.path
import numpy as np
import threading
import select
import time
config = ConfigParser()
config.readfp(open(os.path.join(os.path.dirname(__file__), 'telescope.ini')))
logging.basicConfig(level=logging.DEBUG)
class telescope():
def __init__(self, setmode=None, consoleHost=None, trace2portHost=None):
'''
Initializes a telescope instance. The mode for writing can be
'J2000' or 'AZEL'.
'''
if setmode not in ('J2000', 'AZEL', None):
raise ValueError("Mode must be None, 'J2000' or 'AZEL', not " + setmode)
self.setmode = setmode
if consoleHost is None:
consoleHost = config.get('Console','HostName')
if consoleHost == 'console':
if socket.gethostname()=="mercurius":
logging.warning("You are using the actual console, not a demo!")
else:
raise ValueError("Talking to the actual console can only be done from mercurius")
if trace2portHost is None:
trace2portHost = config.get('Trace2Port','HostName')
self.dist_az = None
self.dist_el = None
self.az = None
self.el = None
self.radec = None
self.speed_az = None
self.speed_el = None
self._outsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if self.setmode == 'J2000':
try:
self._outsocket.connect((consoleHost,
config.getint('Console','Port_Write_J2000')))
except socket.gaierror as e:
raise IOError("Could not connect to DT J2000 port: " + str(e))
elif self.setmode == 'AZEL':
try:
self._outsocket.connect((consoleHost,
config.getint('Console','Port_Write_AzEl')))
except socket.gaierror as e:
raise IOError("Could not connect to DT AZEL port: " + str(e))
# Initialize offset socket for all modes where output is desired
if self.setmode is not None:
self._offsetsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
self._offsetsocket.connect((consoleHost,
config.getint('Console','Port_Write_Offset')))
except socket.gaierror as e:
raise IOError("Could not connect to DT Offset port: " + str(e))
self._event_j2000 = threading.Event()
thread_read_j2000 = threading.Thread(target=self._readj2000, args=(
consoleHost,
config.getint('Console','Port_Read_J2000')))
thread_read_j2000.daemon = True
thread_read_j2000.start()
self._event_traces = threading.Event()
thread_read_traces = threading.Thread(target=self._readtraces, args=(
config.get('Trace2Port', 'Hostname'),
config.getint('Trace2Port', 'Port')))
thread_read_traces.daemon = True
thread_read_traces.start()
def _readj2000(self, consolehost, j2000_read_port):
'''
Poll the socket with the j2000 info, store its values in class members.
'''
try:
self._j2000socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._j2000socket.connect((consolehost, j2000_read_port))
except socket.error:
raise IOError("Cannot connect to j2000 status server")
while True:
readable, writable, exceptional = select.select(
[self._j2000socket], [], [])
if len(exceptional)>0:
raise IOError("Error with j2000 read socket")
msg = readable[0].recv(4096).decode("UTF-8")
vals = msg.strip().split()
(ra, dec) = (vals[2], vals[3])
self.radec = SkyCoord(ra, dec, frame='icrs')
(setpoint_ra, setpoint_dec) = (vals[0], vals[1])
try:
# The setpoint can be invalid
self.setpoint_radec = SkyCoord(setpoint_ra, setpoint_dec, frame='icrs')
except RangeError:
self.setpoint_radec = None
self._event_j2000.set()
def _readtraces(self, tracehost, traceport):
'''
Poll the socket with the traces, store their values in class members
'''
try:
self._tracesocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._tracesocket.connect((tracehost, traceport))
except socket.error:
raise IOError("Cannot connect to trace2port server")
while True:
readable, writable, exceptional = select.select(
[self._tracesocket], [], [])
if len(exceptional)>0:
raise IOError("Error with trace read socket")
msg = readable[0].recv(4096).decode("UTF-8")
vals = msg.strip().split()
self.az = (float(vals[0])*u.rad).to(u.deg)
self.el = (float(vals[1])*u.rad).to(u.deg)
self.dist_az = (float(vals[2])*u.rad).to(u.deg)
self.dist_el = (float(vals[3])*u.rad).to(u.deg)
self.speed_az = float(vals[4])
self.speed_el = float(vals[5])
self._event_traces.set()
def setRaDec(self, setpoint):
'''
Set the J2000 setpoint of the telescope.
Only works if the setmode is 'J2000'.
setpoint can be an astropy SkyCoord or a tuple (ra,dec) in radians
'''
if self.setmode != 'J2000':
raise ValueError("Cannot set Ra/Dec if mode is not J2000")
if isinstance(setpoint, tuple):
coord = SkyCoord(ra =setpoint[0] * u.radian,
dec=setpoint[1] * u.radian)
(ra, dec) = setpoint.to_string('hmsdms').split()
cmd = bytes('{:s}\t{:s}\n'.format(ra, dec.strip('+')), 'UTF-8')
logging.info('J2000 setpoint sent to DT: {}'.format(cmd))
self._outsocket.send(cmd)
def setAzEl(self, setpoint):
'''
Set the AzEl setpoint of the telescope.
Only works if the setmode is 'AZEL'
setpoint should be a tuple of degrees or of astropy Quantity
'''
if self.setmode != 'AZEL':
raise ValueError("Cannot set Az/El if mode is not AZEL")
(az, el) = setpoint
if isinstance(az, u.Quantity):
az = az.to(u.deg).value
if isinstance(el, u.Quantity):
el = el.to(u.deg).value
cmd = bytes('{:f}\t{:f}\n'.format(az, el), 'UTF-8')
logging.info('AzEl sent to DT: {}'.format(cmd))
self._outsocket.send(cmd)
def setOffset(self, offset):
'''
Set the offset of the telescope in Az / El.
offset should be a tuple of degrees or of astropy Quantity
'''
if self.setmode is None:
raise ValueError("Cannot set offset if mode is None")
(off_az, off_el) = offset
if isinstance(off_az, u.Quantity):
off_az = off_az.to(u.rad).value
else:
off_az = np.deg2rad(off_az)
if isinstance(off_el, u.Quantity):
off_el = off_el.to(u.rad).value
else:
off_el = np.deg2rad(off_el)
cmd = bytes('{:f}\t{:f}\n'.format(off_az, off_el), 'UTF-8')
logging.info('Offset sent to DT: {}'.format(cmd))
self._offsetsocket.send(cmd)
def getAzEl(self, waitForUpdate=False):
'''
Get azimut and elevation. If waitForUpdate=True, it waits for the
next signal from the socket.
Returns a tuple (az, el) as astropy quantity
'''
if waitForUpdate or self.az is None or self.radec is None:
self._event_traces.clear()
self._event_traces.wait()
return (self.az, self.el)
def getDistance(self, waitForUpdate=False):
'''
Get distance in azimut and elevation. If waitForUpdate=True, it waits
for the next signal from the socket.
Returns a tuple (dist_az, dist_el) as astropy quantity
'''
if waitForUpdate or self.dist_az is None or self.dist_el is None:
self._event_traces.clear()
self._event_traces.wait()
return (self.dist_az, self.dist_el)
def getRaDec(self, waitForUpdate=False):
'''
Get current Ra/Dec. If waitForUpdate=True, it waits for the
next signal from the socket.
Returns an astropy SkyCoord
'''
if waitForUpdate or self.radec is None:
self._event_traces.clear()
self._event_traces.wait()
return self.radec
def getSetpoint_RaDec(self, waitForUpdate=False):
'''
Get current Ra/Dec setpoint. If waitForUpdate=True, it waits for the
next signal from the socket.
Returns an astropy SkyCoord
'''
if waitForUpdate or self.radec is None:
self._event_traces.clear()
self._event_traces.wait()
return self.setpoint_radec
def waitUntilThere(self, tolerance=0.01*u.deg):
'''
Wait until distance to the setpoint gets within tolerance
'''
logging.info("Waiting to reach setpoint")
distIsSmall = False
while not distIsSmall:
diff = self.getDistance(waitForUpdate=True)
distIsSmall = abs(self.dist_el)<tolerance and abs(self.dist_az)<tolerance
def waitUntilMoving(self, tolerance=0.01*u.deg):
'''
Wait until the telescope reckognizes that it is not at its setpoint
'''
logging.info("Waiting to set setpoint")
distIsBig = False
while not distIsBig:
dist = self.getDistance(waitForUpdate=True)
distIsBig = dist[0]>tolerance or dist[1]>tolerance
def getJ2000(self):
'''
Generates dicts with 'radec' and 'setpoint_radec',
containing radec and setpoint_radec, at whichever
speed the console is outputting them (probably once a second).
'''
self._event_traces.clear()
while True:
self._event_j2000.wait()
self._event_j2000.clear()
yield({'radec': self.radec,
'setpoint_radec': self.setpoint_radec})
def getTraces(self):
'''
Generates a dict of all traces from trace2port at whichever
rate trace2port generates them (probably once a second).
'''
self._event_traces.clear()
while True:
self._event_traces.wait()
self._event_traces.clear()
yield({'position_azel': (self.az, self.el),
'distance': (self.dist_az, self.dist_el),
'speed': (self.speed_az, self.speed_el)})
from telescope import *
telescope.py 0 → 100644
View file @ 77df4636
from __future__ import print_function
import socket
import logging
from astropy.coordinates import SkyCoord, AltAz, RangeError
from astropy import units as u
from configparser import ConfigParser
import os.path
import numpy as np
import threading
import select
import time
config = ConfigParser()
config.readfp(open(os.path.join(os.path.dirname(__file__), 'telescope.ini')))
logging.basicConfig(level=logging.DEBUG)
class telescope():
def __init__(self, setmode=None, consoleHost=None, trace2portHost=None):
'''
Initializes a telescope instance. The mode for writing can be
'J2000' or 'AZEL'.
'''
if setmode not in ('J2000', 'AZEL', None):
raise ValueError("Mode must be None, 'J2000' or 'AZEL', not " + setmode)
self.setmode = setmode
if consoleHost is None:
consoleHost = config.get('Console','HostName')
if consoleHost == 'console':
if socket.gethostname()=="mercurius":
logging.warning("You are using the actual console, not a demo!")
else:
raise ValueError("Talking to the actual console can only be done from mercurius")
if trace2portHost is None:
trace2portHost = config.get('Trace2Port','HostName')
self.dist_az = None
self.dist_el = None
self.az = None
self.el = None
self.radec = None
self.speed_az = None
self.speed_el = None
self._outsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if self.setmode == 'J2000':
try:
self._outsocket.connect((consoleHost,
config.getint('Console','Port_Write_J2000')))
except socket.gaierror as e:
raise IOError("Could not connect to DT J2000 port: " + str(e))
elif self.setmode == 'AZEL':
try:
self._outsocket.connect((consoleHost,
config.getint('Console','Port_Write_AzEl')))
except socket.gaierror as e:
raise IOError("Could not connect to DT AZEL port: " + str(e))
# Initialize offset socket for all modes where output is desired
if self.setmode is not None:
self._offsetsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
self._offsetsocket.connect((consoleHost,
config.getint('Console','Port_Write_Offset')))
except socket.gaierror as e:
raise IOError("Could not connect to DT Offset port: " + str(e))
self._event_j2000 = threading.Event()
thread_read_j2000 = threading.Thread(target=self._readj2000, args=(
consoleHost,
config.getint('Console','Port_Read_J2000')))
thread_read_j2000.daemon = True
thread_read_j2000.start()
self._event_traces = threading.Event()
thread_read_traces = threading.Thread(target=self._readtraces, args=(
config.get('Trace2Port', 'Hostname'),
config.getint('Trace2Port', 'Port')))
thread_read_traces.daemon = True
thread_read_traces.start()
def _readj2000(self, consolehost, j2000_read_port):
'''
Poll the socket with the j2000 info, store its values in class members.
'''
try:
self._j2000socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._j2000socket.connect((consolehost, j2000_read_port))
except socket.error:
raise IOError("Cannot connect to j2000 status server")
while True:
readable, writable, exceptional = select.select(
[self._j2000socket], [], [])
if len(exceptional)>0:
raise IOError("Error with j2000 read socket")
msg = readable[0].recv(4096).decode("UTF-8")
vals = msg.strip().split()
(ra, dec) = (vals[2], vals[3])
self.radec = SkyCoord(ra, dec, frame='icrs')
(setpoint_ra, setpoint_dec) = (vals[0], vals[1])
try:
# The setpoint can be invalid
self.setpoint_radec = SkyCoord(setpoint_ra, setpoint_dec, frame='icrs')
except RangeError:
self.setpoint_radec = None
self._event_j2000.set()
def _readtraces(self, tracehost, traceport):
'''
Poll the socket with the traces, store their values in class members
'''
try:
self._tracesocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._tracesocket.connect((tracehost, traceport))
except socket.error:
raise IOError("Cannot connect to trace2port server")
while True:
readable, writable, exceptional = select.select(
[self._tracesocket], [], [])
if len(exceptional)>0:
raise IOError("Error with trace read socket")
msg = readable[0].recv(4096).decode("UTF-8")
vals = msg.strip().split()
self.az = (float(vals[0])*u.rad).to(u.deg)
self.el = (float(vals[1])*u.rad).to(u.deg)
self.dist_az = (float(vals[2])*u.rad).to(u.deg)
self.dist_el = (float(vals[3])*u.rad).to(u.deg)
self.speed_az = float(vals[4])
self.speed_el = float(vals[5])
self._event_traces.set()
def setRaDec(self, setpoint):
'''
Set the J2000 setpoint of the telescope.
Only works if the setmode is 'J2000'.
setpoint can be an astropy SkyCoord or a tuple (ra,dec) in radians
'''
if self.setmode != 'J2000':
raise ValueError("Cannot set Ra/Dec if mode is not J2000")
if isinstance(setpoint, tuple):
coord = SkyCoord(ra =setpoint[0] * u.radian,
dec=setpoint[1] * u.radian)
(ra, dec) = setpoint.to_string('hmsdms').split()
cmd = bytes('{:s}\t{:s}\n'.format(ra, dec.strip('+')), 'UTF-8')
logging.info('J2000 setpoint sent to DT: {}'.format(cmd))
self._outsocket.send(cmd)
def setAzEl(self, setpoint):
'''
Set the AzEl setpoint of the telescope.
Only works if the setmode is 'AZEL'
setpoint should be a tuple of degrees or of astropy Quantity
'''
if self.setmode != 'AZEL':
raise ValueError("Cannot set Az/El if mode is not AZEL")
(az, el) = setpoint
if isinstance(az, u.Quantity):
az = az.to(u.deg).value
if isinstance(el, u.Quantity):
el = el.to(u.deg).value
cmd = bytes('{:f}\t{:f}\n'.format(az, el), 'UTF-8')
logging.info('AzEl sent to DT: {}'.format(cmd))
self._outsocket.send(cmd)
def setOffset(self, offset):
'''
Set the offset of the telescope in Az / El.
offset should be a tuple of degrees or of astropy Quantity
'''
if self.setmode is None:
raise ValueError("Cannot set offset if mode is None")
(off_az, off_el) = offset
if isinstance(off_az, u.Quantity):
off_az = off_az.to(u.rad).value
else:
off_az = np.deg2rad(off_az)
if isinstance(off_el, u.Quantity):
off_el = off_el.to(u.rad).value
else:
off_el = np.deg2rad(off_el)
cmd = bytes('{:f}\t{:f}\n'.format(off_az, off_el), 'UTF-8')
logging.info('Offset sent to DT: {}'.format(cmd))
self._offsetsocket.send(cmd)
def getAzEl(self, waitForUpdate=False):
'''
Get azimut and elevation. If waitForUpdate=True, it waits for the
next signal from the socket.
Returns a tuple (az, el) as astropy quantity
'''
if waitForUpdate or self.az is None or self.radec is None:
self._event_traces.clear()
self._event_traces.wait()
return (self.az, self.el)
def getDistance(self, waitForUpdate=False):
'''
Get distance in azimut and elevation. If waitForUpdate=True, it waits
for the next signal from the socket.
Returns a tuple (dist_az, dist_el) as astropy quantity
'''
if waitForUpdate or self.dist_az is None or self.dist_el is None:
self._event_traces.clear()
self._event_traces.wait()
return (self.dist_az, self.dist_el)
def getRaDec(self, waitForUpdate=False):
'''
Get current Ra/Dec. If waitForUpdate=True, it waits for the
next signal from the socket.
Returns an astropy SkyCoord
'''
if waitForUpdate or self.radec is None:
self._event_traces.clear()
self._event_traces.wait()
return self.radec
def getSetpoint_RaDec(self, waitForUpdate=False):
'''
Get current Ra/Dec setpoint. If waitForUpdate=True, it waits for the
next signal from the socket.
Returns an astropy SkyCoord
'''
if waitForUpdate or self.radec is None:
self._event_traces.clear()
self._event_traces.wait()
return self.setpoint_radec
def waitUntilThere(self, tolerance=0.01*u.deg):
'''
Wait until distance to the setpoint gets within tolerance
'''
logging.info("Waiting to reach setpoint")
distIsSmall = False
while not distIsSmall:
diff = self.getDistance(waitForUpdate=True)
distIsSmall = abs(self.dist_el)<tolerance and abs(self.dist_az)<tolerance
def waitUntilMoving(self, tolerance=0.01*u.deg):
'''
Wait until the telescope reckognizes that it is not at its setpoint
'''
logging.info("Waiting to set setpoint")
distIsBig = False
while not distIsBig:
dist = self.getDistance(waitForUpdate=True)
distIsBig = dist[0]>tolerance or dist[1]>tolerance
def getJ2000(self):
'''
Generates dicts with 'radec' and 'setpoint_radec',
containing radec and setpoint_radec, at whichever
speed the console is outputting them (probably once a second).
'''
self._event_traces.clear()
while True:
self._event_j2000.wait()
self._event_j2000.clear()
yield({'radec': self.radec,
'setpoint_radec': self.setpoint_radec})
def getTraces(self):
'''
Generates a dict of all traces from trace2port at whichever
rate trace2port generates them (probably once a second).
'''
self._event_traces.clear()
while True:
self._event_traces.wait()
self._event_traces.clear()
yield({'position_azel': (self.az, self.el),
'distance': (self.dist_az, self.dist_el),
'speed': (self.speed_az, self.speed_el)})
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment