Source code for pylablib.core.devio.comm_backend

"""
Routines for defining a unified interface across multiple backends.
"""

from ..utils import funcargparse, general, net, py3, module, functions as func_utils
from . import interface
from . import backend_logger
from .base import DeviceError

import time
import re
import contextlib
import warnings
import functools


### Generic backend interface ###


[docs]class DeviceBackendError(DeviceError): """Generic exception relaying a backend error""" def __init__(self, exc): msg="backend exception: {} ('{}')".format(repr(exc),str(exc)) super().__init__(msg) self.backend_exc=exc
[docs]def reraise(func): """Wrapper for a backend method which intercepts backend exceptions and re-emits them as a subclass of :exc:`DeviceBackendError` defined in the class""" @functools.wraps(func) def wrapped(self, *args, **kwargs): if self.Error is None or self.BackendError is None: return func(self,*args,**kwargs) try: return func(self,*args,**kwargs) except self.BackendError as exc: ReraiseError=getattr(self,"ReraiseError",self.Error) raise ReraiseError(exc) from exc return wrapped
[docs]def logerror(func): """Wrapper for a backend method which logs if any errors escaped""" @functools.wraps(func) def wrapped(self, *args, **kwargs): try: return func(self,*args,**kwargs) except self.Error as exc: self._log("error",str(exc)) raise return wrapped
logger=None
[docs]class IDeviceCommBackend: """ An abstract class for a device communication backend. Connection is automatically opened on creation. Args: conn: Connection parameters (depend on the backend). timeout (float): Default timeout (in seconds). term_write (str): Line terminator for writing operations. term_read (str): Line terminator for reading operations. datatype (str): Type of the returned data; can be ``"bytes"`` (return ``bytes`` object), ``"str"`` (return ``str`` object), or ``"auto"`` (default Python result: ``str`` in Python 2 and ``bytes`` in Python 3) reraise_error: if not ``None``, specifies an error to be re-raised on any backend exception (by default, use backend-specific error); should be a subclass of :exc:`DeviceBackendError`. """ Error=DeviceBackendError BackendError=None """Base class for the errors raised by the backend operations""" _default_operation_cooldown={"default":0.} def __init__(self, conn, timeout=None, term_write=None, term_read=None, datatype="auto", reraise_error=None): # pylint: disable=unused-argument funcargparse.check_parameter_range(datatype,"datatype",{"auto","str","bytes"}) self.datatype=datatype self.conn=conn self.term_write=term_write self.term_read=term_read self._operation_cooldown=dict(self._default_operation_cooldown) if reraise_error is not None: self.Error=reraise_error _conn_params=["addr"] _default_conn=[None] @classmethod def _conn_to_dict(cls, conn): """Turn connection parameters (tuple or dict) into a full dictionary using class-specific parameter names""" if isinstance(conn, dict): return conn if isinstance(conn, (tuple,list)): return dict(zip(cls._conn_params,conn)) return {cls._conn_params[0]:conn}
[docs] @classmethod def combine_conn(cls, conn1, conn2): """Combined two connection parameters into a single dictionary (`conn1` overrides `conn2`)""" conn=cls._conn_to_dict(conn2).copy() conn.update(cls._conn_to_dict(conn1)) return conn
[docs] @classmethod def get_backend_name(cls): """Get string representation of the backend (e.g., ``"serial"``, ``"visa"``, or ``"network"``)""" return getattr(cls,"_backend",None)
def _to_datatype(self, data): if self.datatype=="auto": return data if self.datatype=="str": return py3.as_str(data) return py3.as_bytes(data)
[docs] def open(self): """Open the connection""" pass
[docs] def close(self): """Close the connection""" pass
[docs] def is_opened(self): """Check if the device is connected""" return True
def __bool__(self): return self.is_opened() __nonzero__=__bool__ # Python 2 compatibility def _log(self, operation, value): """Log the operation (used for testing and debugging)""" if logger: logger.log(operation,value)
[docs] def lock(self, timeout=None): """Lock the access to the device from other threads/processes (isn't necessarily implemented)""" pass
[docs] def unlock(self): """Unlock the access to the device from other threads/processes (isn't necessarily implemented)""" pass
[docs] @contextlib.contextmanager def locking(self, timeout=None): # pylint: disable=unused-argument """Context manager for lock & unlock""" yield
[docs] def setup_cooldown(self, **kwargs): """ Setup cooldown times for various operations. The arguments are of the form ``kind=value``, where ``value`` is the cooldown time (in seconds), and ``kind`` is the operation kind (common kinds are ``open``, ``close``, ``read``, ``write``, ``timeout``, and ``flush``). ``kind`` can also be ``default`` (default value for all kind), or ``all`` (reset all cooldown values to this value). The cooldowns of the given kinds are usually called after the corresponding operation (it is necessary for some devices, otherwise the communication can freeze or crush). Default cooldown values are specified by ``_default_operation_cooldown`` class attribute dictionary. """ if "all" in kwargs: self._operation_cooldown={"default":kwargs.pop("all")} self._operation_cooldown.update(kwargs)
[docs] def cooldown(self, kind="default"): """ Cooldown between the operations. ``kind`` specifies the operation kind (common kinds are ``open``, ``close``, ``read``, ``write``, ``timeout``, and ``flush``); ``"default"`` corresponds to the default cooldown (usually, specified as 0). Called automatically by various backend operations, so usually there is no need to call explicitly. """ cooldown=self._operation_cooldown.get(kind,self._operation_cooldown.get("default",0)) if cooldown>0: time.sleep(cooldown)
[docs] def set_timeout(self, timeout): """Set operations timeout (in seconds)""" pass
[docs] def get_timeout(self): """Get operations timeout (in seconds)""" return None
[docs] @contextlib.contextmanager def using_timeout(self, timeout=None): """Context manager for usage of a different timeout inside a block""" if timeout is not None: to=self.get_timeout() # pylint: disable=assignment-from-none if to!=timeout: self.set_timeout(timeout) try: yield finally: if timeout is not None and to!=timeout: self.set_timeout(to)
[docs] def readline(self, remove_term=True, timeout=None, skip_empty=True): """ Read a single line from the device. Args: remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. skip_empty (bool): If ``True``, ignore empty lines (works only for ``remove_term==True``). """ raise NotImplementedError("IDeviceCommBackend.readline")
[docs] def readlines(self, lines_num, remove_term=True, timeout=None, skip_empty=True): """ Read multiple lines from the device. Parameters are the same as in :func:`readline`. """ return [self.readline(remove_term=remove_term,timeout=timeout,skip_empty=skip_empty) for _ in range(lines_num)]
[docs] def read(self, size=None): """ Read data from the device. If `size` is not None, read `size` bytes (the standard timeout applies); otherwise, read all available data (return immediately). """ raise NotImplementedError("IDeviceCommBackend.read")
[docs] def flush_read(self): """Flush the device output (read all the available data; return the number of bytes read)""" return len(self.read())
[docs] def write(self, data, flush=True, read_echo=False, read_echo_delay=0, read_echo_lines=1): """ Write data to the device. If ``flush==True``, flush the write buffer. If ``read_echo==True``, wait for `read_echo_delay` seconds and then perform :func:`readline` (`read_echo_lines` times). """ raise NotImplementedError("IDeviceCommBackend.write")
[docs] def ask(self, query, delay=0., read_all=False): """ Perform a write followed by a read, with `delay` in between. If ``read_all==True``, read all the available data; otherwise, read a single line. """ self.write(query) if delay: time.sleep(delay) if read_all: return self.read() else: return self.readline()
[docs] @staticmethod def list_resources(desc=False): # pylint: disable=unused-argument """ List all available resources for this backend. If ``desc==False``, return list of connections (usually strings or tuples), which can be used to connect to the device. Otherwise, return a list of descriptions, which have more info, but can be backend-dependent. Might not be implemented (depending on the backend), in which case returns ``None``. """ return None
### Helper functions ###
[docs]def remove_longest_term(msg, terms): """ Remove the longest terminator among `terms` from the end of the message. """ tcs=0 for t in terms: if msg.endswith(py3.as_builtin_bytes(t)): tcs=max(tcs,len(t)) return msg[:-tcs]
### Specific backends ### _backends={} _backend_errors={} _backend_install_message=("{name:} package is missing. You can install it via PyPi as 'pip install {pkg:}'. " "If it is installed, check if it imports correctly by running 'import {mod:}'") try: try: import pyvisa as visa except ImportError: import visa
[docs] class DeviceVisaError(DeviceBackendError): """Visa backend operation error"""
[docs] class VisaDeviceBackend(IDeviceCommBackend): """ NIVisa backend (via pyVISA). Connection is automatically opened on creation. Args: conn (str): Connection string. timeout (float): Default timeout (in seconds). term_write (str): Line terminator for writing operations; appended to the data term_read (str): Line terminator for reading operations (specifies when :func:`readline` stops). do_lock (bool): If ``True``, employ locking operations; otherwise, locking function does nothing. datatype (str): Type of the returned data; can be ``"bytes"`` (return `bytes` object), ``"str"`` (return `str` object), or ``"auto"`` (default Python result: `str` in Python 2 and `bytes` in Python 3) reraise_error: if not ``None``, specifies an error to be re-raised on any backend exception (by default, use backend-specific error); should be a subclass of :exc:`DeviceBackendError`. """ _backend="visa" BackendError=visa.VisaIOError """Base class for the errors raised by the backend operations""" Error=DeviceVisaError if module.cmp_versions(visa.__version__,"1.6")=="<": # older pyvisa versions have a slightly different interface @reraise def _set_timeout(self, timeout): self.instr.timeout=timeout @reraise def _get_timeout(self): return self.instr.timeout @reraise def _open_resource(self, conn): if not self.term_write.endswith(self.term_read): raise NotImplementedError("PyVisa version <1.6 doesn't support different terminators for reading and writing; update to a newer version by running 'pip install --upgrade pyvisa'") instr=visa.instrument(conn) # pylint: disable=no-member instr.term_chars=self.term_read self.term_write=self.term_write[:len(self.term_write)-len(self.term_read)] return instr _lock_default=False def _lock(self, timeout=None): raise NotImplementedError("PyVisa version <1.6 doesn't support locking; update to a newer version by running 'pip install --upgrade pyvisa'") def _unlock(self): raise NotImplementedError("PyVisa version <1.6 doesn't support locking; update to a newer version by running 'pip install --upgrade pyvisa'") def _lock_context(self, timeout=None): raise NotImplementedError("PyVisa version <1.6 doesn't support locking; update to a newer version by running 'pip install --upgrade pyvisa'") @reraise def _read_term(self): return py3.as_builtin_bytes(self.instr.term_chars) else: @reraise def _set_timeout(self, timeout): self.instr.timeout=timeout*1000. # in newer versions timeout is in ms @reraise def _get_timeout(self): return self.instr.timeout/1000. # in newer versions timeout is in ms @reraise def _open_resource(self, conn): instr=visa.ResourceManager().open_resource(conn) instr.read_termination=self.term_read instr.write_termination=self.term_write self.term_read=self.term_write="" return instr _lock_default=False @reraise def _lock(self, timeout=None): self.instr.lock(timeout=timeout*1000. if timeout is not None else None) @reraise def _unlock(self): self.instr.unlock() @reraise def _lock_context(self, timeout=None): return self.instr.lock_context(timeout=timeout*1000. if timeout is not None else None) @reraise def _read_term(self): return py3.as_builtin_bytes(self.instr.read_termination)
[docs] @staticmethod def list_resources(desc=False): try: return visa.ResourceManager().list_resources_info() if desc else visa.ResourceManager().list_resources() except VisaDeviceBackend.BackendError as e: raise VisaDeviceBackend.Error(e) from e
if module.cmp_versions(visa.__version__,"1.9")=="<": # older pyvisa versions have a slightly different interface @reraise def _read_raw(self, size): chunk_size=self.instr.chunk_size data=bytearray() with self.instr.ignore_warning(visa.constants.VI_SUCCESS_DEV_NPRESENT,visa.constants.VI_SUCCESS_MAX_CNT): while len(data)<size: to_read=min(chunk_size,size-len(data)) chunk,_=self.instr.visalib.read(self.instr.session,to_read) data.extend(chunk) return bytes(data) else: @reraise def _read_raw(self, size): return self.instr.read_bytes(size) @reraise def _read_all(self): data=bytearray() with self.using_timeout(0): while True: try: chunk=self.instr.read_raw() data.extend(chunk) except visa.VisaIOError as err: if err.abbreviation=="VI_ERROR_TMO": return bytes(data) else: raise def __init__(self, conn, timeout=10., term_write=None, term_read=None, do_lock=None, datatype="auto", reraise_error=None): if term_write is None: term_write=b"\r\n" if term_read is None: term_read=b"\n" IDeviceCommBackend.__init__(self,conn,term_write=term_write,term_read=term_read,datatype=datatype,reraise_error=reraise_error) try: self.instr=self._open_resource(self.conn) self.opened=True self._do_lock=do_lock if do_lock is not None else self._lock_default self.cooldown("open") self.set_timeout(timeout) except self.BackendError as e: raise self.Error(e) from e
[docs] @reraise def open(self): """Open the connection""" self.instr.open() self.opened=True self.cooldown("open")
[docs] @reraise def close(self): """Close the connection""" self.instr.close() self.opened=False self.cooldown("close")
[docs] def is_opened(self): return self.opened
[docs] def lock(self, timeout=None): """Lock the access to the device from other threads/processes""" if self._do_lock: self.lock(timeout=timeout)
[docs] def unlock(self): """Unlock the access to the device from other threads/processes""" if self._do_lock: self.unlock()
[docs] def locking(self, timeout=None): """Context manager for lock & unlock""" if self._do_lock: return self._lock_context(timeout=timeout) else: return general.DummyResource()
[docs] def set_timeout(self, timeout): """Set operations timeout (in seconds)""" if timeout is not None: self._set_timeout(timeout) self.cooldown("timeout")
[docs] def get_timeout(self): """Get operations timeout (in seconds)""" return self._get_timeout()
[docs] @logerror @reraise def readline(self, remove_term=True, timeout=None, skip_empty=True): """ Read a single line from the device. Args: remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. skip_empty (bool): If ``True``, ignore empty lines (works only for ``remove_term==True``). """ with self.using_timeout(timeout): while True: result=self.instr.read_raw() if remove_term: term=self._read_term() if term and result.endswith(term): result=result[:-len(term)] if (not skip_empty) or result: break self.cooldown("read") self._log("read",result) return self._to_datatype(result)
[docs] @logerror def read(self, size=None): """ Read data from the device. If `size` is not None, read `size` bytes (the standard timeout applies); otherwise, read all available data (return immediately). """ result=self._read_all() if size is None else self._read_raw(size=size) self.cooldown("read") self._log("read",result) return self._to_datatype(result)
[docs] @logerror @reraise def write(self, data, flush=True, read_echo=False, read_echo_delay=0, read_echo_lines=1): """ Write data to the device. If ``read_echo==True``, wait for `read_echo_delay` seconds and then perform :func:`readline` (`read_echo_lines` times). `flush` parameter is ignored. """ self._log("write",data) data=py3.as_builtin_bytes(data) if self.term_write: data=data+py3.as_builtin_bytes(self.term_write) self.instr.write_raw(data) self.cooldown("write") if read_echo_delay>0.: time.sleep(read_echo_delay) if read_echo: for _ in range(read_echo_lines): self.readline()
@reraise def __repr__(self): return "VisaDeviceBackend("+self.instr.__repr__()+")"
_backends["visa"]=VisaDeviceBackend except ImportError: pass _backend_errors["visa"]=_backend_install_message.format(name="PyVISA",pkg="pyvisa",mod="pyvisa") try: import serial try: import serial.tools.list_ports as serial_list_ports except ImportError: serial_list_ports=None
[docs] class DeviceSerialError(DeviceBackendError): """Serial backend operation error"""
[docs] class SerialDeviceBackend(IDeviceCommBackend): """ Serial backend (via pySerial). Connection is automatically opened on creation. Args: conn: Connection parameters. Can be either a string (for a port), or a list/tuple ``(port, baudrate, bytesize, parity, stopbits, xonxoff, rtscts, dsrdtr)`` supplied to the serial connection (default is ``('COM1',19200,8,'N',1,0,0,0)``), or a dict with the same parameters. timeout (float): Default timeout (in seconds). term_write (str): Line terminator for writing operations; appended to the data term_read (str): List of possible single-char terminator for reading operations (specifies when :func:`readline` stops). connect_on_operation (bool): If ``True``, the connection is normally closed, and is opened only on the operations (normally two processes can't be simultaneously connected to the same device). open_retry_times (int): Number of times the connection is attempted before giving up. no_dtrrts (bool): If ``True``, turn off DTR and RTS status lines before opening (e.g., turns off reset-on-connection for Arduino controllers). datatype (str): Type of the returned data; can be ``"bytes"`` (return `bytes` object), ``"str"`` (return `str` object), or ``"auto"`` (default Python result: `str` in Python 2 and `bytes` in Python 3) reraise_error: if not ``None``, specifies an error to be re-raised on any backend exception (by default, use backend-specific error); should be a subclass of :exc:`DeviceBackendError`. """ _backend="serial" BackendError=serial.SerialException """Base class for the errors raised by the backend operations""" Error=DeviceSerialError _conn_params=["port","baudrate","bytesize","parity","stopbits","xonxoff","rtscts","dsrdtr"] _default_conn=["COM1",19200,8,"N",1,0,0,0] def __init__(self, conn, timeout=10., term_write=None, term_read=None, connect_on_operation=False, open_retry_times=3, no_dtrrts=False, datatype="auto", reraise_error=None): conn_dict=self.combine_conn(conn,self._default_conn) if term_write is None: term_write=b"\r\n" if term_read is None: term_read=b"\n" if isinstance(term_read,py3.anystring): term_read=[term_read] IDeviceCommBackend.__init__(self,conn_dict.copy(),term_write=term_write,term_read=term_read,datatype=datatype,reraise_error=reraise_error) port=conn_dict.pop("port") try: self.instr=serial.serial_for_url(port,do_not_open=True,**conn_dict) self.opened=True if no_dtrrts: try: self.instr.setDTR(0) self.instr.setRTS(0) except self.BackendError: warnings.warn("Cannot set DTR for an unconnected device") if not connect_on_operation: self.instr.open() self._connect_on_operation=connect_on_operation self._opened_stack=0 self._open_retry_times=open_retry_times self.cooldown("open") self.set_timeout(timeout) except self.BackendError as e: raise self.Error(e) from e @reraise def _do_open(self): general.retry_wait(self.instr.open, self._open_retry_times, 0.3) @reraise def _do_close(self): general.retry_wait(self.instr.close, self._open_retry_times, 0.3)
[docs] def open(self): """Open the connection""" if not self._connect_on_operation and not self.opened: self._do_open() self.opened=True
[docs] def close(self): """Close the connection""" if not self._connect_on_operation and self.opened: self._do_close() self.opened=False
[docs] def is_opened(self): return self.opened
def _op_open(self): if self._connect_on_operation: if not self._opened_stack: self._do_open() self._opened_stack=self._opened_stack+1 def _op_close(self): if self._connect_on_operation: self._opened_stack=self._opened_stack-1 if not self._opened_stack: self._do_close()
[docs] @contextlib.contextmanager def single_op(self): """ Context manager for a single operation. If ``connect_on_operation==True`` during creation, wrapping several command in `single_op` prevents the connection from being closed and reopened between the operations (only opened in the beginning and closed in the end). """ self._op_open() try: yield finally: self._op_close()
[docs] @reraise def set_timeout(self, timeout): """Set operations timeout (in seconds)""" if timeout is not None: self.instr.timeout=timeout self.cooldown("timeout")
[docs] @reraise def get_timeout(self): """Get operations timeout (in seconds)""" return self.instr.timeout
@reraise def _read_terms(self, terms=(), timeout=None, error_on_timeout=True): result=b"" singlechar_terms=all(len(t)==1 for t in terms) terms=[py3.as_builtin_bytes(t) for t in terms] with self.single_op(): with self.using_timeout(timeout): while True: c=self.instr.read(1 if terms else 8) result=result+c if c==b"": if error_on_timeout and terms: raise self.Error("timeout during read") return result if singlechar_terms: if c in terms: return result else: for t in terms: if result.endswith(t): return result
[docs] @logerror def readline(self, remove_term=True, timeout=None, skip_empty=True, error_on_timeout=True): # pylint: disable=arguments-differ """ Read a single line from the device. Args: remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. skip_empty (bool): If ``True``, ignore empty lines (works only for ``remove_term==True``). error_on_timeout (bool): If ``False``, return an incomplete line instead of raising the error on timeout. """ while True: result=self._read_terms(self.term_read or [],timeout=timeout,error_on_timeout=error_on_timeout) self.cooldown("read") if remove_term and self.term_read: result=remove_longest_term(result,self.term_read) if not (skip_empty and remove_term and (not result)): break self._log("read",result) return self._to_datatype(result)
[docs] @logerror @reraise def read(self, size=None): """ Read data from the device. If `size` is not None, read `size` bytes (usual timeout applies); otherwise, read all available data (return immediately). """ with self.single_op(): if size is None: result=self._read_terms(timeout=0,error_on_timeout=False) else: result=self.instr.read(size=size) if len(result)!=size: raise self.Error("read returned less than expected: {} instead of {}".format(len(result),size)) self.cooldown("read") self._log("read",result) return self._to_datatype(result)
[docs] @logerror def read_multichar_term(self, term, remove_term=True, timeout=None, error_on_timeout=True): """ Read a single line with multiple possible terminators. Args: term: Either a string (single multi-char terminator) or a list of strings (multiple terminators). remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. error_on_timeout (bool): If ``False``, return an incomplete line instead of raising the error on timeout. """ if isinstance(term,py3.anystring): term=[term] result=self._read_terms(term,timeout=timeout,error_on_timeout=error_on_timeout) self.cooldown("read") if remove_term and term: result=remove_longest_term(result,term) self._log("read",result) return self._to_datatype(result)
[docs] @logerror @reraise def write(self, data, flush=True, read_echo=False, read_echo_delay=0, read_echo_lines=1): """ Write data to the device. If ``flush==True``, flush the write buffer. If ``read_echo==True``, wait for `read_echo_delay` seconds and then perform :func:`readline` (`read_echo_lines` times). """ self._log("write",data) with self.single_op(): data=py3.as_builtin_bytes(data) if self.term_write: data=data+py3.as_builtin_bytes(self.term_write) self.instr.write(data) self.cooldown("write") if flush: self.instr.flush() self.cooldown("flush") if read_echo_delay>0.: time.sleep(read_echo_delay) if read_echo: for _ in range(read_echo_lines): self.readline()
@reraise def __repr__(self): return "SerialDeviceBackend("+self.instr.__repr__()+")"
[docs] @staticmethod def list_resources(desc=False): if serial_list_ports is not None: try: return [(p if desc else p[0]) for p in serial_list_ports.comports()] except SerialDeviceBackend.BackendError as e: raise SerialDeviceBackend.Error(e) from e
_backends["serial"]=SerialDeviceBackend except (ImportError, AttributeError): pass _backend_errors["serial"]=_backend_install_message.format(name="PySerial",pkg="pyserial",mod="serial") try: import ft232
[docs] class DeviceFT232Error(DeviceBackendError): """FT232 backend operation error"""
[docs] class FT232DeviceBackend(IDeviceCommBackend): """ FT232 backend (via pyft232). Connection is automatically opened on creation. Args: conn: Connection parameters. Can be either a string (for a port), or a list/tuple ``(port, baudrate, bytesize, parity, stopbits, xonxoff, rtscts)`` supplied to the serial connection (default is ``('COM1',19200,8,'N',1,0,0,0)``), or a dict with the same parameters. timeout (float): Default timeout (in seconds). term_write (str): Line terminator for writing operations; appended to the data term_read (str): List of possible single-char terminator for reading operations (specifies when :func:`readline` stops). open_retry_times (int): Number of times the connection is attempted before giving up. datatype (str): Type of the returned data; can be ``"bytes"`` (return `bytes` object), ``"str"`` (return `str` object), or ``"auto"`` (default Python result: `str` in Python 2 and `bytes` in Python 3) reraise_error: if not ``None``, specifies an error to be re-raised on any backend exception (by default, use backend-specific error); should be a subclass of :exc:`DeviceBackendError`. """ _backend="ft232" BackendError=ft232.Ft232Exception """Base class for the errors raised by the backend operations""" Error=DeviceFT232Error _conn_params=["port","baudrate","bytesize","parity","stopbits","xonxoff","rtscts"] _default_conn=[None,9600,8,"N",1,0,0] def __init__(self, conn, timeout=10., term_write=None, term_read=None, open_retry_times=3, datatype="auto", reraise_error=None): conn_dict=self.combine_conn(conn,self._default_conn) if term_write is None: term_write=b"\r\n" if term_read is None: term_read=b"\n" if isinstance(term_read,py3.anystring): term_read=[term_read] conn_dict=conn_dict.copy() conn_dict["port"]=str(conn_dict["port"]) IDeviceCommBackend.__init__(self,conn_dict.copy(),term_write=term_write,term_read=term_read,datatype=datatype,reraise_error=reraise_error) port=conn_dict.pop("port") self.opened=False try: self.instr=self._open_instr(port,conn_dict) self.opened=True self._open_retry_times=open_retry_times self.cooldown("open") self.set_timeout(timeout) self._conn_params=(port,conn_dict,timeout) except self.BackendError as e: raise self.Error(e) from e def _open_instr(self, port, params): sig=func_utils.funcsig(ft232.Ft232) if "serial_number" in sig.arg_names: # pyft232 v0.11 signature change return ft232.Ft232(serial_number=port,**params) else: return ft232.Ft232(port,**params) @reraise def _do_open(self): if self.is_opened(): return def reopen(): self.instr=self._open_instr(self._conn_params[0],self._conn_params[1]) self.set_timeout(self._conn_params[2]) self.opened=True general.retry_wait(reopen, self._open_retry_times, 0.3) @reraise def _do_close(self): if self.is_opened(): general.retry_wait(self.instr.close, self._open_retry_times, 0.3) self.opened=False
[docs] def open(self): """Open the connection""" self._do_open()
[docs] def close(self): """Close the connection""" self._do_close()
[docs] def is_opened(self): return self.opened
[docs] @contextlib.contextmanager def single_op(self): """ Context manager for a single operation. Does nothing. """ yield
[docs] @reraise def set_timeout(self, timeout): """Set operations timeout (in seconds)""" if timeout is not None: if timeout<1E-3: timeout=1E-3 # 0 is infinite timeout self.instr.timeout=timeout self.cooldown("timeout")
[docs] @reraise def get_timeout(self): """Get operations timeout (in seconds)""" return self.instr.timeout
@reraise def _read_terms(self, terms=(), timeout=None, error_on_timeout=True): result=b"" singlechar_terms=all(len(t)==1 for t in terms) terms=[py3.as_builtin_bytes(t) for t in terms] with self.single_op(): with self.using_timeout(timeout): while True: c=self.instr.read(1 if terms else 8) result=result+c if c==b"": if error_on_timeout and terms: raise self.Error("timeout during read") return result if singlechar_terms: if c in terms: return result else: for t in terms: if result.endswith(t): return result
[docs] @logerror def readline(self, remove_term=True, timeout=None, skip_empty=True, error_on_timeout=True): # pylint: disable=arguments-differ """ Read a single line from the device. Args: remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. skip_empty (bool): If ``True``, ignore empty lines (works only for ``remove_term==True``). error_on_timeout (bool): If ``False``, return an incomplete line instead of raising the error on timeout. """ while True: result=self._read_terms(self.term_read or [],timeout=timeout,error_on_timeout=error_on_timeout) self.cooldown("read") if remove_term and self.term_read: result=remove_longest_term(result,self.term_read) if not (skip_empty and remove_term and (not result)): break self._log("read",result) return self._to_datatype(result)
[docs] @logerror @reraise def read(self, size=None): """ Read data from the device. If `size` is not None, read `size` bytes (usual timeout applies); otherwise, read all available data (return immediately). """ with self.single_op(): if size is None: result=self._read_terms(timeout=0,error_on_timeout=False) else: result=self.instr.read(size=size) if len(result)!=size: raise self.Error("read returned less data than expected") self.cooldown("read") self._log("read",result) return self._to_datatype(result)
[docs] @logerror def read_multichar_term(self, term, remove_term=True, timeout=None, error_on_timeout=True): """ Read a single line with multiple possible terminators. Args: term: Either a string (single multi-char terminator) or a list of strings (multiple terminators). remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. error_on_timeout (bool): If ``False``, return an incomplete line instead of raising the error on timeout. """ if isinstance(term,py3.anystring): term=[term] result=self._read_terms(term,timeout=timeout,error_on_timeout=error_on_timeout) self.cooldown("read") if remove_term and term: result=remove_longest_term(result,term) self._log("read",result) return self._to_datatype(result)
[docs] @logerror @reraise def write(self, data, flush=True, read_echo=False, read_echo_delay=0, read_echo_lines=1): """ Write data to the device. If ``flush==True``, flush the write buffer. If ``read_echo==True``, wait for `read_echo_delay` seconds and then perform :func:`readline` (`read_echo_lines` times). """ self._log("write",data) with self.single_op(): data=py3.as_builtin_bytes(data) if self.term_write: data=data+py3.as_builtin_bytes(self.term_write) self.instr.write(data) self.cooldown("write") if flush: self.instr.flush() self.cooldown("flush") if read_echo_delay>0.: time.sleep(read_echo_delay) if read_echo: for _ in range(read_echo_lines): self.readline()
@reraise def __repr__(self): return "FT232DeviceBackend("+self.instr.__repr__()+")" @staticmethod def _as_str(v): if isinstance(v,py3.anystring): return py3.as_str(v) if isinstance(v,(tuple,list)): return type(v)([FT232DeviceBackend._as_str(e) for e in v]) return s
[docs] @staticmethod def list_resources(desc=False): try: devices=[FT232DeviceBackend._as_str(d) for d in ft232.list_devices()] return [d if desc else d[0] for d in devices] except FT232DeviceBackend.BackendError as e: raise FT232DeviceBackend.Error(e) from e
_backends["ft232"]=FT232DeviceBackend except (ImportError,NameError,OSError): pass _backend_errors["ft232"]=_backend_install_message.format(name="pyft232",pkg="pyft232",mod="ft232")
[docs]class DeviceNetworkError(DeviceBackendError): """Network backend operation error"""
[docs]class NetworkDeviceBackend(IDeviceCommBackend): """ Serial backend (via pySerial). Connection is automatically opened on creation. Args: conn: Connection parameters. Can be either a string ``"IP:port"`` (e.g., ``"127.0.0.1:80"``), or a tuple ``(IP,port)``, where `IP` is a string and `port` is a number. timeout (float): Default timeout (in seconds). term_write (str): Line terminator for writing operations; appended to the data term_read (str): List of possible single-char terminator for reading operations (specifies when :func:`readline` stops). datatype (str): Type of the returned data; can be ``"bytes"`` (return `bytes` object), ``"str"`` (return `str` object), or ``"auto"`` (default Python result: `str` in Python 2 and `bytes` in Python 3) reraise_error: if not ``None``, specifies an error to be re-raised on any backend exception (by default, use backend-specific error); should be a subclass of :exc:`DeviceBackendError`. Note: If `term_read` is a string, its behavior is different from the VISA backend: instead of being a multi-char terminator it is assumed to be a set of single-char terminators. If multi-char terminator is required, `term_read` should be a single-element list instead of a string. """ _backend="network" BackendError=net.socket.error """Base class for the errors raised by the backend operations""" Error=DeviceNetworkError def __init__(self, conn, timeout=10., term_write=None, term_read=None, datatype="auto", reraise_error=None): if term_write is None: term_write="\r\n" if term_read is None: term_read="\r\n" if isinstance(term_read,py3.anystring): term_read=[term_read] conn=self._conn_to_dict(conn) self._split_addr(conn) IDeviceCommBackend.__init__(self,conn,term_write=term_write,term_read=term_read,datatype=datatype,reraise_error=reraise_error) try: self.socket=None self.open() self.cooldown("open") self.set_timeout(timeout) except self.BackendError as e: raise self.Error(e) from e _conn_params=["addr","port"] _default_conn=["127.0.0.1",80] @classmethod def _split_addr(cls, conn): addr=conn["addr"] addr_split=addr.split(":") if len(addr_split)==2: conn["addr"],conn["port"]=addr_split[0],int(addr_split[1]) elif len(addr_split)>2: raise ValueError("invalid device address: {}".format(conn))
[docs] @reraise def open(self): """Open the connection""" self.close() self.socket=net.ClientSocket(send_method="fixedlen",recv_method="fixedlen") self.socket.connect(self.conn["addr"],self.conn["port"])
[docs] @reraise def close(self): """Close the connection""" if self.socket is not None: self.socket.close() self.socket=None
[docs] @reraise def is_opened(self): return bool(self.socket)
[docs] @reraise def set_timeout(self, timeout): """Set operations timeout (in seconds)""" self.socket.set_timeout(timeout)
[docs] @reraise def get_timeout(self): """Get operations timeout (in seconds)""" return self.socket.get_timeout()
[docs] @logerror @reraise def readline(self, remove_term=True, timeout=None, skip_empty=True): """ Read a single line from the device. Args: remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. skip_empty (bool): If ``True``, ignore empty lines (works only for ``remove_term==True``). """ while True: with self.using_timeout(timeout): result=self.socket.recv_delimiter(self.term_read,strict=True) self.cooldown("read") if remove_term and self.term_read: result=remove_longest_term(result,self.term_read) if not (skip_empty and remove_term and (not result)): break self._log("read",result) return self._to_datatype(result)
[docs] @logerror @reraise def read(self, size=None): """ Read data from the device. If `size` is not None, read `size` bytes (usual timeout applies); otherwise, read all available data (return immediately). """ if size is None: return self.socket.recv_all() else: result=self.socket.recv_fixedlen(size) self.cooldown("read") self._log("read",result) return self._to_datatype(result)
[docs] @logerror @reraise def read_multichar_term(self, term, remove_term=True, timeout=None): """ Read a single line with multiple possible terminators. Args: term: Either a string (single multi-char terminator) or a list of strings (multiple terminators). remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. """ if isinstance(term,py3.anystring): term=[term] with self.socket.using_timeout(timeout): result=self.socket.recv_delimiter(term,strict=True) self.cooldown("read") if remove_term and term: result=remove_longest_term(result,term) self._log("read",result) return self._to_datatype(result)
[docs] @logerror @reraise def write(self, data, flush=True, read_echo=False, read_echo_delay=0, read_echo_lines=1): """ Write data to the device. If ``read_echo==True``, wait for `read_echo_delay` seconds and then perform :func:`readline` (`read_echo_lines` times). `flush` parameter is ignored. """ self._log("write",data) self.socket.send_delimiter(data,self.term_write) self.cooldown("write") if read_echo_delay>0.: time.sleep(read_echo_delay) if read_echo: for _ in range(read_echo_lines): self.readline()
@reraise def __repr__(self): return "NetworkDeviceBackend("+self.socket.__repr__()+")"
_backends["network"]=NetworkDeviceBackend try: import usb import usb.backend.libusb0 import usb.backend.libusb1 import usb.backend.openusb
[docs] class DeviceUSBError(DeviceBackendError): """USB backend operation error"""
[docs] class PyUSBDeviceBackend(IDeviceCommBackend): """ USB backend (via PyUSB package). Connection is automatically opened on creation. Args: conn: Connection parameters. Can be either a string (for a port), or a list/tuple ``(vendorID, productID, index, endpoint_read, endpoint_write, backend)`` supplied to the connection (default is ``(0x0000,0x0000,0,0x00,0x01,'libusb1')``, which is invalid for most devices), or a dict with the same parameters. ``vendorID`` and ``productID`` specify device kind, ``index`` is an integer index (starting from zero) of the device among several identical (i.e., with the same ids) ones, and ``endpoint_read`` and ``endpoint_write`` specify connection endpoints for the specific device. timeout (float): Default timeout (in seconds). term_write (str): Line terminator for writing operations; appended to the data term_read (str): List of possible single-char terminator for reading operations (specifies when :func:`readline` stops). datatype (str): Type of the returned data; can be ``"bytes"`` (return `bytes` object), ``"str"`` (return `str` object), or ``"auto"`` (default Python result: `str` in Python 2 and `bytes` in Python 3) reraise_error: if not ``None``, specifies an error to be re-raised on any backend exception (by default, use backend-specific error); should be a subclass of :exc:`DeviceBackendError`. """ _backend="pyusb" BackendError=usb.USBError """Base class for the errors raised by the backend operations""" Error=DeviceUSBError _conn_params=["vendorID","productID","index","endpoint_read","endpoint_write","backend"] _default_conn=[0x0000,0x0000,0,0x00,0x01,"libusb1"] _usb_backends={"libusb0":usb.backend.libusb0, "libusb1":usb.backend.libusb1, "openusb":usb.backend.openusb} def __init__(self, conn, timeout=10., term_write=None, term_read=None, check_read_size=True, datatype="auto", reraise_error=None): conn_dict=self.combine_conn(conn,self._default_conn) funcargparse.check_parameter_range(conn_dict["backend"],"usb_backend",self._usb_backends) if isinstance(term_read,py3.anystring): term_read=[term_read] IDeviceCommBackend.__init__(self,conn_dict.copy(),term_write=term_write,term_read=term_read,datatype=datatype,reraise_error=reraise_error) self.timeout=timeout self.check_read_size=check_read_size try: self.open() except self.BackendError as e: raise self.Error(e) from e self.cooldown("open")
[docs] @reraise def open(self): """Open the connection""" idx=self.conn["index"] backend=self._usb_backends[self.conn["backend"]].get_backend() all_devs=list(usb.core.find(idVendor=self.conn["vendorID"],idProduct=self.conn["productID"],backend=backend,find_all=True)) if len(all_devs)<idx+1: raise self.Error("can't find device with VID=0x{:04x} PID=0x{:04x} index {}; {} devices found".format( self.conn["vendorID"],self.conn["productID"],idx,len(all_devs))) self.instr=all_devs[idx] self.ep_read=self.conn["endpoint_read"] self.ep_write=self.conn["endpoint_write"] self.cooldown("open") self.opened=True
[docs] @reraise def close(self): """Close the connection""" self.instr.finalize() self.opened=False
[docs] def is_opened(self): return self.opened
[docs] def set_timeout(self, timeout): """Set operations timeout (in seconds)""" if timeout is not None: self.timeout=timeout
[docs] def get_timeout(self): """Get operations timeout (in seconds)""" return self.timeout
def _timeout(self, timeout=None): timeout=self.timeout if timeout is None else timeout return None if timeout is None else int(timeout*1000) @reraise def _read_terms(self, terms=(), read_block_size=65536, timeout=None, error_on_timeout=True): result=b"" singlechar_terms=all(len(t)==1 for t in terms) terms=[py3.as_builtin_bytes(t) for t in terms] while True: try: c=self.instr.read(self.ep_read,1 if terms else read_block_size,timeout=self._timeout(timeout)).tobytes() except self.BackendError: c=b"" result=result+c if c==b"": if error_on_timeout and terms: raise self.Error("timeout during read") return result if not terms: return result if singlechar_terms: if c in terms: return result else: for t in terms: if result.endswith(t): return result
[docs] @logerror def readline(self, remove_term=True, timeout=None, skip_empty=True, error_on_timeout=True): # pylint: disable=arguments-differ """ Read a single line from the device. Args: remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. skip_empty (bool): If ``True``, ignore empty lines (works only for ``remove_term==True``). error_on_timeout (bool): If ``False``, return an incomplete line instead of raising the error on timeout. """ while True: result=self._read_terms(self.term_read or [],timeout=timeout,error_on_timeout=error_on_timeout) self.cooldown("read") if remove_term and self.term_read: result=remove_longest_term(result,self.term_read) if not (skip_empty and remove_term and (not result)): break self._log("read",result) return self._to_datatype(result)
[docs] @logerror @reraise def read(self, size=None, max_read_size=65536): # pylint: disable=arguments-differ """ Read data from the device. If `size` is not None, read `size` bytes (usual timeout applies); otherwise, read all available data (return immediately). """ if size is None: result=self._read_terms(read_block_size=max_read_size,timeout=1E-3,error_on_timeout=False) else: result=self.instr.read(self.ep_read,size,timeout=self._timeout()).tobytes() if len(result)!=size and self.check_read_size: raise self.Error("read returned less than expected {} instead of {}".format(len(result),size)) self.cooldown("read") self._log("read",result) return self._to_datatype(result)
[docs] @logerror def read_multichar_term(self, term, remove_term=True, timeout=None, error_on_timeout=True): """ Read a single line with multiple possible terminators. Args: term: Either a string (single multi-char terminator) or a list of strings (multiple terminators). remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. error_on_timeout (bool): If ``False``, return an incomplete line instead of raising the error on timeout. """ if isinstance(term,py3.anystring): term=[term] result=self._read_terms(term,timeout=timeout,error_on_timeout=error_on_timeout) self.cooldown("read") if remove_term and term: result=remove_longest_term(result,term) self._log("read",result) return self._to_datatype(result)
[docs] @logerror @reraise def write(self, data, flush=True, read_echo=False, read_echo_delay=0, read_echo_lines=1): """ Write data to the device. If ``read_echo==True``, wait for `read_echo_delay` seconds and then perform :func:`readline` (`read_echo_lines` times). `flush` parameter is ignored. """ self._log("write",data) data=py3.as_builtin_bytes(data) if self.term_write: data=data+py3.as_builtin_bytes(self.term_write) self.instr.write(self.ep_write,data,timeout=self._timeout()) self.cooldown("write") if read_echo: if read_echo_delay>0.: time.sleep(read_echo_delay) for _ in range(read_echo_lines): self.readline()
@reraise def __repr__(self): return "PyUSBDeviceBackend("+self.instr.__repr__()+")"
[docs] @staticmethod def list_resources(desc=False, **kwargs): try: devs=list(usb.core.find(find_all=True,**kwargs)) except PyUSBDeviceBackend.BackendError as e: raise PyUSBDeviceBackend.Error from e if desc: return devs indices={} res=[] for d in devs: vid,pid=d.idVendor,d.idProduct i=indices.get((vid,pid),0) indices[(vid,pid)]=i+1 res.append((vid,pid,i)) return res
_backends["pyusb"]=PyUSBDeviceBackend except ImportError: pass _backend_errors["pyusb"]=_backend_install_message.format(name="PyUSB",pkg="pyusb",mod="usb")
[docs]class DeviceRecordedError(DeviceBackendError): """Recorded backend operation error"""
[docs]class RecordedDeviceBackend(IDeviceCommBackend): """ Recorded backend. Connection is automatically opened on creation. Args: conn: connection parameters (recorded log path) datatype (str): Type of the returned data; can be ``"bytes"`` (return `bytes` object), ``"str"`` (return `str` object), or ``"auto"`` (default Python result: `str` in Python 2 and `bytes` in Python 3) reraise_error: if not ``None``, specifies an error to be re-raised on any backend exception (by default, use backend-specific error); should be a subclass of :exc:`DeviceBackendError`. """ BackendError=IOError _backend="recorded" Error=DeviceRecordedError _conn_params=["path"] _default_conn=[None] def __init__(self, conn, datatype="auto", reraise_error=None): conn_dict=self.combine_conn(conn,self._default_conn) IDeviceCommBackend.__init__(self,conn_dict.copy(),datatype=datatype,reraise_error=reraise_error) self.log=None self.log_section=None self.log_pos=0 try: self.open() except IOError as e: raise self.Error(e) from e
[docs] @reraise def open(self): """Open the connection""" if self.log is None: self.log=dict(backend_logger.load_logfile(self.conn["path"]))
[docs] def close(self): """Close the connection""" self.log=None
[docs] def is_opened(self): return self.log is not None
[docs] def start(self, header): """Start recorded section""" if header not in self.log: raise self.Error(IOError("header {} is missing from the record".format(header))) self.log_section=header self.log_pos=0
[docs] def stop(self): """Stop logging section""" self.log_section=None self.log_pos=0
[docs] @contextlib.contextmanager def section(self, header): self.start(header) try: yield finally: self.stop()
def _get_value(self, operation): if self.log is None: raise self.Error(IOError("device is not opened")) if self.log_section is None: raise self.Error(IOError("log section is not selected")) section=self.log[self.log_section] if len(section)<=self.log_pos: raise self.Error(IOError("section is over")) op,val=section[self.log_pos] self.log_pos+=1 if operation[0]!=op: raise self.Error(IOError("requested operation {}, recorded {}".format(operation,op))) return val
[docs] def readline(self, remove_term=True, timeout=None, skip_empty=True): """ Read a single line from the device. Args: remove_term (bool): If ``True``, remove terminal characters from the result. timeout: Operation timeout. If ``None``, use the default device timeout. skip_empty (bool): If ``True``, ignore empty lines (works only for ``remove_term==True``). """ result=self._get_value("read") return self._to_datatype(result)
[docs] def read(self, size=None): """ Read data from the device. If `size` is not None, read `size` bytes (usual timeout applies); otherwise, read all available data (return immediately). """ result=self._get_value("read") return self._to_datatype(result)
[docs] def write(self, data, flush=True, read_echo=False, read_echo_delay=0, read_echo_lines=1): """ Write data to the device. If ``flush==True``, flush the write buffer. If ``read_echo==True``, wait for `read_echo_delay` seconds and then perform :func:`readline` (`read_echo_lines` times). """ value=self._get_value("write") if value!=data: raise self.Error(IOError("requested write {}, recorded {}".format(data,value))) return value
def __repr__(self): return "RecordedDeviceBackend("+self.conn["path"]+")"
_backends["recorded"]=RecordedDeviceBackend _serial_re=re.compile(r"^com\d+",re.IGNORECASE) def _is_serial_addr(addr): return isinstance(addr,py3.anystring) and bool(_serial_re.match(addr)) _network_re=re.compile(r"(\d+\.){3}\d+(:\d+)?",re.IGNORECASE) def _is_network_addr(addr): return isinstance(addr,py3.anystring) and bool(_network_re.match(addr)) _visa_re=re.compile(r"\w+(::\w+)+",re.IGNORECASE) def _is_visa_addr(addr): return isinstance(addr,py3.anystring) and bool(_visa_re.match(addr))
[docs]def autodetect_backend(conn, default="visa"): """ Try to determine the backend by the connection. `default` specifies the default backend which is returned if the backend is unclear. """ if isinstance(conn, (tuple,list)): if len(conn)>=2 and isinstance(conn[0],int) and (0<=conn[0]<65536) and isinstance(conn[1],int) and (0<=conn[1]<65536): # PID / VID return "pyusb" conn=conn[0] elif isinstance(conn, dict): if "addr" in conn and _is_network_addr(conn["addr"]): return "network" if "port" in conn and _is_serial_addr(conn["port"]): return "serial" if "vendorID" in conn and "productID" in conn: return "pyusb" return default if _is_network_addr(conn): return "network" if _is_serial_addr(conn): return "serial" if _is_visa_addr(conn): return "visa" return default
def _as_backend(backend, conn=None): if backend=="auto": backend=autodetect_backend(conn) if isinstance(backend,tuple) and len(backend)==2 and backend[0]=="auto": backend=autodetect_backend(conn,default=backend[1]) if isinstance(backend,type) and issubclass(backend,IDeviceCommBackend): return backend if backend in _backends: return _backends[backend] error_text=_backend_errors.get(backend,None) raise ValueError("could not find backend {}".format(backend)+(": "+error_text if error_text else ""))
[docs]def new_backend(conn, backend="auto", defaults=None, **kwargs): """ Build new backend with the supplied parameters. Args: conn: Connection parameters (depend on the backend). Can be simply connection parameters (tuple or dict) for the given backend (e.g., ``"192.168.0.1"`` or ``("COM1",19200)``), a tuple ``(backend, conn)`` which specifies both backend and connection (in which case it overrides the supplied backend), or an already opened backend (in which case it is returned as is) backend (str): Backend type. Available backends are ``'auto'`` (try to autodetect based on the connection), ``'visa'``, ``'serial'``, ``'ft232'``, ``'network'``, and ``"pyusb"``. Can also be directly a backend class (more appropriate for custom backends), or a tuple ``('auto', backend)``, which is analogous to ``'auto'``, but it returns the specified ``backend`` if the autodetection fails; by default, the fallback backend is ``'visa'``, so ``'auto'`` is exactly the same as ``('auto', 'visa')``. defaults: if not ``None``, specifies a dictionary ``{backend: params}`` with default connection parameters (depending on the backend), which are added to the connection parameters **kwargs: parameters sent to the backend. """ if isinstance(conn,IDeviceCommBackend): return conn if isinstance(conn,tuple) and conn and (conn[0] in _backends or (isinstance(conn[0],type) and issubclass(conn[0],IDeviceCommBackend))): return new_backend(conn[1],backend=conn[0],**kwargs) backend=_as_backend(backend,conn) backend_name=getattr(backend,"_backend",None) if defaults is not None and backend_name is not None and backend_name in defaults: conn=backend.combine_conn(conn,defaults[backend_name]) return backend(conn,**kwargs)
[docs]def backend_error(backend, conn=None): """ Return error class corresponding to the current backend. Like :func:`new_backend`, allows setting ``backend="auto"``, in which case `conn` is used to try and autodetect the backend kind (not completely reliable, should be avoided). """ return _as_backend(backend,conn).Error
[docs]def list_backend_resources(backend=None, desc=False): """ List all resources for the given backend. If `backend` is ``None``, return dictionary ``{backend: resources}`` for all available backends. If ``desc==False``, return list of connections (usually strings or tuples), which can be used to connect to the device. Otherwise, return a list of descriptions, which have more info, but can be backend-dependent. """ if backend is None: res={n:b.list_resources(desc=desc) for (n,b) in _backends.items()} return {n:r for n,r in res.items() if r is not None} else: return _as_backend(backend).list_resources(desc=desc)
### Interface for a generic device class employing a communication backend ###
[docs]class ICommBackendWrapper(interface.IDevice): """ A base class for an instrument using a communication backend. Args: instr: Backend (assumed to be already opened). """ def __init__(self, instr): super().__init__() self.instr=instr self._connection_parameters=self.instr.conn
[docs] def open(self): """Open the backend""" return self.instr.open()
[docs] def close(self): """Close the backend""" return self.instr.close()
[docs] def is_opened(self): """Check if the device is connected""" return bool(self.instr)
def _get_connection_parameters(self): return self._connection_parameters
[docs] def lock(self, timeout=None): """Lock the access to the device from other threads/processes (isn't necessarily implemented)""" return self.instr.lock(timeout=timeout)
[docs] def unlock(self): """Unlock the access to the device from other threads/processes (isn't necessarily implemented)""" return self.instr.unlock()
[docs] def locking(self, timeout=None): """Context manager for lock & unlock""" return self.instr.locking(timeout=timeout)