Skip to content

Thread Safe Buffer

Thread-safe history buffer for PyEyesWeb.

This module provides a reusable thread-safe buffer implementation.

ThreadSafeHistoryBuffer

Thread-safe history buffer with deque backing.

Parameters:

Name Type Description Default
maxlen int

Maximum size of the history buffer

 required

Examples:

>>> buffer = ThreadSafeHistoryBuffer(maxlen=100)
>>> buffer.append(data)
>>> history = buffer.get_history()
>>> length = len(buffer)
Source code in pyeyesweb/data_models/thread_safe_buffer.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class ThreadSafeHistoryBuffer:
    """Thread-safe history buffer with deque backing.

    Parameters
    ----------
    maxlen : int
        Maximum size of the history buffer

    Examples
    --------
    >>> buffer = ThreadSafeHistoryBuffer(maxlen=100)
    >>> buffer.append(data)
    >>> history = buffer.get_history()
    >>> length = len(buffer)
    """

    def __init__(self, maxlen: int):
        """Initialize thread-safe history buffer.

        Parameters
        ----------
        maxlen : int
            Maximum size of the buffer
        """
        self._history = deque(maxlen=maxlen)
        self._lock = threading.RLock()

    def append(self, item: Any) -> None:
        """Thread-safely append item to history.

        Parameters
        ----------
        item : Any
            Item to append to the history buffer
        """
        with self._lock:
            self._history.append(item)

    def get_history(self) -> List[Any]:
        """Get a thread-safe copy of the history.

        Returns
        -------
        list
            Copy of the current history
        """
        with self._lock:
            return list(self._history)

    def get_array(self) -> np.ndarray:
        """Get history as numpy array (thread-safe).

        Returns
        -------
        np.ndarray
            History converted to numpy array
        """
        with self._lock:
            return np.array(list(self._history))

    def clear(self) -> None:
        """Clear the history buffer (thread-safe)."""
        with self._lock:
            self._history.clear()

    def __len__(self) -> int:
        """Get current size of buffer (thread-safe).

        Returns
        -------
        int
            Current number of items in buffer
        """
        with self._lock:
            return len(self._history)

    def __repr__(self) -> str:
        """String representation of buffer."""
        with self._lock:
            return f"ThreadSafeHistoryBuffer(maxlen={self._history.maxlen}, size={len(self._history)})"

__init__(maxlen)

Initialize thread-safe history buffer.

Parameters:

Name Type Description Default
maxlen int

Maximum size of the buffer

 required
Source code in pyeyesweb/data_models/thread_safe_buffer.py 
28
29
30
31
32
33
34
35
36
37
def __init__(self, maxlen: int):
    """Initialize thread-safe history buffer.

    Parameters
    ----------
    maxlen : int
        Maximum size of the buffer
    """
    self._history = deque(maxlen=maxlen)
    self._lock = threading.RLock()

__len__()

Get current size of buffer (thread-safe).

Returns:

Type Description
int

Current number of items in buffer
Source code in pyeyesweb/data_models/thread_safe_buffer.py 
77
78
79
80
81
82
83
84
85
86
def __len__(self) -> int:
    """Get current size of buffer (thread-safe).

    Returns
    -------
    int
        Current number of items in buffer
    """
    with self._lock:
        return len(self._history)

__repr__()

String representation of buffer.

Source code in pyeyesweb/data_models/thread_safe_buffer.py 
88
89
90
91
def __repr__(self) -> str:
    """String representation of buffer."""
    with self._lock:
        return f"ThreadSafeHistoryBuffer(maxlen={self._history.maxlen}, size={len(self._history)})"

append(item)

Thread-safely append item to history.

Parameters:

Name Type Description Default
item Any

Item to append to the history buffer

 required
Source code in pyeyesweb/data_models/thread_safe_buffer.py 
39
40
41
42
43
44
45
46
47
48
def append(self, item: Any) -> None:
    """Thread-safely append item to history.

    Parameters
    ----------
    item : Any
        Item to append to the history buffer
    """
    with self._lock:
        self._history.append(item)

clear()

Clear the history buffer (thread-safe).

Source code in pyeyesweb/data_models/thread_safe_buffer.py 
72
73
74
75
def clear(self) -> None:
    """Clear the history buffer (thread-safe)."""
    with self._lock:
        self._history.clear()

get_array()

Get history as numpy array (thread-safe).

Returns:

Type Description
ndarray

History converted to numpy array
Source code in pyeyesweb/data_models/thread_safe_buffer.py 
61
62
63
64
65
66
67
68
69
70
def get_array(self) -> np.ndarray:
    """Get history as numpy array (thread-safe).

    Returns
    -------
    np.ndarray
        History converted to numpy array
    """
    with self._lock:
        return np.array(list(self._history))

get_history()

Get a thread-safe copy of the history.

Returns:

Type Description
list

Copy of the current history
Source code in pyeyesweb/data_models/thread_safe_buffer.py 
50
51
52
53
54
55
56
57
58
59
def get_history(self) -> List[Any]:
    """Get a thread-safe copy of the history.

    Returns
    -------
    list
        Copy of the current history
    """
    with self._lock:
        return list(self._history)