Add spec-an and minor fixes

Signed-off-by: Travis Collins <[email protected]>

bench/common.py

@@ -2,19 +2,107 @@
 import os
 
 import pyvisa
+import yaml
 
 common_log = logging.getLogger(__name__)
 
 
+def check_connected(func):
+    def check(*args, **kwargs):
+        self = args[0]
+        if not self._connected:
+            raise Exception("Must connect to instrument first. Run connect method")
+        return func(*args, **kwargs)
+
+    return check
+
+
 class Common:
+
+    config_locations = [".", "/etc/"]
+    """Paths to search for config"""
+
+    use_config_file = False
+    """Use config file to get address for instrument(s)"""
+
+    _config_file = None
+    """Used config at runtime"""
+
+    _config = None
+    """Read in config"""
+
+    _config_filename = "bench.yaml"
+    """Filename used for config"""
+
+    _connected = False
+    """State of connected to instrument"""
+
+    _teardown = False
+
+    def __getattribute__(self, name):
+        if name == "_instr":
+            td = object.__getattribute__(
+                self, "_teardown"
+            )  # Don't trigger on destructor
+            if not object.__getattribute__(self, "_connected") and not td:
+                raise Exception("Must connect to instrument first. Run connect method")
+        return object.__getattribute__(self, name)
+
+    @property
+    def config_file(self):
+        return self._config_file
+
+    @config_file.setter
+    def config_file(self, value):
+        if not os.path.isfile(value):
+            raise Exception(f"No config file found at {value}")
+        self._config_file = value
+
+    def __init__(self, address: str = None, use_config_file=False) -> None:
+        """Initialize the N9040B UXA
+
+        Parameters
+        ----------
+        address : str, optional
+            VISA address of the device. If not provided, the device will be found automatically.
+        """
+        if use_config_file:
+            self.use_config_file = use_config_file
+            self._read_from_config()
+        else:
+            self.address = address
+
+        self._post_init_()
+
+    def _post_init_(self):
+        pass
+
     def __del__(self):
         """Close the instrument on object deletion"""
+        self._teardown = True
         if hasattr(self, "_instr"):
             self._instr.close()
         self.address = None
         if hasattr(self, "_rm"):
             self._rm.close()
 
+    def connect(self):
+        if self._connected:
+            print("Already connected")
+            return
+        if not self.address:
+            self._find_device()
+        else:
+            self._instr = pyvisa.ResourceManager().open_resource(self.address)
+            self._connected = True
+            self._instr.timeout = 15000
+            self._instr.write("*CLS")
+            q_id = self._instr.query("*IDN?")
+            if self.id not in q_id:
+                raise Exception(
+                    f"Device at {self.address} is not a {self.id}. Got {q_id}"
+                )
+
     def _find_dev_ind(self, rm):
         all_resources = rm.list_resources()
         for res in all_resources:
@@ -47,3 +135,45 @@ def _find_device(self):
             return
 
         raise Exception(f"No instrument found with ID: {self.id}")
+
+    def _read_from_config(self):
+
+        if not self._config_file:
+            self._find_config()
+
+        with open(self._config_file, "r") as f:
+            self._config = yaml.load(f, Loader=yaml.FullLoader)
+
+        for device in self._config:
+            item = self._config[device]
+            if "type" not in item.keys():
+                print(f"{item} has no field type. Skipping ...")
+                continue
+            if item["type"] == self.id:
+                if "address" not in item.keys():
+                    print(f"{item} has no field address. Skipping ...")
+                    continue
+                self.address = item["address"]
+                return
+        raise Exception(f"No instruments of ID {self.id} found in config")
+
+    def _find_config(self):
+        """Find config on path"""
+        for path in self.config_locations:
+            fn = os.path.join(path, self._config_filename)
+            if os.path.isfile(fn):
+                self._config_file = fn
+                return
+        raise Exception("No config found")
+
+    def query_error(self):
+        """Queries for PXA error
+
+        :raises AttributeError: Error
+        """
+        err = self._instr.query_ascii_values("SYST:ERR?", converter="s")
+        err = int(err[0]), err[1]
+        if err[0] != 0:
+            self._inst.write("*CLS")
+            print(f"ErrorCode:{err[0]}, Message:{err[1]}")
+            raise AttributeError(err)

bench/hittite/hmct2220.py

@@ -1,7 +1,7 @@
 import logging
 
 import pyvisa
-from bench.common import Common
+from bench.common import Common, check_connected
 
 hmct2220_logger = logging.getLogger(__name__)
 

bench/keysight/__init__.py

@@ -1 +1,2 @@
 from .e36233a import E36233A
+from .n9040b import N9040B

