# This Device Class is published under the terms of the MIT License. # Required Third Party Libraries, which are included in the Device Class # package for convenience purposes, may have a different license. You can # find those in the corresponding folders or contact the maintainer. # # MIT License # # Copyright (c) 2024 SweepMe! GmbH (sweep-me.net) # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. # SweepMe! driver # * Module: Logger # * Instrument: Ahlborn ALMEMO 2890-9 ### !!! Change the above license to your case and needs. !!! ### Although the above MIT license states that you have to add it to copies, ### we do not insist in that. Please feel free to not add the MIT license of this file. ### Each device class must have a name like: "-_". ### New drivers can also be created as a service. ### Please write to "contact@sweep-me.net" if you need a new driver of if you need support with creating one. ### import further python module here as usual, many packages come with SweepMe!, ### all other have to be shipped with the device class import pathlib import time ### If you like to see the source code of EmptyDevice, take a look at the pysweepme package that contains this file from pysweepme.EmptyDeviceClass import EmptyDevice from pysweepme.ErrorMessage import debug, error ### use the next two lines to add the folder of this device class to the PATH variable # from FolderManager import addFolderToPATH # addFolderToPATH() ### if you use 'addFolderToPATH', you can now important packages that are shipped with your device class class Device(EmptyDevice): ### here you can add html formatted description to your device class ### that is shown by modules like 'Logger' or 'Switch' that have a description field. def __init__(self): super().__init__() self.shortname = "ALMEMO 2890-9" # short name will be shown in the sequencer self.variables = [] # define as many variables you need self.units = [] # make sure that you have as many units as you have variables self.plottype = [] # True to plot data, corresponding to self.variables self.savetype = [] # True to save data, corresponding to self.variables for n_channel in range(0,16): self.variables.append("Var_Channel " + str(n_channel)) self.units.append("Unit_Channel " + str(n_channel)) self.plottype.append(True) self.savetype.append(True) ### use/uncomment the next line to use the port manager self.port_manager = True ### use/uncomment the next line to let SweepMe! search for ports of types supported by your instrument ### Also works if self.port_manager is False or commented. self.port_types = ["COM"] # available_ports = self.find_ports() # for port in available_ports: # try: # temp_port = serial.Serial(port) # temp_port.timeout = 1 # temp_port.baudrate = 9600 # temp_port.bytesize = 8 # temp_port.parity = 'N' # temp_port.stopbits = 1 # self.identify_channels(temp_port) # self.port = temp_port # except: # print("no available ports found.") ### use/uncomment the next lines to change port properties, ### you can find all keys here: https://wiki.sweep-me.net/wiki/Port_manager#Port_properties self.port_properties = { "timeout": 1, "baudrate": 9600, "EOLread": '\x03', "EOLwrite": '', "delay": 0.5, } self.data_separator = '/' self.message_terminator = '\x03' def set_GUIparameter(self): # add keys and values to generate GUI elements in the Parameters-Box # If you use this template to create a driver for modules other than Logger or Switch, # you need to use fixed keys that are defined for each module. GUIparameter = { "Channel-Count": len(self.variables), "Channel-List": self.variables, } ### Caution: ### Make sure that you do not use special strings ### such as 'Port', 'Comment', 'Device', 'Data', 'Description', or 'Parameters' ### These strings have special meaning and should not be overwritten. return GUIparameter def get_GUIparameter(self, parameter): ### see all available keys you get from the GUI #debug(parameter) ### the port selected by the user is readout here and saved in a variable ### that can be later used to open the correct port self.port_string = parameter["Port"] # use this string to open the right port object later during 'connect' self.channel_count = parameter["Channel-Count"] def update_GUIparameter(self, parameter): channel_count = parameter.get("Channel-Count") channel_list = parameter.get("Channel-List") new_parameters = { "Channel-Count": len(self.channel_name), "Channel-List": self.channel_name, } return new_parameters def apply_gui_parameters(self, parameters): self.channel_count = parameters.get("Channel-Count") # cut off additional variables self.variables = self.variables[0:self.channel_count-1] self.units = self.units[0:self.channel_count-1] self.plottype = self.plottype[0:self.channel_count-1] self.savetype = self.savetype[0:self.channel_count-1] # def find_ports(self): # """This function is called whenever the user presses 'Find ports' button. # No need to use, if you search for ports using self.port_types in __init__ # Function can be removed if not needed. # """ # debug("find_ports") # ### if you do not use the port manager you can use this function # ### to return a list of strings with possible port items # ### the next lines are an example how to find ports yourself # #import serial # #import serial.tools # if hasattr(self,"port_string"): # port_list = [self.port_string] # else: # port_list = [port.device for port in serial.tools.list_ports.comports()] # port_count = -1 # for port_add in port_list: # port_count += 1 # ser = serial.Serial() # # must be changed according to the device communicaton specifications # ser.port = port_add # ser.timeout = 0.1 # ser.baudrate = 9600 # ser.bytesize = 8 # ser.parity = 'N' # ser.stopbits = 1 # ser.xonxoff = False #disable software flow control # ser.rtscts = False #disable hardware (RTS/CTS) flow control # ser.dsrdtr = False #disable hardware (DSR/DTR) flow control # #ser.writeTimeout = 1 #timeout for write # comm_error = "" # try: # ser.open() # except: # print ("error open serial port: ") # break # if ser.isOpen(): # try: # ser.flushInput() #flush input buffer, discarding all its contents # ser.flushOutput() # ser.write("f1 t0".encode("utf-8")) # print("write data: f1 t0") # time.sleep(0.5) #give the serial port sometime to receive the data # idn = ser.read_until(b'/x03').decode("utf-8") # print("read data: " + idn) # if len(idn) > 1: # #self.shortname = idn[2] # break # ser.close() # except: # print ("error communicating...: ") # else: # print ("cannot open serial port ") # return [port_list[port_count]] def identify_channels(self): ### request_result self.port.write("S1") time.sleep(0.5) ### read_result raw_data = self.read_msg_until(self.message_terminator).split(self.data_separator) if (len(raw_data) > 1): raw_data = [item for item in raw_data if item] raw_data.pop(0) ### process data self.channel_name = [] self.channel_value = [] self.channel_units = [] self.sensor_type = [] result_cnt = 0 for result in raw_data: if (result_cnt == 0): self.timestamp = result if (result_cnt%4 == 1): self.channel_name.append(result) if (result_cnt%4 == 2): self.channel_value.append(float(result)) if (result_cnt%4 == 3): self.channel_units.append(result) if ((result_cnt%4 == 0) and (result_cnt != 0)): self.sensor_type.append(result) result_cnt += 1 def read_msg_until(self,terminator='\n',max_size=1000): bytes_count = 0 response_msg = '' while (bytes_count <= max_size): response_byte = self.port.read(1) if (response_byte == terminator): break if (response_byte == ''): response_byte = '/' response_msg += response_byte bytes_count += 1 return response_msg.replace('///////////','') ### -------------------------------------------------------------------------------------------- """ here, semantic standard functions start that are called by SweepMe! during a measurement """ ### all functions are overridden functions that are called by SweepMe! ### remove those function that you do not needed def connect(self): ### called only once at the start of the measurement ### this function 'connect' is typically not needed if the port manager is activated debug("->connect") ### if you do not use the port manager you can also create your own port objects # rm = visa.ResourceManager() # self.instrument = rm.open_resource(self.port_string) # debug("Instrument identification:", self.instrument.query('*IDN?')) ### of course you can use any other library to open your ports, e.g. pyserial ### just use 'find_Ports' to find all possible ports and then open them here ### based on the string 'self.port_string' that is created during 'get_GUIparameter' def disconnect(self): # called only once at the end of the measurement debug("->disconnect") def initialize(self): # called only once at the start of the measurement debug("->initialize") self.identify_channels() # debug("-> Tempfolder:", self.get_folder("TEMP")) # the folder in which all data is saved before saving # debug("-> External libs:", self.get_folder("EXTLIBS")) # the folder in which all data is saved before saving # debug("-> Custom files:", self.get_folder("CUSTOMFILES")) # the folder in which all data is saved before saving # debug("-> Driver folder:", self.get_folder("SELF")) # the folder where this file is in # In 'initialize' you can check whether the user input is valid. # If not you can abort the run by throwing an exception as shown in the lines below # msg = "Value of ... not valid. Please use ..." # raise Exception(msg) def deinitialize(self): # called only once at the end of the measurement debug("->deinitialize") def configure(self): # called if the measurement procedure enters a branch of the sequencer # and the module has not been used in the previous branch debug("-> configure") def reconfigure(self, parameters, keys): """'reconfigure' is called whenever parameters of the GUI change by using the {...}-parameter system.""" debug("-> reconfigure") # debug("-> Parameters:", parameters) # debug("-> Changed keys:", keys) ### The following two lines are the default behavior that is used by EmptyDevice ### if you do not override 'reconfigure' # self.get_GUIparameter(parameters) # self.configure() def start(self): """'start' can be used to do some first steps before the acquisition of a measurement point starts.""" debug("-> start") def apply(self): """'apply' is used to set the new setvalue that is always available as 'self.value'.""" # apply is not called in the module 'Logger' as logger cannot apply any value, # but it can be used in all other modules that have varying sweep values # apply is only called if the setvalue ("Sweep value") has changed debug("-> apply") # debug("-> New value to apply:", self.value) # self.value is a variable created by SweepMe! and stores the latest sweep value that should be applied # It can be any object. Please make sure to to test the type # and change it to the format you need before you send it to a device. def measure(self): """'measure' should be used to trigger the acquisition of new data. If all drivers use this function for this purpose, the data acquisition can start almost simultaneously. """ debug("-> measure") def request_result(self): """'request_result' can be used to ask an instrument to send data.""" debug("-> request_result") self.port.write("S1") #channel_count = 0 #for channel in self.channel_name: #self.port.write("M" + channel.replace(':','')) # self.port.write("p") # if (channel_count%2 == 1): # self.raw_data.append(self.read_msg_until(self.message_terminator)) # else: # temp_data = self.read_msg_until(self.message_terminator)) # channel_count += 1 def read_result(self): """'read_result' can be used get the data from a buffer that was requested during 'request_result'.""" debug("-> read_result") self.raw_data = self.read_msg_until(self.message_terminator).split(self.data_separator) if (len(self.raw_data) > 1): self.raw_data = [item for item in self.raw_data if item] self.raw_data.pop(0) def process_data(self): """'process_data' can be used for some evaluation of the data before it is returned.""" debug("-> process_data") if hasattr(self,"raw_data"): channel_name = [] channel_value = [] channel_units = [] sensor_type = [] result_cnt = 0 for result in self.raw_data: if (result_cnt == 0): timestamp = result if (result_cnt%4 == 1): channel_name.append(result) if (result_cnt%4 == 2): channel_value.append(float(result)) if (result_cnt%4 == 3): channel_units.append(result) if ((result_cnt%4 == 0) and (result_cnt != 0)): sensor_type.append(result) result_cnt += 1 self.value = channel_value else: self.value = [0]*len(self.channel_name) def call(self): """'call' is a mandatory function that must be used to return as many values as defined in self.variables. This function can only be omitted if no variables are defined in self.variables. """ # most import function: # return exactly the number of values that have been defined by self.variables and self.units debug("-> call") return self.value def finish(self): """'finish' can be used to do some final steps after the acquisition of a measurement point.""" debug("-> finish") #self.port.clear()