diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 276ee446..94e0856d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,7 +30,7 @@ jobs: - name: Install tools run: | python -m pip install --upgrade pip - pip install flake8 black isort + pip install flake8 black isort mypy pip install git+https://github.com/pyvisa/pyvisa.git#egg=pyvisa - name: Isort run: | @@ -41,6 +41,9 @@ jobs: - name: Flake8 run: | flake8 pyvisa-py; + - name: Mypy + run: | + mypy pyvisa; tests: name: Unit tests runs-on: ${{ matrix.os }} @@ -86,5 +89,5 @@ jobs: if: needs.tests.result == 'success' run: exit 0 - name: Mark the job as a failure - if: needs.tests.result == 'failure' + if: needs.tests.result != 'success' run: exit 1 diff --git a/pyvisa-py/common.py b/pyvisa-py/common.py index 7242bdd9..3b9dff68 100644 --- a/pyvisa-py/common.py +++ b/pyvisa-py/common.py @@ -1,14 +1,12 @@ # -*- coding: utf-8 -*- -""" - pyvisa-sim.common - ~~~~~~~~~~~~~~~~~ +"""Common code. - Common code. +:copyright: 2014-2020 by PyVISA-sim Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-sim Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ import logging +from typing import Optional, SupportsBytes from pyvisa import logger @@ -16,25 +14,31 @@ class MockInterface(object): - def __init__(self, resource_name): + + #: Name of the resource used to create this interface + resource_name: str + + def __init__(self, resource_name) -> None: self.resource_name = resource_name class NamedObject(object): - """A class to construct named sentinels. - """ + """A class to construct named sentinels.""" - def __init__(self, name): + #: Name used to identify the sentinel + name: str + + def __init__(self, name) -> None: self.name = name - def __repr__(self): + def __repr__(self) -> str: return "<%s>" % self.name __str__ = __repr__ -def iter_bytes(data, mask=None, send_end=False): - +# XXX can probably be removed +def iter_bytes(data: SupportsBytes, mask: Optional[int] = None, send_end: bool = False): if send_end and mask is None: raise ValueError("send_end requires a valid mask.") diff --git a/pyvisa-py/gpib.py b/pyvisa-py/gpib.py index d2874167..3ef729e1 100644 --- a/pyvisa-py/gpib.py +++ b/pyvisa-py/gpib.py @@ -1,21 +1,21 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.gpib - ~~~~~~~~~~~~~~ +"""GPIB Session implementation using linux-gpib or gpib-ctypes. - GPIB Session implementation using linux-gpib or gpib-ctypes. +:copyright: 2015-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2015-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ import ctypes # Used for missing bindings not ideal from bisect import bisect +from typing import Any, Iterator, List, Optional, Tuple from pyvisa import attributes, constants, logger +from pyvisa.constants import StatusCode, ResourceAttribute from .sessions import Session, UnknownAttribute + try: GPIB_CTYPES = True from gpib_ctypes import gpib @@ -52,7 +52,7 @@ # patch Gpib to avoid double closing of handles -def _patch_Gpib(): +def _patch_Gpib() -> None: if not hasattr(Gpib, "close"): _old_del = Gpib.__del__ @@ -67,9 +67,8 @@ def _inner(self): _patch_Gpib() -def _find_boards(): - """Find GPIB board addresses. - """ +def _find_boards() -> Iterator[Tuple[int, int]]: + """Find GPIB board addresses.""" for board in range(16): try: yield board, gpib.ask(board, 1) @@ -77,9 +76,8 @@ def _find_boards(): logger.debug("GPIB board %i error in _find_boards(): %s", board, repr(e)) -def _find_listeners(): - """Find GPIB listeners. - """ +def _find_listeners() -> Iterator[Tuple[int, int]]: + """Find GPIB listeners.""" for board, boardpad in _find_boards(): for i in range(31): try: @@ -94,9 +92,24 @@ def _find_listeners(): ) -def _analyse_lines_value(value, line): +def _analyse_lines_value(value: int, line: int): """Determine the state of a GPIB line based on the iblines byte. + Parameters + ---------- + value : int + Value returned by iblines. + line : int + One of constants.VI_ATTR_GPIB_***_STATE where the *** can be REN, ATN, + NDAC, or SRQ + + Returns + ------- + constants.LineState + State of the line + StatusCode + Library StatusCode for the operation. + """ if line == constants.VI_ATTR_GPIB_REN_STATE: # REN bit valid = 0x10, REN bit value = 0x100 @@ -116,15 +129,13 @@ def _analyse_lines_value(value, line): value_mask = 0x2000 if not value & validity_mask: - return constants.VI_STATE_UNKNOWN, StatusCode.success + return constants.LineState.unknown, StatusCode.success else: if value & value_mask: - return constants.VI_STATE_ASSERTED, StatusCode.success + return constants.LineState.asserted, StatusCode.success else: - return constants.VI_STATE_UNASSERTED, StatusCode.success - + return constants.LineState.unasserted, StatusCode.success -StatusCode = constants.StatusCode # linux-gpib timeout constants, in seconds. See GPIBSession._set_timeout. TIMETABLE = ( @@ -149,22 +160,29 @@ def _analyse_lines_value(value, line): ) -def convert_gpib_error(error, status, operation): +def convert_gpib_error( + error: gpib.GpibError, status: int, operation: str +) -> StatusCode: """Convert a GPIB error to a VISA StatusCode. - :param error: Error to use to determine the proper status code. - :type error: gpib.GpibError - :param status: Status byte of the GPIB library. - :type status: int - :param operation: Name of the operation that caused an exception. Used in logging. - :type operation: str - :return: Status code matching the GPIB error. - :rtype: constants.StatusCode + Parameters + ---------- + error : gpib.GpibError + Error to use to determine the proper status code. + status : int + Status byte of the GPIB library. + operation : str + Name of the operation that caused an exception. Used in logging. + + Returns + ------- + StatusCode + Status code matching the GPIB error. """ # First check the imeout condition in the status byte if status & 0x4000: - return constants.StatusCode.error_timeout + return StatusCode.error_timeout # All other cases are hard errors. # In particular linux-gpib simply gives a string we could parse but that # feels brittle. As a consequence we only try to be smart when using @@ -172,33 +190,33 @@ def convert_gpib_error(error, status, operation): else: logger.debug("Failed to %s.", exc_info=error) if not GPIB_CTYPES: - return constants.StatusCode.error_system_error + return StatusCode.error_system_error if error.code == 1: - return constants.StatusCode.error_not_cic + return StatusCode.error_not_cic elif error.code == 2: - return constants.StatusCode.error_no_listeners + return StatusCode.error_no_listeners elif error.code == 4: - return constants.StatusCode.error_invalid_mode + return StatusCode.error_invalid_mode elif error.code == 11: - return constants.StatusCode.error_nonsupported_operation + return StatusCode.error_nonsupported_operation elif error.code == 1: - return constants.StatusCode.error_not_cic + return StatusCode.error_not_cic elif error.code == 21: - return constants.StatusCode.error_resource_locked + return StatusCode.error_resource_locked else: - return constants.StatusCode.error_system_error + return StatusCode.error_system_error -def convert_gpib_status(status): +def convert_gpib_status(status: int) -> StatusCode: if status & 0x4000: - return constants.StatusCode.error_timeout + return StatusCode.error_timeout elif status & 0x8000: - return constants.StatusCode.error_system_error + return StatusCode.error_system_error else: - return constants.StatusCode.success + return StatusCode.success -class _GPIBCommon(object): +class _GPIBCommon: """Common base class for GPIB sessions. Both INSTR and INTFC resources share the following attributes: @@ -219,8 +237,11 @@ class _GPIBCommon(object): """ + #: Bus wide controller. + controller: Gpib + @classmethod - def get_low_level_info(cls): + def get_low_level_info(cls) -> str: try: ver = gpib.version() except AttributeError: @@ -228,7 +249,7 @@ def get_low_level_info(cls): return "via Linux GPIB (%s)" % ver - def after_parsing(self): + def after_parsing(self) -> None: minor = int(self.parsed.board) sad = 0 timeout = 13 @@ -259,7 +280,7 @@ def after_parsing(self): attribute = getattr(constants, "VI_ATTR_" + name) self.attrs[attribute] = attributes.AttributesByID[attribute].default - def _get_timeout(self, attribute): + def _get_timeout(self, attribute: constants.ResourceAttribute) -> Optional[int]: if self.interface: # 0x3 is the hexadecimal reference to the IbaTMO (timeout) configuration # option in linux-gpib. @@ -271,8 +292,11 @@ def _get_timeout(self, attribute): self.timeout = None return super(_GPIBCommon, self)._get_timeout(attribute) - def _set_timeout(self, attribute, value): - """ + def _set_timeout( + self, attribute: constants.ResourceAttribute, value: Optional[int] + ): + """Set the timeout value. + linux-gpib only supports 18 discrete timeout values. If a timeout value other than these is requested, it will be rounded up to the closest available value. Values greater than the largest available timout value @@ -295,6 +319,7 @@ def _set_timeout(self, attribute, value): 15 100 seconds 16 300 seconds 17 1000 seconds + """ status = super(_GPIBCommon, self)._set_timeout(attribute, value) if self.interface: @@ -307,19 +332,28 @@ def _set_timeout(self, attribute, value): self.interface.timeout(gpib_timeout) return status - def close(self): + def close(self) -> None: if self.interface: self.interface.close() self.controller.close() - def read(self, count): + def read(self, count: int) -> bytes: """Reads data from device or interface synchronously. Corresponds to viRead function of the VISA library. - :param count: Number of bytes to be read. - :return: data read, return value of the library call. - :rtype: bytes, constants.StatusCode + Parameters + ---------- + count : int + Number of bytes to be read. + + Returns + ------- + bytes + Data read from the interface. + StatusCode + Return value of the library call. + """ # INTFC don't have an interface so use the controller ifc = self.interface or self.controller @@ -331,15 +365,23 @@ def read(self, count): return self._read(reader, count, checker, False, None, False, gpib.GpibError) - def write(self, data): + def write(self, data: bytes) -> Tuple[int, StatusCode]: """Writes data to device or interface synchronously. Corresponds to viWrite function of the VISA library. - :param data: data to be written. - :type data: bytes - :return: Number of bytes actually transferred, return value of the library call. - :rtype: int, VISAStatus + Parameters + ---------- + data : bytes + Data to be written. + + Returns + ------- + int + Number of bytes actually transferred + StatusCode + Return value of the library call. + """ logger.debug("GPIB.write %r" % data) @@ -354,17 +396,24 @@ def write(self, data): except gpib.GpibError as e: return 0, convert_gpib_error(e, ifc.ibsta(), "write") - def gpib_control_ren(self, mode): - """Controls the state of the GPIB Remote Enable (REN) interface line, and optionally the remote/local - state of the device. + def gpib_control_ren(self, mode: constants.RENLineOperation) -> StatusCode: + """Controls the state of the GPIB Remote Enable (REN) interface line. + + Optionally the remote/local state of the device is also controlled. Corresponds to viGpibControlREN function of the VISA library. - :param session: Unique logical identifier to a session. - :param mode: Specifies the state of the REN line and optionally the device remote/local state. - (Constants.VI_GPIB_REN*) - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + mode : constants.RENLineOperation + Specifies the state of the REN line and optionally the device + remote/local state. + + Returns + ------- + StatusCode + Return value of the library call. + """ if self.parsed.interface_type == "INTFC": if mode not in ( @@ -409,14 +458,23 @@ def gpib_control_ren(self, mode): return constants.StatusCode.success - def _get_attribute(self, attribute): + def _get_attribute(self, attribute: ResourceAttribute) -> Tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. - :param attribute: Resource attribute for which the state query is made - :return: The state of the queried attribute for a specified resource, return value of the library call. - :rtype: (unicode | str | list | int, VISAStatus) + Parameters + ---------- + attribute : ResourceAttribute + Attribute for which the state query is made + + Returns + ------- + Any + The state of the queried attribute for a specified resource + StatusCode + Return value of the library call. + """ # TODO implement the following attributes # - VI_ATTR_INTF_INST_NAME RO @@ -428,20 +486,20 @@ def _get_attribute(self, attribute): # INTFC don't have an interface so use the controller ifc = self.interface or self.controller - if attribute == constants.VI_ATTR_GPIB_PRIMARY_ADDR: + if attribute == ResourceAttribute.gpib_primary_address: # IbaPAD 0x1 return ifc.ask(1), StatusCode.success - elif attribute == constants.VI_ATTR_GPIB_SECONDARY_ADDR: + elif attribute == ResourceAttribute.gpib_secondary_address: # IbaSAD 0x2 # Remove 0x60 because National Instruments. - sad = ifc.ask(2) # noqa + _ = ifc.ask(2) if ifc.ask(2): return ifc.ask(2) - 96, StatusCode.success else: return constants.VI_NO_SEC_ADDR, StatusCode.success - elif attribute == constants.VI_ATTR_GPIB_REN_STATE: + elif attribute == ResourceAttribute.gpib_ren_state: try: lines = self.controller.lines() return _analyse_lines_value(lines, attribute) @@ -449,7 +507,7 @@ def _get_attribute(self, attribute): # some versions of linux-gpib do not expose Gpib.lines() return constants.VI_STATE_UNKNOWN, StatusCode.success - elif attribute == constants.VI_ATTR_SEND_END_EN: + elif attribute == ResourceAttribute.send_end_enabled: # Do not use IbaEndBitIsNormal 0x1a which relates to EOI on read() # not write(). see issue #196 # IbcEOT 0x4 @@ -458,24 +516,34 @@ def _get_attribute(self, attribute): else: return constants.VI_FALSE, StatusCode.success - elif attribute == constants.VI_ATTR_INTF_NUM: + elif attribute == ResourceAttribute.interface_number: # IbaBNA 0x200 return ifc.ask(512), StatusCode.success - elif attribute == constants.VI_ATTR_INTF_TYPE: + elif attribute == ResourceAttribute.interface_type: return constants.InterfaceType.gpib, StatusCode.success raise UnknownAttribute(attribute) - def _set_attribute(self, attribute, attribute_state): + def _set_attribute( + self, attribute: constants.ResourceAttribute, attribute_state: Any + ) -> StatusCode: """Sets the state of an attribute. Corresponds to viSetAttribute function of the VISA library. - :param attribute: Attribute for which the state is to be modified. (Attributes.*) - :param attribute_state: The state of the attribute to be set for the specified object. - :return: return value of the library call. - :rtype: VISAStatus + Parameters + ---------- + attribute : constants.ResourceAttribute + Attribute for which the state is to be modified. (Attributes.*) + attribute_state : Any + The state of the attribute to be set for the specified object. + + Returns + ------- + StatusCode + Return value of the library call. + """ # TODO implement the following attributes # - VI_ATTR_DMA_ALLOW_EN RW @@ -486,7 +554,7 @@ def _set_attribute(self, attribute, attribute_state): # INTFC don't have an interface so use the controller ifc = self.interface or self.controller - if attribute == constants.VI_ATTR_GPIB_READDR_EN: + if attribute == ResourceAttribute.gpib_readdress_enabled: # IbcREADDR 0x6 # Setting has no effect in linux-gpib. if isinstance(attribute_state, int): @@ -495,7 +563,7 @@ def _set_attribute(self, attribute, attribute_state): else: return StatusCode.error_nonsupported_attribute_state - elif attribute == constants.VI_ATTR_GPIB_PRIMARY_ADDR: + elif attribute == ResourceAttribute.gpib_primary_address: # IbcPAD 0x1 if isinstance(attribute_state, int) and 0 <= attribute_state <= 30: ifc.config(1, attribute_state) @@ -503,7 +571,7 @@ def _set_attribute(self, attribute, attribute_state): else: return StatusCode.error_nonsupported_attribute_state - elif attribute == constants.VI_ATTR_GPIB_SECONDARY_ADDR: + elif attribute == ResourceAttribute.gpib_secondary_address: # IbcSAD 0x2 # Add 0x60 because National Instruments. if isinstance(attribute_state, int) and 0 <= attribute_state <= 30: @@ -515,7 +583,7 @@ def _set_attribute(self, attribute, attribute_state): else: return StatusCode.error_nonsupported_attribute_state - elif attribute == constants.VI_ATTR_GPIB_UNADDR_EN: + elif attribute == ResourceAttribute.gpib_unadress_enable: # IbcUnAddr 0x1b try: ifc.config(27, attribute_state) @@ -523,7 +591,7 @@ def _set_attribute(self, attribute, attribute_state): except gpib.GpibError: return StatusCode.error_nonsupported_attribute_state - elif attribute == constants.VI_ATTR_SEND_END_EN: + elif attribute == ResourceAttribute.send_end_enabled: # Do not use IbaEndBitIsNormal 0x1a which relates to EOI on read() # not write(). see issue #196 # IbcEOT 0x4 @@ -539,23 +607,23 @@ def _set_attribute(self, attribute, attribute_state): # TODO: Check secondary addresses. @Session.register(constants.InterfaceType.gpib, "INSTR") class GPIBSession(_GPIBCommon, Session): - """A GPIB Session that uses linux-gpib to do the low level communication. - """ + """A GPIB Session that uses linux-gpib to do the low level communication.""" @staticmethod - def list_resources(): + def list_resources() -> List[str]: return ["GPIB%d::%d::INSTR" % (board, pad) for board, pad in _find_listeners()] - def clear(self): + def clear(self) -> StatusCode: """Clears a device. Corresponds to viClear function of the VISA library. - :param session: Unique logical identifier to a session. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` - """ + Returns + ------- + StatusCode + Return value of the library call. + """ logger.debug("GPIB.device clear") try: self.interface.clear() @@ -563,14 +631,21 @@ def clear(self): except gpib.GpibError as e: return convert_gpib_error(e, self.interface.ibsta(), "clear") - def assert_trigger(self, protocol): + def assert_trigger(self, protocol: constants.TriggerProtocol) -> StatusCode: """Asserts hardware trigger. - Only supports protocol = constants.VI_TRIG_PROT_DEFAULT - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` - """ + Parameters + ---------- + protocol : constants.TriggerProtocol + Triggering protocol to use. + Only supports constants.TriggerProtocol.default + + Returns + ------- + StatusCode + Return value of the library call. + """ logger.debug("GPIB.device assert hardware trigger") try: @@ -582,13 +657,16 @@ def assert_trigger(self, protocol): except gpib.GpibError as e: return convert_gpib_error(e, self.interface.ibsta(), "assert trigger") - def read_stb(self): + def read_stb(self) -> Tuple[int, StatusCode]: + """Read the device status byte.""" try: return self.interface.serial_poll(), StatusCode.success except gpib.GpibError as e: return 0, convert_gpib_error(e, self.interface.ibsta(), "read STB") - def _get_attribute(self, attribute): + def _get_attribute( + self, attribute: constants.ResourceAttribute + ) -> Tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. GPIB::INSTR have the @@ -600,9 +678,18 @@ def _get_attribute(self, attribute): - VI_ATTR_GPIB_READDR_EN - VI_ATTR_GPIB_UNADDR_EN - :param attribute: Resource attribute for which the state query is made - :return: The state of the queried attribute for a specified resource, return value of the library call. - :rtype: (unicode | str | list | int, VISAStatus) + Parameters + ---------- + attribute : constants.ResourceAttribute + Resource attribute for which the state query is made + + Returns + ------- + Any + State of the queried attribute for a specified resource + StatusCode + Return value of the library call. + """ # TODO implement the following attributes # - VI_ATTR_TRIG_ID RW or RO see specs @@ -624,15 +711,25 @@ def _get_attribute(self, attribute): return super(GPIBSession, self)._get_attribute(attribute) - def _set_attribute(self, attribute, attribute_state): + def _set_attribute( + self, attribute: ResourceAttribute, attribute_state: Any + ) -> StatusCode: """Sets the state of an attribute. Corresponds to viSetAttribute function of the VISA library. - :param attribute: Attribute for which the state is to be modified. (Attributes.*) - :param attribute_state: The state of the attribute to be set for the specified object. - :return: return value of the library call. - :rtype: VISAStatus + Parameters + ---------- + attribute : ResourceAttribute + Attribute for which the state is to be modified. (Attributes.*) + attribute_state : Any + The state of the attribute to be set for the specified object. + + Returns + ------- + StatusCode + Return value of the library call. + """ # TODO implement the following attributes # - VI_ATTR_TRIG_ID RW or RO see specs @@ -666,33 +763,38 @@ class GPIBInterface(_GPIBCommon, Session): """ @staticmethod - def list_resources(): + def list_resources() -> List[str]: return ["GPIB%d::INTFC" % board for board, pad in _find_boards()] - def gpib_command(self, command_bytes): + def gpib_command(self, command_bytes: bytes) -> Tuple[int, StatusCode]: """Write GPIB command byte on the bus. Corresponds to viGpibCommand function of the VISA library. See: https://linux-gpib.sourceforge.io/doc_html/gpib-protocol.html#REFERENCE-COMMAND-BYTES - :param command_bytes: command bytes to send - :type command_bytes: bytes - :return: Number of written bytes, return value of the library call. - :rtype: int, :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + command_bytes : bytes + Command bytes to send + + Returns + ------- + int + Number of written bytes, + StatusCode + Return value of the library call. + """ try: return self.controller.command(command_bytes), StatusCode.success except gpib.GpibError as e: return convert_gpib_error(e, self.controller.ibsta(), "gpib command") - def gpib_send_ifc(self): + def gpib_send_ifc(self) -> StatusCode: """Pulse the interface clear line (IFC) for at least 100 microseconds. Corresponds to viGpibSendIFC function of the VISA library. - :param session: Unique logical identifier to a session. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` """ logger.debug("GPIB.interface clear") try: @@ -701,16 +803,22 @@ def gpib_send_ifc(self): except gpib.GpibError as e: return convert_gpib_error(e, self.controller.ibsta(), "send IFC") - def gpib_control_atn(self, mode): + def gpib_control_atn(self, mode: constants.ATNLineOperation) -> StatusCode: """Specifies the state of the ATN line and the local active controller state. Corresponds to viGpibControlATN function of the VISA library. - :param session: Unique logical identifier to a session. - :param mode: Specifies the state of the ATN line and optionally the local active controller state. - (Constants.VI_GPIB_ATN*) - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + mode : constants.ATNLineOperation + Specifies the state of the ATN line and optionally the local active + controller state. + + Returns + ------- + StatusCode + Return value of the library call. + """ logger.debug("GPIB.control atn") if mode == constants.VI_GPIB_ATN_ASSERT: @@ -726,18 +834,27 @@ def gpib_control_atn(self, mode): return constants.StatusCode.error_invalid_mode return convert_gpib_status(status) - def gpib_pass_control(self, primary_address, secondary_address): - """Tell the GPIB device at the specified address to become controller in charge (CIC). + def gpib_pass_control( + self, primary_address: int, secondary_address: int + ) -> StatusCode: + """Tell a GPIB device to become controller in charge (CIC). Corresponds to viGpibPassControl function of the VISA library. - :param session: Unique logical identifier to a session. - :param primary_address: Primary address of the GPIB device to which you want to pass control. - :param secondary_address: Secondary address of the targeted GPIB device. - If the targeted device does not have a secondary address, - this parameter should contain the value Constants.VI_NO_SEC_ADDR. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + primary_address : int + Primary address of the GPIB device to which you want to pass control. + secondary_address : int + Secondary address of the targeted GPIB device. + If the targeted device does not have a secondary address, + this parameter should contain the value Constants.VI_NO_SEC_ADDR. + + Returns + ------- + StatusCode + Return value of the library call. + """ # ibpct need to get the device id matching the primary and secondary address logger.debug("GPIB.pass control") @@ -752,7 +869,7 @@ def gpib_pass_control(self, primary_address, secondary_address): status = gpib_lib.ibpct(did) return convert_gpib_status(status) - def _get_attribute(self, attribute): + def _get_attribute(self, attribute: ResourceAttribute) -> Tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. GPIB::INTFC have the @@ -767,9 +884,18 @@ def _get_attribute(self, attribute): - VI_ATTR_GPIB_HS488_CBL_LEN - VI_ATTR_GPIB_ADDR_STATE - :param attribute: Resource attribute for which the state query is made - :return: The state of the queried attribute for a specified resource, return value of the library call. - :rtype: (unicode | str | list | int, VISAStatus) + Parameters + ---------- + attribute: ResourceAttribute + Resource attribute for which the state query is made. + + Returns + ------- + Any + The state of the queried attribute for a specified resource, + StatusCode + Return value of the library call. + """ # TODO implement the following attributes # - VI_ATTR_DEV_STATUS_BYTE RW @@ -799,21 +925,31 @@ def _get_attribute(self, attribute): return super(GPIBSession, self)._get_attribute(attribute) - def _set_attribute(self, attribute, attribute_state): + def _set_attribute( + self, attribute: ResourceAttribute, attribute_state: Any + ) -> StatusCode: """Sets the state of an attribute. Corresponds to viSetAttribute function of the VISA library. - :param attribute: Attribute for which the state is to be modified. (Attributes.*) - :param attribute_state: The state of the attribute to be set for the specified object. - :return: return value of the library call. - :rtype: VISAStatus + Parameters + ---------- + attribute : ResourceAttribute + Attribute for which the state is to be modified. + attribute_state : Any + The state of the attribute to be set for the specified object. + + Returns + ------- + StatusCode + Return value of the library call. + """ # TODO implement the following attributes # - VI_ATTR_GPIB_SYS_CNTRL_STATE # - VI_ATTR_DEV_STATUS_BYTE # INTFC don't have an interface so use the controller - ifc = self.controller # noqa + _ = self.controller return super(GPIBSession, self)._set_attribute(attribute, attribute_state) diff --git a/pyvisa-py/highlevel.py b/pyvisa-py/highlevel.py index 3282487a..cf9bbc56 100644 --- a/pyvisa-py/highlevel.py +++ b/pyvisa-py/highlevel.py @@ -1,38 +1,39 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.highlevel - ~~~~~~~~~~~~~~~~~~~ +"""Highlevel wrapper of the VISA Library. - Highlevel wrapper of the VISA Library. +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ import random from collections import OrderedDict +from typing import Any, Dict, Iterable, Iterator, Optional, Tuple, Union from pyvisa import constants, errors, highlevel, rname +from pyvisa.constants import StatusCode +from pyvisa.typing import VISAEventContext, VISARMSession, VISASession from . import sessions from .common import logger -StatusCode = constants.StatusCode - class PyVisaLibrary(highlevel.VisaLibraryBase): """A pure Python backend for PyVISA. The object is basically a dispatcher with some common functions implemented. - When a new resource object is requested to pyvisa, the library creates a Session object - (that knows how to perform low-level communication operations) associated with a session handle - (a number, usually refered just as session). + When a new resource object is requested to pyvisa, the library creates a + Session object (that knows how to perform low-level communication operations) + associated with a session handle (a number, usually refered just as session). - A call to a library function is handled by PyVisaLibrary if it involves a resource agnosting - function or dispatched to the correct session object (obtained from the session id). + A call to a library function is handled by PyVisaLibrary if it involves a + resource agnostic function or dispatched to the correct session object + (obtained from the session id). + + Importantly, the user is unaware of this. PyVisaLibrary behaves for + the user just as NIVisaLibrary. - Importantly, the user is unaware of this. PyVisaLibrary behaves for the user just as NIVisaLibrary. """ # Try to import packages implementing lower level functionality. @@ -65,22 +66,23 @@ class PyVisaLibrary(highlevel.VisaLibraryBase): logger.debug("GPIBSession was not imported %s." % e) @classmethod - def get_session_classes(cls): + def get_session_classes(cls) -> Dict[sessions.Session]: return sessions.Session._session_classes @classmethod - def iter_session_classes_issues(cls): + def iter_session_classes_issues( + cls, + ) -> Iterator[Tuple[constants.InterfaceType, str], str]: return sessions.Session.iter_session_classes_issues() @staticmethod - def get_library_paths(): + def get_library_paths() -> Iterable[str]: """List a dummy library path to allow to create the library.""" return ("py",) @staticmethod - def get_debug_info(): - """Return a list of lines with backend info. - """ + def get_debug_info() -> Dict[str, Union[str, Dict[str, str]]]: + """Return a list of lines with backend info.""" from . import __version__ d = OrderedDict() @@ -95,19 +97,17 @@ def get_debug_info(): return d - def _init(self): - + def _init(self) -> None: + """Custom initialization code.""" #: map session handle to session object. #: dict[int, session.Session] self.sessions = {} - def _register(self, obj): - """Creates a random but unique session handle for a session object, - register it in the sessions dictionary and return the value + def _register(self, obj: object) -> VISASession: + """Creates a random but unique session handle for a session object. + + Register it in the sessions dictionary and return the value. - :param obj: a session object. - :return: session handle - :rtype: int """ session = None @@ -120,24 +120,36 @@ def _register(self, obj): # noinspection PyShadowingBuiltins def open( self, - session, - resource_name, - access_mode=constants.AccessModes.no_lock, - open_timeout=constants.VI_TMO_IMMEDIATE, - ): + session: VISARMSession, + resource_name: str, + access_mode: constants.AccessModes = constants.AccessModes.no_lock, + open_timeout: Optional[int] = constants.VI_TMO_IMMEDIATE, + ) -> Tuple[VISASession, StatusCode]: """Opens a session to the specified resource. Corresponds to viOpen function of the VISA library. - :param session: Resource Manager session (should always be a session returned from open_default_resource_manager()). - :param resource_name: Unique symbolic name of a resource. - :param access_mode: Specifies the mode by which the resource is to be accessed. (constants.AccessModes) - :param open_timeout: Specifies the maximum time period (in milliseconds) that this operation waits - before returning an error. - :return: Unique logical identifier reference to a session, return value of the library call. - :rtype: session, VISAStatus - """ + Parameters + ---------- + session : typing.VISARMSession + Resource Manager session (should always be a session returned from + open_default_resource_manager()). + resource_name : str + Unique symbolic name of a resource. + access_mode : constants.AccessModes, optional + Specifies the mode by which the resource is to be accessed. + open_timeout : Optional[int] + Specifies the maximum time period (in milliseconds) that this + operation waits before returning an error. + + Returns + ------- + typing.VISASession + Unique logical identifier reference to a session + StatusCode + Return value of the library call. + """ try: open_timeout = int(open_timeout) except ValueError: @@ -159,14 +171,21 @@ def open( return self._register(sess), StatusCode.success - def clear(self, session): + def clear(self, session: VISASession) -> StatusCode: """Clears a device. Corresponds to viClear function of the VISA library. - :param session: Unique logical identifier to a session. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : typin.VISASession + Unique logical identifier to a session. + + Returns + ------- + StatusCode + Return value of the library call. + """ try: sess = self.sessions[session] @@ -174,17 +193,30 @@ def clear(self, session): return constants.StatusCode.error_invalid_object return sess.clear() - def flush(self, session, mask): - """Flushes device buffers. + def flush( + self, session: VISASession, mask: constants.BufferOperation + ) -> constants.StatusCode: + """Flush the specified buffers. + + The buffers can be associated with formatted I/O operations and/or + serial communication. + + Corresponds to viFlush function of the VISA library. - Corresponds to viFlush function of the VISA library. See: - https://pyvisa.readthedocs.io/en/latest/api/visalibrarybase.html?highlight=flush#pyvisa.highlevel.VisaLibraryBase.flush - for valid values of mask. + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + mask : constants.BufferOperation + Specifies the action to be taken with flushing the buffer. + The values can be combined using the | operator. However multiple + operations on a single buffer cannot be combined. + + Returns + ------- + constants.StatusCode + Return value of the library call. - :param session: Unique logical identifier to a session. - :param mask: which buffers to clear. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` """ try: sess = self.sessions[session] @@ -192,45 +224,73 @@ def flush(self, session, mask): return constants.StatusCode.error_invalid_object return sess.flush(mask) - def gpib_command(self, session, command_byte): - """Write GPIB command byte on the bus. + def gpib_command( + self, session: VISASession, command_byte: bytes + ) -> Tuple[int, constants.StatusCode]: + """Write GPIB command bytes on the bus. Corresponds to viGpibCommand function of the VISA library. - See: https://linux-gpib.sourceforge.io/doc_html/gpib-protocol.html#REFERENCE-COMMAND-BYTES - :param command_byte: command byte to send - :type command_byte: int, must be [0 255] - :return: return value of the library call - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + command_byte : bytes + Data to write. + + Returns + ------- + int + Number of written bytes + constants.StatusCode + Return value of the library call. + """ try: return self.sessions[session].gpib_command(command_byte) except KeyError: return constants.StatusCode.error_invalid_object - def assert_trigger(self, session, protocol): - """Asserts software or hardware trigger. + def assert_trigger( + self, session: VISASession, protocol: constants.TriggerProtocol + ) -> constants.StatusCode: + """Assert software or hardware trigger. Corresponds to viAssertTrigger function of the VISA library. - :param session: Unique logical identifier to a session. - :param protocol: Trigger protocol to use during assertion. (Constants.PROT*) - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + protocol : constants.TriggerProtocol + Trigger protocol to use during assertion. + + Returns + ------- + constants.StatusCode + Return value of the library call. + """ try: return self.sessions[session].assert_trigger(protocol) except KeyError: return constants.StatusCode.error_invalid_object - def gpib_send_ifc(self, session): + def gpib_send_ifc(self, session: VISASession) -> constants.StatusCode: """Pulse the interface clear line (IFC) for at least 100 microseconds. Corresponds to viGpibSendIFC function of the VISA library. - :param session: Unique logical identifier to a session. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + + Returns + ------- + constants.StatusCode + Return value of the library call. + """ try: sess = self.sessions[session] @@ -238,17 +298,27 @@ def gpib_send_ifc(self, session): return constants.StatusCode.error_invalid_object return sess.gpib_send_ifc() - def gpib_control_ren(self, session, mode): - """Controls the state of the GPIB Remote Enable (REN) interface line, and optionally the remote/local - state of the device. + def gpib_control_ren( + self, session: VISASession, mode: constants.RENLineOperation + ) -> constants.StatusCode: + """Controls the state of the GPIB Remote Enable (REN) interface line. + + Optionally the remote/local state of the device can also be set. Corresponds to viGpibControlREN function of the VISA library. - :param session: Unique logical identifier to a session. - :param mode: Specifies the state of the REN line and optionally the device remote/local state. - (Constants.VI_GPIB_REN*) - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + mode : constants.RENLineOperation + State of the REN line and optionally the device remote/local state. + + Returns + ------- + constants.StatusCode + Return value of the library call. + """ try: sess = self.sessions[session] @@ -256,16 +326,25 @@ def gpib_control_ren(self, session, mode): return constants.StatusCode.error_invalid_object return sess.gpib_control_ren() - def gpib_control_atn(self, session, mode): + def gpib_control_atn( + self, session: VISASession, mode: constants.ATNLineOperation + ) -> constants.StatusCode: """Specifies the state of the ATN line and the local active controller state. Corresponds to viGpibControlATN function of the VISA library. - :param session: Unique logical identifier to a session. - :param mode: Specifies the state of the ATN line and optionally the local active controller state. - (Constants.VI_GPIB_ATN*) - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + mode : constants.ATNLineOperation + State of the ATN line and optionally the local active controller state. + + Returns + ------- + constants.StatusCode + Return value of the library call. + """ try: sess = self.sessions[session] @@ -273,18 +352,29 @@ def gpib_control_atn(self, session, mode): return constants.StatusCode.error_invalid_object return sess.gpib_control_atn() - def gpib_pass_control(self, session, primary_address, secondary_address): - """Tell the GPIB device at the specified address to become controller in charge (CIC). + def gpib_pass_control( + self, session: VISASession, primary_address: int, secondary_address: int + ) -> constants.StatusCode: + """Tell a GPIB device to become controller in charge (CIC). Corresponds to viGpibPassControl function of the VISA library. - :param session: Unique logical identifier to a session. - :param primary_address: Primary address of the GPIB device to which you want to pass control. - :param secondary_address: Secondary address of the targeted GPIB device. - If the targeted device does not have a secondary address, - this parameter should contain the value Constants.VI_NO_SEC_ADDR. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + primary_address : int + Primary address of the GPIB device to which you want to pass control. + secondary_address : int + Secondary address of the targeted GPIB device. + If the targeted device does not have a secondary address, this parameter + should contain the value Constants.VI_NO_SEC_ADDR. + + Returns + ------- + constants.StatusCode + Return value of the library call. + """ try: sess = self.sessions[session] @@ -292,12 +382,23 @@ def gpib_pass_control(self, session, primary_address, secondary_address): return constants.StatusCode.error_invalid_object return sess.gpib_pass_control() - def read_stb(self, session): + def read_stb(self, session: VISASession) -> Tuple[int, constants.StatusCode]: """Reads a status byte of the service request. + Corresponds to viReadSTB function of the VISA library. - :param session: Unique logical identifier to a session. - :return: Service request status byte, return value of the library call. - :rtype: int, :class:`pyvisa.constants.StatusCode` + + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + + Returns + ------- + int + Service request status byte + constants.StatusCode + Return value of the library call. + """ try: sess = self.sessions[session] @@ -305,14 +406,23 @@ def read_stb(self, session): return 0, constants.StatusCode.error_invalid_object return sess.read_stb() - def close(self, session): + def close( + self, session: Union[VISASession, VISAEventContext, VISARMSession] + ) -> constants.StatusCode: """Closes the specified session, event, or find list. Corresponds to viClose function of the VISA library. - :param session: Unique logical identifier to a session, event, or find list. - :return: return value of the library call. - :rtype: VISAStatus + Parameters + --------- + session : Union[VISASession, VISAEventContext, VISARMSession] + Unique logical identifier to a session, event, resource manager. + + Returns + ------- + constants.StatusCode + Return value of the library call. + """ try: sess = self.sessions[session] @@ -321,22 +431,41 @@ def close(self, session): except KeyError: return StatusCode.error_invalid_object - def open_default_resource_manager(self): + def open_default_resource_manager( + self, + ) -> Tuple[VISARMSession, constants.StatusCode]: """This function returns a session to the Default Resource Manager resource. Corresponds to viOpenDefaultRM function of the VISA library. - :return: Unique logical identifier to a Default Resource Manager session, return value of the library call. - :rtype: session, VISAStatus + Returns + ------- + VISARMSession + Unique logical identifier to a Default Resource Manager session + constants.StatusCode + Return value of the library call. + """ return self._register(self), StatusCode.success - def list_resources(self, session, query="?*::INSTR"): - """Returns a tuple of all connected devices matching query. + def list_resources( + self, session: VISARMSession, query: str = "?*::INSTR" + ) -> Tuple[str, ...]: + """Return a tuple of all connected devices matching query. - :param query: regular expression used to match devices. - """ + Parameters + ---------- + session : VISARMSession + Unique logical identifier to the resource manager session. + query : str + Regular expression used to match devices. + Returns + ------- + Tuple[str, ...] + Resource names of all the connected devices matching the query. + + """ # For each session type, ask for the list of connected resources and # merge them into a single list. @@ -352,15 +481,27 @@ def list_resources(self, session, query="?*::INSTR"): return resources - def read(self, session, count): + def read( + self, session: VISASession, count: int + ) -> Tuple[bytes, constants.StatusCode]: """Reads data from device or interface synchronously. Corresponds to viRead function of the VISA library. - :param session: Unique logical identifier to a session. - :param count: Number of bytes to be read. - :return: data read, return value of the library call. - :rtype: bytes, VISAStatus + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + count : int + Number of bytes to be read. + + Returns + ------- + bytes + Date read + constants.StatusCode + Return value of the library call. + """ # from the session handle, dispatch to the read method of the session object. try: @@ -373,16 +514,27 @@ def read(self, session, count): return ret - def write(self, session, data): - """Writes data to device or interface synchronously. + def write( + self, session: VISASession, data: bytes + ) -> Tuple[int, constants.StatusCode]: + """Write data to device or interface synchronously. Corresponds to viWrite function of the VISA library. - :param session: Unique logical identifier to a session. - :param data: data to be written. - :type data: str - :return: Number of bytes actually transferred, return value of the library call. - :rtype: int, VISAStatus + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + data : bytes + Data to be written. + + Returns + ------- + int + Number of bytes actually transferred + constants.StatusCode + Return value of the library call. + """ # from the session handle, dispatch to the write method of the session object. try: @@ -395,38 +547,79 @@ def write(self, session, data): return ret - def buffer_read(self, session, count): - """Reads data from device or interface through the use of a formatted I/O read buffer. + def buffer_read( + self, session: VISASession, count: int + ) -> Tuple[bytes, constants.StatusCode]: + """Reads data through the use of a formatted I/O read buffer. + + The data can be read from a device or an interface. + Corresponds to viBufRead function of the VISA library. - :param session: Unique logical identifier to a session. - :param count: Number of bytes to be read. - :return: data read, return value of the library call. - :rtype: bytes, VISAStatus + Parameters + ---------- + session : VISASession\ + Unique logical identifier to a session. + count : int + Number of bytes to be read. + + Returns + ------- + bytes + Data read + constants.StatusCode + Return value of the library call. + """ return self.read(session, count) - def buffer_write(self, session, data): + def buffer_write( + self, session: VISASession, data: bytes + ) -> Tuple[int, constants.StatusCode]: """Writes data to a formatted I/O write buffer synchronously. + Corresponds to viBufWrite function of the VISA library. - :param session: Unique logical identifier to a session. - :param data: data to be written. - :type data: str - :return: Number of bytes actually transferred, return value of the library call. - :rtype: int, VISAStatus + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + data : bytes + Data to be written. + + Returns + ------- + int + number of written bytes + constants.StatusCode + return value of the library call. + """ return self.write(session, data) - def get_attribute(self, session, attribute): + def get_attribute( + self, + session: Union[VISASession, VISAEventContext, VISARMSession], + attribute: Union[constants.ResourceAttribute, constants.EventAttribute], + ) -> Tuple[Any, constants.StatusCode]: """Retrieves the state of an attribute. Corresponds to viGetAttribute function of the VISA library. - :param session: Unique logical identifier to a session, event, or find list. - :param attribute: Resource attribute for which the state query is made (see Attributes.*) - :return: The state of the queried attribute for a specified resource, return value of the library call. - :rtype: unicode | str | list | int, VISAStatus + Parameters + ---------- + session : Union[VISASession, VISAEventContext] + Unique logical identifier to a session, event, or find list. + attribute : Union[constants.ResourceAttribute, constants.EventAttribute] + Resource or event attribute for which the state query is made. + + Returns + ------- + Any + State of the queried attribute for a specified resource + constants.StatusCode + Return value of the library call. + """ try: sess = self.sessions[session] @@ -435,18 +628,31 @@ def get_attribute(self, session, attribute): return sess.get_attribute(attribute) - def set_attribute(self, session, attribute, attribute_state): - """Sets the state of an attribute. + def set_attribute( + self, + session: VISASession, + attribute: constants.ResourceAttribute, + attribute_state: Any, + ) -> constants.StatusCode: + """Set the state of an attribute. Corresponds to viSetAttribute function of the VISA library. - :param session: Unique logical identifier to a session. - :param attribute: Attribute for which the state is to be modified. (Attributes.*) - :param attribute_state: The state of the attribute to be set for the specified object. - :return: return value of the library call. - :rtype: VISAStatus - """ + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + attribute : constants.ResourceAttribute + Attribute for which the state is to be modified. + attribute_state : Any + The state of the attribute to be set for the specified object. + Returns + ------- + StatusCode + Return value of the library call. + + """ try: sess = self.sessions[session] except KeyError: @@ -454,18 +660,38 @@ def set_attribute(self, session, attribute, attribute_state): return sess.set_attribute(attribute, attribute_state) - def lock(self, session, lock_type, timeout, requested_key=None): + def lock( + self, + session: VISASession, + lock_type: constants.Lock, + timeout: int, + requested_key: Optional[str] = None, + ) -> Tuple[str, constants.StatusCode]: """Establishes an access mode to the specified resources. Corresponds to viLock function of the VISA library. - :param session: Unique logical identifier to a session. - :param lock_type: Specifies the type of lock requested, either Constants.EXCLUSIVE_LOCK or Constants.SHARED_LOCK. - :param timeout: Absolute time period (in milliseconds) that a resource waits to get unlocked by the - locking session before returning an error. - :param requested_key: This parameter is not used and should be set to VI_NULL when lockType is VI_EXCLUSIVE_LOCK. - :return: access_key that can then be passed to other sessions to share the lock, return value of the library call. - :rtype: str, :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + lock_type : constants.Lock + Specifies the type of lock requested. + timeout : int + Absolute time period (in milliseconds) that a resource waits to get + unlocked by the locking session before returning an error. + requested_key : Optional[str], optional + Requested locking key in the case of a shared lock. For an exclusive + lock it should be None. + + Returns + ------- + Optional[str] + Key that can then be passed to other sessions to share the lock, or + None for an exclusive lock. + StatusCode + Return value of the library call. + """ try: sess = self.sessions[session] @@ -474,14 +700,21 @@ def lock(self, session, lock_type, timeout, requested_key=None): return sess.lock(lock_type, timeout, requested_key) - def unlock(self, session): - """Relinquishes a lock for the specified resource. + def unlock(self, session: VISASession) -> constants.StatusCode: + """Relinquish a lock for the specified resource. Corresponds to viUnlock function of the VISA library. - :param session: Unique logical identifier to a session. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + + Returns + ------- + StatusCode + Return value of the library call. + """ try: sess = self.sessions[session] @@ -490,10 +723,56 @@ def unlock(self, session): return sess.unlock() - def disable_event(self, session, event_type, mechanism): - # TODO: implement this for GPIB finalization + def disable_event( + self, + session: VISASession, + event_type: constants.EventType, + mechanism: constants.EventMechanism, + ) -> StatusCode: + """Disable notification for an event type(s) via the specified mechanism(s). + + Corresponds to viDisableEvent function of the VISA library. + + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + event_type : constants.EventType + Event type. + mechanism : constants.EventMechanism + Event handling mechanisms to be disabled. + + Returns + ------- + StatusCode + Return value of the library call. + + """ pass - def discard_events(self, session, event_type, mechanism): - # TODO: implement this for GPIB finalization + def discard_events( + self, + session: VISASession, + event_type: constants.EventType, + mechanism: constants.EventMechanism, + ) -> StatusCode: + """Discard event occurrences for a given type and mechanisms in a session. + + Corresponds to viDiscardEvents function of the VISA library. + + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + event_type : constans.EventType + Logical event identifier. + mechanism : constants.EventMechanism + Specifies event handling mechanisms to be discarded. + + Returns + ------- + StatusCode + Return value of the library call. + + """ pass diff --git a/pyvisa-py/protocols/__init__.py b/pyvisa-py/protocols/__init__.py index 1f3bf64c..2e843c03 100644 --- a/pyvisa-py/protocols/__init__.py +++ b/pyvisa-py/protocols/__init__.py @@ -1,11 +1,8 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.protocols - ~~~~~~~~~~~~~~~~~~~ +"""Implements protocols on top of lower level libraries to talk to instruments. - Implements protocols on top of lower level libraries to talk to instruments. +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ diff --git a/pyvisa-py/protocols/rpc.py b/pyvisa-py/protocols/rpc.py index 9845e2b5..07349357 100644 --- a/pyvisa-py/protocols/rpc.py +++ b/pyvisa-py/protocols/rpc.py @@ -1,26 +1,22 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.protocols.rpc - ~~~~~~~~~~~~~~~~~~~~~~~ +"""Sun RPC version 2 -- RFC1057 - Sun RPC version 2 -- RFC1057 +This file is drawn from Python's RPC demo, updated for python 3. - This file is drawn from Python's RPC demo, updated for python 3. +XXX There should be separate exceptions for the various reasons why +XXX an RPC can fail, rather than using RuntimeError for everything - XXX There should be separate exceptions for the various reasons why - XXX an RPC can fail, rather than using RuntimeError for everything +XXX The UDP version of the protocol resends requests when it does +XXX not receive a timely reply -- use only for idempotent calls! - XXX The UDP version of the protocol resends requests when it does - XXX not receive a timely reply -- use only for idempotent calls! +Original source: + http://svn.python.org/projects/python/trunk/Demo/rpc/rpc.py - Original source: - http://svn.python.org/projects/python/trunk/Demo/rpc/rpc.py +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ - import enum import select import socket diff --git a/pyvisa-py/protocols/usbraw.py b/pyvisa-py/protocols/usbraw.py index 970da9b3..04c7c96a 100644 --- a/pyvisa-py/protocols/usbraw.py +++ b/pyvisa-py/protocols/usbraw.py @@ -1,17 +1,14 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.protocols.usbraw - ~~~~~~~~~~~~~~~~~~~~~~~~~~ +"""Implements Session to control USB Raw devices - Implements Session to control USB Raw devices +Loosely based on PyUSBTMC:python module to handle +USB-TMC(Test and Measurement class) devices. by Noboru Yamamot, Accl. Lab, KEK, JAPAN - Loosely based on PyUSBTMC:python module to handle USB-TMC(Test and Measurement class) devices. - by Noboru Yamamot, Accl. Lab, KEK, JAPAN +This file is an offspring of the Lantz Project. - This file is an offspring of the Lantz Project. +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ from .usbtmc import USBRaw as USBRaw from .usbutil import find_devices, find_interfaces diff --git a/pyvisa-py/protocols/usbtmc.py b/pyvisa-py/protocols/usbtmc.py index e345bd4c..4863d9a3 100644 --- a/pyvisa-py/protocols/usbtmc.py +++ b/pyvisa-py/protocols/usbtmc.py @@ -1,19 +1,14 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.protocols.usbtmc - ~~~~~~~~~~~~~~~~~~~~~~~~~~ - - Implements Session to control USBTMC instruments +"""Implements Session to control USBTMC instruments - Loosely based on PyUSBTMC:python module to handle USB-TMC(Test and - Measurement class) devices. +Loosely based on PyUSBTMC:python module to handle USB-TMC(Test and +Measurement class) devices. by Noboru Yamamot, Accl. Lab, KEK, JAPAN - by Noboru Yamamot, Accl. Lab, KEK, JAPAN +This file is an offspring of the Lantz Project. - This file is an offspring of the Lantz Project. +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ import enum import struct diff --git a/pyvisa-py/protocols/usbutil.py b/pyvisa-py/protocols/usbutil.py index ad78463f..2b204906 100644 --- a/pyvisa-py/protocols/usbutil.py +++ b/pyvisa-py/protocols/usbutil.py @@ -1,18 +1,15 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.usb - ~~~~~~~~~~~~~ +"""Serial Session implementation using PyUSB. - Serial Session implementation using PyUSB. +See the following link for more information about USB. - See the following link for more information about USB. +http://www.beyondlogic.org/usbnutshell/usb5.shtml - http://www.beyondlogic.org/usbnutshell/usb5.shtml +This file is an offspring of the Lantz Project. - This file is an offspring of the Lantz Project. +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ from fnmatch import fnmatch diff --git a/pyvisa-py/protocols/vxi11.py b/pyvisa-py/protocols/vxi11.py index 531e93c0..0df4d5b7 100644 --- a/pyvisa-py/protocols/vxi11.py +++ b/pyvisa-py/protocols/vxi11.py @@ -1,16 +1,13 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.protocols.vxi11 - ~~~~~~~~~~~~~~~~~~~~~~~~~ +"""Implements a VX11 Session using Python Standard Library. - Implements a VX11 Session using Python Standard Library. +Based on Python Sun RPC Demo and Alex Forencich python-vx11 - Based on Python Sun RPC Demo and Alex Forencich python-vx11 +This file is an offspring of the Lantz project. - This file is an offspring of the Lantz project. +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ import enum import socket diff --git a/pyvisa-py/serial.py b/pyvisa-py/serial.py index 9dac91b5..1408d49d 100644 --- a/pyvisa-py/serial.py +++ b/pyvisa-py/serial.py @@ -1,15 +1,20 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.serial - ~~~~~~~~~~~~~~~~ +"""Serial Session implementation using PySerial. - Serial Session implementation using PySerial. +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ +from typing import Any, List, Tuple + from pyvisa import attributes, constants, logger +from pyvisa.constants import ( + BufferOperation, + ResourceAttribute, + SerialTermination, + StatusCode, +) from . import common from .sessions import Session, UnknownAttribute @@ -26,29 +31,23 @@ raise -def to_state(boolean_input): - """Convert a boolean input into a LineState value - """ +def to_state(boolean_input: bool) -> constants.LineState: + """Convert a boolean input into a LineState value.""" if boolean_input: return constants.LineState.asserted return constants.LineState.unasserted -StatusCode = constants.StatusCode -SerialTermination = constants.SerialTermination - - @Session.register(constants.InterfaceType.asrl, "INSTR") class SerialSession(Session): - """A serial Session that uses PySerial to do the low level communication. - """ + """A serial Session that uses PySerial to do the low level communication.""" @staticmethod - def list_resources(): + def list_resources() -> List[str]: return ["ASRL%s::INSTR" % port[0] for port in comports()] @classmethod - def get_low_level_info(cls): + def get_low_level_info(cls) -> str: try: ver = serial.VERSION except AttributeError: @@ -56,11 +55,8 @@ def get_low_level_info(cls): return "via PySerial (%s)" % ver - def after_parsing(self): - if "mock" in self.parsed: - cls = self.parsed.mock - else: - cls = serial.Serial + def after_parsing(self) -> None: + cls = serial.Serial self.interface = cls( port=self.parsed.board, timeout=self.timeout, write_timeout=self.timeout @@ -77,33 +73,41 @@ def after_parsing(self): attribute = getattr(constants, "VI_ATTR_" + name) self.attrs[attribute] = attributes.AttributesByID[attribute].default - def _get_timeout(self, attribute): + def _get_timeout(self, attribute: ResourceAttribute) -> Tuple[int, StatusCode]: if self.interface: self.timeout = self.interface.timeout return super(SerialSession, self)._get_timeout(attribute) - def _set_timeout(self, attribute, value): + def _set_timeout(self, attribute: ResourceAttribute, value: int) -> StatusCode: status = super(SerialSession, self)._set_timeout(attribute, value) if self.interface: self.interface.timeout = self.timeout self.interface.write_timeout = self.timeout return status - def close(self): + def close(self) -> None: self.interface.close() - def read(self, count): + def read(self, count: int) -> Tuple[bytes, StatusCode]: """Reads data from device or interface synchronously. Corresponds to viRead function of the VISA library. - :param count: Number of bytes to be read. - :return: data read, return value of the library call. - :rtype: bytes, constants.StatusCode - """ + Parameters + ----------- + count : int + Number of bytes to be read. - end_in, _ = self.get_attribute(constants.VI_ATTR_ASRL_END_IN) - suppress_end_en, _ = self.get_attribute(constants.VI_ATTR_SUPPRESS_END_EN) + Returns + ------- + bytes + Data read from the device + StatusCode + Return value of the library call. + + """ + end_in, _ = self.get_attribute(ResourceAttribute.asrl_end_in) + suppress_end_en, _ = self.get_attribute(ResourceAttribute.suppress_end_enabled) reader = lambda: self.interface.read(1) @@ -115,7 +119,7 @@ def read(self, count): checker = lambda current: bool(common.last_int(current) & mask) elif end_in == SerialTermination.termination_char: - end_char, _ = self.get_attribute(constants.VI_ATTR_TERMCHAR) + end_char, _ = self.get_attribute(ResourceAttribute.termchar) checker = lambda current: common.last_int(current) == end_char @@ -132,20 +136,28 @@ def read(self, count): serial.SerialTimeoutException, ) - def write(self, data): + def write(self, data: bytes) -> Tuple[int, StatusCode]: """Writes data to device or interface synchronously. Corresponds to viWrite function of the VISA library. - :param data: data to be written. - :type data: bytes - :return: Number of bytes actually transferred, return value of the library call. - :rtype: int, VISAStatus + Parameters + ---------- + data : bytes + Data to be written. + + Returns + ------- + int + Number of bytes actually transferred + StatusCode + Return value of the library call. + """ logger.debug("Serial.write %r" % data) # TODO: How to deal with VI_ATTR_TERMCHAR_EN - end_out, _ = self.get_attribute(constants.VI_ATTR_ASRL_END_OUT) - send_end, _ = self.get_attribute(constants.VI_ATTR_SEND_END_EN) + end_out, _ = self.get_attribute(ResourceAttribute.asrl_end_out) + send_end, _ = self.get_attribute(ResourceAttribute.send_end_enabled) try: # We need to wrap data in common.iter_bytes to Provide Python 2 and 3 compatibility @@ -178,44 +190,67 @@ def write(self, data): except serial.SerialTimeoutException: return 0, StatusCode.error_timeout - def flush(self, mask): - """Flushes device buffers. + def flush(self, mask: BufferOperation) -> StatusCode: + """Flush the specified buffers. + + The buffers can be associated with formatted I/O operations and/or + serial communication. - Corresponds to viFlush function of the VISA library. See: - https://pyvisa.readthedocs.io/en/latest/api/visalibrarybase.html?highlight=flush#pyvisa.highlevel.VisaLibraryBase.flush - for valid values of mask. + Corresponds to viFlush function of the VISA library. + + Parameters + ---------- + mask : constants.BufferOperation + Specifies the action to be taken with flushing the buffer. + The values can be combined using the | operator. However multiple + operations on a single buffer cannot be combined. + + Returns + ------- + constants.StatusCode + Return value of the library call. - :param mask: which buffers to clear. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` """ if ( - mask & constants.VI_READ_BUF - or mask & constants.VI_READ_BUF_DISCARD - or mask & constants.VI_IO_IN_BUF - or mask & constants.VI_IO_IN_BUF_DISCARD + mask & BufferOperation.discard_read_buffer + or mask & BufferOperation.discard_read_buffer_no_io + or mask & BufferOperation.discard_receive_buffer + or mask & BufferOperation.discard_receive_buffer2 ): self.interface.reset_input_buffer() - if mask & constants.VI_WRITE_BUF or mask & constants.VI_IO_OUT_BUF: + if ( + mask & BufferOperation.flush_write_buffer + or mask & BufferOperation.flush_transmit_buffer + ): self.interface.flush() if ( - mask & constants.VI_WRITE_BUF_DISCARD - or mask & constants.VI_IO_OUT_BUF_DISCARD + mask & BufferOperation.discard_write_buffer + or mask & BufferOperation.discard_transmit_buffer ): self.interface.reset_output_buffer() return StatusCode.success - def _get_attribute(self, attribute): + def _get_attribute( + self, attribute: constants.ResourceAttribute + ) -> Tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. - :param attribute: Resource attribute for which the state query is made - :return: The state of the queried attribute for a specified resource, return value of the library call. - :rtype: (unicode | str | list | int, VISAStatus) - """ + Parameters + ---------- + attribute : ResourceAttribute + Attribute for which the state query is made + Returns + ------- + Any + State of the queried attribute for a specified resource + StatusCode + Return value of the library call. + + """ if attribute == constants.VI_ATTR_ASRL_ALLOW_TRANSMIT: raise NotImplementedError @@ -302,17 +337,26 @@ def _get_attribute(self, attribute): raise UnknownAttribute(attribute) - def _set_attribute(self, attribute, attribute_state): + def _set_attribute( + self, attribute: constants.ResourceAttribute, attribute_state: Any + ) -> StatusCode: """Sets the state of an attribute. Corresponds to viSetAttribute function of the VISA library. - :param attribute: Attribute for which the state is to be modified. (Attributes.*) - :param attribute_state: The state of the attribute to be set for the specified object. - :return: return value of the library call. - :rtype: VISAStatus - """ + Parameters + ---------- + attribute : constants.ResourceAttribute + Attribute for which the state is to be modified. (Attributes.*) + attribute_state : Any + The state of the attribute to be set for the specified object. + Returns + ------- + StatusCode + Return value of the library call. + + """ if attribute == constants.VI_ATTR_ASRL_ALLOW_TRANSMIT: raise NotImplementedError diff --git a/pyvisa-py/sessions.py b/pyvisa-py/sessions.py index 3bf7d4ed..aa73065f 100644 --- a/pyvisa-py/sessions.py +++ b/pyvisa-py/sessions.py @@ -1,29 +1,30 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.session - ~~~~~~~~~~~~~~~~~ +"""Base Session class. - Base Session class. +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ import abc import time +from typing import Any, Callable, Dict, Iterator, Optional, Tuple, Type, TypeVar from pyvisa import attributes, constants, logger, rname +from pyvisa.constants import ResourceAttribute, StatusCode +from pyvisa.typing import VISARMSession -from . import common - -StatusCode = constants.StatusCode +#: Type var used when typing register. +T = TypeVar("T", bounds="Session") class UnknownAttribute(Exception): - def __init__(self, attribute): + """Custom exception signaling a VISA attribute is not supported.""" + + def __init__(self, attribute: constants.ResourceAttribute) -> None: self.attribute = attribute - def __str__(self): + def __str__(self) -> str: attr = self.attribute if isinstance(attr, int): try: @@ -43,69 +44,121 @@ class Session(metaclass=abc.ABCMeta): Just makes sure that common methods are defined and information is stored. - :param resource_manager_session: The session handle of the parent Resource - Manager - :param resource_name: The resource name. - :param parsed: the parsed resource name (optional). - If not provided, the resource_name will be parsed. + Parameters + ---------- + resource_manager_session : VISARMSession + Session handle of the parent Resource Manager + resource_name : str + Name of the resource this session is communicating with + parsed : rname.ResourceName, optional + Parsed representation of the resource name. The default is False meaning + that the provided resource name will be parsed. + """ @abc.abstractmethod - def _get_attribute(self, attribute): + def _get_attribute( + self, attribute: constants.ResourceAttribute + ) -> Tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. - :param attribute: Resource attribute for which the state query is made - :return: The state of the queried attribute for a specified resource, - return value of the library call. - :rtype: (unicode | str | list | int, VISAStatus) + Parameters + ---------- + attribute : constants.ResourceAttribute + Resource attribute for which the state query is made + + Returns + ------- + Any + State of the queried attribute for a specified resource + constants.StatusCode + Return value of the library call. + """ @abc.abstractmethod - def _set_attribute(self, attribute, attribute_state): + def _set_attribute( + self, attribute: constants.ResourceAttribute, attribute_state: Any + ) -> StatusCode: """Set the attribute_state value for a given VISA attribute for this session. Use to implement custom logic for attributes. - :param attribute: Resource attribute for which the state query is made. - :param attribute_state: value. - :return: The return value of the library call. - :rtype: VISAStatus + Parameters + ---------- + attribute : constants.ResourceAttribute + Resource attribute for which the state query is made. + attribute_state : Any + Value to which to set the attribute. + + Returns + ------- + StatusCode + The return value of the library call. + """ @abc.abstractmethod - def close(self): - """Close the session. Use it to do final clean ups. + def close(self) -> None: + """Close the session. + + Use it to do final clean ups. + """ + #: Session handle of the parent Resource Manager + resource_manager_session: VISARMSession + + #: Name of the resource this session is communicating with + resource_name: str + + #: Parsed representation of the resource name. + parsed: rname.ResourceName + + #: Session type as (Interface Type, Resource Class) + session_type: Optional[Tuple[constants.InterfaceType, str]] = None + + #: Timeout in seconds to use when opening the resource. + open_timeout: float + + #: Value of the timeout in seconds used for general operation + timeout: Optional[float] + + #: Used as a place holder for the object doing the lowlevel communication. + interface: Any + + #: Used for attributes not handled by the underlying interface. + #: Values are get or set automatically by get_attribute and set_attribute + attrs: Dict[constants.ResourceAttribute, Any] + #: Maps (Interface Type, Resource Class) to Python class encapsulating that #: resource. #: dict[(Interface Type, Resource Class) , Session] - _session_classes = dict() - - #: Session type as (Interface Type, Resource Class) - session_type = None + _session_classes: Dict[ + Tuple[constants.InterfaceType, str], Type["Session"] + ] = dict() @classmethod - def get_low_level_info(cls): + def get_low_level_info(cls) -> str: + """Get info about the backend used by the session.""" return "" @classmethod - def iter_valid_session_classes(cls): - """Yield (Interface Type, Resource Class), Session class pair for - valid sessions classes. - """ - + def iter_valid_session_classes( + cls, + ) -> Iterator[Tuple[constants.InterfaceType, str], Type["Session"]]: + """Iterator over valid sessions classes infos.""" for key, val in cls._session_classes.items(): if issubclass(val, Session): yield key, val @classmethod - def iter_session_classes_issues(cls): - """Yield (Interface Type, Resource Class), Issues class pair for - invalid sessions classes (i.e. those with import errors). - """ + def iter_session_classes_issues( + cls, + ) -> Iterator[Tuple[constants.InterfaceType, str], str]: + """Iterator over invalid sessions classes (i.e. those with import errors).""" for key, val in cls._session_classes.items(): try: yield key, getattr(val, "session_issue") @@ -113,12 +166,23 @@ def iter_session_classes_issues(cls): pass @classmethod - def get_session_class(cls, interface_type, resource_class): - """Return the session class for a given interface type and resource class. + def get_session_class( + cls, interface_type: constants.InterfaceType, resource_class: str + ) -> Type["Session"]: + """Get the session class for a given interface type and resource class. + + Parameters + ---------- + interface_type : constants.InterfaceType + Type of interface. + resource_class : str + Class of resource. + + Returns + ------- + Sessions + Session subclass the most appropriate for the resource. - :type interface_type: constants.InterfaceType - :type resource_class: str - :return: Session """ try: return cls._session_classes[(interface_type, resource_class)] @@ -128,11 +192,23 @@ def get_session_class(cls, interface_type, resource_class): ) @classmethod - def register(cls, interface_type, resource_class): + def register( + cls, interface_type: constants.InterfaceType, resource_class: str + ) -> Callable[[T], T]: """Register a session class for a given interface type and resource class. - :type interface_type: constants.InterfaceType - :type resource_class: str + Parameters + ---------- + interface_type : constants.InterfaceType + Type of interface. + resource_class : str + Class of the resource + + Returns + ------- + Callable[[T], T] + Decorator function to register a session subclass. + """ def _internal(python_class): @@ -150,23 +226,37 @@ def _internal(python_class): return _internal @classmethod - def register_unavailable(cls, interface_type, resource_class, msg): - """Register an unavailable session class for a given interface type and - resource class. + def register_unavailable( + cls, interface_type: constants.InterfaceType, resource_class: str, msg: str + ) -> Type[object]: + """Register that no session class exists. + + This creates a fake session that will raise a ValueError if called. + + Parameters + ---------- + interface_type : constants.InterfaceType + Type of interface. + resource_class : str + Class of the resource + msg : str + Message detailing why no session class exists for this particular + interface type, resource class pair. + + Returns + ------- + Type[object] + Fake session. - raising a ValueError if called. - - :type interface_type: constants.InterfaceType - :type resource_class: str """ - # noinspection PyUnusedLocal + class _internal(object): - session_issue = msg - def __init__(self, *args, **kwargs): - raise ValueError(msg) + #: Message detailing why no session is available. + session_issue: str = msg - _internal.session_issue = msg + def __init__(self, *args, **kwargs) -> None: + raise ValueError(msg) if (interface_type, resource_class) in cls._session_classes: logger.warning( @@ -178,20 +268,19 @@ def __init__(self, *args, **kwargs): cls._session_classes[(interface_type, resource_class)] = _internal def __init__( - self, resource_manager_session, resource_name, parsed=None, open_timeout=None - ): - if isinstance(resource_name, common.MockInterface): - parsed = rname.parse_resource_name(resource_name.resource_name) - parsed["mock"] = resource_name - - elif parsed is None: + self, + resource_manager_session: VISARMSession, + resource_name: str, + parsed: Optional[rname.ResourceName] = None, + open_timeout: Optional[int] = None, + ) -> None: + if parsed is None: parsed = rname.parse_resource_name(resource_name) self.parsed = parsed self.open_timeout = open_timeout - #: Used as a place holder for the object doing the lowlevel - #: communication. + #: Used as a place holder for the object doing the lowlevel communication. self.interface = None #: Used for attributes not handled by the underlying interface. @@ -199,26 +288,25 @@ def __init__( #: set_attribute #: Add your own by overriding after_parsing. self.attrs = { - constants.VI_ATTR_RM_SESSION: resource_manager_session, - constants.VI_ATTR_RSRC_NAME: str(parsed), - constants.VI_ATTR_RSRC_CLASS: parsed.resource_class, - constants.VI_ATTR_INTF_TYPE: parsed.interface_type, - constants.VI_ATTR_TMO_VALUE: (self._get_timeout, self._set_timeout), + ResourceAttribute.resource_manager_session: resource_manager_session, + ResourceAttribute.resource_name: str(parsed), + ResourceAttribute.resource_class: parsed.resource_class, + ResourceAttribute.interface_type: parsed.interface_type, + ResourceAttribute.timeout_value: (self._get_timeout, self._set_timeout), } #: Timeout expressed in second or None for the absence of a timeout. - #: The default value is set when calling - #: self.set_attribute(attr, default_timeout) + #: The default value is set when calling self.set_attribute(attr, default_timeout) self.timeout = None #: Set the default timeout from constants - attr = constants.VI_ATTR_TMO_VALUE + attr = ResourceAttribute.timeout_value default_timeout = attributes.AttributesByID[attr].default self.set_attribute(attr, default_timeout) self.after_parsing() - def after_parsing(self): + def after_parsing(self) -> None: """Override this method to provide custom initialization code, to be called after the resourcename is properly parsed @@ -242,7 +330,7 @@ def after_parsing(self): Getter has same signature as see Session._get_attribute and setter has same signature as see Session._set_attribute. (It is possible to register also see Session._get_attribute and see Session._set_attribute - as getter/setter). Getter and Setter are registered as tupple. + as getter/setter). Getter and Setter are registered as tuple. For readwrite attribute: ` self.attrs[constants.VI_ATTR_] = (, )` @@ -252,151 +340,243 @@ def after_parsing(self): Session._set_attribute ` self.attrs[constants.VI_ATTR_] = (self._get_attribute, self._set_attribute)` + """ pass - def clear(self): + def clear(self) -> StatusCode: """Clears a device. Corresponds to viClear function of the VISA library. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Returns + ------- + StatusCode + Return value of the library call. + """ return StatusCode.error_nonsupported_operation - def flush(self, mask): - """Flushes device buffers. + def flush(self, mask: constants.BufferOperation) -> StatusCode: + """Flush the specified buffers. - Corresponds to viFlush function of the VISA library. See: - https://pyvisa.readthedocs.io/en/latest/api/visalibrarybase.html?highlight=flush#pyvisa.highlevel.VisaLibraryBase.flush - for valid values of mask. + The buffers can be associated with formatted I/O operations and/or + serial communication. + + Corresponds to viFlush function of the VISA library. + + Parameters + ---------- + mask : constants.BufferOperation + Specifies the action to be taken with flushing the buffer. + The values can be combined using the | operator. However multiple + operations on a single buffer cannot be combined. + + Returns + ------- + constants.StatusCode + Return value of the library call. - :param mask: which buffers to clear. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` """ raise NotImplementedError - def read_stb(self): + def read_stb(self) -> Tuple[int, StatusCode]: """Reads a status byte of the service request. Corresponds to viReadSTB function of the VISA library. - :return: Service request status byte, return value of the library call. - :rtype: int, :class:`pyvisa.constants.StatusCode` + Returns + ------- + int + Service request status byte + StatusCode + Return value of the library call. + """ return 0, StatusCode.error_nonsupported_operation - def lock(self, session, lock_type, timeout, requested_key=None): + def lock( + self, + lock_type: constants.Lock, + timeout: int, + requested_key: Optional[str] = None, + ): """Establishes an access mode to the specified resources. Corresponds to viLock function of the VISA library. - :param session: Unique logical identifier to a session. - :param lock_type: Specifies the type of lock requested, either Constants.EXCLUSIVE_LOCK or Constants.SHARED_LOCK. - :param timeout: Absolute time period (in milliseconds) that a resource waits to get unlocked by the - locking session before returning an error. - :param requested_key: This parameter is not used and should be set to VI_NULL when lockType is VI_EXCLUSIVE_LOCK. - :return: access_key that can then be passed to other sessions to share the lock, return value of the library call. - :rtype: str, :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + lock_type : constants.Lock + Specifies the type of lock requested. + timeout : int + Absolute time period (in milliseconds) that a resource waits to get + unlocked by the locking session before returning an error. + requested_key : Optional[str], optional + Requested locking key in the case of a shared lock. For an exclusive + lock it should be None. + + Returns + ------- + Optional[str] + Key that can then be passed to other sessions to share the lock, or + None for an exclusive lock. + constants.StatusCode + Return value of the library call. + """ return "", StatusCode.error_nonsupported_operation - def unlock(self, session): + def unlock(self) -> StatusCode: """Relinquishes a lock for the specified resource. Corresponds to viUnlock function of the VISA library. - :param session: Unique logical identifier to a session. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Returns + ------- + StatusCode + Return value of the library call. + """ return StatusCode.error_nonsupported_operation - def gpib_command(self, command_byte): - """Write GPIB command byte on the bus. + def gpib_command(self, command_byte: bytes) -> Tuple[int, StatusCode]: + """Write GPIB command bytes on the bus. Corresponds to viGpibCommand function of the VISA library. - See: https://linux-gpib.sourceforge.io/doc_html/gpib-protocol.html#REFERENCE-COMMAND-BYTES - :param command_byte: command byte to send - :type command_byte: int, must be [0 255] - :return: Number of written bytes, return value of the library call. - :rtype: int, :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + data : bytes + Data to write. + + Returns + ------- + int + Number of written bytes + constants.StatusCode + Return value of the library call. + """ return 0, StatusCode.error_nonsupported_operation - def assert_trigger(self, protocol): - """Asserts software or hardware trigger. + def assert_trigger(self, protocol: constants.TriggerProtocol) -> StatusCode: + """Assert software or hardware trigger. Corresponds to viAssertTrigger function of the VISA library. - :param protocol: Trigger protocol to use during assertion. (Constants.PROT*) - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + protocol : constants.TriggerProtocol + Trigger protocol to use during assertion. + + Returns + ------- + constants.StatusCode + Return value of the library call. + """ raise NotImplementedError - def gpib_send_ifc(self): + def gpib_send_ifc(self) -> StatusCode: """Pulse the interface clear line (IFC) for at least 100 microseconds. Corresponds to viGpibSendIFC function of the VISA library. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Returns + ------- + constants.StatusCode + Return value of the library call. + """ return StatusCode.error_nonsupported_operation - def gpib_control_ren(self, mode): - """Controls the state of the GPIB Remote Enable (REN) interface line, and optionally the remote/local - state of the device. + def gpib_control_ren( + self, mode: constants.RENLineOperation + ) -> constants.StatusCode: + """Controls the state of the GPIB Remote Enable (REN) interface line. + + Optionally the remote/local state of the device can also be set. Corresponds to viGpibControlREN function of the VISA library. - :param mode: Specifies the state of the REN line and optionally the device remote/local state. - (Constants.VI_GPIB_REN*) - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + mode : constants.RENLineOperation + State of the REN line and optionally the device remote/local state. + + Returns + ------- + constants.StatusCode + Return value of the library call. + """ return StatusCode.error_nonsupported_operation - def gpib_control_atn(self, mode): + def gpib_control_atn(self, mode: constants.ATNLineOperation) -> StatusCode: """Specifies the state of the ATN line and the local active controller state. Corresponds to viGpibControlATN function of the VISA library. - :param mode: Specifies the state of the ATN line and optionally the local active controller state. - (Constants.VI_GPIB_ATN*) - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + mode : constants.ATNLineOperation + State of the ATN line and optionally the local active controller state. + + Returns + ------- + constants.StatusCode + Return value of the library call. + """ return StatusCode.error_nonsupported_operation - def gpib_pass_control(self, primary_address, secondary_address): - """Tell the GPIB device at the specified address to become controller in charge (CIC). + def gpib_pass_control( + self, primary_address: int, secondary_address: int + ) -> StatusCode: + """Tell a GPIB device to become controller in charge (CIC). Corresponds to viGpibPassControl function of the VISA library. - :param primary_address: Primary address of the GPIB device to which you want to pass control. - :param secondary_address: Secondary address of the targeted GPIB device. - If the targeted device does not have a secondary address, - this parameter should contain the value Constants.VI_NO_SEC_ADDR. - :return: return value of the library call. - :rtype: :class:`pyvisa.constants.StatusCode` + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + primary_address : int + Primary address of the GPIB device to which you want to pass control. + secondary_address : int + Secondary address of the targeted GPIB device. + If the targeted device does not have a secondary address, this parameter + should contain the value Constants.VI_NO_SEC_ADDR. + + Returns + ------- + constants.StatusCode + Return value of the library call. + """ return StatusCode.error_nonsupported_operation - def get_attribute(self, attribute): + def get_attribute(self, attribute: ResourceAttribute) -> Tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Does a few checks before and calls before dispatching to `_get_attribute`. - :param attribute: Resource attribute for which the state query is made - :return: The state of the queried attribute for a specified resource, - return value of the library call. - :rtype: (unicode | str | list | int, VISAStatus) + Parameters + ---------- + attribute : ResourceAttribute + Resource attribute for which the state query is made + + Returns + ------- + Any + The state of the queried attribute for a specified resource. + StatusCode + Return value of the library call. + """ # Check if the attribute value is defined. @@ -435,19 +615,28 @@ def get_attribute(self, attribute): logger.exception(str(e)) return 0, StatusCode.error_nonsupported_attribute - def set_attribute(self, attribute, attribute_state): + def set_attribute( + self, attribute: ResourceAttribute, attribute_state: Any + ) -> StatusCode: """Set the attribute_state value for a given VISA attribute for this session. Does a few checks before and calls before dispatching to `_gst_attribute`. - :param attribute: Resource attribute for which the state query is made. - :param attribute_state: value. - :return: The return value of the library call. - :rtype: VISAStatus - """ + Parameters + ---------- + attribute : ResourceAttribute + Resource attribute for which the state query is made. + attribute_state : Any + Value. + Returns + ------- + StatusCode + The return value of the library call. + + """ # Check if the attribute value is defined. try: attr = attributes.AttributesByID[attribute] @@ -494,38 +683,43 @@ def set_attribute(self, attribute, attribute_state): def _read( self, - reader, - count, - end_indicator_checker, - suppress_end_en, - termination_char, - termination_char_en, - timeout_exception, - ): + reader: Callable[[], bytes], + count: int, + end_indicator_checker: Callable[[bytes], bool], + suppress_end_en: bool, + termination_char: str, + termination_char_en: bool, + timeout_exception: Type[Exception], + ) -> Tuple[bytes, StatusCode]: """Reads data from device or interface synchronously. Corresponds to viRead function of the VISA library. - :param reader: Function to read one or more bytes. - :type reader: () -> bytes - :param count: Number of bytes to be read. - :type count: int - :param end_indicator_checker: Function to check if the message is - complete. - :type end_indicator_checker: (bytes) -> boolean - :param suppress_end_en: suppress end. - :type suppress_end_en: bool - :param termination_char: Stop reading if this character is received. - :type suppress_end_en: int or str - :param termination_char_en: termination char enabled. - :type termination_char_en: boolean - :param: timeout_exception: Exception to capture time out for the given - interface. - :type: Exception - :return: data read, return value of the library call. - :rtype: bytes, constants.StatusCode - """ + Parameters + ---------- + reader : Callable[[], bytes] + Function to read one or more bytes. + count : int + Number of bytes to be read. + end_indicator_checker : Callable[[bytes], bool] + Function to check if the message is complete. + suppress_end_en : bool + Suppress end. + termination_char : str + Stop reading if this character is received. + termination_char_en : bool + Is termination char enabled. + timeout_exception : Type[Exception] + Exception to capture time out for the given interface. + + Returns + ------- + bytes + Data read from the resource. + StatusCode + Return value of the library call. + """ # NOTE: Some interfaces return not only a single byte but a complete # block for each read therefore we must handle the case that the # termination character is in the middle of the block or that the @@ -569,7 +763,7 @@ def _read( if finish_time and time.time() > finish_time: return bytes(out), StatusCode.error_timeout - def _get_timeout(self, attribute): + def _get_timeout(self, attribute: ResourceAttribute) -> Tuple[int, StatusCode]: """ Returns timeout calculated value from python way to VI_ way In VISA, the timeout is expressed in milliseconds or using the @@ -587,7 +781,7 @@ def _get_timeout(self, attribute): ret_value = int(self.timeout * 1000.0) return ret_value, StatusCode.success - def _set_timeout(self, attribute, value): + def _set_timeout(self, attribute: ResourceAttribute, value: Optional[int]): """ Sets timeout calculated value from python way to VI_ way In VISA, the timeout is expressed in milliseconds or using the diff --git a/pyvisa-py/tcpip.py b/pyvisa-py/tcpip.py index 12d8c517..7db2ed20 100644 --- a/pyvisa-py/tcpip.py +++ b/pyvisa-py/tcpip.py @@ -1,27 +1,24 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.tcpip - ~~~~~~~~~~~~~~~ +"""TCPIP Session implementation using Python Standard library. - TCPIP Session implementation using Python Standard library. +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ import random import select import socket import time +from typing import Any, List, Optional, Tuple from pyvisa import attributes, constants, errors +from pyvisa.constants import ResourceAttribute, StatusCode from . import common from .protocols import rpc, vxi11 from .sessions import Session, UnknownAttribute -StatusCode = constants.StatusCode - # Conversion between VXI11 error codes and VISA status # TODO this is so far a best guess, in particular 6 and 29 are likely wrong VXI11_ERRORS_TO_VISA = { @@ -44,17 +41,26 @@ @Session.register(constants.InterfaceType.tcpip, "INSTR") class TCPIPInstrSession(Session): - """A TCPIP Session that uses the network standard library to do the low - level communication using VXI-11 + """A TCPIP Session built on socket standard library using VXI-11 protocol.""" - """ + #: Maximum size of a chunk of data in bytes. + max_recv_size: int + + #: Time to wait before erroring with a timeout when trying to acquire a lock + lock_timeout: int + + #: Unique ID of the client used to authenticate messages. + client_id: int + + #: ID of the link used for VXI-11 communication + link: int @staticmethod - def list_resources(): + def list_resources() -> List[str]: # TODO: is there a way to get this? return [] - def after_parsing(self): + def after_parsing(self) -> None: # TODO: board_number not handled # vx11 expect all timeouts to be expressed in ms and should be integers try: @@ -64,7 +70,6 @@ def after_parsing(self): except rpc.RPCError: raise errors.VisaIOError(constants.VI_ERROR_RSRC_NFOUND) - self.max_recv_size = 1024 self.lock_timeout = 10000 self.client_id = random.getrandbits(31) @@ -82,7 +87,7 @@ def after_parsing(self): attribute = getattr(constants, "VI_ATTR_" + name) self.attrs[attribute] = attributes.AttributesByID[attribute].default - def close(self): + def close(self) -> None: try: self.interface.destroy_link(self.link) except (errors.VisaIOError, socket.error, rpc.RPCError) as e: @@ -92,14 +97,23 @@ def close(self): self.link = None self.interface = None - def read(self, count): + def read(self, count: int) -> Tuple[bytes, StatusCode]: """Reads data from device or interface synchronously. Corresponds to viRead function of the VISA library. - :param count: Number of bytes to be read. - :return: data read, return value of the library call. - :rtype: bytes, VISAStatus + Parameters + ---------- + count : int + Number of bytes to be read. + + Returns + ------- + bytes + Data read + StatusCode + Return value of the library call. + """ if count < self.max_recv_size: chunk_length = count @@ -146,18 +160,24 @@ def read(self, count): return bytes(read_data), status - def write(self, data): + def write(self, data: bytes) -> Tuple[int, StatusCode]: """Writes data to device or interface synchronously. Corresponds to viWrite function of the VISA library. - :param data: data to be written. - :type data: str - :return: Number of bytes actually transferred, return value of the - library call. - :rtype: int, VISAStatus - """ + Parameters + ---------- + data : bytes + Data to be written. + Returns + ------- + int + Number of bytes actually transferred + StatusCode + Return value of the library call. + + """ send_end, _ = self.get_attribute(constants.VI_ATTR_SEND_END_EN) chunk_size = 1024 @@ -190,17 +210,24 @@ def write(self, data): except vxi11.Vxi11Error: return 0, StatusCode.error_timeout - def _get_attribute(self, attribute): + def _get_attribute(self, attribute: ResourceAttribute) -> Tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. - :param attribute: Resource attribute for which the state query is made - :return: The state of the queried attribute for a specified resource, - return value of the library call. - :rtype: (unicode | str | list | int, VISAStatus) - """ + Parameters + ---------- + attribute : + Resource attribute for which the state query is made + Returns + ------- + Any + The state of the queried attribute for a specified resource + StatusCode + Return value of the library call. + + """ if attribute == constants.VI_ATTR_TCPIP_ADDR: return self.parsed.host_address, StatusCode.success @@ -224,85 +251,114 @@ def _get_attribute(self, attribute): raise UnknownAttribute(attribute) - def _set_attribute(self, attribute, attribute_state): + def _set_attribute( + self, attribute: ResourceAttribute, attribute_state: Any + ) -> StatusCode: """Sets the state of an attribute. Corresponds to viSetAttribute function of the VISA library. - :param attribute: Attribute for which the state is to be modified. - (Attributes.*) - :param attribute_state: The state of the attribute to be set for the - specified object. - :return: return value of the library call. - :rtype: VISAStatus - """ + Parameters + ---------- + attribute : ResourceAttribute + Attribute for which the state is to be modified. + attribute_state : Any + The state of the attribute to be set for the specified object. + + Returns + ------- + StatusCode + Return value of the library call. + """ raise UnknownAttribute(attribute) - def assert_trigger(self, protocol): + def assert_trigger(self, protocol: constants.TriggerProtocol): """Asserts software or hardware trigger. Corresponds to viAssertTrigger function of the VISA library. - :param protocol: Trigger protocol to use during assertion. - (Constants.PROT*) - :return: return value of the library call. - :rtype: VISAStatus - """ + Parameters + ---------- + protocol : constants.TriggerProtocol + Trigger protocol to use during assertion. Only default is supported. + + Returns + ------- + StatusCode + Return value of the library call. + """ + # XXX make this nicer (either validate protocol or pass it) error = self.interface.device_trigger( - self.link, 0, self.lock_timeout, self._io_timeout + self.link, 0, self.lock_timeout, self._io_timeout, ) return VXI11_ERRORS_TO_VISA[error] - def clear(self): + def clear(self) -> StatusCode: """Clears a device. Corresponds to viClear function of the VISA library. - :return: return value of the library call. - :rtype: VISAStatus """ - error = self.interface.device_clear( self.link, 0, self.lock_timeout, self._io_timeout ) return VXI11_ERRORS_TO_VISA[error] - def read_stb(self): + def read_stb(self) -> Tuple[int, StatusCode]: """Reads a status byte of the service request. Corresponds to viReadSTB function of the VISA library. - :return: Service request status byte, return value of the library call. - :rtype: int, :class:`pyvisa.constants.StatusCode` - """ + Returns + ------- + int + Service request status byte + StatusCode + Return value of the library call. + """ error, stb = self.interface.device_read_stb( self.link, 0, self.lock_timeout, self._io_timeout ) return stb, VXI11_ERRORS_TO_VISA[error] - def lock(self, lock_type, timeout, requested_key=None): + def lock( + self, + lock_type: constants.Lock, + timeout: int, + requested_key: Optional[str] = None, + ) -> Tuple[str, constants.StatusCode]: """Establishes an access mode to the specified resources. Corresponds to viLock function of the VISA library. - :param lock_type: Specifies the type of lock requested, either - Constants.EXCLUSIVE_LOCK or Constants.SHARED_LOCK. - :param timeout: Absolute time period (in milliseconds) that a resource - waits to get unlocked by the locking session before returning an - error. - :param requested_key: This parameter is not used and should be set to - VI_NULL when lockType is VI_EXCLUSIVE_LOCK. - :return: access_key that can then be passed to other sessions to share - the lock, return value of the library call. - :rtype: str, VISAStatus - """ + Parameters + ---------- + session : VISASession + Unique logical identifier to a session. + lock_type : constants.Lock + Specifies the type of lock requested. + timeout : int + Absolute time period (in milliseconds) that a resource waits to get + unlocked by the locking session before returning an error. + requested_key : Optional[str], optional + Requested locking key in the case of a shared lock. For an exclusive + lock it should be None. + + Returns + ------- + Optional[str] + Key that can then be passed to other sessions to share the lock, or + None for an exclusive lock. + StatusCode + Return value of the library call. + """ # TODO: lock type not implemented flags = 0 @@ -310,19 +366,24 @@ def lock(self, lock_type, timeout, requested_key=None): return "", VXI11_ERRORS_TO_VISA[error] - def unlock(self): - """Relinquishes a lock for the specified resource. + def unlock(self) -> constants.StatusCode: + """Relinquish a lock for the specified resource. Corresponds to viUnlock function of the VISA library. - :return: return value of the library call. - :rtype: VISAStatus + Returns + ------- + StatusCode + Return value of the library call. + """ error = self.interface.device_unlock(self.link) return VXI11_ERRORS_TO_VISA[error] - def _set_timeout(self, attribute, value): + def _set_timeout( + self, attribute: ResourceAttribute, value: Optional[int] + ) -> StatusCode: """ Sets timeout calculated value from python way to VI_ way """ @@ -353,12 +414,15 @@ class TCPIPSocketSession(Session): # timeout. The absolute minimum is 1 ms as a consequence. # This is valid for connect and read operations + #: Maximum size of a chunk of data in bytes. + max_recv_size: int + @staticmethod - def list_resources(): + def list_resources() -> List[str]: # TODO: is there a way to get this? return [] - def after_parsing(self): + def after_parsing(self) -> None: # TODO: board_number not handled ret_status = self._connect() @@ -390,7 +454,7 @@ def after_parsing(self): attribute = getattr(constants, "VI_ATTR_" + name) self.attrs[attribute] = attributes.AttributesByID[attribute].default - def _connect(self): + def _connect(self) -> None: timeout = self.open_timeout / 1000.0 if self.open_timeout else 10.0 try: self.interface = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -424,18 +488,27 @@ def _connect(self): # min_select_timeout select_timout = max(select_timout / 2.0, min_select_timeout) - def close(self): + def close(self) -> None: self.interface.close() self.interface = None - def read(self, count): + def read(self, count: int) -> Tuple[bytes, StatusCode]: """Reads data from device or interface synchronously. Corresponds to viRead function of the VISA library. - :param count: Number of bytes to be read. - :return: data read, return value of the library call. - :rtype: bytes, VISAStatus + Parameters + ----------- + count : int + Number of bytes to be read. + + Returns + ------- + bytes + Data read from the device + StatusCode + Return value of the library call. + """ if count < self.max_recv_size: chunk_length = count @@ -511,18 +584,24 @@ def read(self, count): # min_select_timeout select_timout = max(select_timout / 2.0, min_select_timeout) - def write(self, data): + def write(self, data: bytes) -> Tuple[int, StatusCode]: """Writes data to device or interface synchronously. Corresponds to viWrite function of the VISA library. - :param data: data to be written. - :type data: str - :return: Number of bytes actually transferred, return value of the - library call. - :rtype: int, VISAStatus - """ + Parameters + ---------- + data : bytes + Data to be written. + + Returns + ------- + int + Number of bytes actually transferred + StatusCode + Return value of the library call. + """ chunk_size = 4096 num = sz = len(data) @@ -548,63 +627,90 @@ def write(self, data): return offset, StatusCode.success - def _get_tcpip_nodelay(self, attribute): + def _get_tcpip_nodelay( + self, attribute: ResourceAttribute + ) -> Tuple[constants.VisaBoolean, StatusCode]: if self.interface: value = self.interface.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) return ( - constants.VI_TRUE if value == 1 else constants.VI_FALSE, + constants.VisaBoolean.true + if value == 1 + else constants.VisaBoolean.false, StatusCode.success, ) - return 0, StatusCode.error_nonsupported_attribute + return constants.VisaBoolean.false, StatusCode.error_nonsupported_attribute - def _set_tcpip_nodelay(self, attribute, attribute_state): + def _set_tcpip_nodelay( + self, attribute: ResourceAttribute, attribute_state: bool + ) -> StatusCode: if self.interface: self.interface.setsockopt( socket.IPPROTO_TCP, socket.TCP_NODELAY, 1 if attribute_state else 0 ) return StatusCode.success - return 0, StatusCode.error_nonsupported_attribute + return StatusCode.error_nonsupported_attribute - def _get_tcpip_keepalive(self, attribute): + def _get_tcpip_keepalive( + self, attribute: ResourceAttribute + ) -> Tuple[constants.VisaBoolean, StatusCode]: if self.interface: value = self.interface.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE) return ( - constants.VI_TRUE if value == 1 else constants.VI_FALSE, + constants.VisaBoolean.true + if value == 1 + else constants.VisaBoolean.false, StatusCode.success, ) - return 0, StatusCode.error_nonsupported_attribute + return constants.VisaBoolean.false, StatusCode.error_nonsupported_attribute - def _set_tcpip_keepalive(self, attribute, attribute_state): + def _set_tcpip_keepalive( + self, attribute: ResourceAttribute, attribute_state: bool + ) -> StatusCode: if self.interface: self.interface.setsockopt( socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1 if attribute_state else 0 ) return StatusCode.success - return 0, StatusCode.error_nonsupported_attribute + return StatusCode.error_nonsupported_attribute - def _get_attribute(self, attribute): + def _get_attribute(self, attribute: ResourceAttribute) -> Tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. - :param attribute: Resource attribute for which the state query is made - :return: The state of the queried attribute for a specified resource, - return value of the library call. - :rtype: (unicode | str | list | int, VISAStatus) + Parameters + ---------- + attribute : ResourceAttribute + Attribute for which the state query is made + + Returns + ------- + Any + State of the queried attribute for a specified resource + StatusCode + Return value of the library call. + """ raise UnknownAttribute(attribute) - def _set_attribute(self, attribute, attribute_state): + def _set_attribute( + self, attribute: ResourceAttribute, attribute_state: Any + ) -> StatusCode: """Sets the state of an attribute. Corresponds to viSetAttribute function of the VISA library. - :param attribute: Attribute for which the state is to be modified. - (Attributes.*) - :param attribute_state: The state of the attribute to be set for the - specified object. - :return: return value of the library call. - :rtype: VISAStatus - """ + Parameters + ---------- + attribute : constants.ResourceAttribute + Attribute for which the state is to be modified. (Attributes.*) + attribute_state : Any + The state of the attribute to be set for the specified object. + Returns + ------- + StatusCode + Return value of the library call. + + """ raise UnknownAttribute(attribute) diff --git a/pyvisa-py/usb.py b/pyvisa-py/usb.py index 40decddd..976ac625 100644 --- a/pyvisa-py/usb.py +++ b/pyvisa-py/usb.py @@ -1,17 +1,16 @@ # -*- coding: utf-8 -*- -""" - pyvisa-py.usb - ~~~~~~~~~~~~~ +"""Serial Session implementation using PyUSB. - Serial Session implementation using PyUSB. +:copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. - :copyright: 2014-2020 by PyVISA-py Authors, see AUTHORS for more details. - :license: MIT, see LICENSE for more details. """ import errno +from typing import Any, List, Tuple from pyvisa import attributes, constants +from pyvisa.constants import ResourceAttribute, StatusCode from .common import logger from .sessions import Session, UnknownAttribute @@ -41,9 +40,6 @@ raise -StatusCode = constants.StatusCode - - class USBTimeoutException(Exception): """Exception used internally to indicate USB timeout.""" @@ -56,12 +52,12 @@ class USBSession(Session): """ @staticmethod - def list_resources(): + def list_resources() -> List[str]: """Return list of resources for this type of USB device""" raise NotImplementedError @classmethod - def get_low_level_info(cls): + def get_low_level_info(cls) -> str: try: ver = usb.__version__ except AttributeError: @@ -75,7 +71,22 @@ def get_low_level_info(cls): return "via PyUSB (%s). Backend: %s" % (ver, backend) - def _get_timeout(self, attribute): + def after_parsing(self) -> None: + self.interface = self._intf_cls( + int(self.parsed.manufacturer_id, 0), + int(self.parsed.model_code, 0), + self.parsed.serial_number, + ) + + for name in ("SEND_END_EN", "TERMCHAR", "TERMCHAR_EN"): + attribute = getattr(constants, "VI_ATTR_" + name) + self.attrs[attribute] = attributes.AttributesByID[attribute].default + + # Force setting the timeout to get the proper value + attribute = constants.VI_ATTR_TMO_VALUE + self.set_attribute(attribute, attributes.AttributesByID[attribute].default) + + def _get_timeout(self, attribute: ResourceAttribute) -> int: if self.interface: if self.interface.timeout == 2 ** 32 - 1: self.timeout = None @@ -83,7 +94,7 @@ def _get_timeout(self, attribute): self.timeout = self.interface.timeout / 1000 return super(USBSession, self)._get_timeout(attribute) - def _set_timeout(self, attribute, value): + def _set_timeout(self, attribute: ResourceAttribute, value: int) -> StatusCode: status = super(USBSession, self)._set_timeout(attribute, value) timeout = int(self.timeout * 1000) if self.timeout else 2 ** 32 - 1 timeout = min(timeout, 2 ** 32 - 1) @@ -91,14 +102,23 @@ def _set_timeout(self, attribute, value): self.interface.timeout = timeout return status - def read(self, count): + def read(self, count: int) -> Tuple[bytes, StatusCode]: """Reads data from device or interface synchronously. Corresponds to viRead function of the VISA library. - :param count: Number of bytes to be read. - :return: data read, return value of the library call. - :rtype: (bytes, VISAStatus) + Parameters + ----------- + count : int + Number of bytes to be read. + + Returns + ------- + bytes + Data read from the device + StatusCode + Return value of the library call. + """ def _usb_reader(): @@ -110,15 +130,15 @@ def _usb_reader(): raise USBTimeoutException() raise - supress_end_en, _ = self.get_attribute(constants.VI_ATTR_SUPPRESS_END_EN) + supress_end_en, _ = self.get_attribute(ResourceAttribute.suppress_end_enabled) if supress_end_en: raise ValueError( "VI_ATTR_SUPPRESS_END_EN == True is currently unsupported by pyvisa-py" ) - term_char, _ = self.get_attribute(constants.VI_ATTR_TERMCHAR) - term_char_en, _ = self.get_attribute(constants.VI_ATTR_TERMCHAR_EN) + term_char, _ = self.get_attribute(ResourceAttribute.termchar) + term_char_en, _ = self.get_attribute(ResourceAttribute.termchar_enabled) return self._read( _usb_reader, @@ -130,18 +150,25 @@ def _usb_reader(): USBTimeoutException, ) - def write(self, data): + def write(self, data: bytes) -> Tuple[int, StatusCode]: """Writes data to device or interface synchronously. Corresponds to viWrite function of the VISA library. - :param data: data to be written. - :type data: bytes - :return: Number of bytes actually transferred, return value of the library call. - :rtype: (int, VISAStatus) - """ + Parameters + ---------- + data : bytes + Data to be written. - send_end, _ = self.get_attribute(constants.VI_ATTR_SEND_END_EN) + Returns + ------- + int + Number of bytes actually transferred + StatusCode + Return value of the library call. + + """ + send_end, _ = self.get_attribute(ResourceAttribute.send_end_enabled) count = self.interface.write(data) @@ -150,58 +177,60 @@ def write(self, data): def close(self): self.interface.close() - def _get_attribute(self, attribute): + def _get_attribute( + self, attribute: constants.ResourceAttribute + ) -> Tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. - :param attribute: Resource attribute for which the state query is made - :return: The state of the queried attribute for a specified resource, return value of the library call. - :rtype: (unicode | str | list | int, VISAStatus) - """ + Parameters + ---------- + attribute : ResourceAttribute + Attribute for which the state query is made + Returns + ------- + Any + State of the queried attribute for a specified resource + StatusCode + Return value of the library call. + + """ raise UnknownAttribute(attribute) - def _set_attribute(self, attribute, attribute_state): + def _set_attribute( + self, attribute: constants.ResourceAttribute, attribute_state: Any + ) -> StatusCode: """Sets the state of an attribute. Corresponds to viSetAttribute function of the VISA library. - :param attribute: Attribute for which the state is to be modified. (Attributes.*) - :param attribute_state: The state of the attribute to be set for the specified object. - :return: return value of the library call. - :rtype: VISAStatus - """ - - raise UnknownAttribute(attribute) + Parameters + ---------- + attribute : constants.ResourceAttribute + Attribute for which the state is to be modified. (Attributes.*) + attribute_state : Any + The state of the attribute to be set for the specified object. - def after_parsing(self): - self.interface = self._intf_cls( - int(self.parsed.manufacturer_id, 0), - int(self.parsed.model_code, 0), - self.parsed.serial_number, - ) + Returns + ------- + StatusCode + Return value of the library call. - for name in ("SEND_END_EN", "TERMCHAR", "TERMCHAR_EN"): - attribute = getattr(constants, "VI_ATTR_" + name) - self.attrs[attribute] = attributes.AttributesByID[attribute].default - - # Force setting the timeout to get the proper value - attribute = constants.VI_ATTR_TMO_VALUE - self.set_attribute(attribute, attributes.AttributesByID[attribute].default) + """ + raise UnknownAttribute(attribute) @Session.register(constants.InterfaceType.usb, "INSTR") class USBInstrSession(USBSession): - """Base class for drivers that communicate with instruments - via usb port using pyUSB - """ + """Class for USBTMC devices.""" #: Class to use when instantiating the interface _intf_cls = usbtmc.USBTMC @staticmethod - def list_resources(): + def list_resources() -> List[str]: out = [] fmt = ( "USB%(board)s::%(manufacturer_id)s::%(model_code)s::" @@ -250,15 +279,13 @@ def list_resources(): @Session.register(constants.InterfaceType.usb, "RAW") class USBRawSession(USBSession): - """Base class for drivers that communicate with usb raw devices - via usb port using pyUSB - """ + """Class for RAW devices.""" #: Class to use when instantiating the interface _intf_cls = usbraw.USBRawDevice @staticmethod - def list_resources(): + def list_resources() -> List[str]: out = [] fmt = ( "USB%(board)s::%(manufacturer_id)s::%(model_code)s::" diff --git a/setup.cfg b/setup.cfg index 606a2ffe..9cbd528f 100644 --- a/setup.cfg +++ b/setup.cfg @@ -70,6 +70,8 @@ per-file-ignores = pyvisa-py/protocols/vxi11.py:E221 pyvisa-py/serial.py:C901 +[mypy] +follow_imports = ignore [isort] multi_line_output = 3 @@ -77,4 +79,5 @@ include_trailing_comma = true force_grid_wrap = 0 use_parentheses = true line_length = 88 -known_third_party = pyvisa,numpy,setuptools,typing_extensions +skip = pyvisa-py/__init__.py +known_third_party = numpy,setuptools,typing_extensions,pyvisa