various fixes on mb11/dil5

This commit is contained in:
2022-06-07 11:45:25 +02:00
parent 68e2e06905
commit 9c7b6aeb94
5 changed files with 146 additions and 57 deletions

View File

@ -27,7 +27,7 @@ import time
from secop.core import Drivable, HasIO, Writable, \
Parameter, Property, Readable, StringIO, Attached, Done, IDLE, nopoll
from secop.datatypes import EnumType, FloatRange, StringType, StructOf, BoolType
from secop.datatypes import EnumType, FloatRange, StringType, StructOf, BoolType, TupleOf
from secop.errors import HardwareError
from secop_psi.convergence import HasConvergence
from secop.lib.enum import Enum
@ -172,26 +172,29 @@ class TemperatureSensor(MercuryChannel, Readable):
class HasInput(MercuryChannel):
controlled_by = Parameter('source of target value', EnumType(members={'self': SELF}), default=0)
target = Parameter(readonly=False)
input_modules = ()
# do not know why this? target = Parameter(readonly=False)
input_callbacks = ()
def add_input(self, modobj):
if not self.input_modules:
self.input_modules = []
self.input_modules.append(modobj)
def register_input(self, name, control_off):
"""register input
:param name: the name of the module (for controlled_by enum)
:param control_off: a method on the input module to switch off control
"""
if not self.input_callbacks:
self.input_callbacks = []
self.input_callbacks.append(control_off)
prev_enum = self.parameters['controlled_by'].datatype._enum
# add enum member, using autoincrement feature of Enum
self.parameters['controlled_by'].datatype = EnumType(Enum(prev_enum, **{modobj.name: None}))
self.parameters['controlled_by'].datatype = EnumType(Enum(prev_enum, **{name: None}))
def write_controlled_by(self, value):
if self.controlled_by == value:
return Done
self.controlled_by = value
if value == SELF:
self.log.warning('switch to manual mode')
for input_module in self.input_modules:
if input_module.control_active:
input_module.write_control_active(False)
for control_off in self.input_callbacks:
control_off()
return Done
@ -209,12 +212,17 @@ class Loop(HasConvergence, MercuryChannel, Drivable):
def initModule(self):
super().initModule()
if self.output_module:
self.output_module.add_input(self)
self.output_module.register_input(self.name, self.control_off)
def control_off(self):
if self.control_active:
self.log.warning('switch to manual mode')
self.write_control_active(False)
def set_output(self, active):
if active:
if self.output_module and self.output_module.controlled_by != self.name:
self.output_module.controlled_by = self.name
self.output_module.write_controlled_by(self.name)
else:
if self.output_module and self.output_module.controlled_by != SELF:
self.output_module.write_controlled_by(SELF)
@ -340,7 +348,6 @@ class TemperatureLoop(TemperatureSensor, Loop, Drivable):
ramp = Parameter('ramp rate', FloatRange(0, unit='K/min'), readonly=False)
enable_ramp = Parameter('enable ramp rate', BoolType(), readonly=False)
setpoint = Parameter('working setpoint (differs from target when ramping)', FloatRange(0, unit='$'))
auto_flow = Parameter('enable auto flow', BoolType(), readonly=False)
tolerance = Parameter(default=0.1)
_last_setpoint_change = None
@ -394,11 +401,16 @@ class TemperatureLoop(TemperatureSensor, Loop, Drivable):
def write_enable_ramp(self, value):
return self.change('TEMP:LOOP:RENA', value, off_on)
def read_auto_flow(self):
return self.query('TEMP:LOOP:FAUT', off_on)
def write_auto_flow(self, value):
return self.change('TEMP:LOOP:FAUT', value, off_on)
def set_output(self, active):
if active:
if self.output_module and self.output_module.controlled_by != self.name:
self.output_module.write_controlled_by(self.name)
else:
if self.output_module and self.output_module.controlled_by != SELF:
self.output_module.write_controlled_by(SELF)
status = IDLE, 'control inactive'
if self.status != status:
self.status = status
def read_ramp(self):
result = self.query('TEMP:LOOP:RSET')
@ -450,7 +462,7 @@ class ValvePos(HasInput, MercuryChannel, Drivable):
return self.change('PRES:LOOP:FSET', value)
class PressureLoop(PressureSensor, Loop, Drivable):
class PressureLoop(HasInput, PressureSensor, Loop, Drivable):
channel_type = 'PRES,AUX'
output_module = Attached(ValvePos, mandatory=False)
tolerance = Parameter(default=0.1)
@ -467,12 +479,62 @@ class PressureLoop(PressureSensor, Loop, Drivable):
def read_target(self):
return self.query('PRES:LOOP:PRST')
def set_target(self, value):
"""set the target without switching to manual
might be used by a software loop
"""
self.change('PRES:LOOP:PRST', value)
super().set_target(value)
def write_target(self, value):
target = self.change('PRES:LOOP:PRST', value)
self.set_target(target)
self.write_controlled_by(SELF)
self.set_target(value)
return Done
class HasAutoFlow:
needle_valve = Attached(PressureLoop, mandatory=False)
auto_flow = Parameter('enable auto flow', BoolType(), readonly=False, default=0)
flowpars = Parameter('Tdif(min, max), FlowSet(min, max)',
TupleOf(TupleOf(FloatRange(unit='K'), FloatRange(unit='K')),
TupleOf(FloatRange(unit='mbar'), FloatRange(unit='mbar'))),
readonly=False, default=((1,5), (4,20)))
def read_value(self):
value = super().read_value()
if self.auto_flow:
(dmin, dmax), (fmin, fmax) = self.flowpars
flowset = min(dmax - dmin, max(0, value - self.target - dmin)) / (dmax - dmin) * (fmax - fmin) + fmin
self.needle_valve.set_target(flowset)
return Done
def initModule(self):
super().initModule()
if self.needle_valve:
self.needle_valve.register_input(self.name, self.auto_flow_off)
def write_auto_flow(self, value):
if value:
if self.needle_valve and self.needle_valve.controlled_by != self.name:
self.needle_valve.write_controlled_by(self.name)
else:
if self.needle_valve and self.needle_valve.controlled_by != SELF:
self.needle_valve.write_controlled_by(SELF)
_, (fmin, _) = self.flowpars
self.needle_valve.write_target(fmin)
return value
def auto_flow_off(self):
if self.auto_flow:
self.log.warning('switch auto flow off')
self.write_auto_flow(False)
class TemperatureAutoFlow(HasAutoFlow, TemperatureLoop):
pass
class HeLevel(MercuryChannel, Readable):
"""He level meter channel