Return stream of (iden, provenance stack) tuples at the given offset.
169205
async def provStacks(self, offs, size):
        '''
        
        '''
        count = 0
        for iden, stack in self.cell.provstor.provStacks(offs, size):
            count += 1
            if not count % 1000:
                await asyncio.sleep(0)
            yield s_common.ehex(iden), stack
386359
def get(self, stream, fmt='txt', **kwargs):
        """"""
        sel = ''.join(["&{0}={1}".format(k, v) for (k, v) in kwargs.items()])
        url = "streamds/{0}.{1}?{2}".format(stream, fmt, sel[1:])
        data = self._db._get_content(url)
        if not data:
            log.error("No data found at URL '%s'." % url)
            return
        if (data.startswith("ERROR")):
            log.error(data)
            return
        if fmt == "txt":
            return read_csv(data)
        return data
814965
public InputStream getInputStreamFromOffset(int offset) throws IOException {
        loadFileData();
        return new ByteArrayInputStream(data, offset, data.length - offset);
    }
140897
def _consume(self):
        
        self._stream_offset += self._buff_i - self._buf_checkpoint
        self._buf_checkpoint = self._buff_i
390965
function getTTYStream(stream: UnderlyingStream): ?tty.WriteStream {
  if (
    stream instanceof tty.WriteStream &&
    stream.isTTY &&
    stream.columns >= 1
  ) {
    return stream;
  }
  return null;
}
573901
def _dump(self, stream, pattern=None, object_ids=False):
        """"""
        return self._lists[pattern]._dump(stream, '', object_ids=object_ids)
1242156
def from_stream(cls, stream, marker_code, offset):
        
        # field                 off  type   notes
        # ------------------  ---  -----  ----------------------------
        # segment length       0   short
        # Data precision       2   byte
        # Vertical lines       3   short  px_height
        # Horizontal lines     5   short  px_width
        # ------------------  ---  -----  ----------------------------
        segment_length = stream.read_short(offset)
        px_height = stream.read_short(offset, 3)
        px_width = stream.read_short(offset, 5)
        return cls(marker_code, offset, segment_length, px_width, px_height)
102975
def read_tuple(self):
        """"""
        cmd = self.read_command()
        source = cmd["comp"]
        stream = cmd["stream"]
        values = cmd["tuple"]
        val_type = self._source_tuple_types[source].get(stream)
        return Tuple(
            cmd["id"],
            source,
            stream,
            cmd["task"],
            tuple(values) if val_type is None else val_type(*values),
        )
89292
def as_bin(self, as_spendable=False):
        """"""
        f = io.BytesIO()
        self.stream(f, as_spendable=as_spendable)
        return f.getvalue()
210384
def provStacks(self, offs, size):
        '''
        
        '''
        for _, iden in self.provseq.slice(offs, size):
            stack = self.getProvStack(iden)
            if stack is None:
                continue
            yield (iden, stack)
901490
def from_offset(cls, chunk_type, stream_rdr, offset):
        
        px_width = stream_rdr.read_long(offset)
        px_height = stream_rdr.read_long(offset, 4)
        return cls(chunk_type, px_width, px_height)
234022
def convertToFree(stream, length_limit=True):
    """"""
    linestack = []
        
    for line in stream:
        convline = FortranLine(line, length_limit)
        
        if convline.is_regular:
            if convline.isContinuation and linestack: 
                linestack[0].continueLine()
            for l in linestack:
                yield str(l)
            linestack = []
            
        linestack.append(convline)
        
    for l in linestack:
        yield str(l)
653220
def parse_stream_raw(self, stream, debug=False):
        """"""
        tokens = tokenize.generate_tokens(stream.readline)
        return self.parse_tokens(tokens, debug)
403563
public List<Integer> get(String stream, List<Object> tuple, Collection<Tuple> anchors, Object rootId) {
        List<Integer> outTasks = new ArrayList<>();

        // get grouper, then get which task should tuple be sent to.
        Map<String, MkGrouper> componentCrouping = streamComponentGrouper.get(stream);
        if (componentCrouping == null) {
            // if the target component's parallelism is 0, don't need send to
            // them
            LOG.debug("Failed to get Grouper of " + stream + " when " + debugIdStr);
            return outTasks;
        }

        for (Entry<String, MkGrouper> ee : componentCrouping.entrySet()) {
            String targetComponent = ee.getKey();
            MkGrouper g = ee.getValue();

            if (GrouperType.direct.equals(g.gettype())) {
                throw new IllegalArgumentException("Cannot do regular emit to direct stream");
            }

            outTasks.addAll(g.grouper(tuple));
        }

        if (isDebug(anchors, rootId)) {
            LOG.info(debugIdStr + stream + " to " + outTasks + ":" + tuple.toString());
        }
        int num_out_tasks = outTasks.size();
        taskStats.send_tuple(stream, num_out_tasks);

        return outTasks;
    }
221684
private CompletableFuture<List<EpochRecord>> getEpochRecords(String scope, String stream, int epoch, OperationContext context) {
        List<CompletableFuture<EpochRecord>> list = new ArrayList<>();
        list.add(streamMetadataStore.getEpoch(scope, stream, epoch, context, executor));
        list.add(streamMetadataStore.getActiveEpoch(scope, stream, context, true, executor));
        return Futures.allOfWithResults(list);
    }
80326
def parse_set(self):
        """"""
        lineno = next(self.stream).lineno
        target = self.parse_assign_target()
        self.stream.expect('assign')
        expr = self.parse_tuple()
        return nodes.Assign(target, expr, lineno=lineno)
1373942
def stream_info(self):
        
        accum = {}
        for key in self.keys:
            accum[key] = self.database.xinfo_stream(key)
        return accum