Source code for naginterfaces.base.utils

"""
Base utilities for the `naginterfaces` package.
"""

# NAG Copyright 2017-2021.

# pylint: disable=protected-access,too-many-lines

from __future__ import absolute_import

import array as _array
import collections as _collections
import configparser as _configparser
import ctypes as _ctypes
import functools as _functools
import inspect as _inspect
import io as _io
import logging as _logging
import os as _os
import socket as _socket
import struct as _struct
import subprocess as _subprocess
import sys as _sys
import warnings as _warnings

import numpy as _np

from ._nag_engine_data_consts import (
    NAG_ED_COL_MAJOR as _NAG_ED_COL_MAJOR,
    NAG_ED_ROW_MAJOR as _NAG_ED_ROW_MAJOR,
    NAG_ED_YES as _NAG_ED_YES,
)
from ._nag_p_consts import ( # pylint: disable=unused-import
    IERR_ALLOC_FAIL as _IERR_ALLOC_FAIL,
    IERR_HELPER as _IERR_HELPER,
    IERR_NO_LIC as _IERR_NO_LIC,
    IERR_SUCCESS as _IERR_SUCCESS,
    IERR_UNEXPECTED as _IERR_UNEXPECTED,
)

from .. import _THE_SYSTEM, ENGINE_LIBDIR as _ENGINE_LIBDIR, __version__

_PKG_NAME = 'naginterfaces'
_MODULE_NAME = '.'.join([_PKG_NAME, 'base.utils'])

def _join_sequence(
        sequence,
        separator=', ',
        last_separator=' or ',
):
    """
    Return a string joining the input sequence using separator,
    with last_separator between the final pair of elements.
    """
    if len(sequence) <= 1:
        return separator.join(sequence)

    if len(sequence) == 2:
        return last_separator.join(sequence)

    return last_separator.join([
        separator.join(sequence[:-1]),
        sequence[-1]
    ])

def _bad_instance_msg(
        exp_insts,
        allow_none,
):
    """
    Return a string for when a data-type payload was not any instance
    from the exp_insts list.
    """
    return ' '.join([
        'must be ' + ('' if not allow_none else 'None or ')  +
        'an instance of',
        _join_sequence(exp_insts)
    ])

def _bad_rank_msg(
        exp_rank,
        actual_rank,
):
    """
    Return a string for when a data-type payload was not of the expected
    rank.
    """
    return ' '.join([
        'has rank',
        str(actual_rank) + ',',
        'but must have rank',
        str(exp_rank),
    ])

def _bad_shape_msg(
        exp_shape,
        actual_shape,
):
    """
    Return a string for when a data-type payload was not of the expected
    shape
    """
    if len(exp_shape) == 1:

        if len(actual_shape) == 1:
            return ' '.join([
                'has length',
                str(actual_shape[0]) + ',',
                'but must have length',
                str(exp_shape[0])
            ])

        return ' '.join([
            'has shape',
            str(actual_shape) + ',',
            'but must be rank 1 with length',
            str(exp_shape[0])
        ])

    str_exp_shape = '(' + ', '.join([
        ':' if extent is None else str(extent) for extent in exp_shape
    ]) + ')'
    return ' '.join([
        'has shape',
        str(actual_shape) + ',',
        'but must have shape',
        str_exp_shape,
    ])

