sim.py

class Simulator(object):

SIMO simulator module.

Attributes: - _logger: simulation log, logger instance - _log_name: log name, string - lexicon: simulation lexicon, Lexicon instance - _ctrl: simulation main controls, SimControl instance - _callers: model and table caller instances, dictionary with model/table type as key and caller instance as value - _evaluator: condition evaluator, ConditionEvaluator instance - _input_db: input database: DataDB instance - _data_db: data database: DataDB instance - _operation_db: operation database: OperationDB instance - data: data matrix handler, Handler instance - _iterations: number of iterations, integer - period: current period, integer - timespan: current time span, Timespan object - chain_index: current model chain index, integer - _chain_type: current model chain type: ‘init’, ‘loop’ or ‘operation’ - _chains: model chains of the current time span in a dictionary, chain type as key, list of chains as value - _forced_operations: forced operations for current simulation units, dictionary - level: current evaluation level, integer - main_level: simulation main level, integer - _tasks: currently active tasks in a list - task: currently active task object - value_fixing: value fixing, boolean - _block_growth: target index of objects for which growth is blocked - _block_all: target index of objects that are totally blocked - terminate: terminate simulation, boolean

Some of the caller classes need to access certain simulator data attributes. This access is granted with properties that reveal the needed attributes.

Properties: - operation_memory

def __init__(self, logger, logname, lexicon, ctrl, input_db, data_db, operation_db, warnings, maxunitcount, obj_count_multiplier, branch_count, no_op_branch, separate_branching_groups, branch_limit, max_res_objs, max_errors, max_simunit_errors, deterministic, track_prices, do_forced_ops, copy2iters, create_branch_desc, cache_dir):

>>> import numpy as np
>>> import datetime as dt
>>> from minimock import Mock
>>> execfile('simulation/test/mocks4sim.py')
>>> import simo.simulation.sim
>>> simo.simulation.sim.Handler = fake_handler.Handler

Construct simulator instance:

>>> from simo.simulation.sim import Simulator
>>> sim = Simulator(logger, 'test-log', 'simulation', lexicon, ctrl,
...                 input_db, data_db, operation_db, True, 11, 1, 1,
...                 False, False, 100, 50, 10, 1, False, True, True, False,
...                 True, 'cache')

def _reset(self):

Reset simulator, erase all information about past simulation and set all settings to default.

def simulate(self, iterations):

Run simulation with current simulation setup and data matrix, but substitute data handler and evaluator at this point with mock objects

>>> sim.data = data
>>> sim.data.matrix.iterations = 1
>>> sim._evaluator = evaluator
>>> sim.simulate(1)  
Called InputDB.get_ids('data', 1, both=True, id_list=None)
Called Logger.log_message(
    'test-log',
    'INFO',
    '6 simulation units divided into 1 sets')
Called Logger.log_message(
    'test-log',
    'DEBUG',
    "Simulating ids: [(u'1', u'1'), (u'2', u'2'), (u'3', u'3'),
                      (u'4', u'4'), (u'5', u'5'), ('6', '6')]")
Called HandlerCreator.Handler(
    1, <Mock ... Lexicon>, 5, 1, 1,
    {'bgroup1': [[(None, {'btask1': ['cond1']})], True]},
    False, False, <simo.simulation.sim.Simulator object at ...>,
    6, 1, 100)
Called Logger.log_message(
    'test-log',
    'INFO',
    'filling simulation data matrix...')
Called InputDB.fill_matrix(
    <Mock ... Handler>,
    1,
    [(u'1', u'1'), (u'2', u'2'), (u'3', u'3'),
     (u'4', u'4'), (u'5', u'5'), ('6', '6')])
