Skip to content

CircularBuffer

But de l’exercice

Le but de l’exercice est d’écrire une classe CircularBuffer, dans un fichier circularbuffer.py dont le but est de fournir les fonctionnalités d'un buffer circulaire afin de pouvoir enregistrer et lire des données de manière infinie.

Et voila le script de tests unitaires : test_circularbuffer.py

Note

Désolé mais cet exercice sera certainement plus gros que d'habitude.

Avant-Propos

Qu'est-ce qu'un buffer circulaire? C'est une structure qui permet de rajouter des données de manière continue et de les lire de manière continue elle aussi.

Les fréquence d'ecriture et de lecture pouvant être differentes (on écrit par petit blocks toutes les quart de secondes, et on lit un gros blocks toutes les secondes).

Ce genre de buffer peut être utilisé pour de la visio conférence par exemple. On met les données enregistrées par le micro au fil de l'eau, et on va laisser le système de compression aller chercher les données dont il a besoin.

Pour de plus amples informations, vous pouvez aller lire plus d'informations ici: * Buffer Circulaire[fr] * Circular Buffer[en]

Perspectives

Gardez bien à l'esprit que nous n'allons pas implémenter un buffer circulaire qui soit 100% fonctionnel, pour cela nous aurions besoin de certifier notre classe pour supporter le multi-threading, être super optimisée, et tellement d'autres choses qui ne rentre pas dans le but de ce site.

Par contre, n'ayons pas peur non plus, nous obtiendrons quelque chose de fonctionnel et (re)utilisable.

Exercice de base

Dans l'exercice de base, nous allons créer un buffer circulaire d'une taille donnée dans le constructeur.

Il faudra que notre objet ait 2 fonctions principales: write et read afin de pouvoir ajouter des infos dans le buffer, et de les lire.

>>> buffer = CircularBuffer(3)
>>> buffer.write('1')
>>> buffer.read()
'1'

La fonction write doit prendre soit un élément, soit une liste d'éléments afin de pouvoir ecrire plusieurs éléments en une seule fois.

>>> buffer = CircularBuffer(3)
>>> buffer.write(('1', '2'))
>>> buffer.read()
'1'
>>> buffer.read()
'2'

La fonction read prend quant à elle un nombre de d'éléments à lire en une seule fois. Si le nombre d'éléments n'est pas donné, alors elle retourne seulement un élément.

>>> buffer = CircularBuffer(3)
>>> buffer.write((1, 2, 3))
>>> buffer.read()
1
>>> buffer.read(2)
(2, 3)

Et bien entendu, nous devons pouvoir écrire et lire de manière continue:

>>> buffer = CircularBuffer(3)
>>> success_count = 0
>>> for i in range(500):
...     buffer.write(i)
...     if buffer.read() == i:
...         success_count += 1
>>> success_count
500

Bonus numéro 1

Ajouter 3 fonctions dont le but est d'avoir des informations sur la taille disponible: is_empty, is_full et get_available_items

  • is_empty retourne True que s'il n'y a rien à lire
  • is_full retourne True que si le buffer est plein de choses à lire
  • get_available_items retourne le nombre d'éléments disponibles pour la lecture
>>> buffer = CircularBuffer(3)
>>> buffer.is_empty()
True
>>> buffer.is_full()
False
>>> buffer.get_available_items()
0
>>> buffer.write((1, 2, 3))
>>> buffer.is_empty()
False
>>> buffer.is_full()
True
>>> buffer.get_available_items()
3
>>> buffer.read()
1
>>> buffer.is_empty()
False
>>> buffer.is_full()
False
>>> buffer.get_available_items()
2

Maintenant que nous avons ces fonctions, il va falloir ajouter une representation sous forme de chaine de charactères à notre buffer:

>>> buffer = CircularBuffer(5)
>>> str(buffer)
CircularBuffer(size=5, available=0)
>>> buffer.write(1)
>>> str(buffer)
CircularBuffer(size=5, available=1)

Bonus numéro 2

Faire de notre buffer un iterable afin de pouvoir faire ceci par exemple:

>>> buffer = CircularBuffer(3)
>>> buffer.write((1, 2, 3))
>>> for i in buffer:
...     print(i)
1
2
3

Bonus numéro 3

Il faut gérer les dépassements.

Si on écrit alors que la lecture n'a pas eu lieu, il faudra pouvoir écraser les plus anciennes données avec les nouvelles:

>>> buffer = CircularBuffer(3)
>>> buffer.write((1, 2, 3, 4, 5, 6, 7, 8))
>>> buffer.read()
6

Si on lit trop de données, il faudra lancer une exception de type IndexError:

>>> buffer = CircularBuffer(3)
>>> buffer.read()
Exception raised:
IndexError: "No more item available for read in buffer"

Aide

Si vraiment vous êtes bloqué·e·s, vous pouvez regarder ces articles :

Solution

Voir la solution