FM / python

/usr/local/lib64/python3.6/site-packages/caffe2/python/

/usr/local/lib64/python3.6/site-packages/caffe2/python

Name	Size	Mode	Actions
docs/	-	0755	rm
examples/	-	0755	rm
fakelowp/	-	0755	rm
helpers/	-	0755	rm
ideep/	-	0755	rm
layers/	-	0755	rm
mint/	-	0755	rm
mkl/	-	0755	rm
modeling/	-	0755	rm
models/	-	0755	rm
onnx/	-	0755	rm
operator_test/	-	0755	rm
predictor/	-	0755	rm
rnn/	-	0755	rm
serialized_test/	-	0755	rm
test/	-	0755	rm
trt/	-	0755	rm
__pycache__/	-	0755	rm
allcompare_test.py	2255	0644	edit dl rm
attention.py	12359	0644	edit dl rm
benchmark_generator.py	4912	0644	edit dl rm
binarysize.py	5521	0644	edit dl rm
brew.py	4762	0644	edit dl rm
brew_test.py	11739	0644	edit dl rm
build.py	153	0644	edit dl rm
cached_reader.py	4394	0644	edit dl rm
caffe2_pybind11_state.cpython-36m-x86_64-linux-gnu.so	48299712	0755	edit dl rm
caffe2_pybind11_state_gpu.cpython-36m-x86_64-linux-gnu.so	49048144	0755	edit dl rm
caffe_translator.py	35227	0644	edit dl rm
caffe_translator_test.py	3553	0644	edit dl rm
checkpoint.py	32101	0644	edit dl rm
checkpoint_test.py	13405	0644	edit dl rm
cnn.py	7626	0644	edit dl rm
context.py	2841	0644	edit dl rm
context_test.py	1792	0644	edit dl rm
control.py	19309	0644	edit dl rm
control_ops_grad.py	28893	0644	edit dl rm
control_ops_grad_test.py	1752	0644	edit dl rm
control_ops_util.py	10863	0644	edit dl rm
control_test.py	12276	0644	edit dl rm
convert.py	55	0644	edit dl rm
convert_test.py	201	0644	edit dl rm
convnet_benchmarks.py	20533	0644	edit dl rm
convnet_benchmarks_test.py	839	0644	edit dl rm
core.py	119400	0644	edit dl rm
core_gradients_test.py	38022	0644	edit dl rm
core_test.py	47683	0644	edit dl rm
crf.py	13250	0644	edit dl rm
crf_predict.py	1159	0644	edit dl rm
crf_viterbi_test.py	1663	0644	edit dl rm
dataio.py	23532	0644	edit dl rm
dataio_test.py	17575	0644	edit dl rm
dataset.py	12886	0644	edit dl rm
data_parallel_model.py	83100	0644	edit dl rm
data_parallel_model_test.py	56145	0644	edit dl rm
data_workers.py	15941	0644	edit dl rm
data_workers_test.py	6561	0644	edit dl rm
db_file_reader.py	6608	0644	edit dl rm
db_test.py	1110	0644	edit dl rm
device_checker.py	5157	0644	edit dl rm
dyndep.py	1533	0644	edit dl rm
embedding_generation_benchmark.py	5256	0644	edit dl rm
experiment_util.py	3625	0644	edit dl rm
extension_loader.py	744	0644	edit dl rm
fakefp16_transform_lib.py	322	0644	edit dl rm
filler_test.py	748	0644	edit dl rm
functional.py	4415	0644	edit dl rm
functional_test.py	4204	0644	edit dl rm
fused_8bit_rowwise_conversion_ops_test.py	3945	0644	edit dl rm
gradient_checker.py	15377	0644	edit dl rm
gradient_check_test.py	20729	0644	edit dl rm
gru_cell.py	5129	0644	edit dl rm
hip_test_util.py	405	0644	edit dl rm
hsm_util.py	2259	0644	edit dl rm
hypothesis_test.py	105762	0644	edit dl rm
hypothesis_test_util.py	26853	0644	edit dl rm
ideep_test_util.py	998	0644	edit dl rm
layers_test.py	92931	0644	edit dl rm
layer_model_helper.py	29340	0644	edit dl rm
layer_model_instantiator.py	3935	0644	edit dl rm
layer_parameter_sharing_test.py	9148	0644	edit dl rm
layer_test_util.py	4875	0644	edit dl rm
lazy.py	277	0644	edit dl rm
lazy_dyndep.py	2562	0644	edit dl rm
lazy_dyndep_test.py	3914	0644	edit dl rm
lengths_reducer_fused_8bit_rowwise_ops_test.py	7575	0644	edit dl rm
lengths_reducer_rowwise_8bit_ops_test.py	5710	0644	edit dl rm
lstm_benchmark.py	10649	0644	edit dl rm
memonger.py	34041	0644	edit dl rm
memonger_test.py	36910	0644	edit dl rm
mkl_test_util.py	1142	0644	edit dl rm
model_device_test.py	4777	0644	edit dl rm
model_helper.py	23492	0644	edit dl rm
model_helper_test.py	2336	0644	edit dl rm
modifier_context.py	1772	0644	edit dl rm
muji.py	8131	0644	edit dl rm
muji_test.py	3058	0644	edit dl rm
net_builder.py	27679	0644	edit dl rm
net_builder_test.py	11382	0644	edit dl rm
net_drawer.py	14264	0644	edit dl rm
net_printer.py	12704	0644	edit dl rm
net_printer_test.py	3190	0644	edit dl rm
nomnigraph.py	4216	0644	edit dl rm
nomnigraph_test.py	15427	0644	edit dl rm
nomnigraph_transformations.py	3787	0644	edit dl rm
nomnigraph_transformations_test.py	5767	0644	edit dl rm
normalizer.py	1411	0644	edit dl rm
normalizer_context.py	1007	0644	edit dl rm
normalizer_test.py	487	0644	edit dl rm
numa_benchmark.py	2230	0644	edit dl rm
numa_test.py	1663	0644	edit dl rm
observer_test.py	5316	0644	edit dl rm
operator_fp_exceptions_test.py	1248	0644	edit dl rm
optimizer.py	78813	0644	edit dl rm
optimizer_context.py	1462	0644	edit dl rm
optimizer_test.py	30705	0644	edit dl rm
optimizer_test_util.py	9187	0644	edit dl rm
parallelize_bmuf_distributed_test.py	9908	0644	edit dl rm
parallel_workers.py	7682	0644	edit dl rm
parallel_workers_test.py	3501	0644	edit dl rm
pipeline.py	17283	0644	edit dl rm
pipeline_test.py	2542	0644	edit dl rm
predictor_constants.py	198	0644	edit dl rm
python_op_test.py	9169	0644	edit dl rm
queue_util.py	4459	0644	edit dl rm
record_queue.py	4453	0644	edit dl rm
recurrent.py	13297	0644	edit dl rm
regularizer.py	21120	0644	edit dl rm
regularizer_context.py	1013	0644	edit dl rm
regularizer_test.py	10266	0644	edit dl rm
rnn_cell.py	68233	0644	edit dl rm
schema.py	45621	0644	edit dl rm
schema_test.py	15754	0644	edit dl rm
scope.py	3623	0644	edit dl rm
scope_test.py	5249	0644	edit dl rm
session.py	7642	0644	edit dl rm
session_test.py	2078	0644	edit dl rm
sparse_to_dense_mask_test.py	6565	0644	edit dl rm
sparse_to_dense_test.py	3556	0644	edit dl rm
task.py	24274	0644	edit dl rm
task_test.py	870	0644	edit dl rm
test_util.py	3524	0644	edit dl rm
text_file_reader.py	1990	0644	edit dl rm
timeout_guard.py	4054	0644	edit dl rm
toy_regression_test.py	2822	0644	edit dl rm
transformations.py	1832	0644	edit dl rm
transformations_test.py	11960	0644	edit dl rm
tt_core.py	9349	0644	edit dl rm
tt_core_test.py	2518	0644	edit dl rm
utils.py	14181	0644	edit dl rm
utils_test.py	1399	0644	edit dl rm
visualize.py	6315	0644	edit dl rm
workspace.py	25263	0644	edit dl rm
workspace_test.py	34844	0644	edit dl rm
_import_c_extension.py	2250	0644	edit dl rm
__init__.py	3925	0644	edit dl rm

