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, branch_limit, max_res_objs, max_errors, max_simunit_errors, deterministic, track_prices, do_forced_ops):

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

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, 100, 50, 10, 1, False, True, True, 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)  # doctest: +ELLIPSIS
Called InputDB.get_ids('data', 1, both=True)
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'2', u'3', u'4', u'5', '6']")
Called Handler.reset()
Called Logger.log_message(
    'test-log',
    'INFO',
    'filling simulation data matrix...')
Called InputDB.fill_matrix(
    <Mock ... Handler>,
    1,
    [u'1', u'2', u'3', u'4', u'5', '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-01-01, 2000-01-01, 2000-01-01], 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 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 ...>)
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 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 ...>)
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 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 ...>)
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 ...>)
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.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,
    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-01-02, 2000-01-02, 2000-01-02], 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.release(None, None, None, False, True)
Called Handler.check_errors(1)
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'])
Called Logger.log_message('test-log', 'INFO', 'set simulated in ...')
Called Logger.log_message('test-log', 'INFO', 'avg. time per unit ...')
Called Logger.log_message('test-log', 'INFO', 'avg. time per set ...')
Called Logger.log_message('test-log', 'INFO', 'total time ...')

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)  # doctest: +ELLIPSIS
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 ...>)
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 ...>)
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 ...>)
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.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,
    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-01-02, 2000-01-02, 2000-01-02], 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.]))

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.task_path = 'path/of/this/task'
>>> task.subtasks = []
>>> task.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)  # doctest: +ELLIPSIS
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.task_path = 'path/of/this/task'
>>> subtask_1 = Mock('Task')
>>> subtask_1.condition = None
>>> subtask_1.model = None
>>> subtask_1.task_path = 'path/of/this/task'
>>> subtask_1.subtasks = []
>>> subtask_2 = Mock('Task')
>>> subtask_2.condition = None
>>> subtask_2.model = None
>>> subtask_2.task_path = 'path/of/this/task'
>>> subtask_2.subtasks = [subtask_1, subtask_1]
>>> task.subtasks = [subtask_1, subtask_2]
>>> sim._evaluate_task(task, 0)  # doctest: +ELLIPSIS

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.task_path = 'path/of/this/task'
>>> task.condition = None
>>> subtask = Mock('Task')
>>> subtask.model = None
>>> subtask.task_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)  # doctest: +ELLIPSIS
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 ...>)
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 ...>)
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 ...>)
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 ...>)
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()  # doctest: +ELLIPSIS
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 ...>)
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()  # doctest: +ELLIPSIS
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.block(1, 0, array([[0, 0, 0, 0, 0]]), True)
False
>>> sim._block_all
array([[0, 0, 0, 0, 0]])

>>> sim.timespan.ending.target = dt.date(2002,1,1)
>>> sim._evaluate_end_date()  # doctest: +ELLIPSIS
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]]))
False
>>> sim._block_all

>>> sim.timespan.ending.target = dt.date(1998,1,1)
>>> sim._evaluate_end_date()  # doctest: +ELLIPSIS
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]]))
True
>>> sim._block_all

def _do_grow(self):

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

>>> 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.block(
    1,
    0,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    True)
True
>>> sim._block_growth
array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]])

>>> 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._reset()
>>> 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]]))
False
>>> sim._block_growth

def _do_force_operations(self, timespan):

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._do_force_operations(timespan)
Called Lexicon.get_level_name(1)
Called OperationDB.fill_dictionary(
    u'comp_unit',
    [u'1', u'2', u'3', u'4', u'5', '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]]))
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
>>> 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.block(
    1,
    0,
    array([[0, 0, 1, 0, 0],
       [0, 0, 2, 0, 0]]),
    True)
Called mock_evaluate_chains('forced_operation', [[1], [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 = list([])
>>> 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]]))
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 = list([])
>>> 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]]))
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() # doctest: +NORMALIZE_WHITESPACE
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',
    'ERROR',
    'All simulation units are pass simulation ending date, simulation of
     timespan aborted!')
0

>>> 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 (17) is not divisible with the given time step (5). Forcably rounding up.')
4

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

def _update(self):

Update simulator state and move to next time step:

>>> sim._update(to='current period end')
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([2010-07-01, 2009-07-01, 2020-07-01], 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-07-03, 2004-07-03, 2015-07-03], 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.]))
>>> 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'])  # doctest: +ELLIPSIS
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._divide_data(2, [u'1',u'2',u'3',u'4'], None)
Called InputDB.get_ids('data', 1, both=True)
Called Logger.log_message(
    'test-log',
    'INFO',
    '4 simulation units divided into 1 sets')
[([u'1', u'2', u'3', u'4'], 2, 0)]
>>> sim._divide_data(1, None, (2,4))
Called InputDB.get_ids('data', 1, both=True)
Called Logger.log_message(
    'test-log',
    'INFO',
    '3 simulation units divided into 1 sets')
[([u'3', u'4', u'5'], 1, 0)]
>>> sim._divide_data(3, None, None)
Called InputDB.get_ids('data', 1, both=True)
Called Logger.log_message(
    'test-log',
    'INFO',
    '6 simulation units divided into 2 sets')
[([u'1', u'2', u'3'], 3, 0), ([u'4', u'5', '6'], 3, 0)]
>>> sim._divide_data(5, None, None)
Called InputDB.get_ids('data', 1, both=True)
Called Logger.log_message(
    'test-log',
    'INFO',
    '6 simulation units divided into 3 sets')
[([u'1', u'2'], 5, 0), ([u'3', u'4'], 5, 0), ([u'5', '6'], 5, 0)]
>>> sim._divide_data(6, None, None)
Called InputDB.get_ids('data', 1, both=True)
Called Logger.log_message(
    'test-log',
    'INFO',
    '6 simulation units divided into 6 sets')
[([u'1'], 6, 0), ([u'2'], 6, 0), ([u'3'], 6, 0), ([u'4'], 6, 0), ([u'5'], 6, 0), (['6'], 6, 0)]
>>> sim._divide_data(10, None, None)
Called InputDB.get_ids('data', 1, both=True)
Called Logger.log_message(
    'test-log',
    'INFO',
    '6 simulation units divided into 6 sets')
[([u'1'], 10, 0), ([u'2'], 10, 0), ([u'3'], 10, 0), ([u'4'], 10, 0), ([u'5'], 10, 0), (['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._divide_data(2, [u'2',u'3',u'49',u'139'], None)
Called InputDB.get_ids('data', 1, both=True)
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'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-07-02, 2004-07-02, 2015-07-02], 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.]))

>>> 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-01-01, 2000-01-01, 2005-01-01], 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.]))

>>> 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-01-01, 2000-01-01, 2000-01-01], 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.]))

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.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')]
>>> sim._add_opres_to_db(data_db) # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
Called Handler.get_date(array([[0, 0, 0, 0, 0]]))
Called DataDB.add_opres(
    'comp_unit',
    0,
    [{...}])

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')]
>>> sim._add_opres_to_db(data_db) # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
Called Handler.get_date(array([[0, 0, 0, 0, 0]]))
Called DataDB.add_opres(
    'comp_unit',
    0,
    [{...}])