Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/smmap/buf.py: 16%

1"""Module with a simple buffer implementation using the memory manager""" 

2import sys 

3 

4__all__ = ["SlidingWindowMapBuffer"] 

5 

6 

7class SlidingWindowMapBuffer: 

8 

9 """A buffer like object which allows direct byte-wise object and slicing into 

10 memory of a mapped file. The mapping is controlled by the provided cursor. 

11 

12 The buffer is relative, that is if you map an offset, index 0 will map to the 

13 first byte at the offset you used during initialization or begin_access 

14 

15 **Note:** Although this type effectively hides the fact that there are mapped windows 

16 underneath, it can unfortunately not be used in any non-pure python method which 

17 needs a buffer or string""" 

18 __slots__ = ( 

19 '_c', # our cursor 

20 '_size', # our supposed size 

21 ) 

22 

23 def __init__(self, cursor=None, offset=0, size=sys.maxsize, flags=0): 

24 """Initalize the instance to operate on the given cursor. 

25 :param cursor: if not None, the associated cursor to the file you want to access 

26 If None, you have call begin_access before using the buffer and provide a cursor 

27 :param offset: absolute offset in bytes 

28 :param size: the total size of the mapping. Defaults to the maximum possible size 

29 From that point on, the __len__ of the buffer will be the given size or the file size. 

30 If the size is larger than the mappable area, you can only access the actually available 

31 area, although the length of the buffer is reported to be your given size. 

32 Hence it is in your own interest to provide a proper size ! 

33 :param flags: Additional flags to be passed to os.open 

34 :raise ValueError: if the buffer could not achieve a valid state""" 

35 self._c = cursor 

36 if cursor and not self.begin_access(cursor, offset, size, flags): 

37 raise ValueError("Failed to allocate the buffer - probably the given offset is out of bounds") 

38 # END handle offset 

39 

40 def __del__(self): 

41 self.end_access() 

42 

43 def __enter__(self): 

44 return self 

45 

46 def __exit__(self, exc_type, exc_value, traceback): 

47 self.end_access() 

48 

49 def __len__(self): 

50 return self._size 

51 

52 def __getitem__(self, i): 

53 if isinstance(i, slice): 

54 return self.__getslice__(i.start or 0, i.stop or self._size) 

55 c = self._c 

56 assert c.is_valid() 

57 if i < 0: 

58 i = self._size + i 

59 if not c.includes_ofs(i): 

60 c.use_region(i, 1) 

61 # END handle region usage 

62 return c.buffer()[i - c.ofs_begin()] 

63 

64 def __getslice__(self, i, j): 

65 c = self._c 

66 # fast path, slice fully included - safes a concatenate operation and 

67 # should be the default 

68 assert c.is_valid() 

69 if i < 0: 

70 i = self._size + i 

71 if j == sys.maxsize: 

72 j = self._size 

73 if j < 0: 

74 j = self._size + j 

75 if (c.ofs_begin() <= i) and (j < c.ofs_end()): 

76 b = c.ofs_begin() 

77 return c.buffer()[i - b:j - b] 

78 else: 

79 l = j - i # total length 

80 ofs = i 

81 # It's fastest to keep tokens and join later, especially in py3, which was 7 times slower 

82 # in the previous iteration of this code 

83 md = list() 

84 while l: 

85 c.use_region(ofs, l) 

86 assert c.is_valid() 

87 d = c.buffer()[:l] 

88 ofs += len(d) 

89 l -= len(d) 

90 # Make sure we don't keep references, as c.use_region() might attempt to free resources, but 

91 # can't unless we use pure bytes 

92 if hasattr(d, 'tobytes'): 

93 d = d.tobytes() 

94 md.append(d) 

95 # END while there are bytes to read 

96 return bytes().join(md) 

97 # END fast or slow path 

98 #{ Interface 

99 

100 def begin_access(self, cursor=None, offset=0, size=sys.maxsize, flags=0): 

101 """Call this before the first use of this instance. The method was already 

102 called by the constructor in case sufficient information was provided. 

103 

104 For more information no the parameters, see the __init__ method 

105 :param path: if cursor is None the existing one will be used. 

106 :return: True if the buffer can be used""" 

107 if cursor: 

108 self._c = cursor 

109 # END update our cursor 

110 

111 # reuse existing cursors if possible 

112 if self._c is not None and self._c.is_associated(): 

113 res = self._c.use_region(offset, size, flags).is_valid() 

114 if res: 

115 # if given size is too large or default, we computer a proper size 

116 # If its smaller, we assume the combination between offset and size 

117 # as chosen by the user is correct and use it ! 

118 # If not, the user is in trouble. 

119 if size > self._c.file_size(): 

120 size = self._c.file_size() - offset 

121 # END handle size 

122 self._size = size 

123 # END set size 

124 return res 

125 # END use our cursor 

126 return False 

127 

128 def end_access(self): 

129 """Call this method once you are done using the instance. It is automatically 

130 called on destruction, and should be called just in time to allow system 

131 resources to be freed. 

132 

133 Once you called end_access, you must call begin access before reusing this instance!""" 

134 self._size = 0 

135 if self._c is not None: 

136 self._c.unuse_region() 

137 # END unuse region 

138 

139 def cursor(self): 

140 """:return: the currently set cursor which provides access to the data""" 

141 return self._c 

142 

143 #}END interface