Class JoinData

tensorpack/dataflow/common.py:522–583

Join the components from each DataFlow. See below for its behavior. Note that you can't join a DataFlow that produces lists with one that produces dicts. Example: .. code-block:: none df1 produces: [c1, c2] df2 produces: [c3, c4] joined: [c1, c2, c3, c4]

class JoinData(DataFlow):
    """
    Join the components from each DataFlow. See below for its behavior.

    Note that you can't join a DataFlow that produces lists with one that produces dicts.

    Example:

    .. code-block:: none

        df1 produces: [c1, c2]
        df2 produces: [c3, c4]
        joined: [c1, c2, c3, c4]

        df1 produces: {"a":c1, "b":c2}
        df2 produces: {"c":c3}
        joined: {"a":c1, "b":c2, "c":c3}
    """

    def __init__(self, df_lists):
        """
        Args:
            df_lists (list): a list of DataFlow.
                When these dataflows have different sizes, JoinData will stop when any
                of them is exhausted.
                The list could contain the same DataFlow instance more than once,
                but note that in that case `__iter__` will then also be called many times.
        """
        self.df_lists = df_lists

        try:
            self._size = len(self.df_lists[0])
            for d in self.df_lists:
                assert len(d) == self._size, \
                    "All DataFlow must have the same size! {} != {}".format(len(d), self._size)
        except Exception:
            logger.info("[JoinData] Size check failed for the list of dataflow to be joined!")

    def reset_state(self):
        for d in set(self.df_lists):
            d.reset_state()

    def __len__(self):
        """
        Return the minimum size among all.
        """
        return min(len(k) for k in self.df_lists)

    def __iter__(self):
        itrs = [k.__iter__() for k in self.df_lists]
        try:
            while True:
                all_dps = [next(itr) for itr in itrs]
                if isinstance(all_dps[0], (list, tuple)):
                    dp = list(itertools.chain(*all_dps))
                else:
                    dp = {}
                    for x in all_dps: