.. _modelchain-py:
#############
modelchain.py
#############
*******************************
class ModelChainDef(XmlObject):
*******************************
Model chain definitions.
def __init__(self, typedef):
============================
Initializes the model chain collection by processing the type definition schema::
>>> from simo.builder.modelchain.modelchain import ModelChainDef
>>> from simo.builder import names
>>> from minimock import Mock
>>> tdf = open('../../simulator/xml/schemas/Typedefs_SIMO.xsd')
>>> typedef = tdf.read()
>>> tdf.close()
>>> sf = open('../../simulator/xml/schemas/model_chain.xsd')
>>> schema = sf.read()
>>> sf.close()
>>> xml = u'''
...
...
...
...
... Thinning_limits_JKK
...
...
...
... THINNING_LEVEL
... 0.5
...
...
...
...
...
...
...
...
...
...
... invalid
...
... JUST_THINNED
...
...
... 0
...
...
...
...
...
...
...
... assign_value
...
... JUST_THINNED
...
...
... 0
...
...
...
...
...
...
...
...
...
...
... comp_unit:MAIN_SP in [1, 8] and
... comp_unit:NF eq 0
...
...
... comp_unit:MAIN_SP in [1, 8] and
... comp_unit:SOME eq 0
...
... clearcut
...
...
... false
...
...
... timber_prices
...
...
...
...
...
...
... comp_unit:open_area eq 0
...
...
...
... comp_unit:passed_regen_limit gt 0
...
...
... comp_unit:passed_regen_limit gt 5
...
...
... comp_unit:passed_regen_limit gt 10
...
...
... comp_unit:MAIN_GROUP eq 1 and
... comp_unit:REGENERABLE eq 1
...
... comp_unit:PEAT eq 0
...
...
... clearcut2
...
...
... false
...
...
... timber_prices
...
...
...
...
...
...
... zero_value
...
... REGENERABLE
...
...
... REGENERABLE
... comp_unit
...
...
...
...
...
...
...
...
...
...
... '''
>>> class Lexicon(object):
... def get_variable_ind(self, level, variable, active=False):
... if variable == 'MAIN_SP':
... return (1,1)
... elif variable == 'NF':
... return (1,2)
... if variable == 'REGENERABLE':
... return (1,3)
... elif variable == 'MAIN_GROUP':
... return (1,4)
... elif variable == 'passed_regen_limit':
... return (1,5)
... elif variable == 'PEAT':
... return (1,6)
... else:
... return (None, None)
... def get_level_ind(self, level):
... return 1
... def get_level_name(self, level):
... return 'mock level'
... def get_operation_ind(self, operation):
... return 3
... def is_child(self, level_1, level_2):
... return True
... models = {names.PREDICTION: ['Thinning_limits_JKK'],
... names.OPERATION: ['clearcut', 'clearcut2'],
... names.AGGREGATION: ['assign_value']}
>>> m = Mock('Model')
>>> m.__deepcopy__.mock_returns = m
>>> pm = Mock('PredictionModel')
>>> pm.name = 'mock model'
>>> pm.param_names = ['test']
>>> pm.param_limits = [None]
>>> pm.n_params = 1
>>> pm.__deepcopy__.mock_returns = pm
>>> cf = Mock('Classifier')
>>> cf.__deepcopy__.mock_returns = cf
>>> om1 = Mock('OperationModel')
>>> om1.result_classifiers = {('SP', 'assortment'): [cf, cf]}
>>> om1.result_type = 'operation_result'
>>> om1.result_variables = [('operation_result', 'Volume', None, 'scalar', True), ('operation_result', 'Income', None, 'scalar', True)]
>>> om1.n_params = 0
>>> om1.parameter_tables = None
>>> om1.__deepcopy__.mock_returns = om1
>>> om2 = Mock('OperationModel')
>>> om2.result_classifiers = {('SP', 'something else'): [cf, cf]}
>>> om2.result_type = 'operation_result'
>>> om2.result_variables = [('operation_result', 'Volume', None, 'scalar', True), ('operation_result', 'Income', None, 'scalar', True)]
>>> om2.n_params = 0
>>> om2.parameter_tables = None
>>> om2.__deepcopy__.mock_returns = om2
>>> om3 = Mock('OperationModel')
>>> om3.result_classifiers = {}
>>> om3.result_type = 'data'
>>> om3.result_variables = [('operation_result', 'dummy_1', 1, 'scalar', True), ('operation_result', 'dummy_2', 2, 'scalar', True)]
>>> om3.n_params = 0
>>> om3.parameter_tables = None
>>> om3.__deepcopy__.mock_returns = om3
>>> cft = Mock('CashFlowTable')
>>> cft.classifier_order = []
>>> cft.keycount=0
>>> cft.__deepcopy__.mock_returns = cft
>>> clf = Mock('Classifier')
>>> clf.persist = True
>>> clf.level = 2
>>> clf.__deepcopy__.mock_returns = clf
>>> clf.ind = 2
>>> cft.classifiers = {'SP':clf}
>>> cft.db_cfiers = set(['SP'])
>>> cft.__deepcopy__.mock_returns = cft
>>> mi = {names.PREDICTION: {'Thinning_limits_JKK': pm},
... names.OPERATION: {'clearcut': om1, 'clearcut2': om2,
... 'planting': om3},
... names.AGGREGATION: {'assign_value': m},
... names.CASH_FLOW_TABLE: {'timber_prices': cft},
... names.CASH_FLOW: {},
... names.PARAMETER_TABLE: {}}
>>> mcd = ModelChainDef(typedef)
>>> mcd.schema = schema
def xml_to_obj(self, xmlname, root, lexicon, chaintype, models):
================================================================
Converts the model chain xml definition into Python objects::
>>> try:
... mcd.xml = ('testxml', xml, Lexicon(), 'simulation', mi)
... except ValueError, e:
... print e.message
Called PredictionModel.vars.keys()
Called PredictionModel.__deepcopy__({})
Called Model.__deepcopy__({})
Called OperationModel.vars.keys()
Called OperationModel.__deepcopy__({})
Called OperationModel.vars.keys()
Called OperationModel.__deepcopy__({})
errors in xml to object conversion
>>> mcd.xml['simulation']['testxml'][:13]
u'>> mc = mcd.obj['simulation']['testxml']
>>> mc.chain_groups['Forced clearcut and strip cut']
['Forced clearcut']
>>> mc.chain_groups['Forced thinnings']
['Calculate thinning limits', 'Forced first thinning']
>>> mc.branching_groups # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
{'Final_harvest': [[([('data', (None, None, True)),
('value', 0.0), ('eq', )],
{'Clearcut or seedtree cut': ['Normal clearcut',
'Delayed clearcut, 5 years',
'Delayed clearcut, 10 years']})], False]}
>>> mc.chain_type
'simulation'
>>> len(mc.chains)
3
>>> mc.depth
3
>>> 'SP' in mc.opres_cfiers
True
>>> 'assortment' in mc.opres_cfiers
True
>>> 'something else' in mc.opres_cfiers
True
>>> mc.opres_vars
set(['Volume', 'Income'])
>>> mc.cf_cfiers
set(['SP'])
>>> c = mc.chains[0]
>>> c.condition
>>> c.evaluate_at
'comp_unit'
>>> c.name
'Calculate thinning limits'
>>> len(c.tasks)
1
>>> t = c.tasks[0]
>>> t.name
'Calculate thinning limits'
>>> t.value_fixer
False
>>> t.model # doctest: +ELLIPSIS
>>> t.condition_parser
>>> t.validator
>>> p = t.model_param
>>> p.parameters
[None]
>>> p.rect_factor
1.0
>>> p.risk_level
1
>>> p._validator
>>> c = mc.chains[2]
>>> c.condition[0]
('data', (1, 1, True))
>>> t = c.tasks[0]
>>> 'assortment' in t.opres_cfiers
True
>>> st = t.subtasks[0]
>>> st.condition[3]
('data', (None, None, True))
>>> st.model # doctest: +ELLIPSIS
>>> 'assortment' in st.opres_cfiers
True
>>> p = st.model_param
>>> p.cash_flow_table # doctest: +ELLIPSIS
[]
>>> p.simulation_effects
{'force_step': None, 'erase_memory': False}
>>> t = c.tasks[1]
>>> t.branching_group
'Final_harvest'
>>> t.branching_ops
[3, 3]
>>> for op in t.branch_conditions: print op
... # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
('Normal clearcut', [('data', (1, 5, True)),
('value', 0.0),
('eq', )])
('Delayed clearcut, 5 years', [('data', (1, 5, True)),
('value', 5.0),
('eq', )])
('Delayed clearcut, 10 years', [('data', (1, 5, True)),
('value', 10.0),
('eq', )])
>>> t.condition # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
[('data', (1, 4, True)), ('value', 1.0), ('eq', ),
('data', (1, 3, True)), ('value', 1.0), ('eq', ),
('group', )]
Check that the subtask structure is correctly generated::
>>> c = mc.chains[1]
>>> t = c.tasks[0]
>>> len(t.subtasks)==2
True
>>> t.subtasks[0].name
'Invalid model'
>>> t.subtasks[1].name
'Assign just thinned dummy to zero to avoid'
>>> mc.memory_models # doctest:
set(['mock model'])
Errors generated during conversion::
>>> err = list(mcd.errors)
>>> err.sort()
>>> for e in err: print e # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
Model implementation for model 'invalid' of type 'aggregation' .../Invalid model
Model implementation for model 'zero_value' of type 'aggregation' ...
Parameter 'THINNING_LEVEL' is not a valid parameter for prediction model 'mock model' ...
Parameter 'test' is not defined for prediction model 'mock model' ...
Variable 'JUST_THINNED' not found at level 'comp_unit' ...
Variable 'SOME' not found at level 'comp_unit' ...
Variable 'open_area' not found at level 'comp_unit' ...
>>> mcd.ok
{'testxml': False}
***************************************
class ModelChainCollection(Persistent):
***************************************
A bunch of model chain groups originally in an XML file
*****************************
class ModelChain(Persistent):
*****************************
def __init__(self, ns, root, validator, chaintype, models, bgroups,
opres_cfiers, memory_models):
=============================
Construct model chain from the parsed model chain xml.