flame sample combined T

frappy_psi/parmod.py

@@ -22,11 +22,12 @@
 
 """modules to access parameters"""
 
-from frappy.core import Drivable, IDLE, Attached, StringType, Property, \
-    Parameter, FloatRange, Readable
+from frappy.core import Drivable, EnumType, IDLE, Attached, StringType, Property, \
+    Parameter, FloatRange, Readable, ERROR
 from frappy.errors import ConfigError
 from frappy_psi.convergence import HasConvergence
 from frappy_psi.mixins import HasRamp
+from frappy.lib import merge_status
 
 
 class Par(Readable):
@@ -121,3 +122,77 @@ class Converging(HasConvergence, Driv):
 
 class RampDriv(HasRamp, Driv):
     pass
+
+
+def set_enabled(modobj, value):
+    """set enabled on module if available"""
+    if hasattr(modobj, 'enabled') and modobj.enabled != value:
+        modobj.write_enabled(value)
+
+
+def get_value(self, obj, default):
+    """get the value of given module. if not valid, return the limit (min_high or max_low)"""
+    if not getattr(obj, 'enabled', True):
+        return default
+    return obj.value if IDLE <= obj.status[0] < ERROR else default
+
+
+class SwitchDriv(HasConvergence, Drivable):
+    low = Attached(description='low range module')
+    high = Attached(description='high range module')
+    min_high = Parameter('minimum high target', FloatRange(unit='$'), readonly=False)
+    max_low = Parameter('maximum low target', FloatRange(unit='$'), readonly=False)
+    # disable_other = Parameter('whether to disable unused channel', BoolType(), readonly=False)
+    selected = Parameter('selected module', EnumType(low=0, high=1), readonly=False, default=0)
+
+    # TODO: copy units from attached module
+    # TODO: callbacks for updates
+
+    def doPoll(self):
+        super().doPoll()
+        if not self.isBusy():
+            low = self.get_value(self.low, self.max_low)
+            high = self.get_value(self.high, self.min_high)
+            low_valid = low < self.max_low
+            high_valid = high > self.min_high
+            if low_valid > high_valid:
+                self.write_selected(self.selected.low)
+            elif high_valid > low_valid:
+                self.write_selected(self.selected.high)
+            if high_valid or low_valid:
+                set_enabled(self.low, low_valid)
+                set_enabled(self.high, high_valid)
+            else:  # both invalid: measure both!
+                set_enabled(self.low, True)
+                set_enabled(self.high, True)
+
+    def read_value(self):
+        return self.low.value if self.selected == self.selected.low else self.high.value
+
+    def read_status(self):
+        status = self.low.status if self.selected == self.selected.low else self.high.status
+        # merge with convergence status
+        return merge_status(status, super().read_status())
+
+    def read_target(self):
+        if self.selected == self.selected.low:
+            return self.low.target
+        return self.high.target
+
+    def write_target(self, target):
+        this, other = self.low, self.high
+        selected = self.selected
+        if target > self.max_low:
+            this, other = other, this
+            selected = self.selected.high
+        elif target < self.min_high:
+            selected = self.selected.low
+        elif self.selected == self.selected.high:
+            this, other = other, this
+        if hasattr(other, 'control_off'):
+            other.control_off()
+        set_enabled(this, True)
+        set_enabled(other, False)
+        self.write_selected(selected)
+        self.convergence_start()
+        return this.write_target(target)