Pythonでリングバッファを実装するクラスを作成してみた

はじめに

こんにちは！今回はPythonでリングバッファ(Ring Buffer/Circular Buffer)を実装するプログラムを紹介します。
リングバッファは、一定のサイズのバッファを使用してデータを循環的に管理するデータ構造です。特に、メモリを固定サイズで管理したいときやキューのような操作を効率よく行きたい場合に役立ちます。

リングバッファとは？

リングバッファは、配列を循環的(リング状)に扱うデータ構造です。配列の末尾に達した場合でも、先頭に戻ってデータを格納できるため、一定サイズ内のメモリ管理に最適です。
例えば、サイズが5のリングバッファに対して以下のような操作を行った場合:
・enqueue:データを追加(末尾に追加)
・dequeue:データを削除(先頭から取り出し)
バッファは常に新しいデータで上書きされます。

実装コード

以下のpythonクラスRingBufferでは、バッファの初期化、データの追加(engueue)、データの取り出し(dequeue)、バッファの状態を表示するdisplayメソッドを実装しています。

class RingBuffer:
    def __init__(self, size, initial_values: list = None):
        self.size = size          
        self.buffer = [None] * size  
        self.head = 0             
        self.tail = 0             
        self.count = 0            
        self.operation_count = 0  
        self.dequeue_count = 0    
        self.hoge = []            

        if initial_values:
            for i, value in enumerate(initial_values):
                self.buffer[i] = value
            self.tail = len(initial_values)
            self.count = len(initial_values)
            self.hoge.append(list(self.buffer))

        print("=== 初期状態 ===")
        self.display()

    def enqueue(self, item):
        if self.count == self.size:
            print("バッファが満杯です。エンキューできません。")
            return
        self.buffer[self.tail] = item
        self.tail = (self.tail + 1) % self.size
        self.count += 1
        self.operation_count += 1
        print(f"{self.operation_count}. enqueue(Q, {item})")
        self.hoge.append(list(self.buffer))
        self.display()

    def dequeue(self):
        if self.count == 0:
            print("バッファが空です。デキューできません。")
            return None
        item = self.buffer[self.head]
        self.buffer[self.head] = None
        self.head = (self.head + 1) % self.size
        self.count -= 1
        self.operation_count += 1
        self.dequeue_count += 1
        print(f"{self.operation_count}. {self.dequeue_count}回目のdequeue(Q): {item}")
        self.hoge.append(list(self.buffer))
        self.display()

        return item

    def display(self):
        print(f"  Q[i]: {self.buffer}\n")

使用例

以下のコードでリングバッファを初期化し、データを追加・削除する操作を試します。

# 使用例
Q = RingBuffer(5, [17, 43, 86])
Q.enqueue(71)
Q.dequeue()
Q.enqueue(19)
Q.enqueue(61)
Q.dequeue()
Q.enqueue(34)
Q.dequeue()
Q.dequeue()

import pandas as pd
pd.DataFrame(Q.hoge, index=list(range(len(Q.hoge))), columns=list(range(len(Q.hoge[0]))))

実行結果の例

=== 初期状態 ===
  Q[i]: [17, 43, 86, None, None]

1. enqueue(Q, 71)
  Q[i]: [17, 43, 86, 71, None]

2. 1回目のdequeue(Q): 17
  Q[i]: [None, 43, 86, 71, None]

3. enqueue(Q, 19)
  Q[i]: [None, 43, 86, 71, 19]
  ・
  ・
  ・

まとめ

このリングバッファの実装は、キューのような操作を循環的に行きたいときに便利です。今回のプログラムを応用して、例えばリアルタイムログやセンサーのデータストリームなどを管理する際にも利用できます。最後まで読んでくださり、ありがとうございました。もし改善点や質問があれば、ぜひコメントしてください！