<commit_before>/*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.
 */

#include "pipeline.h"
#include "pipeline_exception.h"

#include "edge.h"

#include <boost/foreach.hpp>

#include <set>

/**
 * \file pipeline.cxx
 *
 * \brief Implementation of the base class for \link pipeline pipelines\endlink.
 */

namespace vistk
{

class pipeline::priv
{
  public:
    priv();
    ~priv();

    typedef std::map<process::name_t, process_t> process_map_t;
    typedef std::pair<process::port_addr_t, process::port_addr_t> connection_t;
    typedef std::vector<connection_t> connections_t;
    typedef std::map<size_t, edge_t> edge_map_t;

    typedef std::map<process::port_t, process::port_addrs_t> input_port_mapping_t;
    typedef std::map<process::port_t, process::port_addr_t> output_port_mapping_t;
    typedef std::pair<input_port_mapping_t, output_port_mapping_t> port_mapping_t;
    typedef std::map<process::name_t, port_mapping_t> group_t;

    connections_t connections;

    process_map_t process_map;
    edge_map_t edge_map;

    group_t groups;
};

pipeline
::pipeline(config_t const& /*config*/)
{
  d = boost::shared_ptr<priv>(new priv);
}

pipeline
::~pipeline()
{
}

void
pipeline
::add_process(process_t process)
{
  if (!process)
  {
    throw null_process_addition();
  }

  process::name_t const name = process->name();

  priv::process_map_t::const_iterator proc_it = d->process_map.find(name);

  if (proc_it != d->process_map.end())
  {
    throw duplicate_process_name(name);
  }

  priv::group_t::const_iterator group_it = d->groups.find(name);

  if (group_it != d->groups.end())
  {
    throw duplicate_process_name(name);
  }

  d->process_map[name] = process;
}

void
pipeline
::add_group(process::name_t const& name)
{
  priv::process_map_t::const_iterator proc_it = d->process_map.find(name);

  if (proc_it != d->process_map.end())
  {
    throw duplicate_process_name(name);
  }

  priv::group_t::const_iterator group_it = d->groups.find(name);

  if (group_it != d->groups.end())
  {
    throw duplicate_process_name(name);
  }

  d->groups[name] = priv::port_mapping_t();
}

void
pipeline
::connect(process::name_t const& upstream_process,
          process::port_t const& upstream_port,
          process::name_t const& downstream_process,
          process::port_t const& downstream_port)
{
  priv::group_t::const_iterator const up_group_it = d->groups.find(upstream_process);

  if (up_group_it != d->groups.end())
  {
    priv::output_port_mapping_t const& mapping = up_group_it->second.second;

    priv::output_port_mapping_t::const_iterator const mapping_it = mapping.find(upstream_port);

    if (mapping_it != mapping.end())
    {
      connect(mapping_it->second.first, mapping_it->second.second,
              downstream_process, downstream_port);

      return;
    }
  }

  priv::group_t::const_iterator const down_group_it = d->groups.find(upstream_process);

  if (down_group_it != d->groups.end())
  {
    priv::input_port_mapping_t const& mapping = down_group_it->second.first;

    priv::input_port_mapping_t::const_iterator const mapping_it = mapping.find(downstream_port);

    if (mapping_it != mapping.end())
    {
      BOOST_FOREACH (process::port_addr_t const& port_addr, mapping_it->second)
      {
        connect(upstream_process, upstream_port,
                port_addr.first, port_addr.second);
      }

      return;
    }
  }

  config_t edge_config = config::empty_config();

  edge_t e = edge_t(new edge(edge_config));

  process::port_addr_t const up_port = process::port_addr_t(upstream_process, upstream_port);
  process::port_addr_t const down_port = process::port_addr_t(downstream_process, downstream_port);
  priv::connection_t const conn = priv::connection_t(up_port, down_port);

  priv::process_map_t::iterator const up_it = d->process_map.find(upstream_process);
  priv::process_map_t::iterator const down_it = d->process_map.find(downstream_process);

  if (up_it == d->process_map.end())
  {
    throw no_such_process(upstream_process);
  }
  if (down_it == d->process_map.end())
  {
    throw no_such_process(downstream_process);
  }

  /// \todo Port type checking here?

  up_it->second->connect_output_port(upstream_port, e);
  down_it->second->connect_input_port(downstream_port, e);

  d->edge_map[d->connections.size()] = e;
  d->connections.push_back(conn);<commit_msg>Make code more readable (no inline pair access)<commit_after>/*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.
 */

#include "pipeline.h"
#include "pipeline_exception.h"

#include "edge.h"

#include <boost/foreach.hpp>

#include <set>

/**
 * \file pipeline.cxx
 *
 * \brief Implementation of the base class for \link pipeline pipelines\endlink.
 */

namespace vistk
{

class pipeline::priv
{
  public:
    priv();
    ~priv();

    typedef std::map<process::name_t, process_t> process_map_t;
    typedef std::pair<process::port_addr_t, process::port_addr_t> connection_t;
    typedef std::vector<connection_t> connections_t;
    typedef std::map<size_t, edge_t> edge_map_t;

    typedef std::map<process::port_t, process::port_addrs_t> input_port_mapping_t;
    typedef std::map<process::port_t, process::port_addr_t> output_port_mapping_t;
    typedef std::pair<input_port_mapping_t, output_port_mapping_t> port_mapping_t;
    typedef std::map<process::name_t, port_mapping_t> group_t;

    connections_t connections;

    process_map_t process_map;
    edge_map_t edge_map;

    group_t groups;
};

pipeline
::pipeline(config_t const& /*config*/)
{
  d = boost::shared_ptr<priv>(new priv);
}

pipeline
::~pipeline()
{
}

void
pipeline
::add_process(process_t process)
{
  if (!process)
  {
    throw null_process_addition();
  }

  process::name_t const name = process->name();

  priv::process_map_t::const_iterator proc_it = d->process_map.find(name);

  if (proc_it != d->process_map.end())
  {
    throw duplicate_process_name(name);
  }

  priv::group_t::const_iterator group_it = d->groups.find(name);

  if (group_it != d->groups.end())
  {
    throw duplicate_process_name(name);
  }

  d->process_map[name] = process;
}

void
pipeline
::add_group(process::name_t const& name)
{
  priv::process_map_t::const_iterator proc_it = d->process_map.find(name);

  if (proc_it != d->process_map.end())
  {
    throw duplicate_process_name(name);
  }

  priv::group_t::const_iterator group_it = d->groups.find(name);

  if (group_it != d->groups.end())
  {
    throw duplicate_process_name(name);
  }

  d->groups[name] = priv::port_mapping_t();
}

void
pipeline
::connect(process::name_t const& upstream_process,
          process::port_t const& upstream_port,
          process::name_t const& downstream_process,
          process::port_t const& downstream_port)
{
  priv::group_t::const_iterator const up_group_it = d->groups.find(upstream_process);

  if (up_group_it != d->groups.end())
  {
    priv::output_port_mapping_t const& mapping = up_group_it->second.second;

    priv::output_port_mapping_t::const_iterator const mapping_it = mapping.find(upstream_port);

    if (mapping_it != mapping.end())
    {
      connect(mapping_it->second.first, mapping_it->second.second,
              downstream_process, downstream_port);

      return;
    }
  }

  priv::group_t::const_iterator const down_group_it = d->groups.find(upstream_process);

  if (down_group_it != d->groups.end())
  {
    priv::input_port_mapping_t const& mapping = down_group_it->second.first;

    priv::input_port_mapping_t::const_iterator const mapping_it = mapping.find(downstream_port);

    if (mapping_it != mapping.end())
    {
      BOOST_FOREACH (process::port_addr_t const& port_addr, mapping_it->second)
      {
        connect(upstream_process, upstream_port,
                port_addr.first, port_addr.second);
      }

      return;
    }
  }

  config_t edge_config = config::empty_config();

  edge_t e = edge_t(new edge(edge_config));

  process::port_addr_t const up_port = process::port_addr_t(upstream_process, upstream_port);
  process::port_addr_t const down_port = process::port_addr_t(downstream_process, downstream_port);
  priv::connection_t const conn = priv::connection_t(up_port, down_port);

  priv::process_map_t::iterator const up_it = d->process_map.find(upstream_process);
  priv::process_map_t::iterator const down_it = d->process_map.find(downstream_process);

  if (up_it == d->process_map.end())
  {
    throw no_such_process(upstream_process);
  }
  if (down_it == d->process_map.end())
  {
    throw no_such_process(downstream_process);
  }

  /// \todo Port type checking here?

  process_t const& up_proc = up_it->second;
  process_t const& down_proc = down_it->second;

  up_proc->connect_output_port(upstream_port, e);
  down_proc->connect_input_port(downstream_port, e);

  d->edge_map[d->connections.size()] = e;
  d->connections.push_back(conn);
{'fraction_non_alphanumeric': 0.1286945141236776, 'fraction_numerical': 0.002617515541498528, 'mean_word_length': 4.520770620108369, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 87, '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': '6251991', 'n_tokens_mistral': 3230, 'n_tokens_neox': 3126, 'n_words': 568}