Edit: /usr/local/lib64/python3.6/site-packages/caffe2/python/control_test.py (12276B)

from caffe2.python import control, core, test_util, workspace

import logging
logger = logging.getLogger(__name__)

class TestControl(test_util.TestCase):
    def setUp(self):
        super(TestControl, self).setUp()
        self.N_ = 10

self.init_net_ = core.Net("init-net")
        cnt = self.init_net_.CreateCounter([], init_count=0)
        const_n = self.init_net_.ConstantFill(
            [], shape=[], value=self.N_, dtype=core.DataType.INT64)
        const_0 = self.init_net_.ConstantFill(
            [], shape=[], value=0, dtype=core.DataType.INT64)

self.cnt_net_ = core.Net("cnt-net")
        self.cnt_net_.CountUp([cnt])
        curr_cnt = self.cnt_net_.RetrieveCount([cnt])
        self.init_net_.ConstantFill(
            [], [curr_cnt], shape=[], value=0, dtype=core.DataType.INT64)
        self.cnt_net_.AddExternalOutput(curr_cnt)

self.cnt_2_net_ = core.Net("cnt-2-net")
        self.cnt_2_net_.CountUp([cnt])
        self.cnt_2_net_.CountUp([cnt])
        curr_cnt_2 = self.cnt_2_net_.RetrieveCount([cnt])
        self.init_net_.ConstantFill(
            [], [curr_cnt_2], shape=[], value=0, dtype=core.DataType.INT64)
        self.cnt_2_net_.AddExternalOutput(curr_cnt_2)

self.cond_net_ = core.Net("cond-net")
        cond_blob = self.cond_net_.LT([curr_cnt, const_n])
        self.cond_net_.AddExternalOutput(cond_blob)

self.not_cond_net_ = core.Net("not-cond-net")
        cond_blob = self.not_cond_net_.GE([curr_cnt, const_n])
        self.not_cond_net_.AddExternalOutput(cond_blob)

self.true_cond_net_ = core.Net("true-cond-net")
        true_blob = self.true_cond_net_.LT([const_0, const_n])
        self.true_cond_net_.AddExternalOutput(true_blob)

self.false_cond_net_ = core.Net("false-cond-net")
        false_blob = self.false_cond_net_.GT([const_0, const_n])
        self.false_cond_net_.AddExternalOutput(false_blob)

self.idle_net_ = core.Net("idle-net")
        self.idle_net_.ConstantFill(
            [], shape=[], value=0, dtype=core.DataType.INT64)

def CheckNetOutput(self, nets_and_expects):
        """
        Check the net output is expected
        nets_and_expects is a list of tuples (net, expect)
        """
        for net, expect in nets_and_expects:
            output = workspace.FetchBlob(
                net.Proto().external_output[-1])
            self.assertEqual(output, expect)

def CheckNetAllOutput(self, net, expects):
        """
        Check the net output is expected
        expects is a list of bools.
        """
        self.assertEqual(len(net.Proto().external_output), len(expects))
        for i in range(len(expects)):
            output = workspace.FetchBlob(
                net.Proto().external_output[i])
            self.assertEqual(output, expects[i])

def BuildAndRunPlan(self, step):
        plan = core.Plan("test")
        plan.AddStep(control.Do('init', self.init_net_))
        plan.AddStep(step)
        self.assertEqual(workspace.RunPlan(plan), True)

def ForLoopTest(self, nets_or_steps):
        step = control.For('myFor', nets_or_steps, self.N_)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(self.cnt_net_, self.N_)])

def testForLoopWithNets(self):
        self.ForLoopTest(self.cnt_net_)
        self.ForLoopTest([self.cnt_net_, self.idle_net_])

def testForLoopWithStep(self):
        step = control.Do('count', self.cnt_net_)
        self.ForLoopTest(step)
        self.ForLoopTest([step, self.idle_net_])

def WhileLoopTest(self, nets_or_steps):
        step = control.While('myWhile', self.cond_net_, nets_or_steps)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(self.cnt_net_, self.N_)])

def testWhileLoopWithNet(self):
        self.WhileLoopTest(self.cnt_net_)
        self.WhileLoopTest([self.cnt_net_, self.idle_net_])

def testWhileLoopWithStep(self):
        step = control.Do('count', self.cnt_net_)
        self.WhileLoopTest(step)
        self.WhileLoopTest([step, self.idle_net_])

def UntilLoopTest(self, nets_or_steps):
        step = control.Until('myUntil', self.not_cond_net_, nets_or_steps)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(self.cnt_net_, self.N_)])

def testUntilLoopWithNet(self):
        self.UntilLoopTest(self.cnt_net_)
        self.UntilLoopTest([self.cnt_net_, self.idle_net_])

def testUntilLoopWithStep(self):
        step = control.Do('count', self.cnt_net_)
        self.UntilLoopTest(step)
        self.UntilLoopTest([step, self.idle_net_])

def DoWhileLoopTest(self, nets_or_steps):
        step = control.DoWhile('myDoWhile', self.cond_net_, nets_or_steps)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(self.cnt_net_, self.N_)])

def testDoWhileLoopWithNet(self):
        self.DoWhileLoopTest(self.cnt_net_)
        self.DoWhileLoopTest([self.idle_net_, self.cnt_net_])

def testDoWhileLoopWithStep(self):
        step = control.Do('count', self.cnt_net_)
        self.DoWhileLoopTest(step)
        self.DoWhileLoopTest([self.idle_net_, step])

def DoUntilLoopTest(self, nets_or_steps):
        step = control.DoUntil('myDoUntil', self.not_cond_net_, nets_or_steps)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(self.cnt_net_, self.N_)])

def testDoUntilLoopWithNet(self):
        self.DoUntilLoopTest(self.cnt_net_)
        self.DoUntilLoopTest([self.cnt_net_, self.idle_net_])

def testDoUntilLoopWithStep(self):
        step = control.Do('count', self.cnt_net_)
        self.DoUntilLoopTest(step)
        self.DoUntilLoopTest([self.idle_net_, step])

def IfCondTest(self, cond_net, expect, cond_on_blob):
        if cond_on_blob:
            step = control.Do(
                'if-all',
                control.Do('count', cond_net),
                control.If('myIf', cond_net.Proto().external_output[-1],
                           self.cnt_net_))
        else:
            step = control.If('myIf', cond_net, self.cnt_net_)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(self.cnt_net_, expect)])

def testIfCondTrueOnNet(self):
        self.IfCondTest(self.true_cond_net_, 1, False)

def testIfCondTrueOnBlob(self):
        self.IfCondTest(self.true_cond_net_, 1, True)

def testIfCondFalseOnNet(self):
        self.IfCondTest(self.false_cond_net_, 0, False)

def testIfCondFalseOnBlob(self):
        self.IfCondTest(self.false_cond_net_, 0, True)

def IfElseCondTest(self, cond_net, cond_value, expect, cond_on_blob):
        if cond_value:
            run_net = self.cnt_net_
        else:
            run_net = self.cnt_2_net_
        if cond_on_blob:
            step = control.Do(
                'if-else-all',
                control.Do('count', cond_net),
                control.If('myIfElse', cond_net.Proto().external_output[-1],
                           self.cnt_net_, self.cnt_2_net_))
        else:
            step = control.If('myIfElse', cond_net,
                              self.cnt_net_, self.cnt_2_net_)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(run_net, expect)])

def testIfElseCondTrueOnNet(self):
        self.IfElseCondTest(self.true_cond_net_, True, 1, False)

def testIfElseCondTrueOnBlob(self):
        self.IfElseCondTest(self.true_cond_net_, True, 1, True)

def testIfElseCondFalseOnNet(self):
        self.IfElseCondTest(self.false_cond_net_, False, 2, False)

def testIfElseCondFalseOnBlob(self):
        self.IfElseCondTest(self.false_cond_net_, False, 2, True)

