frappy_psi.ValvesHP for capillary rhexometry

frappy_psi/ValvesHP.py

@@ -0,0 +1,104 @@
+from frappy.core import (Drivable, Readable, StringIO, HasIO, Writable,
+                         FloatRange, IntRange, EnumType, Parameter, Property,
+                         PersistentParam, Command, IDLE, BUSY, ERROR,
+                         WARN, Attached, Module)
+#import serial
+#import time
+
+#v1addr = '/dev/serial/by-id/usb-Prolific_Technology_Inc._USB-Serial_Controller_A7BVd126810-if00-port0'
+#v4addr = '/dev/serial/by-id/usb-Prolific_Technology_Inc._USB-Serial_Controller_A_CFd126810-if00-port0'
+#v2addr = '/dev/serial/by-id/usb-Prolific_Technology_Inc._USB-Serial_Controller_BPEAd126810-if00-port0'
+#v3addr = '/dev/serial/by-id/usb-Prolific_Technology_Inc._USB-Serial_Controller_FGCFd126810-if00-port0'
+class ValvesCommunication(StringIO):
+
+   end_of_line = '\r'
+
+
+class SelectorValve(HasIO, Drivable):
+
+    # ioClass = ValvesCommunication
+
+    positions=Property('number of positions', IntRange(1,20), default=10)
+    value=Parameter('current valve position', IntRange(1,20), default=1)
+    target=Parameter('target valve position', IntRange(1,20), default=1, readonly=False)
+    status = Parameter()
+    # key_positions = Property('named posiitons', dict, default={"source1": 1, "source2": 2, "waste": 3})
+
+    io = Attached()
+
+#    def initModule(self):
+#
+#        self.value.datatype = IntRange(1, self.positions)
+#        self.target.datatype = IntRange(1, self.positions)
+
+    def initialReads(self):
+        self.value = self.read_value()
+        self.target = self.value
+        self.value = IDLE
+
+    def query(self, cmd, noreply=False):
+        if noreply:
+            self.communicate(cmd, noreply)
+            return
+        else:
+            return self.communicate(cmd).strip()
+
+    def read_value(self):
+        return int(self.query("CP")[2:])
+
+    def read_status(self):
+        current = self.read_value()
+        if current == self.target:
+            return IDLE,"reached positon"
+        else:
+            return BUSY, "not at target"
+
+    def write_target(self, target):
+        #if not (1 <= target <= self.positions):
+        #    raise ValueError("Invalid position")
+        #    return self.value
+        self.query(f"GO{int(target):02d}", noreply=True)
+        self.status = BUSY, 'not at target'
+
+        self.value = self.read_value()
+        return target
+
+
+    def go_to_key(self, name):
+        if name in self.key_positions:
+            return self.write_target(self.key_positions[name])
+        else:
+            raise ValueError("unknown positon: Try the following names:", ", ".join(self.key_positions.keys()))
+
+class SwitchValve(HasIO, Drivable):
+
+    # ioClass = ValvesCommunication
+
+    value = Parameter('current valve position', IntRange(1, 2))
+    target = Parameter('target valve position', IntRange(1, 2))
+    # key_positions = Property('named posiitons', dict, default={"A": 1, "B": 2})
+
+    #def initModule(self):
+     #   self.value.datatype = IntRange(1, 2)
+      #  self.target.datatype = IntRange(1, 2)
+
+    def query(self, cmd):
+        return self.communicate(cmd).strip()
+
+    def read_value(self):
+        return int(self.query("CP"))
+
+    def read_status(self):
+        current = self.read_value()
+        if current == self.target:
+            return IDLE,"reached target"
+        else:
+            return BUSY, "moving to target"
+
+    def write_target(self, value):
+        current = self.read_value()
+        if current == value:
+            return current
+
+        self.communicate("CP")
+        return self.read_value()