openrvdas.logger.readers.text_file_reader
No module-level documentation available.
1#!/usr/bin/env python3 2 3import glob 4import logging 5import sys 6import time 7 8from logger.readers.reader import StorageReader # noqa: E402 9 10 11################################################################################ 12# Open and read single-line records from one or more text files. 13class TextFileReader(StorageReader): 14 """Read lines from one or more text files. Sequentially open all 15 files that match the file_spec. 16 """ 17 ############################ 18 19 def __init__(self, file_spec=None, tail=False, refresh_file_spec=False, 20 retry_interval=0.1, interval=0, eol=None, *args, **kwargs): 21 """ 22 ``` 23 file_spec Possibly wildcarded string speficying files to be opened. 24 Special case: if file_spec is None, read from stdin. 25 26 tail If False, return None upon reaching end of last file; if 27 True, block upon reaching EOF of last file and wait for 28 more records. 29 30 refresh_file_spec 31 If True, refresh the search for matching filenames when 32 reaching last EOF to see if any new matching files have 33 appeared in the interim. 34 35 retry_interval 36 If tail and/or refresh_file_spec are True, how long to 37 wait before looking to see if any new records or files 38 have shown up. 39 40 interval 41 How long to sleep between returning records. In general 42 this should be zero except for debugging purposes. 43 44 eol Optional character by which to recognize the end of a record 45 ``` 46 Note that the order in which files are opened will probably be in 47 alphanumeric by filename, but this is not strictly enforced and 48 depends on how glob returns them. 49 """ 50 51 super().__init__(*args, **kwargs) 52 53 self.file_spec = file_spec 54 self.tail = tail 55 self.refresh_file_spec = refresh_file_spec 56 self.retry_interval = retry_interval 57 self.interval = interval 58 self.eol = eol 59 60 # If interval != 0, we need to keep track of our last_read to know 61 # how long to sleep 62 self.last_read = 0 63 64 # The file we're currently using 65 self.current_file = None 66 67 self.pos = 0 68 self.start_pos = {} 69 self.end_pos = {} 70 71 # Special case if file_spec is None 72 if file_spec is None: 73 self.current_file = sys.stdin 74 self.used_file_list = [] 75 self.unused_file_list = [] 76 self.tail = True 77 return 78 79 # Which files will we use, which haven't we used yet? 80 self.unused_file_list = sorted(glob.glob(file_spec)) 81 if not self.unused_file_list: 82 logging.warning('TextFileReader: file_spec "%s" matches no files', 83 file_spec) 84 self.used_file_list = [] 85 86 ############################ 87 88 def _get_next_file(self): 89 """Internal - Open and assign the next unused file to 90 self.current_file if we can find one. Return None (and don't mess 91 with current_file) if we can't find a next one. 92 """ 93 # If no more unused files, but refresh_file_spec is specified, see 94 # if more files have shown up 95 if not self.unused_file_list and self.refresh_file_spec: 96 matching_files = sorted(glob.glob(self.file_spec)) 97 self.unused_file_list = [f for f in matching_files 98 if f not in self.used_file_list] 99 logging.info('TextFileReader found %d new files matching spec "%s": %s', 100 len(self.unused_file_list), self.file_spec, 101 self.unused_file_list) 102 103 # Are there any more files? If so, get the next one and open it 104 if self.unused_file_list: 105 # First, save the record count for the file we're about to close. 106 if self.used_file_list: 107 prev_filename = self.used_file_list[-1] 108 self.end_pos[prev_filename] = self.pos 109 110 next_filename = self.unused_file_list.pop(0) 111 logging.info('TextFileReader opening next file "%s"', next_filename) 112 self.start_pos[next_filename] = self.pos 113 self.current_file = open(next_filename, 'r') 114 self.used_file_list.append(next_filename) 115 return self.current_file 116 117 # If here, we've found no unused next file. Give up 118 return None 119 120 ############################ 121 def read(self): 122 """Get the next line of text. Return None if there are no more 123 records. To test EOF you'll need to test 124 125 if record is None: 126 no more records... 127 128 rather than simply 129 130 if not record: 131 could be EOF or simply an empty next line 132 """ 133 if self.interval: 134 now = time.time() 135 sleep_time = max(0, self.interval - (now - self.last_read)) 136 logging.debug('Sleeping %f seconds', sleep_time) 137 if sleep_time: 138 time.sleep(sleep_time) 139 140 record = None 141 while not record: 142 # If we've got a current file, or if _get_next_file() gets one 143 # for us, try to read a record. 144 if self.current_file or self._get_next_file(): 145 if not self.eol: 146 record = self.current_file.readline() 147 else: 148 record = self._read_until_eol() 149 if record: 150 self.last_read = time.time() 151 record = record.rstrip('\n') 152 logging.debug('TextFileReader got record "%s"', record) 153 self.pos += 1 154 return record 155 156 # No record: our current_file has reached EOF. See if more 157 # files we should try to read. 158 if self._get_next_file(): 159 # Found a new file to read - loop again right away 160 continue 161 162 # EOF when we're reading from stdin means we're done 163 if not self.file_spec: 164 return None 165 166 # No record, no new files, no tail or refresh directive - 167 # there's nothing left for us to try. Go home empty-handed. 168 if not self.refresh_file_spec and not self.tail: 169 return None 170 171 # User wants refresh or tail, so sleep and try again. 172 logging.debug('TextFileReader - tail/refresh specified, so sleeping ' 173 '%f seconds before trying again', self.retry_interval) 174 time.sleep(self.retry_interval) 175 176 ############################ 177 # If self.eol is a string instead of None, read until we've consumed that 178 # string or reached eof, and return that as a record. 179 def _read_until_eol(self): 180 if not self.eol: 181 logging.fatal('Code error: called _read_until_eof, but no eof string specified') 182 return 183 184 record = '' 185 eol_index = 0 # we're going to count our way through eol characters 186 while eol_index < len(self.eol): 187 # read by character 188 char = self.current_file.read(1) 189 if char == '': 190 break 191 elif char == self.eol[eol_index]: 192 eol_index += 1 193 record += char 194 else: 195 eol_index = 0 196 record += char 197 198 # If we're here because we did in fact get a full eol string, 199 # retroactively snip it from our record. 200 if eol_index == len(self.eol): 201 record = record[:-eol_index] 202 203 return record 204 205 ############################ 206 # Current behavior is to just go to the end if we run out of records, 207 # as io.IOBase.seek() does. 208 # QUESTION: To really behave like seek(), we'd have to keep track of self.pos 209 # beyond the end of the file, e.g. seek(100, 'start') would always return 210 # 100, even if there are < 100 records. Is this what we want? 211 def _seek_forward_from_current(self, offset=0): 212 if offset == 0: 213 return 214 if offset < 0: 215 return self._seek_back_from_current(offset) 216 i = 0 217 while i < offset: 218 if self.current_file or self._get_next_file(): 219 if self.current_file.readline(): 220 i += 1 221 self.pos += 1 222 else: 223 if self._get_next_file() is None: 224 break 225 # TODO: take advantage of self.start_pos and self.end_pos if we've 226 # already processed later files. 227 228 ############################ 229 def _seek_back_from_current(self, offset=0): 230 if offset == 0: 231 return 232 if offset > 0: 233 return self._seek_forward_from_current(offset) 234 target = self.pos + offset 235 if target < 0: 236 raise ValueError("Can't back up past earliest record") 237 238 # Find the right file. 239 current_filename = self.used_file_list[-1] 240 while target < self.start_pos[current_filename]: 241 self.unused_file_list.insert(0, current_filename) 242 self.used_file_list.pop() 243 current_filename = self.used_file_list[-1] 244 245 self.current_file = open(current_filename, 'r') 246 247 # TODO: implement backwards search within the file 248 for _ in range(target - self.start_pos[current_filename]): 249 self.current_file.readline() 250 self.pos = target 251 252 ############################ 253 def _save_state(self): 254 state = { 255 'used_file_list': self.used_file_list[:], 256 'unused_file_list': self.unused_file_list[:], 257 'pos': self.pos 258 } 259 if self.current_file: 260 state['current_filename'] = self.used_file_list[-1] 261 state['current_file_pos'] = self.current_file.tell() 262 return state 263 264 ############################ 265 def _restore_state(self, state): 266 self.used_file_list = state['used_file_list'] 267 self.unused_file_list = state['unused_file_list'] 268 if 'current_filename' in state: 269 self.current_file = open(state['current_filename'], 'r') 270 self.current_file.seek(state['current_file_pos']) 271 else: 272 self.current_file = None 273 self.pos = state['pos'] 274 275 ############################ 276 # Behavior is intended to mimic file seek() behavior but with 277 # respect to records: 'offset' means number of records, and origin 278 # is either 'start', 'current' or 'end'. 279 def seek(self, offset=0, origin='current'): 280 original_state = self._save_state() 281 282 try: 283 if origin == 'start': 284 if offset < 0: 285 raise ValueError("Can't back up past earliest record") 286 self.used_file_list = [] 287 self.unused_file_list = sorted(glob.glob(self.file_spec)) 288 self.current_file = None 289 self.pos = 0 290 self._seek_forward_from_current(offset) 291 292 elif origin == 'current': 293 if offset >= 0: 294 self._seek_forward_from_current(offset) 295 else: 296 self._seek_back_from_current(offset) 297 298 elif origin == 'end': 299 # Have to count lines in all files that haven't been processed yet. 300 # TODO: take self.refresh_file_spec into account 301 file_list = sorted(glob.glob(self.file_spec)) 302 pos = 0 303 for filename in file_list: 304 if filename in self.end_pos: 305 pos = self.end_pos[filename] 306 else: 307 self.start_pos[filename] = pos 308 309 # TODO: this can be made faster, if needed 310 with open(filename) as f: 311 for n, _ in enumerate(f, 1): 312 pass 313 314 pos += n 315 self.end_pos[filename] = pos 316 317 self.used_file_list = file_list 318 self.unused_file_list = [] 319 self.current_file = None 320 self.pos = pos 321 self._seek_back_from_current(offset) 322 323 else: 324 raise ValueError('Unknown origin value: "%s"' % origin) 325 326 except: # noqa: E722 327 self._restore_state(original_state) 328 raise 329 330 return self.pos 331 332 ############################ 333 def read_range(self, start=None, stop=None): 334 """ 335 Read a range of records beginning with record number start, and ending 336 *before* record number stop. 337 """ 338 if start is None: 339 start = 0 340 if stop is None: 341 stop = sys.maxsize 342 self.seek(start, 'start') 343 records = [] 344 for _ in range(stop - start): 345 record = self.read() 346 if record is None: 347 break 348 records.append(record) 349 return records
class
TextFileReader(logger.readers.reader.StorageReader):
14class TextFileReader(StorageReader): 15 """Read lines from one or more text files. Sequentially open all 16 files that match the file_spec. 17 """ 18 ############################ 19 20 def __init__(self, file_spec=None, tail=False, refresh_file_spec=False, 21 retry_interval=0.1, interval=0, eol=None, *args, **kwargs): 22 """ 23 ``` 24 file_spec Possibly wildcarded string speficying files to be opened. 25 Special case: if file_spec is None, read from stdin. 26 27 tail If False, return None upon reaching end of last file; if 28 True, block upon reaching EOF of last file and wait for 29 more records. 30 31 refresh_file_spec 32 If True, refresh the search for matching filenames when 33 reaching last EOF to see if any new matching files have 34 appeared in the interim. 35 36 retry_interval 37 If tail and/or refresh_file_spec are True, how long to 38 wait before looking to see if any new records or files 39 have shown up. 40 41 interval 42 How long to sleep between returning records. In general 43 this should be zero except for debugging purposes. 44 45 eol Optional character by which to recognize the end of a record 46 ``` 47 Note that the order in which files are opened will probably be in 48 alphanumeric by filename, but this is not strictly enforced and 49 depends on how glob returns them. 50 """ 51 52 super().__init__(*args, **kwargs) 53 54 self.file_spec = file_spec 55 self.tail = tail 56 self.refresh_file_spec = refresh_file_spec 57 self.retry_interval = retry_interval 58 self.interval = interval 59 self.eol = eol 60 61 # If interval != 0, we need to keep track of our last_read to know 62 # how long to sleep 63 self.last_read = 0 64 65 # The file we're currently using 66 self.current_file = None 67 68 self.pos = 0 69 self.start_pos = {} 70 self.end_pos = {} 71 72 # Special case if file_spec is None 73 if file_spec is None: 74 self.current_file = sys.stdin 75 self.used_file_list = [] 76 self.unused_file_list = [] 77 self.tail = True 78 return 79 80 # Which files will we use, which haven't we used yet? 81 self.unused_file_list = sorted(glob.glob(file_spec)) 82 if not self.unused_file_list: 83 logging.warning('TextFileReader: file_spec "%s" matches no files', 84 file_spec) 85 self.used_file_list = [] 86 87 ############################ 88 89 def _get_next_file(self): 90 """Internal - Open and assign the next unused file to 91 self.current_file if we can find one. Return None (and don't mess 92 with current_file) if we can't find a next one. 93 """ 94 # If no more unused files, but refresh_file_spec is specified, see 95 # if more files have shown up 96 if not self.unused_file_list and self.refresh_file_spec: 97 matching_files = sorted(glob.glob(self.file_spec)) 98 self.unused_file_list = [f for f in matching_files 99 if f not in self.used_file_list] 100 logging.info('TextFileReader found %d new files matching spec "%s": %s', 101 len(self.unused_file_list), self.file_spec, 102 self.unused_file_list) 103 104 # Are there any more files? If so, get the next one and open it 105 if self.unused_file_list: 106 # First, save the record count for the file we're about to close. 107 if self.used_file_list: 108 prev_filename = self.used_file_list[-1] 109 self.end_pos[prev_filename] = self.pos 110 111 next_filename = self.unused_file_list.pop(0) 112 logging.info('TextFileReader opening next file "%s"', next_filename) 113 self.start_pos[next_filename] = self.pos 114 self.current_file = open(next_filename, 'r') 115 self.used_file_list.append(next_filename) 116 return self.current_file 117 118 # If here, we've found no unused next file. Give up 119 return None 120 121 ############################ 122 def read(self): 123 """Get the next line of text. Return None if there are no more 124 records. To test EOF you'll need to test 125 126 if record is None: 127 no more records... 128 129 rather than simply 130 131 if not record: 132 could be EOF or simply an empty next line 133 """ 134 if self.interval: 135 now = time.time() 136 sleep_time = max(0, self.interval - (now - self.last_read)) 137 logging.debug('Sleeping %f seconds', sleep_time) 138 if sleep_time: 139 time.sleep(sleep_time) 140 141 record = None 142 while not record: 143 # If we've got a current file, or if _get_next_file() gets one 144 # for us, try to read a record. 145 if self.current_file or self._get_next_file(): 146 if not self.eol: 147 record = self.current_file.readline() 148 else: 149 record = self._read_until_eol() 150 if record: 151 self.last_read = time.time() 152 record = record.rstrip('\n') 153 logging.debug('TextFileReader got record "%s"', record) 154 self.pos += 1 155 return record 156 157 # No record: our current_file has reached EOF. See if more 158 # files we should try to read. 159 if self._get_next_file(): 160 # Found a new file to read - loop again right away 161 continue 162 163 # EOF when we're reading from stdin means we're done 164 if not self.file_spec: 165 return None 166 167 # No record, no new files, no tail or refresh directive - 168 # there's nothing left for us to try. Go home empty-handed. 169 if not self.refresh_file_spec and not self.tail: 170 return None 171 172 # User wants refresh or tail, so sleep and try again. 173 logging.debug('TextFileReader - tail/refresh specified, so sleeping ' 174 '%f seconds before trying again', self.retry_interval) 175 time.sleep(self.retry_interval) 176 177 ############################ 178 # If self.eol is a string instead of None, read until we've consumed that 179 # string or reached eof, and return that as a record. 180 def _read_until_eol(self): 181 if not self.eol: 182 logging.fatal('Code error: called _read_until_eof, but no eof string specified') 183 return 184 185 record = '' 186 eol_index = 0 # we're going to count our way through eol characters 187 while eol_index < len(self.eol): 188 # read by character 189 char = self.current_file.read(1) 190 if char == '': 191 break 192 elif char == self.eol[eol_index]: 193 eol_index += 1 194 record += char 195 else: 196 eol_index = 0 197 record += char 198 199 # If we're here because we did in fact get a full eol string, 200 # retroactively snip it from our record. 201 if eol_index == len(self.eol): 202 record = record[:-eol_index] 203 204 return record 205 206 ############################ 207 # Current behavior is to just go to the end if we run out of records, 208 # as io.IOBase.seek() does. 209 # QUESTION: To really behave like seek(), we'd have to keep track of self.pos 210 # beyond the end of the file, e.g. seek(100, 'start') would always return 211 # 100, even if there are < 100 records. Is this what we want? 212 def _seek_forward_from_current(self, offset=0): 213 if offset == 0: 214 return 215 if offset < 0: 216 return self._seek_back_from_current(offset) 217 i = 0 218 while i < offset: 219 if self.current_file or self._get_next_file(): 220 if self.current_file.readline(): 221 i += 1 222 self.pos += 1 223 else: 224 if self._get_next_file() is None: 225 break 226 # TODO: take advantage of self.start_pos and self.end_pos if we've 227 # already processed later files. 228 229 ############################ 230 def _seek_back_from_current(self, offset=0): 231 if offset == 0: 232 return 233 if offset > 0: 234 return self._seek_forward_from_current(offset) 235 target = self.pos + offset 236 if target < 0: 237 raise ValueError("Can't back up past earliest record") 238 239 # Find the right file. 240 current_filename = self.used_file_list[-1] 241 while target < self.start_pos[current_filename]: 242 self.unused_file_list.insert(0, current_filename) 243 self.used_file_list.pop() 244 current_filename = self.used_file_list[-1] 245 246 self.current_file = open(current_filename, 'r') 247 248 # TODO: implement backwards search within the file 249 for _ in range(target - self.start_pos[current_filename]): 250 self.current_file.readline() 251 self.pos = target 252 253 ############################ 254 def _save_state(self): 255 state = { 256 'used_file_list': self.used_file_list[:], 257 'unused_file_list': self.unused_file_list[:], 258 'pos': self.pos 259 } 260 if self.current_file: 261 state['current_filename'] = self.used_file_list[-1] 262 state['current_file_pos'] = self.current_file.tell() 263 return state 264 265 ############################ 266 def _restore_state(self, state): 267 self.used_file_list = state['used_file_list'] 268 self.unused_file_list = state['unused_file_list'] 269 if 'current_filename' in state: 270 self.current_file = open(state['current_filename'], 'r') 271 self.current_file.seek(state['current_file_pos']) 272 else: 273 self.current_file = None 274 self.pos = state['pos'] 275 276 ############################ 277 # Behavior is intended to mimic file seek() behavior but with 278 # respect to records: 'offset' means number of records, and origin 279 # is either 'start', 'current' or 'end'. 280 def seek(self, offset=0, origin='current'): 281 original_state = self._save_state() 282 283 try: 284 if origin == 'start': 285 if offset < 0: 286 raise ValueError("Can't back up past earliest record") 287 self.used_file_list = [] 288 self.unused_file_list = sorted(glob.glob(self.file_spec)) 289 self.current_file = None 290 self.pos = 0 291 self._seek_forward_from_current(offset) 292 293 elif origin == 'current': 294 if offset >= 0: 295 self._seek_forward_from_current(offset) 296 else: 297 self._seek_back_from_current(offset) 298 299 elif origin == 'end': 300 # Have to count lines in all files that haven't been processed yet. 301 # TODO: take self.refresh_file_spec into account 302 file_list = sorted(glob.glob(self.file_spec)) 303 pos = 0 304 for filename in file_list: 305 if filename in self.end_pos: 306 pos = self.end_pos[filename] 307 else: 308 self.start_pos[filename] = pos 309 310 # TODO: this can be made faster, if needed 311 with open(filename) as f: 312 for n, _ in enumerate(f, 1): 313 pass 314 315 pos += n 316 self.end_pos[filename] = pos 317 318 self.used_file_list = file_list 319 self.unused_file_list = [] 320 self.current_file = None 321 self.pos = pos 322 self._seek_back_from_current(offset) 323 324 else: 325 raise ValueError('Unknown origin value: "%s"' % origin) 326 327 except: # noqa: E722 328 self._restore_state(original_state) 329 raise 330 331 return self.pos 332 333 ############################ 334 def read_range(self, start=None, stop=None): 335 """ 336 Read a range of records beginning with record number start, and ending 337 *before* record number stop. 338 """ 339 if start is None: 340 start = 0 341 if stop is None: 342 stop = sys.maxsize 343 self.seek(start, 'start') 344 records = [] 345 for _ in range(stop - start): 346 record = self.read() 347 if record is None: 348 break 349 records.append(record) 350 return records
Read lines from one or more text files. Sequentially open all files that match the file_spec.
TextFileReader( file_spec=None, tail=False, refresh_file_spec=False, retry_interval=0.1, interval=0, eol=None, *args, **kwargs)
20 def __init__(self, file_spec=None, tail=False, refresh_file_spec=False, 21 retry_interval=0.1, interval=0, eol=None, *args, **kwargs): 22 """ 23 ``` 24 file_spec Possibly wildcarded string speficying files to be opened. 25 Special case: if file_spec is None, read from stdin. 26 27 tail If False, return None upon reaching end of last file; if 28 True, block upon reaching EOF of last file and wait for 29 more records. 30 31 refresh_file_spec 32 If True, refresh the search for matching filenames when 33 reaching last EOF to see if any new matching files have 34 appeared in the interim. 35 36 retry_interval 37 If tail and/or refresh_file_spec are True, how long to 38 wait before looking to see if any new records or files 39 have shown up. 40 41 interval 42 How long to sleep between returning records. In general 43 this should be zero except for debugging purposes. 44 45 eol Optional character by which to recognize the end of a record 46 ``` 47 Note that the order in which files are opened will probably be in 48 alphanumeric by filename, but this is not strictly enforced and 49 depends on how glob returns them. 50 """ 51 52 super().__init__(*args, **kwargs) 53 54 self.file_spec = file_spec 55 self.tail = tail 56 self.refresh_file_spec = refresh_file_spec 57 self.retry_interval = retry_interval 58 self.interval = interval 59 self.eol = eol 60 61 # If interval != 0, we need to keep track of our last_read to know 62 # how long to sleep 63 self.last_read = 0 64 65 # The file we're currently using 66 self.current_file = None 67 68 self.pos = 0 69 self.start_pos = {} 70 self.end_pos = {} 71 72 # Special case if file_spec is None 73 if file_spec is None: 74 self.current_file = sys.stdin 75 self.used_file_list = [] 76 self.unused_file_list = [] 77 self.tail = True 78 return 79 80 # Which files will we use, which haven't we used yet? 81 self.unused_file_list = sorted(glob.glob(file_spec)) 82 if not self.unused_file_list: 83 logging.warning('TextFileReader: file_spec "%s" matches no files', 84 file_spec) 85 self.used_file_list = []
file_spec Possibly wildcarded string speficying files to be opened.
Special case: if file_spec is None, read from stdin.
tail If False, return None upon reaching end of last file; if
True, block upon reaching EOF of last file and wait for
more records.
refresh_file_spec
If True, refresh the search for matching filenames when
reaching last EOF to see if any new matching files have
appeared in the interim.
retry_interval
If tail and/or refresh_file_spec are True, how long to
wait before looking to see if any new records or files
have shown up.
interval
How long to sleep between returning records. In general
this should be zero except for debugging purposes.
eol Optional character by which to recognize the end of a record
Note that the order in which files are opened will probably be in alphanumeric by filename, but this is not strictly enforced and depends on how glob returns them.
def
read(self):
122 def read(self): 123 """Get the next line of text. Return None if there are no more 124 records. To test EOF you'll need to test 125 126 if record is None: 127 no more records... 128 129 rather than simply 130 131 if not record: 132 could be EOF or simply an empty next line 133 """ 134 if self.interval: 135 now = time.time() 136 sleep_time = max(0, self.interval - (now - self.last_read)) 137 logging.debug('Sleeping %f seconds', sleep_time) 138 if sleep_time: 139 time.sleep(sleep_time) 140 141 record = None 142 while not record: 143 # If we've got a current file, or if _get_next_file() gets one 144 # for us, try to read a record. 145 if self.current_file or self._get_next_file(): 146 if not self.eol: 147 record = self.current_file.readline() 148 else: 149 record = self._read_until_eol() 150 if record: 151 self.last_read = time.time() 152 record = record.rstrip('\n') 153 logging.debug('TextFileReader got record "%s"', record) 154 self.pos += 1 155 return record 156 157 # No record: our current_file has reached EOF. See if more 158 # files we should try to read. 159 if self._get_next_file(): 160 # Found a new file to read - loop again right away 161 continue 162 163 # EOF when we're reading from stdin means we're done 164 if not self.file_spec: 165 return None 166 167 # No record, no new files, no tail or refresh directive - 168 # there's nothing left for us to try. Go home empty-handed. 169 if not self.refresh_file_spec and not self.tail: 170 return None 171 172 # User wants refresh or tail, so sleep and try again. 173 logging.debug('TextFileReader - tail/refresh specified, so sleeping ' 174 '%f seconds before trying again', self.retry_interval) 175 time.sleep(self.retry_interval)
Get the next line of text. Return None if there are no more records. To test EOF you'll need to test
if record is None: no more records...
rather than simply
if not record: could be EOF or simply an empty next line
def
seek(self, offset=0, origin='current'):
280 def seek(self, offset=0, origin='current'): 281 original_state = self._save_state() 282 283 try: 284 if origin == 'start': 285 if offset < 0: 286 raise ValueError("Can't back up past earliest record") 287 self.used_file_list = [] 288 self.unused_file_list = sorted(glob.glob(self.file_spec)) 289 self.current_file = None 290 self.pos = 0 291 self._seek_forward_from_current(offset) 292 293 elif origin == 'current': 294 if offset >= 0: 295 self._seek_forward_from_current(offset) 296 else: 297 self._seek_back_from_current(offset) 298 299 elif origin == 'end': 300 # Have to count lines in all files that haven't been processed yet. 301 # TODO: take self.refresh_file_spec into account 302 file_list = sorted(glob.glob(self.file_spec)) 303 pos = 0 304 for filename in file_list: 305 if filename in self.end_pos: 306 pos = self.end_pos[filename] 307 else: 308 self.start_pos[filename] = pos 309 310 # TODO: this can be made faster, if needed 311 with open(filename) as f: 312 for n, _ in enumerate(f, 1): 313 pass 314 315 pos += n 316 self.end_pos[filename] = pos 317 318 self.used_file_list = file_list 319 self.unused_file_list = [] 320 self.current_file = None 321 self.pos = pos 322 self._seek_back_from_current(offset) 323 324 else: 325 raise ValueError('Unknown origin value: "%s"' % origin) 326 327 except: # noqa: E722 328 self._restore_state(original_state) 329 raise 330 331 return self.pos
def
read_range(self, start=None, stop=None):
334 def read_range(self, start=None, stop=None): 335 """ 336 Read a range of records beginning with record number start, and ending 337 *before* record number stop. 338 """ 339 if start is None: 340 start = 0 341 if stop is None: 342 stop = sys.maxsize 343 self.seek(start, 'start') 344 records = [] 345 for _ in range(stop - start): 346 record = self.read() 347 if record is None: 348 break 349 records.append(record) 350 return records
Read a range of records beginning with record number start, and ending before record number stop.