a text file circular buffer in python

Question

I need a python script implementing a circular buffer for rows in a text file limited to N rows like this:

        row 1 -> pop
        row 2
        row 3
         |
         |
push -> row N

What's the best solution?

EDIT: This script should create and maintain the text file which only contains the latest N lines. Then it should pop the first line pushed in. Like a fifo buffer.

Streaming

Do you just want to read the last N lines of a text file, or is your script creating and maintaining the text file which only contains the last N lines you've written to it? – Puddle

My script should create and maintain the text file which only contains the latest N lines. Then it should pop the first line pushed in. Like a fifo buffer. – Streaming

Answer 1

Try my recipe and sorry for italian usage:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
#       fifo(.py)
#       
#       Copyright 2011 Fabio Di Bernardini <[email protected]>
#       
#       This program is free software; you can redistribute it and/or modify
#       it under the terms of the GNU General Public License as published by
#       the Free Software Foundation; either version 2 of the License, or
#       (at your option) any later version.
#       
#       This program is distributed in the hope that it will be useful,
#       but WITHOUT ANY WARRANTY; without even the implied warranty of
#       MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#       GNU General Public License for more details.
#       
#       You should have received a copy of the GNU General Public License
#       along with this program; if not, write to the Free Software
#       Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
#       MA 02110-1301, USA.

def string_conditioned(string):
    return string.decode('string_escape').rstrip() + '\n'

def pop(n, size, filename):
    with open(filename, 'r+U') as fd:
        rows = fd.readlines()
    with open(filename, 'w') as fd:
        n = int(n)
        fd.writelines(rows[n:])
        return ''.join(rows[:n])

def trim_fifo(row, size, filename):
    size = int(size)
    with open(filename, 'rU') as fd:
        rows = fd.readlines()
    num_rows = len(rows)
    if num_rows >= size:
        n = string_conditioned(row).count('\n')
        pop(num_rows + n - size, size, filename)

def push(row, size, filename):
    trim_fifo(row, size, filename)
    with open(filename, 'a') as fd:
        fd.write(string_conditioned(row))
    return ''

def main():
    import sys
    try:
        command  = sys.argv[1]
        param    = sys.argv[2]
        size     = sys.argv[3]
        filename = sys.argv[4]
        sys.stdout.write({
        '--push': push,
        '--pop' : pop,
        }[command](param, size, filename))
    except Exception, e:
        print r"""
Uso:
       fifo --push ROW MAX_ROWS FILE
       fifo --pop  NUM MAX_ROWS FILE

fifo implementa un buffer ad anello di righe di testo, Quando viene inserita
una riga che fa superare il numero massimo di righe (MAX_ROWS) elimina la riga
più vecchia.

Comandi:
  --push    accoda la riga di testo ROW nel FILE rimuovendo le righe più vecchie
            se il file supera MAX_ROWS. Usare '\n' per separare righe multiple.
  --pop     stampa le prime NUM righe e le rimuove dal FILE. MAX_ROWS viene
            ignorato ma deve essere comunque specificato.

Esempi:
       fifo --push 'row_one \n row_two' 10 fifo.txt
       fifo --pop 2 10 fifo.txt
"""
        print e

if __name__ == '__main__':
    main()

Harbinger

Answer 2

Use collections.deque. It supports a maxlen parameter.

d = collections.deque(maxlen=10)
for line in f:
    d.append(line)
    # ...

Radioactivity

using a file don't work fd = open(filename, 'r+'); d = collections.deque(fd, maxlen=10) – Streaming

@s23: Works for me. What's the problem? – Radioactivity

...how? after d.append('string') the file is empty. Thank you. – Streaming

Answer 3

import collections

def keep_last_n_and_return_first_of_last_n(filename, n):
    with open(filename, "r") as inp:
         lines= collections.deque(inp, maxlen=n)
    with open(filename, "w") as out:
         out.writelines(lines)
    return lines[0]

Penang