153 lines
5.8 KiB
Python
153 lines
5.8 KiB
Python
#!/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 <markus.zolliker@psi.ch>
|
|
# *****************************************************************************
|
|
"""oxford instruments mercury IPS power supply"""
|
|
|
|
from secop.core import Parameter, EnumType, FloatRange, BoolType
|
|
from secop.lib.enum import Enum
|
|
from secop.errors import BadValueError
|
|
from secop_psi.magfield import Magfield
|
|
from secop_psi.mercury import MercuryChannel, off_on, Mapped
|
|
|
|
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 Field(MercuryChannel, Magfield):
|
|
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)
|
|
forced_persistent_field = Parameter(
|
|
'manual indication that persistent field is bad', BoolType(), readonly=False, default=False)
|
|
|
|
channel_type = 'PSU'
|
|
_field_mismatch = None
|
|
nslaves = 3
|
|
slave_currents = None
|
|
__init = True
|
|
|
|
def read_value(self):
|
|
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 != 0 or self._field_mismatch is None:
|
|
self.forced_persistent_field = False
|
|
self._field_mismatch = False
|
|
return self.current
|
|
self._field_mismatch = abs(self.persistent_field - pf) > self.tolerance
|
|
return pf
|
|
|
|
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_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_switch_heater(self):
|
|
return self.query('PSU:SIG:SWHT', off_on)
|
|
|
|
def write_switch_heater(self, value):
|
|
return self.change('PSU:SIG:SWHT', value, off_on)
|
|
|
|
def read_atob(self):
|
|
return self.query('PSU:ATOB')
|
|
|
|
def read_voltage(self):
|
|
return self.query('PSU:SIG:VOLT')
|
|
|
|
def read_setpoint(self):
|
|
return self.query('PSU:SIG:FSET')
|
|
|
|
def read_current(self):
|
|
if self.slave_currents is None:
|
|
self.slave_currents = [[] for _ in range(self.nslaves + 1)]
|
|
current = self.query('PSU:SIG:CURR')
|
|
for i in range(self.nslaves + 1):
|
|
if i:
|
|
self.slave_currents[i].append(self.query('DEV:PSU.M%d:PSU:SIG:CURR' % i))
|
|
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.nslaves
|
|
max_ = max(self.slave_currents[0]) / self.nslaves
|
|
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 set_and_go(self, value):
|
|
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_field(self, state):
|
|
try:
|
|
self.set_and_go(self.persistent_field)
|
|
except (HardwareError, AssertionError):
|
|
state.switch_undef = self.switch_on_time or state.now
|
|
return self.wait_for_switch
|
|
return self.ramp_to_field
|
|
|
|
def wait_for_switch(self, state):
|
|
if self.now - self.switch_undef < self.wait_switch_on:
|
|
return Retry()
|
|
self.set_and_go(self.persistent_field)
|
|
return self.ramp_to_field
|
|
|
|
def start_ramp_to_target(self, state):
|
|
self.set_and_go(self.target)
|
|
return self.ramp_to_target
|
|
|
|
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 finish_state(self, state):
|
|
self.write_action('hold')
|
|
super().finish_state(state)
|