Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
self._test_handler = None
if self.CHECK_NO_WARNING:
self._test_handler = th = TestHandler()
th.setLevel(logging.WARNING)
logger.addHandler(th)
Exp|exp Matches either the preceding or following expression. The or
operator | matches the entire expression that precedes or
follows it and not just the character that precedes or follows
it. For example, VXI|GPIB means (VXI)|(GPIB), not VX(I|G)PIB.
(exp) Grouping characters or expressions.
:param resources: iterable of resources.
:param query: query expression.
"""
if '{' in query:
query, _ = query.split('{')
logger.warning('optional part of the query expression not supported. '
'See filter2')
try:
query = query.replace('?', '.')
matcher = re.compile(query, re.IGNORECASE)
except re.error:
raise errors.VisaIOError(constants.VI_ERROR_INV_EXPR)
return tuple(res for res in resources if matcher.match(res))
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`
"""
# ibpct need to get the device id matching the primary and secondary address
logger.debug("GPIB.pass control")
try:
did = gpib.dev(self.parsed.board, primary_address, secondary_address)
except gpib.GpibError:
logger.exception(
"Failed to get id for %s, %d", primary_address, secondary_address
)
return StatusCode.error_resource_not_found
status = gpib_lib.ibpct(did)
return convert_gpib_status(status)
session_issue: str = msg
def __init__(self, *args, **kwargs) -> None:
raise ValueError(msg)
def _get_attribute(self, attr):
raise NotImplementedError()
def _set_attribute(self, attr, value):
raise NotImplementedError()
def close(self):
raise NotImplementedError()
if (interface_type, resource_class) in cls._session_classes:
logger.warning(
"%s is already registered in the ResourceManager. "
"Overwriting with unavailable %s",
((interface_type, resource_class), msg),
)
cls._session_classes[(interface_type, resource_class)] = _internal
def _find_listeners():
"""Find GPIB listeners.
"""
for board, boardpad in _find_boards():
for i in range(31):
try:
if boardpad != i and gpib.listener(board, i):
yield board, i
except gpib.GpibError as e:
logger.debug(
"GPIB board %i addr %i error in _find_listeners(): %s",
board,
i,
repr(e),
)
else:
self.attrs[attribute] = attribute_state
return status
# Dispatch to `_set_attribute`, which must be implemented by subclasses
try:
return self._set_attribute(attribute, attribute_state)
except ValueError:
return StatusCode.error_nonsupported_attribute_state
except NotImplementedError:
e = UnknownAttribute(attribute)
logger.exception(str(e))
return StatusCode.error_nonsupported_attribute
except UnknownAttribute as e:
logger.exception(str(e))
return StatusCode.error_nonsupported_attribute
def write(self, data):
"""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
"""
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)
try:
# We need to wrap data in common.iter_bytes to Provide Python 2 and 3 compatibility
if end_out in (SerialTermination.none, SerialTermination.termination_break):
data = common.iter_bytes(data)
elif end_out == SerialTermination.last_bit:
last_bit, _ = self.get_attribute(constants.VI_ATTR_ASRL_DATA_BITS)
mask = 1 << (last_bit - 1)
data = common.iter_bytes(data, mask, send_end)
elif end_out == SerialTermination.termination_char:
mask = 1 << (last_bit - 1)
data = common.iter_bytes(data, mask, send_end)
elif end_out == SerialTermination.termination_char:
term_char, _ = self.get_attribute(constants.VI_ATTR_TERMCHAR)
data = common.iter_bytes(data + common.int_to_byte(term_char))
else:
raise ValueError("Unknown value for VI_ATTR_ASRL_END_OUT: %s" % end_out)
count = 0
for d in data:
count += self.interface.write(d)
if end_out == SerialTermination.termination_break:
logger.debug("Serial.sendBreak")
self.interface.sendBreak()
return count, StatusCode.success
except serial.SerialTimeoutException:
return 0, StatusCode.error_timeout
def _internal(python_class):
if (interface_type, resource_class) in cls._session_classes:
logger.warning(
"%s is already registered in the "
"ResourceManager. Overwriting with %s",
((interface_type, resource_class), python_class),
)
python_class.session_type = (interface_type, resource_class)
cls._session_classes[(interface_type, resource_class)] = python_class
return python_class
def close(self):
"""Closes the VISA session and marks the handle as invalid.
"""
try:
logger.debug('%s - closing', self._resource_name,
extra=self._logging_extra)
self.before_close()
self.visalib.close(self.session)
logger.debug('%s - is closed', self._resource_name,
extra=self._logging_extra)
self.session = None
except errors.InvalidSession:
pass