#!/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 functools 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.core.pyros_django.utils.celme as celme import src.core.celme as celme from devices_controller.logs import * # Local application imports #sys.path.append('../..') #from src.client.socket_client_abstract import UnknownCommandException, SocketClientAbstract ##from src_socket.client.socket_client_abstract import * ##from src_device.client.client_channel import * from devices_controller.channels.client_channel_socket import ClientChannelSocket from devices_controller.channels.client_channel_serial import ClientChannelSerial from devices_controller.channels.client_channel_usb import ClientChannelUSB # 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' ''' # DECORATOR def generic_cmd(func): #def wrapper_generic_cmd(*args, **kwargs): @functools.wraps(func) def wrapper_generic_cmd(self, values_to_set=None): #print("func name is", func.__name__) return self.execute_generic_cmd(func.__name__, values_to_set) return wrapper_generic_cmd 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 UnexpectedCommandReturnCode(Exception): pass class TimeoutException(Exception): pass class UnknownCommandException(Exception): pass ''' def __init__(self,*args,**kwargs): super().__init__(self,*args,**kwargs) ''' #TODO: remove ClientChannelAbstract, and set instead a ClientChannel #class DeviceControllerAbstract(SocketClientAbstract): ##class DeviceControllerAbstract(ClientChannel): class DeviceControllerAbstract(): # ClientChannel used by the device controller (to be set during __init__ via set_client_channel()) my_channel = None # @abstract (to be overriden) _cmd_device_concrete = {} _cmd_device_abstract = {} _cmd = { # GET-SET commands: 'get_timezone': [], 'set_timezone': [], 'get_date': [], 'set_date': [], 'get_time': [], 'set_time': [], # DO commands: 'do_init': ['do_init'], 'do_park': [], } ##def __init__(self, device_host:str="localhost", device_port:int=11110, PROTOCOL:str="TCP", buffer_size=1024, DEBUG=False): def __init__(self, device_host:str="localhost", device_port:int=11110, PROTOCOL:str="SOCKET-TCP", buffer_size=1024, DEBUG=False): ''' :param device_host: server IP or hostname :param device_port: server port :param PROTOCOL: "SOCKET-TCP", "SOCKET-UDP", "SERIAL", or "USB" ''' ##super().__init__(device_host, device_port, PROTOCOL, buffer_size, DEBUG) set_logger(DEBUG) log_d("Logger configured") if PROTOCOL.startswith("SOCKET"): self.my_channel:ClientChannel = ClientChannelSocket(device_host, device_port, PROTOCOL, buffer_size, DEBUG) elif PROTOCOL == "SERIAL": self.my_channel:ClientChannel = ClientChannelSerial(device_host, device_port, buffer_size, DEBUG) elif PROTOCOL == "USB": self.my_channel:ClientChannel = ClientChannelUSB(device_host, device_port, buffer_size, DEBUG) else: raise Exception("Unknown Channel", PROTOCOL) # overwrite abstract _cmd dictionary with subclass native _cmd_native dictionary: #self._cmd = {**self._cmd, **self._cmd_native} self._cmd = {**self._cmd, **self._cmd_device_abstract, **self._cmd_device_concrete} # So that we can use this with the "with" statement (context manager) def __enter__(self): return self def __exit__(self, type, value, traceback): self.my_channel.__exit__(type, value, traceback) ''' def set_logger(self, DEBUG): self.my_channel.set_logger(DEBUG) ''' def _connect_to_device(self): self.my_channel._connect_to_server() #@override ClientChannel send_data def send_data(self, data:str): data_encapsulated = self.format_data_to_send(data) self.my_channel.send_data(data_encapsulated) ''' The chosen way to send data is this: # - :GD# b'00030000:GD#\x00' Another way to send data (which also works), is it better ? # - :GD# ###tsock.mysock.sendto(b'\x00\x00\x00\x01\x00\x00\x00\x00\x3A\x47\x44\x23\x00', (HOST, PORT)) # - :GR# ###tsock.mysock.sendto(b'\x00\x00\x00\x01\x00\x00\x00\x00\x3A\x47\x52\x23\x00', (HOST, PORT)) # - ACK 06 OK !!! : tsock.mysock.sendto(b'\x00\x00\x00\x01\x00\x00\x00\x00\x00\x06\x00\x00', (HOST, PORT)) Which one is the best method ? ''' #log_d("NATIVE Command to send is "+repr(data)) ##encapsulated_data = self.encapsulate_data_to_send(data) #print("before _send", encapsulated_data) #print("before _send", repr(encapsulated_data)) ##self._send_data(encapsulated_data) ##self.my_channel.send_data(encapsulated_data) #log_i(f'Sent: {encapsulated_data}') ''' def _send_data(self, data): self.my_channel._send_data(data) ''' #@override ClientChannel receive_data def receive_data(self)->str: ##data_received_bytes = self._receive_data() data_received = self.my_channel.receive_data() #log_d("Received (all data): {}".format(data_received)) #log_d("data in bytes: "+str(bytes(data_received, "utf-8"))) data = self.unformat_received_data(data_received) log_i("RECEIVED (useful data): {}".format(data)) return data ''' def _receive_data(self): return self.my_channel._receive_data() ''' # Encapsulate useful data to be ready for sending # By default, do nothing #@abstract def format_data_to_send(self, data:str): return self.encapsulate_data_to_send(data) #@deprecated def encapsulate_data_to_send(self, data:str): return data # Extract useful data from received raw data # By default, do nothing #@abstract def unformat_received_data(self, data:str): return self.uncap_received_data(data) #@deprecated def uncap_received_data(self, data:str): #return data_received.decode() return data ''' def encapsulate_data_to_send(self, command:str): return self.my_channel.encapsulate_data_to_send(command) def uncap_received_data(self, data_received:str): return self.my_channel.uncap_received_data(data_received) ''' def get_utc_date(self): return celme.Date("now").iso(0) #return celme.Date("now").ymdhms() def close(self): self.my_channel.close() 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()): return False,False if generic_cmd not in self._cmd: 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)->GenericResult: # 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") res_native = self.execute_native_cmd(raw_input_cmd) return GenericResult(res_native) #def execute_native_cmd(self, request:str, awaited_res_if_ok=None)->GenericResult: def execute_native_cmd(self, native_cmd:str)->str: print("NATIVE Command to send is "+ repr(native_cmd)) #self.send_request(native_cmd) self.send_native_cmd(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) return self.execute_native_cmd(request) def send_native_cmd(self, native_cmd:str)->str: return self.send_data(native_cmd) #@deprecated def send_request(self, request:str)->str: return self.send_native_cmd(request) def print_available_commands(self): print("\nAvailable commands are:") print("- GET commands:") print (list(cmd.replace('_',' ') for cmd in self._cmd.keys() if cmd.startswith('get_'))) print("- SET commands:") print (list(cmd.replace('_',' ') for cmd in self._cmd.keys() if cmd.startswith('set_'))) print("- DO commands:") print (list(cmd.replace('_',' ') 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 #def run_func(self, func, arg=None): def run_func(self, func, *args): #print("args", args) if args: return getattr(self, func)(*args) else: return getattr(self, func)() ''' TELESCOPE COMMANDS (abstract methods) ''' 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) print("\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", "get_radec" if native_cmd == generic_cmd: #print("cmd,val", native_cmd, values_to_set) #res:GenericResult = self.run_func(native_cmd, *values_to_set) if values_to_set: res = self.run_func(native_cmd, *values_to_set) #res = getattr(self, native_cmd)(values_to_set) else: res = self.run_func(native_cmd) #res = getattr(self, native_cmd)() #if res is None: res = 'ok' # res should be a GenericResult if not isinstance(res,GenericResult): raise Exception("Should be a GenericResult", res) return 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) ''' **************************** **************************** GENERIC TELESCOPE COMMANDS (abstract methods) **************************** **************************** ''' ''' **************************** GENERIC GET & SET commands **************************** ''' @generic_cmd def get_timezone(self): pass #def get_timezone(self): return self.execute_generic_cmd('get_timezone') @generic_cmd def set_timezone(self, hh): pass #def set_timezone(self, hh): return self.execute_generic_cmd('set_timezone', hh) @generic_cmd def get_date(self): pass @generic_cmd def set_date(self, mmddyy): pass @generic_cmd def get_time(self): pass @generic_cmd def set_time(self, hhmmss): pass ''' **************************** 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): pass #def do_PARK(self): return self._do("PARK") # @abstract def do_start(self): pass def do_stop(self): pass # @abstract MACRO def do_init(self): return 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() '''