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.
  asset_finder : zipline.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:`zipline.pipeline.engine.default_populate_initial_workspace`
      for more info.
  default_hooks : list, optional
      List of hooks that should be used to instrument all pipelines executed
      by this engine.
  
  See Also
  --------
  :func:`zipline.pipeline.engine.default_populate_initial_workspace`
class_docstr_tok: ["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",".","asset_finder",":","zipline.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",":","`","zipline.pipeline.engine.default_populate_initial_workspace","`","for","more","info",".","default_hooks",":","list",",","optional","List","of","hooks","that","should","be","used","to","instrument","all","pipelines","executed","by","this","engine",".","See","Also","--","--","--","--",":","func",":","`","zipline.pipeline.engine.default_populate_initial_workspace","`"]
docstr: Compute the Pipeline terms in the graph for the requested start and end
  dates.
  
  This is where we do the actual work of running a pipeline.
  
  Parameters
  ----------
  graph : zipline.pipeline.graph.ExecutionPlan
      Dependency graph of the terms to be executed.
  dates : pd.DatetimeIndex
      Row labels for our root mask.
  sids : pd.Int64Index
      Column labels for our root mask.
  workspace : dict
      Map from term -> output.
      Must contain at least entry for `self._root_mask_term` whose shape
      is `(len(dates), len(assets))`, but may contain additional
      pre-computed terms for testing or optimization purposes.
  refcounts : dict[Term, int]
      Dictionary mapping terms to number of dependent terms. When a
      term's refcount hits 0, it can be safely discarded from
      ``workspace``. See TermGraph.decref_dependencies for more info.
  execution_order : list[Term]
      Order in which to execute terms.
  hooks : implements(PipelineHooks)
      Hooks to instrument pipeline execution.
  
  Returns
  -------
  results : dict
      Dictionary mapping requested results to outputs.
