# 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 import serial ### 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 self.is_initialized = False ### 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"] ### 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": '', } self.data_separator = ' ' self.message_terminator = '\x03' self.measure_timer = 0 #s def update_gui_parameters(self,parameters): debug("updating parameters ...") if self.is_initialized: new_parameters = { "Port": self.port_string, "Channel-List": self.channel_name, "Measure-Timer": self.measure_timer, } else: new_parameters = { "Port": [port.device for port in serial.tools.list_ports.comports()], "Channel-List": [""], "Measure-Timer": 0, } return new_parameters def apply_gui_parameters(self, parameters): debug("apply parameters ...") if type(parameters["Measure-Timer"]) is float: self.measure_timer = parameters["Measure-Timer"] if (not self.is_initialized and len(parameters["Channel-List"]) == 0): ser = serial.Serial() # must be changed according to the device communicaton specifications ser.port = parameters["Port"] ser.timeout = 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 try: ser.open() except: print("could not open port") if ser.isOpen(): try: self.identify_channels(ser) ser.close() self.port_string = parameters["Port"] self.variables = self.channel_name self.units = self.channel_units for channel in self.channel_name: self.plottype.append(True) self.savetype.append(True) except: ser.close() print("Could not find ALMEMO on Port:" + ser.port) def identify_channels(self,port): ### request_result port.write("S1".encode("utf-8")) ### read_result raw_data = port.read_until(self.message_terminator).decode("utf-8","ignore").replace('\r\n',' ').replace(' ','').split(self.data_separator) #_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) raw_data.pop(len(raw_data)-1) print(raw_data) ### 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 if (len(raw_data) > 0): self.is_initialized = True # 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") # 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) self.t_start = time.time() #t_start = time.time() #for nCyc in range(1,1000): # self.measure() # self.process_data() #t_end = time.time() #print(t_end-t_start) 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") if (time.time() - self.t_start) > self.measure_timer: self.port.write("S1") self.raw_data = self.port.read().replace('\r\n',' ').replace(' ','').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) self.t_start = time.time() def request_result(self): """'request_result' can be used to ask an instrument to send data.""" debug("-> request_result") def read_result(self): """'read_result' can be used get the data from a buffer that was requested during 'request_result'.""" debug("-> read_result") 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.results = channel_value else: self.results = [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.results def finish(self): """'finish' can be used to do some final steps after the acquisition of a measurement point.""" debug("-> finish")