r"""Codec used when reading a file in universal newlines mode. It wraps another incremental decoder, translating \r\n and \r into \n. It also records the types of newlines encountered. When used with translate=False, it ensures that the newline sequence is returned in one piece.
| 1960 | |
| 1961 | |
| 1962 | class IncrementalNewlineDecoder(codecs.IncrementalDecoder): |
| 1963 | r"""Codec used when reading a file in universal newlines mode. It wraps |
| 1964 | another incremental decoder, translating \r\n and \r into \n. It also |
| 1965 | records the types of newlines encountered. When used with |
| 1966 | translate=False, it ensures that the newline sequence is returned in |
| 1967 | one piece. |
| 1968 | """ |
| 1969 | def __init__(self, decoder, translate, errors='strict'): |
| 1970 | codecs.IncrementalDecoder.__init__(self, errors=errors) |
| 1971 | self.translate = translate |
| 1972 | self.decoder = decoder |
| 1973 | self.seennl = 0 |
| 1974 | self.pendingcr = False |
| 1975 | |
| 1976 | def decode(self, input, final=False): |
| 1977 | # decode input (with the eventual \r from a previous pass) |
| 1978 | if self.decoder is None: |
| 1979 | output = input |
| 1980 | else: |
| 1981 | output = self.decoder.decode(input, final=final) |
| 1982 | if self.pendingcr and (output or final): |
| 1983 | output = "\r" + output |
| 1984 | self.pendingcr = False |
| 1985 | |
| 1986 | # retain last \r even when not translating data: |
| 1987 | # then readline() is sure to get \r\n in one pass |
| 1988 | if output.endswith("\r") and not final: |
| 1989 | output = output[:-1] |
| 1990 | self.pendingcr = True |
| 1991 | |
| 1992 | # Record which newlines are read |
| 1993 | crlf = output.count('\r\n') |
| 1994 | cr = output.count('\r') - crlf |
| 1995 | lf = output.count('\n') - crlf |
| 1996 | self.seennl |= (lf and self._LF) | (cr and self._CR) \ |
| 1997 | | (crlf and self._CRLF) |
| 1998 | |
| 1999 | if self.translate: |
| 2000 | if crlf: |
| 2001 | output = output.replace("\r\n", "\n") |
| 2002 | if cr: |
| 2003 | output = output.replace("\r", "\n") |
| 2004 | |
| 2005 | return output |
| 2006 | |
| 2007 | def getstate(self): |
| 2008 | if self.decoder is None: |
| 2009 | buf = b"" |
| 2010 | flag = 0 |
| 2011 | else: |
| 2012 | buf, flag = self.decoder.getstate() |
| 2013 | flag <<= 1 |
| 2014 | if self.pendingcr: |
| 2015 | flag |= 1 |
| 2016 | return buf, flag |
| 2017 | |
| 2018 | def setstate(self, state): |
| 2019 | buf, flag = state |
no outgoing calls
no test coverage detected
searching dependent graphs…