Called Handler.get_tind(0, 0)
Called Handler.set_value(0, array([[0, 0, 0, 0, 0]]), 3, array([ 0.]))
Called Handler.set_value(0, array([[0, 0, 1, 0, 0]]), 3, array([ 0.]))
Called Handler.set_value(0, array([[0, 0, 2, 0, 0]]), 3, array([ 0.]))
Called Logger.log_message('test-log', 'INFO', 'matrix filling complete')
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.set_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    array([...2000...1...1..., ...2000...1...1..., ...2000...1...1...],
          dtype=object),
    set_init_date=True)
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    3,
    array([ 2000.,  2000.,  2000.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    4,
    array([ 1.,  1.,  1.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    5,
    array([ 1.,  1.,  1.]))
Called Logger.log_message(
    'test-log',
    'INFO',
    'simulating iterations 1..1 for set 1/1 with 6 units')
Called Handler.get_tind(1, 0)
Called Handler.get_tind(0, 0)
Called Handler.set_value(
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2,
    5)
Called InputDB.get_level_ind(<Mock ... ModelChain.evaluate_at>)
Called Handler.get_tind(None, 0)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    <Mock ... ModelChain.condition>,
    None,
    1,
    0,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Timespan>)
Called Handler.block(None, 0, array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.release(
    None,
    0,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_tind(1, 0)
Called Handler.get_tind(1, 0)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    <Mock ... Ctrl.stop_logic>,
    1,
    1,
    0,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Timespan>)
Called Handler.block(1, 0, array([[0, 0, 0, 0, 0]]), True)
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_tind(0, 0)
Called Handler.get_value(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2)
Called InputDB.get_level_ind(<Mock ... ModelChain.evaluate_at>)
Called Handler.get_tind(None, 0)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    <Mock ... ModelChain.condition>,
    None,
    1,
    0,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Timespan>)
Called Handler.block(None, 0, array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.release(
    None,
    0,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called InputDB.get_level_ind(<Mock ... ModelChain.evaluate_at>)
Called Handler.get_tind(None, 0)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    <Mock ... ModelChain.condition>,
    None,
    1,
    0,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Timespan>)
Called Handler.block(None, 0, array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.release(
    None,
    0,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_value(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2)
Called Handler.set_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    array([...2004...12...31..., ...2004...12...31..., ...2004...12...31...],
          dtype=object))
Called DataDB.add_data_from_matrix(
    <Mock ... Handler.matrix.dmx>,
    <Mock ... Handler.matrix.mx>,
    <Mock ... Handler.ind2id>,
    <Mock ... Handler.linkage.links>,
    set([(0, 0, 0)]),
    set([]),
    1,
    <Mock ... Handler.text_data>,
    False,
    0)
Called DataDB.add_branching_info(
    1,
    <Mock ... Handler.brancher.branching_info>,
    <Mock ... Handler.ind2id.ids>,
    0)
Called Handler.brancher.reset_binfo()
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.set_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    array([...2000...1...2..., ...2000...1...2..., ...2000...1...2...],
          dtype=object))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    3,
    array([ 2000.,  2000.,  2000.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    4,
    array([ 1.,  1.,  1.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    5,
    array([ 2.,  2.,  2.]))
Called Handler.release(None, None, None, False, True)
Called Handler.check_errors(1)
Called Handler.block(1, 0, array([[0, 0, 0]]), True)
Called Logger.log_message('test-log', 'INFO',
                          'creating branch descriptions')
Called DataDB.fill_branch_desc_table([u'1', u'2', u'3', u'4', u'5', '6'],
                                     0)
Called Logger.log_message('test-log', 'INFO',
                          'set simulated in 00:00:...')
Called Logger.log_message('test-log', 'INFO',
                          'avg. time per unit 00:00:...')
Called Logger.log_message('test-log', 'INFO',
                          'avg. time per set 00:00:...')
Called Logger.log_message('test-log', 'INFO',
                          'total time 00:00:...')

NB: The last call to Handler.set_date adds a single day to to initial dates. As the Handler is a mock object the previous call didn’t really change the dates, hence the year discrepancy between the calls.

def _evaluate_steps(self, forceops):

Evaluate number of timesteps with possible stopping conditions, simulation chains, forced operations and/or operation chains and update simulator state:

>>> sim._start_step = 0
>>> sim._evaluate_steps(False) 
Called Handler.get_tind(1, 0)
Called Handler.get_tind(1, 0)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    <Mock ... Ctrl.stop_logic>,
    1,
    1,
    0,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Timespan>)
Called Handler.block(1, 0, array([[0, 0, 0, 0, 0]]), True)
Called InputDB.get_level_ind(<Mock ... ModelChain.evaluate_at>)
Called Handler.get_tind(None, 0)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    <Mock ... ModelChain.condition>,
    None,
    1,
    0,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Timespan>)
Called Handler.block(None, 0, array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.release(
    None,
    0,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called InputDB.get_level_ind(<Mock ... ModelChain.evaluate_at>)
Called Handler.get_tind(None, 0)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    <Mock ... ModelChain.condition>,
    None,
    1,
    0,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Timespan>)
Called Handler.block(None, 0, array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.release(
    None,
    0,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_value(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2)
Called Handler.set_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    array([...2004...12...31..., ...2004...12...31..., ...2004...12...31...],
          dtype=object))
Called DataDB.add_data_from_matrix(
    <Mock ... Handler.matrix.dmx>,
    <Mock ... Handler.matrix.mx>,
    <Mock ... Handler.ind2id>,
    <Mock ... Handler.linkage.links>,
    set([(0, 0, 0)]),
    set([]),
    1,
    <Mock ... Handler.text_data>,
    False,
    0)
Called DataDB.add_branching_info(
    1,
    <Mock ... Handler.brancher.branching_info>,
    <Mock ... Handler.ind2id.ids>,
    0)
Called Handler.brancher.reset_binfo()
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.set_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    array([...2000...1...2..., ...2000...1...2..., ...2000...1...2...],
          dtype=object))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    3,
    array([ 2000.,  2000.,  2000.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    4,
    array([ 1.,  1.,  1.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    5,
    array([ 2.,  2.,  2.]))

def _evaluate_chains(self, chaintype, ind=None):

The model chains are evaluated so that the method goes through the top level ‘task’ elements and calls a method to process each of the task elements. chaintype is one of ‘init’, ‘simulation’, ‘operation’ or ‘forced_operation’.

def _evaluate_task(self, task, depth):

Method evaluates a task element of a model chain and moves on recursively to any subtasks if there are subtasks. If condition element is found, evaluate_expression Method is called and if model element is found, evaluate_model Method is called.

Evaluate a task with no condition, model or subtasks:

>>> task = Mock('Task')
>>> task.condition = None
>>> task.model = None
>>> task.path = 'path/of/this/task'
>>> task.subtasks = []
>>> task.task.path = 'path/of/this/task'
>>> sim._evaluate_task(task, 0)

Evaluate a task with condition:

>>> task.condition = Mock('Expression')
>>> sim._evaluate_task(task, 0)  
Called Handler.get_tind(None, 0)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    <Mock ... Expression>,
    None,
    1,
    0,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>)
Called Handler.block(None, 0, array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.release(
    None,
    0,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))

Evaluate a task with subtasks:

>>> task = Mock('Task')
>>> task.condition = None
>>> task.model = None
>>> task.path = 'path/of/this/task'
>>> subtask_1 = Mock('Task')
>>> subtask_1.condition = None
>>> subtask_1.model = None
>>> subtask_1.path = 'path/of/this/task'
>>> subtask_1.subtasks = []
>>> subtask_2 = Mock('Task')
>>> subtask_2.condition = None
>>> subtask_2.model = None
>>> subtask_2.path = 'path/of/this/task'
>>> subtask_2.subtasks = [subtask_1, subtask_1]
>>> task.subtasks = [subtask_1, subtask_2]
>>> sim._evaluate_task(task, 0)  

def _evaluate_subtasks(self, task, depth):

Evaluate a subtask and all it’s possible subtasks. In this example the task is a so-called ‘branch_task’, which means that it is processed in a bit different way than normal task:

>>> task = Mock('Task')
>>> task.type = 'branch_task'
>>> task.path = 'path/of/this/task'
>>> task.condition = None
>>> subtask = Mock('Task')
>>> subtask.model = None
>>> subtask.path = 'path/of/this/task'
>>> subtask.condition = None
>>> subtask.subtasks = []
>>> sim._tasks = [task]
>>> task.subtasks = [subtask]
>>> task.branch_conditions = [('branch1', 'C1'), ('branch2', 'C2')]
>>> task.branching_group = 'bgroup1'
>>> task.branching_group_part = 0
>>> sim._evaluate_subtasks(task, 0)  
Called Handler.get_tind(None, 1)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    'C1',
    None,
    1,
    1,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Timespan>)
Called Handler.block(None, 1, array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.brancher.start(
    <Mock ... Task>,
    'branch1',
    <Mock ... Handler>,
    <Mock ... Lexicon>,
    1,
    1,
    <simo.simulation.caller.predictionmemory.PredictionModelMemory object at ...>,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>)
Called Handler.brancher.stop(
    'branch1',
    <Mock ... Handler>,
    <Mock ... Timespan>,
    1,
    1,
    <simo.simulation.caller.predictionmemory.PredictionModelMemory object at ...>,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Lexicon>,
    0,
    2,
    [],
    [])
Called Handler.release(
    None,
    1,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_tind(None, 1)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    'C2',
    None,
    1,
    1,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Timespan>)
Called Handler.block(None, 1, array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.brancher.start(
    <Mock ... Task>,
    'branch2',
    <Mock ... Handler>,
    <Mock ... Lexicon>,
    1,
    1,
    <simo.simulation.caller.predictionmemory.PredictionModelMemory object at ...>,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>)
Called Handler.brancher.stop(
    'branch2',
    <Mock ... Handler>,
    <Mock ... Timespan>,
    1,
    1,
    <simo.simulation.caller.predictionmemory.PredictionModelMemory object at ...>,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Lexicon>,
    0,
    2,
    [],
    [])
Called Handler.release(
    None,
    1,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))

def _evaluate_model(self, model, depth):

Evaluate a model stored in a model chain. Models can be of types: ‘prediction’, ‘aggregation’, ‘operation’, ‘management’, ‘geo_table’ and ‘parameter_table’.

def _evaluate_expression(self, expr, level, depth=0):

Evaluate a condition expression. Expressions are in postfix deque items and are evaluated with a postfix evaluator. Return target index of all blocked objects and boolean value True if all evaluation level objects are blocked and False if not.

def _evaluate_stop_condition(self):

Evaluate simulation stopping condition and stop simulation if condition evaluates True for all simulation level objects. Block the objects for which the condition was met:

>>> sim._evaluator.evaluate.mock_returns = np.array([True, False, True],
...                                                 dtype=bool)
>>> sim._evaluate_stop_condition()  
Called Handler.get_tind(1, 0)
Called Handler.get_tind(1, 0)
Called Evaluator.evaluate(
    <Mock ... Handler>,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    <Mock ... Ctrl.stop_logic>,
    1,
    1,
    0,
    <simo.simulation.caller.operationmemory.OperationMemory object at ...>,
    <Mock ... Timespan>)
Called Handler.block(
    1,
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 2, 0, 0]]),
    True)
False

def _evaluate_end_date(self):

Check if any of the main level objects have passed the simulation ending date. Block objects that have passed the ending date:

>>> sim.data.get_date.mock_returns = np.array([
...     dt.date(2001,1,1),
...     dt.date(1999,1,1),
...     dt.date(2000,1,1)], dtype=dt.date)
>>> sim.timespan.ending.target = dt.date(2000,1,1)
>>> sim._evaluate_end_date()  
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.set_value(
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2,
    array([ 0.,  1.,  0.]))
Called Handler.block(
    1,
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 2, 0, 0]]),
    True)
False
>>> sim._block_all
array([[0, 0, 0, 0, 0],
       [0, 0, 2, 0, 0]])

>>> sim.timespan.ending.target = dt.date(2002,1,1)
>>> sim._evaluate_end_date()  
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.set_value(
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2,
    array([ 1.,  3.,  2.]))
False
>>> sim._block_all

>>> sim.timespan.ending.target = dt.date(1998,1,1)
>>> sim._evaluate_end_date()  
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.set_value(
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2,
    array([ 0.,  0.,  0.]))
Called Handler.block(
    1,
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    True)
True
>>> sim._block_all
array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]])

def _do_grow(self):

Check if growth should be calculated for the current time step, ie. should the ‘simulation’ model chains be evaluated, and block objects for which the growth simulation should be passed.

If time step unit is year and length is at least a year, this check has to be done only during the first time step.

If time step is less than one year, the growth season should be considered and the sub year steps should be distributed evenly to the growth season.

>>> sim.period = 0
>>> sim._ctrl.growth_season_end_month = 7
>>> sim._ctrl.growth_season_end_day = 1
>>> sim.data.get_date.mock_returns = np.array([
...     dt.date(2000,6,30),
...     dt.date(2000,7,2),
...     dt.date(1999,7,2)], dtype=dt.date)
>>> sim._do_grow()
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_tind(0, 0)
Called Handler.get_value(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2)
Called Handler.set_value(
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2,
    array([ 5.,  4.,  4.]))
True
>>> sim._block_growth

Test a situation with “end_date” as timespan ending type, first timestep out of more than one timesteps and one (the third) unit is past current year’s growth season and the end date is at the end of next year’s growth season. In this case none of the three units should be blocked as the possible time step reductions have already been done in _evaluate_end_date. All units with timestep greater than 0 should be simulated.

>>> sim.period = 0
>>> sim._end_step = 2
>>> sim._ctrl.growth_season_end_month = 7
>>> sim._ctrl.growth_season_end_day = 31
>>> sim.timespan.ending.type = "end_date"
>>> sim.timespan.ending.target = dt.date(2010, 7, 31)
>>> sim.data.get_date.mock_returns = np.array([
...     dt.date(2002,8,20),
...     dt.date(2005,8,20),
...     dt.date(2009,8,20)], dtype=dt.date)
>>> sim.data.get_value.mock_returns = (np.array([5.,5.,1.], dtype=float),
...                                    None)
>>> sim._do_grow()
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_tind(0, 0)
Called Handler.get_value(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2)
Called Handler.set_value(
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2,
    array([ 5.,  5.,  1.]))
