Note: A ZMQStreamer does not activate its stream,
        but allows the zmq_worker to do that.

        Yields
        ------
        data : dict
            Data drawn from `streamer(max_iter)`.
198005
def iterate(self, max_iter=None):
        
        context = zmq.Context()

        if six.PY2:
            warnings.warn('zmq_stream cannot preserve numpy array alignment '
                          'in Python 2', RuntimeWarning)

        try:
            socket = context.socket(zmq.PAIR)

            port = socket.bind_to_random_port('tcp://*',
                                              min_port=self.min_port,
                                              max_port=self.max_port,
                                              max_tries=self.max_tries)
            terminate = mp.Event()

            worker = mp.Process(target=SafeFunction(zmq_worker),
                                args=[port, self.streamer, terminate],
                                kwargs=dict(copy=self.copy,
                                            max_iter=max_iter))

            worker.daemon = True
            worker.start()

            # Yield from the queue as long as it's open
            while True:
                yield zmq_recv_data(socket)

        except StopIteration:
            pass

        except:
            # pylint: disable-msg=W0702
            six.reraise(*sys.exc_info())

        finally:
            terminate.set()
            worker.join(self.timeout)
            if worker.is_alive():
                worker.terminate()
            context.destroy()
1003029
def create_dataset(self, in_stream):
        
        assert XLSX_IMPORT
        from io import BytesIO
        xlsx_book = openpyxl.load_workbook(BytesIO(in_stream), read_only=True)

        dataset = tablib.Dataset()
        sheet = xlsx_book.active

        # obtain generator
        rows = sheet.rows
        dataset.headers = [cell.value for cell in next(rows)]

        for row in rows:
            row_values = [cell.value for cell in row]
            dataset.append(row_values)
        return dataset
1246961
def load(stream, Loader=None):
    
    if Loader is None:
        load_warning('load')
        Loader = FullLoader

    loader = Loader(stream)
    try:
        return loader.get_single_data()
    finally:
        loader.dispose()
969094
def load(stream, Loader=Loader):
    
    loader = Loader(stream)
    try:
        return loader.get_single_data()
    finally:
        loader.dispose()
1282072
def write_to_stream(self, stream_id, data, sandbox=None):
        
        if sandbox is not None:
            raise NotImplementedError

        if stream_id not in self.streams:
            raise StreamNotFoundError("Stream with id '{}' does not exist".format(stream_id))

        writer = self.get_stream_writer(self.streams[stream_id])

        if isinstance(data, StreamInstance):
            data = [data]

        for instance in data:
            if not isinstance(instance, StreamInstance):
                raise ValueError("Expected StreamInstance, got {}".format(str(type(instance))))
            writer(instance)
75500
def streams(self):
        

        if not self.get_objects_by_type('stream'):
            tpld_ids = []
            for index in self.get_attribute('ps_indices').split():
                stream = XenaStream(parent=self, index='{}/{}'.format(self.index, index))
                tpld_ids.append(stream.get_attribute('ps_tpldid'))
            if tpld_ids:
                XenaStream.next_tpld_id = max([XenaStream.next_tpld_id] + [int(t) for t in tpld_ids]) + 1
        return {s.id: s for s in self.get_objects_by_type('stream')}
276415
def _create_listening_stream(self, pull_addr):
        
        sock = self._zmq_context.socket(zmq.PULL)
        sock.connect(pull_addr)
        stream = ZMQStream(sock, io_loop=self.io_loop)

        return stream
9828
async def read_data(self, stream_id: int) -> bytes:
        
        frames = [f async for f in self.stream_frames(stream_id)]
        return b''.join(frames)
635768
def copy(self):
    
    a, b = it.tee(self._data) # 2 generators, not thread-safe
    self._data = a
    return Stream(b)
621662
def parse(self, data: bytes, context=None):
        
        stream = BytesIO(data)
        return self.parse_stream(stream, context)
568436
def copy(self):
    
    if self._iters:
      a, b = it.tee(self._iters[0])
      self._iters[0] = a
      return Stream(b)
    iter(self)
603052
def from_yaml(*streams):
    
    return from_dict(merge_dicts(*map(
        lambda f: yaml.load(f, Loader=Loader),
        list(streams)
    )))
6096
def data(self):
        
        try:
            bytestream = super(MimeData, self).data(self._mimeType).data()
            return pickle.loads(bytestream)
        except:
            raise
320856
def get_streams(self):
        
        self._write_lock.acquire()
        try:
            return itervalues(self._stream_by_id)
        finally:
            self._write_lock.release()
965484
def iter_sync(fileobj, max_read):
    

    read = 0
    size = 2
    last_byte = b""
    is_second = lambda b: ord(b) & 0xe0 == 0xe0

    while read < max_read:
        data_offset = fileobj.tell()
        new_data = fileobj.read(min(max_read - read, size))
        if not new_data:
            return
        read += len(new_data)

        if last_byte == b"\xff" and is_second(new_data[0:1]):
            fileobj.seek(data_offset - 1, 0)
            yield

        size *= 2
        last_byte = new_data[-1:]

        find_offset = 0
        while True:
            index = new_data.find(b"\xff", find_offset)
            # if not found or the last byte -> read more
            if index == -1 or index == len(new_data) - 1:
                break

            if is_second(new_data[index + 1:index + 2]):
                fileobj.seek(data_offset + index, 0)
                yield
            find_offset = index + 1

        fileobj.seek(data_offset + len(new_data), 0)
296992
def encode_streaming(self, data):
        
        # Buffer value and size
        buffer = 0
        size = 0
        for s in data:
            b, v = self._table[s]
            # Shift new bits in the buffer
            buffer = (buffer << b) + v
            size += b
            while size >= 8:
                byte = buffer >> (size - 8)
                yield to_byte(byte)
                buffer = buffer - (byte << (size - 8))
                size -= 8

        # Handling of the final sub-byte chunk.
        # The end of the encoded bit stream does not align necessarily with byte boundaries,
        # so we need an "end of file" indicator symbol (_EOF) to guard against decoding
        # the non-data trailing bits of the last byte.
        # As an optimization however, while encoding _EOF, it is only necessary to encode up to
        # the end of the current byte and cut off there.
        # No new byte has to be started for the remainder, saving us one (or more) output bytes.
        if size > 0:
            b, v = self._table[_EOF]
            buffer = (buffer << b) + v
            size += b
            if size >= 8:
                byte = buffer >> (size - 8)
            else:
                byte = buffer << (8 - size)
            yield to_byte(byte)
1254525
def pull(self):
        
        sock = self.__sock(zmq.PULL)
        return self.__recv_generator(sock)