X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/8128b451b8e3161ae3db8b7e1858f6eef8105c9f..8a0eb69984a93852ec888cd3e02b778b0be758ed:/sdk/python/arvados/stream.py

diff --git a/sdk/python/arvados/stream.py b/sdk/python/arvados/stream.py
index 75b23cc026..afc202e180 100644
--- a/sdk/python/arvados/stream.py
+++ b/sdk/python/arvados/stream.py
@@ -6,65 +6,13 @@ import threading
 import functools
 import copy
 
-from .ranges import *
-from .arvfile import ArvadosFileBase, StreamFileReader
+from ._ranges import locators_and_ranges, Range
+from .arvfile import StreamFileReader
 from arvados.retry import retry_method
 from keep import *
 import config
 import errors
-
-def locator_block_size(loc):
-    s = re.match(r'[0-9a-f]{32}\+(\d+)(\+\S+)*', loc)
-    return long(s.group(1))
-
-def normalize_stream(s, stream):
-    '''
-    s is the stream name
-    stream is a dict mapping each filename to a list in the form [block locator, block size, segment offset (from beginning of block), segment size]
-    returns the stream as a list of tokens
-    '''
-    stream_tokens = [s]
-    sortedfiles = list(stream.keys())
-    sortedfiles.sort()
-
-    blocks = {}
-    streamoffset = 0L
-    # Go through each file and add each referenced block exactly once.
-    for f in sortedfiles:
-        for b in stream[f]:
-            if b.locator not in blocks:
-                stream_tokens.append(b.locator)
-                blocks[b.locator] = streamoffset
-                streamoffset += locator_block_size(b.locator)
-
-    # Add the empty block if the stream is otherwise empty.
-    if len(stream_tokens) == 1:
-        stream_tokens.append(config.EMPTY_BLOCK_LOCATOR)
-
-    for f in sortedfiles:
-        # Add in file segments
-        current_span = None
-        fout = f.replace(' ', '\\040')
-        for segment in stream[f]:
-            # Collapse adjacent segments
-            streamoffset = blocks[segment.locator] + segment.segment_offset
-            if current_span is None:
-                current_span = [streamoffset, streamoffset + segment.segment_size]
-            else:
-                if streamoffset == current_span[1]:
-                    current_span[1] += segment.segment_size
-                else:
-                    stream_tokens.append("{0}:{1}:{2}".format(current_span[0], current_span[1] - current_span[0], fout))
-                    current_span = [streamoffset, streamoffset + segment.segment_size]
-
-        if current_span is not None:
-            stream_tokens.append("{0}:{1}:{2}".format(current_span[0], current_span[1] - current_span[0], fout))
-
-        if not stream[f]:
-            stream_tokens.append("0:0:{0}".format(fout))
-
-    return stream_tokens
-
+from _normalize_stream import normalize_stream
 
 class StreamReader(object):
     def __init__(self, tokens, keep=None, debug=False, _empty=False,
@@ -87,7 +35,7 @@ class StreamReader(object):
             s = re.match(r'^[0-9a-f]{32}\+(\d+)(\+\S+)*$', tok)
             if s:
                 blocksize = long(s.group(1))
-                self._data_locators.append(Range(tok, streamoffset, blocksize))
+                self._data_locators.append(Range(tok, streamoffset, blocksize, 0))
                 streamoffset += blocksize
                 continue
 
@@ -97,7 +45,7 @@ class StreamReader(object):
                 size = long(s.group(2))
                 name = s.group(3).replace('\\040', ' ')
                 if name not in self._files:
-                    self._files[name] = StreamFileReader(self, [Range(pos, 0, size)], name)
+                    self._files[name] = StreamFileReader(self, [Range(pos, 0, size, 0)], name)
                 else:
                     filereader = self._files[name]
                     filereader.segments.append(Range(pos, filereader.size(), size))
@@ -114,13 +62,10 @@ class StreamReader(object):
     def all_files(self):
         return self._files.values()
 
-    def _size(self):
+    def size(self):
         n = self._data_locators[-1]
         return n.range_start + n.range_size
 
-    def size(self):
-        return self._size()
-
     def locators_and_ranges(self, range_start, range_size):
         return locators_and_ranges(self._data_locators, range_start, range_size)
 
@@ -130,10 +75,6 @@ class StreamReader(object):
 
     @retry_method
     def readfrom(self, start, size, num_retries=None):
-        return self._readfrom(start, size, num_retries=num_retries)
-
-    @retry_method
-    def _readfrom(self, start, size, num_retries=None):
         """Read up to 'size' bytes from the stream, starting at 'start'"""
         if size == 0:
             return ''