Table Of Contents

Previous topic

conditionparser.py

Next topic

task.py

This Page

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'''<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

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'<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}

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.