class_docstr: PipelineEngine class that computes each term independently.
  
  Parameters
  ----------
  get_loader : callable
      A function that is given a loadable term and returns a PipelineLoader
      to use to retrieve raw data for that term.
  calendar : DatetimeIndex
      Array of dates to consider as trading days when computing a range
      between a fixed start and end.
  asset_finder : catalyst.assets.AssetFinder
      An AssetFinder instance.  We depend on the AssetFinder to determine
      which assets are in the top-level universe at any point in time.
  populate_initial_workspace : callable, optional
      A function which will be used to populate the initial workspace when
      computing a pipeline. See
      :func:`catalyst.pipeline.engine.default_populate_initial_workspace`
      for more info.
  
  See Also
  --------
  :func:`catalyst.pipeline.engine.default_populate_initial_workspace`
docstr: Compute a pipeline.
  
  The algorithm implemented here can be broken down into the following
  stages:
  
  0. Build a dependency graph of all terms in `pipeline`.  Topologically
     sort the graph to determine an order in which we can compute the
     terms.
  
  1. Ask our AssetFinder for a "lifetimes matrix", which should contain,
     for each date between start_date and end_date, a boolean value for
     each known asset indicating whether the asset existed on that date.
  
  2. Compute each term in the dependency order determined in (0), caching
     the results in a a dictionary to that they can be fed into future
     terms.
  
  3. For each date, determine the number of assets passing
     pipeline.screen.  The sum, N, of all these values is the total
     number of rows in our output frame, so we pre-allocate an output
     array of length N for each factor in `terms`.
  
  4. Fill in the arrays allocated in (3) by copying computed values from
     our output cache into the corresponding rows.
  
  5. Stick the values computed in (4) into a DataFrame and return it.
  
  Step 0 is performed by ``Pipeline.to_graph``.
  Step 1 is performed in ``SimplePipelineEngine._compute_root_mask``.
  Step 2 is performed in ``SimplePipelineEngine.compute_chunk``.
  Steps 3, 4, and 5 are performed in ``SimplePiplineEngine._to_narrow``.
  
  Parameters
  ----------
  pipeline : catalyst.pipeline.Pipeline
      The pipeline to run.
  start_date : pd.Timestamp
      Start date of the computed matrix.
  end_date : pd.Timestamp
      End date of the computed matrix.
  
  Returns
  -------
  result : pd.DataFrame
      A frame of computed results.
  
      The ``result`` columns correspond to the entries of
      `pipeline.columns`, which should be a dictionary mapping strings to
      instances of :class:`catalyst.pipeline.term.Term`.
  
      For each date between ``start_date`` and ``end_date``, ``result``
      will contain a row for each asset that passed `pipeline.screen`.
      A screen of ``None`` indicates that a row should be returned for
      each asset that existed each day.
  
  See Also
  --------
  :meth:`catalyst.pipeline.engine.PipelineEngine.run_pipeline`
  :meth:`catalyst.pipeline.engine.PipelineEngine.run_chunked_pipeline`
