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

frappy_psi.mercury: proper handling of control_active

Browse Source 
Change-Id: I31e846fa6fdf6d642184e3736a66ffd53033bccf
Reviewed-on: https://forge.frm2.tum.de/review/c/secop/frappy/+/31376
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
Reviewed-by: Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
This commit is contained in:
zolliker
2023-06-19 10:55:40 +02:00
parent 644d005dad
commit 997e8e26e9
1 changed files with 26 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
frappy_psi/mercury.py
View File

@ -28,7 +28,7 @@ import time
from frappy.core import Command, Drivable, HasIO, Writable, StatusType, \
Parameter, Property, Readable, StringIO, Attached, IDLE, RAMPING, nopoll
from frappy.datatypes import EnumType, FloatRange, StringType, StructOf, BoolType, TupleOf
from frappy.errors import HardwareError, ProgrammingError, ConfigError, RangeError
from frappy.errors import HardwareError, ProgrammingError, ConfigError
from frappy_psi.convergence import HasConvergence
from frappy.states import Retry, Finish
from frappy.mixins import HasOutputModule, HasControlledBy
@ -225,14 +225,15 @@ class Loop(HasOutputModule, MercuryChannel, Drivable):
)
enable_pid_table = Parameter('', BoolType(), readonly=False)
def set_output(self, active, source='HW'):
def set_output(self, active, source=None):
if active:
self.activate_control()
else:
self.deactivate_control(source)
def set_target(self, target):
self.set_output(True)
if not self.control_active:
self.activate_control()
self.target = target
def read_enable_pid_table(self):
@ -262,7 +263,7 @@ class Loop(HasOutputModule, MercuryChannel, Drivable):
class ConvLoop(HasConvergence, Loop):
def deactivate_control(self, source):
def deactivate_control(self, source=None):
if self.control_active:
super().deactivate_control(source)
self.convergence_state.start(self.inactive_state)
@ -378,16 +379,14 @@ class TemperatureLoop(TemperatureSensor, ConvLoop):
super().doPoll()
self.read_setpoint()
def read_control_active(self):
active = self.query(f'DEV::{self.ENABLE}', off_on)
self.set_output(active)
return active
def set_control_active(self, active):
super().set_control_active(active)
self.change(f'DEV::{self.ENABLE}', active, off_on)
def write_control_active(self, value):
if value:
raise RangeError('write to target to switch control on')
self.set_output(value, 'user')
return self.change(f'DEV::{self.ENABLE}', value, off_on)
def initialReads(self):
# initialize control active from HW
active = self.query(f'DEV::{self.ENABLE}', off_on)
super().set_output(active, 'HW')
@nopoll # polled by read_setpoint
def read_target(self):
@ -419,7 +418,7 @@ class TemperatureLoop(TemperatureSensor, ConvLoop):
self.change(f'DEV::{self.ENABLE}', True, off_on)
super().set_target(target)
def deactivate_control(self, source):
def deactivate_control(self, source=None):
if self.__ramping:
self.__ramping = False
# stop ramping setpoint
@ -514,14 +513,16 @@ class PressureLoop(PressureSensor, HasControlledBy, ConvLoop):
output_module = Attached(ValvePos, mandatory=False)
tolerance = Parameter(default=0.1)
def read_control_active(self):
active = self.query('DEV::PRES:LOOP:FAUT', off_on)
self.set_output(active)
return active
def set_control_active(self, active):
super().set_control_active(active)
if not active:
self.self_controlled() # switches off auto flow
return self.change('DEV::PRES:LOOP:FAUT', active, off_on)
def write_control_active(self, value):
self.set_output(value, 'user')
return self.change('DEV::PRES:LOOP:FAUT', value, off_on)
def initialReads(self):
# initialize control active from HW
active = self.query('DEV::PRES:LOOP:FAUT', off_on)
super().set_output(active, 'HW')
def read_target(self):
return self.query('DEV::PRES:LOOP:PRST')
@ -566,14 +567,15 @@ class HasAutoFlow:
if value:
self.needle_valve.controlled_by = self.name
else:
if self.needle_valve.control_active:
self.needle_valve.set_target(self.flowpars[1][0]) # flow min
if self.needle_valve.controlled_by != SELF:
self.needle_valve.controlled_by = SELF
self.needle_valve.write_target(self.flowpars[1][0]) # flow min
return value
def auto_flow_off(self):
def auto_flow_off(self, source=None):
if self.auto_flow:
self.log.warning('switch auto flow off')
self.log.warning(f'switched auto flow off by {source or self.name}')
self.write_auto_flow(False)