#!/usr/bin/env python3 """Socket Client Telescope (abstract) implementation To be used as a base class (interface) for any concrete socket client telescope class """ # Standard library imports #from enum import Enum import logging import socket import sys import time # Third party imports # from sockets_tele/ sys.path.append("..") # from src_socket/client/ sys.path.append("../../..") import src.utils.celme as celme # Local application imports #sys.path.append('../..') #from src.client.socket_client_abstract import UnknownCommandException, SocketClientAbstract from src_socket.client.socket_client_abstract import * # Execute also "set" and "do" commands GET_ONLY=False # Execute only "get" commands #GET_ONLY=True # Default timeouts TIMEOUT_SEND = 10 TIMEOUT_RECEIVE = 10 ''' class c(Enum): # GET, SET DEC = 'DEC' RA = 'RA' RA_DEC = 'RA_DEC' # DO PARK = 'PARK' WARM_START = 'WARM_START' ''' class GenericResult: ''' Usage: res = execute(command) print("result is", res) if res.ko: raise UnexpectedReturnCode() if res.ok: ... ''' # By default, bad result ok = True ko = False def __init__(self, native_result:str, ok=True): self.txt = native_result self.ok = ok self.ko = not ok def __str__(self): return self.txt ''' def __repr__(self): return self.txt def __get__(self, instance, owner): return self.b def __set__(self, instance, value): self.b = value ''' class Position(): x = 0 y = 0 def __init__(self, x,y): self.x = x self.y = y def get_values(self): return self.x, self.y class UnexpectedCommandReturnCode(Exception): pass class TimeoutException(Exception): pass class UnknownCommandException(Exception): pass ''' def __init__(self,*args,**kwargs): super().__init__(self,*args,**kwargs) ''' class SocketClientTelescopeAbstract(SocketClientAbstract): # @abstract (to be overriden) _cmd = {} """ _cmd_getset = { 'ack': [], 'ra': [], 'dec': [], #'RADEC': [], 'timezone': [], #'TIMEZONE_IS_UTC': [], 'DATE': [], 'TIME': [], 'LONGITUDE': [], 'LATITUDE': [], 'VELOCITY': [], } _cmd_do = { #'INIT': [], 'PARK': [], 'MOVE': [], 'WARM_START': [], 'PREC_REFR': [], } """ def __init__(self, server_host:str="localhost", server_port:int=11110, PROTOCOL:str="TCP", buffer_size=1024, DEBUG=False): ''' :param server_host: server IP or hostname :param server_port: server port :param PROTOCOL: "UDP" or "TCP" ''' super().__init__(server_host, server_port, PROTOCOL, buffer_size, DEBUG) def get_utc_date(self): return celme.Date("now").iso(0) #return celme.Date("now").ymdhms() def get_celme_longitude(self, longitude): return celme.Angle(longitude).sexagesimal("d:+0180.0") def get_celme_latitude(self, latitude): return celme.Angle(latitude).sexagesimal("d:+090.0") def is_generic_cmd(self, raw_input_cmd:str): ''' is this a generic command ? :param raw_input_cmd: like 'get ra' or 'set ra 20:00:00' or 'set radec 20:00:00 90:00:00" or 'do park' or 'do_park'... :return either False or (cmd, [values]) with cmd like "get_ra" (using underscore '_' instead of space ' ') ''' #return cmd.startswith('get_') or cmd.startswith('set_') or cmd.startswith('do_') #cmds = ['get ', 'get_', 'set ', 'set_', 'do ', 'do_'] ''' seps = (" ", "_") #cmds = list(x+y for x in cmd for y in sep) for cmd in cmds: for sep in seps: generic_cmd = cmd+sep if raw_input_cmd.startswith(generic_cmd): # Is there value(s) passed ? if len(raw_input_cmd) > len(generic_cmd): values = raw_input_cmd[len(generic_cmd):] values = values.split(' ') # return cmd like "get_ra", [and values] return generic_cmd.replace(' ','_'), values return False, False ''' values_to_set = None cmds = ("get","set","do") raw_input_cmd = raw_input_cmd.strip() cmd_splitted = raw_input_cmd.split(' ') if len(cmd_splitted) == 1: return False,False # ex: "set_radec" generic_cmd = cmd_splitted[0] + '_' + cmd_splitted[1] # Check this generic command exists if (generic_cmd not in self._cmd.keys()) and (): return False,False # Is there value(s) passed ? if len(cmd_splitted) > 2: values_to_set = cmd_splitted[2:] # ex: return "set_radec", ["20:00:00", "90:00:00"] return generic_cmd, values_to_set def execute_cmd(self, raw_input_cmd:str)->str: # GENERIC command generic_cmd, values = self.is_generic_cmd(raw_input_cmd) if generic_cmd is not False: #print("GENERIC COMMAND") return self.execute_generic_cmd(generic_cmd, values) else: ''' if cmd.startswith('get_'): #generic_cmd,_ = request[4:].split('(') generic_cmd = cmd[4:] if (generic_cmd not in self._cmd_getset.keys()) and (generic_cmd not in self._cmd_do.keys()): #eval(request) return self.get_radec() print("cmd is", generic_cmd) return self._get(generic_cmd) return ''' # NATIVE command #print("NATIVE COMMAND") return self.execute_native_cmd(raw_input_cmd) #def execute_native_cmd(self, request:str, awaited_res_if_ok=None)->GenericResult: def execute_native_cmd(self, native_cmd:str)->str: log_d("NATIVE Command to send is "+ repr(native_cmd)) self.send_request(native_cmd) native_res = self.receive_data() return native_res ''' ok = True if not awaited_res_if_ok else (native_res == awaited_res_if_ok) return GenericResult(native_res, ok) ''' def execute_native_cmd_OLD(self, request:str)->str: self.send_request(request) native_res = self.receive_data() return native_res def execute_unformated_native_cmd(self, request:str)->str: request = self.formated_cmd(request) return self.execute_native_cmd_OLD(request) def send_request(self, request:str)->str: return self.send_data(request) def print_available_commands(self): print("\nAvailable commands are:") print("- GET commands:") print (list(cmd for cmd in self._cmd.keys() if cmd.startswith('get_'))) print("- SET commands:") print (list(cmd for cmd in self._cmd.keys() if cmd.startswith('set_'))) print("- DO commands:") print (list(cmd for cmd in self._cmd.keys() if cmd.startswith('do_'))) def available_commands(self): return list(self._cmd.keys()) # @abstract def formated_cmd(self, cmd:str, value:str=None)->str: return cmd ''' TELESCOPE COMMANDS (abstract methods) ''' ''' The 3 main generic commands : get(), do(), set() These abstract methods must be implemented (overriden) by the child classes, otherwise they raise a NotImplementedError exception ''' def _get(self, generic_param:str)->str: return self._get_set_do(self._cmd_getset, 0, generic_param) def _set(self, generic_param:str, value:str)->str: if GET_ONLY: return 'NOT_DONE' return self._get_set_do(self._cmd_getset, 2, generic_param, value) def _do(self, generic_param:str)->str: if GET_ONLY: return 'NOT_DONE' return self._get_set_do(self._cmd_do, 0, generic_param) def execute_generic_cmd(self, generic_cmd:str, values_to_set:str=None)->str: ''' Execute a generic command :param generic_cmd: str like "get_ra" or "set_ra" or "do_park"... :param value: only for a "set_" cmd ''' log_d("\n\nGENERIC Command to send is "+generic_cmd) # Check if generic_param exists #if generic_cmd not in self._cmd.keys(): raise UnknownCommandException() # if this generic command has no corresponding native command, raise NotImplementedError native_cmd_infos = self._cmd[generic_cmd] if not native_cmd_infos: raise NotImplementedError # Get corresponding native command: native_cmd = native_cmd_infos[0] if not native_cmd: raise NotImplementedError # ex: native_cmd == "do_init" if native_cmd == generic_cmd: res = getattr(self, native_cmd)(values_to_set) return GenericResult(res) # ex: native_cmd == "GR" else: native_cmd = self.formated_cmd(native_cmd,values_to_set) awaited_res_if_ok = None if len(native_cmd_infos) > 1: awaited_res_if_ok = native_cmd_infos[1] #native_res = self.execute_native_cmd(self.formated_cmd(native_cmd,value), awaited_res_ok) native_res = self.execute_native_cmd(native_cmd) ok = True if not awaited_res_if_ok else (native_res == awaited_res_if_ok) return GenericResult(native_res, ok) ''' def _get_set_do(self, dic:dict, indice:int, generic_param:str, value:str=None)->str: generic_cmd = 'do_' if dic == self._cmd_getset: generic_cmd = 'get_' if indice==0 else 'set_' generic_cmd += generic_param log_d("\nGENERIC command: "+generic_cmd) # Check if generic_param exists if not generic_param: raise UnknownCommandException() if generic_param not in dic.keys(): raise UnknownCommandException() # if this command is not overriden by this class, raise NotImplementedError if not dic[generic_param]: raise NotImplementedError if len(dic[generic_param]) < indice+1: raise NotImplementedError # Get corresponding native command: native_cmd = dic[generic_param][indice] # if this command is not overriden by this class, raise NotImplementedError if native_cmd == '': raise NotImplementedError awaited_res_ok = None if len(dic[generic_param]) > 1: awaited_res_ok = dic[generic_param][indice+1] return self.execute_native_cmd(self.formated_cmd(native_cmd,value), awaited_res_ok) ''' ''' def _get_set_do(self, dic:dict, indice:int, generic_param:str, value:str=None)->str: #if generic_param in c: return generic_param.name #if generic_param == c.RA_DEC: return self.get_radec() #return NotImplementedError native_cmds = [] generic_cmd = 'do_' if dic == self._cmd_getset: generic_cmd = 'get_' if indice==0 else 'set_' generic_cmd += generic_param + '()' log_d("\nGENERIC command: "+generic_cmd) if not generic_param: raise UnknownCommandException() #print("keys are", str(self._cmd.keys())) if generic_param not in dic.keys(): raise UnknownCommandException() # if this command is not overriden by this class, raise NotImplementedError if not dic[generic_param]: raise NotImplementedError # specific GET command: specific_cmd = dic[generic_param][indice] # if this command is not overriden by this class, raise NotImplementedError if specific_cmd == '': raise NotImplementedError if isinstance(specific_cmd,str): native_cmds = [specific_cmd] else: native_cmds = specific_cmd res = [] #print("commands are", native_cmds) for cmd in native_cmds: #print("command is", repr(cmd)) #res.append(self.put_read(self.formated_cmd(cmd))) res.append(self.execute_native_cmd_OLD(self.formated_cmd(cmd,value))) if len(res) == 1: return res[0] return res ''' ''' **************************** GENERIC GET & SET commands **************************** ''' # @abstract def get_ack(self): #return self._get("ack") return self.execute_generic_cmd("get_ack") # RA/DEC # @abstract ''' Sets the object's Right Ascension and the object status to "Not Selected". The :Sd# command has to follow to complete the selection. The subsequent use of the :ON...# command is recommended (p106) :Sr:.# or :Sr::# 0 if invalid 1 if valid ''' def get_ra(self): return self.execute_generic_cmd("get_ra") def set_ra(self, ra): return self._set("ra", ra) ''' Sets the object's declination. It is important that the :Sr# command has been send prior. Internal calculations are done that may take up to 0.5 seconds. If the coordinate selection is valid the object status is set to "Selected" :Sd{+-}
{*°}# or :Sd{+- }
{*°:}: 0 if invalid 1 if valid ''' def get_dec(self): return self.execute_generic_cmd("get_dec") def set_dec(self, dec): return self._set("dec", dec) # @abstract # @abstract #def get_radec(self): return self._get("RADEC") def get_radec(self)->tuple: return (self.get_ra(), self.get_dec()) def set_RADEC(self, radec): return self._set("RADEC", radec) #def set_TIMEZONE_IS_UTC(self): return self._set('TIMEZONE_IS_UTC') def get_timezone(self): return self.execute_generic_cmd('get_timezone') def set_timezone(self, hh): return self.execute_generic_cmd('set_timezone', hh) def get_DATE(self): return self._get('DATE') def set_DATE(self, mmddyy): return self._set('DATE', mmddyy) def get_TIME(self): return self._get('TIME') def set_TIME(self, hhmmss): return self._set('TIME', hhmmss) def get_LONGITUDE(self): return self._get('LONGITUDE') def set_LONGITUDE(self, longitude): return self._set('LONGITUDE', longitude) def get_LATITUDE(self): return self._get('LATITUDE') def set_LATITUDE(self, latitude): return self._set('LATITUDE', latitude) def get_VELOCITY(self): return self._get('VELOCITY') ''' **************************** GENERIC DO commands **************************** ''' # @abstract #def do_INIT(self): return self._do("INIT") ''' do_PARK() (p103) - STARTUP position = CWD - :hC# - position required for a Cold or Warm Start, pointing to the celestial pole of the given hemisphere (north or south), with the counterweight pointing downwards (CWD position). From L4, V1.0 up - HOME position parking => par defaut, c'est CWD, mais ca peut etre different - :hP# - defaults to the celestial pole visible at the given hemisphere (north or south) and can be set by the user ''' # @abstract def do_PARK(self): return self._do("PARK") def do_MOVE(self): return self._do("MOVE") def do_STOP(self): return self._do("STOP") # @abstract #def do_GOTO(self, pos:Position): return self._do("GOTO") #def do_WARM_START(self): return self._do("WARM_START") def do_warm_start(self): pass def do_PREC_REFR(self): return self._do("PREC_REFR") # @abstract def do_init(self): ''' 1) Send cde ACK ('06') and check answer to see if telescope is ready (see doc page 100) (utile pour savoir si tout est ok ; par ex, si une raquette est branchée sur le tele, ça peut bloquer le protocole) Usable for testing the serial link and determining the type of mount (German equatorial). Return code can be: - B# while the initial startup message is being displayed (new in L4), - b# while waiting for the selection of the Startup Mode, - S# during a Cold Start (new in L4), - G# after completed startup ==> MEANS ALL IS OK ''' #ACK = self.get("ACK") ACK = self.get_ack() ''' 2) IF telescope is not ready (still starting up), ask it to do a Warm Start ('bW#') During Startup, with a "b#" being returned, the PC can select the startup mode by sending a • bC# for selecting the Cold Start, • bW# for selecting the Warm Start, • bR# for selecting the Warm Restart If not ok (still starting up, no 'G#' in return), send 'bW#' (see above) for selecting the Warm Start ''' #if ACK != 'G': if not ACK.ok: self.do_warm_start() ACK = self.get_ack() elapsed_time = 0 while not ACK.ok: time.sleep(1) elapsed_time += 1 if elapsed_time == TIMEOUT_RECEIVE: raise TimeoutException() ACK = self.get_ack() ''' 3) Set timezone, date, and time (p109) ''' ''' a) set TIMEZONE Set the number of hours by which your local time differs from UTC. If your local time is earlier than UTC set a positive value, if later than UTC set a negative value. The time difference has to be set before setting the calendar date (SC) and local time (SL), since the Real Time Clock is running at UTC => :SG{+-}hh# ''' res = self.get_timezone() print("Current timezone is", res) res = self.set_timezone('+00') #if res != '1': raise UnexpectedCommandReturnCode(res) if not res.ok: raise UnexpectedCommandReturnCode(res) res = self.get_timezone() if res.txt != '+00': raise UnexpectedCommandReturnCode(res) print("NEW timezone set is", res) return ''' b) set DATE Set Calendar Date: months mm, days dd, year yy of the civil time according to the timezone set. The internal calendar/clock uses GMT :SC/
/# 0 if invalid or TODO: 1Updating planetary data#<24 blanks># ''' res = self.get_DATE() print("Current date is", res) # format is 2018-09-26T17:50:21 d = self.get_utc_date() # format to mm/dd/yy now_utc_mm_dd_yy = d[5:7] + '/' + d[8:10] + '/' + d[2:4] #print("date is", now_utc_mm_dd_yy) res = self.set_DATE(now_utc_mm_dd_yy) #res = self.set_DATE(self.get_utc_date()) #if res[0] != '1': raise UnexpectedCommandReturnCode(res) #if not res.startswith('1Updating planetary data'): raise UnexpectedCommandReturnCode(res) if not res: raise UnexpectedCommandReturnCode(res) res = self.get_DATE() if res != now_utc_mm_dd_yy: raise UnexpectedCommandReturnCode(res) print("NEW DATE set is", res) ''' c) set TIME Set RTC Time from the civil time hours hh, minutes mm and seconds ss. The timezone must be set before using this command :SL::# ''' res = self.get_TIME() print("Current time is", res) _,now_utc_hh_mm_ss = d.split('T') #print("time is", now_utc_hh_mm_ss[:5]) res = self.set_TIME(now_utc_hh_mm_ss) #if res != '1': raise UnexpectedCommandReturnCode(res) if not res: raise UnexpectedCommandReturnCode(res) res = self.get_TIME() if res[:5] != now_utc_hh_mm_ss[:5]: raise UnexpectedCommandReturnCode(res) ''' 4) Set LOCATION (lat,long) (p103,110) Pour l'observatoire de Guitalens: Sg = 2.0375 E St = 43.6443 N (attention, 2.0375 E = - 2.0375) ''' ''' a) set Longitude Sets the longitude of the observing site to ddd degrees and mm minutes. The longitude has to be specified positively for western latitudes (west of Greenwich, the plus sign may be omitted) and negatively for eastern longitudes. Alternatively, 360 degrees may be added to eastern longitudes. => :Sg{+-}*# ''' # TELE format is -002°02 (I convert it to -002:02) res = self.get_LONGITUDE() print("Current longitude is", res) # CELME format is -002:02:15 res = self.get_celme_longitude("-2.0375") res_ddd_mm = res[:-3] #res_ddd_mm = '-002:03' #print("celme longitude is", res) ddd,mm,ss = res.split(':') dddmm = ddd+'*'+mm #dddmm = '-002*03' res = self.set_LONGITUDE(dddmm) #if res != '1': raise UnexpectedCommandReturnCode(res) if not res: raise UnexpectedCommandReturnCode(res) res = self.get_LONGITUDE() if res != res_ddd_mm: raise UnexpectedCommandReturnCode(res) ''' b) set Latitude Sets the latitude of the observing site to dd degrees, mm minutes. The minus sign indicates southern latitudes, the positive sign may be omitted. => :St{+-}
*# ''' # TELE format is +43°38 (I convert it to +43:38) res = self.get_LATITUDE() print("Current latitude is", res) # CELME format is +43:38:15 res = self.get_celme_latitude("+43.6443") res_dd_mm = res[:-3] print("res is", res) #res_dd_mm = '+43:50' #print("celme longitude is", res) dd,mm,ss = res.split(':') ddmm = dd+'*'+mm #ddmm = '+43*50' res = self.set_LATITUDE(ddmm) #if res != '1': raise UnexpectedCommandReturnCode(res) if not res: raise UnexpectedCommandReturnCode(res) res = self.get_LATITUDE() if res != res_dd_mm: raise UnexpectedCommandReturnCode(res) ''' 5) Send cde ':p3#' : Precession & Refraction (see page 107) Ask Gemini to do Precession calculation Coordinates transferred to the Gemini refer to the standard epoch J2000.0. Refraction is calculated (From L4, V1.0 up) ''' self.do_PREC_REFR() # @abstract def set_SPEED(self, speed_rate): pass ''' GOTO (p105) - GOTO(position, blocking=Y/N): (MS = move start) = Goto RA=18h23m45s Dec=+34d00m00s J2000 - radec.goto() ''' # @abstract def do_goto(self, ra, dec, speed_rate=None): # 1) set speed if speed_rate: self.set_SPEED(speed_rate) radec = self.get_radec() print("Current position is", radec) # 2) set RA-DEC ''' :Sr18:23:45#:Sd+34:00:00#:MS# ''' res = self.set_ra(ra) if res != '1': raise UnexpectedCommandReturnCode(res) res = self.set_dec(dec) if res != '1': raise UnexpectedCommandReturnCode(res) # 3) MOVE (non blocking by default for GEMINI) self.do_MOVE() # 4) Test velocity until it is "Tracking" ''' After MOVE, test velocity with ':Gv#' (p103) : we should have 'S', then 'C', then 'T' - N (for "no tracking") - T (for Tracking) - G (for Guiding) - C (for Centering) - S (for Slewing) ''' v = None while v != 'T': v = self.get_VELOCITY() print("Velocity is", v) time.sleep(2) time.sleep(2) radec = self.get_radec() print("Current position is", radec) def do_MOVE_NORTH(self): return self._fo('MOVE_NORTH') def do_MOVE_SOUTH(self): return self._fo('MOVE_SOUTH') def do_MOVE_WEST(self): return self._fo('MOVE_WEST') def do_MOVE_EAST(self): return self._fo('MOVE_EAST') # @abstract def do_move_dir(self, dir, nbsec, speed_rate=None): if speed_rate: self.set_SPEED(speed_rate) if dir=="NORTH": self.do_MOVE_NORTH() if dir=="SOUTH": self.do_MOVE_SOUTH() if dir=="WEST": self.do_MOVE_WEST() if dir=="EAST": self.do_MOVE_EAST() time.sleep(int(nbsec)) self.do_STOP() #TODO: delete if not necessary ''' SPECIFIC COMMANDS ''' ''' # @abstract def initialize(self): raise NotImplementedError # @abstract def start(self): raise NotImplementedError # @abstract def park(self): raise NotImplementedError # @abstract def goto(self, position:Position): raise NotImplementedError # @abstract def move(self, ): raise NotImplementedError # @abstract def get_coord_sys(self): raise NotImplementedError # @abstract def set_coord_sys(self): raise NotImplementedError # @abstract def get_position(self)->Position: raise NotImplementedError ''' # TODO: empecher de creer une instance de cette classe abstraite # Avec ABC ? ''' if __name__ == "__main__": #HOST, PORT = "localhost", 9999 #HOST, PORT = "localhost", 20001 HOST, PORT = "localhost", 11110 # Classic usage: #tsock = SocketClient_UDP_TCP(HOST, PORT, "UDP") # More elegant usage, using "with": with SocketClient_ABSTRACT(HOST, PORT, "UDP") as tsock: # 0) CONNECT to server (only for TCP, does nothing for UDP) tsock._connect_to_server() while True: # 1) SEND REQUEST data to server # saisie de la requête au clavier et suppression des espaces des 2 côtés data = input("REQUEST TO SERVER [ex: ':GD#' (Get Dec), ':GR#' (Get RA)']: ").strip() # test d'arrêt if data=="": break #data_to_send = bytes(data + "\n", "utf-8") tsock.send_data(data) #mysock.sendto("%s" % data, (HOST, PORT)) #print("Sent: {}".format(data)) # 2) RECEIVE REPLY data from server data_received = tsock.receive_data() #reponse, adr = mysock.recvfrom(buf) #print("Received: {}".format(data_received)) #print("Useful data received: {}".format(data_useful)) print('\n') #tsock.close() '''