'Internal implementation of `pipeline_output`.'

def _pipeline_output(self, pipeline, chunks):

    today = normalize_date(self.get_datetime())
    data = NO_DATA = object()
    try:
        data = self._pipeline_cache.unwrap(today)
    except Expired:
        pass
    if (data is NO_DATA):
        exc_clear()
        clear_dataframe_indexer_caches(self._pipeline_cache._unsafe_get_value())
        self._pipeline_cache = None
        (data, valid_until) = self._run_pipeline(pipeline, today, next(chunks))
        self._pipeline_cache = CachedObject(data, valid_until)
    try:
        return data.loc[today]
    except KeyError:
        return pd.DataFrame(index=[], columns=data.columns)
