python3-pyvisa 1.8-2 / usr / lib / python3 / dist-packages / pyvisa / rname.py

# -*- coding: utf-8 -*-
"""
    pyvisa.rname
    ~~~~~~~~~~~~

    Functions and classes to parse and assemble resource name.

    :copyright: 2014 by PyVISA Authors, see AUTHORS for more details.
    :license: MIT, see LICENSE for more details.
"""

from __future__ import division, unicode_literals, print_function, absolute_import

import re
import contextlib
from collections import namedtuple, defaultdict

from . import constants, errors, logger

# :type: set[str]
_INTERFACE_TYPES = set()

# Resource Class for Interface type
# :type: dict[str, set[str]]
_RESOURCE_CLASSES = defaultdict(set)

# :type: dict[(str, str), ResourceName]
_SUBCLASSES = {}

# DEFAULT Resource Class for a given interface type.
# :type: dict[str, str]
_DEFAULT_RC = {}


class InvalidResourceName(ValueError):
    """Exception raised when the resource name cannot be parsed.
    """

    def __init__(self, msg):
        self.msg = msg

    @classmethod
    def bad_syntax(cls, syntax, resource_name, ex=None):
        """Exception used when the resource name cannot be parsed.
        """

        if ex:
            msg = "The syntax is '%s' (%s)." % (syntax, ex)
        else:
            msg = "The syntax is '%s'." % syntax

        msg = "Could not parse '%s'. %s" % (resource_name, msg)

        return cls(msg)

    @classmethod
    def subclass_notfound(cls, interface_type_resource_class,
                          resource_name=None):
        """Exception used when the subclass for a given interface type / resource class pair
        cannot be found.
        """

        msg = "Parser for not found (%s)." % interface_type_resource_class

        if resource_name:
            msg = "Could not parse '%s'. %s" % (resource_name, msg)

        return cls(msg)

    @classmethod
    def rc_notfound(cls, interface_type, resource_name=None):
        """Exception used when no resource class is provided and no default is found.
        """

        msg = "Resource class for %s not provided and default not found." % interface_type

        if resource_name:
            msg = "Could not parse '%s'. %s" % (resource_name, msg)

        return cls(msg)

    def __str__(self):
        return self.msg


def register_subclass(cls):
    """Register a subclass for a given interface type and resource class.
    """

    key = cls.interface_type, cls.resource_class

    if key in _SUBCLASSES:
        raise ValueError('Class already registered for %s and %s' % key)

    _SUBCLASSES[(cls.interface_type, cls.resource_class)] = cls

    _INTERFACE_TYPES.add(cls.interface_type)
    _RESOURCE_CLASSES[cls.interface_type].add(cls.resource_class)

    if cls.is_rc_optional:
        if cls.interface_type in _DEFAULT_RC:
            raise ValueError('Default already specified for %s' %
                             cls.interface_type)
        _DEFAULT_RC[cls.interface_type] = cls.resource_class

    return cls


