alpaca/alpacka/batch_steppers.py
def _batch_coroutines(self, cors):
        """Batches a list of coroutines into one.

        Handles waiting for the slowest coroutine and filling blanks in
        prediction requests.
        """
        # Store the final episodes in a list.
        episodes = [None] * len(cors)

        def store_transitions(i, cor):
            episodes[i] = yield from cor
            # End with an infinite stream of Nones, so we don't have
            # to deal with StopIteration later on.
            while True:
                yield None
        cors = [store_transitions(i, cor) for(i, cor) in enumerate(cors)]

        def all_finished(xs):
            return all(x is None for x in xs)

        def batch_requests(xs):
            assert xs

            # Request used as a filler for coroutines that have already
            # finished.
            filler = next(x for x in xs if x is not None)
            # Fill with 0s for easier debugging.
            filler = data.nested_map(np.zeros_like, filler)

            # Substitute the filler for Nones.
            xs = [x if x is not None else filler for x in xs]

            def assert_not_scalar(x):
                assert np.array(x).shape, (
                    'All arrays in a PredictRequest must be at least rank 1.'
                )
            data.nested_map(assert_not_scalar, xs)

            # Stack instead of concatenate to ensure that all requests have
            # the same shape.
            x = data.nested_stack(xs)

            def flatten_first_2_dims(x):
                return np.reshape(x, (-1,) + x.shape[2:])
            # (n_agents, n_requests, ...) -> (n_agents * n_requests, ...)
            return data.nested_map(flatten_first_2_dims, x)

        def unbatch_responses(x):
            def unflatten_first_2_dims(x):
                return np.reshape(
                    x, (len(self._envs_and_agents), -1) + x.shape[1:]
                )
            # (n_agents * n_requests, ...) -> (n_agents, n_requests, ...)
            return data.nested_unstack(
                data.nested_map(unflatten_first_2_dims, x)
            )

        requests = [next(cor) for (i, cor) in enumerate(cors)]
        while not all_finished(requests):
            responses = yield batch_requests(requests)
            requests = [
                cor.send(inp)
                for (i, (cor, inp)) in enumerate(
                    zip(cors, unbatch_responses(responses))
                )
            ]
        return episodes
Batches a list of coroutines into one.

        Handles waiting for the slowest coroutine and filling blanks in
        prediction requests.
        
Batches a list of coroutines into one.
Handles waiting for the slowest coroutine and filling blanks in
prediction requests.
Batches a list of coroutines into one . Handles waiting for the slowest coroutine and filling blanks in prediction requests .
def _batch_coroutines(self, cors):
        episodes = [None] * len(cors)
        def store_transitions(i, cor):
            episodes[i] = yield from cor
            while True:
                yield None
        cors = [store_transitions(i, cor) for(i, cor) in enumerate(cors)]
        def all_finished(xs):
            return all(x is None for x in xs)
        def batch_requests(xs):
            assert xs
            filler = next(x for x in xs if x is not None)
            filler = data.nested_map(np.zeros_like, filler)
            xs = [x if x is not None else filler for x in xs]
            def assert_not_scalar(x):
                assert np.array(x).shape, (
                    'All arrays in a PredictRequest must be at least rank 1.'
                )
            data.nested_map(assert_not_scalar, xs)
            x = data.nested_stack(xs)
            def flatten_first_2_dims(x):
                return np.reshape(x, (-1,) + x.shape[2:])
            return data.nested_map(flatten_first_2_dims, x)
        def unbatch_responses(x):
            def unflatten_first_2_dims(x):
                return np.reshape(
                    x, (len(self._envs_and_agents), -1) + x.shape[1:]
                )
            return data.nested_unstack(
                data.nested_map(unflatten_first_2_dims, x)
            )
        requests = [next(cor) for (i, cor) in enumerate(cors)]
        while not all_finished(requests):
            responses = yield batch_requests(requests)
            requests = [
                cor.send(inp)
                for (i, (cor, inp)) in enumerate(
                    zip(cors, unbatch_responses(responses))
                )
            ]
        return episodes
def _batch_coroutines ( self , cors ) : episodes = [ None ] * len ( cors ) def store_transitions ( i , cor ) : episodes [ i ] = yield from cor while True : yield None cors = [ store_transitions ( i , cor ) for ( i , cor ) in enumerate ( cors ) ] def all_finished ( xs ) : return all ( x is None for x in xs ) def batch_requests ( xs ) : assert xs filler = next ( x for x in xs if x is not None ) filler = data . nested_map ( np . zeros_like , filler ) xs = [ x if x is not None else filler for x in xs ] def assert_not_scalar ( x ) : assert np . array ( x ) . shape , ( 'All arrays in a PredictRequest must be at least rank 1.' ) data . nested_map ( assert_not_scalar , xs ) x = data . nested_stack ( xs ) def flatten_first_2_dims ( x ) : return np . reshape ( x , ( - 1 , ) + x . shape [ 2 : ] ) return data . nested_map ( flatten_first_2_dims , x ) def unbatch_responses ( x ) : def unflatten_first_2_dims ( x ) : return np . reshape ( x , ( len ( self . _envs_and_agents ) , - 1 ) + x . shape [ 1 : ] ) return data . nested_unstack ( data . nested_map ( unflatten_first_2_dims , x ) ) requests = [ next ( cor ) for ( i , cor ) in enumerate ( cors ) ] while not all_finished ( requests ) : responses = yield batch_requests ( requests ) requests = [ cor . send ( inp ) for ( i , ( cor , inp ) ) in enumerate ( zip ( cors , unbatch_responses ( responses ) ) ) ] return episodes
Batches a list of coroutines into one.
Batches a list of coroutines into one .
"""Batches a list of coroutines into one.

        Handles waiting for the slowest coroutine and filling blanks in
        prediction requests.
        """ # Store the final episodes in a list. # End with an infinite stream of Nones, so we don't have # to deal with StopIteration later on. # Request used as a filler for coroutines that have already # finished. # Fill with 0s for easier debugging. # Substitute the filler for Nones. # Stack instead of concatenate to ensure that all requests have # the same shape. # (n_agents, n_requests, ...) -> (n_agents * n_requests, ...) # (n_agents * n_requests, ...) -> (n_agents, n_requests, ...)
self
cors