True
>>> sim._block_growth

Test again a situation with “steps” as timespan ending type and all units have timesteps greater than 1

>>> sim._ctrl.growth_season_end_month = 7
>>> sim._ctrl.growth_season_end_day = 1
>>> sim.timespan.ending.type = "steps"
>>> sim.timespan.ending.target = dt.date(1998, 1, 1)
>>> sim.data.get_date.mock_returns = np.array([
...     dt.date(2000,7,2),
...     dt.date(1999,7,2),
...     dt.date(2010,7,2)], dtype=dt.date)
>>> sim.data.get_value.mock_returns = (np.array([5.,4.,4.], dtype=float),
...                                    None)
>>> sim._reset()
>>> sim.data = data
>>> sim._do_grow()
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_tind(0, 0)
Called Handler.get_value(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2)
Called Handler.set_value(
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2,
    array([ 4.,  3.,  3.]))
True
>>> sim._block_growth

def _do_force_operations(self):

Check if forced operations should be done to any of the current simulation units during the current timespan

>>> sim._operation_db.fill_dictionary.mock_returns = {
...     u'2': 'forced operations for unit 2',
...     u'3': 'forced operations for unit 3',
...     u'5': 'forced operations for unit 5'}
>>> sim.lexicon.get_level_name.mock_returns = u'comp_unit'
>>> sim.data.get_id.mock_returns = [u'2', u'3', u'4']
>>> sim.data.get_tind.mock_returns = (np.array([[0, 0, 0, 0, 0],
...                                             [0, 0, 1, 0, 0],
...                                             [0, 0, 2, 0, 0]],
...                                            dtype=int), None)
>>> timespan.forced_operations = True
>>> chain1 = Mock('Chain')
>>> chain2 = Mock('Chain')
>>> chain3 = Mock('Chain')
>>> chain1.name = 'Chain one'
>>> chain2.name = 'Chain two'
>>> chain3.name = 'Chain three'
>>> mcc = Mock('ModelChainCollection')
>>> mcc.chains = [chain1, chain2]
>>> mcc2 = Mock('ModelChainCollection')
>>> mcc2.chains = [chain3]
>>> timespan.chain_collections['forced_operation'] = [mcc, mcc2]
>>> sim.timespan = timespan
>>> sim._do_force_operations() 
Called Lexicon.get_level_name(1)
Called OperationDB.fill_dictionary(
    u'comp_unit',
    [(u'1', u'1'), (u'2', u'2'), (u'3', u'3'),
     (u'4', u'4'), (u'5', u'5'), ('6', '6')])
