#!/usr/bin/env python # NOTE: updated for telemetry service # Author: Stephen Potter # Date: 4 July 2025 #from socket import socket import socket from threading import Thread from time import sleep from math import sin 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 = 2 PORT = 12351 # port address for telemetry socket thread HOST = "10.2.16.90" class State(): def __init__(self): self.guide_flag = 0 self.guide_point = -1 self.xc0 = 0 self.yc0 = 0 self.xc = 0 self.yc = 0 self.slide_x = 0 self.slide_y = 0 self.exptime = 0 self.XYSmessage = 0 self.ASmessage = 0 self.main_message = 0 self.tel_ra = "00 00 00" self.tel_dec = "00 00 00" self.tel_secz = 0 self.tel_ha = "00 00 00" self.SideRealTime = 0 self.tel_power = 0 self.tel_tracking = 0 self.RA_SLOW_FAST = 0 self.Dec_SLOW_FAST = 0 self.guide_mirror = 0 self.tcs_floor = 0 self.dome_shutter_stat = 0 self.mirror_cover = 0 self.dome_rot = 0 self.dome_pos = 0 self.dome_reqd = 0 self.julian_date = 0 self.UniversalTime = 0 self.SASTTime = 0 self.DateToday = 0 self.CoordEpoch = 0 self.tel_focus = 0 self.auto_tel = 0 self.acq_focus = 0 def __repr__(self): return """State: guide_flag = {} guide_point = {} xc0 = {} yc0 = {} xc = {} yc = {} slide_x = {} slide_y = {} exptime = {} XYSmessage = {} ASmessage = {} main_message = {} tel_ra = {} tel_dec = {} tel_secz = {} tel_ha = {} SideRealTime = {} tel_power = {} tel_tracking = {} RA_SLOW_FAST = {} Dec_SLOW_FAST = {} guide_mirror = {} tcs_floor = {} dome_shutter_stat = {} mirror_cover = {} dome_rot = {} dome_pos = {} dome_reqd = {} julian_date = {} UniversalTime = {} SASTTime = {} DateToday = {} CoordEpoch = {} tel_focus = {} auto_tel = {} acq_focus = {} """.format(self.guide_flag, self.guide_point, self.xc0, self.yc0, self.xc, self.yc, self.slide_x, self.slide_y, self.exptime, self.XYSmessage, self.ASmessage, self.main_message, self.tel_ra, self.tel_dec, self.tel_secz, self.tel_ha, self.SideRealTime, self.tel_power, self.tel_tracking, self.RA_SLOW_FAST, self.Dec_SLOW_FAST, self.guide_mirror, self.tcs_floor, self.dome_shutter_stat, self.mirror_cover, self.dome_rot, self.dome_pos, self.dome_reqd, self.julian_date, self.UniversalTime, self.SASTTime, self.DateToday, self.CoordEpoch, self.tel_focus, self.auto_tel, self.acq_focus ) def __str__(self): return self.__repr__() def parse_state(self, state_str): state_list = state_str.split('\n') self.guide_flag = state_list[0] # guiding is off/on 0/1 self.guide_point = state_list[1] # guide point defined no/yes 0/1 self.xc0 = state_list[2] # x coordinate of guide point - if 0 then not defined self.yc0 = state_list[3] # y coordinate of guide point - if 0 then not defined self.xc = state_list[4] # x coordinate of guide star self.yc = state_list[5] # x coordinate of guide star self.slide_x = state_list[6] # actual x slide coord self.slide_y = state_list[7] # actual y slide coord self.exptime = state_list[8] # The currently set exposure time ... in non zero than camera ding continuous integrations self.XYSmessage = state_list[9] # timestamped message from XY messages self.ASmessage = state_list[10] # timestamped message from guiding self.main_message = state_list[11] # timestamped general messages self.tel_ra = state_list[12] self.tel_dec = state_list[13] self.tel_secz = state_list[14] self.tel_ha = state_list[15] self.SideRealTime = state_list[16] self.tel_power = state_list[17] # 0=off 1=on self.tel_tracking = state_list[18] # 0=off 1=on self.RA_SLOW_FAST = state_list[19]# 0=slow 1=fast self.Dec_SLOW_FAST = state_list[20]# 0=slow 1=fast self.guide_mirror = state_list[21]# 0=out-of-beam 1=in-beam self.tcs_floor = state_list[22]# 0=floor 1=tcs self.dome_shutter_stat = state_list[23] #dome shutter status string- CLOSED or OPEN or MOVING or PART OPEN self.mirror_cover = state_list[24] #mirror_cover status string- CLOSED or OPEN or MOVING self.dome_rot = state_list[25] # Auto or Man self.dome_pos = state_list[26] # 0 - 359 %4.1f as a string self.dome_reqd = state_list[27] # 0 - 359 %4.1f as a string self.julian_date = state_list[28] self.UniversalTime = state_list[29] self.SASTTime = state_list[30] self.DateToday = state_list[31] self.CoordEpoch = state_list[32] self.tel_focus = state_list[33] self.auto_tel = state_list[34] self.acq_focus = state_list[35] print("In parse state tcs40in-sdk:telescope.py: ", state_list[21]) class Telescope(): def __init__(self, host=HOST, port=PORT, test=False): self.test = test print("Test = ", test) # Spawn server thread #if not self.test: # self.connect(host, port) self.host = host self.port = port self.lock = threading.Lock() self.state = State() self.connect() self.socket_connected = True # def connect(self, host, port): #self.sock = socket() #self.sock.connect((host, port)) def connect(self): self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(3) try: self.sock.connect((self.host, self.port)) except socket.timeout as e: print(f"{datetime.datetime.now().isoformat()} [ERROR][telem.py][telescope.py][Telescope] While connecting to Telescope, got socket timeout: ", e) sleep(5) except socket.error as e: print(f"{datetime.datetime.now().isoformat()} [ERROR][telem.py][telescope.py][Telescope] While connecting, 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][telescope.py][Telescope] While connecting got other error : ", e) self.sock.close() self.socket_connected = False sleep(5) else: print(f"{datetime.datetime.now().isoformat()} [INFO][telem.py][telescope.py][Telescope] Socket connection was successful ") self.socket_connected = True print(f"{datetime.datetime.now().isoformat()} [INFO][telem.py][telescope.py][Telescope] Socket connected is " , self.socket_connected) def __del__(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): if self.test: print("Can't send command {} to remote server in test mode".format(cmd)) return -5 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 attempted") return -1 try: #print("Sending command ", cmd) print(f"{datetime.datetime.now().isoformat()} [INFO][telem.py][telescope.py][Telescope] Sending command: ", cmd) self.sock.sendall(cmd.encode('utf-8')) except socket.timeout as e: print(f"{datetime.datetime.now().isoformat()} [ERROR][telem.py][telescope.py][Telescope] Socket timeout sending command {cmd}: {e}: ") sleep(1) return -2 except socket.error as e: print(f"{datetime.datetime.now().isoformat()} [ERROR][telem.py][telescope.py][Telescope] Socket error sending 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): # get the image """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_data(self, msglen): # from get_image .... replaced with receive_defined_data # chunks = [] # bytes_recd = 0 # while bytes_recd < msglen: # chunk = self.sock.recv(min(msglen - bytes_recd, 2048)) # if chunk == b'': # raise RuntimeError("socket connection broken") # chunks.append(chunk) # bytes_recd = bytes_recd + len(chunk) # return b''.join(chunks) def receive_freeform_data(self): # recieve the telescope status string """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 520x696 pixels (ULTRASTAR), even when binned #img_h = 520 #img_w = 696 # The image is fixed at 582x752 pixels (lodestar), 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 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_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("Telescope.py: 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_mirror_inv(self): print("Telescope: sending GuideMirr ") self.lock.acquire() print("Acquired lock") retval = self.send_command("GuideMirr") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("GuideMirr message not sent: ", reason) return retval, reason def set_exposure_time(self, exposure_time): print("Telescope.py: 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("Telescope: setting goto_x_coord ", goto_x_coord) if self.test: self.goto_x_coord = goto_x_coord # self.send_command("ExpTime {} ".format(exposure_time)) def set_goto_y_coord(self, goto_y_coord): # just set the value .... will be sent with gotoXY_slide print("Telescope: setting goto_y_coord ", goto_y_coord) if self.test: self.goto_y_coord = goto_y_coord def stop_exposures(self): print("Telescope.py: 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 tel_power(self): print("Telescope: sending TelPower ") self.lock.acquire() print("Acquired lock") retval = self.send_command("TelPower") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("TelPower message not sent: ", reason) return retval, reason def tel_track(self): print("Telescope: sending TelTrack ") self.lock.acquire() print("Acquired lock") retval = self.send_command("TelTrack") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("TelTrask message not sent: ", reason) return retval, reason def Ra_fast_slow(self): print("Telescope: sending RAFastSlow ") self.lock.acquire() print("Acquired lock") retval = self.send_command("RAFastSlow") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("RAFastSlow message not sent: ", reason) return retval, reason def Dec_fast_slow(self): print("Telescope: sending DecFastSlow ") self.lock.acquire() print("Acquired lock") retval = self.send_command("DecFastSlow") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("DecFastSlow message not sent: ", reason) return retval, reason def ALLSTOP(self): print("Telescope: sending ALLSTOP ") self.lock.acquire() print("Acquired lock") retval = self.send_command("ALLSTOP") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("ALLSTOP message not sent: ", reason) return retval, reason def TELSTOP(self): print("Telescope: sending TELSTOP ") self.lock.acquire() print("Acquired lock") retval = self.send_command("TELSTOP") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("TELSTOP message not sent: ", reason) return retval, reason def Req_TCS_Cont(self): print("Telescope: sending Req_TCS_cont ") self.lock.acquire() print("Acquired lock") retval = self.send_command("Req_TCS_Cont") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("Req_TCS_Cont message not sent: ", reason) return retval, reason def DomeAutoMan(self): print("Telescope: sending DomeAutoMan ") self.lock.acquire() print("Acquired lock") retval = self.send_command("DomeAutoMan") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("DomeAutoMan message not sent: ", reason) return retval, reason def Open_Dome_Shutter(self): print("Telescope: sending OpenDomeShutter ") self.lock.acquire() print("Acquired lock") retval = self.send_command("OpenDomeShutter") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("OpenDomeShutter message not sent: ", reason) return retval, reason def Clos_Dome_Shutter(self): print("Telescope: sending ClosDomeShutter ") self.lock.acquire() print("Acquired lock") retval = self.send_command("ClosDomeShutter") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("ClosDomeShutter message not sent: ", reason) return retval, reason def Stop_Dome_Shutter(self): print("Telescope: sending StopDomeShutter ") self.lock.acquire() print("Acquired lock") retval = self.send_command("StopDomeShutter") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("StopDomeShutter message not sent: ", reason) return retval, reason def Open_Mirr_Cover(self): print("Telescope: sending OpenMirrCover ") self.lock.acquire() print("Acquired lock") retval = self.send_command("OpenMirrCover") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("OpenMirrCover message not sent: ", reason) return retval, reason def Clos_Mirr_Cover(self): print("Telescope: sending ClosMirrCover ") self.lock.acquire() print("Acquired lock") retval = self.send_command("ClosMirrCover") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("ClosMirrCover message not sent: ", reason) return retval, reason def Switch_Circle_Lights(self): # no status associated with lights ... switch only print("Telescope: sending SwitchCircLight ") self.lock.acquire() print("Acquired lock") retval = self.send_command("SwitchCircLight") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("SwitchCircLight message not sent: ", reason) return retval, reason def Switch_Floures_Lights(self): # no status associated with lights ... switch only print("Telescope: sending SwitchFlresLight ") self.lock.acquire() print("Acquired lock") retval = self.send_command("SwitchFlresLight") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("SwitchFlresLight message not sent: ", reason) return retval, reason def Raise_Wind_Blind(self): print("Telescope: sending RaiseWindBlind ") self.lock.acquire() print("Acquired lock") retval = self.send_command("RaiseWindBlind") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("RaiseWindBlind message not sent: ", reason) return retval, reason def Lower_Wind_Blind(self): print("Telescope: sending LowerWindBlind ") self.lock.acquire() print("Acquired lock") retval = self.send_command("LowerWindBlind") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("LowerWindBlind message not sent: ", reason) return retval, reason def Stop_Wind_Blind(self): print("Telescope: sending StopWindBlind ") self.lock.acquire() print("Acquired lock") retval = self.send_command("StopWindBlind") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("StopWindBlind message not sent: ", reason) return retval, reason def Park_Tel_Dome(self): print("Telescope: sending ParkTelDome ") self.lock.acquire() print("Acquired lock") retval = self.send_command("ParkTelDome") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("ParkTelDome message not sent: ", reason) return retval, reason def Tel2Zenith(self): print("Telescope: sending Tel2Zenith ") self.lock.acquire() print("Acquired lock") retval = self.send_command("Tel2Zenith") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("Tel2Zenith message not sent: ", reason) return retval, reason def set_gotoTargetcoords(self,gotoTargetcoords): x = gotoTargetcoords[0]; y = gotoTargetcoords[1]; print("Telescope: sending gotoTargetcoords " + x + y ) self.lock.acquire() print("Acquired lock") retval = self.send_command("Tel2Target {} {} {} ".format(x," ",y)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("Tel2Target message not sent: ", reason) return retval, reason def set_tel_focus(self, val_tel_focus_input): x = val_tel_focus_input[0]; print("Telescope: setting telescope focus to ", x) self.lock.acquire() print("Acquired lock") retval = self.send_command("TelFocs {} ".format(x)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("TelFocs message not sent: ", reason) return retval, reason def set_acq_focus(self, val_acq_focus_input): x = val_acq_focus_input[0]; print("Telescope: setting camera focus to ", x) self.lock.acquire() print("Acquired lock") retval = self.send_command("AcqFocs {} ".format(x)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("AcqFocs message not sent: ", reason) return retval, reason def NOWstop_acq_focus(self): print("Telescope: stop camera focus ") self.lock.acquire() print("Acquired lock") retval = self.send_command("StopAcqFocs") self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("StopAcqFocs message not sent: ", reason) return retval, reason def set_nudge_ra_plus(self, val_ra_nudge): x = val_ra_nudge[0]; print("Telescope: setting ra nudge to ", x) self.lock.acquire() print("Acquired lock") retval = self.send_command("RaNPlus {} ".format(x)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("RaNPlus message not sent: ", reason) return retval, reason def set_nudge_ra_minus(self, val_ra_nudge): x = val_ra_nudge[0]; print("Telescope: setting ra nudge to ", x) self.lock.acquire() print("Acquired lock") retval = self.send_command("RaNMins {} ".format(x)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("RaNMins message not sent: ", reason) return retval, reason def set_nudge_dec_plus(self, val_dec_nudge): x = val_dec_nudge[0]; print("Telescope: setting dec nudge to ", x) self.lock.acquire() print("Acquired lock") retval = self.send_command("DecPlus {} ".format(x)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("DecPlus message not sent: ", reason) return retval, reason def set_nudge_dec_minus(self, val_dec_nudge): x = val_dec_nudge[0]; print("Telescope: setting dec nudge to ", x) self.lock.acquire() print("Acquired lock") retval = self.send_command("DecMins {} ".format(x)) self.lock.release() print("Released lock") reason = self.translate_error(retval) if retval != 0: print("DecMins 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) #if self.state.main_message is None: # self.state.main_message = "" return state_str #return self.state.__dict__ if __name__ == "__main__": #g = Telescope(test=True) g = Telescope(test=False) sleep(2) print("Getting state") state = g.get_state() print("State: ", state) #print("Setting exposure time") #g.set_exposure_time(2000) sleep(1) state = g.get_state() print("Exposure time was set to ", state["exptime"]) sleep(3) # print("Stopping exposures") # g.stop_exposures() # sleep(5) # state = g.get_state() NUMCOUNTS = 4 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("Initialising default slide") # g.initialise_slide() # Close the socket # sleep(3) del g