class ResourceName(object):
    """Base class for ResourceNames to be used as a Mixin
    """

    # Interface type string
    interface_type = ''

    # Resource class string
    resource_class = ''

    # Specifices if the resource class part of the string is optional.
    is_rc_optional = False

    # Formatting string for canonical
    _canonical_fmt = ''

    # VISA syntax for resource
    _visa_syntax = ''

    # Resource name provided by the user (not empty only when parsing)
    user = ''

    @property
    def interface_type_const(self):
        try:
            return getattr(constants.InterfaceType, self.interface_type.lower())
        except:
            return constants.InterfaceType.unknown

    @classmethod
    def from_string(cls, resource_name):
        """Parse a resource name and return a ResourceName

        :type resource_name: str
        :rtype: ResourceName

        :raises InvalidResourceName: if the resource name is invalid.
        """
        # TODO Remote VISA

        uname = resource_name.upper()

        for interface_type in _INTERFACE_TYPES:

            # Loop through all known interface types until we found one
            # that matches the beginning of the resource name
            if not uname.startswith(interface_type):
                continue

            if len(resource_name) == len(interface_type):
                parts = ()
            else:
                parts = resource_name[len(interface_type):].split('::')

            # Try to match the last part of the resource name to
            # one of the known resource classes for the given interface type.
            # If not possible, use the default resource class
            # for the given interface type.
            if parts and parts[-1] in _RESOURCE_CLASSES[interface_type]:
                parts, resource_class = parts[:-1], parts[-1]
            else:
                try:
                    resource_class = _DEFAULT_RC[interface_type]
                except KeyError:
                    raise InvalidResourceName.rc_notfound(interface_type,
                                                          resource_name)

            # Look for the subclass
            try:
                subclass = _SUBCLASSES[(interface_type, resource_class)]
            except KeyError:
                raise InvalidResourceName.subclass_notfound(
                    (interface_type, resource_class), resource_name)

            # And create the object
            try:
                rn = subclass.from_parts(*parts)
                rn.user = resource_name
                return rn
            except ValueError as ex:
                raise InvalidResourceName.bad_syntax(subclass._visa_syntax,
                                                     resource_name, ex)

        raise InvalidResourceName('Could not parse %s: unknown interface type'
                                  % resource_name)

    @classmethod
    def from_kwargs(cls, **kwargs):
        interface_type = kwargs.pop('interface_type')

        try:
            resource_class = kwargs.pop('resource_class',
                                        _DEFAULT_RC[interface_type])
        except KeyError:
            raise InvalidResourceName.rc_notfound(interface_type)

        # Look for the subclass
        try:
            subclass = _SUBCLASSES[(interface_type, resource_class)]
        except KeyError:
            raise InvalidResourceName.subclass_notfound(
                (interface_type, resource_class))

        # And create the object
        try:
            return subclass(**kwargs)
        except ValueError as ex:
            raise InvalidResourceName(str(ex))

    def __str__(self):
        return self._canonical_fmt.format(self)


def build_rn_class(interface_type, resource_parts, resource_class,
                   is_rc_optional=True):
    """Builds a resource name class by mixing a named tuple and ResourceName.

    It also registers the class.

    The field names are changed to lower case and the spaces replaced
    by underscores ('_').

    :param interface_type: the interface type
    :type: interface_type: str
    :param resource_parts: each of the parts of the resource name indicating
                           name and default value.
                           Use None for mandatory fields.
    :type resource_parts: tuple[(str, str)]
    :param resource_class: the resource class
    :type resource_class: str
    :param is_rc_optional: indicates if the resource class part is optional
    :type is_rc_optional: boolean.
    """

    interface_type = interface_type.upper()
    resource_class = resource_class.upper()

    syntax = interface_type
    fmt = interface_type
    fields = []

    # Contains the resource parts but using python friendly names
    # (all lower case and replacing spaces by underscores)
    p_resource_parts = []

    kwdoc = []

    # Assemble the syntax and format string based on the resource parts
    for ndx, (name, default_value) in enumerate(resource_parts):
        pname = name.lower().replace(' ', '_')
        fields.append(pname)
        p_resource_parts.append((pname, default_value))

        sep = '::' if ndx else ''

        fmt += sep + '{0.%s}' % pname

        if default_value is None:
            syntax += sep + name
        else:
            syntax += '[' + sep + name + ']'

        kwdoc.append('- %s (%s)' % (pname, 'required' if default_value is None
                                    else default_value))

    fmt += '::' + resource_class

    if not is_rc_optional:
        syntax += '::' + resource_class
    else:
        syntax += '[' + '::' + resource_class + ']'

    class _C(namedtuple('Internal', ' '.join(fields)), ResourceName):
        """%s %s"

        Can be created with the following keyword only arguments:
            %s

        Format :
            %s
        """ % (resource_class, interface_type, '    \n'.join(kwdoc), syntax)

        def __new__(cls, **kwargs):
            new_kwargs = dict(p_resource_parts, **kwargs)

            for key, value in new_kwargs.items():
                if value is None:
                    raise ValueError(key + ' is a required parameter')

            return super(_C, cls).__new__(cls, **new_kwargs)

        @classmethod
        def from_parts(cls, *parts):

            if len(parts) < sum(1 for _, v in p_resource_parts
                                if v is not None):
                raise ValueError('not enough parts')
            elif len(parts) > len(p_resource_parts):
                raise ValueError('too many parts')

            (k, default), rp = p_resource_parts[0], p_resource_parts[1:]

            # The first part (just after the interface_type) is the only
            # optional part which can be and empty and therefore the
            # default value should be used.
            p, pending = parts[0], parts[1:]
            kwargs = {k: default if p == '' else p}

            # The rest of the parts are consumed when mandatory elements are required.
            while len(pending) < len(rp):
                (k, default), rp = rp[0], rp[1:]
                if default is None:
                    if not parts:
                        raise ValueError(k + ' part is mandatory')
                    p, pending = pending[0], pending[1:]
                    if not p:
                        raise ValueError(k + ' part is mandatory')
                    kwargs[k] = p
                else:
                    kwargs[k] = default

            # When the length of the pending provided and resource parts
            # are equal, we just consume everything.
            kwargs.update((k, p) for (k, v), p in zip(rp, pending))

            return cls(**kwargs)

    _C.interface_type = interface_type
    _C.resource_class = resource_class
    _C.is_rc_optional = is_rc_optional
    _C._canonical_fmt = fmt
    _C._visa_syntax = syntax

    _C.__name__ = str(interface_type + resource_class.title())

    return register_subclass(_C)

