#!/usr/bin/env python # -*- coding: utf-8 -*- # ***************************************************************************** # 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 # ***************************************************************************** """oxford instruments mercury IPS power supply""" import time from secop.core import Parameter, EnumType, FloatRange, BoolType, IntRange, StringType, Property, BUSY from secop.lib.enum import Enum from secop.errors import BadValueError, HardwareError from secop_psi.magfield import Magfield, SimpleMagfield, Status from secop_psi.mercury import MercuryChannel, off_on, Mapped from secop.lib.statemachine import Retry Action = Enum(hold=0, run_to_set=1, run_to_zero=2, clamped=3) hold_rtoz_rtos_clmp = Mapped(HOLD=Action.hold, RTOS=Action.run_to_set, RTOZ=Action.run_to_zero, CLMP=Action.clamped) CURRENT_CHECK_SIZE = 2 class SimpleField0(MercuryChannel, SimpleMagfield): nunits = Property('number of IPS subunits', IntRange(1, 6), default=1) action = Parameter('action', EnumType(Action), readonly=False) setpoint = Parameter('field setpoint', FloatRange(unit='T'), default=0) voltage = Parameter('leads voltage', FloatRange(unit='V'), default=0) atob = Parameter('field to amp', FloatRange(0, unit='A/T'), default=0) working_ramp = Parameter('effective ramp', FloatRange(0, unit='T/min'), default=0) channel_type = 'PSU' slave_currents = None def read_value(self): return self.query('PSU:SIG:FLD') def read_ramp(self): return self.query('PSU:SIG:RFST') def write_ramp(self, value): return self.change('PSU:SIG:RFST', value) def read_action(self): return self.query('PSU:ACTN', hold_rtoz_rtos_clmp) def write_action(self, value): return self.change('PSU:ACTN', value, hold_rtoz_rtos_clmp) def read_atob(self): return self.query('PSU:ATOB') def read_voltage(self): return self.query('PSU:SIG:VOLT') def read_working_ramp(self): return self.query('PSU:SIG:RFLD') def read_setpoint(self): return self.query('PSU:SIG:FSET') def set_and_go(self, value): self.setpoint = self.change('PSU:SIG:FSET', value) assert self.write_action('hold') == 'hold' assert self.write_action('run_to_set') == 'run_to_set' def start_ramp_to_target(self, state): # if self.action != 'hold': # assert self.write_action('hold') == 'hold' # return Retry() self.set_and_go(state.target) state.try_cnt = 5 return self.ramp_to_target def ramp_to_target(self, state): try: return super().ramp_to_target(state) except HardwareError: state.try_cnt -= 1 if state.try_cnt < 0: raise self.set_and_go(state.target) return Retry() def final_status(self, *args, **kwds): self.write_action('hold') return super().final_status(*args, **kwds) def on_restart(self, state): self.write_action('hold') return super().on_restart(state) class Field0(SimpleField0, Magfield): wait_switch_on = Parameter( 'wait time to ensure switch is on', FloatRange(0, unit='s'), readonly=True, default=60) wait_switch_off = Parameter( 'wait time to ensure switch is off', FloatRange(0, unit='s'), readonly=True, default=60) forced_persistent_field = Parameter( 'manual indication that persistent field is bad', BoolType(), readonly=False, default=False) _field_mismatch = None __init = True __reset_switch_time = False def doPoll(self): super().doPoll() self.read_current() def startModule(self, start_events): self.switch_time = [0, 0] self.switch_heater = self.query('PSU:SIG:SWHT', off_on) super().startModule(start_events) # on restart, assume switch is changed long time ago, if not, the mercury # # will complain and this will be handled in start_ramp_to_field def read_value(self): current = self.query('PSU:SIG:FLD') pf = self.query('PSU:SIG:PFLD') if self.__init: self.__init = False self.persistent_field = pf if self.switch_heater == self.switch_heater.on or self._field_mismatch is None: self.forced_persistent_field = False self._field_mismatch = False return current self._field_mismatch = abs(self.persistent_field - pf) > self.tolerance return pf def read_current(self): if self.slave_currents is None: self.slave_currents = [[] for _ in range(self.nunits + 1)] current = self.query('PSU:SIG:CURR') if self.nunits > 1: for i in range(self.nunits + 1): if i: curri = self.query('DEV:PSU.M%d:PSU:SIG:CURR' % i) volti = self.query('DEV:PSU.M%d:PSU:SIG:VOLT' % i) setattr(self, 'I%d' % i, curri) setattr(self, 'V%d' % i, volti) self.slave_currents[i].append(curri) else: self.slave_currents[i].append(current) min_i = min(self.slave_currents[i]) max_i = max(self.slave_currents[i]) min_ = min(self.slave_currents[0]) / self.nunits max_ = max(self.slave_currents[0]) / self.nunits if len(self.slave_currents[i]) > CURRENT_CHECK_SIZE: self.slave_currents[i] = self.slave_currents[i][-CURRENT_CHECK_SIZE:] if i and (min_i - 1 > max_ or min_ > max_i + 1): self.log.warning('individual currents mismatch %r', self.slave_currents) if self.atob: return current / self.atob return 0 def write_persistent_field(self, value): if self.forced_persistent_field: self._field_mismatch = False return value raise BadValueError('changing persistent field needs forced_persistent_field=True') def write_target(self, target): if self._field_mismatch: self.forced_persistent_field = True raise BadValueError('persistent field does not match - set persistent field to guessed value first') return super().write_target(target) def read_switch_heater(self): value = self.query('PSU:SIG:SWHT', off_on) now = time.time() switch_time = self.switch_time[self.switch_heater] if value != self.switch_heater: self.__reset_switch_time = True if now < (switch_time or 0) + 10: # probably switch heater was changed, but IPS reply is not yet updated return self.switch_heater elif self.__reset_switch_time: self.__reset_switch_time = False self.switch_time[value] = now return value def read_wait_switch_on(self): return self.query('PSU:SWONT') * 0.001 def read_wait_switch_off(self): return self.query('PSU:SWOFT') * 0.001 def write_switch_heater(self, value): if value == self.read_switch_heater(): self.log.info('switch heater already %r', value) # we do not want to restart the timer return value return self.change('PSU:SIG:SWHT', value, off_on) def start_ramp_to_field(self, state): if abs(self.current - self.persistent_field) <= self.tolerance: self.log.info('leads %g are already at %g', self.current, self.persistent_field) return self.ramp_to_field try: self.set_and_go(self.persistent_field) except (HardwareError, AssertionError) as e: if self.switch_heater: self.log.warn('switch is already on!') return self.ramp_to_field self.log.warn('wait first for switch off current=%g pf=%g', self.current, self.persistent_field) return Retry() self.status = Status.PREPARING, 'wait for switch off' state.wait_for = 0 return self.wait_for_switch return self.ramp_to_field def start_ramp_to_target(self, state): state.try_cnt = 5 try: self.set_and_go(state.target) except (HardwareError, AssertionError) as e: self.log.warn('switch not yet ready %r', e) self.status = Status.PREPARING, 'wait for switch on' state.wait_for = 1 return self.wait_for_switch return self.ramp_to_target def wait_for_switch(self, state): if not state.delta(10): # wait at least 10 seconds return Retry() try: # try again self.set_and_go(self.persistent_field) except (HardwareError, AssertionError) as e: return Retry() return self.ramp_to_target if state.wait_for else self.ramp_to_field def start_ramp_to_zero(self, state): assert self.write_action('hold') == 'hold' assert self.write_action('run_to_zero') == 'run_to_zero' return self.ramp_to_zero def Field(name, logger, cfgdict, srv, base=Field0): nunits = cfgdict.get('nunits', 1) if nunits == 1: return base(name, logger, cfgdict, srv) # create individual current and voltage parameters dynamically attrs = {} for i in range(1, nunits + 1): attrs['I%d' % i] = Parameter('slave %s current' % i, FloatRange(unit='A'), default=0) attrs['V%d' % i] = Parameter('slave %s voltage' % i, FloatRange(unit='V'), default=0) return type(base.__name__.replace('0', str(nunits)), (base,), attrs)(name, logger, cfgdict, srv) def SimpleField(name, logger, cfgdict, srv): return Field(name, logger, cfgdict, srv, SimpleField0)