1 from __future__ import print_function
2 from __future__ import absolute_import
3 from future.utils import listvalues
4 from builtins import object
13 from ._ranges import locators_and_ranges, Range
14 from .arvfile import StreamFileReader
15 from arvados.retry import retry_method
16 from arvados.keep import *
19 from ._normalize_stream import normalize_stream
21 class StreamReader(object):
22 def __init__(self, tokens, keep=None, debug=False, _empty=False,
24 self._stream_name = None
25 self._data_locators = []
26 self._files = collections.OrderedDict()
28 self.num_retries = num_retries
34 if debug: print('tok', tok)
35 if self._stream_name is None:
36 self._stream_name = tok.replace('\\040', ' ')
39 s = re.match(r'^[0-9a-f]{32}\+(\d+)(\+\S+)*$', tok)
41 blocksize = int(s.group(1))
42 self._data_locators.append(Range(tok, streamoffset, blocksize, 0))
43 streamoffset += blocksize
46 s = re.search(r'^(\d+):(\d+):(\S+)', tok)
49 size = int(s.group(2))
50 name = s.group(3).replace('\\040', ' ')
51 if name not in self._files:
52 self._files[name] = StreamFileReader(self, [Range(pos, 0, size, 0)], name)
54 filereader = self._files[name]
55 filereader.segments.append(Range(pos, filereader.size(), size))
58 raise errors.SyntaxError("Invalid manifest format")
61 return self._stream_name
67 return listvalues(self._files)
70 n = self._data_locators[-1]
71 return n.range_start + n.range_size
73 def locators_and_ranges(self, range_start, range_size):
74 return locators_and_ranges(self._data_locators, range_start, range_size)
77 def _keepget(self, locator, num_retries=None):
78 return self._keep.get(locator, num_retries=num_retries)
81 def readfrom(self, start, size, num_retries=None):
82 """Read up to 'size' bytes from the stream, starting at 'start'"""
85 if self._keep is None:
86 self._keep = KeepClient(num_retries=self.num_retries)
88 for lr in locators_and_ranges(self._data_locators, start, size):
89 data.append(self._keepget(lr.locator, num_retries=num_retries)[lr.segment_offset:lr.segment_offset+lr.segment_size])
92 def manifest_text(self, strip=False):
93 manifest_text = [self.name().replace(' ', '\\040')]
95 for d in self._data_locators:
96 m = re.match(r'^[0-9a-f]{32}\+\d+', d.locator)
97 manifest_text.append(m.group(0))
99 manifest_text.extend([d.locator for d in self._data_locators])
100 manifest_text.extend([' '.join(["{}:{}:{}".format(seg.locator, seg.range_size, f.name.replace(' ', '\\040'))
101 for seg in f.segments])
102 for f in listvalues(self._files)])
103 return ' '.join(manifest_text) + '\n'