quantnn/data.py
def run(self):
        """
        Collects batches from child process and puts them on the batch
        queue.
        """
        super().run()

        while True:

            batches = []
            # Collect batches from workers.
            while (self.done_queue.qsize() < len(self.files)):
                for i in range(2):
                    batch_queue = self.batch_queues[i]
                    while batch_queue.qsize():
                        try:
                            b = batch_queue.get()
                            batches.append(b)
                        except FileNotFoundError as e:
                            _LOGGER.warning(
                                "FileNotFoundError occured when retrieving batch from "
                                "loader process.", e
                            )
                            continue
                        except queue.Empty:
                            continue

                        if self.aggregate is None:
                            for b in batches:
                                self.merged_queue.put(b)
                            batches = []
                        else:
                            while len(batches) >= self.aggregate:
                                b = self.aggregate_batches(batches[:self.aggregate])
                                batches = batches[self.aggregate:]
                                self.merged_queue.put(b)

            for i in range(2):
                batch_queue = self.batch_queues[i]
                while batch_queue.qsize():
                    try:
                        b = batch_queue.get()
                        batches.append(b)
                    except FileNotFoundError as e:
                        _LOGGER.warning(
                            "FileNotFoundError occured when retrieving batch from "
                            "loader process.", e
                        )
                        continue
                    except queue.Empty:
                        continue

            # Process remaining batches.
            if self.aggregate is None:
                for b in batches:
                    self.merged_queue.put(b)
                batches = []
            else:
                while batches:
                    b = self.aggregate_batches(batches[:self.aggregate])
                    batches = batches[self.aggregate:]
                    self.merged_queue.put(b)

            self.done_flag.set()
            while self.done_flag.is_set():
                sleep(0.01)

        Collects batches from child process and puts them on the batch
        queue.
        
Collects batches from child process and puts them on the batch
queue.
Collects batches from child process and puts them on the batch queue .
def run(self):
        super().run()
        while True:
            batches = []
            while (self.done_queue.qsize() < len(self.files)):
                for i in range(2):
                    batch_queue = self.batch_queues[i]
                    while batch_queue.qsize():
                        try:
                            b = batch_queue.get()
                            batches.append(b)
                        except FileNotFoundError as e:
                            _LOGGER.warning(
                                "FileNotFoundError occured when retrieving batch from "
                                "loader process.", e
                            )
                            continue
                        except queue.Empty:
                            continue
                        if self.aggregate is None:
                            for b in batches:
                                self.merged_queue.put(b)
                            batches = []
                        else:
                            while len(batches) >= self.aggregate:
                                b = self.aggregate_batches(batches[:self.aggregate])
                                batches = batches[self.aggregate:]
                                self.merged_queue.put(b)
            for i in range(2):
                batch_queue = self.batch_queues[i]
                while batch_queue.qsize():
                    try:
                        b = batch_queue.get()
                        batches.append(b)
                    except FileNotFoundError as e:
                        _LOGGER.warning(
                            "FileNotFoundError occured when retrieving batch from "
                            "loader process.", e
                        )
                        continue
                    except queue.Empty:
                        continue
            if self.aggregate is None:
                for b in batches:
                    self.merged_queue.put(b)
                batches = []
            else:
                while batches:
                    b = self.aggregate_batches(batches[:self.aggregate])
                    batches = batches[self.aggregate:]
                    self.merged_queue.put(b)
            self.done_flag.set()
            while self.done_flag.is_set():
                sleep(0.01)
def run ( self ) : super ( ) . run ( ) while True : batches = [ ] while ( self . done_queue . qsize ( ) < len ( self . files ) ) : for i in range ( 2 ) : batch_queue = self . batch_queues [ i ] while batch_queue . qsize ( ) : try : b = batch_queue . get ( ) batches . append ( b ) except FileNotFoundError as e : _LOGGER . warning ( "FileNotFoundError occured when retrieving batch from " "loader process." , e ) continue except queue . Empty : continue if self . aggregate is None : for b in batches : self . merged_queue . put ( b ) batches = [ ] else : while len ( batches ) >= self . aggregate : b = self . aggregate_batches ( batches [ : self . aggregate ] ) batches = batches [ self . aggregate : ] self . merged_queue . put ( b ) for i in range ( 2 ) : batch_queue = self . batch_queues [ i ] while batch_queue . qsize ( ) : try : b = batch_queue . get ( ) batches . append ( b ) except FileNotFoundError as e : _LOGGER . warning ( "FileNotFoundError occured when retrieving batch from " "loader process." , e ) continue except queue . Empty : continue if self . aggregate is None : for b in batches : self . merged_queue . put ( b ) batches = [ ] else : while batches : b = self . aggregate_batches ( batches [ : self . aggregate ] ) batches = batches [ self . aggregate : ] self . merged_queue . put ( b ) self . done_flag . set ( ) while self . done_flag . is_set ( ) : sleep ( 0.01 )
Collects batches from child process and puts them on the batch
queue.
Collects batches from child process and puts them on the batch queue .
"""
        Collects batches from child process and puts them on the batch
        queue.
        """ # Collect batches from workers. # Process remaining batches.
self