Called Handler.get_tind(1, 0)
Called Handler.get_id(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_init_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
True

>>> sim._forced_operations[u'2']
'forced operations for unit 2'
>>> sim._forced_operations[u'3']
'forced operations for unit 3'
>>> sim._forced_operations[u'5']
'forced operations for unit 5'

>>> sim._forced_chain_index
[{'Chain two': 1, 'Chain one': 0}, {'Chain three': 0}]

def _evaluate_forced_operations(self):

Evaluate the forced operations for the current units. Forced operations can be operation that actually have happened when simulating from past to present (updating simulations) or operations can be forced to simulation units for some other reason

>>> sim.data.get_date.mock_returns = np.array([
...     dt.date(2005,7,2),
...     dt.date(2004,7,2),
...     dt.date(2015,7,2)], dtype=dt.date)
>>> fop = Mock('ForcedOperation')
>>> fop.id = u'2'
>>> fop.level = u'comp_unit'
>>> fop.name = u'clearcut'
>>> fop.timing_type = 'step'
>>> fop.time_step = 1
>>> fop.date = dt.date(2006,1,1)
>>> fop.step_unit = 'year'
>>> fop.chain_names = ['Chain two', 'Chain three']
>>> fop.chain_indices = None
>>> fop.parameters = None
>>> fop.data_values = {(u'comp_unit', u'FORCED_OP_VAR'): 100.}
>>> sim._forced_operations = {(0,0,u'2'): [fop]}
>>> sim.data.get_id.mock_returns = ['2','5','6']
>>> sim._evaluate_chains = Mock('mock_evaluate_chains')

>>> sim._evaluate_forced_operations()
Called Handler.get_tind(1, 0)
Called Handler.get_id(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_value(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2)
Called Handler.block(
    1,
    0,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    True)
Called Lexicon.get_variable_ind(u'comp_unit', u'FORCED_OP_VAR')
Called Handler.get_value(array([[0, 0, 0, 0, 0]]), 0)
Called Handler.set_value(1, array([[0, 0, 0, 0, 0]]), 0, 100.0)
Called mock_evaluate_chains('forced_operation', [[1], [0]])
Called Lexicon.get_variable_ind(u'comp_unit', u'FORCED_OP_VAR')
Called Handler.get_value(array([[0, 0, 0, 0, 0]]), 0)
Called Handler.set_value(1, array([[0, 0, 0, 0, 0]]), 0, 4.0)
Called Handler.release(
    1,
    0,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    True,
    False)
False

Set the current dates so that forced operations should not be evaluated yet

>>> sim.data.get_date.mock_returns = np.array([
...     dt.date(2000,7,2),
...     dt.date(1999,7,2),
...     dt.date(2005,7,2)], dtype=dt.date)
>>> sim._forced_operations = {(0,0,u'2'): [fop]}
>>> sim._blocked_forced = np.array([], dtype=int)
>>> sim._blocked_forced.shape = (0, 5)
>>> sim._evaluate_forced_operations()
Called Handler.get_tind(1, 0)
Called Handler.get_id(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_value(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2)
True

Set the current dates so that forced operations should already have been evaluated

>>> sim.data.get_date.mock_returns = np.array([
...     dt.date(2015,7,2),
...     dt.date(2014,7,2),
...     dt.date(2025,7,2)], dtype=dt.date)
>>> sim._blocked_forced = np.array([], dtype=int)
>>> sim._blocked_forced.shape = (0, 5)
>>> sim._evaluate_forced_operations()
Called Handler.get_tind(1, 0)
Called Handler.get_id(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_value(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2)
True

def _get_num_of_steps(self):

Get the maximum number of timesteps for the current timespan. As each simulation unit might have different date stamp, the number of time steps will be the maximum over all units.:

>>> sim.data.get_date.mock_returns = np.array([
...     dt.date(2005,7,2),
...     dt.date(2004,7,2),
...     dt.date(2015,7,2)], dtype=dt.date)
>>> sim.timespan.ending.type = 'end_date'
>>> sim._get_num_of_steps() 
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Logger.log_message(
    'test-log',
    'INFO',
    'all simulation units are past simulation ending date (1998-01-01), time step of timespan n.o 1 set to 0')
True

>>> sim.timespan.ending.target = dt.date(2020,1,1)
>>> sim._get_num_of_steps()
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Logger.log_message(
    'test-log',
    'WARNING',
    'The number of years (16) is not divisible with the given time step (5). Forcably rounding up.')
True

>>> sim._end_step
4
>>> sim.timespan.ending.type = 'steps'
>>> sim.timespan.ending.target = 10
>>> sim._get_num_of_steps()
True
>>> sim._end_step
10

def _update(self):

Update simulator state and move to next time step:

>>> sim._update(to='current period end') 
Called Handler.set_value(
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2,
    array([ 5.,  4.,  4.]))
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_value(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    2)
Called Handler.set_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    array([...2009...7...1..., ...2007...7...1..., ...2018...7...1...],
          dtype=object))
>>> sim.period
0

Note that the ‘next period beginning’ update only adds a single day to the current day. That’s because the ‘current period end’ is assumed to be made before this call. However, as the mock object used here doesn’t really change the date, a single day is here added to the initial date.

>>> sim._update(to='next period beginning') 
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.set_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    array([...2005...7...3..., ...2004...7...3..., ...2015...7...3...],
          dtype=object))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    3,
    array([ 2005.,  2004.,  2015.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    4,
    array([ 7.,  7.,  7.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    5,
    array([ 3.,  3.,  3.]))
>>> sim.period
1

def _set_data(self, iterations, start, ids):

Create data handler if first simulation run and reset existing handler otherwise. Fill data matrix from input database and store init variable values to data matrix

iterations – number of iterations, int start – starting iteration round ids – list of simulation unit id strings

>>> sim.data.matrix.iterations = 1
>>> sim._ctrl.init_variables = [((1,1),0.0), ((0,5),15.0)]
>>> sim._set_data(1, 0, ['1', '2', '3'])  
Called Handler.reset()
Called Logger.log_message(
    'test-log',
    'INFO',
    'filling simulation data matrix...')
Called InputDB.fill_matrix(<Mock ... Handler>, 1, ['1', '2', '3'])
Called Handler.get_tind(0, 0)
Called Handler.set_value(0, array([[0, 0, 0, 0, 0]]), 3, array([ 0.]))
Called Handler.set_value(0, array([[0, 0, 1, 0, 0]]), 3, array([ 0.]))
Called Handler.set_value(0, array([[0, 0, 2, 0, 0]]), 3, array([ 0.]))
Called Handler.get_tind(0, 0)
Called Handler.set_value(
    0,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    [5],
    [15.0])
Called Handler.get_tind(1, 0)
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    [1],
    [0.0])
Called Logger.log_message('test-log', 'INFO', 'matrix filling complete')

def _divide_data(self, iterations, idlist, indseq):

Divide the simulation target data into reasonable blocks:

>>> sim._iterations = 2
>>> sim._divide_data([u'1',u'2',u'3',u'4'], None)
Called InputDB.get_ids('data', 1, both=True, id_list=[u'1', u'2', u'3', u'4'])
Called Logger.log_message(
    'test-log',
    'INFO',
    '4 simulation units divided into 1 sets')
[([(u'1', u'1'), (u'2', u'2'), (u'3', u'3'), (u'4', u'4')], 2, 0)]
>>> sim._iterations = 1
>>> sim._divide_data(None, (2,4))
Called InputDB.get_ids('data', 1, both=True, id_list=None)
Called Logger.log_message(
    'test-log',
    'INFO',
    '3 simulation units divided into 1 sets')
[([(u'2', u'2'), (u'3', u'3'), (u'4', u'4')], 1, 0)]
>>> sim._iterations = 3
>>> sim._divide_data(None, None) 
Called InputDB.get_ids('data', 1, both=True, id_list=None)
Called Logger.log_message(
    'test-log',
    'INFO',
    '6 simulation units divided into 2 sets')
[([(u'1', u'1'), (u'2', u'2'), (u'3', u'3')], 3, 0),
 ([(u'4', u'4'), (u'5', u'5'), ('6', '6')], 3, 0)]
>>> sim._iterations = 5
>>> sim._divide_data(None, None) 
Called InputDB.get_ids('data', 1, both=True, id_list=None)
Called Logger.log_message(
    'test-log',
    'INFO',
    '6 simulation units divided into 3 sets')
[([(u'1', u'1'), (u'2', u'2')], 5, 0),
 ([(u'3', u'3'), (u'4', u'4')], 5, 0),
 ([(u'5', u'5'), ('6', '6')], 5, 0)]
>>> sim._iterations = 6
>>> sim._divide_data(None, None) 
Called InputDB.get_ids('data', 1, both=True, id_list=None)
Called Logger.log_message(
    'test-log',
    'INFO',
    '6 simulation units divided into 6 sets')
[([(u'1', u'1')], 6, 0), ([(u'2', u'2')], 6, 0), ([(u'3', u'3')], 6, 0),
 ([(u'4', u'4')], 6, 0), ([(u'5', u'5')], 6, 0), ([('6', '6')], 6, 0)]
>>> sim._iterations = 10
>>> sim._divide_data(None, None) 
Called InputDB.get_ids('data', 1, both=True, id_list=None)
Called Logger.log_message(
    'test-log',
    'INFO',
    '6 simulation units divided into 6 sets')
[([(u'1', u'1')], 10, 0), ([(u'2', u'2')], 10, 0), ([(u'3', u'3')], 10, 0),
 ([(u'4', u'4')], 10, 0), ([(u'5', u'5')], 10, 0), ([('6', '6')], 10, 0)]

Info message should be generated if invalid simulation units are given in idlist:

>>> sim._input_db.get_ids.returns = [(u'1', u'1') ,(u'2', u'2'),
...                              (u'3', u'3'), (u'4', u'4'), (u'5', u'5')]
>>> sim._iterations = 2
>>> sim._divide_data([u'2',u'3',u'49',u'139'], None)
... 
Called InputDB.get_ids(
    'data',
    1,
    both=True,
    id_list=[u'2', u'3', u'49', u'139'])
Called Logger.log_message(
    'test-log',
    'INFO',
    u'following simulation units were not found in input database:
      49, 139')
Called Logger.log_message(
    'test-log',
    'INFO',
    '2 simulation units divided into 1 sets')
[([(u'2', u'2'), (u'3', u'3')], 2, 0)]

def _set_dates(self):

Set initial dates for all main level objects:

>>> sim._ctrl.use_data_date = True
>>> sim._set_dates() 
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.set_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    array([...2005...7...2..., ...2004...7...2..., ...2015...7...2...],
           dtype=object),
    set_init_date=True)
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    3,
    array([ 2005.,  2004.,  2015.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    4,
    array([ 7.,  7.,  7.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    5,
    array([ 2.,  2.,  2.]))

>>> sim.data.get_date.mock_returns = np.array([
...     dt.date(1999,1,1),
...     0,
...     dt.date(2005,1,1)], dtype=dt.date)
>>> sim._set_dates() 
Called Handler.get_tind(1, 0)
Called Handler.get_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.set_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    array([...1999...1...1..., ...2000...1...1..., ...2005...1...1...],
          dtype=object),
    set_init_date=True)
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    3,
    array([ 1999.,  2000.,  2005.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    4,
    array([ 1.,  1.,  1.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    5,
    array([ 1.,  1.,  1.]))

>>> sim._ctrl.use_data_date = False
>>> sim._ctrl.use_today = False
>>> sim._set_dates() 
Called Handler.get_tind(1, 0)
Called Handler.set_date(
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    array([...2000...1...1..., ...2000...1...1..., ...2000...1...1...],
          dtype=object),
    set_init_date=True)
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    3,
    array([ 2000.,  2000.,  2000.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    4,
    array([ 1.,  1.,  1.]))
Called Handler.set_value(
    1,
    array([[0, 0, 0, 0, 0],
       [0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    5,
    array([ 1.,  1.,  1.]))

def add_error(self, errorstr, level, obj, tind=None):

Log error message and update error counters.

Add a general error message ::

>>> sim.add_error("This is a general error", None, None)
Called Logger.log_message('test-log', 'ERROR', 'This is a general error')

def add_warning(self, warningstr, level, obj, tind=None):

Log warning message.

Add a general warning message ::

>>> sim.add_warning("This is a general warning", None, None)
Called Logger.log_message('test-log', 'WARNING', 'This is a general warning')

def _add2log(self, msg, level, obj, tind, mtype):

Add messages to log. Messages can be general or object-specific, in which case object indice(s) must be given as an input parameter. Target index is an optional parameter.

>>> tind = np.array([[1, 2, 0, 0, 0],
...                  [0, 0, 1, 0, 0],
...                  [0, 0, 2, 0, 0]], dtype=int)
>>> sim.task = Mock('Task')
>>> sim.task.path = 'path/of/this/task'

Add a general message

>>> sim._add2log("This is a general message", None, None, 'TEST')
Called Logger.log_message(
    'test-log',
    'TEST',
    'SIMULATION STEP 1, TASK: path/of/this/task, This is a general message')

Add a message for a single object at simulation main level

>>> sim.level = sim.main_level = 1
>>> sim.data.get_id.mock_returns = [(u'3', u'3')]
>>> sim.data.get_date.mock_returns = [dt.date(2003,3,3)]
>>> etind = np.array([[0, 0, 2, 0, 0]], dtype=int)
>>> sim._add2log("This is a message for a single object", 1, etind, 'TEST')
Called Handler.get_id(1, array([[0, 0, 2, 0, 0]]))
Called Handler.get_date(array([[0, 0, 2, 0, 0]]))
Called Lexicon.get_level_name(1)
Called Logger.log_message(
    'test-log',
    'TEST',
    u"TASK: path/of/this/task, DATE: 2003-03-03, SIMULATION UNIT: '3', ITERATION: 0, BRANCH: 0, OBJECT: comp_unit '3': This is a message for a single object",
    u'3')

Add a message for multiple objects at simulation main level

>>> sim.level = sim.main_level = 1
>>> sim.data.get_id.mock_returns = [(u'3', u'3'), (u'7', u'7')]
>>> sim.data.get_date.mock_returns = [dt.date(2003,3,3),dt.date(2007,7,7)]
>>> etind = np.array([[1, 2, 0, 0, 0],
...                  [0, 0, 2, 0, 0]], dtype=int)
>>> sim._add2log("This is a message for a single object", 1, etind, 'TEST')
Called Handler.get_id(1, array([[1, 2, 0, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Handler.get_date(array([[1, 2, 0, 0, 0],
       [0, 0, 2, 0, 0]]))
Called Lexicon.get_level_name(1)
Called Logger.log_message(
    'test-log',
    'TEST',
    u"TASK: path/of/this/task, DATE: 2007-07-07, SIMULATION UNIT: '7', ITERATION: 0, BRANCH: 0, OBJECT: comp_unit '7': This is a message for a single object",
    u'7')
Called Logger.log_message(
    'test-log',
    'TEST',
    u"TASK: path/of/this/task, DATE: 2003-03-03, SIMULATION UNIT: '3', ITERATION: 1, BRANCH: 2, OBJECT: comp_unit '3': This is a message for a single object",
    u'3')

Add a message for a single object on simulation main level’s child level

>>> sim.level = 3
>>> sim.data.get_id = Mock('get_id')
>>> sim.data.get_id.mock_returns_iter = iter([[(u'4-2-11', u'4-2-11')],
...                                           [(u'4', u'4')]])
>>> sim.data.get_parent.mock_returns = \
...     np.array([[1,3,0,0,0]], dtype=int), None
>>> sim.lexicon.get_level_name.mock_returns = 'tree'
>>> sim.data.get_date.mock_returns = [dt.date(2004,4,4)]
>>> etind = np.array([[1, 2, 0, 0, 0]], dtype=int)
>>> sim._add2log("This is a message for a single object", 3, etind, 'TEST')
Called Handler.get_parent(3, array([[1, 2, 0, 0, 0]]), 1)
Called get_id(3, array([[1, 2, 0, 0, 0]]))
Called Handler.get_date(array([[1, 3, 0, 0, 0]]))
Called get_id(1, array([[1, 3, 0, 0, 0]]))
Called Lexicon.get_level_name(3)
Called Logger.log_message(
    'test-log',
    'TEST',
    u"TASK: path/of/this/task, DATE: 2004-04-04, SIMULATION UNIT: '4', ITERATION: 1, BRANCH: 2, OBJECT: tree '4-2-11': This is a message for a single object",
    u'4')

def _add_opres_to_db(self):

Add the queued operation results to operation database.:

>>> import numpy
>>> sim._opres_queue = [('simulated', numpy.array([[0, 0, 0, 0, 0]]),
...     (0, 0, 'comp_unit', 's1'), 'fop1',
...     [([{'cash_flow': -100.0}], None)],  'planting', 'silviculture',
...     'notes', ['material', 'another material'])]
>>> sim._add_opres_to_db(data_db) 
Called Handler.get_date(array([[0, 0, 0, 0, 0]]))
Called DataDB.add_opres(
    'comp_unit',
    [{...}])

Test adding operation results with delay:

>>> import numpy
>>> sim._opres_queue = [('simulated', numpy.array([[0, 0, 0, 0, 0]]),
...     (0, 0, 'comp_unit', 's1'), 'fop1', [([{'cash_flow': -100.0}], 5)],
...     'planting', 'silviculture', 'notes', ['material',
...     'another material'])]
>>> sim._add_opres_to_db(data_db) 
Called Handler.get_date(array([[0, 0, 0, 0, 0]]))
Called DataDB.add_opres(
    'comp_unit',
    [{...}])

def _check_timestep_saving(self, step):

Check if data from current time step should be stored to result db.:

>>> sim._blocked_from_db
set([])
>>> sim.timespan.save_steps_to_db = 'all'
>>> sim._check_timestep_saving(1)
True

>>> sim.timespan.save_steps_to_db = 'none'
>>> sim._check_timestep_saving(1)
False

>>> sim.timespan.save_steps_to_db = 'last'
>>> sim._end_step
10

>>> sim._last_step_objs = set([(0, 0, 0), (0, 0, 1), (0, 0, 2)])

>>> sim._check_timestep_saving(1)
Called Handler.get_tind(1, None)
True

>>> sim._end_step = 3
>>> sim._last_step_objs = None
>>> sim._check_timestep_saving(1)
False

>>> sim._check_timestep_saving(2)
True

def _temporary_activation(self, block_active):

Activate just blocked objects temporarily. This can be in two cases:

1. only last time step should be stored, and some of the objects have
   just reached their last time step (due to end date or stop
   condition, but some of the objects are still simulated
2. all time steps should be stored, but some of the units are blocked
   from db already during first time step because their date is past
   the time span end date

Parameters

block_active -- indicates whether currently active objects should be

Set the blocked and last step objects and activate temporarily without blockin the active objects

>>> sim.data.get_tind.mock_returns = numpy.array([[0,0,4,0,0],
...                                               [0,0,5,0,0],
...                                               [0,0,6,0,0]], dtype=int)
>>> sim._blocked_for_good = set([(0,0,0), (0,0,1), (0,0,2)])
>>> sim._blocked_from_db = set([(0,0,2), (0,0,3)])
>>> sim._last_step_objs = set([(0,0,4), (0,0,5)])
>>> sim._temporary_activation(False)
>>> sim._blocked_from_db
set([(0, 0, 0), (0, 0, 1), (0, 0, 2), (0, 0, 3)])
>>> sim._temporarily_activated
set([(0, 0, 2), (0, 0, 3)])

Set the blocked and last step objects and activate temporarily, and also block the active objects

>>> sim.data.get_tind.mock_returns = numpy.array([[0,0,4,0,0],
...                                               [0,0,5,0,0],
...                                               [0,0,6,0,0]],
...                                              dtype=int), None
>>> sim._blocked_for_good = set([(0,0,0), (0,0,1), (0,0,2)])
>>> sim._blocked_from_db = set([(0,0,2), (0,0,3), (0,0,4)])
>>> sim._last_step_objs = set([(0,0,3), (0,0,4)])
>>> sim._temporary_activation(True)
Called Handler.get_tind(1, None)
>>> sim._blocked_from_db
set([(0, 0, 5), (0, 0, 0), (0, 0, 2), (0, 0, 4), (0, 0, 6), (0, 0, 1)])
>>> sim._temporarily_activated
set([(0, 0, 2), (0, 0, 3), (0, 0, 4)])

def _revert_blocks(self):

Revert the blocks made in ‘_check_timestep_storing’ method.:

>>> sim._revert_blocks()
>>> sim._last_step_objs
set([(0, 0, 3), (0, 0, 4)])
>>> sim._blocked_from_db
set([(0, 0, 2), (0, 0, 3), (0, 0, 4)])
>>> sim._temporarily_activated

def _sub_year_step(self):

Check if the current time span has time steps less than one year

>>> sim._sub_year_step()
False
>>> sim.timespan.unit = 'month'
>>> sim.timespan.time_step = 60
>>> sim._sub_year_step()
False
>>> sim.timespan.time_step = 12
    >>> sim._sub_year_step()
    False
    >>> sim.timespan.time_step = 3
    >>> sim._sub_year_step()
    True