#! /usr/bin/env python # -*- Encoding: utf-8 -*- # ================================================================== # # DICE / StarDICE@OHP # # Serial interface (minimal) to the Gemini Pulsar2 controler (LUPM) # # Authors: Johann Cohen-Tanugi, Laurent Le Guillou, Kelly Joaquina # # Code History: # Carbon copy of temma/temma.py interface # in June 2017 (JCT) # # ================================================================== import sys import os, os.path import time import serial # ================================================================== #from exceptions import Exception class MountError(Exception): pass # ============ Class Temma controller ============================== def dms2deg(dms): ''' Convert a d:m:s string to decimal degree ''' d, m, s = dms.split(":") sig = -1 if d[0] == '-' else 1. ddeg = int(d) + sig * int(m) / 60. + sig * int(s)/3600. return ddeg class Pulsar2(object): """ The Temma class represents the Temma controller for the Takahashi EM-200 mount. """ # ---------- Constructor --------------------------------- def __init__(self, # port = '/dev/ttyUSB0', port = '/dev/pulsar2', serial_port = None, # to provide if serial port already created debug = True): self.port = port self.serial_port = serial_port self.baudrate = 9600 self.timeout = 0.5 # Non-blocking & non waiting mode self.repr_mode = 0 self.parity = serial.PARITY_NONE self.bytesize = serial.EIGHTBITS self.stopbits = serial.STOPBITS_ONE self.rtscts = False self.EOL = '#' self.BOL = '#:' # ---- debug mode self.debug = debug # ---- action timeout self.action_timeout = 10 self.open() # ---------- Open the controller device ------------------ def open(self): """ Open the Temma device. Check if there's something connected (echotest). """ #create the internal serial port instance if self.serial_port == None: if self.debug: print(("Opening port %s ..." % self.port)) keys = {'port': self.port, 'baudrate': self.baudrate, 'bytesize': self.bytesize, 'parity': self.parity, 'stopbits': self.stopbits, 'rtscts': self.rtscts, 'timeout': self.timeout} if self.port == 'dummy': from . import fakepulsar2 self.serial_port = fakepulsar2.FakePulsar2(**keys) self.serial_port.open() else: self.serial_port = serial.Serial(**keys) if ( (self.serial_port == None) or not(self.serial_port.isOpen()) ): raise IOError("Failed to open serial port %s" % self.port) # if serial port has already been created # check whether it is open, and if not reopen it: elif self.serial_port != None and not self.serial_port.isOpen(): self.serial_port.open() #paranoid flush self.serial_port.flushOutput() #port is already open else: if self.debug: print(( "port %s already open"%self.port)) if not(self.echotest()): raise IOError(("Not echoing on serial port %s") % self.port) if self.debug: print(( "Opening port %s done." % self.port))#, file=sys.stderr) # just in case, and this also set self.GOTO_ACTIVE self.GOTO_ACTIVE = None self.stop() #more paranoid : ensure long format self.set_long_format() return def set_long_format(self): """ pulsar2 can be set to use XX:YY.Z or XX:YY:ZZ format We only work with the latter. """ answer = self.send('GR') if answer[5]=='.': if not self.send_and_check('U'): print('currently in short format, and failed to switch to long format') return # ---------- Close the controller device ----------------- def close(self): if ( self.serial_port and self.serial_port.isOpen() ): self.serial_port.close() # ----------------- write command ----------------------- def write(self, command): """ Send a command through the serial port. """ if not( self.serial_port and self.serial_port.isOpen() ): raise IOError("Port %s not yet opened" % self.port) if self.debug: print(("Sending command [" + command + "]"))#, file=sys.stderr) # A trailing space is required self.serial_port.write((self.BOL + command + self.EOL).encode()) # ----------------- read command ----------------------- def read(self, timeout = None): """ Read the answer from the serial port. Return it as a string. If is specified, the function will wait for data with the specified timeout (instead of the default one). """ if not( self.serial_port and self.serial_port.isOpen() ): raise IOError("Port %s not yet opened" % self.port) if self.debug: print("Reading serial port buffer")#, file=sys.stderr) if timeout != None: self.serial_port.timeout = timeout if self.debug: print(("Timeout specified: ", timeout))# file=sys.stderr) answer = self.serial_port.readline() # return buffer # Restoring timeout to default one self.serial_port.timeout = self.timeout # answer = answer.strip() if self.debug: print(("Received [" + str(answer) + "]"))#, file=sys.stderr) return answer.decode() # ----------------- Purge the serial port --------------- def purge(self): """ Purge the serial port to avoid framing errors. """ self.serial_port.flushOutput() self.serial_port.flushInput() self.serial_port.readlines() # ---------- Echo test ---------------------------------- def echotest(self): cmd = self.BOL + "YV" + self.EOL print(cmd) self.serial_port.write(cmd.encode()) time.sleep(1) answer = self.serial_port.readline() print(("echotest: answer=[%s]" % answer)) # if (len(answer) < 3) or (answer[0:3] != 'ver'): if answer == '': return False return True # ---------- Send a command and get the answer ----------- def send(self, cmd, delay=0.1): """ Send a command (Temma language) to the motor and return the answer (if any). @param cmd: the command to send. """ command = cmd # Now send it self.write(command) # Parsing the answer (to detect errors) time.sleep(delay) answer = self.read() return answer.strip('#') def send_and_check(self, cmd, delay=0.1): answer = self.send(cmd, delay) if answer != "1": print(("%s not interpreted!"%cmd)) return False return True # ---------- Get the motor serial number ----------------- def get_version(self): """ Return the current axis position. """ answer = self.send("YV") return answer serial = property(get_version, doc="Version") # ---------- Setup: set latitude ------------------------- def set_latitude(self, d, m, s): # 'sDD\xdfMM:SS' if d<0: lat="-%02d\xdf%02d:%02d"%(abs(d),m,s) else: lat="+%02d\xdf%02d:%2d"%(d,m,s) return self.send_and_check("St %s" % lat) def get_latitude(self): answer = self.send("Gt") #returns 'sDD\xdfMM:SS#' DD,MM=answer.split('\xdf') MM,SS=MM.split(':') return int(DD),int(MM),int(SS) # ---------- Setup: set latitude ------------------------- def set_longitude(self, d, m, s): #if DDD is negative, it will be transformed internally if d<0: lon="-%03d\xdf%02d:%02d"%(abs(d),m,s) else: lon="+%03d\xdf%02d:%02d"%(d,m,s) answer = self.send("Sg %s" % lon) return True def get_longitude(self): answer = self.send("Gg") #returns 'DDD\xdfMM:SS#' DDD,MM=answer.split('\xdf') MM, SS = MM.split(':') return int(DDD),int(MM), int(SS) # ---------- Setup: set sideral time (LST) --------------- def set_LST(self, hour, minute, sec): #check that inputs are ints? lst = "%02d:%02d:%02d"%(hour, minute, sec) return self.send_and_check("SS %s" %lst) def get_LST(self): answer = self.send("GS") HH,MM,SS = answer.split(':') return int(HH),int(MM),int(SS) def get_time(self,fmt='24'): if fmt=='24': answer = self.send("GL") else: answer = self.send("Ga") HH,MM,SS = answer.split(':') return int(HH),int(MM),int(SS) def set_time(self,HH,MM,SS): #always input in 24h format return self.send_and_check("SL %02d:%02d:%02d" %(HH,MM,SS)) def get_date(self): answer = self.send("GC") MM,DD,YY = answer.split('/') return int(MM), int(DD), int(YY) def set_date(self,MM,DD,YY): #switch 4digit year into 2digit year if "%04d"%YY==str(YY): YY=int(str(YY)[2:]) return self.send_and_check("SC %02d/%02d/%02d" % (MM,DD,YY)) # # ---------- Init at Zenith ------------------------------ # Can not work: Pulsar2 refuses to validate a ra,dec that is not in the South. # def zenith(self) # """ # Initialisation: put the telescope to zenith # and call this method. # """ # h,m,s = self.get_LST() # d,m,s = self.get_latitude() # self.define_radec(h,m,s,d,m,s) # return False # ---------- Current motor position ---------------------- def get_radec(self): """ Return current RA, DEC of current position """ ra_answer = self.send("GR") dec_answer = self.send("GD") h, m, s =ra_answer.split(":") deg, arcmin, arcsec = dec_answer.split(":") return int(h), int(m), int(s), int(deg), int(arcmin), int(arcsec) def get_altaz(self): """ Return current altitude, azimuth of current position """ alt_answer = dms2deg(self.send("GA").replace('\xdf', ':')) az_answer = dms2deg(self.send("GZ").replace('\xdf', ':')) return alt_answer, az_answer def get_object_radec(self): """ Return current RA, DEC of current tracked object """ ra_answer = self.send("Gr") dec_answer = self.send("Gd") h, m, s =ra_answer.split(":") deg, arcmin, arcsec = dec_answer.split(":") return h, m, s, int(deg), int(arcmin), int(arcsec) # ------ Define current position (Initialization) --------- def define_radec(self, hour, minute, sec, deg, arcmin, arcsec): """ set the current telescope position to the provided radec. Warning: only works if telescope to the South """ answer = self.send_and_check("Sr %02d:%02d:%02d"%(hour,minute,sec)) if not answer: return False if deg<0: answer = self.send_and_check("Sd -%02d:%02d:%02d"%(-deg,arcmin,arcsec)) else: answer = self.send_and_check("Sd +%02d:%02d:%02d"%(deg,arcmin,arcsec)) if not answer: print("problem with set_radec") return False else: return self.sync() def set_object_radec(self, hour, minute, sec, deg, arcmin, arcsec): """ Set the radec of an object or position to goto """ answer = self.send_and_check("Sr %02d:%02d:%02d"%(hour,minute,sec)) if not answer: return False if deg<0: answer = self.send_and_check("Sd -%02d:%02d:%02d"%(-deg,arcmin,arcsec)) else: answer = self.send_and_check("Sd +%02d:%02d:%02d"%(deg,arcmin,arcsec)) if not answer: print("problem with set_radec") return False else: return True def start_slewing(self): """ command to actually launch slewing. It returns 0# after quite some delay, so account for it. """ answer = self.send("MS", delay=2) if answer != "0": print("command MS for goto has not been executed") return False return True def sync(self): """ Synchronize to internal catalog entry if available """ return self.send('CM') def goto_radec(self, hour, minute, sec, deg, arcmin, arcsec): """ Point to RA, DEC ('goto') """ if self.is_pointing(): raise MountError("Already moving (goto command ongoing)") answer = self.set_object_radec(hour, minute, sec,deg, arcmin, arcsec) if not answer: raise("object radec setting failed!") # self.sync() self.start_slewing() self.GOTO_ACTIVE = True # #https://github.com/indilib/indi/blob/master/libindi/drivers/telescope/lx200pulsar2.cpp#L1629 documents an issue with slewing bool set to 1 only a few seconds after slewing has started. Thus return only after a delay here # time.sleep(0.1) return True # ---------- Is it on E/W side ? ------------------------- def get_side(self): """ Tell us if the telescope is on the East(E) or West(W) side. """ answer = self.send('YGN') if answer=='0': return "E" elif answer=='1': return "W" else: print(('get_side got an unexpected answer from mount : ',answer)) return answer def set_side(self,side): """ Set telescope side wrt the pier """ if side not in [0,1,'E','W','e','w']: print("side must be either integer 0(E),1(W) or letters E, W (upper or lower case).") return False if side.upper()=='E': side=0 else: side=1 return self.send_and_check("YSN%d"%side) # ---------- Tell the mount to change side (E/W) --------- def flip_side(self): """ convenience wrap-up of set_side """ answer = self.get_side() if answer == "E": return self.set_side('W') elif answer == "W": return self.set_side("E") else: return False # ---------- Stop goto ----------------------------------- def stop(self): """ Stop an on-going command. """ answer = self.send("Q") if answer=='': self.GOTO_ACTIVE = False return True else: return False # ---------- Is the telescope doing a goto ? ------------- def is_pointing(self): """ A goto command is ongoing ? """ return self.send_and_check("YGi") # ---------- Continuous movements (simulating handset) -------- def move(self, direction): ''' move N/S/W/E as when using the handset, but the move is continuous until a Q command is sent. speed is defined by the guiding speed. ''' # Check that no goto is currently executed if self.is_pointing(): raise MountError('GOTO running, you need to stop GOTO first') direction = direction.lower() if direction not in ['e','w','s','n']: raise MountError('direction must be provided as e,w,s, or n') answer = self.send_and_check("M%s" % direction, delay=1) return answer def stop_move(self, direction): direction = direction.lower() if direction not in ['e','w','s','n']: raise MountError('direction must be provided as e,w,s, or n') answer = self.send("Q%s" % direction) return answer # ---------- Standby mode (on = RA motors off) ----------- def standby(self, on=True): """ set the mount to STILL state, which should stop tracking """ if on: #use still setting to stop tracking answer = self.set_tracking_type('STOP') else: #use sidereal setting to go back to tracking answer = self.set_tracking_type('SID') return True def is_standby(self): """ Is the telescope RA motors in stand-by ? """ return self.get_tracking_type() == 'STOP' def set_motion_speed(self, motion): if motion not in ['G', 'C', 'M', 'S']: print("motion argument needs to be either G for guide rate, C for center rate, M for find rate, or S for slew rate.") return False self.send("R%s"%motion) return True def flip_needed(self): """ flip est-west needed """ return self.send_and_check("YGW") ############# parking commands ##################### def is_home_set(self): """ Return True if a parking/home position is defined """ return self.send_and_check("YGh") def is_parked(self): """ return True if telescope is in parked state """ return self.send_and_check("YGk") def is_parking(self): """ Return True if telescope is currently moving to its parking position """ return self.send_and_check("YGj") def park(self): """ park the telescope. """ if self.is_parked(): print("Telescope already parked") elif self.is_parking(): print("parking in process") elif not self.is_home_set(): print("No home/park position defined!") elif self.is_pointing(): print("Telescope is currently slewing. Stop first!") else: return self.send_and_check("YH") return False def unpark(self): """ unpark the telescope """ if not self.is_parked(): print('Telescope is not parked') return False else : return self.send_and_check("YL") def set_speed(self, motion_type, value1,value2): """ value1 and 2 are in 10th of sid rate for guiding, and in sid rate unit for all others. Range is 1-9 for guiding, and 1-999 for all others. """ if (value1!=int(value1)) or (value2!=int(value2)): print("value1 and value2 must be integers, in units of 10th of sidereal rate for guiding (motion_type=A), and in units of sidereal rates for the other motion types.") return False motion_type = motion_type.upper() if motion_type not in ['A', 'B', 'C', 'D', 'E']: print("motion argument needs to be either A for guiding, B for centering, C for finding, D for slewing, or E for goto.") return False cmd = "YS%s%03d,%03d"%(motion_type,int(value1),int(value2)) return self.send_and_check(cmd) def get_speed(self, motion_type): motion_type = motion_type.upper() if motion_type not in ['A', 'B', 'C', 'D', 'E']: print("motion argument needs to be either A for guiding, B for centering, C for finding, D for slewing, or E for goto.") return False cmd = "YG%s"%motion_type return self.send(cmd) def get_location(self): """ returns geographical coordinates aaa,bbb, bbb<0 for East """ answer = self.send("YGI") return answer def set_location(self, lat_deg,lon_deg): """ lat first in deg, lon second in deg, negative for East. Ex: +47.3996,-018.7005 """ return self.send_and_check("YSI%s,%s"%(str(lat_deg),str(lon_deg))) def get_mount_type(self): """ returns 3=AltAz, 2=fork,1=German """ answer = self.send("YGM") return answer def set_mount_type(self,value): if value not in [1,2,3]: print("value must be integer 1(German), 2(flok), or 3(AltAz)") return False return self.send_and_check("YSM%d"%value) def allow_pole_crossing(self, value): if value not in [0, 1, False, True]: print("value must be 0,1 or bool False,True") return False, if value: value=1 else: value=0 return self.send_and_check("YSQ%d"%value) def pole_crossing_allowed(self): answer = self.send("YGQ") if answer=='0': return False elif answer=='1': return True else: print(("unexpected answer to YGQ :",answer)) return False def allow_refraction_correction(self, in_ra, in_dec): if (in_ra not in [0,1,False, True]) or (in_dec not in [0,1,False, True]): print("Inputs must be int 0/1 or bool False/True") return False in_ra=1 if in_ra else 0 in_dec=1 if in_dec else 0 return self.send_and_check('YSR%d,%d'%(in_ra,in_dec)) def refraction_correction_allowed(self): answer = self.send("YGR") if "," not in answer: raise("Parsing error for refraction_correction_allowed returned string") return answer.split(',') def set_tracking_type(self, tracking_type): val_tracking = {'STOP':0, 'SID':1, 'LUN':2, 'SOL':3, 'USER1':4, 'USER2':5, 'USER3':6} tracking_type = tracking_type.upper() if tracking_type not in ['STOP', 'SID', 'LUN', 'SOL', 'USER1', 'USER2', 'USER3']: raise('Wrong argument! Choose between STOP, SID, LUN, SOL, USER1, USER2, and USER3') return self.send_and_check('YSS%d,0'%val_tracking[tracking_type]) def get_tracking_type(self): """ get the current tracking rate type, and value in case of user defined YGS returns X,1 where X is what needs decoding. """ type_string ={0:'STOP',1:'SID',2:'LUN',3:'SOL',4:'USER1',5:'USER2',6:'USER3'} answer = self.send("YGS") print(answer) answer = int(answer.strip(',')[0]) if answer<4: return type_string[answer], None else: userid = answer-3 val = self.send('YGZ%d'%userid) return "USER%d"%userid,val def set_speed_rate(self, rate_type, value=None): """ set tracking rate type, and value (rad/min) in case of user defined tracking rate """ if rate_type>3: if value is None: print("value needs to be set in format %07d,%07d if user defined tracking rate is desired") return False else: userid=rate_type-3 return send_and_check("YSZ%d%s"%(userid, value)) def set_immediate_rate(self,value): "value format %07d,%07d. Max allowed value 4.1887902 rad/min id est 4 deg/sec" return self.send_and_check("YSZ%s"%value) def get_immediate_rate(self): return self.send("YGZ") def get_home(self): """ return the Alt/Az position of the telescope when homed" """ return self.send("YGX") def pulse_guide(self, direction, value, 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. """ #set the motion speed self.set_motion_speed(speed) d = direction.lower() if d not in ['w','e','n','s']: print("direction w,e,n, or s must be provided, followed by 3digit integer (10ms unit, so max is 999*10ms)") return False #3 below is dummy return self.send_and_check("Mg%s%03d3"%(d,value)) def is_autostop_enabled(self): answer = self.send("YGa") #second parameter always 0 ans = int(answer.split(',')[0]) if ans == 0: return False elif ans == 1: return True else: print(("YGA returned unexpected answer: ", answer)) return False # ----------- miscellaneous info ---------------- def display_data(self): return self.send("YGd") def get_latlon(self): """ Return latitude and longitude in decimal degrees """ return self.send("YGl") # ----------- motors info ---------------- def get_motor_info(self, which,val1,val2): which=which.lower() if which not in ["backlash","b","resolution","e",\ "maingear","m", "rotation", "n",\ "ramp","p","reduction","r"]: print("Wrong current type") return False tag=which[0] if which=="resolution": tag="e" elif which=="rotation": tag="n" elif which=="ramp": tag="p" return self.send("YG%s"%tag) def set_motor_info(self, which, val1, val2): which=which.lower() if which not in ["backlash","b","resolution","e","maingear","m", "rotation", "n","ramp","p","reduction","r"]: print("Wrong current type") return False tag=which[0] if which=="backlash": if val1>9 or val2>59: print("input out of range") return False valstring="%d,%02d"%(minute,sec) elif which=="resolution": tag="e" if str(val1) not in ['100','200','400'] or str(val2) not in ['100','200','400']: print("input must be 100,200, or 400") return False valstring = "%d,%d"%(val1,val2) elif which=="rotation": val1=str(val1) val2=str(val2) if val1 not in ['0','1'] or \ val2 not in ['0','1']: return False tag="n" valstring="%s,%s"%(val1, val2) elif which=="ramp": if int(val1) not in list(range(1,11)) or int(val2) not in list(range(1,11)): print("input value should be 1 to 10") return False tag="p" valstring="%d,%d"%(int(val1),int(val2)) elif which=="maingear" or which=="reduction": if int(val1)<100 or int(val2)<100 or int(val1)>6000 or int(val2)>6000: print("input value out of range") return False valstring="%04d,%04d"%(val1,val2) return self.send("YS%s"%valstring) def get_backlash(self): """ Return backlash in format 0-9:0-59 (min,sec) """ return self.send("YGb") def set_backlash(self, minute, sec): """ Set backlash in format 0-9,0-59 (min,sec) """ valstring="%d,%02d"%(minute,sec) return self.send_and_check("YSb%s"%valstring) def get_resolution(self): """ Return motor resolution in format 100/200/400,100/200/400# (step number) """ return self.send("YGe") def get_maingear(self): """ in format 100-6000,100-6000 """ return self.send("YGm") def set_maingear(self, valstring): """ in format 100-6000,100-6000 """ return self.send_and_check("YSm"+valstring) def get_rotation(self): """ Return 0/1,0/1 (Right=0/Left=1) """ answer = self.send("YGn") return answer def set_rotation(self, valstring): """ Return 0/1,0/1 (Right=0/Left=1) """ return self.send_and_check("YSn"+valstring) def get_ramp(self): """ 1-10,1-10# (Ra and DEC) """ return self.send("YGp") def set_ramp(self, valstring): """ string 'X,X' for 1-10,1-10# (Ra and DEC) setting """ return self.send_and_check("YSp"+valstring) def get_reduction(self): """ 100-6000,100-6000 """ return self.send("YGr") def set_reduction(self, valstring): """ string 'XXXX,YYYY' for 100-6000,100-6000 setting """ return self.send("YSr"+valstring) #------------------ currents ------------------------------- def get_stop_current(self): """ 0-2000, 0-2000 (mA) """ return self.send("YGs") def set_stop_current(self, valstring): """ string 'XXXX,YYYY' for 0-2000, 0-2000 (mA) setting """ return self.send_and_check("YSs"+valstring) def get_track_current(self): """ 0-2000, 0-2000 (mA) """ return self.send("YGt") def set_track_current(self, valstring): """ string 'XXXX,YYYY' for 0-2000, 0-2000 (mA) setting """ return self.send_and_check("YSt"+valstring) def get_goto_current(self): """ Return goto current in format 0-2000, 0-2000 (mA) """ return self.send("YGg") def set_goto_current(self, valstring): """ Set goto current in format 0-2000, 0-2000 (mA) """ return self.send("YSg"+valstring) def set_current(self, which, value1, value2): """ Set currents for track, stop, and goto. Currents are 0-2000 mA in 4 digits. """ which=which.lower() if which not in ["track","goto","stop","t","g","s"]: print("Wrong current type") return False valstring="%04d,%04d"%(value1, value2) return self.send_and_check("YS%s%s"%(which[0],valstring)) def get_current(self, which): """ Get currents for track, stop, and goto. Currents are returned in format XXXX,YYYY 0-2000 mA in 4 digits. """ which=which.lower() if which not in ["track","goto","stop","t","g","s"]: print("Wrong current type") return False return self.send("YG%s"%which[0]) # =================================================================== if __name__ == "__main__": mount = Pulsar2(port='/dev/pulsar2') # mount = Pulsar2(port='dummy')