[docs]class NagException(Exception): """ Base class for errors from this package. Attributes ---------- errno : int or None When a call to the underlying NAG Library Engine flags an error or warning, this attribute is set to an integer identifier for the exit case. The value may be used to cross reference the failure in the corresponding NAG Library document for the routine in question. The ``__str__`` representation of an exception having non-``None`` ``errno`` will display a substring of the form ``code x[:y[,z]]`` in which the ``x`` integer is the same as the ``errno`` value (and where the ``y`` and ``z`` values, which may be absent, only have internal significance to NAG). return_data : collections.namedtuple (nag_return_data) or None When a call to the underlying NAG Engine flags a non-fatal warning and you have configured the Python ``warnings`` module to escalate NAG warnings to exceptions, this attribute is populated with the return arguments for the routine in question (in argument order). Thus after catching the warning you may access the routine's return data via this attribute if desired. The tuple subclass - i.e., the type of this attribute - is named ``nag_return_data``. Notes ----- Concrete subclasses ``NagLicenceError`` and ``NagMemoryError`` are potential exceptions that may be raised by any function in the `naginterfaces` package. """ _contact_nag = True _contact_nag_msg = 'Please contact NAG.' def __init__( self, locus, msg, ): """ Store the exception locus dict and exception message. """ Exception.__init__(self, locus, msg) self.locus = locus self.msg = msg self._errcodes = ( self.locus['errcodes'] if 'errcodes' in self.locus else None ) self.errno = self._errcodes[0] if self._errcodes is not None else None self.return_data = self.locus.get('return_data') def __str__( self, ): err_str = [] l_t_s = self._locus_to_string() if l_t_s: err_str.append(l_t_s) if self.msg and self.msg[0] != '\n': err_str.append(' ') err_str.append(self.msg) if self._contact_nag: if self.msg[-1] != '.': err_str[-1] += '.' err_str.extend([ '\n', '** ' + self._contact_nag_msg ]) return ''.join(err_str) def _locus_to_string( self ): """ Return a string formatting the current locus for printing. """ msg = [] if 'fun_name' in self.locus: msg.append('NAG Python function ' + self.locus['fun_name']) if 'returned' in self.locus: msg.append('returned ' + self.locus['returned']) elif 'entity' in self.locus: msg.append('argument ' + self.locus['entity']) if 'element' in self.locus: msg.append('element ' + str(self.locus['element'])) if 'errcodes' in self.locus: codes_str = 'code ' + str(self.errno) if len(self.locus['errcodes']) > 1: codes_str += ':' + ','.join( [str(l) for l in self.locus['errcodes'][1:]] ) msg.append(codes_str) if msg: return '(' + ', '.join(msg) + ')' return ''
[docs]class NagWarning(NagException, Warning): # pylint: disable=line-too-long """ Base class for warnings from NAG wrappers. As a subclass of the Python ``Warning`` class the behaviour of a ``NagWarning`` can be controlled using the methods in the core ``warnings`` module. Examples -------- To escalate occurrences of ``NagAlgorithmicWarning`` to exceptions and to never show occurrences of ``NagWithdrawalWarning`` (which would not be recommended in practice): >>> import warnings >>> from naginterfaces.base import utils >>> warnings.simplefilter('error', utils.NagAlgorithmicWarning) >>> warnings.simplefilter('ignore', utils.NagWithdrawalWarning) Any promoted ``NagAlgorithmicWarning`` exits can then be caught and their ``errno`` and ``return_data`` attributes accessed as required: >>> try: # doctest: +SKIP ... some_nag_call(...) ... except NagAlgorithmicWarning as exc: ... if exc.errno == xyz: ... look at exc.return_data See Also -------- :func:`naginterfaces.library.examples.glopt.nlp_multistart_sqp_lsq_ex.main` : This example catches a ``NagAlgorithmicWarning`` and retrieves some of its ``return_data``. :func:`naginterfaces.library.examples.roots.sys_func_rcomm_ex.main` : This example promotes instances of ``NagAlgorithmicWarning`` to exceptions. """ # pylint: enable=line-too-long _contact_nag = False
[docs]class NagAlgorithmicWarning(NagWarning): """ Concrete class for warnings from NAG wrappers that were identified by the NAG Library Engine. """ _contact_nag = False
[docs]class NagAlgorithmicMajorWarning(NagAlgorithmicWarning): """ Concrete class for major warnings from NAG wrappers that were identified by the NAG Library Engine. """ _contact_nag = False
[docs]class NagAttributeError(NagException, AttributeError): """ Concrete class for attribute errors from NAG wrappers. """ _contact_nag = False
[docs]class NagCallbackTerminateWarning(NagAlgorithmicWarning): """ Concrete class for warnings when early termination from a callback was triggered by the user. Attributes ---------- cb_excs : list of UserCallbackTerminate The original exception(s) raised by the user. """ _contact_nag = False def __init__( self, locus, msg, ): """ Extract the original exception from the locus. """ NagAlgorithmicWarning.__init__(self, locus, msg) self.cb_excs = self.locus['cb_excs']
[docs]class NagDeprecatedWarning(NagWarning): """ Concrete class for warnings about NAG wrappers that are deprecated. This category is not suppressed by default. """ _contact_nag = False
[docs]class NagIndexError(NagException, IndexError): """ Concrete class for index errors from NAG wrappers. """ _contact_nag = False
[docs]class NagKeyError(NagException, KeyError): """ Concrete class for key errors from NAG wrappers. """ _contact_nag = False
[docs]class NagLicenceError(NagException): """ Concrete class for licensing errors from NAG wrappers. """
[docs]class NagMemoryError(NagException, MemoryError): """ Concrete class for allocation errors from NAG wrappers. """ _contact_nag = False
[docs]class NagNotImplementedError(NagException, NotImplementedError): """ Concrete class for not-implemented errors from NAG wrappers. """ _contact_nag = True
[docs]class NagOSError(NagException, OSError): """ Concrete class for OS errors from NAG wrappers. """ _contact_nag = False
[docs]class NagOverflowError(NagException, OverflowError): """ Concrete class for overflow errors from NAG wrappers. """ _contact_nag = False
[docs]class NagTypeError(NagException, TypeError): """ Concrete class for type errors from NAG wrappers. """ _contact_nag = False
[docs]class NagValueError(NagException, ValueError): """ Concrete class for value errors from NAG wrappers. """ _contact_nag = False
[docs]class NagShapeError(NagValueError): """ Concrete class for shape errors from NAG wrappers. """
[docs]class NagWithdrawalWarning(NagDeprecatedWarning): """ Concrete class for warnings about NAG wrappers that are scheduled for withdrawal. This category is not suppressed by default. """
[docs]class UserCallbackTerminate(Exception): """ Concrete class for user-raised callback termination requests. """
_IS_IL32 = (_ctypes.c_int is _ctypes.c_long) # ctypes integers for this platform: _SYS_CTYPES_INTS = { 32: (_ctypes.c_long if _IS_IL32 else _ctypes.c_int), 64: (_ctypes.c_longlong if _IS_IL32 else _ctypes.c_long) } def _get_sys_data_model(): """ Return a string for the data model used by the Python interpreter on this platform. """ if _struct.calcsize('P') == 4: # pragma: no cover return 'ILP32' if _THE_SYSTEM == 'Windows': return 'IL32P64' return 'I32LP64' class _EngineData(_ctypes.Structure): """ State structure for the NAG Engine. """ # pylint: disable=too-few-public-methods _fields_ = [ ('cpuser', _ctypes.c_void_p), ('hlperr', _SYS_CTYPES_INTS[32]), ('ad_handle', _ctypes.c_void_p), ('storage_order', _SYS_CTYPES_INTS[32]), ('reorder_packed', _SYS_CTYPES_INTS[32]), ('allocate_workspace', _SYS_CTYPES_INTS[32]), ('wrapptr1', _ctypes.c_void_p), ('wrapptr2', _ctypes.c_void_p), ('y90_print_rec', _ctypes.c_void_p), ('num_threads', _SYS_CTYPES_INTS[32]), ('thread_num', _SYS_CTYPES_INTS[32]), ('hlperr_thread_num', _SYS_CTYPES_INTS[32]), ] default_sorder = 'C' default_spiked_sorder = 'F' def __init__( self, sorder=default_sorder, py_c_print_rec=None, ): """ Return an initialized _EngineData structure. """ _ctypes.Structure.__init__(self) y90haan_ctypes = _EngineState._get_symbol('y90haan') y90haan_ctypes.restype = None y90haan_ctypes.argtypes = ( _BYREF_ENGINEDATA, ) y90haan_ctypes(self) self._set_sorder(sorder) self.allocate_workspace = _NAG_ED_YES self._set_print_rec(py_c_print_rec) def _set_print_rec( self, py_c_print_rec, ): """ Update the y90_print_rec field. """ if ( py_c_print_rec is None or _EngineState._debug_level == _EngineState._debug_off ): return c_p_r = _C_PRINT_REC_FUNCTYPE(py_c_print_rec) self.y90_print_rec = _ctypes.cast(c_p_r, _ctypes.c_void_p) def _set_sorder( self, sorder, ): """ Update the storage-order field. """ if sorder == 'F': self.storage_order = _NAG_ED_COL_MAJOR else: self.storage_order = _NAG_ED_ROW_MAJOR _BYREF_ENGINEDATA = _ctypes.POINTER(_EngineData) def _cast_callback( proto, callback, locus, ): """ Return a ctypes FUNCTYPE for the input callback. """ if callback is None: return _ctypes.cast(callback, proto) try: return proto(callback) except TypeError: pass raise NagException( # pragma: no cover locus, 'unexpected failure casting to C function' ) def _get_symbol_from_lib_handle( libh, symbol_name, ): """ Utility for loading symbol_name from library handle libh. """ return getattr(libh, _mangle_sym_name(symbol_name)) def _mangle_sym_name( symbol_name, ): """ Mangle symbol_name into a valid shared-library symbol name on this platform. """ if ( ( _THE_SYSTEM == 'Windows' or _THE_SYSTEM.startswith('CYGWIN') ) and _ENGINE_DATA_MODEL != 'ILP32' ): return symbol_name.strip('_').upper() return symbol_name.strip('_').lower() + '_' _SPEC_CFG_FILE = _os.path.join( _os.path.abspath(_ENGINE_LIBDIR), 'lib_spec.cfg', ) if not _os.path.isfile(_SPEC_CFG_FILE): raise NagException( {}, 'missing ' + _SPEC_CFG_FILE, ) _CF = _configparser.ConfigParser() _CF.read(_SPEC_CFG_FILE) _ENGINE_COMPILERS_DICT = {} for _n, _v in _CF.items('compilers'): if _n in ['fortran', 'c', 'c++']: _dict_n = _n[0].upper() + _n[1:] else: # pragma: no cover _dict_n = _n _ENGINE_COMPILERS_DICT[_dict_n] = _v _ENGINE_DATA_MODEL = _CF.get('datamodel', 'model') _ENGINE_INT_BITS = 64 if _ENGINE_DATA_MODEL == 'ILP64' else 32 _ENGINE_BUILD_ID_STR = ( _CF.get('buildid', 'idnum') if ( _CF.has_section('buildid') and _CF.has_option('buildid', 'idnum') ) else None ) _ENGINE_PCODE = ( _CF.get('productcode', 'pcode') if ( _CF.has_section('productcode') and _CF.has_option('productcode', 'pcode') ) else None ) _ENGINE_LIB_NAME_FROM_SPEC = ( _CF.get('meta', 'lib_name') if ( _CF.has_section('meta') and _CF.has_option('meta', 'lib_name') ) else None ) if _ENGINE_PCODE is None: # pragma: no cover _ENGINE_PCODE = _ENGINE_LIB_NAME_FROM_SPEC if _ENGINE_PCODE is None: # pragma: no cover raise NagException( {}, 'missing productcode and lib_name in ' + _SPEC_CFG_FILE, ) def _get_sys_shlib_ext(): """ Return a string for the file extension of shared libraries on this platform. """ if _THE_SYSTEM == 'Darwin': return '.dylib' if _THE_SYSTEM == 'Linux': return '.so' if _THE_SYSTEM == 'Windows' or _THE_SYSTEM.startswith('CYGWIN'): return '.dll' raise NagException( # pragma: no cover {'fun_name': '_get_sys_shlib_ext'}, 'inconceivable platform' ) def _is_ascii( data, locus, ): """ Python 3+ check for data being an ASCII string. """ if isinstance(data, str): try: data.encode('ascii') except UnicodeError: return False else: return True elif isinstance(data, bytes): try: data.decode('ascii') except UnicodeError: return False else: return True else: raise NagException( # pragma: no cover locus, 'inconceivable string class' ) class _EngineExit(object): """ Base class for categorization of error exits from the NAG Engine. """ # pylint: disable=too-few-public-methods __slots__ = ['fun_name', 'entity', 'msg_lines', 'errcodes', 'return_data'] def __init__( self, fun_name=None, entity=None, msg_lines=None, errcodes=None, return_data=None, ): # pylint: disable=too-many-arguments self.fun_name = fun_name self.entity = entity self.msg_lines = msg_lines self.errcodes = errcodes self.return_data = return_data def manage(self): """ Take the appropriate action for this exit category. """ raise NotImplementedError # pragma: no cover def _set_locus(self): """ Construct the exception locus. """ locus = {} if self.fun_name is not None: locus['fun_name'] = self.fun_name if self.errcodes is not None: locus['errcodes'] = self.errcodes if self.entity is not None: locus['entity'] = self.entity if self.return_data is not None: locus['return_data'] = self.return_data return locus class _EngineNoErrorExit(_EngineExit): """ Concrete class for successful exits from the NAG Engine. """ # pylint: disable=too-few-public-methods __slots__ = [] def manage(self): return class _EngineWarningExit(_EngineExit): """ Concrete class for warning exits from the NAG Engine. """ # pylint: disable=too-few-public-methods __slots__ = [] def manage(self): _raise_warning( NagAlgorithmicWarning, self.msg_lines, self._set_locus(), ) class _EngineMajorWarningExit(_EngineWarningExit): """ Concrete class for major warning exits from the NAG Engine. """ # pylint: disable=too-few-public-methods __slots__ = [] def manage(self): _raise_warning( NagAlgorithmicMajorWarning, self.msg_lines, self._set_locus(), ) class _EngineErrorExit(_EngineExit): """ Concrete class for routine-specific error exits from the NAG Engine. """ # pylint: disable=too-few-public-methods __slots__ = [] def manage(self): raise NagValueError( self._set_locus(), _format_msg_lines(self.msg_lines) ) class _EngineAllocationErrorExit(_EngineExit): """ Concrete class for memory-allocation error exits from the NAG Engine. """ # pylint: disable=too-few-public-methods __slots__ = [] def manage(self): locus = self._set_locus() locus['errcodes'] = (_IERR_ALLOC_FAIL,) raise NagMemoryError(locus, _format_msg_lines( [ 'Dynamic memory allocation failed.', ] )) class _UserCallbackExit(_EngineWarningExit): """ Concrete class for early termination exits from callbacks. """ # pylint: disable=too-few-public-methods __slots__ = ['cb_excs'] def __init__( self, fun_name=None, entity=None, msg_lines=None, errcodes=None, return_data=None, cb_excs=None, ): # pylint: disable=too-many-arguments _EngineWarningExit.__init__( self, fun_name, entity, msg_lines, errcodes, return_data, ) self.cb_excs = cb_excs def manage(self): locus = self._set_locus() locus['cb_excs'] = self.cb_excs _raise_warning( NagCallbackTerminateWarning, self.msg_lines, locus, ) def _resolve_kpath(k_util): """ Return the absolute path of `k_util`. """ k_util_name = k_util if _THE_SYSTEM == 'Windows' or _THE_SYSTEM.startswith('CYGWIN'): k_util_name += '.exe' k_util_path = _os.path.join( _os.path.abspath(_ENGINE_LIBDIR), k_util_name ) if not _os.path.isfile(k_util_path): raise NagException( # pragma: no cover {}, 'Missing utility ' + k_util_path ) return k_util_path def _capture_hostid(): """ Return the output of ``khostid``. """ return _EngineCharScalarType.to_py( _subprocess.check_output(_resolve_kpath('khostid')) ) def _capture_lcheck(pcode, quiet=False): """ Return the output of ``klcheck`` with some postprocessing. """ if not pcode.endswith('L'): return ( 'Licence available; this product is not licence managed' ) cmd_list = [_resolve_kpath('klcheck')] if quiet: cmd_list.append('-quiet') cmd_list.append(pcode) cmd_output = _subprocess.check_output(cmd_list) try: lcheck_decode = cmd_output.decode(encoding='utf-8') except UnicodeDecodeError: lcheck_decode = cmd_output.decode(encoding='latin-1') lcheck_txt = lcheck_decode.rstrip().splitlines() for lcheck_i, lcheck_line in enumerate(lcheck_txt): if lcheck_line.startswith('Licence available'): lcheck_txt[lcheck_i] = ( 'Licence available; the required ' + pcode + ' licence key ' 'for this product is valid' ) elif lcheck_line in [ 'Error: Licence file not found', 'Error: No licence found for this product' ]: lcheck_txt[lcheck_i] = ( 'Error: Licence not found; this product requires a key for ' + pcode ) return '\n'.join(lcheck_txt) def _lcheck_lines(pcode, quiet=False): """ Return a list of lines forming a licensing message. """ l_line = _capture_lcheck(pcode, quiet=quiet) if 'Licence available' in l_line: return [l_line] lc_lines = [ 'The NAG Library for Python on this platform uses', 'underlying Library ' + pcode + '.', 'This Library has been installed as part of the package', 'and it requires a valid licence key.', 'No such key could be validated:', 'the key may not have been installed correctly or', 'it may have expired.', 'The Kusari licence-check utility reports the following:', ] lc_lines.append(l_line) lc_lines.append( 'The above information has been generated on machine ' + _socket.gethostname() ) lc_lines.extend([ 'For information on how to obtain a licence, please see', 'https://www.nag.com/numeric/py/nagdoc_latest/' 'naginterfaces.kusari.html', ]) return lc_lines class _EngineLicenceErrorExit(_EngineExit): """ Concrete class for licensing error exits from the NAG Engine. """ # pylint: disable=too-few-public-methods __slots__ = [] def manage(self): from .info import impl_details # pylint:disable=bad-option-value,import-outside-toplevel pcode = impl_details()['product_code'] locus = self._set_locus() locus['errcodes'] = (_IERR_NO_LIC,) msg_lines = _lcheck_lines(pcode) msg_lines.extend([ _capture_hostid(), 'PRODUCT CODE = ' + pcode, ]) raise NagLicenceError(locus, _format_msg_lines(msg_lines)) class _EngineUnexpectedExit(_EngineExit): """ Concrete class for unexpected exits from the NAG Engine. """ # pylint: disable=too-few-public-methods __slots__ = [] def manage(self): locus = self._set_locus() new_errcodes = [_IERR_UNEXPECTED] if len(locus['errcodes']) > 1: new_errcodes.extend(locus['errcodes'][1:]) locus['errcodes'] = tuple(new_errcodes) raise NagException(locus, _format_msg_lines(['Unexpected error.'])) def _format_msg_lines(msg_lines): """ Formats msg_lines into a single string for printing as an error/warning message. """ return '\n' + '\n'.join( ['** ' + line for line in msg_lines] ) class _EngineState(object): """ Class for the data state of the underlying NAG Engine on this platform. Raises ------ NagOSError if running using 32-bit Python with a 64-bit Engine or vice versa. NagOSError if the underlying Engine is 32-bit and the system is not Windows. NagTypeError upon instantiation (only the single class is permitted). """ # pylint: disable=too-few-public-methods _compilers = _ENGINE_COMPILERS_DICT _sys_functype = staticmethod(_ctypes.CFUNCTYPE) if ( _THE_SYSTEM != 'Windows' and not _THE_SYSTEM.startswith('CYGWIN') and _ENGINE_DATA_MODEL == 'ILP32' ): # pragma: no cover raise NagOSError( {}, 'this product does not support use with a 32-bit NAG library ' 'on this system.' ) if ( _ENGINE_DATA_MODEL[-2:] != _get_sys_data_model()[-2:] ): # pragma: no cover raise NagOSError( {}, 'the Python interpreter is ' + _get_sys_data_model()[-2:] + '-bit ' 'but the underlying NAG library is ' + _ENGINE_DATA_MODEL[-2:] + '-bit. Please run using a matching pair.' ) _int_model = { 'bits': _ENGINE_INT_BITS, 'ctypes': _SYS_CTYPES_INTS[_ENGINE_INT_BITS], 'numpy': getattr(_np, 'int' + str(_ENGINE_INT_BITS)) } if _compilers['Fortran'] == 'Intel' and _ENGINE_DATA_MODEL != 'ILP32': _chrlen_ctype = _SYS_CTYPES_INTS[64] else: _chrlen_ctype = _SYS_CTYPES_INTS[32] if _compilers['Fortran'] == 'Intel': _literal_true_int = -1 elif _compilers['Fortran'] == 'NAG': _literal_true_int = 1 else: raise NagException( # pragma: no cover {}, 'NYI for Fortran compiler' ) _literal_true_cint = _int_model['ctypes'](_literal_true_int) if ( _compilers['Fortran'] == 'Intel' or ( _compilers['Fortran'] == 'NAG' and ( _THE_SYSTEM == 'Windows' or _THE_SYSTEM.startswith('CYGWIN') ) and _ENGINE_DATA_MODEL != 'ILP32' ) ): _complex_fun_rtn = 'as argument' else: _complex_fun_rtn = 'as retval' _engine_lib_base = ( _ENGINE_LIB_NAME_FROM_SPEC if _ENGINE_LIB_NAME_FROM_SPEC is not None else _ENGINE_PCODE ) _engine_lib = _engine_lib_base + _get_sys_shlib_ext() if _THE_SYSTEM == 'Windows' or _THE_SYSTEM.startswith('CYGWIN'): _cwd = _os.getcwd() if not _os.path.isdir(_ENGINE_LIBDIR): # pragma: no cover _have_handle = False msg = 'No such directory ' + _ENGINE_LIBDIR else: try: dll_dirs = _os.add_dll_directory(_ENGINE_LIBDIR) except AttributeError: # pragma: no cover dll_dirs = None try: _ctypes.CDLL("libiomp5md") except: # pylint: disable=bare-except pass _os.chdir(_ENGINE_LIBDIR) try: _engine_handle = _ctypes.CDLL(_engine_lib) except OSError as exc: _have_handle = False msg = str(exc) else: _have_handle = True _os.chdir(_cwd) if dll_dirs is not None: # pragma: no cover dll_dirs.close() else: try: _engine_handle = _ctypes.CDLL( _os.path.join(_ENGINE_LIBDIR, _engine_lib) ) except OSError as exc: _have_handle = False msg = str(exc) else: _have_handle = True if not _have_handle: pretty_bits = ('32' if _struct.calcsize('P') == 4 else '64') + '-bit' if _THE_SYSTEM == 'Windows' or _THE_SYSTEM.startswith('CYGWIN'): pretty_sys = 'Windows' elif _THE_SYSTEM == 'Darwin': pretty_sys = 'Mac' elif _THE_SYSTEM == 'Linux': pretty_sys = 'Linux' else: raise NagException( # pragma: no cover {}, 'inconceivable system ' + _THE_SYSTEM, ) diag_msg_lines = [ ' '.join([ 'The', pretty_bits, pretty_sys, _PKG_NAME, 'package', 'version', __version__ ]), 'failed to obtain a handle to the NAG library ' + _ENGINE_PCODE + '.', 'It is possible that incompatible Intel MKL or compiler runtimes ' 'have been loaded from the environment.', 'Please read the compatibility notes at', 'https://www.nag.com/numeric/py/nagdoc_latest/readme.html' '#software-stack-compatibility-notes', msg, ] raise NagException( {}, _format_msg_lines(diag_msg_lines), ) _errbuf_sep = b'\1' ( _debug_off, _debug_args_short, _debug_args_full, _debug_naninf_counts ) = range(4) _debug_level = _debug_off _break_ctypes_ptr_ref_cycles = False _cast_pointer_to_ctypes_array = True def __new__( cls, *args, **kwargs ): # pylint: disable=unused-argument raise NagTypeError( {}, 'instances of _EngineState class not permitted' ) @classmethod def _call_char_fun( cls, fun, chrlen, *args ): """ Return the result of calling the fixed-length character-valued ctypes function fun with arguments args. """ result = _EngineCharScalarType.empty_ctypes(chrlen=chrlen) fun(result, chrlen, *args) return result @classmethod def _format_char_cfun_proto( cls, *args ): """ Return a ctypes FUNCTYPE for a callback function having arguments args and returning fixed-length character data. """ return _EngineState._sys_functype( None, *cls._format_char_fun_args( *args ) ) @classmethod def _format_char_fun_proto( cls, fun, *args ): """ Update the ctypes prototype attributes for function fun having arguments args, when fun returns fixed-length character data. """ fun.restype = None fun.argtypes = cls._format_char_fun_args(*args) @classmethod def _format_char_fun_args( cls, *args ): """ Return an argument tuple for a function having arguments args and returning fixed-length character data. """ argtypes = [_EngineCharScalarType.c_type, _EngineState._chrlen_ctype] argtypes.extend(args) return tuple(argtypes) @classmethod def _call_complex_fun( cls, fun, *args ): """ Return the result of calling the complex-valued ctypes function fun with arguments args. """ if cls._complex_fun_rtn == 'as argument': result = fun.argtypes[0]._type_() fun(result, *args) else: result = fun(*args) return result @classmethod def _format_complex_cfun_proto( cls, fun_type, *args ): """ Return a ctypes FUNCTYPE for a callback function having arguments args and returning complex data of type fun_type. """ return _EngineState._sys_functype( cls._format_complex_fun_rtn(fun_type), *cls._format_complex_fun_args( fun_type, *args ) ) @classmethod def _format_complex_fun_args( cls, fun_type, *args ): """ Return an argument tuple for a function having arguments args and returning complex data of type fun_type. """ if cls._complex_fun_rtn == 'as argument': argtypes = [_ctypes.POINTER(fun_type)] argtypes.extend(args) return tuple(argtypes) return args @classmethod def _format_complex_fun_proto( cls, fun, fun_type, *args ): """ Update the ctypes prototype attributes for function fun having arguments args, when fun returns complex data of type fun_type. """ fun.restype = cls._format_complex_fun_rtn(fun_type) fun.argtypes = cls._format_complex_fun_args(fun_type, *args) @classmethod def _format_complex_fun_rtn( cls, fun_type, ): """ Return a ctypes function-result type for a function returning complex data of type fun_type. """ return None if cls._complex_fun_rtn == 'as argument' else fun_type @classmethod def _from_bool_to_int( cls, bool_data, ): """ Return an Engine-compatible integer for the input boolean. """ return 0 if not bool_data else cls._literal_true_int @classmethod def _get_symbol( cls, symbol_name, ): """ Utility for loading symbol_name from the Engine handle. """ return _get_symbol_from_lib_handle( cls._engine_handle, symbol_name, ) @classmethod def _modify_unit( cls, modify_mode, unit_name, unit, ): """ Set or get the Engine unit_name from unit number unit (a ctypes int). Raises NagException if modify_mode or unit_name are invalid. """ allowed_modes = ['set', 'get'] if modify_mode not in allowed_modes: raise NagException( # pragma: no cover { 'fun_name': '_EngineState._modify_unit', 'entity': 'modify_mode' }, ' '.join([ 'must be', _join_sequence(['"' + a_m + '"' for a_m in allowed_modes]) ]) ) allowed_names = ['adv', 'err'] if unit_name not in allowed_names: raise NagException( # pragma: no cover { 'fun_name': '_EngineState._modify_unit', 'entity': 'unit_name' }, ' '.join([ 'must be', _join_sequence(['"' + a_n + '"' for a_n in allowed_names]) ]) ) if unit_name == 'adv': unit_manager_name = 'x04abft' elif unit_name == 'err': unit_manager_name = 'x04aaft' else: raise NagException( # pragma: no cover {'fun_name': '_EngineState._modify_unit'}, 'inconceivable unit name' ) unit_manager = _get_symbol_from_lib_handle( cls._engine_handle, unit_manager_name ) unit_manager.restype = None unit_manager.argtypes = ( _BYREF_ENGINEDATA, _BYREF_INT, # Input, iflag. _BYREF_INT, # Input/output, nerr/nadv resp. ) _en_data = _EngineData() if modify_mode == 'set': _iflag = _EngineIntScalarType.c_type(1) elif modify_mode == 'get': _iflag = _EngineIntScalarType.c_type(0) else: raise NagException( # pragma: no cover {'fun_name': '_EngineState._modify_unit'}, 'inconceivable modify mode' ) old_debug_level = cls._debug_level cls._set_debug_level(cls._debug_off) unit_manager(_en_data, _iflag, unit) cls._set_debug_level(old_debug_level) return _EngineIntScalarType.to_py(unit) @classmethod def _set_unit( cls, unit_name, unit, ): """ Set the Engine unit_name to unit number unit. """ _unit = _EngineIntScalarType( unit, {'fun_name': '_EngineState._set_unit', 'entity': 'unit'} ).to_ctypes() cls._modify_unit('set', unit_name, _unit) @classmethod def _get_unit( cls, unit_name, ): """ Return an int for the unit number of Engine unit unit_name. """ _unit = _EngineIntScalarType.empty_ctypes() return cls._modify_unit('get', unit_name, _unit) @classmethod def _set_debug_level( cls, level, ): """ Set the current debug printing or NaN/infinity detection flag for arguments in NAG computational and dummy callback routines. When printing is enabled, each argument is printed on input and/or output (depending on its input/output class), and if possible the memory address of the argument is also shown. Output is done using Python's `logging` module, to the logger named 'naginterfaces.base.<chapter>.<function>', at logging level `DEBUG`. The default is no printing and no NaN/infinity detection, which may be explicitly requested by calling with `level` = `_debug_off`. A call with `level` = `_debug_args_full` enables printing of all arguments on entry and exit in subsequent calls of procedures in the NAG Library Engine, and this includes full printing of arrays (ignoring array increments). A diagnostic message will be output if for any reason the extent of an array is not known at run time. A call with `level` = `_debug_args_short` requests elided array printing (only the first and last elements being displayed). A call with `level` = `_debug_naninf_counts` indicates that no printing is to take place, but all floating-point scalar and array values are examined and a count of NaN/infinity values maintained. Thread-Safety Considerations ---------------------------- The debug printing level in the NAG Engine is managed by a global variable, and setting the level using `_set_debug_level` requires modifying this global variable. Examples -------- To send output from `naginterfaces.library.lapacklin.dgesv` (sic) to file 'spam.txt': >>> from logging import DEBUG, getLogger ... >>> logger = getLogger('naginterfaces.base.lapacklin.dgesv') >>> logger.setLevel(DEBUG) >>> fh = FileHandler('spam.txt') # doctest: +SKIP >>> logger.addHandler(fh) # doctest: +SKIP """ if level not in [ cls._debug_off, cls._debug_args_short, cls._debug_args_full, cls._debug_naninf_counts, ]: raise NagValueError( { 'fun_name': '.'.join( [_MODULE_NAME, '_EngineState._set_debug_level'] ), 'entity': 'level' }, 'must be _debug_off, _debug_args_short, ' + '_debug_args_full or _debug_naninf_counts' ) y90gcft_ctypes = _get_symbol_from_lib_handle( cls._engine_handle, 'y90gcft' ) y90gcft_ctypes.restype = None y90gcft_ctypes.argtypes = ( _ctypes.POINTER(cls._int_model['ctypes']), # Input, ival. ) _level = _EngineIntScalarType( level, { 'fun_name': '.'.join( [_MODULE_NAME, '_EngineState._set_debug_level'] ), 'entity': 'level', }, ).to_ctypes() y90gcft_ctypes(_level) cls._debug_level = level @classmethod def _handle_global_errors( cls, fun_name, errbuf, ifail, ): """ Process global (allocation, licence, ...) failures. Return split errbuf data otherwise. """ if ifail == _IERR_ALLOC_FAIL: _EngineAllocationErrorExit(fun_name=fun_name).manage() elif ifail == _IERR_NO_LIC: _EngineLicenceErrorExit(fun_name=fun_name).manage() elif ifail == _IERR_SUCCESS: return 0, None eb_data = cls._split_errbuf(errbuf) return int(eb_data[1]), eb_data[2:] @classmethod def _split_errbuf( cls, errbuf, ): """ Return a list of the split contents of the input errbuf c_char_Array. """ split_eb = [ _EngineCharScalarType.to_py(ed).lstrip() for ed in errbuf[:].strip(b'\0 ' + cls._errbuf_sep).split(cls._errbuf_sep) if ed ] if len(split_eb) < 3: raise NagException( # pragma: no cover {'fun_name': '_split_errbuf'}, 'inconceivable errbuf structure' ) return split_eb
[docs]class EngineIntCType(_EngineState._int_model['ctypes']): # pylint: disable=too-few-public-methods """ Class for the ``ctypes`` integer type that is compatible with the NAG Library Engine's integer type. The base class is platform dependent. Examples -------- >>> from naginterfaces.base import utils >>> i = utils.EngineIntCType(42) """
[docs]class EngineIntNumPyType(_EngineState._int_model['numpy']): # pylint: disable=too-few-public-methods """ Class for the NumPy integer type that is compatible with the NAG Library Engine's integer type. The base class is platform dependent. Examples -------- >>> import numpy as np >>> from naginterfaces.base import utils >>> i = np.array([42, 53], dtype=utils.EngineIntNumPyType) """
class _Debug(object): """ Context-management class for NAG Library Engine debug dumping. """ # pylint: disable=too-few-public-methods __slots__ = [ '_filename', '_level', '_old_log_level', '_old_debug_level', '_pop_handler', ] def __init__(self, filename=None, level=_EngineState._debug_args_short): """ Initialize `level`-level Engine debug dumping. The default level is `_EngineState._debug_args_short`. (See also `_EngineState._set_debug_level`.) Examples -------- >>> from naginterfaces.base import utils >>> with utils._Debug(): ... pass """ self._filename = filename self._level = level self._old_log_level = None self._old_debug_level = None self._pop_handler = False def __enter__(self): """ Activate Engine debug dumping. Save the current logging and debug levels. Output using Python's `logging` module, via the root logger. This is configured using `basicConfig(filename, DEBUG)`; i.e., a root logger with handlers already configured will not be affected. Note that if `filename` was not supplied to the instance initializer, or was None, this implies that output through a previously unconfigured root logger is sent to `sys.stderr`. """ self._old_log_level = _logging.getLogger().getEffectiveLevel() self._old_debug_level = _EngineState._debug_level self._pop_handler = not _logging.getLogger().handlers _logging.basicConfig(filename=self._filename, level=_logging.DEBUG) _EngineState._set_debug_level(self._level) def __exit__(self, *args): """ Restore the previous debug state. If on entry the root logger had no pre-configured handlers, here the list of root handlers is cleared. """ _EngineState._set_debug_level(self._old_debug_level) _logging.getLogger().setLevel(self._old_log_level) if self._pop_handler: our_handler = _logging.getLogger().handlers.pop() our_handler.close() class _NagLibOutput(object): """ Context-management class for restoring NAG Library output I/O units. Thread-Safety Considerations ---------------------------- The default output I/O units in the NAG Engine are managed by global variables, and this context manager requires modifying these global variables. Examples -------- >>> from naginterfaces.base import utils >>> with utils._NagLibOutput(): ... pass """ # pylint: disable=too-few-public-methods __slots__ = [] def __enter__(self): FileObjManager._set_default_units() def __exit__(self, *args): FileObjManager._set_ni_units() def _break_ref_cycles(func): """ Decorate with forced breaking of ctypes pointer reference cycles. """ @_functools.wraps(func) def wrapper(*args, **kwargs): # pylint: disable=missing-docstring old_break = _EngineState._break_ctypes_ptr_ref_cycles _EngineState._break_ctypes_ptr_ref_cycles = True result = func(*args, **kwargs) _EngineState._break_ctypes_ptr_ref_cycles = old_break return result return wrapper
[docs]class Handle(_ctypes.c_void_p): """ Class for communication handles in the NAG Library Engine. Overrides ``__copy__``. Overrides ``__deepcopy__``. Raises ------ NagAttributeError Shallow or deep copy was requested; these operations are prohibited for this class. See Also -------- :func:`naginterfaces.library.examples.blgm.lm_formula_ex.main` : This example uses ``Handle`` instances. """ # pylint: disable=too-few-public-methods def __copy__(self): locus = {'fun_name': '.'.join([_MODULE_NAME, 'Handle.__copy__'])} raise NagAttributeError( locus, 'shallow copy not permitted' ) def __deepcopy__(self, memo): locus = {'fun_name': '.'.join([_MODULE_NAME, 'Handle.__deepcopy__'])} raise NagAttributeError( locus, 'deep copy not permitted' )
_BYREF_HANDLE = _ctypes.POINTER(Handle) _BYREF_C_PTR = _ctypes.POINTER(_ctypes.c_void_p) class _EngineType(object): """ Base class for data passed to the NAG Engine. """ # pylint: disable=too-few-public-methods __slots__ = ['allow_none', 'data', 'locus'] _module_key_ctypes = 'ctypes' _module_key_ni = _MODULE_NAME _module_key_numpy = 'numpy' _module_key_python = 'builtin' def __init__( self, data, locus, allow_none=False, ): """ Store the payload data and wrapper locus. """ self.data = data self.locus = locus self.allow_none = allow_none class _EngineScalarType(_EngineType): """ Base class for scalar NAG Engine data. Extends __init__. """ # pylint: disable=too-few-public-methods __slots__ = [] c_type = lambda *args: None # pragma: no cover numpy_type = lambda *args: None # pragma: no cover type_map = {} def __init__( self, data, locus, allow_none=False, ): """ Raises NagTypeError if the payload is not a compatible instance for scalar Engine data. """ _EngineType.__init__(self, data, locus, allow_none) self._check_scalar_type_compat() @staticmethod def _copy_ctypes_data( data, target, ): """ Copy data from ctypes data to ctypes target. """ if data is not None: target.value = data.value @classmethod def empty_ctypes( cls, chrlen=None, # pylint: disable=unused-argument ): """ Return an empty ctypes scalar with the same data type as the payload. """ return (cls.c_type)() def refresh_ctypes_buffer( self, ctypes_buffer, ): """ Refresh ctypes_buffer with the contents of the payload. """ self._copy_ctypes_data(self.to_ctypes(), ctypes_buffer.contents) def to_ctypes( self, ): """ Return the underlying C-compatible cast of the payload. """ if self.allow_none and self.data is None: return self.data if isinstance(self.data, self.c_type): # pylint: disable=isinstance-second-argument-not-valid-type return self.data return self.c_type(self.data) def _check_scalar_type_compat( self, ): """ Raises NagTypeError if the payload is not an instance of any scalar type in the type_map. """ if self.allow_none and self.data is None: return if not isinstance( self.data, tuple([t for ttup in self.type_map.values() for t in ttup]) # pylint: disable=consider-using-generator ): raise NagTypeError( self.locus, _bad_instance_msg( self._type_map_to_strlist(), allow_none=self.allow_none, ) ) def _type_map_to_strlist( self, ): """ Return a list of sorted qualified type names from the type_map. """ type_names = [] for mkey, mval in self.type_map.items(): for type_val in mval: full_name = [] if ( mkey == self._module_key_ctypes and type_val.__name__ == 'Handle' ): continue if mkey != self._module_key_python: full_name.append(mkey) full_name.append(type_val.__name__) type_names.append('.'.join(full_name)) return sorted(type_names) class _EngineCharScalarType(_EngineScalarType): """ Concrete class for character scalar NAG Engine data. Extends __init__. Overrides empty_ctypes. Overrides refresh_ctypes_buffer. Overrides to_ctypes. """ # pylint: disable=too-few-public-methods __slots__ = ['chrlen'] _valid_np_string_classes = (_np.bytes_, _np.str_) _valid_py_string_classes = (bytes, str) numpy_type = _np.str_ c_type = _ctypes.POINTER(_ctypes.c_char) type_map = { _EngineScalarType._module_key_numpy: _valid_np_string_classes, _EngineScalarType._module_key_python: _valid_py_string_classes, } def __init__( self, data, locus, exp_chrlen, allow_none=False, ): """ Raises NagTypeError from the super __init__ if the payload data is not the correct type. Raises NagValueError if the payload contains non-ASCII characters. Raises NagValueError if the payload exceeds a required length. """ _EngineScalarType.__init__(self, data, locus, allow_none) if self.allow_none and self.data is None: self.chrlen = 0 elif exp_chrlen is None: self.chrlen = len(self.data) else: self.chrlen = exp_chrlen self._check_is_ascii() @classmethod def empty_ctypes( cls, chrlen=None, ): """ Return an empty ctypes string buffer with the same data type as the payload. """ if chrlen is None or chrlen <= 0: return _NullEngineCharArray() new_arr_cls = _ctypes.c_char*chrlen globals()[new_arr_cls.__name__] = new_arr_cls return new_arr_cls() def refresh_ctypes_buffer( self, ctypes_buffer, ): """ Refresh ctypes_buffer with the contents of the payload. """ if self.allow_none and self.data is None: return ct_data = self.to_ctypes() for i, ct_char in enumerate(ct_data): ctypes_buffer[i] = ct_char def to_ctypes( self, ): """ Return the underlying C-compatible cast of the payload taking care to explicitly encode to ASCII in Python 3+. If the payload is shorter than a non-None chrlen then it is blank padded on the right to this length. """ if self.allow_none and self.data is None: return self.data if isinstance(self.data, (str, _np.str_)): str_to_cast = self.data.encode('ascii') else: str_to_cast = self.data if self.chrlen > len(self.data): str_to_cast = str_to_cast.ljust(self.chrlen) return _ctypes.create_string_buffer( str_to_cast[:self.chrlen], size=self.chrlen ) @classmethod def to_py( cls, data, chrlen=None, ): """ Return the value of ctypes (char_p) data as a str. """ if data is None: return data data_block = ( data[:chrlen] if chrlen is not None else data[:] ) return data_block.decode(encoding='ascii').rstrip() to_py_from_retval = to_py def _check_is_ascii( self, ): """ Raises NagValueError if the payload contains non-ASCII characters. """ if self.allow_none and self.data is None: return if not _is_ascii(self.data[:self.chrlen], self.locus): raise NagValueError( self.locus, 'must ' + ('' if not self.allow_none else 'be None or ') + 'comprise only ASCII characters' ) _BYREF_STRING = _EngineCharScalarType.c_type class _EngineComplexScalarType(_EngineScalarType): # pylint: disable=too-few-public-methods """ Abstract class for complex scalar NAG Engine data. Overrides _copy_ctypes_data. Overrides to_ctypes. """ __slots__ = [] @staticmethod def _copy_ctypes_data( data, target, ): """ Copy data from ctypes data to ctypes target. """ if data is None: return target.real = data.real target.imag = data.imag def to_ctypes( self, ): """ Cast to the associated Structure. """ if self.allow_none and self.data is None: return self.data return self.c_type(self.data.real, self.data.imag) @classmethod def _to_py_complex( cls, data, ): """ Cast back from the Structure to Python native complex. """ return complex(data.real, data.imag) def _set_complex_fields(c_type): """ Return a _fields_ list for a complex C structure of type c_type. """ return [ ('real', c_type), ('imag', c_type), ] class _EngineComplexStructure(_ctypes.Structure): # pylint: disable=too-few-public-methods """ Abstract class for complex scalar NAG Engine data. """ _component_c_type = lambda *args: None # pragma: no cover class _EngineComplex64Structure(_EngineComplexStructure): # pylint: disable=too-few-public-methods """ Concrete C structure for single-precision complex. """ _component_c_type = _ctypes.c_float _fields_ = _set_complex_fields(_component_c_type) class _EngineComplex64ScalarType(_EngineComplexScalarType, _EngineScalarType): """ Concrete class for single-precision complex scalar NAG Engine data. """ # pylint: disable=too-few-public-methods __slots__ = [] c_type = _EngineComplex64Structure numpy_type = _np.complex64 type_map = { _EngineScalarType._module_key_numpy: (numpy_type,), } @classmethod def to_py( cls, data, chrlen=None, # pylint: disable=unused-argument ): """ Return the complex64 representation of the input data. """ if data is None: return data return cls.numpy_type(cls._to_py_complex(data)) to_py_from_retval = to_py _BYREF_COMPLEX64 = _ctypes.POINTER(_EngineComplex64ScalarType.c_type) class _EngineComplex128Structure(_EngineComplexStructure): # pylint: disable=too-few-public-methods """ Concrete C structure for double-precision complex. """ _component_c_type = _ctypes.c_double _fields_ = _set_complex_fields(_component_c_type) class _EngineComplex128ScalarType(_EngineComplexScalarType, _EngineScalarType): """ Concrete class for double-precision complex scalar NAG Engine data. """ # pylint: disable=too-few-public-methods __slots__ = [] c_type = _EngineComplex128Structure numpy_type = _np.complex128 type_map = { _EngineScalarType._module_key_numpy: (numpy_type,), _EngineScalarType._module_key_python: (complex,), } @classmethod def to_py( cls, data, chrlen=None, # pylint: disable=unused-argument ): """ Return the complex128 representation of the input data. """ if data is None: return data return cls._to_py_complex(data) to_py_from_retval = to_py _BYREF_COMPLEX128 = _ctypes.POINTER(_EngineComplex128ScalarType.c_type) class _EngineCptrScalarType(_EngineScalarType): """ Concrete class for void* scalar NAG Engine data. Overrides to_ctypes. """ # pylint: disable=too-few-public-methods __slots__ = [] c_type = Handle type_map = { _EngineScalarType._module_key_ctypes: (c_type,), _EngineScalarType._module_key_ni: (Handle,), } def to_ctypes( self, ): return self.data @classmethod def to_py( cls, data, chrlen=None, # pylint: disable=unused-argument ): """ Return the original data (which must therefore already be a Handle). """ return data _BUILTIN_INT_TYPES = (int,) _INT_TYPE_MAP = { _EngineScalarType._module_key_ctypes: ( _SYS_CTYPES_INTS[32], _SYS_CTYPES_INTS[64] ), _EngineScalarType._module_key_numpy: (_np.int32, _np.int64), _EngineScalarType._module_key_python: _BUILTIN_INT_TYPES, } _PROMOTABLE_INTS = tuple([ # pylint: disable=consider-using-generator t for tkey in ( _EngineScalarType._module_key_numpy, _EngineScalarType._module_key_python, ) for t in _INT_TYPE_MAP[tkey] ]) def _is_promotable_int(data): """ Return True if we consider scalar int data to be appropriate for promotion to float. """ return ( data is not None and isinstance(data, _PROMOTABLE_INTS) ) class _EngineFloatScalarType(_EngineScalarType): # pylint: disable=too-few-public-methods """ Abstract class for float scalar NAG Engine data. Extends __init__. """ __slots__ = [] def __init__( self, data, locus, allow_none=False, ): """ Promotes incoming data if an appropriate int type. """ _EngineScalarType.__init__( self, data, locus, allow_none, ) if _is_promotable_int(self.data): self.data = self.numpy_type(self.data) class _EngineFloat32ScalarType(_EngineFloatScalarType, _EngineScalarType): """ Concrete class for 32-bit float scalar NAG Engine data. """ # pylint: disable=too-few-public-methods __slots__ = [] c_type = _ctypes.c_float numpy_type = _np.float32 _numpy_types = [] _numpy_types.extend(_INT_TYPE_MAP[_EngineScalarType._module_key_numpy]) _numpy_types.append(numpy_type) _builtin_types = [] _builtin_types.extend(_INT_TYPE_MAP[_EngineScalarType._module_key_python]) type_map = { _EngineScalarType._module_key_ctypes: (c_type,), _EngineScalarType._module_key_numpy: tuple(_numpy_types), _EngineScalarType._module_key_python: tuple(_builtin_types), } @classmethod def _to_py_from_float( cls, data, ): """ Return the data as np.float32. """ if data is None: return data return cls.numpy_type(data) to_py_from_retval = _to_py_from_float @classmethod def to_py( cls, data, chrlen=None, # pylint: disable=unused-argument ): """ Return the float32 representation of the input data. """ if data is None: return data return cls._to_py_from_float( data.value if isinstance(data, cls.c_type) else data ) _BYREF_FLOAT32 = _ctypes.POINTER(_EngineFloat32ScalarType.c_type) class _EngineFloat64ScalarType(_EngineFloatScalarType, _EngineScalarType): """ Concrete class for 64-bit float scalar NAG Engine data. """ # pylint: disable=too-few-public-methods __slots__ = [] c_type = _ctypes.c_double numpy_type = _np.float64 _numpy_types = [] _numpy_types.extend(_INT_TYPE_MAP[_EngineScalarType._module_key_numpy]) _numpy_types.append(numpy_type) _builtin_types = [float] _builtin_types.extend(_INT_TYPE_MAP[_EngineScalarType._module_key_python]) type_map = { _EngineScalarType._module_key_ctypes: (c_type,), _EngineScalarType._module_key_numpy: tuple(_numpy_types), _EngineScalarType._module_key_python: tuple(_builtin_types), } @classmethod def _to_py_from_float( cls, data, ): """ Return the data as float. """ if data is None: return data return float(data) to_py_from_retval = _to_py_from_float @classmethod def to_py( cls, data, chrlen=None, # pylint: disable=unused-argument ): """ Return the float64 representation of the input data. """ if data is None: return data return cls._to_py_from_float( data.value if isinstance(data, cls.c_type) else data ) _BYREF_FLOAT64 = _ctypes.POINTER(_EngineFloat64ScalarType.c_type) class _EngineIntScalarType(_EngineScalarType): """ Concrete class for integer scalar NAG Engine data. This could be 32- or 64-bits depending on the underlying Engine. Extends __init__. Extends cast. """ # pylint: disable=too-few-public-methods __slots__ = [] _builtin_types = _BUILTIN_INT_TYPES c_type = _EngineState._int_model['ctypes'] numpy_type = _EngineState._int_model['numpy'] _ctypes_int64 = _SYS_CTYPES_INTS[64] type_map = _INT_TYPE_MAP def __init__( self, data, locus, allow_none=False, ): """ Raises NagTypeError from the super __init__ if the payload data is not the correct type. Raises NagOverflowError if the payload overflows the underlying integer type used by the NAG Engine. """ _EngineScalarType.__init__(self, data, locus, allow_none) self._check_is_repr() def to_ctypes( self, ): """ Return the underlying C-compatible cast of the payload taking care to account for 32- and 64-bit input. """ if self.allow_none and self.data is None: return self.data if isinstance( self.data, self.type_map[_EngineScalarType._module_key_ctypes] ): return self.c_type(self.data.value) return _EngineScalarType.to_ctypes(self) @classmethod def _to_py_from_int( cls, data, ): """ Return the data as int. """ if data is None: return data return int(data) to_py_from_retval = _to_py_from_int @classmethod def to_py( cls, data, chrlen=None, # pylint: disable=unused-argument ): """ Return the int representation of the input data. """ if data is None: return data return cls._to_py_from_int( data.value if isinstance(data, cls.c_type) else data ) def _check_is_repr( self ): """ Raises NagOverflowError if the payload overflows the underlying integer type used by the NAG Engine. """ if self.allow_none and self.data is None: return if self._payload_overflows(): raise NagOverflowError( self.locus, 'value must not overflow ' + str(_EngineState._int_model['bits']) + '-bit NAG Library Engine integer' ) def _do_overflow_check( self, ): """ Return a bool shortcutting for whether overflow checking is relevant for the payload. """ if isinstance(self.data, int): return True if _EngineState._int_model['bits'] == 64: # We don't allow long on platforms # where the Engine int is 64 bit return False return isinstance( self.data, (_SYS_CTYPES_INTS[64], _np.int64) ) def _payload_overflows( self, ): """ Return a bool for when the payload is outside the range of the integer type used by the NAG Engine. """ if not self._do_overflow_check(): return False if isinstance(self.data, _SYS_CTYPES_INTS[64]): val_to_check = self.data.value else: val_to_check = self.data return ( val_to_check >= 2**(_EngineState._int_model['bits'] - 1) or val_to_check <= -2**(_EngineState._int_model['bits'] - 1) - 1 ) _BYREF_INT = _ctypes.POINTER(_EngineIntScalarType.c_type) class _EngineLogicalScalarType(_EngineScalarType): """ Concrete class for logical scalar NAG Engine data. Overrides to_ctypes. """ # pylint: disable=too-few-public-methods __slots__ = [] c_type = _EngineState._int_model['ctypes'] numpy_type = _np.bool_ type_map = { _EngineScalarType._module_key_numpy: (numpy_type,), _EngineScalarType._module_key_python: (bool,), } _f_comp = _EngineState._compilers['Fortran'] def to_ctypes( self, ): """ Return the underlying C-compatible cast using the correct integer representation. """ if self.allow_none and self.data is None: return self.data return self.c_type( _EngineState._from_bool_to_int(self.data) ) @classmethod def _to_py_from_int( cls, data, ): """ Return the bool representation of the input int. """ if data is None: return data if cls._f_comp == 'Intel': return bool(data & 1) if cls._f_comp == 'NAG': return data != 0 raise NagException( # pragma: no cover {'fun_name': '_EngineLogicalScalarType._to_py_from_int'}, 'NYI for Fortran compiler ' + cls._f_comp ) to_py_from_retval = _to_py_from_int @classmethod def to_py( cls, data, chrlen=None, # pylint: disable=unused-argument ): """ Return the bool representation of the input int. """ if data is None: return data return cls._to_py_from_int( data.value if isinstance(data, cls.c_type) else data ) _BYREF_LOGICAL = _ctypes.POINTER(_EngineLogicalScalarType.c_type) _IS_DISCONTIG = lambda a: not ( a.flags.c_contiguous or a.flags.f_contiguous ) _CONTIG_ERR_STR = ( 'must be contiguous, ' + 'in either the C sense, the Fortran sense, or both.' ) def _get_sorder(locus, array): """ Return a string for the storage order used by `array`, or None if not multidimensional. """ if ( not isinstance(array, _np.ndarray) or array.ndim == 1 ): return None if array.flags.f_contiguous and array.flags.c_contiguous: return 'A' if array.flags.f_contiguous: return 'F' if array.flags.c_contiguous: return 'C' raise NagTypeError( locus, 'all arrays in the argument list ' + _CONTIG_ERR_STR ) def _get_mutual_sorder( locus, arrays, in_callback=False, spiked_sorder=_EngineData.default_spiked_sorder, ): """ Return a string for the storage order used by the input sequence of arrays, or None if no multidimensional arrays feature. Raises NagTypeError if the order is not uniform across all the arrays. Raises NagTypeError if spiked_sorder is required but not a str. Raises NagValueError if spiked_sorder is required but invalid. """ sorder = None for array, array_name in arrays: this_sorder = _get_sorder(locus, array) if this_sorder is None: if not in_callback: continue locus['returned'] = array_name raise NagException( locus, 'Inconceivable; not a multidimensional ndarray', ) if ( sorder is None or ( sorder == 'A' and sorder != this_sorder ) ): sorder = this_sorder elif this_sorder == 'A': continue elif sorder != this_sorder: locus['entity' if not in_callback else 'returned'] = array_name raise NagTypeError( locus, 'arrays ' + ( 'on input' if not in_callback else 'returned from this callback' ) + ' must use a consistent storage order' ) if sorder == 'A' or not arrays: _spiked_sorder = _EngineCharScalarType( spiked_sorder, locus, exp_chrlen=1, ).to_ctypes() _spiked_sorder_py = _EngineCharScalarType.to_py(_spiked_sorder) if _spiked_sorder_py not in ['C', 'F']: raise NagValueError( locus, 'supplied spiked_sorder was not \'C\' or \'F\'.', ) return _spiked_sorder_py if sorder is None: # Some arrays, all were None: return _EngineData.default_sorder return sorder def _get_size_from_shape( shape, chrlen=None, ): """ Return the product of the shape tuple. """ size = 1 for extent in shape: size *= extent if chrlen is not None: size *= chrlen return size def _to_sized_buffer( pointer, buffer_size, ): """ Return the input ctypes pointer type as an addressable buffer. """ return _ctypes.cast( pointer, _ctypes.POINTER(pointer._type_ * buffer_size), ).contents def _from_pointer_to_buffer( pointer, shape, dtype, force_ctypes_cast, chrlen=None, ): """ Return the input ctypes pointer type as a flat ndarray. """ size = _get_size_from_shape(shape, chrlen) if ( force_ctypes_cast or _EngineState._cast_pointer_to_ctypes_array ): sized_buffer = _to_sized_buffer( pointer, size, ) np_buffer = _np.frombuffer(sized_buffer, dtype=dtype) else: np_buffer = _np.ctypeslib.as_array(pointer, shape=(size,)) return np_buffer def _unlink_pointer( pointer, ): """ Break ctypes pointer reference cycles. """ if ( _EngineState._break_ctypes_ptr_ref_cycles and _EngineState._cast_pointer_to_ctypes_array ): del pointer._objects[id(pointer)] def _check_shape_good(data, exp_shape): """ Return a Boolean for `data` having the expected shape `exp_shape`. """ if isinstance(data, _np.ndarray): if ( data.size == 0 and any(extent == 0 for extent in exp_shape) ): return True if data.ndim != len(exp_shape): return False for extent_i, extent in enumerate(exp_shape): if extent is None: continue if data.shape[extent_i] != extent: return False return True if len(exp_shape) != 1: raise NagException( # pragma: no cover {'fun_name': '_check_shape_good'}, 'inconceivable in shape check' ) if exp_shape[0] is None: return True return len(data) == exp_shape[0] def _check_shape_nonneg(locus, shape): """ Raises NagShapeError if an entry in `shape` was negative. """ for dim_i, dim_n in enumerate(shape): if dim_n is None: continue if dim_n < 0: raise NagShapeError( locus, ( 'expected extent in dimension ' + str(dim_i + 1) + ' was ' + str(dim_n) + ' but must not be negative. ' 'Check that all supplied scalar integer arguments ' 'are correct.' ) ) class _EngineArrayType(_EngineType): """ Base class for array NAG Engine data. Extends __init__. """ # pylint: disable=too-few-public-methods,too-many-instance-attributes __slots__ = [ 'exp_shape', 'check_contents', 'rank', 'shape', 'size', 'sorder', ] containers = tuple() array_typecodes = tuple() _underlying_scalar = lambda *args: None # pragma: no cover c_type = lambda *args: None # pragma: no cover _null_ctypes = None _force_ctypes_cast = False _dtypes_equiv = staticmethod( lambda data_dtype, np_dtype: data_dtype == np_dtype ) _allowed_np_dtypes = (None,) def __init__( self, data, locus, exp_shape, mutual_sorder=None, check_contents=True, allow_none=False, ): """ Store the expected shape tuple. Register whether to iterate over the payload contents in order to deep-check scalar contents for compatibility. Raises NagTypeError if the payload is not a compatible container for array Engine data. Raises NagTypeError if the container contents are not type compatible for the Engine array type. Raises NagShapeError if the payload does not have the expected shape. """ # pylint: disable=too-many-arguments _EngineType.__init__(self, data, locus, allow_none) self.exp_shape = exp_shape self.check_contents = ( False if self.allow_none and data is None else check_contents ) self._check_container_compat( self.data, self.locus, len(self.exp_shape), self.allow_none, ) self._check_shape() ( self.rank, self.shape, self.size, ) = self._get_rank_shape_and_size( self.data, len(self.exp_shape), allow_none, ) self._check_contents_type_compat() self.sorder = mutual_sorder if self.rank > 1 else None @classmethod def empty_ctypes( cls, size, chrlen=None, # pylint: disable=unused-argument ): """ Return an empty ctypes array with the same data type as the payload. """ if size <= 0: return cls._null_ctypes new_arr_cls = cls.c_type._type_*size globals()[new_arr_cls.__name__] = new_arr_cls return new_arr_cls() @staticmethod def _make_shape_nonneg(shape): """ Make the input shape tuple have non-negative elements. """ return [max(shape_el, 0) for shape_el in shape] @classmethod def empty_ndarray( cls, shape, sorder=_EngineData.default_sorder, chrlen=None, # pylint: disable=unused-argument ): """ Return an empty ndarray with the same data type as the payload. """ return _np.empty( cls._make_shape_nonneg(shape), dtype=cls._underlying_scalar.numpy_type, order=sorder ) @classmethod def from_pointer( cls, pointer, shape, **kwargs ): """ Return the input ctypes pointer type as an ndarray. """ np_buffer = _from_pointer_to_buffer( pointer, cls._make_shape_nonneg(shape), cls._underlying_scalar.numpy_type, cls._force_ctypes_cast, chrlen=kwargs.get('chrlen'), ) if len(shape) == 1: return np_buffer return np_buffer.reshape( cls._make_shape_nonneg(shape), order=kwargs.get('sorder') ) @classmethod def _check_and_get_element( # pylint: disable=too-many-branches cls, data, locus, element_tuple, validate, ): """ Return element element_tuple from data as Python data. Raises NagIndexError if element_tuple is out of range. Raises exception from the scalar constructor if the element is invalid. """ el_tuple = ( element_tuple[0] if len(element_tuple) == 1 else element_tuple ) if not validate: data_el = data[el_tuple] else: try: data_el = data[el_tuple] index_ok = True except IndexError: index_ok = False locus_el = ', '.join([str(i) for i in element_tuple]) if len(element_tuple) > 1: locus_el = '[' + locus_el + ']' if not index_ok: raise NagIndexError( locus, 'requested index value ' + locus_el + ' is out of range' ) if isinstance(data, _np.ndarray): cls._check_dtype(data, locus) if cls is _EngineCharArrayType: locus['element'] = locus_el cls._underlying_scalar(data_el, locus, None) elif ( cls is _EngineIntArrayType and data.dtype != _np.int32 ): locus['element'] = locus_el cls._underlying_scalar(data_el, locus) elif isinstance(data, _array.array): cls._check_typecode(data, locus) if ( cls is _EngineIntArrayType and _EngineState._int_model['bits'] == 32 ): locus['element'] = locus_el cls._underlying_scalar(data_el, locus) elif isinstance(data, _ctypes.Array): cls._check_ctype(data, locus) if ( cls is _EngineIntArrayType and _EngineState._int_model['bits'] == 32 and data._type_ is not _SYS_CTYPES_INTS[32] ): locus['element'] = locus_el cls._underlying_scalar(data_el, locus) elif isinstance(data, list): locus['element'] = locus_el if cls is _EngineCharArrayType: cls._underlying_scalar(data_el, locus, None) else: cls._underlying_scalar(data_el, locus) else: raise NagException( # pragma: no cover locus, 'Inconceivable data instance', ) if ( cls is _EngineCharArrayType and isinstance(data_el, (bytes, _np.bytes_)) ): data_el = data_el.decode(encoding='ascii') return ( data_el.value if ( isinstance( data_el, ( cls._underlying_scalar.c_type, _ctypes.c_int, _ctypes.c_long, _ctypes.c_longlong, ) ) and not isinstance( data_el, ( _EngineComplex64Structure, _EngineComplex128Structure, ) ) ) else data_el ) @classmethod def get_element( cls, data, locus, element_tuple, validate=True, ): """ Return element element_tuple from data. """ if validate: cls._check_container_compat( data, locus, exp_rank=len(element_tuple), allow_none=False, ) cls._check_rank( data, locus, exp_rank=len(element_tuple), allow_none=False, ) return cls._check_and_get_element( data, locus, element_tuple, validate, ) @classmethod def get_nd( cls, data, locus, exp_rank, sorder, dim_type, allow_none=False, ): """ Return the lead/second/... dimension for data. """ # pylint: disable=too-many-arguments cls._check_container_compat(data, locus, exp_rank, allow_none) cls._check_rank(data, locus, exp_rank, allow_none) if exp_rank < 2: raise NagException( # pragma: no cover {'fun_name': 'get_nd'}, 'inconceivable dim request', ) if allow_none and data is None: return 0 if dim_type == 'l': if sorder == 'F': shape_index = 0 elif sorder == 'C': shape_index = -1 else: raise NagException( # pragma: no cover {'fun_name': 'get_nd'}, 'inconceivable sorder' ) elif dim_type == 's': if sorder == 'F': shape_index = 1 elif sorder == 'C': shape_index = -2 else: raise NagException( # pragma: no cover {'fun_name': 'get_nd'}, 'inconceivable sorder' ) else: raise NagException( # pragma: no cover {'fun_name': 'get_nd'}, 'inconceivable dim_type' ) return cls._get_rank_shape_and_size( data, exp_rank, allow_none )[1][shape_index] def _get_max( self, start_i, stop_i, ): """ Return the max of the (1d) payload over the requested range. """ if ( not isinstance(self.data, list) or not isinstance( self.data[start_i], ( self._underlying_scalar.c_type, _ctypes.c_int, _ctypes.c_long, _ctypes.c_longlong, ) ) ): return max(self.data[start_i:stop_i + 1]) return max( data_el.value for data_el in self.data[start_i:stop_i + 1] ) def _get_product( self, start_i, stop_i, ): """ Return the product of the (1d) payload over the requested range. """ product_val = 1 if ( not isinstance(self.data, list) or not isinstance( self.data[start_i], ( self._underlying_scalar.c_type, _ctypes.c_int, _ctypes.c_long, _ctypes.c_longlong, ) ) ): for data_el in self.data[start_i:stop_i + 1]: product_val *= data_el else: for data_el in self.data[start_i:stop_i + 1]: product_val *= data_el.value return product_val @classmethod def get_shape( cls, data, locus, exp_rank, allow_none=False, validate=True, ): """ Return the shape tuple for data. """ # pylint: disable=too-many-arguments if validate: cls._check_container_compat(data, locus, exp_rank, allow_none) cls._check_rank(data, locus, exp_rank, allow_none) return cls._get_rank_shape_and_size(data, exp_rank, allow_none)[1] def _get_sum( self, start_i, stop_i, ): """ Return the sum of the (1d) payload over the requested range. """ if ( not isinstance(self.data, list) or not isinstance( self.data[start_i], ( self._underlying_scalar.c_type, _ctypes.c_int, _ctypes.c_long, _ctypes.c_longlong, ) ) ): return sum(self.data[start_i:stop_i + 1]) return sum( data_el.value for data_el in self.data[start_i:stop_i + 1] ) def _is_monotonic( self, start_i, stop_i, direction, ): """ Return a Boolean for when the (1d) payload is monotonic over the requested range in the requested direction. """ incr = True decr = True take_val = ( isinstance(self.data, list) and isinstance( self.data[start_i], ( self._underlying_scalar.c_type, _ctypes.c_int, _ctypes.c_long, _ctypes.c_longlong, ) ) ) data_el = lambda i: self.data[i].value if take_val else self.data[i] for i in range(start_i, stop_i): if incr: incr = incr and data_el(i) <= data_el(i+1) if decr: decr = decr and data_el(i) >= data_el(i+1) if direction == 'either' and not incr and not decr: return False if direction == 'increasing' and not incr: return False if direction == 'decreasing' and not decr: return False return True def _is_nonnegative( self, start_i, stop_i, ): """ Return a Boolean for when the (1d) payload is non-negative over the requested range. """ take_val = ( isinstance(self.data, list) and isinstance( self.data[start_i], ( self._underlying_scalar.c_type, _ctypes.c_int, _ctypes.c_long, _ctypes.c_longlong, ) ) ) data_el = lambda i: self.data[i].value if take_val else self.data[i] for i in range(start_i, stop_i): if data_el(i) < 0: return False return True def to_ctypes( self, **kwargs ): """ Return the underlying C-compatible cast of the payload. """ if self.allow_none and self.data is None: return None if isinstance(self.data, _np.ndarray): return self._to_ctypes_from_ndarray(**kwargs) if isinstance(self.data, _array.array): return self._to_ctypes_from_array(**kwargs) if isinstance(self.data, _ctypes.Array): return self._to_ctypes_from_ctypes(**kwargs) if isinstance(self.data, list): return self._to_ctypes_from_list(**kwargs) raise NagException( # pragma: no cover self.locus, 'inconceivable container' ) def _to_ctypes_from_array( self, **_kwargs ): """ Return the underlying C-compatible cast of the payload array.array. """ return _ctypes.cast( self.data.buffer_info()[0], self.c_type ) def _to_ctypes_from_ctypes( self, **_kwargs ): """ Return the underlying C-compatible cast of the payload ctypes.Array. """ return self.data def _to_ctypes_from_list( self, **kwargs ): """ Return the underlying C-compatible cast of the payload list. """ if kwargs.get('can_reference_contents', True): return (self.c_type._type_*self.size)(*self.data) return self.empty_ctypes(self.size) def _to_ctypes_from_ndarray( self, **_kwargs ): """ Return the underlying C-compatible cast of the payload numpy.ndarray. """ return self.data.ctypes.data_as(self.c_type) @classmethod def _check_container_compat( cls, data, locus, exp_rank, allow_none, ): """ Raises NagTypeError if data is not an instance of any valid container. """ if allow_none and data is None: return if ( exp_rank <= 1 and not isinstance(data, cls.containers) ): raise NagTypeError( locus, _bad_instance_msg(cls._containers_to_strlist(), allow_none) ) if ( exp_rank > 1 and not isinstance(data, _np.ndarray) ): raise NagTypeError( locus, 'multidimensional array data ' + _bad_instance_msg(['numpy.ndarray'], allow_none) ) if isinstance(data, _np.ndarray) and _IS_DISCONTIG(data): raise NagTypeError( locus, 'numpy.ndarray data ' + _CONTIG_ERR_STR, ) def _check_contents_type_compat( self, ): """ Raises NagTypeError if the contents of the payload container are not compatible types for this array class. """ if self.allow_none and self.data is None: return if isinstance(self.data, _np.ndarray): self._check_contents_ndarray() elif isinstance(self.data, _array.array): self._check_contents_array() elif isinstance(self.data, _ctypes.Array): self._check_contents_ctypes() elif isinstance(self.data, list): self._check_contents_list() else: raise NagException( # pragma: no cover self.locus, 'inconceivable container ' + type(self.data).__name__ ) def _check_contents_1dim( self, ): """ Raises NagTypeError if the contents of the rank-1 payload fail their respective type checks. """ for arg_i, arg_el in enumerate(self.data): self.locus['element'] = arg_i self._underlying_scalar(arg_el, self.locus) def _check_contents_anydim( self, ): """ Raises NagTypeError if the contents of the payload fail their respective type checks. """ if ( isinstance(self.data, _np.ndarray) and self.data.ndim > 1 ): self._check_contents_multidim() else: self._check_contents_1dim() def _check_contents_multidim( # pylint: disable=no-self-use self, ): """ Raises NagTypeError if the contents of the multidimensional payload fail their respective type checks. """ raise NagException( # pragma: no cover {'fun_name': '_check_contents_multidim'}, 'not yet implemented', ) @classmethod def _check_typecode( cls, data, locus, ): """ Raises NagTypeError if the contents of the data array.array are not compatible types for this array class. """ if data.typecode not in cls.array_typecodes: raise NagTypeError( locus, 'typecode for array.array entity was ' + str(data.typecode) + ' but must be ' + _join_sequence( ['"' + a_t + '"' for a_t in cls.array_typecodes] ) ) def _check_contents_array( self, ): """ Raises NagTypeError if the contents of the payload array.array are not compatible types for this array class. """ self._check_typecode(self.data, self.locus) @classmethod def _check_ctype( cls, data, locus, ): """ Raises NagTypeError if the contents of the data ctypes.Array are not compatible types for this array class. """ allowed_ctypes = cls._underlying_scalar.type_map[ _EngineType._module_key_ctypes ] if data._type_ not in allowed_ctypes: raise NagTypeError( locus, '_type_ for ctypes.Array entity must be ' + _join_sequence( [ ( ctype.__name__ if ctype.__name__ == 'Handle' else '.'.join(['ctypes', ctype.__name__]) ) for ctype in allowed_ctypes ] ) ) def _check_contents_ctypes( self, ): """ Raises NagTypeError if the contents of the payload ctypes.Array are not compatible types for this array class. """ self._check_ctype(self.data, self.locus) def _check_contents_list( self, ): """ Raises NagTypeError if the contents of the payload list are not compatible types for this array class. """ if not self.check_contents: return self._check_contents_1dim() if self.size <= 1: return for arg_el in self.data[1:]: if not isinstance(arg_el, type(self.data[0])): raise NagTypeError( self.locus, 'must contain elements all of the same instance' ) def _check_contents_ndarray( self, ): """ Raises NagTypeError if the contents of the payload numpy.ndarray are not compatible types for this array class. """ self._check_dtype(self.data, self.locus) @classmethod def _check_dtype( cls, data, locus, allowed_dtypes=None, ): """ Raises NagTypeError if data is not one of the allowed numpy dtypes for this array class. """ a_dtypes = ( cls._allowed_np_dtypes if allowed_dtypes is None else allowed_dtypes ) for np_dtype in a_dtypes: if cls._dtypes_equiv(data.dtype, np_dtype): return raise NagTypeError( locus, 'dtype for numpy.ndarray entity must be an instance of ' + _join_sequence( ['.'.join(['numpy', ndtype.__name__]) for ndtype in a_dtypes] ) ) @classmethod def _check_rank( cls, data, locus, exp_rank, allow_none, ): """ Check the rank of data. Raises NagShapeError if the payload container does not have the expected rank. """ if allow_none and data is None: return rank, _, _ = cls._get_rank_shape_and_size(data, exp_rank, allow_none) if rank != exp_rank: raise NagShapeError( locus, _bad_rank_msg(exp_rank, rank) ) def _check_shape( self, ): """ Check the payload's shape. Raises NagShapeError if the payload container does not have the expected shape. Raises NagShapeError if any element of the expected shape was negative. """ if self.allow_none and self.data is None: return if _check_shape_good(self.data, self.exp_shape): return _check_shape_nonneg(self.locus, self.exp_shape) raise NagShapeError( self.locus, _bad_shape_msg( self.exp_shape, ( self.data.shape if isinstance(self.data, _np.ndarray) else (len(self.data),) ), ), ) @classmethod def _containers_to_strlist( cls, ): """ Return a list of sorted qualified type names from the containers tuple. """ type_names = [] for container in cls.containers: full_name = [] if container is _array.array: full_name.append('array') elif container is _ctypes.Array: full_name.append('ctypes') elif container is _np.ndarray: full_name.append('numpy') elif container is list: pass else: raise NagException( # pragma: no cover {'fun_name': '_containers_to_strlist'}, 'inconceivable container' ) full_name.append(container.__name__) type_names.append('.'.join(full_name)) return sorted(type_names) @staticmethod def _get_rank_shape_and_size( data, exp_rank, allow_none, ): """ Return the rank, shape and size of data. """ if allow_none and data is None: return exp_rank, tuple([0]*exp_rank), 0 if isinstance(data, _np.ndarray): return data.ndim, data.shape, data.size if isinstance(data, (_array.array, _ctypes.Array, list)): return 1, (len(data),), len(data) raise NagException( # pragma: no cover {'fun_name': '_get_rank_shape_and_size'}, 'inconceivable container', ) @classmethod def refresh_ctypes_buffer( cls, _data, _locus, ctypes_buffer, chrlen=None # pylint: disable=unused-argument ): # pylint: disable=too-many-arguments """ Refresh ctypes_buffer with the contents of data. This is the version where no back-conversion is necessary, e.g. for float64. """ def _refresh_py_list( self, data, ): """ Fill the payload list with ctypes `data`. """ self.data[:] = self._to_py_list(data) def _refresh_py_ndarray_copy( self, data, ): """ Fill the payload ndarray with ctypes `data`. This is the copy-back version. """ self.data[:] = self._to_py_ndarray( data, self.data.shape, sorder=self.sorder, chrlen=getattr(self, 'chrlen', None), ) def _refresh_py_noop( # pylint: disable=no-self-use self, data, # pylint: disable=unused-argument ): """ Fill the payload with ctypes `data`. This is the no-op version assuming that the memory buffer has been updated directly (and no copying was required). """ return _refresh_py_ndarray = _refresh_py_noop _refresh_py_array = _refresh_py_noop _refresh_py_ctarray = _refresh_py_noop def refresh_py( self, data, ): """ Fill the payload with ctypes `data`. """ if self.allow_none and self.data is None: return if isinstance(self.data, list): self._refresh_py_list(data) elif isinstance(self.data, _np.ndarray): self._refresh_py_ndarray(data) elif isinstance(self.data, _array.array): self._refresh_py_array(data) elif isinstance(self.data, _ctypes.Array): self._refresh_py_ctarray(data) else: raise NagException( # pragma: no cover {'fun_name': 'refresh_py'}, 'inconceivable container' ) def _to_py_list( self, data, ): """ Return ctypes `data` reshaped into a list. """ if not self.data: return [] cast_type = ( self._underlying_scalar.to_py if self.data[0] is None else type(self.data[0]) ) return [cast_type(data[i]) for i in range(self.size)] _to_py_ndarray = from_pointer class _NullEngineCharArray(_ctypes.Array): """ Null array. """ # pylint: disable=too-few-public-methods _length_ = 0 _type_ = _EngineCharScalarType.c_type._type_ def _byteorder_stripped_dtype( data, ): """ Return a stripped copy of the dtype of numpy `data` with byte-order markers removed. """ return data.dtype.str.lstrip('|<>=') class _EngineCharArrayType(_EngineArrayType): """ Concrete class for character array NAG Engine data. Extends __init__. Extends _check_contents_ndarray. Extends _check_contents_type_compat. Overrides _check_contents_1dim. Overrides empty_ctypes. Overrides empty_ndarray. Overrides from_pointer. Overrides refresh_ctypes_buffer. Overrides _refresh_py_ndarray. Overrides _to_ctypes_from_list. Overrides _to_ctypes_from_ndarray. Overrides _to_py_list. Overrides _to_py_ndarray. Supplies _check_contents_multidim. """ # pylint: disable=too-few-public-methods __slots__ = ['chrlen', 'flat_seq'] containers = (list, _np.ndarray) _underlying_scalar = _EngineCharScalarType c_type = _underlying_scalar.c_type _null_ctypes = _NullEngineCharArray() _force_ctypes_cast = True _dtypes_equiv = staticmethod(_np.issubdtype) _allowed_np_dtypes = _underlying_scalar.type_map[ _EngineType._module_key_numpy ] _numpy_kind = 'U' def __init__( self, data, locus, exp_chrlen, exp_shape, mutual_sorder=None, check_contents=True, allow_none=False, ): """ Raises NagShapeError from the super __init__ if the payload data is not the correct shape. Raises NagTypeError from the super __init__ if the payload data is not the correct type. Raises NagValueError if any payload element exceeds a required length. """ # pylint: disable=too-many-arguments self.chrlen = exp_chrlen self.flat_seq = None _EngineArrayType.__init__( self, data, locus, exp_shape, mutual_sorder, check_contents, allow_none, ) def _check_contents_1dim( self, ): """ Raises NagTypeError if the contents of the rank-1 payload fail their respective type checks. """ for arg_i, arg_el in enumerate(self.data): self.locus['element'] = arg_i self._underlying_scalar( arg_el, self.locus, self.chrlen, ) @staticmethod def _check_contents_multidim_direct( data, locus, chrlen, ): """ Raises NagTypeError if the contents of data fail their respective type checks. """ if data.size == 0: return data_it = _np.nditer(data, flags=['multi_index']) while not data_it.finished: locus['element'] = ', '.join([str(i) for i in data_it.multi_index]) if data.ndim > 1: locus['element'] = '[' + locus['element'] + ']' _EngineCharScalarType( data[data_it.multi_index], locus, chrlen, ) data_it.iternext() def _check_contents_multidim( self, ): """ Raises NagTypeError if the contents of the multidimensional payload fail their respective type checks. """ if self.size == 0: return self._check_contents_multidim_direct( self.data, self.locus, self.chrlen ) def _check_contents_type_compat( self, ): """ Set or compute chrlen before checking. """ self.flat_seq = self._set_flat_seq(self.data, self.allow_none) self._infer_chrlen() _EngineArrayType._check_contents_type_compat(self) @classmethod def empty_ctypes( cls, size, chrlen=None, ): """ Return an empty string buffer. """ if ( size <= 0 or chrlen is None or chrlen <= 0 ): return cls._null_ctypes new_arr_cls = _ctypes.c_char*(size*chrlen) globals()[new_arr_cls.__name__] = new_arr_cls return new_arr_cls() @classmethod def empty_ndarray( cls, shape, sorder=_EngineData.default_sorder, chrlen=None, ): """ Return an empty ndarray with the same data type as the payload. """ return _np.empty( cls._make_shape_nonneg(shape), dtype=( cls._numpy_kind + ('0' if chrlen is None else str(max(chrlen, 0))) ), order=sorder, ) @classmethod def from_pointer( cls, pointer, shape, **kwargs ): """ Return the input ctypes pointer type as an ndarray of byte chars. """ chrlen = kwargs.get('chrlen') _np_buffer = _from_pointer_to_buffer( pointer, cls._make_shape_nonneg(shape), 'S' + str(chrlen), cls._force_ctypes_cast, chrlen=chrlen, ) np_buffer = _np_buffer if kwargs.get('can_reference_contents', True): for buff_i, buff_el in enumerate(np_buffer): np_buffer[buff_i] = buff_el.rstrip() else: np_buffer[:] = [b'\0']*len(np_buffer) if len(shape) == 1: return np_buffer return np_buffer.reshape( cls._make_shape_nonneg(shape), order=kwargs.get('sorder') ) def _infer_chrlen(self): """ Compute maximal chrlen from the payload data if not already known. """ if self.chrlen is not None: return if isinstance(self.flat_seq, _np.ndarray): self.chrlen = int(self.flat_seq.dtype.str[2:]) return if isinstance(self.flat_seq, list): self.chrlen = ( 0 if not self.flat_seq else max([ len(c) for c in self.flat_seq if isinstance(c, _collections.abc.Sized) ]) ) return raise NagException( # pragma: no cover {'fun_name': '_infer_chrlen'}, 'inconceivable chrlen inference' ) @classmethod def refresh_ctypes_buffer( cls, data, locus, ctypes_buffer, chrlen=None ): """ This version back-casts to short-enough ASCII correctly. """ # pylint: disable=too-many-arguments cls._check_contents_multidim_direct(data, locus, chrlen) _ctypes_buffer = cls._to_ctypes_from_ndarray_data( data, chrlen, allow_none=False, ) for i in range(data.size*chrlen): ctypes_buffer[i] = _ctypes_buffer[i] _refresh_py_ndarray = _EngineArrayType._refresh_py_ndarray_copy @staticmethod def _set_flat_seq(data, allow_none): """ Return data as a flat ASCII sequence. """ if allow_none and data is None: return [] if isinstance(data, _np.ndarray): flat_data = data if data.ndim == 1 else data.flatten(order='K') elif isinstance(data, list): flat_data = data else: raise NagException( # pragma: no cover {}, 'inconceivable instance for _set_flat_seq' ) if ( len(flat_data) and # pylint: disable=len-as-condition isinstance(flat_data[0], (str, _np.str_)) ): strs_to_cast = [ c.encode('ascii') for c in flat_data if ( isinstance(c, (str, _np.str_)) and _is_ascii(c, {}) ) ] else: strs_to_cast = flat_data return ( strs_to_cast if isinstance(data, list) else _np.array( strs_to_cast, dtype='S' + _byteorder_stripped_dtype(data)[1:] ) ) def _to_ctypes_from_list( self, **kwargs ): """ Pad and join the payload to be contiguous. """ if kwargs.get('can_reference_contents', True): casted = self._pad_and_join_to_string_seq( self.flat_seq, self.chrlen ) return _ctypes.create_string_buffer(casted, size=len(casted)) return self.empty_ctypes(self.size, chrlen=self.chrlen) @classmethod def _to_ctypes_from_ndarray_data( cls, data, chrlen, allow_none, ): """ Direct creation of a string buffer for the data. """ casted = cls._pad_and_join_to_string_seq( cls._set_flat_seq(data, allow_none), chrlen, ) return _ctypes.create_string_buffer(casted, size=len(casted)) def _to_ctypes_from_ndarray( self, **kwargs ): """ Return a contiguous string with padding. If the payload is shorter than a non-None chrlen then it is blank padded on the right to this length. """ if kwargs.get('can_reference_contents', True): return self._to_ctypes_from_ndarray_data( self.data, self.chrlen, self.allow_none, ) return self.empty_ctypes(self.size, chrlen=self.chrlen) @classmethod def _to_bytes_list_from_ctypes( cls, data, size, chrlen=None, ): """ Return ctypes `data` reshaped into a list. """ return [ data[(i * chrlen):((i + 1) * chrlen)].rstrip() for i in range(size) ] def _to_py_list( self, data, ): """ Return ctypes `data` reshaped into a list. """ bytes_list = self._to_bytes_list_from_ctypes( data, self.size, chrlen=self.chrlen ) if ( not bytes_list or isinstance(self.data[0], bytes) ): return bytes_list return [c.decode(encoding='ascii') for c in bytes_list] def _to_py_ndarray( self, data, shape, **kwargs ): bytes_array = self.from_pointer(data, shape, **kwargs) data_char_class = _byteorder_stripped_dtype(self.data) return ( bytes_array if data_char_class[0] == 'S' else _np.array(bytes_array, dtype=data_char_class) ) def _check_contents_ndarray( self, ): """ Extend the super method to also check for ASCII and short-enough input. """ _EngineArrayType._check_contents_ndarray(self) if not self.check_contents: return self._check_contents_anydim() @staticmethod def _pad_and_join_to_string_seq( strs_to_cast, chrlen, ): """ Return a contiguous string with padding. If the payload is shorter than a non-None chrlen then it is blank padded on the right to this length. """ return b''.join( [c_str.ljust(chrlen)[:chrlen] for c_str in strs_to_cast] ) class _EngineComplexArrayType(_EngineArrayType): """ Abstract class for complex array NAG Engine data. Overrides _to_ctypes_from_list. Overrides _to_py_list. """ __slots__ = [] # Complex ctype arrays become void numpy dtypes when using as_array, so # are useless in that form. _force_ctypes_cast = True def _to_ctypes_from_list( self, **kwargs ): """ Return a ctypes Array of Complex Structures. """ if kwargs.get('can_reference_contents', True): return (self.c_type._type_*self.size)( *[(z.real, z.imag) for z in self.data] ) return self.empty_ctypes(self.size) def _to_py_list( self, data, ): """ Return ctypes `data` reshaped into a list. """ if not self.data: return [] return [ self._underlying_scalar.to_py(data[i]) if self.data[0] is None else type(self.data[0])(self._underlying_scalar.to_py(data[i])) for i in range(self.size) ] class _NullEngineComplex64Array(_ctypes.Array): """ Null array. """ # pylint: disable=too-few-public-methods _length_ = 0 _type_ = _EngineComplex64ScalarType.c_type class _EngineComplex64ArrayType(_EngineComplexArrayType, _EngineArrayType): """ Concrete class for single-precision complex array NAG Engine data. """ # pylint: disable=too-few-public-methods __slots__ = [] containers = (list, _np.ndarray) _underlying_scalar = _EngineComplex64ScalarType c_type = _ctypes.POINTER(_underlying_scalar.c_type) _null_ctypes = _NullEngineComplex64Array() _allowed_np_dtypes = _underlying_scalar.type_map[ _EngineType._module_key_numpy ] class _NullEngineComplex128Array(_ctypes.Array): """ Null array. """ # pylint: disable=too-few-public-methods _length_ = 0 _type_ = _EngineComplex128ScalarType.c_type class _EngineComplex128ArrayType(_EngineComplexArrayType, _EngineArrayType): """ Concrete class for double-precision complex array NAG Engine data. """ # pylint: disable=too-few-public-methods __slots__ = [] containers = (list, _np.ndarray) _underlying_scalar = _EngineComplex128ScalarType c_type = _ctypes.POINTER(_underlying_scalar.c_type) _null_ctypes = _NullEngineComplex128Array() _allowed_np_dtypes = _underlying_scalar.type_map[ _EngineType._module_key_numpy ] class _NullEngineCptrArray(_ctypes.Array): """ Null array. """ # pylint: disable=too-few-public-methods _length_ = 0 _type_ = _EngineCptrScalarType.c_type class _EngineCptrArrayType(_EngineArrayType): """ Concrete class for void* array NAG Engine data. Overrides empty_ctypes. Overrides empty_ndarray. Overrides from_pointer. Overrides refresh_ctypes_buffer. Overrides refresh_py. """ # pylint: disable=too-few-public-methods __slots__ = [] containers = (_ctypes.Array, list) _underlying_scalar = _EngineCptrScalarType c_type = _ctypes.POINTER(_underlying_scalar.c_type) _null_ctypes = _NullEngineCptrArray() @classmethod def empty_ndarray( cls, shape, sorder=_EngineData.default_sorder, chrlen=None, ): raise NotImplementedError @classmethod def from_pointer( cls, pointer, shape, **kwargs ): raise NotImplementedError @classmethod def refresh_ctypes_buffer( cls, data, locus, ctypes_buffer, chrlen=None, ): raise NotImplementedError def refresh_py( self, data, ): raise NotImplementedError class _EngineFloatArrayType(_EngineArrayType): """ Abstract class for float array NAG Engine data. Overrides _check_contents_1dim. """ __slots__ = [] def _check_contents_1dim( self, ): """ Promote promotable ints. """ for arg_i, arg_el in enumerate(self.data): self.locus['element'] = arg_i self._underlying_scalar(arg_el, self.locus) if ( isinstance(self.data, list) and _is_promotable_int(arg_el) ): self.data[arg_i] = self._underlying_scalar.numpy_type( self.data[arg_i] ) class _NullEngineFloat32Array(_ctypes.Array): """ Null array. """ # pylint: disable=too-few-public-methods _length_ = 0 _type_ = _EngineFloat32ScalarType.c_type class _EngineFloat32ArrayType(_EngineFloatArrayType, _EngineArrayType): """ Concrete class for 32-bit float array NAG Engine data. """ # pylint: disable=too-few-public-methods __slots__ = [] containers = (_array.array, _ctypes.Array, list, _np.ndarray) array_typecodes = ('f',) _underlying_scalar = _EngineFloat32ScalarType c_type = _ctypes.POINTER(_underlying_scalar.c_type) _null_ctypes = _NullEngineFloat32Array() _allowed_np_dtypes = (_underlying_scalar.numpy_type,) class _NullEngineFloat64Array(_ctypes.Array): """ Null array. """ # pylint: disable=too-few-public-methods _length_ = 0 _type_ = _EngineFloat64ScalarType.c_type class _EngineFloat64ArrayType(_EngineFloatArrayType, _EngineArrayType): """ Concrete class for 64-bit float array NAG Engine data. """ # pylint: disable=too-few-public-methods __slots__ = [] containers = (_array.array, _ctypes.Array, list, _np.ndarray) array_typecodes = ('d',) _underlying_scalar = _EngineFloat64ScalarType c_type = _ctypes.POINTER(_underlying_scalar.c_type) _null_ctypes = _NullEngineFloat64Array() _allowed_np_dtypes = (_underlying_scalar.numpy_type,) get_max = _EngineArrayType._get_max get_product = _EngineArrayType._get_product get_sum = _EngineArrayType._get_sum is_monotonic = _EngineArrayType._is_monotonic is_nonnegative = _EngineArrayType._is_nonnegative class _NullEngineIntArray(_ctypes.Array): """ Null array. """ # pylint: disable=too-few-public-methods _length_ = 0 _type_ = _EngineIntScalarType.c_type class _EngineIntArrayType(_EngineArrayType): """ Concrete class for integer array NAG Engine data. This could be 32- or 64-bits depending on the underlying Engine. Overrides _to_ctypes_from_list. Extends _check_contents_array. Extends _check_contents_ctypes. Extends _check_contents_ndarray. Extends _refresh_py_array. Extends _refresh_py_ctarray. Extends _refresh_py_ndarray. Extends _to_ctypes_from_array. Extends _to_ctypes_from_ctypes. Extends _to_ctypes_from_ndarray. Supplies _check_contents_multidim (int32 only). """ # pylint: disable=too-few-public-methods __slots__ = [] containers = (_array.array, _ctypes.Array, list, _np.ndarray) array_typecodes = ('l', 'q') _underlying_scalar = _EngineIntScalarType c_type = _ctypes.POINTER(_underlying_scalar.c_type) _null_ctypes = _NullEngineIntArray() _allowed_np_dtypes = _underlying_scalar.type_map[ _EngineType._module_key_numpy ] get_max = _EngineArrayType._get_max get_product = _EngineArrayType._get_product get_sum = _EngineArrayType._get_sum is_monotonic = _EngineArrayType._is_monotonic is_nonnegative = _EngineArrayType._is_nonnegative if _EngineState._int_model['bits'] == 32: def _to_ctypes_from_array( self, **kwargs ): """ Override the super method to take care to downcast 64-bit integer input. """ if ( self.data.typecode == 'q' or ( self.data.typecode == 'l' and not _IS_IL32 ) ): if kwargs.get('can_reference_contents', True): data_32 = _np.array( self.data, dtype=_EngineState._int_model['numpy'] ) return data_32.ctypes.data_as(self.c_type) return self.empty_ctypes(self.size) return _EngineArrayType._to_ctypes_from_array(self, **kwargs) def _refresh_py_array( self, data, ): """ Fill the payload array.array with ctypes `data`. This is the mixed copy-back or no-op version. """ if ( self.data.typecode == 'q' or ( self.data.typecode == 'l' and not _IS_IL32 ) ): for i in range(self.size): self.data[i] = data[i] return _EngineArrayType._refresh_py_array(self, data) def _refresh_py_ctarray( self, data, ): """ Fill the payload ctypes.Array with ctypes `data`. This is the mixed copy-back or no-op version. """ if data._type_ is not self.data._type_: self.data[:] = (self.data._type_*self.size)(*data) return _EngineArrayType._refresh_py_ctarray(self, data) def _refresh_py_ndarray( self, data, ): """ Fill the payload ndarray with ctypes `data`. This is the mixed copy-back or no-op version. """ refresh_fun = ( self._refresh_py_ndarray_copy if self._has_different_int_width(self.data) else self._refresh_py_noop ) return refresh_fun(data) def _to_ctypes_from_ctypes( self, **kwargs ): """ Override the super method to take care to cast to Engine integer input. """ if self.data._type_ is not self.c_type._type_: return (self.c_type._type_*self.size)(*self.data) return _EngineArrayType._to_ctypes_from_ctypes(self, **kwargs) def _to_ctypes_from_list( self, **kwargs ): """ Return the underlying C-compatible cast of the payload list. """ if kwargs.get('can_reference_contents', True): return (self.c_type._type_*self.size)( *[ _EngineState._int_model['ctypes']( x.value if isinstance(x, ( _ctypes.c_int, _ctypes.c_long, _ctypes.c_longlong )) else x ) for x in self.data ] ) return self.empty_ctypes(self.size) @staticmethod def _has_different_int_width( data, ): """ Return True when numpy int `data` is of a different type to the Engine int. """ return ( ( _EngineState._int_model['bits'] == 32 and data.dtype == _np.int64 ) or ( _EngineState._int_model['bits'] == 64 and data.dtype == _np.int32 ) ) @classmethod def _to_ctypes_from_flat_np_data( cls, data, ): """ Direct conversion of 1d data to Engine integer. """ return (cls.c_type._type_*len(data))( *[ _EngineState._int_model['numpy'](x) for x in data ] ) def _to_ctypes_from_ndarray( self, **kwargs ): """ Override the super method to take care to cast to Engine integer input. """ if self._has_different_int_width(self.data): if kwargs.get('can_reference_contents', True): return self._to_ctypes_from_flat_np_data( self.data.flatten(order='K') ) return self.empty_ctypes(self.size) return _EngineArrayType._to_ctypes_from_ndarray(self, **kwargs) if _EngineState._int_model['bits'] == 32: def _check_contents_array( self, ): """ Extend the super method to also check for in-range input. """ _EngineArrayType._check_contents_array(self) if not self.check_contents: return self._check_contents_1dim() def _check_contents_ctypes( self, ): """ Extend the super method to also check for in-range input. """ _EngineArrayType._check_contents_ctypes(self) if not self.check_contents: return if self.data._type_ is _SYS_CTYPES_INTS[32]: return self._check_contents_1dim() def _check_contents_ndarray( self, ): """ Extend the super method to also check for in-range input. """ _EngineArrayType._check_contents_ndarray(self) if not self.check_contents: return if self.data.dtype == _np.int32: return self._check_contents_anydim() def _check_contents_multidim( self, ): if self.size == 0: return data_it = _np.nditer(self.data, flags=['multi_index']) while not data_it.finished: self.locus['element'] = ( '[' + ', '.join( [str(i) for i in data_it.multi_index] ) + ']' ) self._underlying_scalar( self.data[data_it.multi_index], self.locus ) data_it.iternext() class _NullEngineLogicalArray(_ctypes.Array): """ Null array. """ # pylint: disable=too-few-public-methods _length_ = 0 _type_ = _EngineLogicalScalarType.c_type class _EngineLogicalArrayType(_EngineArrayType): """ Concrete class for logical array NAG Engine data. Overrides from_pointer. Overrides refresh_ctypes_buffer. Overrides _refresh_py_ndarray. Overrides _to_ctypes_from_list. Overrides _to_ctypes_from_ndarray. """ # pylint: disable=too-few-public-methods __slots__ = [] containers = (list, _np.ndarray) _underlying_scalar = _EngineLogicalScalarType c_type = _ctypes.POINTER(_underlying_scalar.c_type) _null_ctypes = _NullEngineLogicalArray() _force_ctypes_cast = True _allowed_np_dtypes = _underlying_scalar.type_map[ _EngineType._module_key_numpy ] @classmethod def from_pointer( cls, pointer, shape, **kwargs ): """ Convert from int to bool. """ int_buffer = _from_pointer_to_buffer( pointer, cls._make_shape_nonneg(shape), _EngineState._int_model['numpy'], cls._force_ctypes_cast, ) if kwargs.get('can_reference_contents', True): bool_ndarray = _np.array( [ cls._underlying_scalar._to_py_from_int(int_buffer[i]) for i in range(len(int_buffer)) ], dtype=cls._underlying_scalar.numpy_type ) else: bool_ndarray = _np.array( [False]*len(int_buffer), dtype=cls._underlying_scalar.numpy_type ) if len(shape) == 1: return bool_ndarray return bool_ndarray.reshape( cls._make_shape_nonneg(shape), order=kwargs.get('sorder') ) @classmethod def refresh_ctypes_buffer( cls, data, _locus, ctypes_buffer, chrlen=None ): """ This version back-casts to integer correctly. """ # pylint: disable=too-many-arguments _ctypes_buffer = cls._to_ctypes_from_flat_data(data.flatten(order='K')) for i in range(data.size): ctypes_buffer[i] = _ctypes_buffer[i] _refresh_py_ndarray = _EngineArrayType._refresh_py_ndarray_copy def _to_ctypes_from_list( self, **kwargs ): """ Return the underlying C-compatible cast of the payload list. """ if kwargs.get('can_reference_contents', True): return self._to_ctypes_from_flat_data(self.data) return self.empty_ctypes(self.size) @classmethod def _to_ctypes_from_flat_data( cls, data, ): """ Direct conversion of 1d data to Engine integer. """ return (cls.c_type._type_*len(data))( *[ _EngineState._from_bool_to_int(x) for x in data ] ) def _to_ctypes_from_ndarray( self, **kwargs ): """ Convert to Engine int and then return as a pointer. """ if kwargs.get('can_reference_contents', True): return self._to_ctypes_from_flat_data(self.data.flatten(order='K')) return self.empty_ctypes(self.size) _to_py_ndarray = from_pointer class _EngineErrbuf(_ctypes.Array): """ NAG Engine error-state string. """ # pylint: disable=too-few-public-methods _length_ = 200 _type_ = _ctypes.c_char class _EngineRoutineName(_ctypes.Array): """ NAG Engine short routine name using during e.g. option listing. """ # pylint: disable=too-few-public-methods _length_ = 6 _type_ = _ctypes.c_char # ctypes function prototype for the Engine's explicitly C-interoperable # print handler. _C_PRINT_REC_FUNCTYPE = _EngineState._sys_functype( None, _BYREF_ENGINEDATA, _BYREF_INT, # Input, nout. _EngineCharArrayType.c_type, # Input, rec(len=1)[200]. _BYREF_INT, # Input, length_rec. _BYREF_INT, # Output, ifail. ) # ctypes args for the Engine's default print handler. _PRINT_REC_ARGS = ( _BYREF_ENGINEDATA, _BYREF_INT, # Input, nout. _BYREF_STRING, # Input, rec(len=*). _BYREF_INT, # Output, ifail. _EngineState._chrlen_ctype, # Input, hidden chrlen of rec. ) # ctypes function prototype for the Engine's default # print handler. _PRINT_REC_FUNCTYPE = _EngineState._sys_functype( None, *_PRINT_REC_ARGS ) _UNUSED_PRINT_REC = _cast_callback( _PRINT_REC_FUNCTYPE, None, {'fun_name': _MODULE_NAME} ) # ctypes function prototype for the Engine's default # print handler helper. _PRINT_RECH_FUNCTYPE = _EngineState._sys_functype( None, _PRINT_REC_FUNCTYPE, *_PRINT_REC_ARGS ) # ctypes args for the Engine's default read handler. _READ_REC_ARGS = ( _BYREF_ENGINEDATA, _BYREF_INT, # Input, nin. _BYREF_STRING, # Output, recin(len=*). _BYREF_STRING, # Output, errbuf(len=200). _BYREF_INT, # Output, ifail. _EngineState._chrlen_ctype, # Input, hidden chrlen of recin. _EngineState._chrlen_ctype, # Input, hidden chrlen of errbuf. ) # ctypes function prototype for the Engine's default # read handler. _READ_REC_FUNCTYPE = _EngineState._sys_functype( None, *_READ_REC_ARGS ) _UNUSED_READ_REC = _cast_callback( _READ_REC_FUNCTYPE, None, {'fun_name': _MODULE_NAME} ) # ctypes function prototype for the Engine's default # read handler helper. _READ_RECH_FUNCTYPE = _EngineState._sys_functype( None, _READ_REC_FUNCTYPE, *_READ_REC_ARGS ) # ctypes args for the Engine's default nonadvancing-read handler. _NONADV_READ_REC_ARGS = ( _BYREF_ENGINEDATA, _BYREF_INT, # Input, nin. _BYREF_STRING, # Output, recin(len=*). _BYREF_INT, # Output, readlen. _BYREF_INT, # Output, inform. _EngineState._chrlen_ctype, # Input, hidden chrlen of recin. ) # ctypes function prototype for the Engine's default # nonadvancing-read handler. _NONADV_READ_REC_FUNCTYPE = _EngineState._sys_functype( None, *_NONADV_READ_REC_ARGS ) _UNUSED_NONADV_READ_REC = _cast_callback( _NONADV_READ_REC_FUNCTYPE, None, {'fun_name': _MODULE_NAME} ) # ctypes function prototype for the Engine's default # nonadvancing-read handler helper. _NONADV_READ_RECH_FUNCTYPE = _EngineState._sys_functype( None, _NONADV_READ_REC_FUNCTYPE, *_NONADV_READ_REC_ARGS ) # Dummy callback prototype and definition for unreferenced # Engine callbacks. _UNUSED_FUNCTYPE = _EngineState._sys_functype(None) _UNUSED_NULL_CALLBACK = _cast_callback( _UNUSED_FUNCTYPE, None, {'fun_name': _MODULE_NAME} ) def _unused_callback(): # pragma: no cover return None _UNUSED_NONNULL_CALLBACK = _cast_callback( _UNUSED_FUNCTYPE, _unused_callback, {'fun_name': _MODULE_NAME} ) def _cast_to_unused_callback(user_cb): return ( _UNUSED_NULL_CALLBACK if user_cb is None else _UNUSED_NONNULL_CALLBACK )
[docs]class FileObjManager(object): # pylint: disable=line-too-long """ Class for translating between Python file objects and NAG Library Engine I/O units. NAG Engine I/O units are roughly synonymous with Unix file descriptors. File objects may be registered with I/O units using the instance initializer or the :meth:`register` method. Convenience methods :meth:`register_to_errunit` and :meth:`register_to_advunit` are supplied to associate file objects with the Engine error unit (analogous to standard error, stderr) and the Engine advisory unit (analogous to standard output, stdout), respectively. A Python file object is considered valid for use with this class if it has a ``fileno`` method returning its underlying file descriptor. In particular, you may use any concrete subclass of ``io.IOBase`` (including the ``file`` attribute of a file object returned by ``tempfile.NamedTemporaryFile``). Naturally, the object must also have ``read``/``readline`` and ``write`` methods. Resource management (opening/closing/deleting/... files) remains the reponsibility of the caller. In particular, objects to be used for output streams must have been opened for writing (mut. mut. for objects to be used for input) before their associated units are used by the Engine. To communicate I/O unit numbers to the NAG Engine the :meth:`unit_from_fileobj` method must be used. This maps a (registered) file object to a valid unit number. Writes requested from the Engine to a unit number that does not have a file object associated with it will fall back to being sent to ``sys.stdout``. By default written data is prefixed with the name of the calling NAG Python function. This behaviour can be disabled when constructing an instance by using the initializer's `locus_in_output` argument. Reads requested by the Engine from a unit number that does not have a file object associated with it will cause **NagOSError** to be raised. File objects may be removed from the class's registry of I/O units by calling the :meth:`deregister` method. The special nature of the Engine advisory and error units means that these have persistent file-object associations; they are not affected by a call to ``deregister``. The methods :meth:`deregister_errunit` and :meth:`deregister_advunit` are available to restore the associations with ``sys.stderr`` and ``sys.stdout``, respectively. A module-level instance of this class called ``GLOBAL_IO_MANAGER`` is supplied for convenience. This is queried as the fallback I/O manager whenever the Engine requires an I/O manager but one has not been explicitly supplied to the calling Python function. It is constructed using the initializer's default argument set: in particular, prefixing of output data with NAG Python function names is enabled. Attributes ---------- errunit : int The value denoting the Engine error unit. This attribute should be considered read only. By default `errunit` is associated with ``sys.stderr``. advunit : int The value denoting the Engine advisory unit. This attribute should be considered read only. By default `advunit` is associated with ``sys.stdout``. Parameters ---------- fileobjs : None or tuple, optional If not ``None``, the file objects to register. locus_in_output : bool, optional Controls whether output managed by this instance is prefixed with the name of the calling NAG Python function. Raises ------ NagTypeError `fileobjs` is neither ``None`` nor an instance of ``tuple``. NagOSError An element of the `fileobjs` ``tuple`` is not a valid file-object instance. Examples -------- To send monitoring from a (fictional) optimization function to a temporary file: >>> import os >>> import tempfile >>> t_fd, t_path = tempfile.mkstemp() >>> f1 = open(t_path, 'w') >>> io_m = FileObjManager((f1,)) >>> io_m_unit = io_m.unit_from_fileobj(f1) >>> OptimizerOptionSet("Monitoring File", io_m_unit) # doctest: +SKIP ... >>> Optimizer(..., io_manager=io_m) # doctest: +SKIP ... >>> f1.close() >>> os.close(t_fd) ... >>> os.unlink(t_path) To display iteration statistics from another (fictional) optimization function, in which the statistics are documented as being sent to the advisory unit, delegating handling to ``GLOBAL_IO_MANAGER``: >>> AnotherOptimizerOptionSet("Print Level = 1") # doctest: +SKIP # Don't supply a value for io_manager: >>> AnotherOptimizer(...) # doctest: +SKIP See Also -------- :func:`naginterfaces.library.examples.opt.handle_solve_bounds_foas_ex.main` : This example uses a ``FileObjManager``. """ # pylint: enable=line-too-long __slots__ = ['_locus_in_output', '_mappings'] _unit_shift = 10 _default_errunit = _EngineState._get_unit('err') errunit = _unit_shift - 2 _EngineState._set_unit('err', errunit) _default_advunit = _EngineState._get_unit('adv') advunit = _unit_shift - 1 _EngineState._set_unit('adv', advunit) _print_err = -10000 _read_err_unassoc = -20000 _read_err_nonascii = -20001 _read_err = -20002 _allowed_fo_insts = (_io.IOBase,) _allowed_fo_insts_str = 'io.IOBase' def __init__(self, fileobjs=None, locus_in_output=True): """ Register the supplied file objects with this manager instance. Parameters ---------- fileobjs : None or tuple, optional If not ``None``, the file objects to register. locus_in_output : bool, optional Controls whether output managed by this instance is prefixed with the name of the calling NAG Python function. Raises ------ NagTypeError `fileobjs` is neither ``None`` nor an instance of ``tuple``. NagOSError An element of the `fileobjs` ``tuple`` is not a valid file-object instance. """ self._locus_in_output = locus_in_output self._mappings = { self.errunit: _sys.stderr, self.advunit: _sys.stdout, } if fileobjs is None: return locus = { 'fun_name': '.'.join([_MODULE_NAME, 'FileObjManager.__init__']), 'entity': 'fileobjs' } self._check_valid_tuple(fileobjs, locus) for fileobj in fileobjs: self._check_valid_fo_and_fd(fileobj, locus) self.register(fileobjs) @classmethod def _check_type_compat(cls, data, locus): """ Raises NagTypeError if non-None `data` is not an instance of `cls`. """ if ( data is None or isinstance(data, cls) ): return raise NagTypeError( locus, _bad_instance_msg( ['.'.join([_MODULE_NAME, 'FileObjManager'])], allow_none=False, ) ) @staticmethod def _check_valid_fd(fileobj, locus): """ Raises NagOSError if `fileobj` does not use a file descriptor. """ try: fileobj.fileno() except OSError: has_fd = False else: has_fd = True if not has_fd: raise NagOSError( locus, 'must implement the fileno method' ) def _check_valid_fo_and_fd(self, fileobj, locus): """ Raises NagOSError if `fileobj` is not a valid file-object instance. Raises NagOSError if `fileobj` does not use a file descriptor. """ if not isinstance(fileobj, self._allowed_fo_insts): raise NagOSError( locus, 'must be an instance of ' + self._allowed_fo_insts_str ) self._check_valid_fd(fileobj, locus) @staticmethod def _check_valid_tuple(fileobjs, locus): """ Raises NagTypeError if argument `fileobjs` is not an instance of tuple. """ if not isinstance(fileobjs, tuple): raise NagTypeError( locus, 'must be an instance of tuple' )
[docs] def deregister(self, fileobjs): """ Deregister the supplied file objects from this manager instance. Parameters ---------- fileobjs : tuple The file objects to deregister. Raises ------ NagTypeError `fileobjs` is not an instance of ``tuple``. """ locus = { 'fun_name': '.'.join([_MODULE_NAME, 'FileObjManager.deregister']), 'entity': 'fileobjs' } self._check_valid_tuple(fileobjs, locus) for fileobj in fileobjs: unit_to_dereg = self._fileobj_to_unit( fileobj, locus, check_range=False ) self._deregister_unit(unit_to_dereg)
[docs] def deregister_advunit(self): """ Remove the current association for the Engine advisory unit and restore it to ``sys.stdout``. """ self._mappings[self.advunit] = _sys.stdout
[docs] def deregister_errunit(self): """ Remove the current association for the Engine error unit and restore it to ``sys.stderr``. """ self._mappings[self.errunit] = _sys.stderr
def _deregister_unit(self, unit): """ Deregister the supplied unit number. """ if unit not in self._mappings: return del self._mappings[unit] @classmethod def _fileno_to_unit(cls, fileno, check_range=True): """ Map the input file-descriptor number to a 'safe' (unreserved) unit number for the Engine. """ unitno = fileno + cls._unit_shift if check_range and unitno > 2147483647: raise NagException( # pragma: no cover {'fun_name': '_fileno_to_unit'}, 'unit number will be out of range. ' 'Consider closing other open files', ) return unitno def _fileobj_from_unit(self, unit): """ Return the file object registered with the input unit number, or None. """ return self._mappings.get(unit) def _fileobj_to_unit(self, fileobj, locus, check_range=True): """ Return the unit number for the input file object. `fileobj` does not need to have been registered. Raises NagOSError if argument `fileobj` is not a valid file-object instance. Raises NagOSError if argument `fileobj` does not use a file descriptor. """ self._check_valid_fo_and_fd(fileobj, locus) return self._fileno_to_unit(fileobj.fileno(), check_range=check_range)
[docs] def register(self, fileobjs): """ Register the supplied file objects with this manager instance. Parameters ---------- fileobjs : tuple The file objects to register. Raises ------ NagTypeError `fileobjs` is not an instance of ``tuple``. NagOSError An element of the `fileobjs` ``tuple`` is not a valid file-object instance. An element of the `fileobjs` ``tuple`` does not use a file descriptor. """ locus = { 'fun_name': '.'.join([_MODULE_NAME, 'FileObjManager.register']), 'entity': 'fileobjs' } self._check_valid_tuple(fileobjs, locus) for fileobj in fileobjs: unit_to_reg = self._fileobj_to_unit(fileobj, locus) self._register_to_unit(fileobj, unit_to_reg, locus)
[docs] def register_to_advunit(self, fileobj): """ Register the supplied file object with the NAG Library Engine advisory output unit (equivalent to stdout). Parameters ---------- fileobj : file object The file object to register. Raises ------ NagOSError `fileobj` is not a valid file-object instance. `fileobj` does not use a file descriptor. """ self._register_to_unit( fileobj, self.advunit, { 'fun_name': '.'.join( [_MODULE_NAME, 'FileObjManager.register_to_advunit'] ), 'entity': 'fileobj' }, )
[docs] def register_to_errunit(self, fileobj): """ Register the supplied file object with the NAG Library Engine error output unit (equivalent to stderr). Parameters ---------- fileobj : file object The file object to register. Raises ------ NagOSError `fileobj` is not a valid file-object instance. `fileobj` does not use a file descriptor. """ self._register_to_unit( fileobj, self.errunit, { 'fun_name': '.'.join( [_MODULE_NAME, 'FileObjManager.register_to_errunit'] ), 'entity': 'fileobj' }, )
def _register_to_unit(self, fileobj, unit, locus): """ Register the input file object with the input unit number. Raises NagOSError if argument `fileobj` is not a valid file-object instance. Raises NagOSError if argument `fileobj` does not use a file descriptor. """ self._check_valid_fo_and_fd(fileobj, locus) self._mappings[unit] = fileobj @classmethod def _set_default_units(cls): """ Set the default values for the advisory and error units. """ _EngineState._set_unit('err', cls._default_errunit) _EngineState._set_unit('adv', cls._default_advunit) @classmethod def _set_ni_units(cls): """ Set the modified values for the advisory and error units. """ _EngineState._set_unit('err', cls.errunit) _EngineState._set_unit('adv', cls.advunit)
[docs] def unit_from_fileobj(self, fileobj): """ Return a NAG Library Engine unit number for the input file object. Parameters ---------- fileobj : file object The file object to query. It is registered with this manager instance if it has not already been. Returns ------- int The NAG Engine unit number corresponding to `fileobj`. Raises ------ NagOSError `fileobj` is not a valid file-object instance. `fileobj` does not use a file descriptor. """ locus = { 'fun_name': '.'.join( [_MODULE_NAME, 'FileObjManager.unit_from_fileobj'] ), 'entity': 'fileobj' } unit = self._fileobj_to_unit(fileobj, locus) if unit not in self._mappings: self._register_to_unit(fileobj, unit, locus) return unit
#: The fallback I/O manager, used whenever the NAG Library Engine requires an #: I/O manager but one has not been explicitly supplied to the calling #: Python function. It has been constructed such that prefixing of #: output data with NAG Python function names is enabled. GLOBAL_IO_MANAGER = FileObjManager() def _py_print_rec( # pylint: disable=too-many-arguments,too-many-branches en_data, nout, rec, io_manager, excs, ifail, logger=None, fun_name=None ): """ Write the input string rec. If logger is None, write to Engine unit nout. Else write using logger.debug. Update the excs exception list with the tuple for any exception generated. Update ifail.value with the exit flag. If logger is None, the file object registered with nout is extracted from the input io_manager if this argument is not None, or from the global manager otherwise. If logger is None and no file object has been registered the object associated with the advisory unit is used instead. """ if io_manager is not None: this_io_manager = io_manager else: this_io_manager = GLOBAL_IO_MANAGER if logger is None: this_fo = this_io_manager._fileobj_from_unit(nout) if this_fo is None: this_fo = this_io_manager._fileobj_from_unit( this_io_manager.advunit ) writer = this_fo.write else: writer = logger.debug if ( logger is None and isinstance(this_fo, (_io.BufferedIOBase, _io.RawIOBase)) ): rec_to_write = rec if this_io_manager._locus_in_output: rec_to_write = b': '.join([fun_name.encode('ascii'), rec_to_write]) rec_to_write = rec_to_write.rstrip() + b'\n' elif ( logger is not None or not isinstance(this_fo, (_io.BufferedIOBase, _io.RawIOBase)) ): rec_to_write = _EngineCharScalarType.to_py(rec) if logger is None: if this_io_manager._locus_in_output: rec_to_write = ': '.join([fun_name, rec_to_write]) rec_to_write = rec_to_write.rstrip() + '\n' else: raise NagException( # pragma: no cover {'fun_name': '_py_print_rec'}, 'could not initialize record to write', ) try: writer(rec_to_write) except: # pylint: disable=bare-except _print_err( _sys.exc_info(), excs, en_data, ifail, ) else: ifail.value = 0 def _print_err(exc_info, excs, en_data, ifail): """ Trigger printing error. """ excs.append(exc_info) en_data.hlperr = FileObjManager._print_err ifail.value = 1 def _py_read_rec( # pylint: disable=too-many-arguments en_data, nin, rec, errbuf, rec_len, io_manager, excs, ifail ): """ Read a record from Engine unit nin. rec is a ctypes c_char * rec_len array and will be modified in place. errbuf is a ctypes c_char * 200 array and will be modified in place. Its contents are not referenced by the caller. Update the excs exception list with the tuple for any exception generated. Update ifail.value with the exit flag. The file object registered with nin is extracted from the input io_manager if this argument is not None, or from the global manager otherwise. If no file object has been registered with nin in a non-None local io_manager, the global manager will be queried instead. Raises NagOSError if no file object has been registered with nin. Raises NagValueError if the read record contains non-ASCII characters. """ locus = {'fun_name': '_py_read_rec'} errbuf_a = _ctypes.cast( errbuf, _ctypes.POINTER(_ctypes.c_char * 200) ).contents errbuf_a[:200] = b'\0' * 200 if io_manager is not None: this_io_manager = io_manager else: this_io_manager = GLOBAL_IO_MANAGER this_fo = this_io_manager._fileobj_from_unit(nin) if ( this_fo is None and io_manager is not None ): this_fo = GLOBAL_IO_MANAGER._fileobj_from_unit(nin) if this_fo is None: _read_err( ( None, NagOSError( locus, 'no file object registered with ' + 'NAG Library Engine I/O unit ' + str(nin) ) ), excs, FileObjManager._read_err_unassoc, en_data, ifail, ) return try: _rec = this_fo.readline() except: # pylint: disable=bare-except _read_err( _sys.exc_info(), excs, FileObjManager._read_err, en_data, ifail, ) return if not _rec: ifail.value = 1 return if not _is_ascii(_rec, locus): _read_err( ( None, NagValueError( locus, 'data read from NAG Library Engine I/O unit ' + str(nin) + ' must comprise only ASCII characters' ) ), excs, FileObjManager._read_err_nonascii, en_data, ifail, ) return if isinstance(_rec, str): _rec = _rec.encode('ascii') _rec = _rec.rstrip(b'\r\n') rec_a = _ctypes.cast( rec, _ctypes.POINTER(_ctypes.c_char * rec_len) ).contents minlen = min(rec_len, len(_rec)) rec_a[:minlen] = _rec[:minlen] rec_a[minlen:rec_len] = b' ' * (rec_len - minlen) ifail.value = 0 return def _read_err(exc_info, excs, err_val, en_data, ifail): """ Trigger (advancing) reading error. """ excs.append(exc_info) en_data.hlperr = err_val ifail.value = 1 def _py_nonadv_read_rec( # pylint: disable=too-many-arguments,too-many-statements en_data, nin, rec, rec_len, io_manager, excs, inform ): """ Read a record from Engine unit nin. Nonadvancing. rec is a ctypes c_char * rec_len array and will be modified in place. Update the excs exception list with the tuple for any exception generated. Update inform.value with the exit flag. Return no. chars read. The file object registered with nin is extracted from the input io_manager if this argument is not None, or from the global manager otherwise. If no file object has been registered with nin in a non-None local io_manager, the global manager will be queried instead. Raises NagOSError if no file object has been registered with nin. Raises NagValueError if the read record contains non-ASCII characters. """ locus = {'fun_name': '_py_nonadv_read_rec'} def read_chars(): """ Return a bytes string of length rec_len reading from this_fo as far as '\n'. """ the_string = [] if isinstance(this_fo, (_io.BufferedIOBase, _io.RawIOBase)): line_feed = b'\n' join_char = b'' else: line_feed = '\n' join_char = '' i = 0 while i < rec_len: one_char = this_fo.read(1) if not one_char: return b'', 2 if one_char == line_feed: break the_string.append(one_char) i += 1 strng = join_char.join(the_string) if not _is_ascii(strng, locus): return None, 3 if isinstance(this_fo, _io.TextIOBase): strng = strng.encode('ascii') strng = strng.rstrip(b'\r') if not strng: return strng, 1 strng = strng.ljust(rec_len) return strng, 0 if io_manager is not None: this_io_manager = io_manager else: this_io_manager = GLOBAL_IO_MANAGER this_fo = this_io_manager._fileobj_from_unit(nin) if ( this_fo is None and io_manager is not None ): this_fo = GLOBAL_IO_MANAGER._fileobj_from_unit(nin) if this_fo is None: return _nonadv_read_err( ( None, NagOSError( locus, 'no file object registered with ' + 'NAG Library Engine I/O unit ' + str(nin) ) ), excs, FileObjManager._read_err_unassoc, en_data, inform, ) try: _rec, inform.value = read_chars() except: # pylint: disable=bare-except return _nonadv_read_err( _sys.exc_info(), excs, FileObjManager._read_err, en_data, inform, ) if _rec is None: return _nonadv_read_err( ( None, NagValueError( locus, 'data read from NAG Library Engine I/O unit ' + str(nin) + ' must comprise only ASCII characters' ) ), excs, FileObjManager._read_err_nonascii, en_data, inform, ) readlen = len(_rec.rstrip()) if not _rec: return readlen rec_a = _ctypes.cast( rec, _ctypes.POINTER(_ctypes.c_char * rec_len) ).contents rec_a[:readlen] = _rec[:readlen] rec_a[readlen:rec_len] = b' ' * (rec_len - readlen) if readlen < rec_len: inform.value = 1 return readlen def _nonadv_read_err(exc_info, excs, err_val, en_data, inform): """ Trigger nonadvancing reading error. """ excs.append(exc_info) en_data.hlperr = err_val inform.value = 3 readlen = 0 return readlen def _handle_cb_exception(excs): """ Reraise an exception stored in the excs list. """ for exc in excs: if exc[0] is not UserCallbackTerminate: _raise_exc_info(exc) def _raise_exc_info(exc_info): """ (Re)raise a sys.exc_info-style exception. """ raise exc_info[1].with_traceback(exc_info[2]) def _violated_constraint_err_text(constraint): """ Standard text for a violated input-constraint error. """ return ' '.join( ['On entry, the following constraint was not satisfied:', constraint] ) _lapack_constraint_err_text = _violated_constraint_err_text # pylint: disable=invalid-name def _raise_warning(warn_class, msg_lines, locus): """ Raises warn_class with formatted `msg_lines`. """ _warnings.warn( warn_class( locus, _format_msg_lines(msg_lines), ), stacklevel=_get_user_stack_level(), ) _STR_FMT_STR = '{:s}' _STR_FMT_INT_AS_STR = '{:s}' _STR_FMT_INT = '{:d}' _STR_FMT_FLOAT_AS_STR = '{:s}' _STR_FMT_FLOAT = '{:22.15e}' def _withdrawal(fun_name, mkwdrn, advice=None): """ Decorate with a deprecation warning if called from outside the package. """ def decorate(func): # pylint: disable=missing-docstring @_functools.wraps(func) def wrapper(*args, **kwargs): # pylint: disable=missing-docstring caller_dirname = _os.path.dirname(_inspect.stack()[1][1]) caller_parent_dirbase = _os.path.split( _os.path.split(caller_dirname)[0] )[-1] if caller_parent_dirbase != _PKG_NAME: msg_lines = [] if mkwdrn is None: msg_lines.append( 'This function is deprecated.' ) warn_cls = NagDeprecatedWarning else: msg_lines.append( 'This function is scheduled for withdrawal, ' 'no earlier than Mark ' + mkwdrn + '.', ) warn_cls = NagWithdrawalWarning if advice is None: msg_lines.append( 'There is no replacement for this routine.' ) else: msg_lines.append( 'The following advice is given for making a ' 'replacement:' ) msg_lines.extend(advice) _warnings.warn( warn_cls( {'fun_name': fun_name}, '\n' + '\n'.join(msg_lines), ), stacklevel=2 ) result = func(*args, **kwargs) return result return wrapper return decorate @_withdrawal('.'.join([_MODULE_NAME, '_withdrawal_test']), '42') def _withdrawal_test(): """ Test of withdrawal decorator. """ @_withdrawal('.'.join([_MODULE_NAME, '_withdrawal_warn_test']), '42') def _withdrawal_warn_test(): """ Test of withdrawal decorator on a function that raises a warning. """ _EngineWarningExit( fun_name='_withdrawal_warn_test', msg_lines=['spam', 'eggs'], errcodes=(43, 44, 45), ).manage() @_withdrawal( '.'.join([_MODULE_NAME, '_withdrawal_test']), None, ['Replacement base advice.'] ) def _deprecation_test(): """ Test of withdrawal decorator. """ def _get_user_stack_level(): """ Return the frame number for the first non-naginterfaces stack frame. """ stack_frames = _inspect.stack() for frame_i, frame in enumerate(stack_frames): frame_dirname = _os.path.dirname(frame[1]) frame_parent_dirbase = _os.path.split( _os.path.split(frame_dirname)[0] )[-1] if frame_parent_dirbase != _PKG_NAME: return frame_i raise NagException( # pragma: no cover {'fun_name': '_get_user_stack_level'}, 'inconceivable stack' ) def _check_comm(comm, locus, req_keys=(), opt_keys=(), allow_none=False): """ Validate the comm blob. Return True if opt_keys was non-empty and none were present in comm. Raises NagTypeError if argument comm is not a dict. Raises NagKeyError if the dictionary keys in comm are incorrect for the communication structure it is supposed to represent. """ # pylint: disable=too-many-branches if allow_none and comm is None: return True if not isinstance(comm, dict): raise NagTypeError( locus, 'must be an instance of dict' ) if not comm: if req_keys: raise NagKeyError( locus, 'must not be empty. ' 'You must pass a communication structure that was returned by ' 'the initializer for this routine (which may be this routine ' 'itself)' ) if opt_keys: return True raise NagException( locus, 'communication structure seems totally redundant', ) for key in req_keys: if key not in comm.keys(): raise NagKeyError( locus, 'invalid communication structure for this function: ' + 'key \'' + key + '\' must be present. ' 'You must pass a communication structure that was returned by ' 'the initializer for this routine (which may be this routine ' 'itself)' ) if not opt_keys: if not req_keys: raise NagException( locus, 'communication structure seems totally redundant', ) return False n_opt_keys_present = len([key for key in opt_keys if key in comm.keys()]) if n_opt_keys_present == 0: return True if len(opt_keys) == n_opt_keys_present: return False for key in opt_keys: if key not in comm.keys(): raise NagKeyError( locus, 'invalid communication structure for this function: ' + 'key \'' + key + '\' must be present' ) raise NagException( # pragma: no cover locus, 'inconceivable exit from _check_comm', ) def _check_callable(cble, locus, allow_none=False): """ Validate the callable `cble`. Raises NagTypeError if argument `cble` is not callable, and not None if None is permitted. """ if ( callable(cble) or ( allow_none and cble is None ) ): return msg = 'must be ' if allow_none: msg += 'None or ' msg += 'callable' raise NagTypeError( locus, msg, ) def _d01mazn(handle): """ Query handle and extract problem dimensions. """ raise NotImplementedError def _e04ptan(handle): """ Query handle and extract problem dimensions. """ fun_name = '.'.join([_MODULE_NAME, '_e04ptan']) _handle = _EngineCptrScalarType( handle, {'fun_name': fun_name, 'entity': 'handle'}, ).to_ctypes() _nvar = _EngineIntScalarType.empty_ctypes() _nnzu = _EngineIntScalarType.empty_ctypes() _nnzuc = _EngineIntScalarType.empty_ctypes() _nnzua = _EngineIntScalarType.empty_ctypes() _ierr = _EngineIntScalarType.empty_ctypes() engine_ctypes = _EngineState._get_symbol('e04ptan') engine_ctypes.restype = None engine_ctypes.argtypes = ( _BYREF_HANDLE, # Input, handle. _BYREF_INT, # Output, nvar. _BYREF_INT, # Output, nnzu. _BYREF_INT, # Output, nnzuc. _BYREF_INT, # Output, nnzua. _BYREF_INT, # Output, ierr. ) engine_ctypes( _handle, _nvar, _nnzu, _nnzuc, _nnzua, _ierr, ) _ierr_py = _EngineIntScalarType.to_py(_ierr) if _ierr_py != _IERR_SUCCESS: _EngineErrorExit( fun_name=fun_name, msg_lines=[ 'handle does not belong to the NAG optimization modelling ' 'suite,', 'has not been initialized properly or is corrupted.', ], errcodes=(_ierr_py, _ierr_py, _ierr_py), ).manage() return _nvar, _nnzu, _nnzuc, _nnzua def _f08mdan(call_vendor, n): # pylint: disable=invalid-name """ Return the required size of Q and IQ in F08MDF. """ # pylint: disable=invalid-name fun_name = '.'.join([_MODULE_NAME, '_f08mdan']) _n = _EngineIntScalarType( n, {'fun_name': fun_name, 'entity': 'n'}, ).to_ctypes() _lq = _EngineIntScalarType.empty_ctypes() _liq = _EngineIntScalarType.empty_ctypes() engine_ctypes = _EngineState._get_symbol('f08mdan') engine_ctypes.restype = None engine_ctypes.argtypes = ( _BYREF_LOGICAL, # Input, call_vendor. _BYREF_INT, # Input, n. _BYREF_INT, # Output lq. _BYREF_INT, # Output liq. ) _call_vendor = _EngineLogicalScalarType( call_vendor, {'fun_name': fun_name, 'entity': 'call_vendor'}, ).to_ctypes() engine_ctypes( _call_vendor, _n, _lq, _liq, ) return (_lq, _liq) def _f12jzyn(handle, irevcm, m0_in): """ Query handle and extract problem dimensions. """ fun_name = '.'.join([_MODULE_NAME, '_f12jzyn']) _handle = _EngineCptrScalarType( handle, {'fun_name': fun_name, 'entity': 'handle'}, ).to_ctypes() _irevcm = _EngineIntScalarType( irevcm, {'fun_name': fun_name, 'entity': 'irevcm'}, ).to_ctypes() _m0_in = _EngineIntScalarType( m0_in, {'fun_name': fun_name, 'entity': 'm0_in'}, ).to_ctypes() _m0_out = _EngineIntScalarType.empty_ctypes() engine_ctypes = _EngineState._get_symbol('f12jzyn') engine_ctypes.restype = None engine_ctypes.argtypes = ( _BYREF_HANDLE, # Input, handle. _BYREF_INT, # Input, irevcm. _BYREF_INT, # Input, m0_in. _BYREF_INT, # Output, m0_out. ) engine_ctypes( _handle, _irevcm, _m0_in, _m0_out, ) return _m0_out def _g02zkbn(init, routine_name, ip, ntau, comm, statecomm): # pylint: disable=invalid-name,too-many-arguments,too-many-locals """ Extract size of optional arrays from comm. """ fun_name = '.'.join([_MODULE_NAME, '_g02zkbn']) engine_ctypes = _EngineState._get_symbol('g02zkbn') engine_ctypes.restype = None engine_ctypes.argtypes = ( _BYREF_INT, # Input, init. _BYREF_STRING, # Input, routine_name(len=6). _BYREF_INT, # Input, ip. _BYREF_INT, # Input, ntau. _EngineIntArrayType.c_type, # Input, iopts[*]. _EngineIntArrayType.c_type, # Input, state[*]. _BYREF_INT, # Output, sdres. _BYREF_INT, # Output, rip. _BYREF_INT, # Output, tdch. _BYREF_INT, # Output, liopts. _BYREF_INT, # Output, lopts. _BYREF_INT, # Output, lstate. _BYREF_INT, # Output, calcib. _EngineState._chrlen_ctype, # Input, hidden chrlen of routine_name. ) _init = _EngineIntScalarType( init, {'fun_name': fun_name, 'entity': 'init'}, ).to_ctypes() _init_py = _EngineIntScalarType.to_py(_init) if _init_py == 0: _routine_name = _EngineCharScalarType( routine_name, {'fun_name': fun_name, 'entity': 'routine_name'}, exp_chrlen=6, ).to_ctypes() else: _routine_name = _EngineRoutineName(*b'unknow') _ip = _EngineIntScalarType( ip, {'fun_name': fun_name, 'entity': 'ip'}, ).to_ctypes() _ntau = _EngineIntScalarType( ntau, {'fun_name': fun_name, 'entity': 'ntau'}, ).to_ctypes() if comm is None: _iopts = None else: _iopts = comm['iopts'] if statecomm is None: _state = _EngineIntArrayType.empty_ctypes(1) else: _state = statecomm['state'] _lstate_py = _EngineIntArrayType.get_shape( _state, {'fun_name': fun_name, 'entity': '_state'}, exp_rank=1, allow_none=True, validate=False, )[0] _sdres = _EngineIntScalarType.empty_ctypes() _rip = _EngineIntScalarType.empty_ctypes() _tdch = _EngineIntScalarType.empty_ctypes() _liopts = _EngineIntScalarType.empty_ctypes() _lopts = _EngineIntScalarType.empty_ctypes() _lstate = _EngineIntScalarType.empty_ctypes() _calcib = _EngineIntScalarType.empty_ctypes() engine_ctypes( _init, _routine_name, _ip, _ntau, _iopts, _state, _sdres, _rip, _tdch, _liopts, _lopts, _lstate, _calcib, _routine_name._length_, ) return ( _sdres, _rip, _tdch, _liopts, _lopts, _lstate, _calcib ) #: A named constant for use with certain functions in the ``blast`` module. NAG_ONE_NORM = 171 #: A named constant for use with certain functions in the ``blast`` module. NAG_TWO_NORM = 173 #: A named constant for use with certain functions in the ``blast`` module. NAG_FROBENIUS_NORM = 174 #: A named constant for use with certain functions in the ``blast`` module. NAG_INF_NORM = 175 #: A named constant for use with certain functions in the ``blast`` module. NAG_MAX_NORM = 177