(self)
| 2338 | return position, dec_flags, bytes_to_feed, bool(need_eof), chars_to_skip |
| 2339 | |
| 2340 | def tell(self): |
| 2341 | if not self._seekable: |
| 2342 | raise UnsupportedOperation("underlying stream is not seekable") |
| 2343 | if not self._telling: |
| 2344 | raise OSError("telling position disabled by next() call") |
| 2345 | self.flush() |
| 2346 | position = self.buffer.tell() |
| 2347 | decoder = self._decoder |
| 2348 | if decoder is None or self._snapshot is None: |
| 2349 | if self._decoded_chars: |
| 2350 | # This should never happen. |
| 2351 | raise AssertionError("pending decoded text") |
| 2352 | return position |
| 2353 | |
| 2354 | # Skip backward to the snapshot point (see _read_chunk). |
| 2355 | dec_flags, next_input = self._snapshot |
| 2356 | position -= len(next_input) |
| 2357 | |
| 2358 | # How many decoded characters have been used up since the snapshot? |
| 2359 | chars_to_skip = self._decoded_chars_used |
| 2360 | if chars_to_skip == 0: |
| 2361 | # We haven't moved from the snapshot point. |
| 2362 | return self._pack_cookie(position, dec_flags) |
| 2363 | |
| 2364 | # Starting from the snapshot position, we will walk the decoder |
| 2365 | # forward until it gives us enough decoded characters. |
| 2366 | saved_state = decoder.getstate() |
| 2367 | try: |
| 2368 | # Fast search for an acceptable start point, close to our |
| 2369 | # current pos. |
| 2370 | # Rationale: calling decoder.decode() has a large overhead |
| 2371 | # regardless of chunk size; we want the number of such calls to |
| 2372 | # be O(1) in most situations (common decoders, sensible input). |
| 2373 | # Actually, it will be exactly 1 for fixed-size codecs (all |
| 2374 | # 8-bit codecs, also UTF-16 and UTF-32). |
| 2375 | skip_bytes = int(self._b2cratio * chars_to_skip) |
| 2376 | skip_back = 1 |
| 2377 | assert skip_bytes <= len(next_input) |
| 2378 | while skip_bytes > 0: |
| 2379 | decoder.setstate((b'', dec_flags)) |
| 2380 | # Decode up to temptative start point |
| 2381 | n = len(decoder.decode(next_input[:skip_bytes])) |
| 2382 | if n <= chars_to_skip: |
| 2383 | b, d = decoder.getstate() |
| 2384 | if not b: |
| 2385 | # Before pos and no bytes buffered in decoder => OK |
| 2386 | dec_flags = d |
| 2387 | chars_to_skip -= n |
| 2388 | break |
| 2389 | # Skip back by buffered amount and reset heuristic |
| 2390 | skip_bytes -= len(b) |
| 2391 | skip_back = 1 |
| 2392 | else: |
| 2393 | # We're too far ahead, skip back a bit |
| 2394 | skip_bytes -= skip_back |
| 2395 | skip_back = skip_back * 2 |
| 2396 | else: |
| 2397 | skip_bytes = 0 |
no test coverage detected