# Build subclasses for each resource

GPIBInstr = build_rn_class('GPIB',
                           (('board', '0'), ('primary address', None),
                            ('secondary address', constants.VI_NO_SEC_ADDR)),
                           'INSTR')

GPIBIntfc = build_rn_class('GPIB', (('board', '0'), ), 'INTFC', False)

ASRLInstr = build_rn_class('ASRL', (('board', '0'), ), 'INSTR')

TCPIPInstr = build_rn_class('TCPIP', (('board', '0'), ('host address', None),
                                      ('LAN device name', 'inst0'), ), 'INSTR')

TCPIPSocket = build_rn_class('TCPIP', (('board', '0'), ('host address', None),
                                       ('port', None), ), 'SOCKET', False)

USBInstr = build_rn_class('USB',
                          (('board', '0'), ('manufacturer ID', None),
                           ('model code', None), ('serial number', None),
                           ('USB interface number', '0')), 'INSTR')

USBRaw = build_rn_class('USB', (('board', '0'), ('manufacturer ID', None),
                                ('model code', None), ('serial number', None),
                                ('USB interface number', '0')), 'RAW', False)

PXIBackplane = build_rn_class('PXI', (('interface', '0'),
                                      ('chassis number', None)), 'BACKPLANE',
                              False)

PXIMemacc = build_rn_class('PXI', (('interface', '0'), ), 'MEMACC', False)

VXIBackplane = build_rn_class('VXI', (('board', '0'),
                                      ('VXI logical address', '0')),
                              'BACKPLANE', False)

VXIInstr = build_rn_class('VXI', (('board', '0'),
                                  ('VXI logical address', None)), 'INSTR',
                          True)

VXIMemacc = build_rn_class('VXI', (('board', '0'), ), 'MEMACC', False)

VXIServant = build_rn_class('VXI', (('board', '0'), ), 'SERVANT', False)

# TODO 3 types of PXI INSTR
# TODO ENET-Serial INSTR
# TODO Remote NI-VISA


def assemble_canonical_name(**kwargs):
    """Given a set of keyword arguments defining a resource name,
    return the canonical resource name.
    """
    return str(ResourceName.from_kwargs(**kwargs))


def to_canonical_name(resource_name):
    """Parse a resource name and return the canonical version.

    :type resource_name: str
    :rtype: str
    """
    return str(ResourceName.from_string(resource_name))


parse_resource_name = ResourceName.from_string


