Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/joblib/backports.py: 27%

1""" 

2Backports of fixes for joblib dependencies 

3""" 

4 

5import os 

6import re 

7import time 

8from multiprocessing import util 

9from os.path import basename 

10 

11 

12class Version: 

13 """Backport from deprecated distutils 

14 

15 We maintain this backport to avoid introducing a new dependency on 

16 `packaging`. 

17 

18 We might rexplore this choice in the future if all major Python projects 

19 introduce a dependency on packaging anyway. 

20 """ 

21 

22 def __init__(self, vstring=None): 

23 if vstring: 

24 self.parse(vstring) 

25 

26 def __repr__(self): 

27 return "%s ('%s')" % (self.__class__.__name__, str(self)) 

28 

29 def __eq__(self, other): 

30 c = self._cmp(other) 

31 if c is NotImplemented: 

32 return c 

33 return c == 0 

34 

35 def __lt__(self, other): 

36 c = self._cmp(other) 

37 if c is NotImplemented: 

38 return c 

39 return c < 0 

40 

41 def __le__(self, other): 

42 c = self._cmp(other) 

43 if c is NotImplemented: 

44 return c 

45 return c <= 0 

46 

47 def __gt__(self, other): 

48 c = self._cmp(other) 

49 if c is NotImplemented: 

50 return c 

51 return c > 0 

52 

53 def __ge__(self, other): 

54 c = self._cmp(other) 

55 if c is NotImplemented: 

56 return c 

57 return c >= 0 

58 

59 

60class LooseVersion(Version): 

61 """Backport from deprecated distutils 

62 

63 We maintain this backport to avoid introducing a new dependency on 

64 `packaging`. 

65 

66 We might rexplore this choice in the future if all major Python projects 

67 introduce a dependency on packaging anyway. 

68 """ 

69 

70 component_re = re.compile(r"(\d+ | [a-z]+ | \.)", re.VERBOSE) 

71 

72 def __init__(self, vstring=None): 

73 if vstring: 

74 self.parse(vstring) 

75 

76 def parse(self, vstring): 

77 # I've given up on thinking I can reconstruct the version string 

78 # from the parsed tuple -- so I just store the string here for 

79 # use by __str__ 

80 self.vstring = vstring 

81 components = [x for x in self.component_re.split(vstring) if x and x != "."] 

82 for i, obj in enumerate(components): 

83 try: 

84 components[i] = int(obj) 

85 except ValueError: 

86 pass 

87 

88 self.version = components 

89 

90 def __str__(self): 

91 return self.vstring 

92 

93 def __repr__(self): 

94 return "LooseVersion ('%s')" % str(self) 

95 

96 def _cmp(self, other): 

97 if isinstance(other, str): 

98 other = LooseVersion(other) 

99 elif not isinstance(other, LooseVersion): 

100 return NotImplemented 

101 

102 if self.version == other.version: 

103 return 0 

104 if self.version < other.version: 

105 return -1 

106 if self.version > other.version: 

107 return 1 

108 

109 

110try: 

111 import numpy as np 

112 

113 def make_memmap( 

114 filename, 

115 dtype="uint8", 

116 mode="r+", 

117 offset=0, 

118 shape=None, 

119 order="C", 

120 unlink_on_gc_collect=False, 

121 ): 

122 """Custom memmap constructor compatible with numpy.memmap. 

123 

124 This function: 

125 - is a backport the numpy memmap offset fix (See 

126 https://github.com/numpy/numpy/pull/8443 for more details. 

127 The numpy fix is available starting numpy 1.13) 

128 - adds ``unlink_on_gc_collect``, which specifies explicitly whether 

129 the process re-constructing the memmap owns a reference to the 

130 underlying file. If set to True, it adds a finalizer to the 

131 newly-created memmap that sends a maybe_unlink request for the 

132 memmaped file to resource_tracker. 

133 """ 

134 util.debug( 

135 "[MEMMAP READ] creating a memmap (shape {}, filename {}, pid {})".format( 

136 shape, basename(filename), os.getpid() 

137 ) 

138 ) 

139 

140 mm = np.memmap( 

141 filename, dtype=dtype, mode=mode, offset=offset, shape=shape, order=order 

142 ) 

143 if LooseVersion(np.__version__) < "1.13": 

144 mm.offset = offset 

145 if unlink_on_gc_collect: 

146 from ._memmapping_reducer import add_maybe_unlink_finalizer 

147 

148 add_maybe_unlink_finalizer(mm) 

149 return mm 

150except ImportError: 

151 

152 def make_memmap( 

153 filename, 

154 dtype="uint8", 

155 mode="r+", 

156 offset=0, 

157 shape=None, 

158 order="C", 

159 unlink_on_gc_collect=False, 

160 ): 

161 raise NotImplementedError( 

162 "'joblib.backports.make_memmap' should not be used " 

163 "if numpy is not installed." 

164 ) 

165 

166 

167if os.name == "nt": 

168 # https://github.com/joblib/joblib/issues/540 

169 access_denied_errors = (5, 13) 

170 from os import replace 

171 

172 def concurrency_safe_rename(src, dst): 

173 """Renames ``src`` into ``dst`` overwriting ``dst`` if it exists. 

174 

175 On Windows os.replace can yield permission errors if executed by two 

176 different processes. 

177 """ 

178 max_sleep_time = 1 

179 total_sleep_time = 0 

180 sleep_time = 0.001 

181 while total_sleep_time < max_sleep_time: 

182 try: 

183 replace(src, dst) 

184 break 

185 except Exception as exc: 

186 if getattr(exc, "winerror", None) in access_denied_errors: 

187 time.sleep(sleep_time) 

188 total_sleep_time += sleep_time 

189 sleep_time *= 2 

190 else: 

191 raise 

192 else: 

193 raise 

194else: 

195 from os import replace as concurrency_safe_rename # noqa