ungetc в Python

Некоторые функции чтения файлов (readlines()) в Python
копируют содержимое файла в память (в виде списка)

Мне нужно обработать файл, который слишком велик для
копирования в память, и поэтому мне нужно использовать
файловый указатель (для доступа к файлу по одному байту
за раз) -- как в C getc( ).

У меня есть дополнительное требование:
я хочу перемотать указатель файла на предыдущие
байты, как в C ungetc().

Есть ли способ сделать это в Python?

Кроме того, в Python я могу читать по одной строке
с помощью readline().

Есть ли способ прочитать предыдущую строку
в обратном порядке?


person Community    schedule 16.04.2010    source источник
comment
худшее / стихотворение / когда-либо. :)   -  person kennytm    schedule 16.04.2010
comment
Как насчет ungetch из модуля msvcrt?   -  person    schedule 17.04.2010


Ответы (6)


Если вы хотите использовать указатель файла напрямую (хотя я думаю, что предложение Майка Грэма лучше), вы можете использовать объект файла seek(), который позволяет установить внутренний указатель в сочетании с read(), который поддерживает параметр option, указывающий, сколько байтов вы хотите прочитать.

person Josh Wright    schedule 16.04.2010

  • Вам не нужны файловые указатели, которых Python не имеет или не хочет.

  • Чтобы просмотреть файл построчно, не считывая все это в память, просто выполните итерацию по самому файловому объекту, т.е.

     with open(filename, "r") as f:
         for line in f:
             ...
    

    Обычно следует избегать использования readlines.

  • Вернуться назад — это не то, что вы можете сделать очень легко. Если вам никогда не нужно возвращаться более чем на одну строку, ознакомьтесь с рецептом pairwise в itertools документации< /а>.

person Mike Graham    schedule 16.04.2010

Хорошо, вот что я придумал. Спасибо Бренде за идею создания класса.
Спасибо Джошу за идею использовать 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()
person Community    schedule 13.05.2010

Напишите класс для чтения и буферизации ввода для вас и реализуйте для него 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
person Brenda Holloway    schedule 16.04.2010
comment
Я нахожусь в процессе создания класса для буферизованного чтения файла через seek(). - person ; 30.04.2010

Я не хочу делать миллиарды небуферизованных чтений файлов с одним символом, плюс мне нужен способ
отлаживать положение указателя файла. Поэтому я решил вернуть позицию файла
в дополнение к символу или строке и использовать mmap для сопоставления файла с памятью. (и пусть mmap
обрабатывает пейджинг) Я думаю, это будет проблемой, если файл действительно очень большой. (например, больше, чем объем физической памяти). Именно тогда mmap начнет входить в виртуальную память, и все может стать очень медленным. На данный момент он обрабатывает файл размером 50 МБ примерно за 4 минуты.

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()
person Community    schedule 19.05.2010

Первоначально вопрос был вызван моей потребностью в создании лексического анализатора.
getc() и ungetc() полезны в первую очередь (чтобы избавиться от ошибок чтения и
построить конечный автомат). машина готова,
getc() и ungetc() становятся помехой, так как они занимают слишком много времени для прямого чтения из хранилища.

Когда конечный автомат был готов (отладил все проблемы с вводом-выводом,
доработал состояния), я оптимизировал лексический анализатор.

Чтение исходного файла фрагментами (или страницами) в память и запуск
конечного автомата на каждой странице дает наилучший результат по времени.

Я обнаружил, что значительно экономится время, если getc() и ungetc() не используются
для прямого чтения из файла.

person Community    schedule 20.05.2010
comment
Мне это кажется странным, поскольку открытые/прочитанные питоны буферизуют ввод-вывод, поэтому вы все равно читаете из памяти. Если вы не используете os.open - person Erik Aronesty; 05.03.2019