#!/usr/bin/env python # # Exposes autoguider + XYslides comamnds # author: Stephen Potter # updated: 22 June 2025 import socket from threading import Thread from time import sleep from math import sin, cos, sqrt, exp from math import pi as PI #from PIL import Image from io import BytesIO #from random import random import struct import threading import datetime TIMEOUT_RETRY = 0.5 TIMEOUT_NUMTRIES = 1 PORT = 12351 #HOST = "localhost" # a bit confusing, but lesedi_host is not used here ... instead has its own defintion and use in gaia_list.py when called as a libray/function LESEDI_HOST = "10.2.2.31" # IP address of 1ms1.suth where main_prog runs #LESEDI_HOST = "10.2.2.32" # IP address of lesedis2.suth where main_prog runs #globalXC = 0 #globalYC = 0 class State(): def __init__(self): self.slide_moving = -1 self.slide_error = -1 self.slide_init = 0 self.which_slide = -1 self.x_upper_lim = 0 self.x_lowr_lim = 0 self.y_upper_lim = 0 self.y_lowr_lim = 0 self.XY_slide_state = 0 self.slide_x = 0 self.slide_y = 0 self.slide_x_err_code = 0 self.slide_y_err_code = 0 self.below_pier_temp = 0 self.at_PLC_temp = 0 self.at_Primary_mir_temp = 0 self.at_Secndary_mir_temp = 0 self.status0 = 0 self.rotator0_ang = 0 self.tel_parallactic_ang0 = 0 self.t_p_a0rate = 0 self.skyangle0 = 0 self.status1 = 0 self.rotator1_ang = 0 self.tel_parallactic_ang1 = 0 self.t_p_a1rate = 0 self.skyangle1 = 0 self.guide_flag = -1 self.guide_point = -1 self.NS_Flag = -1 self.EW_Flag = -1 self.fake_flag = -1 self.xc0 = 0 self.yc0 = 0 self.ext_continuous = -1 self.binning = -1 self.xc = 0 self.yc = 0 self.message = None self.exptime = 0 self.no_lodestars = -1 def __repr__(self): return """State: slide_moving = {} slide_error = {} slide_init = {} which_slide = {} x_upper_lim = {} x_lowr_lim = {} y_upper_lim = {} y_lowr_lim = {} XY_slide_state = {} slide_x = {} slide_y = {} slide_x_err_code = {} slide_y_err_code = {} below_pier_temp = {} at_PLC_temp = {} at_Primary_mir_temp = {} at_Secndary_mir_temp = {} status0 = {} rotator0_ang = {} tel_parallactic_ang0 = {} t_p_a0rate = {} skyangle0 = {} status1 = {} rotator1_ang = {} tel_parallactic_ang1 = {} t_p_a1rate = {} skyangle1 = {} guide_flag = {} guide_point = {} NS_Flag = {} EW_Flag = {} fake_flag = {} xc0 = {} yc0 = {} ext_continuous = {} binning = {} xc = {} yc = {} message = {} exptime = {} no_lodestars = {} """.format(self.slide_moving, self.slide_error, self.slide_init, self.which_slide, self.x_upper_lim, self.x_lowr_lim, self.y_upper_lim, self.y_lowr_lim, self.XY_slide_state, self.slide_x, self.slide_y, self.slide_x_err_code, self.slide_y_err_code, self.below_pier_temp, self.at_PLC_temp, self.at_Primary_mir_temp, self.at_Secndary_mir_temp, self.status0, self.rotator0_ang, self.tel_parallactic_ang0, self.t_p_a0rate, self.skyangle0, self.status1, self.rotator1_ang, self.tel_parallactic_ang1, self.t_p_a1rate, self.skyangle1, self.guide_flag, self.guide_point, self.NS_Flag, self.EW_Flag, self.fake_flag, self.xc0, self.yc0, self.ext_continuous, self.binning, self.xc, self.yc, self.message, self.exptime, self.no_lodestars ) def __str__(self): return self.__repr__() def parse_state(self, state_str): state_list = state_str.split('\n') temp = state_list[1] self.slide_moving = temp[15] self.slide_error = temp[14] self.slide_init = temp[13] self.which_slide = temp[12] self.x_upper_lim = temp[11] self.x_lowr_lim = temp[10] self.y_upper_lim = temp[9] self.y_lowr_lim = temp[8] self.XY_slide_state = state_list[1] self.slide_x = state_list[4] self.slide_y = state_list[6] self.slide_x_err_code = state_list[10] self.slide_y_err_code = state_list[11] self.below_pier_temp = state_list[12] #need to divide by 10 self.at_PLC_temp = state_list[13] #need to divide by 10 self.at_Primary_mir_temp = state_list[14] #need to divide by 10 self.at_Secndary_mir_temp = state_list[15] #need to divide by 10 self.status0 = state_list[25] self.rotator0_ang = state_list[26] self.tel_parallactic_ang0 = state_list[27] self.t_p_a0rate = state_list[28] self.skyangle0 = state_list[29] self.status1 = state_list[30] self.rotator1_ang = state_list[31] self.tel_parallactic_ang1 = state_list[32] self.t_p_a1rate = state_list[33] self.skyangle1 = state_list[34] self.guide_flag = state_list[35] # guiding is off/on 0/1 self.guide_point = state_list[36] # guide point defined no/yes 0/1 self.NS_Flag = state_list[37] # NS guide flip off/on 0/1 self.EW_Flag = state_list[38] # EW guide flip off/on 0/1 self.fake_flag = state_list[39] # fake guiding off/on 0/1 self.xc0 = state_list[40] # x coordinate of guide point - if 0 then not defined self.yc0 = state_list[41] # y coordinate of guide point - if 0 then not defined self.ext_continuous = state_list[42] # 0 = continuous exposures 1 = exit continuous exp self.binning = state_list[43] # default is 2 but can be 1 self.xc = state_list[44] # x coordinate of guide star self.yc = state_list[45] # x coordinate of guide star self.message = state_list[46] # timestamped message from main_prog.c self.exptime = state_list[47] # The currently set exposure time self.no_lodestars = state_list[48] # class Guider(): def __init__(self, host=LESEDI_HOST, port=PORT, test=False): self.test = test #print("Test = ", test) # Spawn server thread if self.test: self.image_cx = 376 self.image_cy = 291 self.host = host self.port = port self.lock = threading.Lock() self.state = State() self.socket_connected = False def connect(self): self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(3) #self.sock.setblocking(False) # x = self.sock.getsockopt( socket.SOL_SOCKET, socket.SO_KEEPALIVE) # if( x == 0): # print('Socket Keepalive off, turning on') # x = self.sock.setsockopt( socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) # print('setsockopt=', x) # else: # print('Socket Keepalive already on') try: self.sock.connect((self.host, self.port)) except socket.timeout as e: print(f"{datetime.datetime.now().isoformat()} [ERROR][telem.py][guider.py][Guider] While connecting to Guider, got socket timeout: ", e) sleep(5) except socket.error as e: print(f"{datetime.datetime.now().isoformat()} [ERROR][telem.py][guider.py][Guider] While connecting to Guider, got socket error: ", e) self.sock.close() self.socket_connected = False sleep(5) except Exception as e: print(f"{datetime.datetime.now().isoformat()} [ERROR][telem.py][guider.py][Guider] While connecting to Guider got other error: ",e) self.sock.close() self.socket_connected = False sleep(5) else: print(f"{datetime.datetime.now().isoformat()} [INFO][telem.py][guider.py][Guider] Socket connection to Guider was successful") self.socket_connected = True print(f"{datetime.datetime.now().isoformat()} [INFO][telem.py][guider.py][Guider] Socket connected to Guider is ", self.socket_connected) def __del__(self): if hasattr(self, 'socket'): self.sock.close() # def disconnect(self): # if hasattr(self, 'socket'): # self.sock.close() def disconnect(self): if getattr(self, 'sock', None): try: self.sock.close() except Exception as e: print(f"Error closing socket: {e}") finally: self.sock = None self.socket_connected = False def translate_error(self, errval): """Translate a return value from send or receive commands to a human-readable error""" if errval == 0: return "No error" elif errval == -1: return "Connection not open" elif errval == -2: return "Operation timeout" elif errval == -3: return "Socket error" elif errval == -4: return "Max read attempts exceeded" else: return "Error reading or writing" def send_command(self, cmd): """Send the command to the socket. Attempts to connect to the socket if not connected. Returns 0 if message sent successfully, -1 if the connection was not open -2 if the send or recieve timed out -3 if there was another socket error -5 for other errors """ if self.test: print("Can't send command {} to remote server in test mode".format(cmd)) return -5 if not self.socket_connected: msg = "Unable to send command. Socket not currently connected." print(msg) self.connect() #print("Connection from possibly Mookodi attempted") return -1 try: #print("Sending command ", cmd) self.sock.sendall(cmd.encode('utf-8')) #print("Sent command ", cmd) print(f"{datetime.datetime.now().isoformat()} [INFO][main][guider.py][send_command] sent command to main_prog", cmd) except socket.timeout as e: print(f"{datetime.datetime.now().isoformat()} [ERROR][main][guider.py][send_command] Socket timeout sending to main_prog command {cmd}: {e} ") sleep(1) return -2 except socket.error as e: print(f"{datetime.datetime.now().isoformat()} [ERROR][main][guider.py][send_command] Socket error sending to main_prog command {cmd}: {e}") self.sock.close() self.socket_connected = False sleep(2) return -3 sleep(0.5) return 0 def receive_defined_data(self, msglen): """Read a message of defined length. Returns 0 if message sent successfully, -1 if the connection was not open -2 if the send or recieve timed out -3 if there was another socket error -4 if the number of tries exceeds the max set -5 for other errors """ chunks = [] bytes_recd = 0 max_tries = TIMEOUT_NUMTRIES tries =0 if msglen <= 0: print("Cant try to read 0 bytes") return None, -5 while bytes_recd < msglen: if not self.socket_connected: msg = "Unable to receive data. Socket not currently connected." print(msg) self.connect() print("Connection attempted") return None, -1 if tries >= max_tries: print("Couldn't read message after {} attempts".format(max_tries)) return None, -4 try: chunk = self.sock.recv(min(msglen - bytes_recd, 2048)) except socket.timeout as e: tries += 1 if tries < max_tries: print("Socket timeout receiving defined length data: {}. Wait {}s before trying to read again ".format(e, TIMEOUT_RETRY)) sleep(TIMEOUT_RETRY) except socket.error as e: print("Socket error: ", e) self.sock.close() self.socket_connected = False return None, -3 else: chunks.append(chunk) bytes_recd = bytes_recd + len(chunk) return b''.join(chunks), 0 def receive_freeform_data(self): """Read a message of undefined length. Returns 0 if message sent successfully, -1 if the connection was not open -2 if the send or recieve timed out -3 if there was another socket error -4 if the number of tries exceeds the max set -5 for other errors """ chunks = [] bytes_recd = 0 max_tries = TIMEOUT_NUMTRIES tries =0 if not self.socket_connected: msg = "Unable to receive data. Socket not currently connected." print(msg) self.connect() print("Connection attempted") return None, -1 while tries < max_tries: try: data = self.sock.recv(2048) except socket.timeout as e: tries += 1 if tries < max_tries: print("Socket timeout reading freeform data: {}. Wait {}s before trying to read again ".format(e, TIMEOUT_RETRY)) sleep(TIMEOUT_RETRY) except socket.error as e: print("Socket error: ", e) self.sock.close() self.socket_connected = False return None, -3 else: return data, 0 print("Couldn't read message after {} attempts".format(max_tries)) return None, -4 def get_image(self): """Get the image from the server. Returns the image as an array of data, as well as the values of the first two pixels e.g. image, p1, p2 """ print("Getting image") # The image is fixed at 582x752 pixels, even when binned img_h = 582 img_w = 752 bytes_per_pixel = 8 image_size = img_w * img_h * bytes_per_pixel if self.test: print("Getting dummy image") self.image_cx = self.image_cx + 1 self.image_cy = self.image_cy + 1 if self.image_cx > 376 + 50: self.image_cx = 376 - 50 if self.image_cy > 291 + 50: self.image_cy = 291 - 50 pixels = [] star_sd = 3 star_max_val = pow(2, 14) print("Cx = {} Cy = {}".format(self.image_cx, self.image_cy)) for i in range(img_h): for j in range(img_w): centre_dist = sqrt((self.image_cx - j)**2 + (self.image_cy - i)**2) if centre_dist <= star_sd * 4: val = star_max_val * exp( -1/2 * (centre_dist/star_sd)**2) pixels.append(val) else: pixels.append(0) else: print("Sending getimage command") self.lock.acquire() #print("Acquired lock") retval = self.send_command("getimage") if retval != 0: reason = self.translate_error(retval) print("Error sending getimage command: ", reason) self.lock.release() print("Released lock") sleep(1) return None, -1, -1 else: print("Image requested. Waiting for data...") pixels, retval = self.receive_defined_data(image_size) if retval != 0: reason = self.translate_error(retval) print("Error sending getimage command: ", reason) self.lock.release() print("Released lock") return None, -1, -1 pixels = [num[0] for num in struct.iter_unpack("L", pixels)] print("Got THE image 1", pixels[200:205]) self.lock.release() print("Released lock") image_size = (img_w, img_h) image_mode = 'I' image = Image.new(image_mode,image_size) image.putdata(pixels) with BytesIO() as output: image.save(output, format="PNG") contents = output.getvalue() print("Got THE image 3") return contents, pixels[0], pixels[1] def initialise_slide(self, **kwargs): # i think kwargs needs to be removed here """Request the remote end to update the slide. The slide number may be 0 or 1 but that is taken care of in the backend """ if self.test: self.state["slide_init"] = 1 return self.lock.acquire() #print("Acquired lock") retval = self.send_command("Initsld") self.lock.release() #print("Released lock") reason = self.translate_error(retval) if retval != 0: print("Initsld message not sent: ", reason) return retval, reason def reset_slide(self, **kwargs): # i think kwargs needs to be removed here """Request the remote end to reset the slide. The slide number may be 0 or 1 but that is taken care of in the backend """ if self.test: self.state["slide_reset"] = 1 return self.lock.acquire() print("Acquired lock") retval = self.send_command("resetsld") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("resetsld message not sent: ", reason) return retval, reason # def initialise_slide(self, **kwargs): # """Request the remote end to update the slide. # The slide number may be 0 or 1 # """ # slide_num = kwargs.get('slide_num', None) # if slide_num is None: # state = self.get_state() # slide_num = state['which_slide'] # print("Slide number not specified, but {} read from state".format(slide_num)) # print("Initialising slide ", slide_num) # if self.test: # self.state["slide_init"] = 1 # return # self.send_command("initsld{}".format(slide_num)) # def stop_slide(self, **kwargs): # slide_num = kwargs.get(slide_num, None) # if slide_num is None: # slide_num = self.which_slide # self.send_command("EStpsld{}".format(slide_num)) def stop_slide(self): """Request the remote end to stop the slide. The slide number may be 0 or 1 but that is taken care of in the backend """ self.lock.acquire() print("Acquired lock") retval = self.send_command("EStpsld") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("EStpsld message not sent: ", reason) return retval, reason def set_gotoXYcoords(self,gotoXYcoords): x = gotoXYcoords[0]; y = gotoXYcoords[1]; print("Guider: sending gotoXYsld " + x + y ) self.lock.acquire() print("Acquired lock") retval = self.send_command("gXYsld {} {} ".format(x,y)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("gXYsld message not sent: ", reason) return retval, reason def set_slide(self, slide_num): self.which_slide = slide_num self.lock.acquire() print("Acquired lock") retval = self.send_command("seltsld{}".format(slide_num)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("setsld message not sent: ", reason) return retval, reason def set_guide_coords(self, guide_coords): if self.test: self.state.xc0 = self.state.xc = guide_coords[0] self.state.yc0 = self.state.yc = guide_coords[1] print("Set guide coords to {} {}".format(guide_coords[0], guide_coords[1])) return #state = self.get_state() x = guide_coords[0] #state['xc0'] y = guide_coords[1] # state['yc0'] print("Sending mouse coords {} {}".format(x, y)) self.lock.acquire() print("Acquired lock") retval = self.send_command("gclick {} {} ".format(x,y)) # make sure there is a space at the end of this command self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("gclick message not sent: ", reason) return retval, reason def guide(self): print("Guider: sending GuideInv ") self.lock.acquire() print("Acquired lock") retval = self.send_command("GuideInv") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("GuideInv message not sent: ", reason) return retval, reason def guide_off(self): print("Guider: sending Guide_off ") self.lock.acquire() print("Acquired lock") retval = self.send_command("Guide_of") # needs to be 8 or less characters self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("Guide_off message not sent: ", reason) return retval, reason def guide_on(self): print("Guider: sending Guide_on ") self.lock.acquire() print("Acquired lock") retval = self.send_command("Guide_on") # needs to be 8 or less characters self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("Guide_on message not sent: ", reason) return retval, reason def fake(self): self.lock.acquire() print("Acquired lock") retval = self.send_command("FakeInv") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("FakeInv message not sent: ", reason) return retval, reason # def set_binning(self, binfact): # retval = self.send_command("Bin{}x{}".format(binfact, binfact)) def set_binning(self, binning): self.lock.acquire() print("Acquired lock") retval = self.send_command("Bin{}x{}".format(binning+1, binning+1)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("Bin message not sent: ", reason) return retval, reason def set_exposure_time(self, exposure_time): print("Guider: setting exposure time to ", exposure_time) if self.test: self.exposure_time = exposure_time self.lock.acquire() print("Acquired lock") retval = self.send_command("ExpTime {} ".format(exposure_time)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("ExpTime message not sent: ", reason) return retval, reason def set_goto_x_coord(self, goto_x_coord): # just set the value .... will be sent with gotoXY_slide print("Guider: setting goto_x_coord ", goto_x_coord) if self.test: self.goto_x_coord = goto_x_coord def set_goto_y_coord(self, goto_y_coord): # just set the value .... will be sent with gotoXY_slide print("Guider: setting goto_y_coord ", goto_y_coord) if self.test: self.goto_y_coord = goto_y_coord def start_exposures(self, continuous=True): if continuous: print("Guider: starting continuous exposures") self.lock.acquire() print("Acquired lock") retval = self.send_command("ContExp") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("ContExp message not sent: ", reason) else: print("Guider: starting single exposure") self.lock.acquire() print("Acquired lock") retval = self.send_command("Expose1") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("Expose1 message not sent: ", reason) return retval, reason def stop_exposures(self): print("Guider: stopping exposures") if self.test: return self.lock.acquire() print("Acquired lock") retval = self.send_command("StopExps") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("StopExps message not sent: ", reason) return retval, reason def get_state(self): if self.test: self.state.slide_x = random() self.state.slide_y = random() else: self.lock.acquire() print("Acquired lock") # Request state from back-end print("Requesting state from socket") retval = self.send_command("update") reason = self.translate_error(retval) #print("Result of update command: {} - {}".format(retval, reason)) if retval != 0: print("update message not sent: ", reason) self.lock.release() print("Released lock") return {"message": "Error getting state: {}".format(reason)} #print("Requested state from socket") sleep(0.5) # TBD: confirm if needed # Read the state. If the state isn't received within a short while, give up state_str, retval = self.receive_freeform_data() self.lock.release() #print("Released lock") reason = self.translate_error(retval) if state_str is None or len(state_str) == 0: print("Returning no state message") return {"message": "Error getting state: {}".format(reason)} #print("Got here Stephen: ") # print("Got raw state: ", state_str) try: state_str = state_str.decode('utf-8') except UnicodeDecodeError as e: msg = "Error getting state: {}".format(e) return {"message": msg} self.state.parse_state(state_str) #print("Retrieved state: ", self.state) if self.state.message is None: self.state.message = "" return self.state.__dict__ if __name__ == "__main__": #g = Guider(test=True) # g = Guider(host="10.2.2.31", test=False) g = Guider(host=LESEDI_HOST, test=False) g.connect() sleep(5) #print("Connected Encarni") while True: state = g.get_state() print("State: ") print(state) img, p1, p2 = g.get_image() if img is not None: with open("tmp.png", 'wb') as f: f.write(img) sleep(3) import sys sys.exit(0) print("Setting exposure time") g.set_exposure_time(2000) state = g.get_state() print("Exposure time was set to ", state["exptime"]) sleep(1) print("Doing single exposure") g.start_exposures(continuous=False) sleep(10) state = g.get_state() print("Continuously exposing? ", state["ext_continuous"]) print("Starting continuous exposures") g.start_exposures() sleep(3) state = g.get_state() print("Continuously exposing? ", state["ext_continuous"]) print("Stopping exposures") g.stop_exposures() sleep(5) state = g.get_state() print("Continuously exposing? ", state["ext_continuous"]) NUMCOUNTS = 1 count = 0 while count < NUMCOUNTS: count += 1 state = g.get_state() sleep(3) img = g.get_image() print("Image received") with open("tmp.png", 'wb') as f: f.write(img) sleep(3) print("Setting slide to 0") g.set_slide(0) sleep(1) state = g.get_state() print("Current slide: ", state["which_slide"]) sleep(1) # print("Setting slide to 1") # g.set_slide(1) # sleep(1) # state = g.get_state() # print(state) # sleep(1) # print("Initialising default slide") # g.initialise_slide() # state = g.get_state() # print("Current slide: {} Slide initialised: ".format(state["which_slide"], state["slide_init"])) # sleep(1) # print("Initialising slide 0") # g.initialise_slide(slide_num=0) # state = g.get_state() # print("Current slide: {} Slide initialised: ".format(state["which_slide"], state["slide_init"])) # Close the socket sleep(5) del g