For testing seek/tell behavior with a stateful, buffering decoder. Input is a sequence of words. Words may be fixed-length (length set by input) or variable-length (period-terminated). In variable-length mode, extra periods are ignored. Possible words are: - 'i' followed b
| 2702 | # - The number of output characters can vary depending on previous input. |
| 2703 | |
| 2704 | class StatefulIncrementalDecoder(codecs.IncrementalDecoder): |
| 2705 | """ |
| 2706 | For testing seek/tell behavior with a stateful, buffering decoder. |
| 2707 | |
| 2708 | Input is a sequence of words. Words may be fixed-length (length set |
| 2709 | by input) or variable-length (period-terminated). In variable-length |
| 2710 | mode, extra periods are ignored. Possible words are: |
| 2711 | - 'i' followed by a number sets the input length, I (maximum 99). |
| 2712 | When I is set to 0, words are space-terminated. |
| 2713 | - 'o' followed by a number sets the output length, O (maximum 99). |
| 2714 | - Any other word is converted into a word followed by a period on |
| 2715 | the output. The output word consists of the input word truncated |
| 2716 | or padded out with hyphens to make its length equal to O. If O |
| 2717 | is 0, the word is output verbatim without truncating or padding. |
| 2718 | I and O are initially set to 1. When I changes, any buffered input is |
| 2719 | re-scanned according to the new I. EOF also terminates the last word. |
| 2720 | """ |
| 2721 | |
| 2722 | def __init__(self, errors='strict'): |
| 2723 | codecs.IncrementalDecoder.__init__(self, errors) |
| 2724 | self.reset() |
| 2725 | |
| 2726 | def __repr__(self): |
| 2727 | return '<SID %x>' % id(self) |
| 2728 | |
| 2729 | def reset(self): |
| 2730 | self.i = 1 |
| 2731 | self.o = 1 |
| 2732 | self.buffer = bytearray() |
| 2733 | |
| 2734 | def getstate(self): |
| 2735 | i, o = self.i ^ 1, self.o ^ 1 # so that flags = 0 after reset() |
| 2736 | return bytes(self.buffer), i*100 + o |
| 2737 | |
| 2738 | def setstate(self, state): |
| 2739 | buffer, io = state |
| 2740 | self.buffer = bytearray(buffer) |
| 2741 | i, o = divmod(io, 100) |
| 2742 | self.i, self.o = i ^ 1, o ^ 1 |
| 2743 | |
| 2744 | def decode(self, input, final=False): |
| 2745 | output = '' |
| 2746 | for b in input: |
| 2747 | if self.i == 0: # variable-length, terminated with period |
| 2748 | if b == ord('.'): |
| 2749 | if self.buffer: |
| 2750 | output += self.process_word() |
| 2751 | else: |
| 2752 | self.buffer.append(b) |
| 2753 | else: # fixed-length, terminate after self.i bytes |
| 2754 | self.buffer.append(b) |
| 2755 | if len(self.buffer) == self.i: |
| 2756 | output += self.process_word() |
| 2757 | if final and self.buffer: # EOF terminates the last word |
| 2758 | output += self.process_word() |
| 2759 | return output |
| 2760 | |
| 2761 | def process_word(self): |