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
| 42 | # - The number of output characters can vary depending on previous input. |
| 43 | |
| 44 | class StatefulIncrementalDecoder(codecs.IncrementalDecoder): |
| 45 | """ |
| 46 | For testing seek/tell behavior with a stateful, buffering decoder. |
| 47 | |
| 48 | Input is a sequence of words. Words may be fixed-length (length set |
| 49 | by input) or variable-length (period-terminated). In variable-length |
| 50 | mode, extra periods are ignored. Possible words are: |
| 51 | - 'i' followed by a number sets the input length, I (maximum 99). |
| 52 | When I is set to 0, words are space-terminated. |
| 53 | - 'o' followed by a number sets the output length, O (maximum 99). |
| 54 | - Any other word is converted into a word followed by a period on |
| 55 | the output. The output word consists of the input word truncated |
| 56 | or padded out with hyphens to make its length equal to O. If O |
| 57 | is 0, the word is output verbatim without truncating or padding. |
| 58 | I and O are initially set to 1. When I changes, any buffered input is |
| 59 | re-scanned according to the new I. EOF also terminates the last word. |
| 60 | """ |
| 61 | |
| 62 | def __init__(self, errors='strict'): |
| 63 | codecs.IncrementalDecoder.__init__(self, errors) |
| 64 | self.reset() |
| 65 | |
| 66 | def __repr__(self): |
| 67 | return '<SID %x>' % id(self) |
| 68 | |
| 69 | def reset(self): |
| 70 | self.i = 1 |
| 71 | self.o = 1 |
| 72 | self.buffer = bytearray() |
| 73 | |
| 74 | def getstate(self): |
| 75 | i, o = self.i ^ 1, self.o ^ 1 # so that flags = 0 after reset() |
| 76 | return bytes(self.buffer), i*100 + o |
| 77 | |
| 78 | def setstate(self, state): |
| 79 | buffer, io = state |
| 80 | self.buffer = bytearray(buffer) |
| 81 | i, o = divmod(io, 100) |
| 82 | self.i, self.o = i ^ 1, o ^ 1 |
| 83 | |
| 84 | def decode(self, input, final=False): |
| 85 | output = '' |
| 86 | for b in input: |
| 87 | if self.i == 0: # variable-length, terminated with period |
| 88 | if b == ord('.'): |
| 89 | if self.buffer: |
| 90 | output += self.process_word() |
| 91 | else: |
| 92 | self.buffer.append(b) |
| 93 | else: # fixed-length, terminate after self.i bytes |
| 94 | self.buffer.append(b) |
| 95 | if len(self.buffer) == self.i: |
| 96 | output += self.process_word() |
| 97 | if final and self.buffer: # EOF terminates the last word |
| 98 | output += self.process_word() |
| 99 | return output |
| 100 | |
| 101 | def process_word(self): |
no outgoing calls
searching dependent graphs…