import os
import shutil
from yggdrasil.drivers.ModelDriver import ModelDriver
_top_dir = os.path.normpath(os.path.join(os.path.dirname(__file__), '../'))
[docs]class InterpretedModelDriver(ModelDriver):
r"""Base class for models written in interpreted languages.
Args:
name (str): Driver name.
args (str or list): Argument(s) for running the model on the command
line. This driver will check to see if there is an interpreter
included in the args provided. If not, one will be added unless
skip_interpreter is True.
interpreter (str, optional): Name or path of interpreter executable that
should be used to run the model. If not provided, the interpreter
will be determined based on configuration options for the language
(if present) and the default_interpreter class attribute.
interpreter_flags (list, optional): Flags that should be passed to the
interpreter when running the model. If not provided, the flags are
determined based on configuration options for the language (if
present) and the default_interpreter_flags class attribute.
skip_interpreter (bool, optional): If True, no interpreter will be
added to the arguments. This should only be used for subclasses
that will not be invoking the model via the command line.
Defaults to False.
**kwargs: Additional keyword arguments are passed to the parent class.
Class Attributes:
default_interpreter (str): Name of interpreter that will be used if not
set explicitly by instance or config file. Defaults to the language
name if not set.
default_interpreter_flags (list): Flags that will be passed to the
interpreter when running the model by default if not set explicitly
by instance or config file.
Attributes:
interpreter (str): Name or path to the interpreter that will be used.
interpreter_flags (list): Flags that will be passed to the interpreter
when running a model.
path_env_variable (str): Name of the environment variable containing
path information for the interpreter for this language.
paths_to_add (list): Paths that should be added to the path_env_variable
for this language on the process the model is run in.
comm_atexit (function): Function taking a comm instance as input that
performs any necessary operations during exit. If None, no additional
actions are taken.
comm_linger (bool): If True, interface comms will linger during close.
This should only be required if the language will disreguard Python
threads at exit (e.g. when using a Matlab engine).
decode_format (function: Function decoding format string created in this
language. If None, no additional actions are taken.
recv_converters (dict): Mapping between the names of message types (e.g.
'array', 'pandas') and functions that should be used to prepare such
objects for return when they are received.
send_converters (dict): Mapping between the names of message types (e.g.
'array', 'pandas') and functions that should be used to prepare such
objects for sending.
"""
_schema_properties = {
'interpreter': {'type': 'string'},
'interpreter_flags': {'type': 'array', 'items': {'type': 'string'},
'default': []},
'skip_interpreter': {'type': 'boolean', 'default': False}}
executable_type = 'interpreter'
default_interpreter = None
default_interpreter_flags = []
path_env_variable = None
paths_to_add = [_top_dir]
comm_atexit = None
comm_linger = False
decode_format = None
recv_converters = {}
send_converters = {}
_config_attr_map = [{'attr': 'default_interpreter',
'key': 'interpreter'},
{'attr': 'default_interpreter_flags',
'key': 'interpreter_flags',
'type': list}]
# def __init__(self, name, args, **kwargs):
# super(InterpretedModelDriver, self).__init__(name, args, **kwargs)
# # Set defaults from attributes
# for k0 in ['interpreter']:
# for k in [k0, '%s_flags' % k0]:
# v = getattr(self, k, None)
# if v is None:
# setattr(self, k, getattr(self, 'default_%s' % k))
[docs] @staticmethod
def after_registration(cls, **kwargs):
r"""Operations that should be performed to modify class attributes after
registration. For compiled languages this includes selecting the
default compiler. The order of precedence is the config file 'compiler'
option for the language, followed by the environment variable set by
_compiler_env, followed by the existing class attribute.
"""
ModelDriver.after_registration(cls, **kwargs)
if kwargs.get('second_pass', False):
return
if cls.language is not None:
# Set default interpreter based on language
if cls.default_interpreter is None:
cls.default_interpreter = cls.language
# Add directory containing the interface
try:
cls.paths_to_add.append(cls.get_language_dir())
except ValueError:
pass
[docs] @classmethod
def are_dependencies_installed(cls, **kwargs):
r"""Determine if the dependencies are installed for the interface (not
including dependencies needed by a particular communication type).
Returns:
bool: True if the dependencies are installed. False otherwise.
"""
# Short cut by checking if yggdrasil installed
if not (cls.full_language and cls.is_interface_installed()): # pragma: config
return super(InterpretedModelDriver, cls).are_dependencies_installed(**kwargs)
return True
[docs] def parse_arguments(self, *args, **kwargs):
r"""Sort model arguments to determine which one is the executable
and which ones are arguments.
Args:
*args: Arguments are passed to the parent class's method.
**kwargs: Keyword arguments are passed to the parent class's method.
"""
super(InterpretedModelDriver, self).parse_arguments(*args, **kwargs)
self.model_src = self.model_file
[docs] @classmethod
def get_interpreter(cls):
r"""Command required to run a model written in this language from
the command line.
Returns:
str: Name of (or path to) interpreter executable.
"""
out = getattr(cls, 'interpreter', getattr(cls, 'default_interpreter'))
if out is None:
raise NotImplementedError("Interpreter not set for language '%s'."
% cls.language)
if not os.path.isfile(out):
out_full = shutil.which(out)
if out_full:
out = out_full
return out
[docs] @classmethod
def get_interpreter_flags(cls):
r"""Get the flags that should be passed to the interpreter when using it
to run a model on the command line.
Returns:
list: Flags that should be passed to the interpreter on the command
line.
"""
out = getattr(cls, 'interpreter_flags',
getattr(cls, 'default_interpreter_flags'))
return out
[docs] @classmethod
def language_executable(cls):
r"""Command required to compile/run a model written in this language
from the command line.
Returns:
str: Name of (or path to) compiler/interpreter executable required
to run the compiler/interpreter from the command line.
"""
return cls.get_interpreter()
[docs] def run_model(self, *args, **kwargs):
r"""Run the model. Unless overridden, the model will be run using
run_executable.
Args:
*args: Arguments are passed to the parent class's method.
**kwargs: Keyword arguments are passed to the parent class's
method.
"""
if self.interpreter:
kwargs.setdefault('interpreter', self.interpreter)
if self.interpreter_flags:
kwargs.setdefault('interpreter_flags', self.interpreter_flags)
return super(InterpretedModelDriver, self).run_model(*args, **kwargs)
[docs] @classmethod
def executable_command(cls, args, exec_type='interpreter', unused_kwargs=None,
skip_interpreter_flags=False, interpreter=None,
interpreter_flags=None, **kwargs):
r"""Compose a command for running a program in this language with the
provied arguments. If not already present, the interpreter command and
interpreter flags are prepended to the provided arguments.
Args:
args (list): The program that returned command should run and any
arguments that should be provided to it.
exec_type (str, optional): Type of executable command that will be
returned. If 'interpreter', a command using the interpreter is
returned and if 'direct', the raw args being provided are
returned. Defaults to 'interpreter'.
skip_interpreter_flags (bool, optional): If True, interpreter flags
will not be added to the command after the interpreter. Defaults
to False. Interpreter flags will not be added, reguardless of
this keyword, if the first element of args is already an
interpreter.
unused_kwargs (dict, optional): Existing dictionary that unused
keyword arguments should be added to. Defaults to {}.
**kwargs: Additional keyword arguments are ignored.
Returns:
list: Arguments composing the command required to run the program
from the command line using the interpreter for this language.
Raises:
ValueError: If exec_type is not 'interpreter' or 'direct'.
"""
ext = cls.language_ext
assert isinstance(ext, (tuple, list))
if exec_type == 'interpreter':
if not cls.is_interpreter(args[0]):
if interpreter is None:
interpreter = cls.get_interpreter()
new_args = [interpreter]
if not skip_interpreter_flags:
new_args += cls.get_interpreter_flags()
if interpreter_flags:
new_args += interpreter_flags
args = new_args + args
elif exec_type != 'direct':
raise ValueError("Invalid exec_type '%s'" % exec_type)
if isinstance(unused_kwargs, dict):
unused_kwargs.update(kwargs)
return args
[docs] @classmethod
def is_interpreter(cls, cmd):
r"""Determine if a command line argument is an interpreter.
Args:
cmd (str): Command that should be checked.
Returns:
bool: True if the command is an interpreter, False otherwise.
"""
# (cls.language not in cmd)
out = ((shutil.which(cmd) is not None)
and (not any([cmd.endswith(e) for e in cls.language_ext])))
return out
[docs] def set_env(self, **kwargs):
r"""Get environment variables that should be set for the model process.
Returns:
dict: Environment variables for the model process.
"""
out = super(InterpretedModelDriver, self).set_env(**kwargs)
if self.path_env_variable is not None: # pragma: debug
if self.language != 'matlab':
raise NotImplementedError(
("Language %s sets path_env_variable. "
"Move part of MatlabModelDriver set_env method "
"to InterpretedModelDriver in place of this "
"warning message.") % self.language)
return out
# Methods for handling type conversions
[docs] @classmethod
def python2language(cls, pyobj):
r"""Prepare a python object for transformation in target
language.
Args:
pyobj (object): Python object.
Returns:
object: Python object in a form that is friendly to the
target language.
"""
return pyobj
[docs] @classmethod
def language2python(cls, pyobj):
r"""Prepare an object from the target language for receipt
in Python.
Args:
pyobj (object): Python object transformed from the target
language.
Returns:
object: Python object in a form that conforms with the
expected Python type.
"""
return pyobj