Merge pull request #668 from pyvisa/bin-write

Improve write_binary_values by bypassing the struct module when possible

.github/workflows/ci.yml

@@ -59,6 +59,7 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
+          pip install numpy
       - name: Install project
         run: |
           pip install -e .

CHANGES

@@ -4,6 +4,8 @@ PyVISA Changelog
 1.12.0 (unreleased)
 -------------------
 
+- optimize write_binary_values when passing in bytes, bytearray or numpy arrays PR #668
+  We avoid going through the struct module which can cause large memory overheads.
 - fix collection of debug info on the ctypes wrapper PR #598
 - allow an IEEE binary block or an HP block to be empty PR #581
   This is more correct and can affect real instruments. However this introduces
@@ -19,9 +21,9 @@ PyVISA Changelog
   custom resource opening #660
 - changed constant ControlFlow enum type from IntEnum to IntFlag PR#652
   This change also triggers a change in attributes.py to include a new Attribute
-  class, FlagAttribute where the enum type is declared at IntFlag. 
+  class, FlagAttribute where the enum type is declared at IntFlag.
   Class AttrVI_ATTR_ASRL_FLOW_CNTRL now inherits from FlagAttribute.
-  Flow control attribute per ivi foundation definition is a flag and 
+  Flow control attribute per ivi foundation definition is a flag and
   multiple flow control types can be set.
 
 1.11.3 (07-11-2020)

pyvisa/testsuite/keysight_assisted_tests/__init__.py

@@ -22,22 +22,35 @@
 
 import pytest
 
+_KEYSIGHT_AVAILABLE = "PYVISA_KEYSIGHT_VIRTUAL_INSTR" in os.environ
+
 require_virtual_instr = pytest.mark.skipif(
-    "PYVISA_KEYSIGHT_VIRTUAL_INSTR" not in os.environ,
+    not _KEYSIGHT_AVAILABLE,
     reason="Requires the Keysight virtual instrument. Run on PyVISA buildbot.",
 )
 
 
-RESOURCE_ADDRESSES = {
-    # "USB::INSTR": "USB::",
-    "TCPIP::INSTR": "TCPIP::192.168.0.2::INSTR",
-    "TCPIP::SOCKET": "TCPIP::192.168.0.2::5025::SOCKET",
-    # "GPIB::INSTR": "GPIB::19::INSTR",
-}
+# We are testing locally with only TCPIP resources on the loop back address
+if _KEYSIGHT_AVAILABLE and os.environ["PYVISA_KEYSIGHT_VIRTUAL_INSTR"] == "0":
+    RESOURCE_ADDRESSES = {
+        "TCPIP::INSTR": "TCPIP::127.0.0.1::INSTR",
+        "TCPIP::SOCKET": "TCPIP::127.0.0.1::5025::SOCKET",
+    }
+
+    ALIASES = {}
+
+# We are running on the bot with all supported resources.
+else:
+    RESOURCE_ADDRESSES = {
+        # "USB::INSTR": "USB::",
+        "TCPIP::INSTR": "TCPIP::192.168.0.2::INSTR",
+        "TCPIP::SOCKET": "TCPIP::192.168.0.2::5025::SOCKET",
+        # "GPIB::INSTR": "GPIB::19::INSTR",
+    }
 
-ALIASES = {
-    "TCPIP::192.168.0.2::INSTR": "tcpip",
-}
+    ALIASES = {
+        "TCPIP::192.168.0.2::INSTR": "tcpip",
+    }
 
 
 # Even a deepcopy is not a true copy of a function.

pyvisa/testsuite/keysight_assisted_tests/messagebased_resource_utils.py

@@ -303,7 +303,8 @@ def test_write_ascii_values(self):
             assert self.instr.read() == "\r1;2;3;4;5\r"
 
     @pytest.mark.parametrize(
-        "hfmt, prefix", zip(("ieee", "hp", "empty"), (b"#212", b"#A\x0c\x00", b""))
+        "hfmt, prefix",
+        list(zip(("ieee", "hp", "empty"), (b"#212", b"#A\x0c\x00", b""))),
     )
     def test_write_binary_values(self, hfmt, prefix):
         """Test writing binary data."""
@@ -469,7 +470,7 @@ def test_handling_malformed_binary(self):
         """ """
         pass
 
-    @pytest.mark.parametrize("hfmt, header", zip(("ieee", "empty"), ("#0", "")))
+    @pytest.mark.parametrize("hfmt, header", list(zip(("ieee", "empty"), ("#0", ""))))
     def test_read_binary_values_unreported_length(self, hfmt, header):
         """Test reading binary data."""
         self.instr.read_termination = "\r"

pyvisa/testsuite/test_util.py

@@ -294,10 +294,10 @@ def test_bytes_binary_block(self):
             (util.to_ieee_block, util.to_hp_block),
             (util.from_ieee_block, util.from_hp_block),
         ):
-            for fmt in "sp":
-                block = tb(values, datatype="s")
-                print(block)
-                rt = fb(block, datatype="s", container=bytes)
+            for fmt in "sbB":
+                block = tb(values, datatype=fmt)
+                print(fmt, block)
+                rt = fb(block, datatype=fmt, container=bytes)
                 assert values == rt
 
     def test_malformed_binary_block_header(self):

pyvisa/util.py

@@ -706,7 +706,7 @@ def from_binary_block(
 
 
 def to_binary_block(
-    iterable: Sequence[Union[int, float]],
+    iterable: Union[bytes, bytearray, Sequence[Union[int, float]]],
     header: Union[str, bytes],
     datatype: BINARY_DATATYPES,
     is_big_endian: bool,
@@ -730,12 +730,27 @@ def to_binary_block(
         Binary block of data preceded by the specified header
 
     """
-    array_length = len(iterable)
+    if isinstance(header, str):
+        header = header.encode("ascii")
+
+    if isinstance(iterable, (bytes, bytearray)):
+        if datatype not in "sbB":
+            warnings.warn(
+                "Using the formats 's', 'p', 'b' or 'B' is more efficient when "
+                "directly writing bytes",
+                UserWarning,
+            )
+        else:
+            return header + iterable
+
     endianess = ">" if is_big_endian else "<"
-    fullfmt = "%s%d%s" % (endianess, array_length, datatype)
 
-    if isinstance(header, str):
-        header = bytes(header, "ascii")
+    if _use_numpy_routines(type(iterable)):
+        assert np and isinstance(iterable, np.ndarray)  # For typing
+        return header + iterable.astype(endianess + datatype).tobytes()
+
+    array_length = len(iterable)
+    fullfmt = "%s%d%s" % (endianess, array_length, datatype)
 
     if datatype in ("s", "p"):
         block = struct.pack(fullfmt, iterable)