#! /usr/bin/env python # # ================================================================== # # DICE/StarDICE # # XML-RPC server to control the Pulsar2 mount controler # default port: 8810 # # Authors: Laurent Le Guillou # (recycled from DICE control system code) # # ================================================================== import sys import os, os.path import time import subprocess import sys, os import signal import shutil import socket # to get the hostname import datetime import numpy as np # ================================================================== from pulsar2 import Pulsar2, MountError import ephem, inspect # ================================================================== import logging #logging.basicConfig(level=logging.DEBUG,format='%(asctime)s: %(message)s') # ================================================================== import inspect import xmlrpc.client from xmlrpc.server import SimpleXMLRPCServer, list_public_methods # ================================================================== DEFAULT_HOSTNAME = 'dicetel' DEFAULT_PORT = 8810 DEFAULT_HOSTNAME = "192.168.200.178" SERVER_HOSTNAME = os.getenv("MOUNT_SERVER_HOSTNAME", DEFAULT_HOSTNAME) SERVER_PORT = int(os.getenv("MOUNT_SERVER_PORT", DEFAULT_PORT)) # ================================================================== class MountServer(SimpleXMLRPCServer): def serve_forever(self): self.quit = 0 while not self.quit: self.handle_request() def server_quit(): logging.info("Exiting. Bye.") server.quit = 1 return 1 # def __getattr__(self, name): # if name == '__doc__': # return 'sub special' # else: # raise AttributeError, name # ================================================================== class Pulsar2Remote(object): instance = None # -------------------------------------------------------------- # def __new__(cls, *args, **kargs): # if cls.instance is None: # cls.instance = object.__new__(cls, *args, **kargs) # return cls.instance def __init__(self, port = "/dev/pulsar2", debug = True): self.state = 1 self.observer = ephem.Observer() self.mount = Pulsar2(port = port, debug = debug) # # name: "OHP" # # longitude: "+05d42m44s" # E # # latitude: "+43d55m54s" # N # self.observer.long = "05:42:44" # E # self.observer.lat = "43:55:54" # N # self.observer.elevation = 650.0 # m def set_observer(self, lon, lat, elev, name): """ lon and lat are in radians """ self.observer.long = lon self.observer.lat = lat d,m,s=str(self.observer.lat).split(':') self.set_latitude(float(d), float(m), float(s)) h,m,s=str(self.observer.lon).split(':') self.set_longitude(float(h), float(m), float(s)) self.observer.elev = elev self.observer.name = name self.update_LST() return True # -------------------------------------------------------------- def status(self): """ Return the status of the system. """ return self.state # -------------------------------------------------------------- def open(self): """ Open the connection with the Takahashi mount. """ logging.info("Pulsar2.open() called.") self.mount.open() logging.info("Pulsar2.open() done.") return True def get_version(self): """ Open the connection with the Takahashi mount. """ logging.info("Pulsar2.get_version() called.") self.mount.get_version() logging.info("Pulsar2.get_version() done.") return True def close(self): """ Close the connection with the Takahashi mount. """ logging.info("Pulsar2.close() called.") self.mount.close() logging.info("Pulsar2.close() done.") return True def park(self): """ """ logging.info("Pulsar2.park() called.") self.mount.park() logging.info("Pulsar2.park() done.") return True def unpark(self): """ """ logging.info("Pulsar2.unpark() called.") self.mount.unpark() logging.info("Pulsar2.unpark() done.") return True # ----------------------- Pulsar2 basic send---------------------- def send(self, cmd): """ Send a low level command to the Takahashi mount. """ logging.info("Pulsar2.send(%s) called." % cmd) answer = self.mount.send(cmd).replace('\xdf', ':') logging.info("Pulsar2.send() done.") return answer # ----------------------- setup: latitude, LST, zenith ---------- def allow_refraction_correction(self, in_ra, in_dec): logging.info("Pulsar2.allow_refraction_correction() called.") answer = self.mount.allow_refraction_correction(in_ra, in_dec) logging.info("Pulsar2.allow_refraction_correction() called.") return answer def refraction_correction_allowed(self): logging.info("Pulsar2.refraction_correction_allowed() called.") answer = self.refraction_correction_allowed() logging.info("Pulsar2.refraction_correction_allowed() called.") return answer def allow_pole_crossing(self, value): logging.info("Pulsar2.allow_pole_crossing() called.") answer = self.mount.allow_pole_crossing(value) logging.info("Pulsar2.allow_pole_crossing() called.") return answer def pole_crossing_allowed(self): logging.info("Pulsar2.pole_crossing_allowed() called.") answer = self.pole_crossing_allowed() logging.info("Pulsar2.pole_crossing_allowed() called.") return answer def set_longitude(self, d, m, s): """ Set the longitude for the observing site (ephem). Note: we keep ephem convention as input : longitudes are +E. Unfortunately, pulsar2 uses the convention -E, which is taken int account here. """ print("Pulsar2.set_longitude(%f,%f,%f) called." % (d,m,s)) logging.info("Pulsar2.set_longitude(%f,%f,%f) called." % (d,m,s)) self.observer.long = "%02d:%02d:%02d" % (d,m,s) answer = self.mount.set_longitude(-d,m,s) logging.info("Pulsar2.set_longitude() done.") return answer def set_latitude(self, d, m, s): """ Tell the latitude of the observing site to the Takahashi mount. Set the latitude for the observing site (ephem). """ logging.info("Pulsar2.set_latitude(%f,%f,%f) called." % (d,m,s)) self.observer.lat = "%02d:%02d:%02d" % (d,m,s) answer = self.mount.set_latitude(d,m,s) logging.info("Pulsar2.set_latitude() done.") return answer def get_tracking_type(self): logging.info("Pulsar2.get_tracking_type() called.") answer = self.mount.get_tracking_type() logging.info("Pulsar2.get_tracking_type() done.") return answer def set_tracking_type(self, tracking_type): logging.info("Pulsar2.set_tracking_type() called.") self.mount.set_tracking_type(tracking_type) logging.info("Pulsar2.set_tracking_type() done.") return True def get_latitude(self): """ Get the latitude of the observing site (see set_latitude). """ logging.info("Pulsar2.get_latitude() called.") answer = self.mount.get_latitude() logging.info("Pulsar2.get_latitude() done.") return answer def set_elevation(self, elevation): """ Set the elevation of the observing site (in meter). """ logging.info("Pulsar2.set_elevation() called.") self.observer.elevation = elevation logging.info("Pulsar2.set_elevation() done.") return True def set_LST(self, hour, minute, sec): """ Set the sidereal time for the Takahashi mount. """ logging.info("Pulsar2.set_LST(%f,%f,%f) called." % (hour, minute, sec)) answer = self.mount.set_LST(hour, minute, sec) logging.info("Pulsar2.set_LST() done.") return answer def get_LST(self): """ Get the sidereal time loaded in the Takahashi mount. """ logging.info("Pulsar2.get_LST() called.") answer = self.mount.get_LST() logging.info("Pulsar2.get_LST() done.") return answer def define_radec(self, hour, minute, sec, deg, arcmin, arcsec): """ Set current RA, DEC (initialisation). """ logging.info("Pulsar2.define_radec() called.") try: answer = self.mount.define_radec(hour, minute, sec, deg, arcmin, arcsec) except MountError as msg: logging.info("Pulsar2.define_radec() failed: returns [%s]." % msg) return False logging.info("Pulsar2.define_radec() done.") return answer def set_UTC(self, now=None): """ Resync the UTC time of the mount-server. (Important for the Raspberry Pi which has no real-time clock). """ if now is None: now = datetime.datetime.utcnow() date_str = "%02d%02d%02d%02d%04d.%02d" % (now.month, now.day, now.hour, now.minute, now.year, now.second) elif isinstance(now, str): date_str = now now = datetime.datetime.utcnow() elif isinstance(now, datetime.datetime): date_str = "%02d%02d%02d%02d%04d.%02d" % (now.month, now.day, now.hour, now.minute, now.year, now.second) else: print('set_UTC error with input ', date_str) return False ret = os.system("sudo date -u %s" % date_str) if ret != 0: return False logging.info("Setting UTC on OS: ", date_str) ans = self.mount.set_date(now.month, now.day, now.year) ans2 = self.mount.set_LST(now.hour, now.minute, now.second) return ans & ans2 # invoke the Unix command: # sudo date -u [MMDDhhmm[[CC]YY][.ss]] #now = datetime.datetime.utcnow() #date_str = "%02d%02d%02d%02d%04d.%02d" % (now.month, now.day, now.hour, now.minute, now.year, now.second) def update_LST(self): """ Compute the local sidereal time from the server os clock and update it in the mount. (and for ephem observer). This is used only once for the pulsqr2, as the latter has an internal clock """ logging.info("Pulsar2.update_LST() called.") self.observer.date = datetime.datetime.utcnow() lst = self.observer.sidereal_time() # un peu pourri... lst_h, lst_m, lst_s = [int(float(i)) for i in str(lst).split(":")[0:3]] self.mount.set_LST(lst_h, lst_m, lst_s) logging.info("Pulsar2.update_LST() done.") return True # ----------------------- Pulsar2 movements --------------------- def sync(self): logging.info("Pulsar2.sync() called.") self.mount.sync() logging.info("Pulsar2.sync() called.") def get_radec(self): """ Return current RA, DEC, E/W, H??? """ logging.info("Pulsar2.get_radec() called.") answer = self.mount.get_radec() logging.info("Pulsar2.get_radec() done.") return answer def get_altaz(self): """ Return current alt and az in decimal degrees """ logging.info("Pulsar2.get_altaz() called.") answer = self.mount.get_altaz() logging.info("Pulsar2.get_altaz() done.") return answer def goto_radec(self, hour, minute, sec, deg, arcmin, arcsec): logging.info("Pulsar2.goto_radec() called.") try: answer = self.mount.goto_radec(hour, minute, sec, deg, arcmin, arcsec) except MountError as msg: logging.info("Pulsar2.goto_radec() failed: returns [%s]." % msg) return False logging.info("Pulsar2.goto_radec() done.") return answer def goto(self, ra, dec): """ ra, dec in degrees. High level goto() method. Recomputes the LST, and then repoint the telescope. """ ra = str(ephem.hours(np.radians(ra))) hour, minute, sec = list(map(float, ra.split(':'))) dec = str(ephem.degrees(np.radians(dec))) deg, arcmin, arcsec = list(map(float, dec.split(':'))) logging.info("High level Pulsar2.goto() called.") result = self.goto_radec(hour, minute, sec, deg, arcmin, arcsec) logging.info("High level Pulsar2.goto() done.") return result # ---------- Stop goto ----------------------------------- def stop(self): """ Stop an on-going command. """ logging.info("Pulsar2.stop() called.") answer = self.mount.stop() logging.info("Pulsar2.stop() done.") return answer # ---------- Is it on E/W side ? ------------------------- def get_side(self): """ Tell us if the telescope is on the East(E) or West(W) side. """ logging.info("Pulsar2.get_side() called.") answer = self.mount.get_side() logging.info("Pulsar2.get_side() done.") return answer def set_side(self, value): """ Set the telescope on the East(E) or West(W) side. """ logging.info("Pulsar2.set_side() called.") answer = self.mount.set_side(value) logging.info("Pulsar2.set_side() done.") return answer # ---------- Tell the mount to change side (E/W) --------- def flip_side(self): """ Switch the mount side (E/W). Does not move the mount, but tell it on which side it actually is. IMPORTANT to avoid collision of the telescope tube. See also the zenith() method. When turned on, the mount believes it is on the West side. Please check with get_side() after! """ logging.info("Pulsar2.flip_side() called.") answer = self.mount.flip_side() logging.info("Pulsar2.flip_side() done.") return answer # ---------- Is the telescope doing a goto ? ------------- def is_pointing(self): """ A goto command is ongoing ? """ logging.info("Pulsar2.is_pointing() called.") answer = self.mount.is_pointing() logging.info("Pulsar2.is_pointing() done.") return answer # ---------- Handset move ------------- def move(self, direction, fast = False): """ Move N/S/W/E as when using the handset. speed is defined by the correction speed. Depending on the E/W position of the scope, the S/N direction is reversed. """ logging.info("Pulsar2.move() called.") answer = self.mount.move(direction) logging.info("Pulsar2.move() done.") return answer def pulse_guide(self, direction, ms, speed): """ send a guide pulse of a given value in units of 10ms in the provided direction. The param 'speed' provides the motion speed : G,C,M,S for guiding, centering, finding, and slewing, in increasing order. """ logging.info("Pulsar2.pulse_guide() called.") answer = self.mount.pulse_guide(direction, ms, speed) logging.info("Pulsar2.pulse_guide() done.") return answer # ---------- Standby mode (on = RA motors off) ----------- def standby(self, on=True): """ Initialisation: put the telescope RA motors on standby standby ON = RA OFF standby OFF = RA ON """ logging.info("Pulsar2.standby(%d) called." % on) answer = self.mount.standby(on) logging.info("Pulsar2.standby(%d) done." % on) return answer def is_standby(self): """ Is the telescope RA motors in stand-by ? """ logging.info("Pulsar2.is_standby() called.") answer = self.mount.is_standby() logging.info("Pulsar2.is_standby() done.") return answer # ----------- currents -------------------------- def set_current(self, which, value1, value2): logging.info("Pulsar2.set_current() called.") answer = self.mount.set_current(which, value1, value2) logging.info("Pulsar2.set_current() done.") return answer def get_current(self, which): logging.info("Pulsar2.set_current() called.") answer = self.mount.get_current(which) logging.info("Pulsar2.set_current() done.") return answer # ----------- motor info -------------------------- def set_motor_info(self, which, value1, value2): logging.info("Pulsar2.set_motor_info() called.") answer = self.mount.set_motor_info(which, value1, value2) logging.info("Pulsar2.set_motor_info() done.") return answer def get_motor_info(self, which): logging.info("Pulsar2.set_motor_info() called.") answer = self.mount.get_motor_info(which) logging.info("Pulsar2.set_motor_info() done.") return answer # ----------------------- Introspection ------------------------ def _listMethods(self): logging.info("Pulsar2._listMethods() called.") methods = list_public_methods(self) logging.info("Pulsar2._listMethods() done.") return methods def _methodHelp(self, method): f = getattr(self, method) return inspect.getdoc(f) # ================================================================== # ------------- Missing functions and tests ------------------------ def server_quit(): logging.info("Server going down.") server.quit = 1 return 1 # ================================================================== # ------------- Daemonization --------------------------------------- # Default working directory for the daemon. WORKDIR = "/" def redirect_stream(system_stream, target_stream): """ Redirect a system stream to a specified file. 'system_stream' is a standard system stream such as ''sys.stdout''. 'target_stream' is an open file object that should replace the corresponding system stream object. If 'target_stream' is ''None'', defaults to opening the operating system's null device and using its file descriptor. """ if target_stream is None: target_fd = os.open(os.devnull, os.O_RDWR) else: target_fd = target_stream.fileno() os.dup2(target_fd, system_stream.fileno()) def daemonize(options, args): try: pid = os.fork() if pid > 0: # exit first parent sys.exit(0) except OSError as e: print("fork #1 failed: %d (%s)" % (e.errno, e.strerror), file=sys.stderr) sys.exit(1) # Become leader of a new session to decouple from the controlling tty os.setsid() # We stay with the same umask # os.umask(0) # do second fork and kill session leader to ensure one will never get attach to a TTY. try: pid = os.fork() if pid > 0: # exit from second parent, print eventual PID before print("Starting server as daemon with PID %d ..." % pid) sys.exit(0) except OSError as e: print("fork #2 failed: %d (%s)" % (e.errno, e.strerror), file=sys.stderr) sys.exit(1) # start the daemon main main(options, args) def main(options, args): logging.basicConfig(filename=options.log_file, level=logging.DEBUG, format='%(asctime)s: %(message)s') # Now that logging is set up decouple from parent environnement redirect_stream(sys.stdin, None) redirect_stream(sys.stdout, None) redirect_stream(sys.stderr, None) # General Control Functions server.register_function(Pulsar2.status, "status") server.register_function(Pulsar2.open, "open") server.register_function(Pulsar2.get_version, "get_version") server.register_function(Pulsar2.close, "close") server.register_function(Pulsar2.unpark, "unpark") server.register_function(Pulsar2.park, "park") server.register_function(Pulsar2.send, "send") # Pulsar2 init server.register_function(Pulsar2.set_UTC, "set_UTC") server.register_function(Pulsar2.set_observer, "set_observer") server.register_function(Pulsar2.set_longitude, "set_longitude") server.register_function(Pulsar2.set_latitude, "set_latitude") server.register_function(Pulsar2.get_latitude, "get_latitude") server.register_function(Pulsar2.set_elevation, "set_elevation") server.register_function(Pulsar2.set_LST, "set_LST") server.register_function(Pulsar2.get_LST, "get_LST") #server.register_function(Pulsar2.update_LST, "update_LST") server.register_function(Pulsar2.define_radec, "define_radec") server.register_function(Pulsar2.allow_pole_crossing, "allow_pole_crossing") server.register_function(Pulsar2.pole_crossing_allowed, "pole_crossing_allowed") server.register_function(Pulsar2.allow_refraction_correction, "allow_refraction_correction") server.register_function(Pulsar2.refraction_correction_allowed, "refraction_correction_allowed") server.register_function(Pulsar2.sync, "sync") # Pulsar2 motion server.register_function(Pulsar2.get_tracking_type, "get_tracking_type") server.register_function(Pulsar2.set_tracking_type, "set_tracking_type") server.register_function(Pulsar2.get_radec, "get_radec") server.register_function(Pulsar2.get_altaz, "get_altaz") server.register_function(Pulsar2.goto_radec, "goto_radec") server.register_function(Pulsar2.goto, "goto") server.register_function(Pulsar2.stop, "stop") server.register_function(Pulsar2.get_side, "get_side") server.register_function(Pulsar2.set_side, "set_side") server.register_function(Pulsar2.flip_side, "flip_side") server.register_function(Pulsar2.is_pointing, "is_pointing") server.register_function(Pulsar2.move, "move") server.register_function(Pulsar2.pulse_guide, "pulse_guide") server.register_function(Pulsar2.standby, "standby") server.register_function(Pulsar2.is_standby, "is_standby") server.register_function(Pulsar2.set_current, "set_current") server.register_function(Pulsar2.get_current, "get_current") server.register_function(Pulsar2.set_current, "set_motor_info") server.register_function(Pulsar2.get_current, "get_motor_info") # misc server.register_function(server_quit, "quit") # for remote introspection (tab completion with ipython) server.register_function(Pulsar2._listMethods, "__dir__") server.register_function(Pulsar2._listMethods, "system.listMethods") server.register_function(Pulsar2._listMethods, "trait_names") server.register_function(Pulsar2._listMethods, "_getAttributeNames") # TODO: implement: system.methodSignature server.register_function(Pulsar2._methodHelp, "system.methodHelp") logging.info('Opening the connection') Pulsar2.open() logging.info("Server going up.") server.serve_forever() # ------------------------------------------------------------------ # ================================================================== if __name__ == '__main__': now = datetime.datetime.now() logdir = os.path.join(os.getenv("HOME"), "logs") logname = os.path.join(logdir, "mount-server-%s.log" % now.date().isoformat()) # recreate the symlink mount-server.log logsymlink = os.path.join(logdir, "mount-server.log") if os.path.islink(logsymlink): try: os.unlink(logsymlink) os.symlink(logname, logsymlink) except OSError: pass import optparse parser = optparse.OptionParser(usage=""" %prog [-l log-file] [-d] Start the Pulsar2 control server for the Takahashi EM-200 mount. """) parser.add_option('-d', '--daemon', default=False, action='store_true', help='Run as a background daemon') parser.add_option('--dummy', default=False, action='store_true', help='Run a fake instance instead') parser.add_option('-p', '--port', default=SERVER_PORT, action='store', type='int', help='Listen on port') parser.add_option('-H', '--hostname', default=SERVER_HOSTNAME, action='store', help='Listen adress') parser.add_option('-l', '--log-file', default=logname, action='store', help='Specify a file for daemon logs.') parser.add_option('-t', '--tty', default='/dev/pulsar2', dest='tty', action='store', help='specify the serial port') (options, args) = parser.parse_args() SERVER_HOSTNAME = options.hostname SERVER_PORT = options.port if options.dummy: #Pulsar2 = Pulsar2Dummy(port=options.tty) Pulsar2 = Pulsar2Remote(port='dummy') else: Pulsar2 = Pulsar2Remote(port=options.tty) server = MountServer((SERVER_HOSTNAME, SERVER_PORT)) server.register_introspection_functions() print("Pulsar2: Listening on port %s:%d. Waiting for commands." % (SERVER_HOSTNAME, SERVER_PORT)) if options.daemon: daemonize(options, args) else: main(options, args) # ==================================================================