docstr_tok: ["Compute","the","Pipeline","terms","in","the","graph","for","the","requested","start","and","end","dates",".","This","is","where","we","do","the","actual","work","of","running","a","pipeline",".","Parameters","--","--","--","--","--","graph",":","zipline.pipeline.graph.ExecutionPlan","Dependency","graph","of","the","terms","to","be","executed",".","dates",":","pd.DatetimeIndex","Row","labels","for","our","root","mask",".","sids",":","pd.Int64Index","Column","labels","for","our","root","mask",".","workspace",":","dict","Map","from","term","-",">","output",".","Must","contain","at","least","entry","for","`","self._root_mask_term","`","whose","shape","is","`","(","len","(","dates",")",",","len","(","assets",")",")","`",",","but","may","contain","additional","pre-computed","terms","for","testing","or","optimization","purposes",".","refcounts",":","dict","[","Term",",","int","]","Dictionary","mapping","terms","to","number","of","dependent","terms",".","When","a","term","'s","refcount","hits","0",",","it","can","be","safely","discarded","from","``","workspace","``",".","See","TermGraph.decref_dependencies","for","more","info",".","execution_order",":","list","[","Term","]","Order","in","which","to","execute","terms",".","hooks",":","implements","(","PipelineHooks",")","Hooks","to","instrument","pipeline","execution",".","Returns","--","--","--","-","results",":","dict","Dictionary","mapping","requested","results","to","outputs","."]
code: def compute_chunk(
      self,
      graph,
      dates,
      sids,
      workspace,
      refcounts,
      execution_order,
      hooks,
  ):
      """
      Compute the Pipeline terms in the graph for the requested start and end
      dates.
  
      This is where we do the actual work of running a pipeline.
  
      Parameters
      ----------
      graph : zipline.pipeline.graph.ExecutionPlan
          Dependency graph of the terms to be executed.
      dates : pd.DatetimeIndex
          Row labels for our root mask.
      sids : pd.Int64Index
          Column labels for our root mask.
      workspace : dict
          Map from term -> output.
          Must contain at least entry for `self._root_mask_term` whose shape
          is `(len(dates), len(assets))`, but may contain additional
          pre-computed terms for testing or optimization purposes.
      refcounts : dict[Term, int]
          Dictionary mapping terms to number of dependent terms. When a
          term's refcount hits 0, it can be safely discarded from
          ``workspace``. See TermGraph.decref_dependencies for more info.
      execution_order : list[Term]
          Order in which to execute terms.
      hooks : implements(PipelineHooks)
          Hooks to instrument pipeline execution.
  
      Returns
      -------
      results : dict
          Dictionary mapping requested results to outputs.
      """
      self._validate_compute_chunk_params(graph, dates, sids, workspace)
  
      get_loader = self._get_loader
  
      # Copy the supplied initial workspace so we don't mutate it in place.
      workspace = workspace.copy()
      domain = graph.domain
  
      # Many loaders can fetch data more efficiently if we ask them to
      # retrieve all their inputs at once. For example, a loader backed by a
      # SQL database can fetch multiple columns from the database in a single
      # query.
      #
      # To enable these loaders to fetch their data efficiently, we group
      # together requests for LoadableTerms if they are provided by the same
      # loader and they require the same number of extra rows.
      #
      # The extra rows condition is a simplification: we don't currently have
      # a mechanism for asking a loader to fetch different windows of data
      # for different terms, so we only batch requests together when they're
      # going to produce data for the same set of dates. That may change in
      # the future if we find a loader that can still benefit significantly
      # from batching unequal-length requests.
      def loader_group_key(term):
          loader = get_loader(term)
          extra_rows = graph.extra_rows[term]
          return loader, extra_rows
  
      # Only produce loader groups for the terms we expect to load.  This
      # ensures that we can run pipelines for graphs where we don't have a
      # loader registered for an atomic term if all the dependencies of that
      # term were supplied in the initial workspace.
      will_be_loaded = graph.loadable_terms - viewkeys(workspace)
      loader_groups = groupby(
          loader_group_key,
          (t for t in execution_order if t in will_be_loaded),
      )
  
      for term in execution_order:
          # `term` may have been supplied in `initial_workspace`, or we may
          # have loaded `term` as part of a batch with another term coming
          # from the same loader (see note on loader_group_key above). In
          # either case, we already have the term computed, so don't
          # recompute.
          if term in workspace:
              continue
  
          # Asset labels are always the same, but date labels vary by how
          # many extra rows are needed.
          mask, mask_dates = graph.mask_and_dates_for_term(
              term,
              self._root_mask_term,
              workspace,
              dates,
          )
  
          if isinstance(term, LoadableTerm):
              loader = get_loader(term)
              to_load = sorted(
                  loader_groups[loader_group_key(term)],
                  key=lambda t: t.dataset,
              )
              self._ensure_can_load(loader, to_load)
              with hooks.loading_terms(to_load):
                  loaded = loader.load_adjusted_array(
                      domain,
                      to_load,
                      mask_dates,
                      sids,
                      mask,
                  )
              assert set(loaded) == set(to_load), (
                  "loader did not return an AdjustedArray for each column\n"
                  "expected: %r\n"
                  "got:      %r"
                  % (
                      sorted(to_load, key=repr),
                      sorted(loaded, key=repr),
                  )
              )
              workspace.update(loaded)
          else:
              with hooks.computing_term(term):
                  workspace[term] = term._compute(
                      self._inputs_for_term(
                          term,
                          workspace,
                          graph,
                          domain,
                          refcounts,
                      ),
                      mask_dates,
                      sids,
                      mask,
                  )
              if term.ndim == 2:
                  assert workspace[term].shape == mask.shape
              else:
                  assert workspace[term].shape == (mask.shape[0], 1)
  
              # Decref dependencies of ``term``, and clear any terms
              # whose refcounts hit 0.
              for garbage in graph.decref_dependencies(term, refcounts):
                  del workspace[garbage]
  
      # At this point, all the output terms are in the workspace.
      out = {}
      graph_extra_rows = graph.extra_rows
      for name, term in iteritems(graph.outputs):
          # Truncate off extra rows from outputs.
          out[name] = workspace[term][graph_extra_rows[term] :]
      return out
  
imports: from abc import ABCMeta, abstractmethod
  from functools import partial
  from six import iteritems, with_metaclass, viewkeys
  from numpy import array, arange
  from pandas import DataFrame, MultiIndex
  from toolz import groupby
  from zipline.lib.adjusted_array import ensure_adjusted_array, ensure_ndarray
  from zipline.errors import NoFurtherDataError
  from zipline.utils.input_validation import expect_types
  from zipline.utils.numpy_utils import as_column, repeat_first_axis, repeat_last_axis
  from zipline.utils.pandas_utils import explode
  from zipline.utils.string_formatting import bulleted_list
  from .domain import Domain, GENERIC
  from .graph import maybe_specialize
  from .hooks import DelegatingHooks
  from .term import AssetExists, InputDates, LoadableTerm
  from zipline.utils.date_utils import compute_date_range_chunks
  from zipline.utils.pandas_utils import categorical_df_concat
label_desc: Use image node_id 7 for calling the SimplePipelineEngine obj's underlying member method code with example usage: obj.compute_chunk(graph, dates, sids, workspace, refcounts, execution_order, hooks) and returns: out, loader,  extra_rows