def IfNotCondTest(self, cond_net, expect, cond_on_blob):
        if cond_on_blob:
            step = control.Do(
                'if-not',
                control.Do('count', cond_net),
                control.IfNot('myIfNot', cond_net.Proto().external_output[-1],
                              self.cnt_net_))
        else:
            step = control.IfNot('myIfNot', cond_net, self.cnt_net_)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(self.cnt_net_, expect)])

def testIfNotCondTrueOnNet(self):
        self.IfNotCondTest(self.true_cond_net_, 0, False)

def testIfNotCondTrueOnBlob(self):
        self.IfNotCondTest(self.true_cond_net_, 0, True)

def testIfNotCondFalseOnNet(self):
        self.IfNotCondTest(self.false_cond_net_, 1, False)

def testIfNotCondFalseOnBlob(self):
        self.IfNotCondTest(self.false_cond_net_, 1, True)

def IfNotElseCondTest(self, cond_net, cond_value, expect, cond_on_blob):
        if cond_value:
            run_net = self.cnt_2_net_
        else:
            run_net = self.cnt_net_
        if cond_on_blob:
            step = control.Do(
                'if-not-else',
                control.Do('count', cond_net),
                control.IfNot('myIfNotElse',
                              cond_net.Proto().external_output[-1],
                              self.cnt_net_, self.cnt_2_net_))
        else:
            step = control.IfNot('myIfNotElse', cond_net,
                                 self.cnt_net_, self.cnt_2_net_)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(run_net, expect)])

def testIfNotElseCondTrueOnNet(self):
        self.IfNotElseCondTest(self.true_cond_net_, True, 2, False)

def testIfNotElseCondTrueOnBlob(self):
        self.IfNotElseCondTest(self.true_cond_net_, True, 2, True)

def testIfNotElseCondFalseOnNet(self):
        self.IfNotElseCondTest(self.false_cond_net_, False, 1, False)

def testIfNotElseCondFalseOnBlob(self):
        self.IfNotElseCondTest(self.false_cond_net_, False, 1, True)

def testSwitch(self):
        step = control.Switch(
            'mySwitch',
            (self.false_cond_net_, self.cnt_net_),
            (self.true_cond_net_, self.cnt_2_net_)
        )
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(self.cnt_net_, 0), (self.cnt_2_net_, 2)])

def testSwitchNot(self):
        step = control.SwitchNot(
            'mySwitchNot',
            (self.false_cond_net_, self.cnt_net_),
            (self.true_cond_net_, self.cnt_2_net_)
        )
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(self.cnt_net_, 1), (self.cnt_2_net_, 0)])

def testBoolNet(self):
        bool_net = control.BoolNet(('a', True))
        step = control.Do('bool', bool_net)
        self.BuildAndRunPlan(step)
        self.CheckNetAllOutput(bool_net, [True])

bool_net = control.BoolNet(('a', True), ('b', False))
        step = control.Do('bool', bool_net)
        self.BuildAndRunPlan(step)
        self.CheckNetAllOutput(bool_net, [True, False])

bool_net = control.BoolNet([('a', True), ('b', False)])
        step = control.Do('bool', bool_net)
        self.BuildAndRunPlan(step)
        self.CheckNetAllOutput(bool_net, [True, False])

def testCombineConditions(self):
        # combined by 'Or'
        combine_net = control.CombineConditions(
            'test', [self.true_cond_net_, self.false_cond_net_], 'Or')
        step = control.Do('combine',
                          self.true_cond_net_,
                          self.false_cond_net_,
                          combine_net)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(combine_net, True)])

# combined by 'And'
        combine_net = control.CombineConditions(
            'test', [self.true_cond_net_, self.false_cond_net_], 'And')
        step = control.Do('combine',
                          self.true_cond_net_,
                          self.false_cond_net_,
                          combine_net)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(combine_net, False)])

def testMergeConditionNets(self):
        # merged by 'Or'
        merge_net = control.MergeConditionNets(
            'test', [self.true_cond_net_, self.false_cond_net_], 'Or')
        step = control.Do('merge', merge_net)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(merge_net, True)])