def filter(resources, query):
    """Filter a list of resources according to a query expression.

    The search criteria specified in the query parameter has two parts:
      1. a VISA regular expression over a resource string.
      2. optional logical expression over attribute values
         (not implemented in this function, see below).

    .. note: The VISA regular expression syntax is not the same as the
             Python regular expression syntax. (see below)

    The regular expression is matched against the resource strings of resources
    known to the VISA Resource Manager. If the resource string matches the
    regular expression, the attribute values of the resource are then matched
    against the expression over attribute values. If the match is successful,
    the resource has met the search criteria and gets added to the list of
    resources found.

    By using the optional attribute expression, you can construct flexible
    and powerful expressions with the use of logical ANDs (&&), ORs(||),
    and NOTs (!). You can use equal (==) and unequal (!=) comparators to
    compare attributes of any type, and other inequality comparators
    (>, <, >=, <=) to compare attributes of numeric type. Use only global
    attributes in the attribute expression. Local attributes are not allowed
    in the logical expression part of the expr parameter.


        Symbol      Meaning
        ----------  ----------

        ?           Matches any one character.

        \           Makes the character that follows it an ordinary character
                    instead of special character. For example, when a question
                    mark follows a backslash (\?), it matches the ? character
                    instead of any one character.

        [list]      Matches any one character from the enclosed list. You can
                    use a hyphen to match a range of characters.

        [^list]     Matches any character not in the enclosed list. You can use
                    a hyphen to match a range of characters.

        *           Matches 0 or more occurrences of the preceding character or
                    expression.

        +           Matches 1 or more occurrences of the preceding character or
                    expression.

        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))


def filter2(resources, query, open_resource):
    """Filter a list of resources according to a query expression.

    It accepts the optional part of the expression.

    .. warning: This function is experimental and unsafe as it uses eval,
                It also might require to open the resource.

    :param resources: iterable of resources.
    :param query: query expression.
    :param open_resource: function to open the resource.
    """

    if '{' in query:
        try:
            query, optional = query.split('{')
            optional, _ = optional.split('}')
        except ValueError:
            raise errors.VisaIOError(constants.VI_ERROR_INV_EXPR)
    else:
        optional = None

    filtered = filter(resources, query)

    if not optional:
        return filtered

    optional = optional.replace('&&', 'and').replace('||', 'or').replace('!', 'not ')
    optional = optional.replace('VI_', 'res.VI_')

    class AttrGetter():

        def __init__(self, resource_name):
            self.resource_name = resource_name
            self.parsed = parse_resource_name(resource_name)
            self.resource = None

        def __getattr__(self, item):
            if item == 'VI_ATTR_INTF_NUM':
                   return int(self.parsed.board)
            elif item == 'VI_ATTR_MANF_ID':
                   return self.parsed.manufacturer_id
            elif item == 'VI_ATTR_MODEL_CODE':
                   return self.parsed.model_code
            elif item == 'VI_ATTR_USB_SERIAL_NUM':
                   return self.parsed.serial_number
            elif item == 'VI_ATTR_USB_INTFC_NUM':
                   return int(self.parsed.board)
            elif item == 'VI_ATTR_TCPIP_ADDR':
                   return self.parsed.host_address
            elif item == 'VI_ATTR_TCPIP_DEVICE_NAME':
                   return self.parsed.lan_device_name
            elif item == 'VI_ATTR_TCPIP_PORT':
                   return int(self.parsed.port)
            elif item == 'VI_ATTR_INTF_NUM':
                   return int(self.parsed.board)
            elif item == 'VI_ATTR_GPIB_PRIMARY_ADDR':
                   return int(self.parsed.primary_address)
            elif item == 'VI_ATTR_GPIB_SECONDARY_ADDR':
                   return int(self.parsed.secondary_address)
            elif item == 'VI_ATTR_PXI_CHASSIS':
                return self.parsed.chassis_number
            elif item == 'VI_ATTR_MAINFRAME_LA':
                return self.parsed.vxi_logical_address

            if self.resource is None:
                self.resource = open_resource(self.resource_name)

            return self.resource.get_visa_attribute(item)

    @contextlib.contextmanager
    def open_close(resource_name):
        getter = AttrGetter(resource_name)
        yield getter
        if getter.resource is not None:
            getter.resource.close()

    selected = []
    for rn in filtered:
        with open_close(rn) as getter:
            if eval(optional, None, dict(res=getter)):
                selected.append(rn)