UNGETC dans Python
Question
Quelques fonctions de lecture de fichiers (readlines ()) dans Python
Copiez le contenu du fichier en mémoire (comme une liste)
J'ai besoin de traiter un fichier trop grand pour
être copié en mémoire et en tant que tel besoin d'utiliser
un pointeur de fichier (pour accéder au fichier un octet
à la fois) - comme dans c getc ().
L'exigence supplémentaire que j'ai est que
Je voudrais rembobiner le pointeur de fichier vers
octets comme dans c ungetc ().
Existe-t-il un moyen de faire cela dans Python?
De plus, à Python, je peux lire une ligne à un
Temps avec ReadLine ()
Existe-t-il un moyen de lire la ligne précédente
revenir en arrière?
La solution
Si vous souhaitez utiliser un pointeur de fichier directement (je pense que la suggestion de Mike Graham est meilleure cependant), vous pouvez utiliser l'objet de fichier chercher() Méthode qui vous permet de définir le pointeur interne, combiné avec le lis() Méthode, qui prend en charge un argument d'option spécifiant le nombre d'octets que vous souhaitez lire.
Autres conseils
Vous n'avez pas besoin de pointeurs de fichiers, que Python n'a pas ou ne veut pas.
Pour passer par un fichier ligne par ligne sans lire le tout en mémoire, il suffit d'itérer sur l'objet de fichier lui-même, c'est-à-dire
with open(filename, "r") as f: for line in f: ...
Utilisant
readlines
est généralement à éviter.Retourner une ligne n'est pas quelque chose que vous pouvez faire avec mal à l'aise. Si vous n'avez jamais besoin de revenir en arrière plus d'une ligne, consultez le
pairwise
recette dans leitertools
Documentation.
Ok, voici ce que j'ai trouvé. Merci Brenda pour l'idée de construire une classe.
Merci Josh pour l'idée d'utiliser C Like Fonctions Seek () et 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()
Écrivez un cours les lectures et les tampons entrées pour vous, et implémentez ungetc dessus - quelque chose comme ça peut-être (avertissement: non testé, écrit pendant la compilation):
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
Je ne veux pas faire des milliards de lectures de fichiers de char single à se déshabillé et je voulais un moyen
Pour déboguer la position du pointeur de fichier. Par conséquent, j'ai décidé de retourner la position du fichier
En plus d'un char ou d'une ligne et pour utiliser MMAP pour mapper le fichier en mémoire. (Et laissez mmap
gérer la pagination) Je pense que ce sera un peu un problème si le fichier est vraiment, vraiment grand. (comme dans plus grand que la quantité de mémoire physique) C'est à ce moment que MMAP commencerait à entrer dans la mémoire virtuelle et les choses pourraient devenir très lentes. Pour l'instant, il traite un fichier de 50 Mo en environ 4 min.
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()
La question a été initialement invitée par mon besoin de construire un analyseur lexical.
getc () et ungetc () sont utiles au début (pour éliminer les bugs de lecture et
Pour construire la machine d'état) une fois la machine d'état terminée,
getc () et ungetc () deviennent un passif car ils mettent trop de temps à lire directement dans le stockage.
Une fois la machine d'État terminée (déboguant tous les problèmes d'IO,
Finalisé les états), j'ai optimisé l'analyseur lexical.
Lire le fichier source en morceaux (ou pages) dans la mémoire et en cours d'exécution
La machine d'état sur chaque page donne le meilleur résultat de temps.
J'ai constaté que le temps considérable est enregistré si getc () et ungetc () ne sont pas utilisés
pour lire directement le fichier.