docstr_tok: ["Compute","a","pipeline",".","The","algorithm","implemented","here","can","be","broken","down","into","the","following","stages",":","0",".","Build","a","dependency","graph","of","all","terms","in","`","pipeline","`",".","Topologically","sort","the","graph","to","determine","an","order","in","which","we","can","compute","the","terms",".","1",".","Ask","our","AssetFinder","for","a","``","lifetimes","matrix","''",",","which","should","contain",",","for","each","date","between","start_date","and","end_date",",","a","boolean","value","for","each","known","asset","indicating","whether","the","asset","existed","on","that","date",".","2",".","Compute","each","term","in","the","dependency","order","determined","in","(","0",")",",","caching","the","results","in","a","a","dictionary","to","that","they","can","be","fed","into","future","terms",".","3",".","For","each","date",",","determine","the","number","of","assets","passing","pipeline.screen",".","The","sum",",","N",",","of","all","these","values","is","the","total","number","of","rows","in","our","output","frame",",","so","we","pre-allocate","an","output","array","of","length","N","for","each","factor","in","`","terms","`",".","4",".","Fill","in","the","arrays","allocated","in","(","3",")","by","copying","computed","values","from","our","output","cache","into","the","corresponding","rows",".","5",".","Stick","the","values","computed","in","(","4",")","into","a","DataFrame","and","return","it",".","Step","0","is","performed","by","``","Pipeline.to_graph","``",".","Step","1","is","performed","in","``","SimplePipelineEngine._compute_root_mask","``",".","Step","2","is","performed","in","``","SimplePipelineEngine.compute_chunk","``",".","Steps","3",",","4",",","and","5","are","performed","in","``","SimplePiplineEngine._to_narrow","``",".","Parameters","--","--","--","--","--","pipeline",":","catalyst.pipeline.Pipeline","The","pipeline","to","run",".","start_date",":","pd.Timestamp","Start","date","of","the","computed","matrix",".","end_date",":","pd.Timestamp","End","date","of","the","computed","matrix",".","Returns","--","--","--","-","result",":","pd.DataFrame","A","frame","of","computed","results",".","The","``","result","``","columns","correspond","to","the","entries","of","`","pipeline.columns","`",",","which","should","be","a","dictionary","mapping","strings","to","instances","of",":","class",":","`","catalyst.pipeline.term.Term","`",".","For","each","date","between","``","start_date","``","and","``","end_date","``",",","``","result","``","will","contain","a","row","for","each","asset","that","passed","`","pipeline.screen","`",".","A","screen","of","``","None","``","indicates","that","a","row","should","be","returned","for","each","asset","that","existed","each","day",".","See","Also","--","--","--","--",":","meth",":","`","catalyst.pipeline.engine.PipelineEngine.run_pipeline","`",":","meth",":","`","catalyst.pipeline.engine.PipelineEngine.run_chunked_pipeline","`"]
code: def run_pipeline(self, pipeline, start_date, end_date):
      """
      Compute a pipeline.
  
      The algorithm implemented here can be broken down into the following
      stages:
  
      0. Build a dependency graph of all terms in `pipeline`.  Topologically
         sort the graph to determine an order in which we can compute the
         terms.
  
      1. Ask our AssetFinder for a "lifetimes matrix", which should contain,
         for each date between start_date and end_date, a boolean value for
         each known asset indicating whether the asset existed on that date.
  
      2. Compute each term in the dependency order determined in (0), caching
         the results in a a dictionary to that they can be fed into future
         terms.
  
      3. For each date, determine the number of assets passing
         pipeline.screen.  The sum, N, of all these values is the total
         number of rows in our output frame, so we pre-allocate an output
         array of length N for each factor in `terms`.
  
      4. Fill in the arrays allocated in (3) by copying computed values from
         our output cache into the corresponding rows.
  
      5. Stick the values computed in (4) into a DataFrame and return it.
  
      Step 0 is performed by ``Pipeline.to_graph``.
      Step 1 is performed in ``SimplePipelineEngine._compute_root_mask``.
      Step 2 is performed in ``SimplePipelineEngine.compute_chunk``.
      Steps 3, 4, and 5 are performed in ``SimplePiplineEngine._to_narrow``.
  
      Parameters
      ----------
      pipeline : catalyst.pipeline.Pipeline
          The pipeline to run.
      start_date : pd.Timestamp
          Start date of the computed matrix.
      end_date : pd.Timestamp
          End date of the computed matrix.
  
      Returns
      -------
      result : pd.DataFrame
          A frame of computed results.
  
          The ``result`` columns correspond to the entries of
          `pipeline.columns`, which should be a dictionary mapping strings to
          instances of :class:`catalyst.pipeline.term.Term`.
  
          For each date between ``start_date`` and ``end_date``, ``result``
          will contain a row for each asset that passed `pipeline.screen`.
          A screen of ``None`` indicates that a row should be returned for
          each asset that existed each day.
  
      See Also
      --------
      :meth:`catalyst.pipeline.engine.PipelineEngine.run_pipeline`
      :meth:`catalyst.pipeline.engine.PipelineEngine.run_chunked_pipeline`
      """
      if end_date < start_date:
          raise ValueError(
              "start_date must be before or equal to end_date \n"
              "start_date=%s, end_date=%s" % (start_date, end_date)
          )
  
      screen_name = uuid4().hex
      graph = pipeline.to_execution_plan(
          screen_name,
          self._root_mask_term,
          self._calendar,
          start_date,
          end_date,
      )
      extra_rows = graph.extra_rows[self._root_mask_term]
      root_mask = self._compute_root_mask(
          start_date, end_date, extra_rows
      )
      dates, assets, root_mask_values = explode(root_mask)
  
      initial_workspace = self._populate_initial_workspace(
          {
              self._root_mask_term: root_mask_values,
              self._root_mask_dates_term: as_column(dates.values),
          },
          self._root_mask_term,
          graph,
          dates,
          assets,
      )
  
      results = self.compute_chunk(
          graph,
          dates,
          assets,
          initial_workspace,
      )
  
      return self._to_narrow(
          graph.outputs,
          results,
          results.pop(screen_name),
          dates[extra_rows:],
          assets,
      )
  
imports: from abc import ABCMeta, abstractmethod
  from uuid import uuid4
  import six
  from six import iteritems, with_metaclass
  from numpy import array
  from pandas import DataFrame, MultiIndex
  from toolz import groupby, juxt
  from toolz.curried.operator import getitem
  from catalyst.lib.adjusted_array import ensure_adjusted_array, ensure_ndarray
  from catalyst.errors import NoFurtherDataError
  from catalyst.utils.numpy_utils import as_column, repeat_first_axis, repeat_last_axis
  from catalyst.utils.pandas_utils import explode
  from .term import AssetExists, InputDates, LoadableTerm
  from catalyst.utils.date_utils import compute_date_range_chunks
  from catalyst.utils.pandas_utils import categorical_df_concat
  from catalyst.utils.sharedoc import copydoc
label_desc: Use image node_id 2 for calling the SimplePipelineEngine obj's underlying member method code with example usage: obj.run_pipeline(pipeline, start_date, end_date) and returns: self