# -*- python -*-
# -*- coding: utf8 -*-
#
# OpenAlea.OALab
#
# Copyright 2014 INRIA - CIRAD - INRA
#
# File author(s): Julien Coste <julien.coste@inria.fr>
# Guillaume Baty <guillaume.baty@inria.fr>
#
# File contributor(s):
#
# Distributed under the Cecill-C License.
# See accompanying file LICENSE.txt or copy at
# http://www.cecill.info/licences/Licence_CeCILL-C_V1-en.html
#
# OpenAlea WebSite : http://openalea.gforge.inria.fr
#
###############################################################################
"""
A Model is an object that can be run.
It is generally used to define algorithms or process.
A model has inputs, outputs and an internal state.
Inputs and outputs are used to allow communication between environnement and models and between models.
Internal state is used by the model itself but cannot be reached from outside.
This internal step is generally used when user want to execute model step by step,
each step depending on previous one.
To define model instructions, you need to define at least "step" instructions.
To do that, you can use:
- :meth:`~IModel.set_step_code`
To use a Model, you also need to define set_code method that is able to extract inputs, outpus and step, init, ...
from given source code.
Animate notification are not yet implemented.
To reset internal state, use :meth:`~openalea.core.model.IModel.init` method.
To run one step, use :meth:`~openalea.core.model.IModel.step` method.
Two convenience methods, see :meth:`~openalea.core.model.IModel.run` and "call"
can be used to reset internal state and run a given number of step (one by default)
By default, functions are generated for "init", "run" and "animate"
"""
from ast import literal_eval
from copy import copy
[docs]class IModel(object):
dtype = None
mimetype = None
def __init__(self, **kwds):
"""
keywords:
- name: model name
"""
def __call__(self, *args, **kwargs):
"""
Equivalent to :meth:`~openalea.core.model.IModel.run`.
"""
[docs] def run(self, *args, **kwargs):
"""
Init model and run "nstep" step(s).
"""
raise NotImplementedError
[docs] def init(self, *args, **kwargs):
"""
Reset internal state, fill namespace and run "init code" if defined.
"""
raise NotImplementedError
[docs] def step(self, *args, **kwargs):
"""
run "step code" if defined.
"""
raise NotImplementedError
[docs] def stop(self, *args, **kwargs):
"""
Stop execution.
"""
raise NotImplementedError
[docs] def animate(self, *args, **kwargs):
"""
Like run but send notification at each step.
"""
raise NotImplementedError
[docs] def set_code(self, code):
"""
extract and set inputs, outputs and functions from code.
For example, if "code" is::
'''
input = a, b:int=5
output = c
'''
def step():
c = a+b
you can extract input, output and step function.
So,
m = Model("m1")
m.set_code(code)
is equivalent to
m = Model("m1")
m.set_func_code("step", "c=a+b")
m.inputs_info = [InputObj('a'), InputObj('b:int=5')]
m.outputs_info = [OutputObj('c')]
"""
[docs] def set_func_code(self, fname, code):
"""
:param fname: function name: "step", "init", "animate", ...
:param code: python source code
"""
[docs] def set_step_code(self, code):
"""
convenience method equivalent to set_func_code("step", code)
"""
[docs]class Model(object):
icon = ''
def __init__(self, name=None, **kwds):
from openalea.core.service.ipython import interpreter
self.interp = interpreter()
self.inputs_info = []
self.outputs_info = []
self.name = name
self._ns = {}
self._code = {}
self._initial_code = ''
self.outputs = []
code = kwds.pop('code', None)
if code:
self.set_code(code)
def __copy__(self):
m = self.__class__(name=self.name)
m.inputs_info = list(self.inputs_info)
m.outputs_info = list(self.outputs_info)
for fname, code in self._code.iteritems():
m.set_func_code(fname, code)
return m
def _run_code(self, code):
if isinstance(code, basestring):
self.interp.run_cell(code)
else:
self.interp.run_code(code)
[docs] def set_code(self, code):
self.set_step_code(code)
[docs] def set_func_code(self, fname, code):
self._code[fname] = code
[docs] def set_step_code(self, code):
self.set_func_code('step', code)
def _push_ns(self):
# Save default namespace
self._old_ns = copy(self.interp.user_ns)
def _fill_namespace(self, *args, **kwds):
# Create a new namespace with
# - interpreter namespace
# - initial namespace given by user (namespace keyword)
# - passed variables
# Then, replace input variable names with right values
initial_ns = kwds.pop('namespace', {})
global_ns = {}
global_ns.update(self._old_ns)
global_ns.update(initial_ns)
global_ns.update(kwds)
global_ns['this'] = self
kwargs = self.inputs_from_ns(self.inputs_info, global_ns, *args, **kwds)
global_ns.update(kwargs)
self._ns = global_ns
self.interp.user_ns.clear()
self.interp.user_ns.update(self._ns)
def _populate_ns(self):
# add vars defined in init function
self._ns.update(self.interp.user_ns)
#for k in self.interp.user_ns:
# if k not in self._ns:
# self._ns[k] = self.interp.user_ns[k]
def _pop_ns(self):
# Restore original namespace
self.interp.user_ns.clear()
self.interp.user_ns.update(self._old_ns)
[docs] def init(self, *args, **kwds):
self._push_ns()
self._fill_namespace(*args, **kwds)
# Run init code
if 'init' in self._code:
self._run_code(self._code['init'])
self._populate_ns()
self._pop_ns()
return self.output_from_ns(self._ns)
def __call__(self, *args, **kwds):
return self.run(*args, **kwds)
def _exec(self, fname='step'):
# Save namespace
old_ns = {}
old_ns.update(self.interp.user_ns)
self.interp.user_ns.clear()
self.interp.user_ns.update(self._ns)
# Run code
if fname in self._code:
self._run_code(self._code[fname])
outputs = self.output_from_ns(self.interp.user_ns)
self.interp.user_ns.clear()
self.interp.user_ns.update(old_ns)
self.outputs = outputs
return outputs
[docs] def run_code(self, code, namespace):
# Save namespace
ns = {}
ns.update(namespace)
old_ns = self.interp.user_ns
self.interp.user_ns = ns
self._run_code(code)
self.interp.user_ns = old_ns
final_ns = {}
for key in ns:
if key in old_ns:
continue
final_ns[key] = ns[key]
return final_ns
[docs] def execute(self, code):
self.set_func_code('selection', code)
outputs = self._exec('selection')
del self._code['selection']
return outputs
[docs] def step(self, *args, **kwds):
return self._exec('step')
[docs] def animate(self, *args, **kwds):
if 'animate' in self._code:
return self._exec('animate')
else:
nstep = kwds.pop('nstep', 1)
self.init(*args, **kwds)
out = []
for i in range(nstep):
out = self.step()
# refresh world
return out
[docs] def run(self, *args, **kwds):
if 'run' in self._code:
rvalues = self._exec('run')
else:
nstep = kwds.pop('nstep', 1)
self.init(*args, **kwds)
out = []
for i in range(nstep):
out = self.step()
rvalues = out
if kwds.get('run_in_shell', False):
self.interp.user_ns.update(self._ns)
return rvalues
[docs] def eval_value(self, value):
return literal_eval(value)
[docs] def output_from_ns(self, namespace):
"""
Get outputs from namespace and set them inside self.outputs
:param namespace: dict where the model will search the outputs
"""
outputs = []
if self.outputs_info:
if len(self.outputs_info) > 0:
for outp in self.outputs_info:
if outp.name in namespace:
outputs.append(namespace[outp.name])
self._ns[outp.name] = namespace[outp.name]
if len(outputs) == 0:
return None
elif len(outputs) == 1:
return outputs[0]
else:
return outputs
[docs] def get_documentation(self):
"""
:return: a string with the documentation of the model
"""
return ''
@property
def step_code(self):
return self._code['step']
@step_code.setter
[docs] def step_code(self, code):
self._code['step'] = code
def _set_code(self, code):
self.set_code(code)
[docs] def repr_code(self):
raise NotImplementedError
code = property(fset=_set_code)
[docs]class PythonModel(Model):
dtype = 'Python'
mimetype = 'text/x-python'
default_name = 'Python'
def __init__(self, **kwargs):
Model.__init__(self, **kwargs)
[docs] def get_documentation(self):
"""
:return: a string with the documentation of the model
"""
try:
return self._doc
except AttributeError:
return ''
[docs] def repr_code(self):
try:
return self._initial_code
except AttributeError:
code = '"""\n'
if self.inputs_info:
code += 'input = %s\n' % (', '.join([inp.repr_code() for inp in self.inputs_info]))
if self.outputs_info:
code += 'output = %s\n' % (', '.join([out.repr_code() for out in self.outputs_info]))
code += '"""\n'
if 'init' in self._code:
code += self._code['init'] + '\n'
for fname in ['step', 'run', 'animate', 'stop']:
if fname in self._code:
code += 'def %s():\n' % fname
for l in self._code[fname].split('\n'):
code += ' ' + l
code += '\n'
return code
[docs] def set_code(self, code):
from openalea.oalab.model.parse import parse_docstring, get_docstring, extract_functions
self._initial_code = code
model, self.inputs_info, self.outputs_info = parse_docstring(code)
funcs = extract_functions(code)
self.set_func_code('init', code)
for fname in ['step', 'run', 'animate']:
if fname in funcs:
self.set_func_code(fname, funcs[fname])
self._doc = get_docstring(code)