# ***************************************************************************** # This program is free software; you can redistribute it and/or modify it under # the terms of the GNU General Public License as published by the Free Software # Foundation; either version 2 of the License, or (at your option) any later # version. # # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. See the GNU General Public License for more # details. # # You should have received a copy of the GNU General Public License along with # this program; if not, write to the Free Software Foundation, Inc., # 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # # Module authors: # Markus Zolliker # ***************************************************************************** """Andeen Hagerling capacitance bridge creates up to two additional modules for 'loss' and 'freq' in the configuration file, only the capacitance module needs to be configured, while the loss module will be created automatically. the name of the loss (and freq) module may be configured, or disabled by choosing an empty name """ import re import time import threading from frappy.core import HasIO, Parameter, Readable, StringIO, \ Attached, Property, Command, Writable, BUSY, IDLE, WARN from frappy.datatypes import FloatRange, IntRange, StringType, TupleOf from frappy.modules import Acquisition from frappy.dynamic import Pinata from frappy.errors import ProgrammingError, IsBusyError class IO(StringIO): end_of_line = ('\r\n', '\r') # for writing, '\r\n' also accepted on AH2700 # for reading, '\r\n' would be correct, but '\r' does also works # when stripping the reply timeout = 5 @Command(StringType(), result=StringType()) def communicate(self, command, noreply=False): """communicate and save the last command""" # this is also called by writeline reply = super().communicate(command, noreply) return reply and reply.strip() CONTINUOUS = 0 STARTING = 1 RUNNING = 2 FINISHED = 3 class AHBase(HasIO, Pinata, Acquisition): value = Parameter('capacitance', FloatRange(unit='pF')) freq = Parameter('frequency', FloatRange(unit='Hz'), default=1000) voltage = Parameter('upper voltage limit', FloatRange(0, 15, unit='V', fmtstr='%.1f'), readonly=False, default=0) loss = Parameter('loss', FloatRange(unit=''), default=0) averexp = Parameter('base 2 exponent of goal', IntRange(0, 15), readonly=False, default=0) goal = Parameter('value for averexp for the next go()', IntRange(0, 15), readonly=False, default=0) meas_time = Parameter('measured measuring time', FloatRange(unit='s', fmtstr='%.4g'), default=0) calculated_time = Parameter('calculated measuring time', FloatRange(unit='s', fmtstr='%.4g'), default=0) loss_module = Property( '''name of loss module (default: _loss) configure '' to disable the creation of the loss module ''', StringType(), default='') pollinterval = Parameter(datatype=FloatRange(0.001, 1), export=False, value=0.001) export = True # for a Pinata module, the default is False! ioClass = IO COMMANDS = ['AV', 'VO', 'SI', 'SH', 'FR'] _error = '' _last_start = None _params = None _mode = CONTINUOUS # or RUNNING or FINISHED _cont_deadline = 0 # when to switch back to continuous after finished _averexp_deadline = 0 # to make sure averexp is polled periodically # to be overridden: PATTERN = None # a list of patterns to parse replies MEAS_PAT = None # the pattern to parse the measurement reply def scanModules(self): if self.loss_module: # if loss_module is not empty, we tell the framework to create # a module for the loss with this name, and config below yield self.loss_module.replace('$', self.name), { 'cls': Loss, 'description': f'loss value of {self.name}', 'cap': self.name} def initModule(self): self.io.setProperty('identification', [('\rSERIAL ECHO OFF;SH MODEL', 'ILLEGAL WORD: MODEL')]) super().initModule() pattern = self.PATTERN + [self.MEAS_PAT] self.pattern = {p[:2]: re.compile(p) for p in pattern} if len(self.pattern) != len(pattern): raise ProgrammingError('need more than two letters ' 'to distinguish patterns') self.echo = re.compile('|'.join(self.COMMANDS)) self._params = {} self._lock = threading.RLock() def initialReads(self): # UN 2 does also return the results of the last measurement # (including the frequency for AH2700) self.io.writeline('SH FR;UN 2') self.freq = self.interprete('freq') self.verify_averexp() self.goal = self.averexp self.single_meas() def interprete(self, wait_for=None, tmo=None): """ :param wait_for: name of parameter to wait for or None to wait for measurement :param tmo: :return: """ if tmo is None: tmo = self.io.timeout reply = self.io.readline(tmo) now = time.time() while reply: pattern = self.pattern.get(reply[:2]) if pattern: match = pattern.match(reply) if not match: self.log.warning('unexpected reply syntax: %r', reply) break values = match.groupdict() if len(values) == 1: key, value = next(iter(values.items())) self._params[key] = float(value) else: self._params['meas'] = values elif self.echo.match(reply): # this is probably an echo # we may have lost connection, so query again averexp self.writeline('\r\nSERIAL ECHO OFF;SH AV') elif reply: self.log.warning('unknown reply %r', reply) if (wait_for or 'meas') in self._params: break reply = self.io.readline(tmo) self.log.debug('doPoll %r params %r wait_for %r %d', reply, list(self._params), wait_for, self._mode) result = self._params.pop(wait_for, None) if self._mode == FINISHED and now > self._cont_deadline: self._mode = CONTINUOUS self.status = IDLE, '' self.single_meas() return if result is None and wait_for: self.log.info(f'missing reply for {wait_for}') return getattr(self, wait_for, None) return result def change(self, short, value, param): if self._mode == RUNNING: raise IsBusyError('can not change parameters while measuring') with self._lock: self.io.writeline(f'{short} {value};SH {short}') result = self.interprete(param) self.retrigger_meas() return result def retrigger_meas(self): if self._mode == CONTINUOUS: self.single_meas() def single_meas(self): self._last_start = time.time() self.writeline('SI') def doPoll(self): # this polls measurement results # we can not do polling of other parameters, as they would # interrupt measurements. averexp needs a special treatment self.interprete(tmo=1) with self._lock: for param in list(self._params): value = self._params.pop(param, None) if param == 'meas': self.update_meas(**value) self.retrigger_meas() elif param == 'averexp': self.update_averexp(value) elif param == 'freq': self.update_freq(value) elif param == 'voltage': self.voltage = value def update_freq(self, value): self.freq = value self._calculate_time(self.averexp, value) def update_averexp(self, value): self.averexp = value self._averexp_deadline = time.time() + 15 self._calculate_time(value, self.freq) def update_meas(self, cap, loss, lossunit, voltage, error, freq=None): """update given arguments (these are strings!)""" self._error = error if self._error: status = WARN, self._error else: status = IDLE, '' if self._mode == CONTINUOUS else 'finished' if status != self.status: self.status = status now = time.time() if self._mode == RUNNING: self._cont_deadline = now + 5 self._mode = FINISHED if freq: self.freq = float(freq) self._calculate_time(self.averexp, self.freq) self.value = float(cap) self.voltage = float(voltage) if lossunit != self.UNIT: if self._last_start == 0: self.log.warning('change unit to %s failed', self.UNIT) else: self.io.writeline('UN 2') # this will trigger a measurement reply # skip calculation of meas_time while interpreting result self._last_start = 0 self.interprete('meas') self.retrigger_meas() return if self._last_start: self.meas_time = now - self._last_start self._last_start = 0 if now > self._averexp_deadline and self._mode == CONTINUOUS: self.verify_averexp() self.loss = float(loss) def write_voltage(self, value): return round(self.change('VO', f'{value:.1f}', 'voltage'), 1) def write_averexp(self, value): self.update_averexp(self.change('AV', f'{value}', 'averexp')) def verify_averexp(self): # we do not want to use read_averexp for this, # as it will stop the measurement when polled self.io.writeline('SH AV') self.update_averexp(self.interprete('averexp')) def _calculate_time(self, averexp, freq): self.calculated_time = self.calculate_time(averexp, freq) def go(self): """start acquisition""" prevmode = self._mode self._mode = STARTING if prevmode != FINISHED or time.time() > self._averexp_deadline: # this also makes sure we catch a previous meas reply self.verify_averexp() if self.averexp != self.goal: self.log.info('changed averexp') self.write_averexp(self.goal) self.status = BUSY, 'started' self.single_meas() self._mode = RUNNING def stop(self): """stops measurement""" if self._mode == RUNNING: self.verify_averexp() self.status = WARN, 'stopped' self._mode = FINISHED self._cont_deadline = time.time() + 5 class Loss(Readable): cap = Attached() value = Parameter('loss', FloatRange(unit=''), default=0) def initModule(self): super().initModule() self.cap.addCallback('loss', self.update_loss) # auto update status def update_freq(self, freq): self.freq = float(freq) def update_loss(self, loss): self.value = float(loss) class Freq(Writable): cap = Attached() value = Parameter('', FloatRange(unit='Hz'), default=0) def initModule(self): super().initModule() self.cap.addCallback('freq', self.update_freq) # auto update status def update_freq(self, freq): self.value = freq def write_target(self, target): self.cap.write_freq(target) class AH2550(AHBase): PATTERN = [ r'AVERAGE_AVEREXP *(?P[0-9]*)', r'VOLTAGE_HIGHEST *(?P[0-9.E+-]+)', r'FREQUENCY *(?P[0-9.E+-]+)', ] MEAS_PAT = ( r'C= *(?P[0-9.E+-]+) *PF,' r'L= *(?P[0-9.E+-]+) *(?P[A-Z]*),' r'V= *(?P[0-9.E+-]+) *V,A,*(?P.*)$' ) UNIT = 'DF' # empirically determined - may vary with noise # differs drastically from the table in the manual MEAS_TIME_CONST = [0.2, 0.3, 0.4, 1.0, 1.3, 1.6, 2.2, 3.3, 5.5, 8.3, 14, 25, 47, 91, 180, 360] def initModule(self): self.io.setProperty('identification', [('\rSERIAL ECHO OFF;SH MODEL', 'ILLEGAL WORD: MODEL')]) super().initModule() def _calculate_time(self, averexp, freq): self.calculated_time = self.calculate_time(averexp) @Command(TupleOf(IntRange(0, 15)), result=FloatRange()) def calculate_time(self, averexp): """calculate estimated measuring time""" return self.MEAS_TIME_CONST[int(averexp)] class AH2700(AHBase): freq = Parameter(datatype=FloatRange(50, 20000, unit='Hz', fmtstr='%.1f'), readonly=False) freq_module = Property('''name of freq module default: not created ''', StringType(), default='') PATTERN = [ r'AVERAGE *AVEREXP=(?P[0-9]*)', r'VOLTAGE HIGHEST *(?P[0-9.E+-]+)', r'FREQUENCY *(?P[0-9.E+-]+)', ] UNIT = 'DS' MEAS_PAT = ( r'F= *(?P[0-9.E+-]+) *HZ ' r'C= *(?P[0-9.E+-]+) *PF ' r'L= *(?P[0-9.E+-]+) *(?P[A-Z]*) ' f'V= *(?P[0-9.E+-]+) *V *(?P.*)$' ) def initModule(self): super().initModule() self.io.setProperty('identification', [('\r\nSERIAL ECHO OFF;SH MODEL', 'MODEL/OPTIONS *AH2700')]) def scanModules(self): yield from super().scanModules() if self.freq_module: yield self.freq_module.replace('$', self.name), { 'cls': Freq, 'description': f'freq module of {self.name}', 'cap': self.name} def write_freq(self, value): self.change('FR', f'{value:g}', 'freq') self.update_freq(value) return round(value, 1) MEAS_TIME_CONST = [ # (upper freq limit, meas time @ avrexp=7 - 0.8) (75, 20), (150, 10), (270, 5.62), (550, 2.34), (1100, 2.73), (4500, 1.02), (20000, 0.51), ] @Command(TupleOf(IntRange(0, 15), FloatRange(50, 20000)), result=FloatRange()) def calculate_time(self, averexp, freq): """calculate estimated measuring time from time efficiency considerations averexp > 7 is recommended especially for freq < 550 no time may be saved with averexp <= 7 """ for f, c in self.MEAS_TIME_CONST: if f > freq: const = c break else: const = self.MEAS_TIME_CONST[-1][1] if averexp >= 8: result = 0.8 + const * (0.5 + 2 ** (averexp - 8)) elif freq < 550: result = 0.8 + const else: result = 0.6 + 2 ** (averexp - 7) * const return round(result, 1)