X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/06f92553e100b58d2d72a3190fb8e57abac820bb..90fc79852a995fd8e665cf48ae20c49a9bbc78eb:/sdk/python/arvados/stream.py

diff --git a/sdk/python/arvados/stream.py b/sdk/python/arvados/stream.py
index dc90d8e57c..e13e1a6eff 100644
--- a/sdk/python/arvados/stream.py
+++ b/sdk/python/arvados/stream.py
@@ -1,7 +1,6 @@
 import gflags
 import httplib
 import httplib2
-import logging
 import os
 import pprint
 import sys
@@ -71,7 +70,7 @@ def locators_and_ranges(data_locators, range_start, range_size, debug=False):
         block_size = data_locators[i][BLOCKSIZE]
         block_start = data_locators[i][OFFSET]
         block_end = block_start + block_size
-    
+
     while i < len(data_locators):
         locator, block_size, block_start = data_locators[i]
         block_end = block_start + block_size
@@ -135,9 +134,12 @@ class StreamFileReader(object):
             return ''
 
         data = ''
-        for locator, blocksize, segmentoffset, segmentsize in locators_and_ranges(self.segments, self._filepos, size):
-            data += self._stream.readfrom(locator+segmentoffset, segmentsize)
-            self._filepos += len(data)
+        available_chunks = locators_and_ranges(self.segments, self._filepos, size)
+        if available_chunks:
+            locator, blocksize, segmentoffset, segmentsize = available_chunks[0]
+            data = self._stream.readfrom(locator+segmentoffset, segmentsize)
+
+        self._filepos += len(data)
         return data
 
     def readfrom(self, start, size):
@@ -169,7 +171,7 @@ class StreamFileReader(object):
             dc = bz2.BZ2Decompressor()
             return self.decompress(lambda segment: dc.decompress(segment), size)
         elif re.search('\.gz$', self._name):
-            dc = zlib.decompressobj(16+zlib.MAX_WBITS)            
+            dc = zlib.decompressobj(16+zlib.MAX_WBITS)
             return self.decompress(lambda segment: dc.decompress(dc.unconsumed_tail + segment), size)
         else:
             return self.readall(size)
@@ -205,11 +207,10 @@ class StreamReader(object):
         self._data_locators = []
         self._files = collections.OrderedDict()
 
-        if keep != None:
-            self._keep = keep
-        else:
-            self._keep = Keep.global_client_object()
-            
+        if keep is None:
+            keep = KeepClient()
+        self._keep = keep
+
         streamoffset = 0L
 
         # parse stream
@@ -264,11 +265,16 @@ class StreamReader(object):
         for locator, blocksize, segmentoffset, segmentsize in locators_and_ranges(self._data_locators, start, size):
             data += self._keep.get(locator)[segmentoffset:segmentoffset+segmentsize]
         return data
-    
-    def manifest_text(self):
+
+    def manifest_text(self, strip=False):
         manifest_text = [self.name().replace(' ', '\\040')]
-        manifest_text.extend([d[LOCATOR] for d in self._data_locators])
-        manifest_text.extend([' '.join(["{}:{}:{}".format(seg[LOCATOR], seg[BLOCKSIZE], f.name().replace(' ', '\\040')) 
+        if strip:
+            for d in self._data_locators:
+                m = re.match(r'^[0-9a-f]{32}\+\d+', d[LOCATOR])
+                manifest_text.append(m.group(0))
+        else:
+            manifest_text.extend([d[LOCATOR] for d in self._data_locators])
+        manifest_text.extend([' '.join(["{}:{}:{}".format(seg[LOCATOR], seg[BLOCKSIZE], f.name().replace(' ', '\\040'))
                                         for seg in f.segments])
                               for f in self._files.values()])
         return ' '.join(manifest_text) + '\n'