# merged by 'And'
        merge_net = control.MergeConditionNets(
            'test', [self.true_cond_net_, self.false_cond_net_], 'And')
        step = control.Do('merge', merge_net)
        self.BuildAndRunPlan(step)
        self.CheckNetOutput([(merge_net, False)])

from caffe2.python import control, core, test_util, workspace import logging logger = logging.getLogger(__name__) class TestControl(test_util.TestCase): def setUp(self): super(TestControl, self).setUp() self.N_ = 10 self.init_net_ = core.Net("init-net") cnt = self.init_net_.CreateCounter([], init_count=0) const_n = self.init_net_.ConstantFill( [], shape=[], value=self.N_, dtype=core.DataType.INT64) const_0 = self.init_net_.ConstantFill( [], shape=[], value=0, dtype=core.DataType.INT64) self.cnt_net_ = core.Net("cnt-net") self.cnt_net_.CountUp([cnt]) curr_cnt = self.cnt_net_.RetrieveCount([cnt]) self.init_net_.ConstantFill( [], [curr_cnt], shape=[], value=0, dtype=core.DataType.INT64) self.cnt_net_.AddExternalOutput(curr_cnt) self.cnt_2_net_ = core.Net("cnt-2-net") self.cnt_2_net_.CountUp([cnt]) self.cnt_2_net_.CountUp([cnt]) curr_cnt_2 = self.cnt_2_net_.RetrieveCount([cnt]) self.init_net_.ConstantFill( [], [curr_cnt_2], shape=[], value=0, dtype=core.DataType.INT64) self.cnt_2_net_.AddExternalOutput(curr_cnt_2) self.cond_net_ = core.Net("cond-net") cond_blob = self.cond_net_.LT([curr_cnt, const_n]) self.cond_net_.AddExternalOutput(cond_blob) self.not_cond_net_ = core.Net("not-cond-net") cond_blob = self.not_cond_net_.GE([curr_cnt, const_n]) self.not_cond_net_.AddExternalOutput(cond_blob) self.true_cond_net_ = core.Net("true-cond-net") true_blob = self.true_cond_net_.LT([const_0, const_n]) self.true_cond_net_.AddExternalOutput(true_blob) self.false_cond_net_ = core.Net("false-cond-net") false_blob = self.false_cond_net_.GT([const_0, const_n]) self.false_cond_net_.AddExternalOutput(false_blob) self.idle_net_ = core.Net("idle-net") self.idle_net_.ConstantFill( [], shape=[], value=0, dtype=core.DataType.INT64) def CheckNetOutput(self, nets_and_expects): """ Check the net output is expected nets_and_expects is a list of tuples (net, expect) """ for net, expect in nets_and_expects: output = workspace.FetchBlob( net.Proto().external_output[-1]) self.assertEqual(output, expect) def CheckNetAllOutput(self, net, expects): """ Check the net output is expected expects is a list of bools. """ self.assertEqual(len(net.Proto().external_output), len(expects)) for i in range(len(expects)): output = workspace.FetchBlob( net.Proto().external_output[i]) self.assertEqual(output, expects[i]) def BuildAndRunPlan(self, step): plan = core.Plan("test") plan.AddStep(control.Do('init', self.init_net_)) plan.AddStep(step) self.assertEqual(workspace.RunPlan(plan), True) def ForLoopTest(self, nets_or_steps): step = control.For('myFor', nets_or_steps, self.N_) self.BuildAndRunPlan(step) self.CheckNetOutput([(self.cnt_net_, self.N_)]) def testForLoopWithNets(self): self.ForLoopTest(self.cnt_net_) self.ForLoopTest([self.cnt_net_, self.idle_net_]) def testForLoopWithStep(self): step = control.Do('count', self.cnt_net_) self.ForLoopTest(step) self.ForLoopTest([step, self.idle_net_]) def WhileLoopTest(self, nets_or_steps): step = control.While('myWhile', self.cond_net_, nets_or_steps) self.BuildAndRunPlan(step) self.CheckNetOutput([(self.cnt_net_, self.N_)]) def testWhileLoopWithNet(self): self.WhileLoopTest(self.cnt_net_) self.WhileLoopTest([self.cnt_net_, self.idle_net_]) def testWhileLoopWithStep(self): step = control.Do('count', self.cnt_net_) self.WhileLoopTest(step) self.WhileLoopTest([step, self.idle_net_]) def UntilLoopTest(self, nets_or_steps): step = control.Until('myUntil', self.not_cond_net_, nets_or_steps) self.BuildAndRunPlan(step) self.CheckNetOutput([(self.cnt_net_, self.N_)]) def testUntilLoopWithNet(self): self.UntilLoopTest(self.cnt_net_) self.UntilLoopTest([self.cnt_net_, self.idle_net_]) def testUntilLoopWithStep(self): step = control.Do('count', self.cnt_net_) self.UntilLoopTest(step) self.UntilLoopTest([step, self.idle_net_]) def DoWhileLoopTest(self, nets_or_steps): step = control.DoWhile('myDoWhile', self.cond_net_, nets_or_steps) self.BuildAndRunPlan(step) self.CheckNetOutput([(self.cnt_net_, self.N_)]) def testDoWhileLoopWithNet(self): self.DoWhileLoopTest(self.cnt_net_) self.DoWhileLoopTest([self.idle_net_, self.cnt_net_]) def testDoWhileLoopWithStep(self): step = control.Do('count', self.cnt_net_) self.DoWhileLoopTest(step) self.DoWhileLoopTest([self.idle_net_, step]) def DoUntilLoopTest(self, nets_or_steps): step = control.DoUntil('myDoUntil', self.not_cond_net_, nets_or_steps) self.BuildAndRunPlan(step) self.CheckNetOutput([(self.cnt_net_, self.N_)]) def testDoUntilLoopWithNet(self): self.DoUntilLoopTest(self.cnt_net_) self.DoUntilLoopTest([self.cnt_net_, self.idle_net_]) def testDoUntilLoopWithStep(self): step = control.Do('count', self.cnt_net_) self.DoUntilLoopTest(step) self.DoUntilLoopTest([self.idle_net_, step]) def IfCondTest(self, cond_net, expect, cond_on_blob): if cond_on_blob: step = control.Do( 'if-all', control.Do('count', cond_net), control.If('myIf', cond_net.Proto().external_output[-1], self.cnt_net_)) else: step = control.If('myIf', cond_net, self.cnt_net_) self.BuildAndRunPlan(step) self.CheckNetOutput([(self.cnt_net_, expect)]) def testIfCondTrueOnNet(self): self.IfCondTest(self.true_cond_net_, 1, False) def testIfCondTrueOnBlob(self): self.IfCondTest(self.true_cond_net_, 1, True) def testIfCondFalseOnNet(self): self.IfCondTest(self.false_cond_net_, 0, False) def testIfCondFalseOnBlob(self): self.IfCondTest(self.false_cond_net_, 0, True) def IfElseCondTest(self, cond_net, cond_value, expect, cond_on_blob): if cond_value: run_net = self.cnt_net_ else: run_net = self.cnt_2_net_ if cond_on_blob: step = control.Do( 'if-else-all', control.Do('count', cond_net), control.If('myIfElse', cond_net.Proto().external_output[-1], self.cnt_net_, self.cnt_2_net_)) else: step = control.If('myIfElse', cond_net, self.cnt_net_, self.cnt_2_net_) self.BuildAndRunPlan(step) self.CheckNetOutput([(run_net, expect)]) def testIfElseCondTrueOnNet(self): self.IfElseCondTest(self.true_cond_net_, True, 1, False) def testIfElseCondTrueOnBlob(self): self.IfElseCondTest(self.true_cond_net_, True, 1, True) def testIfElseCondFalseOnNet(self): self.IfElseCondTest(self.false_cond_net_, False, 2, False) def testIfElseCondFalseOnBlob(self): self.IfElseCondTest(self.false_cond_net_, False, 2, True) def IfNotCondTest(self, cond_net, expect, cond_on_blob): if cond_on_blob: step = control.Do( 'if-not', control.Do('count', cond_net), control.IfNot('myIfNot', cond_net.Proto().external_output[-1], self.cnt_net_)) else: step = control.IfNot('myIfNot', cond_net, self.cnt_net_) self.BuildAndRunPlan(step) self.CheckNetOutput([(self.cnt_net_, expect)]) def testIfNotCondTrueOnNet(self): self.IfNotCondTest(self.true_cond_net_, 0, False) def testIfNotCondTrueOnBlob(self): self.IfNotCondTest(self.true_cond_net_, 0, True) def testIfNotCondFalseOnNet(self): self.IfNotCondTest(self.false_cond_net_, 1, False) def testIfNotCondFalseOnBlob(self): self.IfNotCondTest(self.false_cond_net_, 1, True) def IfNotElseCondTest(self, cond_net, cond_value, expect, cond_on_blob): if cond_value: run_net = self.cnt_2_net_ else: run_net = self.cnt_net_ if cond_on_blob: step = control.Do( 'if-not-else', control.Do('count', cond_net), control.IfNot('myIfNotElse', cond_net.Proto().external_output[-1], self.cnt_net_, self.cnt_2_net_)) else: step = control.IfNot('myIfNotElse', cond_net, self.cnt_net_, self.cnt_2_net_) self.BuildAndRunPlan(step) self.CheckNetOutput([(run_net, expect)]) def testIfNotElseCondTrueOnNet(self): self.IfNotElseCondTest(self.true_cond_net_, True, 2, False) def testIfNotElseCondTrueOnBlob(self): self.IfNotElseCondTest(self.true_cond_net_, True, 2, True) def testIfNotElseCondFalseOnNet(self): self.IfNotElseCondTest(self.false_cond_net_, False, 1, False) def testIfNotElseCondFalseOnBlob(self): self.IfNotElseCondTest(self.false_cond_net_, False, 1, True) def testSwitch(self): step = control.Switch( 'mySwitch', (self.false_cond_net_, self.cnt_net_), (self.true_cond_net_, self.cnt_2_net_) ) self.BuildAndRunPlan(step) self.CheckNetOutput([(self.cnt_net_, 0), (self.cnt_2_net_, 2)]) def testSwitchNot(self): step = control.SwitchNot( 'mySwitchNot', (self.false_cond_net_, self.cnt_net_), (self.true_cond_net_, self.cnt_2_net_) ) self.BuildAndRunPlan(step) self.CheckNetOutput([(self.cnt_net_, 1), (self.cnt_2_net_, 0)]) def testBoolNet(self): bool_net = control.BoolNet(('a', True)) step = control.Do('bool', bool_net) self.BuildAndRunPlan(step) self.CheckNetAllOutput(bool_net, [True]) bool_net = control.BoolNet(('a', True), ('b', False)) step = control.Do('bool', bool_net) self.BuildAndRunPlan(step) self.CheckNetAllOutput(bool_net, [True, False]) bool_net = control.BoolNet([('a', True), ('b', False)]) step = control.Do('bool', bool_net) self.BuildAndRunPlan(step) self.CheckNetAllOutput(bool_net, [True, False]) def testCombineConditions(self): # combined by 'Or' combine_net = control.CombineConditions( 'test', [self.true_cond_net_, self.false_cond_net_], 'Or') step = control.Do('combine', self.true_cond_net_, self.false_cond_net_, combine_net) self.BuildAndRunPlan(step) self.CheckNetOutput([(combine_net, True)]) # combined by 'And' combine_net = control.CombineConditions( 'test', [self.true_cond_net_, self.false_cond_net_], 'And') step = control.Do('combine', self.true_cond_net_, self.false_cond_net_, combine_net) self.BuildAndRunPlan(step) self.CheckNetOutput([(combine_net, False)]) def testMergeConditionNets(self): # merged by 'Or' merge_net = control.MergeConditionNets( 'test', [self.true_cond_net_, self.false_cond_net_], 'Or') step = control.Do('merge', merge_net) self.BuildAndRunPlan(step) self.CheckNetOutput([(merge_net, True)]) # merged by 'And' merge_net = control.MergeConditionNets( 'test', [self.true_cond_net_, self.false_cond_net_], 'And') step = control.Do('merge', merge_net) self.BuildAndRunPlan(step) self.CheckNetOutput([(merge_net, False)])