UNGETC在Python中
https://stackoverflow.com/questions/2655643
italiano
english
français
española
中国
日本の
العربية
Deutsch
한국어
Português
Russian题
python中的一些文件读取（readlines（））函数
将文件内容复制到内存（作为列表）
我需要处理一个太大的文件
在记忆中复制，因此需要使用
文件指针（要访问文件一个字节
一次） - 如c getc（）。
我的其他要求是
我想将文件指针倒入以上
字节如c ungetc（）中的字节。
有没有办法在Python中做到这一点？
另外，在Python，我可以在A中阅读一行
带有readline（）的时间
有没有办法阅读上一行
向后？
其他提示
您不需要Python没有或想要的文件指针。
要逐行浏览文件，而无需将整个内容阅读到内存中，只是在文件对象本身上迭代，即
with open(filename, "r") as f: for line in f: ...使用
readlines通常要避免。返回一行并不是您可以超级地做的事情。如果您永远不需要返回多个行，请查看
pairwise食谱itertools文档.
好的，这就是我想出的。感谢布伦达（Brenda）建立课程的想法。
感谢Josh使用C型函数seek（）和read（）的想法
#!/bin/python
# Usage: BufRead.py inputfile
import sys, os, string
from inspect import currentframe
# Debug function usage
#
# if DEBUG:
# debugLogMsg(currentframe().f_lineno,currentframe().f_code.co_filename)
# print ...
def debugLogMsg(line,file,msg=""):
print "%s:%s %s" % (file,line,msg)
# Set DEBUG off.
DEBUG = 0
class BufRead:
def __init__(self,filename):
self.__filename = filename
self.__file = open(self.__filename,'rb')
self.__fileposition = self.__file.tell()
self.__file.seek(0, os.SEEK_END)
self.__filesize = self.__file.tell()
self.__file.seek(self.__fileposition, os.SEEK_SET)
def close(self):
if self.__file is not None:
self.__file.close()
self.__file = None
def seekstart(self):
if self.__file == None:
self.__file.seek(0, os.SEEK_SET)
self.__fileposition = self.__file.tell()
def seekend(self):
if self.__file == None:
self.__file.seek(-1, os.SEEK_END)
self.__fileposition = self.__file.tell()
def getc(self):
if self.__file == None:
return None
self.__fileposition = self.__file.tell()
if self.__fileposition < self.__filesize:
byte = self.__file.read(1)
self.__fileposition = self.__file.tell()
return byte
else:
return None
def ungetc(self):
if self.__file == None:
return None
self.__fileposition = self.__file.tell()
if self.__fileposition > 0:
self.__fileposition = self.__fileposition - 1
self.__file.seek(self.__fileposition, os.SEEK_SET)
byte = self.__file.read(1)
self.__file.seek(self.__fileposition, os.SEEK_SET)
return byte
else:
return None
# uses getc() and ungetc()
def getline(self):
if self.__file == None:
return None
self.__fileposition = self.__file.tell()
if self.__fileposition < self.__filesize:
startOfLine = False
line = ""
while True:
if self.__fileposition == 0:
startOfLine = True
break
else:
c = self.ungetc()
if c == '\n':
c = self.getc()
startOfLine = True
break
if startOfLine:
c = self.getc()
if c == '\n':
return '\n'
else:
self.ungetc()
while True:
c = self.getc()
if c == '\n':
line += c
c = self.getc()
if c == None:
return line
if c == '\n':
self.ungetc()
return line
elif c == None:
return line
else:
line += c
else:
return None
# uses getc() and ungetc()
def ungetline(self):
if self.__file == None:
return None
self.__fileposition = self.__file.tell()
if self.__fileposition > 0:
endOfLine = False
line = ""
while True:
if self.__fileposition == self.__filesize:
endOfLine = True
break
else:
c = self.getc()
if c == '\n':
c = self.ungetc()
endOfLine = True
break
if endOfLine:
c = self.ungetc()
if c == '\n':
return '\n'
else:
self.getc()
while True:
c = self.ungetc()
if c == None:
return line
if c == '\n':
line += c
c = self.ungetc()
if c == None:
return line
if c == '\n':
self.getc()
return line
elif c == None:
return line
else:
line = c + line
else:
return None
def main():
if len(sys.argv) == 2:
print sys.argv[1]
b = BufRead(sys.argv[1])
sys.stdout.write(
'----------------------------------\n' \
'- TESTING GETC \n' \
'----------------------------------\n')
while True:
c = b.getc()
if c == None:
sys.stdout.write('\n')
break
else:
sys.stdout.write(c)
sys.stdout.write(
'----------------------------------\n' \
'- TESTING UNGETC \n' \
'----------------------------------\n')
while True:
c = b.ungetc()
if c == None:
sys.stdout.write('\n')
break
else:
sys.stdout.write(c)
sys.stdout.write(
'----------------------------------\n' \
'- TESTING GETLINE \n' \
'----------------------------------\n')
b.seekstart()
while True:
line = b.getline()
if line == None:
sys.stdout.write('\n')
break
else:
sys.stdout.write(line)
sys.stdout.write(
'----------------------------------\n' \
'- TESTING UNGETLINE \n' \
'----------------------------------\n')
b.seekend()
while True:
line = b.ungetline()
if line == None:
sys.stdout.write('\n')
break
else:
sys.stdout.write(line)
b.close()
if __name__=="__main__": main()
为您编写读取和缓冲区输入的类，并在其中实现UNGETC-也许是这样的（警告：未经测试，编译时编写）：
class BufRead:
def __init__(self,filename):
self.filename = filename
self.fn = open(filename,'rb')
self.buffer = []
self.bufsize = 256
self.ready = True
def close(self):
if self.fn is not None:
self.fn.close()
self.fn = None
self.ready = False
def read(self,size=1):
l = len(self.buffer)
if not self.ready: return None
if l <= size:
s = self.buffer[:size]
self.buffer = self.buffer[size:]
return s
s = self.buffer
size = size - l
self.buffer = self.fn.read(min(self.bufsize,size))
if self.buffer is None or len(self.buffer) == 0:
self.ready = False
return s
return s + self.read(size)
def ungetc(self,ch):
if self.buffer is None:
self.buffer = [ch]
else:
self.buffer.append(ch)
self.ready = True
我不想做数十亿个无情的单炭文件读取和我想要一种方法
调试文件指针的位置。因此，我决心返回文件位置
除了字符或线路外，还使用MMAP将文件映射到内存。 （并让MMAP
处理分页）我认为，如果文件确实非常大，这将是一个问题。 （就像大于物理内存的数量一样）MMAP开始进入虚拟内存，并且事情可能会变得非常慢。目前，它在大约4分钟内处理50 MB文件。
import sys, os, string, re, time
from mmap import mmap
class StreamReaderDb:
def __init__(self,stream):
self.__stream = mmap(stream.fileno(), os.path.getsize(stream.name))
self.__streamPosition = self.__stream.tell()
self.__stream.seek(0 , os.SEEK_END)
self.__streamSize = self.__stream.tell()
self.__stream.seek(self.__streamPosition, os.SEEK_SET)
def setStreamPositionDb(self,streamPosition):
if self.__stream == None:
return None
self.__streamPosition = streamPosition
self.__stream.seek(self.__streamPosition, os.SEEK_SET)
def streamPositionDb(self):
if self.__stream == None:
return None
return self.__streamPosition
def streamSize(self):
if self.__stream == None:
return None
return self.__streamSize
def close(self):
if self.__stream is not None:
self.__stream.close()
self.__stream = None
def seekStart(self):
if self.__stream == None:
return None
self.setStreamPositionDb(0)
def seekEnd(self):
if self.__stream == None:
return None
self.__stream.seek(-1, os.SEEK_END)
self.setStreamPositionDb(self.__stream.tell())
def getcDb(self):
if self.__stream == None:
return None,None
self.setStreamPositionDb(self.__stream.tell())
if self.streamPositionDb() < self.streamSize():
byte = self.__stream.read(1)
self.setStreamPositionDb(self.__stream.tell())
return byte,self.streamPositionDb()
else:
return None,self.streamPositionDb()
def unGetcDb(self):
if self.__stream == None:
return None,None
self.setStreamPositionDb(self.__stream.tell())
if self.streamPositionDb() > 0:
self.setStreamPositionDb(self.streamPositionDb() - 1)
byte = self.__stream.read(1)
self.__stream.seek(self.streamPositionDb(), os.SEEK_SET)
return byte,self.streamPositionDb()
else:
return None,self.streamPositionDb()
def seekLineStartDb(self):
if self.__stream == None:
return None
self.setStreamPositionDb(self.__stream.tell())
if self.streamPositionDb() < self.streamSize():
# Back up to the start of the line
while True:
if self.streamPositionDb() == 0:
return self.streamPositionDb()
else:
c,fp = self.unGetcDb()
if c == '\n':
c,fp = self.getcDb()
return fp
else:
return None
def seekPrevLineEndDb(self):
if self.__stream == None:
return None
self.setStreamPositionDb(self.__stream.tell())
if self.streamPositionDb() < self.streamSize():
# Back up to the start of the line
while True:
if self.streamPositionDb() == 0:
return self.streamPositionDb()
else:
c,fp = self.unGetcDb()
if c == '\n':
return fp
else:
return None
def seekPrevLineStartDb(self):
if self.__stream == None:
return None
self.setStreamPositionDb(self.__stream.tell())
if self.streamPositionDb() < self.streamSize():
# Back up to the start of the line
while True:
if self.streamPositionDb() == 0:
return self.streamPositionDb()
else:
c,fp = self.unGetcDb()
if c == '\n':
return self.seekLineStartDb()
else:
return None
def seekLineEndDb(self):
if self.__stream == None:
return None
self.setStreamPositionDb(self.__stream.tell())
if self.streamPositionDb() > 0:
while True:
if self.streamPositionDb() == self.streamSize():
return self.streamPositionDb()
else:
c,fp = self.getcDb()
if c == '\n':
c,fp = self.unGetcDb()
return fp
else:
return None
def seekNextLineEndDb(self):
if self.__stream == None:
return None
self.setStreamPositionDb(self.__stream.tell())
if self.streamPositionDb() > 0:
while True:
if self.streamPositionDb() == self.streamSize():
return self.streamPositionDb()
else:
c,fp = self.getcDb()
if c == '\n':
return fp
else:
return None
def seekNextLineStartDb(self):
if self.__stream == None:
return None
self.setStreamPositionDb(self.__stream.tell())
if self.streamPositionDb() > 0:
while True:
if self.streamPositionDb() == self.streamSize():
return self.streamPositionDb()
else:
c,fp = self.getcDb()
if c == '\n':
return self.seekLineStartDb()
else:
return None
# uses getc() and ungetc()
def getLineDb(self):
if self.__stream == None:
return None,None
self.setStreamPositionDb(self.__stream.tell())
line = ""
if self.seekLineStartDb() != None:
c,fp = self.getcDb()
if c == '\n':
return c,self.streamPositionDb()
else:
self.unGetcDb()
while True:
c,fp = self.getcDb()
if c == '\n':
line += c
c,fp = self.getcDb()
if c == None:
return line,self.streamPositionDb()
self.unGetcDb()
return line,self.streamPositionDb()
elif c == None:
return line,self.streamPositionDb()
else:
line += c
else:
return None,self.streamPositionDb()
# uses getc() and ungetc()
def unGetLineDb(self):
if self.__stream == None:
return None,None
self.setStreamPositionDb(self.__stream.tell())
line = ""
if self.seekLineEndDb() != None:
c,fp = self.unGetcDb()
if c == '\n':
return c,self.streamPositionDb()
else:
self.getcDb()
while True:
c,fp = self.unGetcDb()
if c == None:
return line,self.streamPositionDb()
if c == '\n':
line += c
c,fp = self.unGetcDb()
if c == None:
return line,self.streamPositionDb()
self.getcDb()
return line,self.streamPositionDb()
elif c == None:
return line,self.streamPositionDb()
else:
line = c + line
else:
return None,self.streamPositionDb()
最初，我需要建立词汇分析仪的需求引起了问题。
getC（）和ungetc（）一开始很有用（以使读取错误的方式和
要构建状态机）
getC（）和ungetc（）成为责任，因为它们花费了太长时间直接从存储中阅读。
当状态机器完成时（调试任何IO问题，
最终确定了状态），我优化了词汇分析仪。
将源文件（或页面）中的源文件读取到内存和运行
每个页面上的状态机都会产生最佳的结果。
我发现如果不使用getC（）和ungetc（），可以节省大量时间
直接从文件中读取。
不隶属于 StackOverflow
