Model chain definitions.
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'''<model_chains
... xsi:schemaLocation="http://www.simo-project.org/simo
... ../../schemas/model_chain.xsd"
... xmlns="http://www.simo-project.org/simo"
... xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
... <chain_group name="Forced thinnings">
... <model_chain evaluate_at="comp_unit"
... name="Calculate thinning limits">
... <task name="Calculate thinning limits">
... <model>
... <name>Thinning_limits_JKK</name>
... <prediction>
... <parameters>
... <parameter>
... <name>THINNING_LEVEL</name>
... <value>0.5</value>
... </parameter>
... </parameters>
... </prediction>
... </model>
... </task>
... </model_chain>
... <model_chain evaluate_at="comp_unit" name="Forced first thinning">
... <task name="Basic thinning for pine in southern Finland">
... <task name="Invalid model">
... <model>
... <name>invalid</name>
... <aggregation>
... <target>JUST_THINNED</target>
... <operands>
... <operand>
... <value>0</value>
... </operand>
... </operands>
... </aggregation>
... </model>
... </task>
... <task name="Assign just thinned dummy to zero to avoid">
... <model>
... <name>assign_value</name>
... <aggregation>
... <target>JUST_THINNED</target>
... <operands>
... <operand>
... <value>0</value>
... </operand>
... </operands>
... </aggregation>
... </model>
... </task>
... </task>
... </model_chain>
... </chain_group>
... <chain_group name="Forced clearcut and strip cut">
... <model_chain evaluate_at="comp_unit" name="Forced clearcut">
... <condition>comp_unit:MAIN_SP in [1, 8] and
... comp_unit:NF eq 0</condition>
... <task name="just making the structure deeper">
... <task name="clearcut">
... <condition>comp_unit:MAIN_SP in [1, 8] and
... comp_unit:SOME eq 0</condition>
... <model>
... <name>clearcut</name>
... <operation>
... <simulation_effects>
... <erase_memory>false</erase_memory>
... </simulation_effects>
... <cash_flow_tables>
... <cash_flow_table>timber_prices</cash_flow_table>
... </cash_flow_tables>
... </operation>
... </model>
... </task>
... </task>
... <branching_group name="Final_harvest" free_branching="false">
... <condition>comp_unit:open_area eq 0</condition>
... <branch_task
... branching_operations="seedtree_position;clearcut"
... name="Clearcut or seedtree cut">
... <branch_conditions>
... <condition name="Normal clearcut">
... comp_unit:passed_regen_limit gt 0
... </condition>
... <condition name="Delayed clearcut, 5 years">
... comp_unit:passed_regen_limit gt 5
... </condition>
... <condition name="Delayed clearcut, 10 years">
... comp_unit:passed_regen_limit gt 10
... </condition>
... </branch_conditions>
... <condition>comp_unit:MAIN_GROUP eq 1 and
... comp_unit:REGENERABLE eq 1</condition>
... <task name="Clearcut">
... <condition>comp_unit:PEAT eq 0</condition>
... <task name="clearcut">
... <model>
... <name>clearcut2</name>
... <operation>
... <simulation_effects>
... <erase_memory>false</erase_memory>
... </simulation_effects>
... <cash_flow_tables>
... <cash_flow_table>timber_prices</cash_flow_table>
... </cash_flow_tables>
... </operation>
... </model>
... </task>
... <task name="zero REGENERABLE comp_unit">
... <model>
... <name>zero_value</name>
... <aggregation>
... <target>REGENERABLE</target>
... <operands>
... <operand>
... <variable>REGENERABLE</variable>
... <level>comp_unit</level>
... </operand>
... </operands>
... </aggregation>
... </model>
... </task>
... </task>
... </branch_task>
... </branching_group>
... </model_chain>
... </chain_group>
... </model_chains>'''
>>> 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
>>> 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'<model_chains'
>>> 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
{'Final_harvest': [[([('data', (None, None, True)),
('value', 0.0), ('eq', <function eee at ...>)],
{'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
<Mock ... PredictionModel>
>>> 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
<Mock ... OperationModel>
>>> 'assortment' in st.opres_cfiers
True
>>> p = st.model_param
>>> p.cash_flow_table
[<Mock ... CashFlowTable>]
>>> 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
...
('Normal clearcut', [('data', (1, 5, True)),
('value', 0.0),
('eq', <function gte at ...>)])
('Delayed clearcut, 5 years', [('data', (1, 5, True)),
('value', 5.0),
('eq', <function gte at ...>)])
('Delayed clearcut, 10 years', [('data', (1, 5, True)),
('value', 10.0),
('eq', <function gte at ...>)])
>>> t.condition
[('data', (1, 4, True)), ('value', 1.0), ('eq', <function eee at ...>),
('data', (1, 3, True)), ('value', 1.0), ('eq', <function eee at ...>),
('group', <function and_ at ...>)]
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
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}
A bunch of model chain groups originally in an XML file
def __init__(self, ns, root, validator, chaintype, models, bgroups, opres_cfiers, memory_models): =============================
Construct model chain from the parsed model chain xml.