IPS mercury

+ mb11, dil5, dil2, variox configs

secop_psi/ips_mercury.py

@@ -0,0 +1,141 @@
+#!/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'), readonly=False, 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):
+        super().write_switch_heater(value)
+        return self.change('PSU:SIG:SWHT', value, off_on)
+
+    def read_atob(self):
+        return self.query('PSU:ATOB')
+
+    def read_setpoint(self):
+        return self.query('PSU:SIG:FSET')
+
+    def write_setpoint(self, value):
+        return self.query('PSU:SIG:FSET', value)
+
+    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 start_ramp_to_field(self, state):
+        self.change('PSU:SIG:FSET', self.persistent_field)
+        assert self.write_action('hold') == 'hold'
+        assert self.write_action('run_to_set') == 'run_to_set'
+        return self.ramp_to_field
+
+    def start_ramp_to_target(self, state):
+        self.change('PSU:SIG:FSET', self.target)
+        assert self.write_action('hold') == 'hold'
+        assert self.write_action('run_to_set') == 'run_to_set'
+        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)