Merge pull request #547 from necst-telescope/#534-Anritsu_sg

#534 anritsu sg

neclib/devices/membrane/cpz2724.py

@@ -7,6 +7,22 @@
 
 
 class CPZ2724(Membrane):
+    """Digital I/O board for membrane opener of NANTEN2.
+
+    Notes
+    -----
+    Configuration items for this device:
+
+    rsw_id : {0, 1, ..., 16} or {"0", "1", ..., "9", "A", ..., "F"}
+        Board identifier. This should be set to the same value as the rotary switch
+        "RSW1" mounted on the side of the board. The board is shipped with default RSW1
+        setting of 0. This ID would be non-zero, when multiple PCI board of same model
+        are mounted on a single FA (Factory Automation) controller.
+
+    See defaults setting file in ``neclib/defaults/config.toml``.
+
+    """
+
     Manufacturer = "Interface"
     Model = "CPZ2724"
 

neclib/devices/signal_generator/__init__.py

@@ -1,4 +1,5 @@
-from .E8257D import E8257D  # noqa: F401
+from .e8257d import E8257D  # noqa: F401
 from .fsw0010 import FSW0010  # noqa: F401
 from .fsw0020 import FSW0020  # noqa: F401
+from .mg3692c import MG3692C  # noqa: F401
 from .simulator import SignalGeneratorSimulator  # noqa: F401

neclib/devices/signal_generator/E8257D.py → neclib/devices/signal_generator/e8257d.py

@@ -1,4 +1,5 @@
 import time
+from typing import Optional, Union
 
 import astropy.units as u
 import ogameasure
@@ -40,7 +41,7 @@ class E8257D(SignalGenerator):
     Identifier = "host"
 
     @skip_on_simulator
-    def __init__(self):
+    def __init__(self) -> None:
         self.logger = get_logger(self.__class__.__name__)
 
         if self.Config.communicator == "GPIB":
@@ -54,43 +55,43 @@ def __init__(self):
             )
         self.sg = ogameasure.Agilent.E8257D(com)
 
-    def set_freq(self, freq_GHz):
+    def set_freq(self, freq_GHz: Union[int, float]) -> None:
         with busy(self, "busy"):
             self.sg.freq_set(freq_GHz)
             time.sleep(1)
             return
 
-    def set_power(self, power_dBm):
+    def set_power(self, power_dBm: Union[int, float]) -> None:
         with busy(self, "busy"):
             self.sg.power_set(power_dBm)
             time.sleep(1)
             return
 
-    def get_freq(self):
+    def get_freq(self) -> u.Quantity:
         with busy(self, "busy"):
             f = self.sg.freq_query()
             time.sleep(1)
             return f * u.Hz
 
-    def get_power(self):
+    def get_power(self) -> u.Quantity:
         with busy(self, "busy"):
             f = self.sg.power_query()
             time.sleep(1)
             return f * dBm
 
-    def start_output(self):
+    def start_output(self) -> None:
         with busy(self, "busy"):
             self.sg.output_on()
             time.sleep(1)
             return
 
-    def stop_output(self):
+    def stop_output(self) -> None:
         with busy(self, "busy"):
             self.sg.output_off()
             time.sleep(1)
             return
 
-    def get_output_status(self):
+    def get_output_status(self) -> Optional[bool]:
         with busy(self, "busy"):
             f = self.sg.output_query()
             time.sleep(1)
@@ -101,9 +102,6 @@ def get_output_status(self):
             else:
                 return None
 
-    def finalize(self):
+    def finalize(self) -> None:
         self.stop_output()
-        try:
-            self.sg.com.close()
-        except AttributeError:
-            pass
+        self.sg.com.close()

neclib/devices/signal_generator/mg3692c.py

