diff --git a/dev-requirements.txt b/dev-requirements.txt index a5adf1678..58edc2bb1 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -3,3 +3,4 @@ pytest==6.1.1 pytest-mock hypothesis==4.28.2 pylint==2.4.4 +pyvisa-sim diff --git a/instruments/abstract_instruments/comm/visa_communicator.py b/instruments/abstract_instruments/comm/visa_communicator.py index d32cbb5b8..154fde4a4 100644 --- a/instruments/abstract_instruments/comm/visa_communicator.py +++ b/instruments/abstract_instruments/comm/visa_communicator.py @@ -29,9 +29,6 @@ class VisaCommunicator(io.IOBase, AbstractCommunicator): def __init__(self, conn): super(VisaCommunicator, self).__init__(self) - if pyvisa is None: - raise ImportError("PyVISA required for accessing VISA instruments.") - version = int(pyvisa.__version__.replace(".", "").ljust(3, "0")) # pylint: disable=no-member if (version < 160 and isinstance(conn, pyvisa.Instrument)) or \ @@ -118,7 +115,7 @@ def read_raw(self, size=-1): elif size == -1: # Read the whole contents, appending the buffer we've already read. - msg = self._buf + self._conn.read() + msg = self._buf + self._conn.read_raw() # Reset the contents of the buffer. self._buf = bytearray() else: @@ -136,10 +133,10 @@ def write_raw(self, msg): self._conn.write_raw(msg) def seek(self, offset): # pylint: disable=unused-argument,no-self-use - return NotImplemented + raise NotImplementedError def tell(self): # pylint: disable=no-self-use - return NotImplemented + raise NotImplementedError def flush_input(self): """ diff --git a/instruments/tests/test_comm/test_visa_communicator.py b/instruments/tests/test_comm/test_visa_communicator.py new file mode 100644 index 000000000..2bccd7f44 --- /dev/null +++ b/instruments/tests/test_comm/test_visa_communicator.py @@ -0,0 +1,179 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +Unit tests for the VISA communication layer +""" + +# IMPORTS #################################################################### + + +import pytest +import pyvisa + +from instruments.units import ureg as u +from instruments.abstract_instruments.comm import VisaCommunicator + + +# TEST CASES ################################################################# + +# pylint: disable=protected-access,redefined-outer-name + +# create a visa instrument +@pytest.fixture() +def visa_inst(): + """Create a default visa-sim instrument and return it.""" + inst = pyvisa.ResourceManager("@sim").open_resource("ASRL1::INSTR") + return inst + + +def test_visacomm_init(visa_inst): + """Initialize visa communicator.""" + comm = VisaCommunicator(visa_inst) + assert comm._conn == visa_inst + assert comm._terminator == "\n" + assert comm._buf == bytearray() + + +def test_visacomm_init_wrong_type(): + """Raise TypeError if not a VISA instrument.""" + with pytest.raises(TypeError) as err: + VisaCommunicator(42) + err_msg = err.value.args[0] + assert err_msg == "VisaCommunicator must wrap a VISA Instrument." + + +def test_visacomm_address(visa_inst): + """Get / Set instrument address.""" + comm = VisaCommunicator(visa_inst) + assert comm.address == visa_inst.resource_name + with pytest.raises(NotImplementedError) as err: + comm.address = "new address" + err_msg = err.value.args[0] + assert err_msg == ( + "Changing addresses of a VISA Instrument is not supported." + ) + + +def test_visacomm_terminator(visa_inst): + """Get / Set terminator.""" + comm = VisaCommunicator(visa_inst) + comm.terminator = ("\r") + assert comm.terminator == "\r" + + +def test_visacomm_terminator_not_string(visa_inst): + """Raise TypeError if terminator is set with non-string character.""" + comm = VisaCommunicator(visa_inst) + with pytest.raises(TypeError) as err: + comm.terminator = 42 + err_msg = err.value.args[0] + assert err_msg == ( + "Terminator for VisaCommunicator must be specified as a single " + "character string." + ) + + +def test_visacomm_terminator_not_single_char(visa_inst): + """Raise ValueError if terminator longer than one character.""" + comm = VisaCommunicator(visa_inst) + with pytest.raises(ValueError) as err: + comm.terminator = "\r\n" + err_msg = err.value.args[0] + assert err_msg == ( + "Terminator for VisaCommunicator must only be 1 character long." + ) + + +def test_visacomm_timeout(visa_inst): + """Set / Get timeout of VISA communicator.""" + comm = VisaCommunicator(visa_inst) + comm.timeout = 3 + assert comm.timeout == u.Quantity(3, u.s) + comm.timeout = u.Quantity(40000, u.ms) + assert comm.timeout == u.Quantity(40, u.s) + + +def test_visacomm_close(visa_inst, mocker): + """Raise an IOError if comms cannot be closed.""" + io_error_mock = mocker.Mock() + io_error_mock.side_effect = IOError + mock_close = mocker.patch.object(visa_inst, 'close', io_error_mock) + comm = VisaCommunicator(visa_inst) + comm.close() + mock_close.assert_called() # but error will just pass! + + +def test_visacomm_read_raw(visa_inst, mocker): + """Read raw data from instrument without size specification.""" + comm = VisaCommunicator(visa_inst) + mock_read_raw = mocker.patch.object( + visa_inst, 'read_raw', return_value=b'asdf' + ) + comm.read_raw() + mock_read_raw.assert_called() + assert comm._buf == bytearray() + + +def test_visacomm_read_raw_size(visa_inst, mocker): + """Read raw data from instrument with size specification.""" + comm = VisaCommunicator(visa_inst) + size = 3 + mock_read_bytes = mocker.patch.object( + visa_inst, 'read_bytes', return_value=b'123' + ) + ret_val = comm.read_raw(size=size) + assert ret_val == b'123' + mock_read_bytes.assert_called() + assert comm._buf == bytearray() + + +def test_visacomm_read_raw_wrong_size(visa_inst): + """Raise ValueError if size is invalid.""" + comm = VisaCommunicator(visa_inst) + with pytest.raises(ValueError) as err: + comm.read_raw(size=-3) + err_msg = err.value.args[0] + assert err_msg == ( + "Must read a positive value of characters, or -1 for all characters." + ) + + +def test_visacomm_write_raw(visa_inst, mocker): + """Write raw message to instrument.""" + mock_write = mocker.patch.object(visa_inst, 'write_raw') + comm = VisaCommunicator(visa_inst) + msg = b'12345' + comm.write_raw(msg) + mock_write.assert_called_with(msg) + + +def test_visacomm_seek_not_implemented(visa_inst): + """Raise NotImplementedError when calling seek.""" + comm = VisaCommunicator(visa_inst) + with pytest.raises(NotImplementedError): + comm.seek(42) + + +def test_visacomm_tell_not_implemented(visa_inst): + """Raise NotImplementedError when calling tell.""" + comm = VisaCommunicator(visa_inst) + with pytest.raises(NotImplementedError): + comm.tell() + + +def test_visacomm_sendcmd(visa_inst, mocker): + """Write to device.""" + mock_write = mocker.patch.object(VisaCommunicator, 'write') + comm = VisaCommunicator(visa_inst) + msg = 'asdf' + comm._sendcmd(msg) + mock_write.assert_called_with(msg + comm.terminator) + + +def test_visacomm_query(visa_inst, mocker): + """Query device.""" + mock_query = mocker.patch.object(visa_inst, 'query') + comm = VisaCommunicator(visa_inst) + msg = 'asdf' + comm._query(msg) + mock_query.assert_called_with(msg + comm.terminator)