brukeropus.control.opus

  1import os
  2from brukeropus.control import DDEClient
  3from brukeropus import read_opus
  4
  5__docformat__ = "google"
  6
  7ERROR_CODES = {
  8    1: 'Not an Opus Command',
  9    2: 'Unknown Opus Command',
 10    3: 'Missing Square Bracket in Command',
 11    4: 'Function Not Available (Possible missing parameter)',
 12    5: 'Parameter Name Is Incorrect',
 13    6: 'Parameter Set Is Incomplete',
 14    7: 'File Parameter Is Incorrectly Formatted',
 15    8: 'File(s) Missing Or Corrupt',
 16    9: 'Opus Could Not Complete The Command',
 17}
 18
 19
 20class Opus:
 21    '''Class for communicating with currently running OPUS software using DDE interface.  Class automatically attempts
 22    to connect to OPUS software upon initialization.'''
 23    dde = None
 24    connected = False
 25    error_string = 'Error'
 26
 27    def connect(self):
 28        '''Connects class to OPUS software through the DDE interface.  Sets the `connected` attribute to `True` if
 29        successful.  By default, initializing an `Opus` class will automatically attempt to connect to OPUS.'''
 30        try:
 31            self.dde = DDEClient("Opus", "System")
 32            self.connected = True
 33        except Exception as e:
 34            self.connected = False
 35            raise Exception("Failed to connect to OPUS Software: " + str(e))
 36
 37    def disconnect(self):
 38        '''Disconnects DDE client/server connection.'''
 39        if self.connected:
 40            self.dde.__del__()
 41            self.dde = None
 42            self.connected = False
 43
 44    def raw_query(self, req_str: str, timeout=10000):
 45        '''Sends command/request string (`req_str`) to OPUS and returns the response in byte format.
 46
 47        Args:
 48            req_str: The request string to send to OPUS over DDE
 49            timeout: timeout in milliseconds.  If a response is not recieved within the timeout period, an exception is
 50                raised.
 51
 52        Returns:
 53            response: response from OPUS software through DDE request in bytes format.'''
 54        return self.dde.request(req_str, timeout)
 55
 56    def parse_response(self, byte_response: bytes, decode='ascii'):
 57        '''Parses the byte response from a raw DDE request query.  If an error is detected in the request, an Exception
 58        is raised.  If successful, a boolean, string or list of strings will be returned as appropriate.
 59
 60        Args:
 61            byte_response: response from OPUS software through DDE request in bytes format.
 62            decode: format used to decode bytes into string (e.g. 'ascii' or 'utf-8')
 63
 64        Returns:
 65            response: parsed response from OPUS software (bool, string, or list of strings depending on request)'''
 66        str_response = byte_response.decode(decode)
 67        responses = str_response.split('\n')
 68        responses = [r for r in responses if r != '']
 69        if len(responses) == 0:
 70            raise Exception('Error with DDE request: "' + req_str + '"; no response recieved...')
 71        elif responses[0].startswith(self.error_string):
 72            error = self._parse_error(responses[0])
 73            raise Exception('Error with DDE request: "' + req_str + '"; ' + error)
 74        else:
 75            responses = [r for r in responses if r != 'OK']
 76            if len(responses) == 0:
 77                return True
 78            elif len(responses) == 1:
 79                return responses[0]
 80            else:
 81                return responses
 82
 83    def _parse_error(self, response: str):
 84        try:
 85            code = int(response[response.rfind('ID: ') + 4:])
 86            if code in ERROR_CODES.keys():
 87                return ERROR_CODES[code]
 88            else:
 89                return response
 90        except:
 91            return response
 92
 93    def query(self, req_str: str, timeout=10000, decode='ascii'):
 94        '''Sends a command/request and returns the parsed response.
 95
 96        Args:
 97            req_str: The request string to send to OPUS over DDE
 98            timeout: timeout in milliseconds.  If a response is not recieved within the timeout period, an exception is
 99                raised.
100            decode: format used to decode bytes into string (e.g. 'ascii' or 'utf-8')
101
102        Returns:
103            response: parsed response from OPUS software (bool, string, or list of strings depending on request)
104        '''
105        response = self.raw_query(req_str=req_str, timeout=timeout)
106        return self.parse_response(response, decode=decode)
107
108    def close_opus(self):
109        '''Closes the OPUS application. Returns `True` if successful.'''
110        return self.query('CLOSE_OPUS')
111
112    def get_param_label(self, param: str):
113        '''Get the label for a three character parameter code (e.g. BMS, APT, DTC, etc...).
114
115        Args:
116            param: three character parameter code (case insensitive)
117
118        Returns:
119            label: short descriptive label that defines the parameter'''
120        return self.query('PARAM_STRING ' + param.upper())
121
122    def get_param_options(self, param: str):
123        '''Get the parameter setting options for a three character parameter code. Only valid for
124        enum type parameters (e.g. BMS, APT, DTC, etc...).
125
126        Args:
127            param: three character parameter code (case insensitive)
128
129        Returns:
130            options: list of valid options (strings) for the given parameter'''
131        result = self.query('ENUM_STRINGS ' + param.upper())
132        if type(result) is list:
133            return result[1:]
134        else:
135            return False
136
137    def get_version(self):
138        '''Get the OPUS software version information'''
139        return self.query('GET_VERSION_EXTENDED')
140
141    def get_opus_path(self):
142        '''Get the absolute path to the OPUS software directory (where PARAMTEXT.bin and other instrument specific files
143        are located)'''
144        return self.query('GET_OPUSPATH')
145
146    def send_command(self, text_command: str, timeout=10000):
147        '''Used to send "Direct Commands" to the optics bench. Useful for manually moving motors, etc. from accessories
148        and other low-level operations such as controlling the scanning mirror movement.
149
150        Examples:
151            send_command('VAC=5') # vents the sample compartment
152            send_command('VAC=4') # evacuates sample compartment
153
154        Args:
155            text_command: string command as you would enter into "Direct Command" input of OPUS
156            timeout: timeout in milliseconds to wait for response
157
158        Returns:
159            response: parsed response from OPUS software (typically boolean confirmation)'''
160
161        return self.query('SendCommand(0, {UNI=' + text_command + '})')
162
163    def evacuate_sample(self):
164        '''Evacuates the sample compartment'''
165        return self.send_command('VAC=4')
166
167    def vent_sample(self):
168        '''Vents the sample compartment'''
169        return self.send_command('VAC=5')
170
171    def close_flaps(self):
172        '''Closes vacumm flaps between optics bench and sample compartment'''
173        return self.send_command('FLP=1')
174
175    def open_flaps(self):
176        '''Opens vacumm flaps between optics bench and sample compartment'''
177        return self.send_command('FLP=0')
178
179    def unload_file(self, filepath: str):
180        '''Unloads a file from the OPUS software from its `filepath`
181
182        Args:
183            filepath: full path of the file to be unloaded in the software.
184
185        Returns:
186            response: `True` if successful.'''
187        return self.query('UNLOAD_FILE "' + filepath + '"')
188
189    def unload_all(self):
190        '''Unloads all files from OPUS software'''
191        return self.query('UnloadAll()')
192
193    def measure_ref(self, timeout=1000000, **kwargs):
194        '''Takes a reference measurement using the current settings from advanced experiment.  Also
195        takes option **kwargs input which use the OPUS 3-letter parameter keys and values as input
196        to customize the measurement.  example:
197
198            measure_ref(nrs=100, res=4) # measures reference with current settings but overriding averages to 100 and
199                resolution to 4
200
201        Args:
202            timeout: timeout in milliseconds to wait for response
203            kwargs: any valid three character parameter code (case insensitive)
204
205        Returns:
206            response: `True` if successful
207            '''
208        params = self._param_str(**kwargs)
209        ok = self.query('MeasureReference(0,' + params + ')', timeout=timeout)
210        return ok
211
212    def measure_sample(self, unload=False, timeout=1000000, **kwargs):
213        '''Takes a reference measurement using the current settings from advanced experiment.  Also
214        takes option **kwargs input which use the OPUS 3-letter parameter keys and values as input
215        to customize the measurement.  example:
216
217            measure_sample(nss=100, res=4) # measures sample with current settings but overriding averages to 100 and
218                resolution to 4
219
220        Args:
221            unload: whether to unload the file from OPUS after measurement is complete (to allow moving/renaming, etc.)
222            timeout: timeout in milliseconds to wait for response
223            kwargs: any valid three character parameter code (case insensitive)
224
225        Returns:
226            filepath: absolute filepath to measured sample file'''
227        params = self._param_str(**kwargs)
228        output = self.query('MeasureSample(0,' + params + ')', timeout=timeout)
229        filepath = output[1][1:-3]
230        if unload:
231            self.unload_file(filepath)
232        return filepath
233
234    def check_signal(self, nss=1, **kwargs):
235        '''Performs a quick (typically 1 sample) measurement using the current FTIR settings. Current settings can be
236        overridden using **kwargs. After measurement is finished, the file is unloaded from OPUS and deleted. The
237        function returns an `OPUSFile` object before it deletes the quick measurement file.
238
239        Args:
240            nss: number of sample scans to average (default is 1, i.e. no averaging)
241            kwargs: any valid three character parameter code (case insensitive)
242
243        Returns:
244            opus_file: `OPUSFile` object generated by quick measurement'''
245        filepath = self.measure_sample(unload=True, nss=nss, **kwargs)
246        opus_file = read_opus(filepath)
247        os.remove(filepath)
248        return opus_file
249
250    def save_ref(self):
251        '''Saves current reference to file (according to current filename and path set in advanced experiment) and
252        returns the filename.
253
254        Returns:
255            filepath: absolute path to saved reference file'''
256        output = self.query('SaveReference()')
257        filepath = output[1][1:-3]
258        return filepath
259
260    def _param_str(self, **kwargs):
261        '''Takes in an arbitrary number of: key=val kwargs and returns a param string of the following format:
262
263        _param_str(nss=100, res=4)
264        returns: {NSS=100,RES=4}
265
266        These param strings are used by the measure_ref and measure_sample functions to specify
267        experimental parameters.'''
268        params = []
269        for arg, val in kwargs.items():
270            params.append(arg.upper() + '=' + str(val))
271        if len(params) > 0:
272            return '{' + ','.join(params) + '}'
273        else:
274            return ''
275
276    def __bool__(self):
277        return self.connected
278
279    def __init__(self):
280        self.connect()
281        if self.connected:
282            self.opus_path = self.get_opus_path()