bench/keysight/e36233a.py

@@ -1,9 +1,7 @@
 import logging
 
 import pyvisa
-from bench.common import Common
-
-logging.basicConfig(level=logging.DEBUG)
+from bench.common import Common, check_connected
 
 
 class channel:
@@ -14,6 +12,10 @@ def __init__(self, parent, channel_number):
         self.parent = parent
         self.channel_number = channel_number
 
+    @property
+    def _connected(self):
+        return self.parent._connected
+
     def _set_channel(self):
         self.parent._instr.write(f"INST CH{self.channel_number}")
 
@@ -22,6 +24,7 @@ def _to_float(self, value):
         return float(numbers[0]) * pow(10, int(numbers[1]))
 
     @property
+    @check_connected
     def voltage(self):
         """Get or set the voltage of the channel"""
         self._set_channel()
@@ -85,27 +88,7 @@ class E36233A(Common):
     num_channels = 2
     """Number of channels on the device"""
 
-    def __init__(self, address: str = None) -> None:
-        """Initialize the E36233A power supply
-
-        Parameters
-        ----------
-        address : str, optional
-            VISA address of the device. If not provided, the device will be found automatically.
-        """
-        if not address:
-            self._find_device()
-        else:
-            self.address = address
-            self._instr = pyvisa.ResourceManager().open_resource(self.address)
-            self._instr.timeout = 15000
-            self._instr.write("*CLS")
-            q_id = self._instr.query("*IDN?")
-            if self.id not in q_id:
-                raise Exception(
-                    f"Device at {self.address} is not a {self.id}. Got {q_id}"
-                )
-
+    def _post_init_(self, address: str = None, use_config_file=False) -> None:
         self.ch1 = channel(self, 1)
         self.ch2 = channel(self, 2)
 

bench/keysight/n9040b.py

@@ -0,0 +1,81 @@
+import logging
+import time
+import os
+from typing import Union
+
+import pyvisa
+
+from bench.common import Common
+
+
+class N9040B(Common):
+    """Keysight N9040B UXA"""
+
+    id = "N9040B"
+    """Substring returned by IDN query to identify the device"""
+
+    _markers = [*range(1,12+1)]
+
+    def peak_search_marker(self, marker=1):
+        """Set max peak search for a specific marker"""
+        if marker not in self._markers:
+            raise Exception(f"Valid markers are: {self._markers}")
+        self._instr.write(f"CALC:MARK{marker}:MAX")
+
+    def get_marker_amplitude(self, marker=1):
+        """Get current value of marker in dBm"""
+        if marker not in self._markers:
+            raise Exception(f"Valid markers are: {self._markers}")
+        return float(self._instr.query(f"CALC:MARK{marker}:Y?"))
+
+    @property
+    def reset(self):
+        """Reset the instrument"""
+        self._instr.write("*RST")
+
+    @property
+    def span(self) -> float:
+        """Returns the Span in HZ"""
+        return float(self._instr.query("FREQ:SPAN?"))
+
+    @span.setter
+    def span(self, value: float):
+        """Sets the span in Hz"""
+        self._instr.write("FREQ:SPAN "+str(value))
+
+    @property
+    def center_frequency(self) -> float:
+        """Returns the current center frequency in Hz"""
+        return float(self._instr.query("FREQ:CENT?"))
+
+    @center_frequency.setter
+    def center_frequency(self, value: Union[str, float]):
+        """Sets the center frequency in Hz"""
+        self._instr.write("FREQ:CENT "+str(value))
+
+    def screenshot(self, filename: str = "N9040B_sc.png"):
+        """Takes a screenshot on PXA and returns it locally
+
+        Args:
+            filename (str): filename+path to screenshot
+        """
+        path = os.path.dirname(filename)
+        if path:
+            if not os.path.isdir(path):
+                os.mkdir(path)
+
+        if '.png' not in filename:
+            filename = f'{filename}.png'
+
+        just_filename = os.path.basename(filename)
+        self._instr.write(f':MMEM:STOR:SCR "D:\\{just_filename}"')
+        self.query_error()
+        time.sleep(0.5)
+        self._instr.write(f':MMEM:DATA? "D:\\{just_filename}"')
+        data = self._instr.read_raw()
+        # Extract PNG data
+        start = data.index(b'PNG') - 1
+        data = data[start:]
+        with open(filename, 'wb') as f:
+            f.write(data)
+        print(f"Wrote file: {filename}")

pyproject.toml

@@ -32,6 +32,7 @@ dependencies = [
     'pyvisa-py',
     'pyserial',
     'pyusb',
+    'pyyaml',
 ]
 
 [tool.setuptools.dynamic]