Merge pull request #546 from necst-telescope/#545-memb

#545 add membrane

neclib/defaults/config.toml

@@ -62,6 +62,10 @@ port = "/dev/ttyUSB0"
 _ = "ND287"
 port = "/dev/ttyUSB1"
 
+[membrane]
+_ = "CPZ2724"
+rsw_id = 2
+
 [ccd_controller]
 _ = "M100"
 host = "192.168.100.47"

neclib/devices/membrane/__init__.py

@@ -0,0 +1 @@
+from .cpz2724 import CPZ2724  # noqa: F401

neclib/devices/membrane/cpz2724.py

@@ -0,0 +1,71 @@
+__all__ = ["CPZ2724"]
+
+import time
+
+from ... import get_logger, utils
+from .membrane_base import Membrane
+
+
+class CPZ2724(Membrane):
+    Manufacturer = "Interface"
+    Model = "CPZ2724"
+
+    Identifier = "rsw_id"
+
+    def __init__(self) -> None:
+        self.logger = get_logger(self.__class__.__name__)
+        self.rsw_id = self.Config.rsw_id
+
+        self.io = self._initialize_io()
+
+    @utils.skip_on_simulator
+    def _initialize_io(self):
+        import pyinterface
+
+        self.io = pyinterface.open(2724, self.rsw_id)
+        if self.io is None:
+            raise RuntimeError("Cannot communicate with the CPZ board.")
+
+        self.io.initialize()
+
+        return self.io
+
+    def get_memb_status(self):
+        ret = self.dio.input_point(8, 3)
+        if ret[0] == 0:
+            self.memb_act = "OFF"
+        else:
+            self.memb_act = "DRIVE"
+
+        if ret[1] == 0:
+            if ret[2] == 0:
+                self.memb_pos = "MOVE"
+            else:
+                self.memb_pos = "CLOSE"
+        else:
+            self.memb_pos = "OPEN"
+        return [self.memb_act, self.memb_pos]
+
+    def memb_open(self) -> None:
+        ret = self.get_memb_status()
+        if ret[1] != "OPEN":
+            buff = [1, 1]
+            self.io.output_point(buff, 7)
+            while ret[1] != "OPEN":
+                time.sleep(5)
+                ret = self.get_memb_status()
+        buff = [0, 0]
+        self.io.output_point(buff, 7)
+        return
+
+    def memb_close(self) -> None:
+        ret = self.get_memb_status()
+        if ret[1] != "CLOSE":
+            buff = [0, 1]
+            self.io.output_point(buff, 7)
+            while ret[1] != "CLOSE":
+                time.sleep(5)
+                ret = self.get_memb_status()
+        buff = [0, 0]
+        self.io.output_point(buff, 7)
+        return

neclib/devices/membrane/membrane_base.py

@@ -0,0 +1,26 @@
+from abc import abstractmethod
+
+from ..device_base import DeviceBase
+
+
+class Membrane(DeviceBase):
+
+    Manufacturer: str = ""
+    Model: str
+
+    @abstractmethod
+    def memb_open(self) -> None:
+        """Open membrane."""
+        ...
+
+    @abstractmethod
+    def memb_close(self) -> None:
+        """Close membrane."""
+        ...
+
+    @abstractmethod
+    def get_memb_status(self):
+        # Type hints are not written because the way to write them
+        # differs depending on the version.
+        """Get status of membrane."""
+        ...