linse/frappy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mlz/demo: move old examples to Attached

Browse Source 
change very early version of module attachments in GarfieldMagnet and
MagnetigField to use Attached

Change-Id: I616ad17bc72cd93d86e1b3e3609543cfe90edcd8
Reviewed-on: https://forge.frm2.tum.de/review/c/secop/frappy/+/32250
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
Reviewed-by: Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
This commit is contained in:
Alexander Zaft 2023-10-04 12:43:07 +02:00 committed by Markus Zolliker
parent ec664d9268
commit ae2a731161
4 changed files with 84 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cfg/amagnet_cfg.py
View File

@ -74,10 +74,10 @@ Mod('currentsource',
Mod('mf', Mod('mf',
'frappy_mlz.amagnet.GarfieldMagnet', 'frappy_mlz.amagnet.GarfieldMagnet',
'magnetic field module, handling polarity switching and stuff', 'magnetic field module, handling polarity switching and stuff',
subdev_currentsource = 'currentsource', currentsource = 'currentsource',
subdev_enable = 'enable', enable = 'enable',
subdev_polswitch = 'polarity', polswitch = 'polarity',
subdev_symmetry = 'symmetry', symmetry = 'symmetry',
target = Param(unit='T'), target = Param(unit='T'),
value = Param(unit='T'), value = Param(unit='T'),
userlimits = (-0.35, 0.35), userlimits = (-0.35, 0.35),

4
cfg/demo_cfg.py
View File

@ -41,6 +41,6 @@ Mod('label',
'frappy_demo.modules.Label', 'frappy_demo.modules.Label',
'some label indicating the state of the magnet `mf`.', 'some label indicating the state of the magnet `mf`.',
system = 'Cryomagnet MX15', system = 'Cryomagnet MX15',
subdev_mf = 'mf', mf = 'mf',
subdev_ts = 'ts', ts = 'ts',
) )

41
frappy_demo/modules.py
View File

@ -16,6 +16,7 @@
# #
# Module authors: # Module authors:
# Enrico Faulhaber <enrico.faulhaber@frm2.tum.de> # Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
# Alexander Zaft <a.zaft@fz-juelich.de>
# #
# ***************************************************************************** # *****************************************************************************
"""testing devices""" """testing devices"""
@ -28,9 +29,8 @@ import time
from frappy.datatypes import ArrayOf, BoolType, EnumType, \ from frappy.datatypes import ArrayOf, BoolType, EnumType, \
FloatRange, IntRange, StringType, StructOf, TupleOf FloatRange, IntRange, StringType, StructOf, TupleOf
from frappy.lib.enum import Enum from frappy.lib.enum import Enum
from frappy.modules import Drivable from frappy.modules import Drivable, Readable, Attached
from frappy.modules import Parameter as SECoP_Parameter from frappy.modules import Parameter as SECoP_Parameter
from frappy.modules import Readable
from frappy.properties import Property from frappy.properties import Property
@ -119,9 +119,7 @@ class MagneticField(Drivable):
default=1, datatype=EnumType(persistent=1, hold=0), default=1, datatype=EnumType(persistent=1, hold=0),
readonly=False, readonly=False,
) )
heatswitch = Parameter('name of heat switch device', heatswitch = Attached(Switch, description='name of heat switch device')
datatype=StringType(), export=False,
)
Status = Enum(Drivable.Status, PERSIST=PERSIST, PREPARE=301, RAMPING=302, FINISH=303) Status = Enum(Drivable.Status, PERSIST=PERSIST, PREPARE=301, RAMPING=302, FINISH=303)
@ -130,7 +128,6 @@ class MagneticField(Drivable):
def initModule(self): def initModule(self):
super().initModule() super().initModule()
self._state = Enum('state', idle=1, switch_on=2, switch_off=3, ramp=4).idle self._state = Enum('state', idle=1, switch_on=2, switch_off=3, ramp=4).idle
self._heatswitch = self.DISPATCHER.get_module(self.heatswitch)
_thread = threading.Thread(target=self._thread) _thread = threading.Thread(target=self._thread)
_thread.daemon = True _thread.daemon = True
_thread.start() _thread.start()
@ -165,10 +162,10 @@ class MagneticField(Drivable):
if self.target != self.value: if self.target != self.value:
self.log.debug('got new target -> switching heater on') self.log.debug('got new target -> switching heater on')
self._state = self._state.enum.switch_on self._state = self._state.enum.switch_on
self._heatswitch.write_target('on') self.heatswitch.write_target('on')
if self._state == self._state.enum.switch_on: if self._state == self._state.enum.switch_on:
# wait until switch is on # wait until switch is on
if self._heatswitch.read_value() == 'on': if self.heatswitch.read_value() == 'on':
self.log.debug('heatswitch is on -> ramp to %.3f', self.log.debug('heatswitch is on -> ramp to %.3f',
self.target) self.target)
self._state = self._state.enum.ramp self._state = self._state.enum.ramp
@ -178,7 +175,7 @@ class MagneticField(Drivable):
if self.mode: if self.mode:
self.log.debug('at field -> switching heater off') self.log.debug('at field -> switching heater off')
self._state = self._state.enum.switch_off self._state = self._state.enum.switch_off
self._heatswitch.write_target('off') self.heatswitch.write_target('off')
else: else:
self.log.debug('at field -> hold') self.log.debug('at field -> hold')
self._state = self._state.enum.idle self._state = self._state.enum.idle
@ -189,7 +186,7 @@ class MagneticField(Drivable):
self.value += step self.value += step
if self._state == self._state.enum.switch_off: if self._state == self._state.enum.switch_off:
# wait until switch is off # wait until switch is off
if self._heatswitch.read_value() == 'off': if self.heatswitch.read_value() == 'off':
self.log.debug('heatswitch is off at %.3f', self.value) self.log.debug('heatswitch is off at %.3f', self.value)
self._state = self._state.enum.idle self._state = self._state.enum.idle
self.read_status() # update async self.read_status() # update async
@ -269,12 +266,8 @@ class Label(Readable):
system = Parameter("Name of the magnet system", system = Parameter("Name of the magnet system",
datatype=StringType(), export=False, datatype=StringType(), export=False,
) )
subdev_mf = Parameter("name of subdevice for magnet status", mf = Attached(MagneticField, description="subdevice for magnet status")
datatype=StringType(), export=False, ts = Attached(SampleTemp, description="subdevice for sample temp")
)
subdev_ts = Parameter("name of subdevice for sample temp",
datatype=StringType(), export=False,
)
value = Parameter("final value of label string", default='', value = Parameter("final value of label string", default='',
datatype=StringType(), datatype=StringType(),
) )
@ -282,18 +275,16 @@ class Label(Readable):
def read_value(self): def read_value(self):
strings = [self.system] strings = [self.system]
dev_ts = self.DISPATCHER.get_module(self.subdev_ts) if self.ts:
if dev_ts: strings.append(f"at {self.ts.read_value():.3f} {self.ts.parameters['value'].datatype.unit}")
strings.append(f"at {dev_ts.read_value():.3f} {dev_ts.parameters['value'].datatype.unit}")
else: else:
strings.append('No connection to sample temp!') strings.append('No connection to sample temp!')
dev_mf = self.DISPATCHER.get_module(self.subdev_mf) if self.mf:
if dev_mf: mf_stat = self.mf.read_status()
mf_stat = dev_mf.read_status() mf_mode = self.mf.mode
mf_mode = dev_mf.mode mf_val = self.mf.value
mf_val = dev_mf.value mf_unit = self.mf.parameters['value'].datatype.unit
mf_unit = dev_mf.parameters['value'].datatype.unit
if mf_stat[0] == self.Status.IDLE: if mf_stat[0] == self.Status.IDLE:
state = 'Persistent' if mf_mode else 'Non-persistent' state = 'Persistent' if mf_mode else 'Non-persistent'
else: else:

135
frappy_mlz/amagnet.py
View File

@ -17,6 +17,7 @@
# #
# Module authors: # Module authors:
# Enrico Faulhaber <enrico.faulhaber@frm2.tum.de> # Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
# Alexander Zaft <a.zaft@fz-juelich.de>
# #
# ***************************************************************************** # *****************************************************************************
@ -28,10 +29,10 @@
import math import math
from frappy.datatypes import ArrayOf, FloatRange, StringType, StructOf, TupleOf from frappy.datatypes import ArrayOf, FloatRange, StructOf, TupleOf
from frappy.errors import ConfigError, DisabledError from frappy.errors import ConfigError, DisabledError
from frappy.lib.sequence import SequencerMixin, Step from frappy.lib.sequence import SequencerMixin, Step
from frappy.modules import Drivable, Parameter from frappy.modules import Drivable, Parameter, Attached
class GarfieldMagnet(SequencerMixin, Drivable): class GarfieldMagnet(SequencerMixin, Drivable):
@ -47,19 +48,12 @@ class GarfieldMagnet(SequencerMixin, Drivable):
the symmetry setting selects which. the symmetry setting selects which.
""" """
# attached submodules
currentsource = Attached(description='(bipolar) Powersupply')
enable = Attached(description='Switch to set for on/off')
polswitch = Attached(description='Switch to set for polarity')
symmetry = Attached(description='Switch to read for symmetry')
# parameters # parameters
subdev_currentsource = Parameter('(bipolar) Powersupply',
datatype=StringType(),
readonly=True, export=False)
subdev_enable = Parameter('Switch to set for on/off',
datatype=StringType(), readonly=True,
export=False)
subdev_polswitch = Parameter('Switch to set for polarity',
datatype=StringType(), readonly=True,
export=False)
subdev_symmetry = Parameter('Switch to read for symmetry',
datatype=StringType(), readonly=True,
export=False)
userlimits = Parameter('User defined limits of device value', userlimits = Parameter('User defined limits of device value',
datatype=TupleOf(FloatRange(unit='$'), datatype=TupleOf(FloatRange(unit='$'),
FloatRange(unit='$')), FloatRange(unit='$')),
@ -111,7 +105,7 @@ class GarfieldMagnet(SequencerMixin, Drivable):
Note: This may be overridden in derived classes. Note: This may be overridden in derived classes.
""" """
# binary search/bisection # binary search/bisection
maxcurr = self._currentsource.abslimits[1] maxcurr = self.currentsource.abslimits[1]
mincurr = -maxcurr mincurr = -maxcurr
maxfield = self._current2field(maxcurr) maxfield = self._current2field(maxcurr)
minfield = -maxfield minfield = -maxfield
@ -143,26 +137,21 @@ class GarfieldMagnet(SequencerMixin, Drivable):
def initModule(self): def initModule(self):
super().initModule() super().initModule()
self._enable = self.DISPATCHER.get_module(self.subdev_enable)
self._symmetry = self.DISPATCHER.get_module(self.subdev_symmetry)
self._polswitch = self.DISPATCHER.get_module(self.subdev_polswitch)
self._currentsource = self.DISPATCHER.get_module(
self.subdev_currentsource)
self.init_sequencer(fault_on_error=False, fault_on_stop=False) self.init_sequencer(fault_on_error=False, fault_on_stop=False)
self._symmetry.read_value() self.symmetry.read_value()
def read_calibration(self): def read_calibration(self):
try: try:
try: try:
return self.calibrationtable[self._symmetry.value] return self.calibrationtable[self.symmetry.value]
except KeyError: except KeyError:
return self.calibrationtable[self._symmetry.value.name] return self.calibrationtable[self.symmetry.value.name]
except KeyError: except KeyError:
minslope = min(entry[0] minslope = min(entry[0]
for entry in self.calibrationtable.values()) for entry in self.calibrationtable.values())
self.log.error( self.log.error(
'unconfigured calibration for symmetry %r', 'unconfigured calibration for symmetry %r',
self._symmetry.value) self.symmetry.value)
return [minslope, 0, 0, 0, 0] return [minslope, 0, 0, 0, 0]
def _checkLimits(self, limits): def _checkLimits(self, limits):
@ -182,22 +171,22 @@ class GarfieldMagnet(SequencerMixin, Drivable):
return limits return limits
def read_abslimits(self): def read_abslimits(self):
maxfield = self._current2field(self._currentsource.abslimits[1]) maxfield = self._current2field(self.currentsource.abslimits[1])
# limit to configured value (if any) # limit to configured value (if any)
maxfield = min(maxfield, max(self.accessibles['abslimits'].default)) maxfield = min(maxfield, max(self.accessibles['abslimits'].default))
return -maxfield, maxfield return -maxfield, maxfield
def read_ramp(self): def read_ramp(self):
# This is an approximation! # This is an approximation!
return self.calibration[0] * abs(self._currentsource.ramp) return self.calibration[0] * abs(self.currentsource.ramp)
def write_ramp(self, newramp): def write_ramp(self, newramp):
# This is an approximation! # This is an approximation!
self._currentsource.ramp = float(newramp) / self.calibration[0] self.currentsource.ramp = float(newramp) / self.calibration[0]
def _get_field_polarity(self): def _get_field_polarity(self):
sign = int(self._polswitch.read_value()) sign = int(self.polswitch.read_value())
if self._enable.read_value(): if self.enable.read_value():
return sign return sign
return 0 return 0
@ -210,35 +199,35 @@ class GarfieldMagnet(SequencerMixin, Drivable):
return return
if current_pol == 0: if current_pol == 0:
# safe to switch # safe to switch
self._polswitch.write_target( self.polswitch.write_target(
'+1' if polarity > 0 else str(polarity)) '+1' if polarity > 0 else str(polarity))
return return
if self._currentsource.value < 0.1: if self.currentsource.value < 0.1:
self._polswitch.write_target('0') self.polswitch.write_target('0')
return return
# unsafe to switch, go to safe state first # unsafe to switch, go to safe state first
self._currentsource.write_target(0) self.currentsource.write_target(0)
def read_value(self): def read_value(self):
return self._current2field( return self._current2field(
self._currentsource.read_value() * self.currentsource.read_value() *
self._get_field_polarity()) self._get_field_polarity())
def readHwStatus(self): def readHwStatus(self):
# called from SequencerMixin.read_status if no sequence is running # called from SequencerMixin.read_status if no sequence is running
if self._enable.value == 'Off': if self.enable.value == 'Off':
return self.Status.WARN, 'Disabled' return self.Status.WARN, 'Disabled'
if self._enable.read_status()[0] != self.Status.IDLE: if self.enable.read_status()[0] != self.Status.IDLE:
return self._enable.status return self.enable.status
if self._polswitch.value in ['0', 0]: if self.polswitch.value in ['0', 0]:
return self.Status.IDLE, 'Shorted, ' + self._currentsource.status[1] return self.Status.IDLE, 'Shorted, ' + self.currentsource.status[1]
if self._symmetry.value in ['short', 0]: if self.symmetry.value in ['short', 0]:
return self._currentsource.status[ return self.currentsource.status[
0], 'Shorted, ' + self._currentsource.status[1] 0], 'Shorted, ' + self.currentsource.status[1]
return self._currentsource.read_status() return self.currentsource.read_status()
def write_target(self, target): def write_target(self, target):
if target != 0 and self._symmetry.read_value() in ['short', 0]: if target != 0 and self.symmetry.read_value() in ['short', 0]:
raise DisabledError( raise DisabledError(
'Symmetry is shorted, please select another symmetry first!') 'Symmetry is shorted, please select another symmetry first!')
@ -251,7 +240,7 @@ class GarfieldMagnet(SequencerMixin, Drivable):
seq.append(Step('preparing', 0, self._prepare_ramp)) seq.append(Step('preparing', 0, self._prepare_ramp))
seq.append(Step('recover', 0, self._recover)) seq.append(Step('recover', 0, self._recover))
if current_polarity != wanted_polarity: if current_polarity != wanted_polarity:
if self._currentsource.read_value() > 0.1: if self.currentsource.read_value() > 0.1:
# switching only allowed if current is low enough -> ramp down # switching only allowed if current is low enough -> ramp down
# first # first
seq.append( seq.append(
@ -281,54 +270,54 @@ class GarfieldMagnet(SequencerMixin, Drivable):
# steps for the sequencing # steps for the sequencing
def _prepare_ramp(self, store, *args): def _prepare_ramp(self, store, *args):
store.old_window = self._currentsource.window store.old_window = self.currentsource.window
self._currentsource.window = 1 self.currentsource.window = 1
def _finish_ramp(self, store, *args): def _finish_ramp(self, store, *args):
self._currentsource.window = max(store.old_window, 10) self.currentsource.window = max(store.old_window, 10)
def _recover(self, store): def _recover(self, store):
# check for interlock # check for interlock
if self._currentsource.read_status()[0] != self.Status.ERROR: if self.currentsource.read_status()[0] != self.Status.ERROR:
return return
# recover from interlock # recover from interlock
ramp = self._currentsource.ramp ramp = self.currentsource.ramp
self._polswitch.write_target('0') # short is safe... self.polswitch.write_target('0') # short is safe...
self._polswitch._hw_wait() self.polswitch._hw_wait()
self._enable.write_target('On') # else setting ramp won't work self.enable.write_target('On') # else setting ramp won't work
self._enable._hw_wait() self.enable._hw_wait()
self._currentsource.ramp = 60000 self.currentsource.ramp = 60000
self._currentsource.target = 0 self.currentsource.target = 0
self._currentsource.ramp = ramp self.currentsource.ramp = ramp
# safe state.... if anything of the above fails, the tamperatures may # safe state.... if anything of the above fails, the tamperatures may
# be too hot! # be too hot!
def _ramp_current(self, store, target): def _ramp_current(self, store, target):
if abs(self._currentsource.value - target) <= 0.05: if abs(self.currentsource.value - target) <= 0.05:
# done with this step if no longer BUSY # done with this step if no longer BUSY
return self._currentsource.read_status()[0] == 'BUSY' return self.currentsource.read_status()[0] == 'BUSY'
if self._currentsource.status[0] != 'BUSY': if self.currentsource.status[0] != 'BUSY':
if self._enable.status[0] == 'ERROR': if self.enable.status[0] == 'ERROR':
self._enable.reset() self.enable.reset()
self._enable.read_status() self.enable.read_status()
self._enable.write_target('On') self.enable.write_target('On')
self._enable._hw_wait() self.enable._hw_wait()
self._currentsource.write_target(target) self.currentsource.write_target(target)
return True # repeat return True # repeat
def _ramp_current_cleanup(self, store, step_was_busy, target): def _ramp_current_cleanup(self, store, step_was_busy, target):
# don't cleanup if step finished # don't cleanup if step finished
if step_was_busy: if step_was_busy:
self._currentsource.write_target(self._currentsource.read_value()) self.currentsource.write_target(self.currentsource.read_value())
self._currentsource.window = max(store.old_window, 10) self.currentsource.window = max(store.old_window, 10)
def _set_polarity(self, store, target): def _set_polarity(self, store, target):
if self._polswitch.read_status()[0] == self.Status.BUSY: if self.polswitch.read_status()[0] == self.Status.BUSY:
return True return True
if int(self._polswitch.value) == int(target): if int(self.polswitch.value) == int(target):
return False # done with this step return False # done with this step
if self._polswitch.read_value() != 0: if self.polswitch.read_value() != 0:
self._polswitch.write_target(0) self.polswitch.write_target(0)
else: else:
self._polswitch.write_target(target) self.polswitch.write_target(target)
return True # repeat return True # repeat