.. _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.