<commit_before>#!/usr/bin/env python
#ckwg +5
# Copyright 2011 by Kitware, Inc. All Rights Reserved. Please refer to
# KITWARE_LICENSE.TXT for licensing information, or contact General Counsel,
# Kitware, Inc., 28 Corporate Drive, Clifton Park, NY 12065.


def log(msg):
    import sys
    sys.stderr.write("%s\n" % msg)


def test_import():
    try:
        import vistk.pipeline.pipeline
    except:
        log("Error: Failed to import the pipeline module")


def test_create():
    from vistk.pipeline import config
    from vistk.pipeline import pipeline

    c = config.empty_config()

    pipeline.Pipeline(c)


def test_api_calls():
    from vistk.pipeline import config
    from vistk.pipeline import edge
    from vistk.pipeline import modules
    from vistk.pipeline import pipeline
    from vistk.pipeline import process
    from vistk.pipeline import process_registry

    c = config.empty_config()

    p = pipeline.Pipeline(c)

    proc_type1 = 'numbers'
    proc_type2 = 'print_number'

    proc_name1 = 'src'
    proc_name2 = 'snk'

    port_name1 = 'number'
    port_name2 = 'number'

    group_name = 'group'
    group_iport = 'iport'
    group_oport = 'oport'

    modules.load_known_modules()

    reg = process_registry.ProcessRegistry.self()

    c.set_value(process_registry.Process.config_name, proc_name1)
    proc1 = reg.create_process(proc_type1, c)

    conf_name = 'output'

    c.set_value(process_registry.Process.config_name, proc_name2)
    c.set_value(conf_name, 'test-python-pipeline-api_calls-print_number.txt')
    proc2 = reg.create_process(proc_type2, c)

    p.add_process(proc1)
    p.add_process(proc2)
    p.add_group(group_name)
    p.connect(proc_name1, port_name1,
              proc_name2, port_name2)
    p.map_input_port(group_name, group_iport,
                     proc_name2, port_name2,
                     process.PortFlags())
    p.map_output_port(group_name, group_oport,
                      proc_name1, port_name1,
                      process.PortFlags())
    p.process_names()
    p.process_by_name(proc_name1)
    p.upstream_for_process(proc_name2)
    p.upstream_for_port(proc_name2, port_name2)
    p.downstream_for_process(proc_name1)
    p.downstream_for_port(proc_name1, port_name1)
    p.sender_for_port(proc_name2, port_name2)
    p.receivers_for_port(proc_name1, port_name1)
    p.edge_for_connection(proc_name1, port_name1,
                          proc_name2, port_name2)
    p.input_edges_for_process(proc_name2)
    p.input_edge_for_port(proc_name2, port_name2)
    p.output_edges_for_process(proc_name1)
    p.output_edges_for_port(proc_name1, port_name1)
    p.groups()
    p.input_ports_for_group(group_name)
    p.output_ports_for_group(group_name)
    p.mapped_group_input_port_flags(group_name, group_iport)
    p.mapped_group_output_port_flags(group_name, group_oport)
    p.mapped_group_input_ports(group_name, group_iport)
    p.mapped_group_output_port(group_name, group_oport)

    p.setup_pipeline()


def main(testname):
    if testname == 'import':
        test_import()
    elif testname == 'create':
        test_create()
    elif testname == 'api_calls':
        test_api_calls()
    else:
        log("Error: No such test '%s'" % testname)


if __name__ == '__main__':
    import os
    import sys

    if not len(sys.argv) == 4:
        log("Error: Expected three arguments")
        sys.exit(1)

    testname = sys.argv[1]<commit_msg>Test the new pipeline API calls from Python<commit_after>#!/usr/bin/env python
#ckwg +5
# Copyright 2011 by Kitware, Inc. All Rights Reserved. Please refer to
# KITWARE_LICENSE.TXT for licensing information, or contact General Counsel,
# Kitware, Inc., 28 Corporate Drive, Clifton Park, NY 12065.


def log(msg):
    import sys
    sys.stderr.write("%s\n" % msg)


def test_import():
    try:
        import vistk.pipeline.pipeline
    except:
        log("Error: Failed to import the pipeline module")


def test_create():
    from vistk.pipeline import config
    from vistk.pipeline import pipeline

    c = config.empty_config()

    pipeline.Pipeline(c)


def test_api_calls():
    from vistk.pipeline import config
    from vistk.pipeline import edge
    from vistk.pipeline import modules
    from vistk.pipeline import pipeline
    from vistk.pipeline import process
    from vistk.pipeline import process_registry

    c = config.empty_config()

    p = pipeline.Pipeline(c)

    proc_type1 = 'numbers'
    proc_type2 = 'print_number'

    proc_name1 = 'src'
    proc_name2 = 'snk'

    port_name1 = 'number'
    port_name2 = 'number'

    group_name = 'group'
    group_iport = 'iport'
    group_oport = 'oport'

    modules.load_known_modules()

    reg = process_registry.ProcessRegistry.self()

    c.set_value(process_registry.Process.config_name, proc_name1)
    proc1 = reg.create_process(proc_type1, c)

    conf_name = 'output'

    c.set_value(process_registry.Process.config_name, proc_name2)
    c.set_value(conf_name, 'test-python-pipeline-api_calls-print_number.txt')
    proc2 = reg.create_process(proc_type2, c)

    p.add_process(proc1)
    p.add_process(proc2)
    p.add_group(group_name)
    p.connect(proc_name1, port_name1,
              proc_name2, port_name2)
    p.map_input_port(group_name, group_iport,
                     proc_name2, port_name2,
                     process.PortFlags())
    p.map_output_port(group_name, group_oport,
                      proc_name1, port_name1,
                      process.PortFlags())
    p.process_names()
    p.process_by_name(proc_name1)
    p.upstream_for_process(proc_name2)
    p.upstream_for_port(proc_name2, port_name2)
    p.downstream_for_process(proc_name1)
    p.downstream_for_port(proc_name1, port_name1)
    p.sender_for_port(proc_name2, port_name2)
    p.receivers_for_port(proc_name1, port_name1)
    p.edge_for_connection(proc_name1, port_name1,
                          proc_name2, port_name2)
    p.input_edges_for_process(proc_name2)
    p.input_edge_for_port(proc_name2, port_name2)
    p.output_edges_for_process(proc_name1)
    p.output_edges_for_port(proc_name1, port_name1)
    p.groups()
    p.input_ports_for_group(group_name)
    p.output_ports_for_group(group_name)
    p.mapped_group_input_port_flags(group_name, group_iport)
    p.mapped_group_output_port_flags(group_name, group_oport)
    p.mapped_group_input_ports(group_name, group_iport)
    p.mapped_group_output_port(group_name, group_oport)

    p.setup_pipeline()

    p.is_setup()
    p.setup_successful()

    p.start()
    p.stop()

    p.reset()


def main(testname):
    if testname == 'import':
        test_import()
    elif testname == 'create':
        test_create()
    elif testname == 'api_calls':
        test_api_calls()
    else:
        log("Error: No such test '%s'" % testname)


if __name__ == '__main__':
    import os
    import sys

    if not len(sys.argv) == 4:
        log("Error: Expected three arguments")
        sys.exit(1)

    testname = sys.argv[1]
{'fraction_non_alphanumeric': 0.09297994269340974, 'fraction_numerical': 0.0166189111747851, 'mean_word_length': 3.202889825406382, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '971023', 'n_tokens_mistral': 2486, 'n_tokens_neox': 2437, 'n_words': 449}