@@ -0,0 +1,89 @@
+import time
+from typing import Optional, Union
+
+import astropy.units as u
+import ogameasure
+
+from ...core.security import busy
+from ...core.units import dBm
+from ...utils import skip_on_simulator
+from .signal_generator_base import SignalGenerator
+
+
+class MG3692C(SignalGenerator):
+    """Signal Generator, which can supply Local Signal.
+
+    Notes
+    -----
+
+    Configuration items for this device:
+
+    host : str
+        IP address for ethernet communicator.
+
+    port : int
+        ethernet port of using devices.
+
+    """
+
+    Manufacturer: str = "Anritsu"
+    Model = "MG3692C"
+
+    Identifier = "host"
+
+    @skip_on_simulator
+    def __init__(self) -> None:
+        com = ogameasure.gpib_prologix(host=self.Config.host, gpibport=self.Config.port)
+        self.sg = ogameasure.Anritsu.mg3692c(com)
+
+    def set_freq(self, GHz: Union[int, float]) -> None:
+        with busy(self, "busy"):
+            self.sg.freq_set(GHz)
+            time.sleep(0.1)
+        return
+
+    def set_power(self, dBm: Union[int, float]) -> None:
+        with busy(self, "busy"):
+            self.sg.power_set(dBm)
+            time.sleep(0.1)
+        return
+
+    def get_freq(self) -> u.Quantity:
+        with busy(self, "busy"):
+            f = self.sg.freq_query()
+            time.sleep(0.1)
+            return f * u.Hz
+
+    def get_power(self) -> u.Quantity:
+        with busy(self, "busy"):
+            f = self.sg.power_query()
+            time.sleep(0.1)
+            return f * dBm
+
+    def start_output(self) -> None:
+        with busy(self, "busy"):
+            self.sg.output_on()
+            time.sleep(0.1)
+        return
+
+    def stop_output(self) -> None:
+        with busy(self, "busy"):
+            self.sg.output_off()
+            time.sleep(0.1)
+        return
+
+    def get_output_status(self) -> Optional[bool]:
+        with busy(self, "busy"):
+            f = self.sg.output_query()
+            time.sleep(0.1)
+            if f == 1:
+                return True
+            elif f == 0:
+                return False
+            else:
+                return None
+
+    def finalize(self) -> None:
+        self.stop_output()
+        self.sg.close()
+        return

tests/devices/attenuator/test_cpz340516.py

@@ -0,0 +1,17 @@
+import pytest
+
+from neclib.devices.attenuator.cpz340516 import CPZ340516
+
+from ..conftest import get_instance
+
+pytestmark = pytest.mark.skipif(
+    get_instance(CPZ340516) is None, reason="cpz340516 is not configured"
+)
+
+
+class TestRhio10:
+    def test_config_type(self):
+        attenuator = get_instance(CPZ340516)
+        assert type(attenuator.Config.rsw_id) is int
+        assert type(attenuator.Config.range) is str
+        assert type(attenuator.Config.channel) is str

tests/devices/membrane/test_cpz2724.py

@@ -0,0 +1,15 @@
+import pytest
+
+from neclib.devices.membrane.cpz2724 import CPZ2724
+
+from ..conftest import get_instance
+
+pytestmark = pytest.mark.skipif(
+    get_instance(CPZ2724) is None, reason="CPZ2724 is not configured"
+)
+
+
+class TestCPZ3177:
+    def test_config_type(self):
+        membrane = get_instance(CPZ2724)
+        assert type(membrane.Config.rsw_id) is int

tests/devices/signal_generator/test_e8257d.py

@@ -0,0 +1,18 @@
+import pytest
+
+from neclib.devices.signal_generator.e8257d import E8257D
+
+from ..conftest import get_instance
+
+pytestmark = pytest.mark.skipif(
+    get_instance(E8257D) is None, reason="E8257D is not configured"
+)
+
+
+class TestML2437A:
+    def test_config_type(self):
+        sg = get_instance(E8257D)
+        assert type(sg.Config.communicator) is str
+        assert type(sg.Config.host) is str
+        assert type(sg.Config.gpib_port) is int
+        assert type(sg.Config.lan_port) is int

tests/devices/signal_generator/test_mg3692c.py

@@ -0,0 +1,16 @@
+import pytest
+
+from neclib.devices.signal_generator.mg3692c import MG3692C
+
+from ..conftest import get_instance
+
+pytestmark = pytest.mark.skipif(
+    get_instance(MG3692C) is None, reason="MG3692C is not configured"
+)
+
+
+class TestML2437A:
+    def test_config_type(self):
+        sg = get_instance(MG3692C)
+        assert type(sg.Config.host) is str
+        assert type(sg.Config.port) is int