#!/usr/bin/env python # -*- encoding: utf-8 -*- # # ============================================================================= # # LSST / StarDICE # # Low level control for the ArDICE source # # Authors: Marc Betoule, Laurent Le Guillou # # version 2019-05-05 # This version should work with Python 2.7 & Python 3.6 (LLG) # # ============================================================================= import sys import os, os.path import time import datetime # import socket import telnetlib # ============================================================================= class ArdiceError(Exception): def __init__(self, value): self.value = value def __str__(self): return repr(self.value) # ============================================================================= class Ardice(object): default_host = '192.168.1.177' default_port = 8756 # default_nleds = 16 default_nleds = 64 default_currents = { 0: 1000, 1: 1142, 2: 440, 3: 313, 4: 431, 5: 387, 6: 1512, 7: 1122, 8: 473, 9: 341, 10: 273, 11: 226, 12: 206, 13: 360, 14: 396, 15: 372, 16: 0, 17: 0, 18: 0, 19: 0, 20: 0, 21: 0, 22: 0, 23: 0, 24: 0, 25: 0, 26: 0, 27: 0, 28: 0, 29: 0, 30: 0, 31: 0, 32: 0, 33: 0, 34: 0, 35: 0, 36: 0, 37: 0, 38: 0, 39: 0, 40: 0, 41: 0, 42: 0, 43: 0, 44: 0, 45: 0, 46: 0, 47: 0, 48: 0, 49: 0, 50: 0, 51: 0, 52: 0, 53: 0, 54: 0, 55: 0, 56: 0, 57: 0, 58: 0, 59: 0, 60: 0, 61: 0, 62: 0, 63: 0 } # -------------------------------------------------------------- def __init__(self, host = default_host, port = default_port, debug = True): """ Create an ArDICE instance. Does not open the connection (yet), see open(). """ self.host = host self.port = port self.timeout = 0.5 # Non-blocking & non waiting mode self.EOL = '\n' self.EOM = self.EOL + self.EOL # end of answer from the ArDICE controller # self.MAXBUFFER = 1024 self.socket = None # ---- debug mode self.debug = debug # ---- timeout to act self.action_timeout = 0.1 # -------------------------------------------------------------- def open(self): """ Open and initialize the Ethernet port to communicate with the ArDICE device. """ if self.debug: sys.stderr.write( "ArDICE: Opening socket to %s:%d ...\n" % ( self.host, self.port ) ) # self.socket = socket.create_connection(((self.host), self.port), self.timeout) self.socket = telnetlib.Telnet(self.host, self.port, self.timeout) if not(self.echotest()): raise ArdicelError( "ArDICE: No answer (*IDN?) from the device. Stop.") if self.debug: sys.stderr.write( "ArDICE: Opening socket to %s:%d done.\n" % ( self.host, self.port ) ) # -------------------------------------------------------------- def close(self): """ Close the socket. """ self.socket.close() # -------------------------------------------------------------- def __del__(self): self.close() # -------------------------------------------------------------- # # Basic I/O with debugging information def write(self, command): """ Send a command through the Ethernet port """ if self.debug: sys.stderr.write("ArDICE: Sending command [" + command + "]\n") # self.socket.send(command + self.EOL) if (self.socket is None) or self.socket.eof: raise IOError("ArDICE connection closed (or not yet opened). Invoke open() method.") msg = (command + self.EOL).encode('ASCII') self.socket.write(msg) def read(self, timeout = None): """ Read the answer from the Ethernet port. Return it as a string. """ if self.debug: sys.stderr.write("ArDICE: Reading the Ethernet port buffer\n") # msg = self.socket.recv(self.MAXBUFFER) bytesmsg = self.socket.read_until(self.EOM.encode('ASCII')) msg = bytesmsg.decode('ASCII') if self.debug: sys.stderr.write("ArDICE: Received [" + repr(msg) + "]\n") return msg # -------------------------------------------------------------- def send(self, command, timeout = None): """ Send a command through the Ethernet port. Read the answer. Answer from ArDICE is multilines : ---------------------------------- XX Status message Extra info (if any) ---------------------------------- Return the formated answer as a tuple: ( (status_number, status_msg), answer_msg ) """ # Sending the command self.write(command) time.sleep(self.action_timeout) # needed to read the whole buffer # Reading the command msg = self.read() answers = msg.strip().splitlines() if len(answers) < 1: raise IOError("ArDICE: malformed request answer [empty answer].") # Remove lines starting with a '#' (debug messages) # python2: answers = filter(lambda line: (len(line) == 0) or (line[0] != '#'), answers) answers = list(filter(lambda line: (len(line) == 0) or (line[0] != '#'), answers)) status_line = answers[0].strip() status_line_elts = status_line.split() # print status_line_elts if len(status_line_elts) < 2: raise IOError("ArDICE: malformed request answer [malformed status line].") status = int(status_line_elts[0]) status_msg = " ".join(status_line_elts[1:]) extra_info = [] if len(answers) > 1: extra_info = [ans.strip() for ans in answers[1:]] if self.debug: sys.stderr.write("ArDICE: Status number: " + str(status) + '\n' + \ "ArDICE: Status message: " + status_msg + '\n') if extra_info != []: sys.stderr.write("ArDICE: Answer extra info: " + str(extra_info) + "\n") return status, status_msg, extra_info # ------------------------------------------------------------- def echotest(self): """ Verify communications with the ArDICE Arduino Controller. Should return True if the communication has been established, and False otherwise. """ # Send the command *IDN? to get the apparatus ID self.write("*IDN?") answer = self.read().strip() if not(answer): return False return True # -------------------------------------------------------------- def get_serial(self): status, status_msg, extra_info = self.send("*IDN?") if status != 0: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "*IDN? command failed (invalid answer).") if len(extra_info) < 1: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "*IDN? command failed (incomplete answer).") return extra_info[0] # -------------------------------------------------------------- def set_ledmask(self, mask): """ Set the LED binary mask : for each LED, 0 = OFF, 1 = ON. mask is a long long unsigned integer """ # first convert to binary, remove the '0b' prefix and reverse binrevmask = bin(mask)[2:][::-1] command = "MASK %s" % binrevmask status, status_msg, extra_info = self.send(command) if status != 0: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "%s command failed." % command) def get_ledmask(self): """ Read the LED binary mask : for each LED, 0 = OFF, 1 = ON. """ command = "MASK?" status, status_msg, extra_info = self.send(command) if status != 0: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "%s command failed." % command) if len(extra_info) < 1: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "*IDN? command failed (incomplete answer).") mask = int("0b" + (extra_info[0].strip())[::-1], 2) return mask ledmask = property(get_ledmask, set_ledmask, doc = "ArDICE LEDs binary mask (0=Off,1=On)") # -------------------------------------------------------------- def set_ledcurrent(self, led, current=None): """ Set the LED current value for LED . If is not specified, a default nominal value is used instead. """ if (led < 0) or (led >= self.default_nleds): raise ValueError("invalid LED channel, should be in [0-%d]" % self.default_nleds) if current is None: reqcurrent = self.default_currents[led] else: if (current < 0) or (current >= 16383): raise ValueError("invalid LED current value, should be in [0-16383]") reqcurrent = current command = "CURR %d %d" % (led, reqcurrent) status, status_msg, extra_info = self.send(command) if status != 0: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "%s command failed." % command) def get_ledcurrents(self): """ Get the LED current value for all LEDs. Returns the values as a list. """ command = "CURR?" status, status_msg, extra_info = self.send(command) if status != 0: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "%s command failed." % command) if len(extra_info) < 1: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "*IDN? command failed (incomplete answer).") try: currents = [int(v) for v in extra_info[0].strip().replace('[','').replace(']','').split(',')] except ValueError: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "%s command failed: malformed answer[%s]." % (command, extra_info) ) return currents # -------------------------------------------------------------- def led_off(self, led): # Operation order is important to avoid damages on the UV LEDs self.set_ledcurrent(led, 0) # mask = self.get_ledmask() mask = self.ledmask if ( mask & (1 << led) ): # if this led is on # self.set_ledmask(mask ^ 2**led) self.ledmask = mask ^ (1 << led) def led_on(self, led, current = None): # Operation order is important to avoid damages on the UV LEDs # First set the current to zero self.set_ledcurrent(led, 0) # then turn it on # mask = self.get_ledmask() mask = self.ledmask if not( mask & (1 << led) ): # if this led is off # self.set_ledmask(mask | 2**led) self.ledmask = mask | (1 << led) # then set the proper current self.set_ledcurrent(led, current) # None case already managed def all_led_off(self): # self.set_ledmask(0) self.ledmask = 0 # -------------------------------------------------------------- def shutter_close(self): command = "SC" status, status_msg, extra_info = self.send(command) if status != 0: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "%s command failed." % command) def shutter_open(self): command = "SO" status, status_msg, extra_info = self.send(command) if status != 0: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "%s command failed." % command) def set_shutter(self, state): """ Get the current shutter state: 0 = closed, 1 = Opened. """ if state == 0: self.shutter_close() else: self.shutter_open() def get_shutter(self): """ Get the current shutter state: 0 = closed, 1 = Opened. """ command = "S?" status, status_msg, extra_info = self.send(command) if status != 0: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "%s command failed." % command) if len(extra_info) < 1: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "*IDN? command failed (incomplete answer).") try: shutter_state = int(extra_info[0].split()[0]) except ValueError: raise ArDICE( ( "ArDICE: status = %d (%s)" % (status, status_msg) ) + "%s command failed: malformed answer[%s]." % (command, extra_info) ) return shutter_state shutter = property(get_shutter, set_shutter, doc = "ArDICE shutter state (0=Closed,1=Open)") # ----------------------------------------------------------------------------- # Aliases Ardice.sclose = Ardice.shutter_close Ardice.sopen = Ardice.shutter_open # =============================================================================