Priority Queues and Ring Buffers in Rust
Build a sensor alert system using ring buffers, sliding windows, and priority queues.
Introduction
30 Second Summary
Think about how a smart thermostat monitors your home. It doesn't store every reading from the past year. It keeps the most recent ones, watches for unusual patterns, and alerts you when something needs attention.
In this project, you will build a sensor alert monitoring system that processes temperature readings using three fundamental data structures: a ring buffer, a sliding window, and a priority queue. You'll implement each from scratch in both Rust and Python, then compare your versions against standard library equivalents.
What You'll Build
A sensor monitoring system that stores recent readings in fixed memory, computes rolling statistics over them, and surfaces priority-ranked alerts when thresholds are exceeded.
By the end of this project, you'll have:
- A working ring buffer implementation in both Rust and Python that stores fixed-capacity sensor readings with O(1) insert.
- A sliding window module that computes running max, min, and average over the ring buffer in both languages.
- A priority queue that ranks sensor alerts by severity, implemented from scratch and compared to stdlib equivalents (BinaryHeap in Rust and heapq in Python).
- Secret Mission: Add a monotonic deque to solve the sliding window maximum problem in O(n) time, and benchmark it against the naive O(n*k) approach.
Are there any prerequisites?
You need basic familiarity with at least one programming language (variables, functions, loops). No prior Rust or Python experience is required, but it is helpful.
Before We Start
To build a sensor alert system, you need two languages set up on your machine: Rust for zero-cost abstractions and Python for ergonomic prototyping.
In this step, you will install both toolchains and create the project folders that will hold your ring buffer, sliding window, and priority queue implementations.
In this step, get ready to:
- Install Rust via rustup (the official installer and toolchain manager).
- Verify Python 3.12 or newer is available.
- Create the project directory structure for both languages.
Install Rust
Rust is installed through a tool called rustup. This single installer gives you the compiler (rustc), the package manager (cargo), and the ability to switch between Rust versions.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
What is rustup?
Rustup is the official Rust toolchain installer and version management tool. It installs rustc (the compiler), cargo (the build tool and package manager), and rustup itself.
You can update everything later with a single command.
- Install Rust by running the install command for your platform:
macOS
- Run the official rustup installer (this command also works on Linux):
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh- The installer will ask you to choose an installation option. Press 1 and then Enter to proceed with the default installation.
- Once it finishes, reload your shell configuration by running:
source $HOME/.cargo/envWindows
- Go to https://www.rust-lang.org/tools/install in your browser.
- Download and run rustup-init.exe.
- You may be prompted to install the Visual Studio C++ Build Tools. Select option 1 (Quick install via the Visual Studio Community installer) if prompted.
- When asked to choose a Rust installation option, select option 1 (default installation).
- Close and reopen your PowerShell window so the PATH changes take effect.
- Verify Rust is installed by running:
rustc --version✔️ I see version 1.96.0 or higher
Rust is ready to go. Confirm cargo is also available.
- Verify Cargo by running:
cargo --versionYou should see a version number in the output. Move on to the next substep.
ⓧ I see an older version
You have Rust installed, but it needs to be updated to 1.96.0 or newer.
- Update to the latest stable version by running:
rustup update stable- Run rustc --version again. You should now see 1.96.0 or newer.
ⓧ Command not found
Rust is not installed yet. Follow the platform-specific install instructions above, then try rustc --version again.
Still seeing command not found?
Your terminal may not have picked up the new PATH. On macOS or Linux, run source $HOME/.cargo/env and try again. On Windows, close and reopen PowerShell.
help me debug why rustc is not found after installing rustup
Verify Python
This project requires Python 3.12 or newer. Many systems come with Python pre-installed, so check what you have before installing anything new.
- Check your Python version by running:
python3 --version✔️ I see version 3.12 or higher
Python is ready. Move on to the next substep.
ⓧ I see a version lower than 3.12
You have Python installed, but you need a newer version.
macOS
- Download and install the latest Python from python.org/downloads. These instructions also work for Linux.
- On macOS, the system python3 may still point to the old version after installation. Try running python3.12 --version or python3.14 --version to confirm the new install is accessible.
- If you use the versioned binary (e.g. python3.14), use that command instead of python3 throughout this project.
- On Linux, you can also install via your package manager:
# Ubuntu/Debian
sudo apt install python3
# Fedora
sudo dnf install python3Windows
- Download the latest Python installer from python.org/downloads.
- Run the installer. Make sure to check Add python.exe to PATH on the first screen.
- Close and reopen PowerShell, then run python3 --version again.
ⓧ Command not found
Python is not installed yet.
macOS
- Download the latest Python from python.org/downloads. These instructions also work for Linux.
- Run the installer and follow the prompts.
- On Linux you can also install via your package manager:
# Ubuntu/Debian
sudo apt install python3
# Fedora
sudo dnf install python3- After installation, run python3 --version to confirm.
Windows
- Download the latest Python installer from python.org/downloads.
- Run the installer. Make sure to check Add python.exe to PATH on the first screen.
- Close and reopen PowerShell, then run python3 --version.
Create the project structure
You will create a sensor-monitor folder on your Desktop containing two subfolders: one for Rust and one for Python.
- Move to your Desktop by running this command:
cd ~/Desktop- Create the project folder structure and initialize both projects by running these commands:
mkdir -p sensor-monitor/{rust,python}
cd sensor-monitor/rust
cargo init --name sensor_monitor
cd ../python
mkdir -p sensor_monitor
touch sensor_monitor/__init__.py
cd ..What do these commands do?
- mkdir -p sensor-monitor/{rust,python} creates the top-level folder with two subdirectories in one go.
- cargo init --name sensor_monitor creates a new Rust project inside the rust/ folder with a Cargo.toml and src/main.rs.
- mkdir -p sensor_monitor creates the Python package directory.
- touch sensor_monitor/__init__.py makes the folder a Python package by adding an empty init file.
- cd .. at the end brings you back to the sensor-monitor root folder.
- Confirm the structure is correct by running:
lsYou should see python and rust listed in the output. You are now inside the sensor-monitor root folder.
Don't see both folders?
Make sure you ran all the commands in sequence. If cargo init failed, check that Rust is installed by running cargo --version.
If the brace expansion ({rust,python}) did not work, your shell may not support it. Create the folders manually with mkdir sensor-monitor followed by mkdir sensor-monitor/rust and mkdir sensor-monitor/python.
help me fix my project folder structure for sensor-monitor
- Open your code editor. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Cursor and press Enter to open it.
- In Cursor, click File in the top menu bar.
- Click Open Folder.
- Navigate to Desktop and select the sensor-monitor folder, then click Open.
You should see the rust/ and python/ folders in the Explorer sidebar on the left.
Your development environment is ready. Next up, you will implement the ring buffer data structure in Python.
Implement the Ring Buffer in Python
Your project structure is in place. Now it's time to build the first data structure that powers our sensor monitor.
A ring buffer is a fixed-size container that stores the most recent N items. When the buffer is full and a new item arrives, it overwrites the oldest one. This makes it perfect for sensor data where you only care about the latest readings and can't let memory grow forever.
In this step, get ready to:
- Build a RingBuffer class with push, get, and iteration support.
- Compare your implementation to Python's built-in stdlib equivalent.
- Write unit tests to verify correctness.
Build the RingBuffer class
The trick behind a ring buffer is modulo arithmetic. Instead of shifting elements when the buffer is full, you wrap the write position back to the start of the array using the % operator. Two pointers track where the oldest item sits (head) and where the next write goes (tail).
- Create a new file at sensor-monitor/python/sensor_monitor/ring_buffer.py by opening the sensor_monitor folder inside sensor-monitor/python/ in Cursor's Explorer sidebar.
- Right-click the sensor_monitor folder and select New File.
- Name it ring_buffer.py.
- Add the import and the class definition with the __init__ and push methods by pasting the following code:
from collections import deque
class RingBuffer:
"""Fixed-capacity circular buffer. Overwrites oldest on overflow."""
def __init__(self, capacity: int):
if capacity <= 0:
raise ValueError("Capacity must be positive")
self.capacity = capacity
self._buffer = [None] * capacity
self._head = 0 # index of oldest element
self._tail = 0 # index of next write position
self._size = 0
def push(self, value):
"""Insert a value. If full, overwrites the oldest element."""
self._buffer[self._tail] = value
if self._size == self.capacity:
# Overwriting oldest: advance head
self._head = (self._head + 1) % self.capacity
else:
self._size += 1
self._tail = (self._tail + 1) % self.capacityWhat does this code do?
The __init__ pre-allocates a list of None slots at the requested capacity. This means the buffer never grows or shrinks in memory.
The push method writes at the _tail position, then advances _tail using modulo (% self.capacity) so it wraps around to index 0 when it reaches the end. If the buffer is already full, it also advances _head to discard the oldest value.
You won't see output yet. The next chunk adds the methods that let you read data back out of the buffer.
- Add the following methods below the push method in the same class, keeping the indentation inside the RingBuffer class:
def get(self, index: int):
"""Access element at logical index (0 = oldest)."""
if index < 0 or index >= self._size:
raise IndexError(f"Index {index} out of range (size={self._size})")
actual = (self._head + index) % self.capacity
return self._buffer[actual]
def __len__(self) -> int:
return self._size
def is_full(self) -> bool:
return self._size == self.capacity
def __iter__(self):
"""Iterate from oldest to newest."""
for i in range(self._size):
yield self._buffer[(self._head + i) % self.capacity]What does this code do?
- get(index) translates a logical index (0 = oldest) into the actual position in the underlying list using the same modulo arithmetic.
- __len__ and is_full expose the current fill state so calling code can check before reading.
- __iter__ yields elements from oldest to newest. This lets you use list(rb) or a for loop directly on a RingBuffer instance.
You won't see output yet. We'll verify everything works when we add tests in substep 2.3.
Compare with collections.deque
Python's standard library already ships a data structure with ring buffer semantics. Adding a note about it at the bottom of your file helps you remember the stdlib shortcut for future projects.
- Add the following comment block at the very bottom of ring_buffer.py, outside the class (no indentation):
# --- stdlib equivalent ---
# collections.deque with maxlen provides the same ring buffer semantics:
# window = deque(maxlen=5)
# window.append(42) # auto-evicts oldest when fullWhy build from scratch when deque exists?
A collections.deque with maxlen automatically evicts the oldest item when you append past capacity. It gives you ring buffer semantics in one line.
Building your own first teaches you the modulo arithmetic and pointer management that makes ring buffers work. When you later use deque in production, you'll understand what it's doing under the hood.
- Save ring_buffer.py.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
Here's the complete ring_buffer.py file for comparison:
from collections import deque
class RingBuffer:
"""Fixed-capacity circular buffer. Overwrites oldest on overflow."""
def __init__(self, capacity: int):
if capacity <= 0:
raise ValueError("Capacity must be positive")
self.capacity = capacity
self._buffer = [None] * capacity
self._head = 0 # index of oldest element
self._tail = 0 # index of next write position
self._size = 0
def push(self, value):
"""Insert a value. If full, overwrites the oldest element."""
self._buffer[self._tail] = value
if self._size == self.capacity:
# Overwriting oldest: advance head
self._head = (self._head + 1) % self.capacity
else:
self._size += 1
self._tail = (self._tail + 1) % self.capacity
def get(self, index: int):
"""Access element at logical index (0 = oldest)."""
if index < 0 or index >= self._size:
raise IndexError(f"Index {index} out of range (size={self._size})")
actual = (self._head + index) % self.capacity
return self._buffer[actual]
def __len__(self) -> int:
return self._size
def is_full(self) -> bool:
return self._size == self.capacity
def __iter__(self):
"""Iterate from oldest to newest."""
for i in range(self._size):
yield self._buffer[(self._head + i) % self.capacity]
# --- stdlib equivalent ---
# collections.deque with maxlen provides the same ring buffer semantics:
# window = deque(maxlen=5)
# window.append(42) # auto-evicts oldest when fullWrite unit tests
Tests confirm that modulo arithmetic and pointer management actually work. You'll cover two key scenarios: inserting when the buffer has room, and inserting when it's full (triggering the overwrite behavior).
- Create a new file at sensor-monitor/python/test_ring_buffer.py (in the python/ folder, outside the sensor_monitor package).
- Paste the following test code into the file:
from sensor_monitor.ring_buffer import RingBuffer
def test_ring_buffer_below_capacity():
rb = RingBuffer(5)
rb.push(1)
rb.push(2)
rb.push(3)
assert len(rb) == 3
assert rb.get(0) == 1
assert rb.get(2) == 3
assert list(rb) == [1, 2, 3]
def test_ring_buffer_at_capacity():
rb = RingBuffer(3)
rb.push(1)
rb.push(2)
rb.push(3)
assert rb.is_full()
rb.push(4) # overwrites 1
assert len(rb) == 3
assert list(rb) == [2, 3, 4]
rb.push(5) # overwrites 2
assert list(rb) == [3, 4, 5]
if __name__ == "__main__":
test_ring_buffer_below_capacity()
test_ring_buffer_at_capacity()
print("Ring buffer tests passed!")What do these tests check?
- test_ring_buffer_below_capacity pushes 3 items into a buffer of size 5. It verifies length, index access, and iteration order all work before the buffer hits full capacity.
- test_ring_buffer_at_capacity fills a buffer of size 3, then pushes two more items. It confirms that each new push evicts the oldest element and keeps the correct order.
- Save test_ring_buffer.py.
- Run the tests by entering these commands in your terminal:
cd python
python3 test_ring_buffer.py- You should see Ring buffer tests passed! printed in the terminal.
Seeing a ModuleNotFoundError?
- Make sure you're running the command from inside the sensor-monitor/python/ directory, not the project root. The sensor_monitor package needs to be in the current working directory for the import to resolve.
- Check that the __init__.py file exists inside sensor_monitor/. Without it, Python doesn't recognize the folder as a package.
- Verify the file is named ring_buffer.py (with an underscore, not a hyphen). Python import paths require underscores.
Help me fix my ModuleNotFoundError when running test_ring_buffer.py
✔️ Awesome, I've got everything!
Great. Make sure both files are saved before continuing.
ⓧ I'd like to double check the full code
Here's the complete test_ring_buffer.py file:
from sensor_monitor.ring_buffer import RingBuffer
def test_ring_buffer_below_capacity():
rb = RingBuffer(5)
rb.push(1)
rb.push(2)
rb.push(3)
assert len(rb) == 3
assert rb.get(0) == 1
assert rb.get(2) == 3
assert list(rb) == [1, 2, 3]
def test_ring_buffer_at_capacity():
rb = RingBuffer(3)
rb.push(1)
rb.push(2)
rb.push(3)
assert rb.is_full()
rb.push(4) # overwrites 1
assert len(rb) == 3
assert list(rb) == [2, 3, 4]
rb.push(5) # overwrites 2
assert list(rb) == [3, 4, 5]
if __name__ == "__main__":
test_ring_buffer_below_capacity()
test_ring_buffer_at_capacity()
print("Ring buffer tests passed!")Your ring buffer is tested and working. Next up, you'll port this same data structure to Rust and see how generics and ownership change the implementation.
Implement the Ring Buffer in Rust
You have a working ring buffer in Python. Now you'll translate the same algorithm into Rust, where generics and ownership rules give you compile-time guarantees that Python can only catch at runtime.
The logic is identical: modulo arithmetic for wrapping, head/tail pointers for tracking position. The difference is that Rust forces you to be explicit about types, mutability, and lifetimes. You'll also implement a custom iterator, which is how Rust collections expose their elements.
In this step, get ready to:
- Implement a generic RingBuffer<T> struct with push, get, and iter methods.
- Compare your implementation to Rust's stdlib VecDeque.
- Write unit tests using Rust's built-in test framework.
Create the ring_buffer module
Rust organizes code into modules, each living in its own file. You'll create a ring_buffer.rs file that defines the generic struct and all its methods.
- Create a new file called ring_buffer.rs inside your sensor-monitor/rust/src/ folder.
- Add the struct definition and core methods by pasting the following code:
/// A fixed-capacity circular buffer that overwrites the oldest element when full.
pub struct RingBuffer<T> {
buffer: Vec<Option<T>>,
head: usize,
tail: usize,
size: usize,
capacity: usize,
}
impl<T: Clone> RingBuffer<T> {
pub fn new(capacity: usize) -> Self {
assert!(capacity > 0, "Capacity must be positive");
RingBuffer {
buffer: vec![None; capacity],
head: 0,
tail: 0,
size: 0,
capacity,
}
}
/// Insert a value. If full, overwrites the oldest element.
pub fn push(&mut self, value: T) {
self.buffer[self.tail] = Some(value);
if self.size == self.capacity {
self.head = (self.head + 1) % self.capacity;
} else {
self.size += 1;
}
self.tail = (self.tail + 1) % self.capacity;
}
/// Access element at logical index (0 = oldest).
pub fn get(&self, index: usize) -> Option<&T> {
if index >= self.size {
return None;
}
let actual = (self.head + index) % self.capacity;
self.buffer[actual].as_ref()
}
pub fn len(&self) -> usize {
self.size
}
pub fn is_full(&self) -> bool {
self.size == self.capacity
}
}What does this code do?
This is the same ring buffer logic you wrote in Python, translated into Rust's type system.
- RingBuffer<T> is generic over any type T that implements Clone. The Clone bound is needed so vec![None; capacity] can initialize the buffer.
- Vec<Option<T>> lets each slot hold either Some(value) or None. This avoids needing a Default trait bound.
- get() returns Option<&T> instead of raising an exception. Rust handles out-of-bounds access with None rather than panicking.
- &mut self on push means the method takes a mutable reference. Only one mutable reference can exist at a time, preventing data races at compile time.
You won't see a change yet. The next chunk adds iteration support so the buffer can be used with Rust's for loops and .collect().
- Add the iterator method and struct by pasting the following code below the closing } of the impl block. First, add this iter method inside the impl block (before the closing }):
/// Iterate from oldest to newest.
pub fn iter(&self) -> RingBufferIter<T> {
RingBufferIter {
buffer: self,
index: 0,
}
}- Then add the RingBufferIter struct and its Iterator implementation after the closing } of the impl block:
pub struct RingBufferIter<'a, T> {
buffer: &'a RingBuffer<T>,
index: usize,
}
impl<'a, T: Clone> Iterator for RingBufferIter<'a, T> {
type Item = &'a T;
fn next(&mut self) -> Option<Self::Item> {
if self.index >= self.buffer.size {
return None;
}
let actual = (self.buffer.head + self.index) % self.buffer.capacity;
self.index += 1;
self.buffer.buffer[actual].as_ref()
}
}What is a custom iterator in Rust?
In Python, you used __iter__ with yield. Rust doesn't have generators, so you create a separate struct (RingBufferIter) that holds a reference to the buffer and tracks its position.
The 'a lifetime annotation tells the compiler that the iterator cannot outlive the buffer it references. This prevents dangling pointer bugs at compile time.
- Save ring_buffer.rs.
The file won't compile on its own yet. You need to declare the module in main.rs so Rust knows it exists.
- Open sensor-monitor/rust/src/main.rs in Cursor.
- Replace the entire file contents with the following:
mod ring_buffer;
fn main() {
println!("Hello, world!");
}What does mod ring_buffer do?
The mod ring_buffer; line tells Rust to look for a file called ring_buffer.rs (or ring_buffer/mod.rs) and include it as a module. Without this declaration, your ring buffer code is invisible to the rest of the project.
- Save main.rs.
- Verify the project compiles by running this command from the sensor-monitor/rust directory:
cd rust && cargo buildYou should see Compiling sensor_monitor followed by Finished with no errors.
Seeing a compile error?
- Check that ring_buffer.rs is inside src/ (not the project root). Rust expects it at src/ring_buffer.rs.
- Make sure the iter method is inside the impl<T: Clone> RingBuffer<T> block, and the RingBufferIter struct is outside it (after the closing }).
- If you see a warning about unused code, that's fine. Rust warns about pub items that nothing calls yet.
Help me fix a compile error in my Rust ring buffer
Compare with VecDeque
Rust's standard library includes VecDeque, which is documented as "a double-ended queue implemented with a growable ring buffer." It doesn't enforce a fixed capacity on its own, but you can replicate ring buffer semantics by manually popping the front when the length exceeds your desired size.
- In ring_buffer.rs, add the following comment block at the bottom of the file (below the RingBufferIter Iterator impl):
// --- stdlib equivalent ---
// std::collections::VecDeque is a growable ring buffer.
// For fixed capacity, manually pop_front when len exceeds desired size:
// use std::collections::VecDeque;
// let mut buf: VecDeque<f64> = VecDeque::with_capacity(5);
// buf.push_back(value);
// if buf.len() > 5 { buf.pop_front(); }Why build your own when VecDeque exists?
VecDeque is growable. It doesn't enforce a fixed capacity or auto-evict the oldest element. Your RingBuffer<T> guarantees a fixed memory footprint with automatic eviction. In embedded systems or real-time sensors, that guarantee matters.
Building it from scratch also teaches you how modulo arithmetic and pointer management work inside ring buffers. The VecDeque abstraction hides those details.
- Save ring_buffer.rs.
Write unit tests
Rust has a built-in test framework. You add a #[cfg(test)] module directly in the same file as your implementation. The compiler only includes this module when you run cargo test.
- Add the test module at the bottom of ring_buffer.rs (below the stdlib equivalent comments):
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_below_capacity() {
let mut rb = RingBuffer::new(5);
rb.push(1);
rb.push(2);
rb.push(3);
assert_eq!(rb.len(), 3);
assert_eq!(rb.get(0), Some(&1));
assert_eq!(rb.get(2), Some(&3));
let items: Vec<&i32> = rb.iter().collect();
assert_eq!(items, vec![&1, &2, &3]);
}
#[test]
fn test_at_capacity() {
let mut rb = RingBuffer::new(3);
rb.push(1);
rb.push(2);
rb.push(3);
assert!(rb.is_full());
rb.push(4); // overwrites 1
assert_eq!(rb.len(), 3);
let items: Vec<&i32> = rb.iter().collect();
assert_eq!(items, vec![&2, &3, &4]);
rb.push(5); // overwrites 2
let items: Vec<&i32> = rb.iter().collect();
assert_eq!(items, vec![&3, &4, &5]);
}
}What do these tests verify?
- test_below_capacity checks that inserting fewer elements than the capacity works correctly for length, index access, and iteration.
- test_at_capacity checks the overwrite behavior. After pushing a 4th element into a capacity-3 buffer, the oldest element (1) disappears and the iteration order shifts.
- use super::*; imports everything from the parent module so the tests can access RingBuffer directly.
- Save ring_buffer.rs.
- Run the tests by executing this command from sensor-monitor/rust/:
cargo testYou should see output showing both tests passing: test ring_buffer::tests::test_below_capacity ... ok and test ring_buffer::tests::test_at_capacity ... ok.
Tests failing or not found?
- If you see "0 tests run", make sure the #[cfg(test)] attribute is directly above mod tests with no blank line between them.
- If a test panics with "index out of bounds", double-check your modulo arithmetic in push and get. The % self.capacity ensures indices wrap around correctly.
- Make sure you're running cargo test from the sensor-monitor/rust/ directory (where Cargo.toml lives).
Help me debug my Rust ring buffer tests
✔️ Awesome, I've got everything!
Great. Double check you've saved both ring_buffer.rs and main.rs.
ⓧ I'd like to double check the full code
Here's the complete ring_buffer.rs file:
/// A fixed-capacity circular buffer that overwrites the oldest element when full.
pub struct RingBuffer<T> {
buffer: Vec<Option<T>>,
head: usize,
tail: usize,
size: usize,
capacity: usize,
}
impl<T: Clone> RingBuffer<T> {
pub fn new(capacity: usize) -> Self {
assert!(capacity > 0, "Capacity must be positive");
RingBuffer {
buffer: vec![None; capacity],
head: 0,
tail: 0,
size: 0,
capacity,
}
}
/// Insert a value. If full, overwrites the oldest element.
pub fn push(&mut self, value: T) {
self.buffer[self.tail] = Some(value);
if self.size == self.capacity {
self.head = (self.head + 1) % self.capacity;
} else {
self.size += 1;
}
self.tail = (self.tail + 1) % self.capacity;
}
/// Access element at logical index (0 = oldest).
pub fn get(&self, index: usize) -> Option<&T> {
if index >= self.size {
return None;
}
let actual = (self.head + index) % self.capacity;
self.buffer[actual].as_ref()
}
pub fn len(&self) -> usize {
self.size
}
pub fn is_full(&self) -> bool {
self.size == self.capacity
}
/// Iterate from oldest to newest.
pub fn iter(&self) -> RingBufferIter<T> {
RingBufferIter {
buffer: self,
index: 0,
}
}
}
pub struct RingBufferIter<'a, T> {
buffer: &'a RingBuffer<T>,
index: usize,
}
impl<'a, T: Clone> Iterator for RingBufferIter<'a, T> {
type Item = &'a T;
fn next(&mut self) -> Option<Self::Item> {
if self.index >= self.buffer.size {
return None;
}
let actual = (self.buffer.head + self.index) % self.buffer.capacity;
self.index += 1;
self.buffer.buffer[actual].as_ref()
}
}
// --- stdlib equivalent ---
// std::collections::VecDeque is a growable ring buffer.
// For fixed capacity, manually pop_front when len exceeds desired size:
// use std::collections::VecDeque;
// let mut buf: VecDeque<f64> = VecDeque::with_capacity(5);
// buf.push_back(value);
// if buf.len() > 5 { buf.pop_front(); }
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_below_capacity() {
let mut rb = RingBuffer::new(5);
rb.push(1);
rb.push(2);
rb.push(3);
assert_eq!(rb.len(), 3);
assert_eq!(rb.get(0), Some(&1));
assert_eq!(rb.get(2), Some(&3));
let items: Vec<&i32> = rb.iter().collect();
assert_eq!(items, vec![&1, &2, &3]);
}
#[test]
fn test_at_capacity() {
let mut rb = RingBuffer::new(3);
rb.push(1);
rb.push(2);
rb.push(3);
assert!(rb.is_full());
rb.push(4); // overwrites 1
assert_eq!(rb.len(), 3);
let items: Vec<&i32> = rb.iter().collect();
assert_eq!(items, vec![&2, &3, &4]);
rb.push(5); // overwrites 2
let items: Vec<&i32> = rb.iter().collect();
assert_eq!(items, vec![&3, &4, &5]);
}
}And here's main.rs:
mod ring_buffer;
fn main() {
println!("Hello, world!");
}Your Rust ring buffer is complete and tested. Next up, you'll build the sliding window module on top of it in both Python and Rust to compute rolling statistics over your sensor data.
Implement the Sliding Window in Both Languages
Your ring buffer stores the last N readings in fixed memory. But raw data alone doesn't answer the question every monitoring system needs to answer: is this sensor trending dangerously?
In this step, you'll build a SlidingWindow that wraps your RingBuffer and computes rolling statistics (average, max, min) over whatever readings the buffer currently holds.
In this step, get ready to:
- Build the Python SlidingWindow class with rolling statistics methods.
- Build the Rust SlidingWindow struct using iterator combinators.
- Test both implementations with known sequences that fill and overflow the window.
Build the Python SlidingWindow class
The sliding window is a thin layer on top of your ring buffer. It delegates storage to the RingBuffer and adds statistical methods that iterate over whatever elements the buffer currently holds.
- Create a new file at sensor-monitor/python/sensor_monitor/sliding_window.py and add the following code:
from sensor_monitor.ring_buffer import RingBuffer
class SlidingWindow:
"""Sliding window statistics over a fixed-size ring buffer."""
def __init__(self, window_size: int):
self._buffer = RingBuffer(window_size)
def push(self, value: float):
"""Add a reading to the window."""
self._buffer.push(value)
def average(self) -> float:
"""Compute the mean of the current window."""
if len(self._buffer) == 0:
return 0.0
total = sum(self._buffer)
return total / len(self._buffer)
def maximum(self) -> float:
"""Return the max of the current window."""
if len(self._buffer) == 0:
return float("-inf")
return max(self._buffer)
def minimum(self) -> float:
"""Return the min of the current window."""
if len(self._buffer) == 0:
return float("inf")
return min(self._buffer)
def __len__(self) -> int:
return len(self._buffer)What does this code do?
The SlidingWindow class delegates all storage to a RingBuffer instance. It never manages indices or capacity itself.
- push(value) passes the reading straight to the ring buffer. Once full, old readings are automatically overwritten.
- average(), maximum(), and minimum() each iterate the buffer using Python's built-in sum(), max(), and min() functions. These are O(k) operations where k is the window size.
- Each method guards against an empty buffer by returning a safe default (0.0 for average, negative/positive infinity for max/min).
- Save sliding_window.py.
You won't see a visible change yet. The next substep tests this module.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before continuing.
ⓧ I'd like to double check the full code
from sensor_monitor.ring_buffer import RingBuffer
class SlidingWindow:
"""Sliding window statistics over a fixed-size ring buffer."""
def __init__(self, window_size: int):
self._buffer = RingBuffer(window_size)
def push(self, value: float):
"""Add a reading to the window."""
self._buffer.push(value)
def average(self) -> float:
"""Compute the mean of the current window."""
if len(self._buffer) == 0:
return 0.0
total = sum(self._buffer)
return total / len(self._buffer)
def maximum(self) -> float:
"""Return the max of the current window."""
if len(self._buffer) == 0:
return float("-inf")
return max(self._buffer)
def minimum(self) -> float:
"""Return the min of the current window."""
if len(self._buffer) == 0:
return float("inf")
return min(self._buffer)
def __len__(self) -> int:
return len(self._buffer)Build the Rust SlidingWindow struct
The Rust version wraps a RingBuffer<f64> and uses iterator combinators instead of Python's built-in functions to compute statistics.
- Create a new file at sensor-monitor/rust/src/sliding_window.rs and add the following code:
use crate::ring_buffer::RingBuffer;
/// Sliding window statistics over a fixed-size ring buffer.
pub struct SlidingWindow {
buffer: RingBuffer<f64>,
}
impl SlidingWindow {
pub fn new(window_size: usize) -> Self {
SlidingWindow {
buffer: RingBuffer::new(window_size),
}
}
pub fn push(&mut self, value: f64) {
self.buffer.push(value);
}
pub fn average(&self) -> f64 {
if self.buffer.len() == 0 {
return 0.0;
}
let sum: f64 = self.buffer.iter().sum();
sum / self.buffer.len() as f64
}
pub fn maximum(&self) -> f64 {
self.buffer
.iter()
.copied()
.fold(f64::NEG_INFINITY, f64::max)
}
pub fn minimum(&self) -> f64 {
self.buffer
.iter()
.copied()
.fold(f64::INFINITY, f64::min)
}
pub fn len(&self) -> usize {
self.buffer.len()
}
}What does this code do?
The Rust SlidingWindow stores f64 readings directly. It uses Rust's iterator combinators to compute statistics without allocating temporary collections.
- .iter().sum() sums all buffered values lazily. The type annotation let sum: f64 tells the compiler which Sum implementation to use.
- .iter().copied().fold(f64::NEG_INFINITY, f64::max) finds the maximum. The .copied() converts &f64 references into f64 values so fold can compare them.
- Rust's f64 doesn't implement Ord (because of NaN), so you can't use .max() directly. The fold pattern with f64::max handles this correctly.
- Save sliding_window.rs.
Now you need to declare this module in your crate root so Rust knows about it.
- In sensor-monitor/rust/src/main.rs, find this line:
mod ring_buffer;- Add mod sliding_window; on the line below it, so the top of the file looks like this:
mod ring_buffer;
mod sliding_window;
fn main() {
println!("Hello, world!");
}- Save main.rs.
- Verify that the project compiles by running this command from the sensor-monitor/rust directory:
cd rust && cargo buildYou should see Compiling sensor_monitor followed by Finished with no errors.
Seeing a compilation error?
- If you see file not found for module, make sure your file is named exactly sliding_window.rs (with underscores) and is in the src/ directory.
- If you see a type error about Sum not being implemented, check that the let sum: f64 type annotation is present in your average() method.
Why won't my Rust sliding window compile?
Test both implementations
You'll push a known sequence into each sliding window and verify that the rolling statistics are correct as the window fills and starts overwriting.
- Replace the contents of sensor-monitor/python/test_ring_buffer.py with the following code that adds a sliding window test:
from sensor_monitor.ring_buffer import RingBuffer
from sensor_monitor.sliding_window import SlidingWindow
def test_ring_buffer_below_capacity():
rb = RingBuffer(5)
rb.push(1); rb.push(2); rb.push(3)
assert len(rb) == 3
assert rb.get(0) == 1
assert list(rb) == [1, 2, 3]
def test_ring_buffer_at_capacity():
rb = RingBuffer(3)
rb.push(1); rb.push(2); rb.push(3)
assert rb.is_full()
rb.push(4)
assert list(rb) == [2, 3, 4]
def test_sliding_window():
sw = SlidingWindow(3)
sw.push(10.0); sw.push(20.0); sw.push(30.0)
assert sw.average() == 20.0
assert sw.maximum() == 30.0
assert sw.minimum() == 10.0
sw.push(40.0)
assert sw.average() == 30.0
assert sw.maximum() == 40.0
if __name__ == "__main__":
test_ring_buffer_below_capacity()
test_ring_buffer_at_capacity()
test_sliding_window()
print("All sliding window tests passed!")What does the sliding window test verify?
- With values [10, 20, 30] in a size-3 window, the average should be 20.0, max 30.0, and min 10.0.
- After pushing 40.0, the window becomes [20, 30, 40] (10.0 is overwritten). The average jumps to 30.0 and the max becomes 40.0.
- Save test_ring_buffer.py.
- Run the Python tests from the sensor-monitor/python directory by running:
cd ../python && python3 test_ring_buffer.pyYou should see All sliding window tests passed! printed to the terminal.
Now add the Rust tests. The #[cfg(test)] module lives inside the same sliding_window.rs file, just like you did with the ring buffer.
- Open sensor-monitor/rust/src/sliding_window.rs and add the following test module at the bottom of the file:
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_sliding_window() {
let mut sw = SlidingWindow::new(3);
sw.push(10.0); sw.push(20.0); sw.push(30.0);
assert!((sw.average() - 20.0).abs() < 1e-10);
assert!((sw.maximum() - 30.0).abs() < 1e-10);
assert!((sw.minimum() - 10.0).abs() < 1e-10);
sw.push(40.0);
assert!((sw.average() - 30.0).abs() < 1e-10);
assert!((sw.maximum() - 40.0).abs() < 1e-10);
}
}Why use abs() < 1e-10 instead of ==?
Floating-point arithmetic can produce tiny rounding errors. Comparing with a small epsilon (1e-10) avoids false test failures from insignificant precision differences.
- Save sliding_window.rs.
- Run the Rust tests from the sensor-monitor/rust directory by running:
cd ../rust && cargo testYou should see all tests passing, including both the ring buffer tests from earlier and the new sliding window test.
Tests failing?
- For the Python test, make sure you're running from the sensor-monitor/python directory so the sensor_monitor package can be found.
- For the Rust test, check that use crate::ring_buffer::RingBuffer; is at the top of sliding_window.rs. Without it, the module can't find the ring buffer.
Help me debug my sliding window test failures.
✔️ Awesome, I've got everything!
Great. Make sure both test suites pass before moving on.
ⓧ I'd like to double check the full code
Here is the full sliding_window.rs file including tests:
use crate::ring_buffer::RingBuffer;
/// Sliding window statistics over a fixed-size ring buffer.
pub struct SlidingWindow {
buffer: RingBuffer<f64>,
}
impl SlidingWindow {
pub fn new(window_size: usize) -> Self {
SlidingWindow {
buffer: RingBuffer::new(window_size),
}
}
pub fn push(&mut self, value: f64) {
self.buffer.push(value);
}
pub fn average(&self) -> f64 {
if self.buffer.len() == 0 {
return 0.0;
}
let sum: f64 = self.buffer.iter().sum();
sum / self.buffer.len() as f64
}
pub fn maximum(&self) -> f64 {
self.buffer
.iter()
.copied()
.fold(f64::NEG_INFINITY, f64::max)
}
pub fn minimum(&self) -> f64 {
self.buffer
.iter()
.copied()
.fold(f64::INFINITY, f64::min)
}
pub fn len(&self) -> usize {
self.buffer.len()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_sliding_window() {
let mut sw = SlidingWindow::new(3);
sw.push(10.0); sw.push(20.0); sw.push(30.0);
assert!((sw.average() - 20.0).abs() < 1e-10);
assert!((sw.maximum() - 30.0).abs() < 1e-10);
assert!((sw.minimum() - 10.0).abs() < 1e-10);
sw.push(40.0);
assert!((sw.average() - 30.0).abs() < 1e-10);
assert!((sw.maximum() - 40.0).abs() < 1e-10);
}
}And here is the full test_ring_buffer.py file:
from sensor_monitor.ring_buffer import RingBuffer
from sensor_monitor.sliding_window import SlidingWindow
def test_ring_buffer_below_capacity():
rb = RingBuffer(5)
rb.push(1); rb.push(2); rb.push(3)
assert len(rb) == 3
assert rb.get(0) == 1
assert list(rb) == [1, 2, 3]
def test_ring_buffer_at_capacity():
rb = RingBuffer(3)
rb.push(1); rb.push(2); rb.push(3)
assert rb.is_full()
rb.push(4)
assert list(rb) == [2, 3, 4]
def test_sliding_window():
sw = SlidingWindow(3)
sw.push(10.0); sw.push(20.0); sw.push(30.0)
assert sw.average() == 20.0
assert sw.maximum() == 30.0
assert sw.minimum() == 10.0
sw.push(40.0)
assert sw.average() == 30.0
assert sw.maximum() == 40.0
if __name__ == "__main__":
test_ring_buffer_below_capacity()
test_ring_buffer_at_capacity()
test_sliding_window()
print("All sliding window tests passed!")Your sliding window computes rolling statistics over a fixed-size window of readings. Next up, you'll build a priority queue to rank alerts by severity so the most critical issues surface first.
Implement the Priority Queue in Both Languages
Your sensor system needs a way to rank alerts by severity. When multiple sensors trigger at once, a technician needs to handle the most critical issue first.
A priority queue solves this by automatically maintaining elements in priority order. Under the hood, it uses a binary heap to guarantee O(log n) insertion and removal. In this step, you'll build a min-heap from scratch in both Python and Rust, then see how the standard libraries provide the same functionality.
In this step, get ready to:
- Implement a PriorityQueue class in Python with manual sift-up and sift-down.
- Implement a MinHeap struct in Rust using generics and the Ord trait.
- Test both implementations and compare them to stdlib equivalents.
Build the Python PriorityQueue class
A min-heap stores elements in a list where the smallest element is always at index 0. The heap property means every parent is smaller than its children. Two internal operations maintain this: sift-up (after inserting) and sift-down (after removing the root).
- Create a new file at sensor-monitor/python/sensor_monitor/priority_queue.py and paste the following code:
import heapq
class PriorityQueue:
"""Min-heap priority queue implemented from scratch."""
def __init__(self):
self._heap = []
def push(self, priority: float, item):
"""Insert an item with the given priority (lower = higher priority)."""
self._heap.append((priority, item))
self._sift_up(len(self._heap) - 1)
def pop(self):
"""Remove and return (priority, item) with the lowest priority value."""
if not self._heap:
raise IndexError("Pop from empty priority queue")
self._heap[0], self._heap[-1] = self._heap[-1], self._heap[0]
result = self._heap.pop()
if self._heap:
self._sift_down(0)
return result
def peek(self):
"""View the highest-priority (lowest value) item without removing."""
if not self._heap:
raise IndexError("Peek on empty priority queue")
return self._heap[0]
def __len__(self) -> int:
return len(self._heap)
def _sift_up(self, index: int):
while index > 0:
parent = (index - 1) // 2
if self._heap[index][0] < self._heap[parent][0]:
self._heap[index], self._heap[parent] = self._heap[parent], self._heap[index]
index = parent
else:
break
def _sift_down(self, index: int):
size = len(self._heap)
while True:
smallest = index
left = 2 * index + 1
right = 2 * index + 2
if left < size and self._heap[left][0] < self._heap[smallest][0]:
smallest = left
if right < size and self._heap[right][0] < self._heap[smallest][0]:
smallest = right
if smallest != index:
self._heap[index], self._heap[smallest] = self._heap[smallest], self._heap[index]
index = smallest
else:
break
# --- stdlib equivalent ---
# import heapq
# pq = []
# heapq.heappush(pq, (priority, counter, item))
# priority, counter, item = heapq.heappop(pq)What does this code do?
The PriorityQueue class stores (priority, item) tuples in a flat list. Lower priority values mean higher urgency.
- push appends to the end, then calls _sift_up to bubble the new element toward the root until the heap property is restored. This is O(log n).
- pop swaps the root (smallest) with the last element, removes the last, then calls _sift_down to push the new root into its correct position. This is also O(log n).
- The stdlib equivalent at the bottom shows that Python's heapq module provides the same min-heap semantics using heappush and heappop on a plain list. A counter is used as a tie-breaker for equal priorities.
- Save priority_queue.py.
Seeing an import error?
The import heapq at the top is there for reference in the stdlib comment. If you get an error when testing later, make sure the file is saved inside sensor_monitor/ (with an underscore, not a hyphen).
help me debug my Python priority queue import error
Build the Rust MinHeap struct
Rust's type system lets you build a generic heap that works with any type implementing Ord. The logic is identical to the Python version, but Rust's ownership model means you swap elements in place rather than creating tuples on the fly.
- Create a new file at sensor-monitor/rust/src/priority_queue.rs and paste the following code:
/// A min-heap priority queue.
/// Elements must implement Ord. The smallest element is at the root.
pub struct MinHeap<T: Ord> {
heap: Vec<T>,
}
impl<T: Ord> MinHeap<T> {
pub fn new() -> Self {
MinHeap { heap: Vec::new() }
}
pub fn push(&mut self, value: T) {
self.heap.push(value);
self.sift_up(self.heap.len() - 1);
}
pub fn pop(&mut self) -> Option<T> {
if self.heap.is_empty() {
return None;
}
let last = self.heap.len() - 1;
self.heap.swap(0, last);
let result = self.heap.pop();
if !self.heap.is_empty() {
self.sift_down(0);
}
result
}
pub fn peek(&self) -> Option<&T> {
self.heap.first()
}
pub fn len(&self) -> usize {
self.heap.len()
}
pub fn is_empty(&self) -> bool {
self.heap.is_empty()
}
fn sift_up(&mut self, mut index: usize) {
while index > 0 {
let parent = (index - 1) / 2;
if self.heap[index] < self.heap[parent] {
self.heap.swap(index, parent);
index = parent;
} else {
break;
}
}
}
fn sift_down(&mut self, mut index: usize) {
let size = self.heap.len();
loop {
let mut smallest = index;
let left = 2 * index + 1;
let right = 2 * index + 2;
if left < size && self.heap[left] < self.heap[smallest] {
smallest = left;
}
if right < size && self.heap[right] < self.heap[smallest] {
smallest = right;
}
if smallest != index {
self.heap.swap(index, smallest);
index = smallest;
} else {
break;
}
}
}
}
// --- stdlib equivalent ---
// std::collections::BinaryHeap is a max-heap by default.
// For a min-heap, wrap values with std::cmp::Reverse:
// use std::collections::BinaryHeap;
// use std::cmp::Reverse;
// let mut heap = BinaryHeap::new();
// heap.push(Reverse((priority, item)));
// let Reverse((p, item)) = heap.pop().unwrap();
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_priority_queue() {
let mut pq = MinHeap::new();
pq.push(3); pq.push(1); pq.push(2);
assert_eq!(pq.peek(), Some(&1));
assert_eq!(pq.pop(), Some(1));
assert_eq!(pq.pop(), Some(2));
assert_eq!(pq.pop(), Some(3));
assert_eq!(pq.pop(), None);
}
#[test]
fn test_with_tuples() {
let mut pq: MinHeap<(i32, &str)> = MinHeap::new();
pq.push((3, "low")); pq.push((1, "high")); pq.push((2, "medium"));
assert_eq!(pq.pop(), Some((1, "high")));
assert_eq!(pq.pop(), Some((2, "medium")));
assert_eq!(pq.pop(), Some((3, "low")));
}
}What does this code do?
MinHeap<T: Ord> is a generic struct that works with any type that can be ordered. The Ord trait bound tells Rust the elements can be compared with < operators.
- sift_up and sift_down use self.heap.swap(index, parent) instead of Python's tuple-swap syntax. Rust moves values without copying thanks to its ownership model.
- The stdlib equivalent uses std::collections::BinaryHeap, which is a max-heap by default. Wrapping values with std::cmp::Reverse flips the comparison to create a min-heap.
- The test_with_tuples test demonstrates that the same struct handles (priority, label) pairs since Rust compares tuples element by element.
Now you need to register this module with the Rust compiler so it knows about the new file.
- In sensor-monitor/rust/src/main.rs, find the module declarations at the top of the file:
mod ring_buffer;
mod sliding_window;- Add mod priority_queue; above the existing declarations so the top of the file looks like this:
mod priority_queue;
mod ring_buffer;
mod sliding_window;
fn main() {
println!("Hello, world!");
}- Save both priority_queue.rs and main.rs.
Compiler error about unresolved module?
Make sure the file is named exactly priority_queue.rs (with an underscore) and lives directly inside src/. Rust expects module file names to match the mod declaration exactly.
help me fix my Rust module declaration error for priority_queue
Test both implementations
Your Python test file already covers the ring buffer and sliding window. You will add a priority queue test to the same file, then run both test suites.
- Open sensor-monitor/python/test_ring_buffer.py and replace its entire contents with the following updated test file:
from sensor_monitor.ring_buffer import RingBuffer
from sensor_monitor.priority_queue import PriorityQueue
from sensor_monitor.sliding_window import SlidingWindow
def test_ring_buffer_below_capacity():
rb = RingBuffer(5)
rb.push(1); rb.push(2); rb.push(3)
assert len(rb) == 3
assert rb.get(0) == 1
assert rb.get(2) == 3
assert list(rb) == [1, 2, 3]
def test_ring_buffer_at_capacity():
rb = RingBuffer(3)
rb.push(1); rb.push(2); rb.push(3)
assert rb.is_full()
rb.push(4)
assert len(rb) == 3
assert list(rb) == [2, 3, 4]
rb.push(5)
assert list(rb) == [3, 4, 5]
def test_sliding_window():
sw = SlidingWindow(3)
sw.push(10.0); sw.push(20.0); sw.push(30.0)
assert sw.average() == 20.0
assert sw.maximum() == 30.0
assert sw.minimum() == 10.0
sw.push(40.0)
assert sw.average() == 30.0
assert sw.maximum() == 40.0
def test_priority_queue():
pq = PriorityQueue()
pq.push(3, "low")
pq.push(1, "high")
pq.push(2, "medium")
assert pq.peek() == (1, "high")
assert pq.pop() == (1, "high")
assert pq.pop() == (2, "medium")
assert pq.pop() == (3, "low")
if __name__ == "__main__":
test_ring_buffer_below_capacity()
test_ring_buffer_at_capacity()
test_sliding_window()
test_priority_queue()
print("All tests passed!")What does the priority queue test verify?
The test inserts three items with priorities 3, 1, and 2. It verifies that peek returns the lowest priority (1, "high") without removing it. Successive pop calls return items in ascending priority order: 1, then 2, then 3.
- Run the Python tests by running this command:
cd sensor-monitor/python && python3 test_ring_buffer.py- You should see All tests passed! printed in your terminal.
- Run the Rust tests by running this command:
cd sensor-monitor/rust && cargo test- You should see all tests passing, including the new priority_queue::tests::test_priority_queue and priority_queue::tests::test_with_tuples tests.
Python test says ModuleNotFoundError?
Make sure you run the command from inside the sensor-monitor/python directory. Python needs to find the sensor_monitor package folder (with its __init__.py) relative to where you execute the script.
help me fix ModuleNotFoundError when running my Python tests
✔️ Awesome, I've got everything!
Great. Double check you have saved all your files before moving on.
ⓧ I'd like to double check the full code
Here is the full code for each file you created or modified in this step.
priority_queue.py:
import heapq
class PriorityQueue:
"""Min-heap priority queue implemented from scratch."""
def __init__(self):
self._heap = []
def push(self, priority: float, item):
"""Insert an item with the given priority (lower = higher priority)."""
self._heap.append((priority, item))
self._sift_up(len(self._heap) - 1)
def pop(self):
"""Remove and return (priority, item) with the lowest priority value."""
if not self._heap:
raise IndexError("Pop from empty priority queue")
self._heap[0], self._heap[-1] = self._heap[-1], self._heap[0]
result = self._heap.pop()
if self._heap:
self._sift_down(0)
return result
def peek(self):
"""View the highest-priority (lowest value) item without removing."""
if not self._heap:
raise IndexError("Peek on empty priority queue")
return self._heap[0]
def __len__(self) -> int:
return len(self._heap)
def _sift_up(self, index: int):
while index > 0:
parent = (index - 1) // 2
if self._heap[index][0] < self._heap[parent][0]:
self._heap[index], self._heap[parent] = self._heap[parent], self._heap[index]
index = parent
else:
break
def _sift_down(self, index: int):
size = len(self._heap)
while True:
smallest = index
left = 2 * index + 1
right = 2 * index + 2
if left < size and self._heap[left][0] < self._heap[smallest][0]:
smallest = left
if right < size and self._heap[right][0] < self._heap[smallest][0]:
smallest = right
if smallest != index:
self._heap[index], self._heap[smallest] = self._heap[smallest], self._heap[index]
index = smallest
else:
break
# --- stdlib equivalent ---
# import heapq
# pq = []
# heapq.heappush(pq, (priority, counter, item))
# priority, counter, item = heapq.heappop(pq)priority_queue.rs:
/// A min-heap priority queue.
/// Elements must implement Ord. The smallest element is at the root.
pub struct MinHeap<T: Ord> {
heap: Vec<T>,
}
impl<T: Ord> MinHeap<T> {
pub fn new() -> Self {
MinHeap { heap: Vec::new() }
}
pub fn push(&mut self, value: T) {
self.heap.push(value);
self.sift_up(self.heap.len() - 1);
}
pub fn pop(&mut self) -> Option<T> {
if self.heap.is_empty() {
return None;
}
let last = self.heap.len() - 1;
self.heap.swap(0, last);
let result = self.heap.pop();
if !self.heap.is_empty() {
self.sift_down(0);
}
result
}
pub fn peek(&self) -> Option<&T> {
self.heap.first()
}
pub fn len(&self) -> usize {
self.heap.len()
}
pub fn is_empty(&self) -> bool {
self.heap.is_empty()
}
fn sift_up(&mut self, mut index: usize) {
while index > 0 {
let parent = (index - 1) / 2;
if self.heap[index] < self.heap[parent] {
self.heap.swap(index, parent);
index = parent;
} else {
break;
}
}
}
fn sift_down(&mut self, mut index: usize) {
let size = self.heap.len();
loop {
let mut smallest = index;
let left = 2 * index + 1;
let right = 2 * index + 2;
if left < size && self.heap[left] < self.heap[smallest] {
smallest = left;
}
if right < size && self.heap[right] < self.heap[smallest] {
smallest = right;
}
if smallest != index {
self.heap.swap(index, smallest);
index = smallest;
} else {
break;
}
}
}
}
// --- stdlib equivalent ---
// std::collections::BinaryHeap is a max-heap by default.
// For a min-heap, wrap values with std::cmp::Reverse:
// use std::collections::BinaryHeap;
// use std::cmp::Reverse;
// let mut heap = BinaryHeap::new();
// heap.push(Reverse((priority, item)));
// let Reverse((p, item)) = heap.pop().unwrap();
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_priority_queue() {
let mut pq = MinHeap::new();
pq.push(3); pq.push(1); pq.push(2);
assert_eq!(pq.peek(), Some(&1));
assert_eq!(pq.pop(), Some(1));
assert_eq!(pq.pop(), Some(2));
assert_eq!(pq.pop(), Some(3));
assert_eq!(pq.pop(), None);
}
#[test]
fn test_with_tuples() {
let mut pq: MinHeap<(i32, &str)> = MinHeap::new();
pq.push((3, "low")); pq.push((1, "high")); pq.push((2, "medium"));
assert_eq!(pq.pop(), Some((1, "high")));
assert_eq!(pq.pop(), Some((2, "medium")));
assert_eq!(pq.pop(), Some((3, "low")));
}
}main.rs:
mod priority_queue;
mod ring_buffer;
mod sliding_window;
fn main() {
println!("Hello, world!");
}test_ring_buffer.py:
from sensor_monitor.ring_buffer import RingBuffer
from sensor_monitor.priority_queue import PriorityQueue
from sensor_monitor.sliding_window import SlidingWindow
def test_ring_buffer_below_capacity():
rb = RingBuffer(5)
rb.push(1); rb.push(2); rb.push(3)
assert len(rb) == 3
assert rb.get(0) == 1
assert rb.get(2) == 3
assert list(rb) == [1, 2, 3]
def test_ring_buffer_at_capacity():
rb = RingBuffer(3)
rb.push(1); rb.push(2); rb.push(3)
assert rb.is_full()
rb.push(4)
assert len(rb) == 3
assert list(rb) == [2, 3, 4]
rb.push(5)
assert list(rb) == [3, 4, 5]
def test_sliding_window():
sw = SlidingWindow(3)
sw.push(10.0); sw.push(20.0); sw.push(30.0)
assert sw.average() == 20.0
assert sw.maximum() == 30.0
assert sw.minimum() == 10.0
sw.push(40.0)
assert sw.average() == 30.0
assert sw.maximum() == 40.0
def test_priority_queue():
pq = PriorityQueue()
pq.push(3, "low")
pq.push(1, "high")
pq.push(2, "medium")
assert pq.peek() == (1, "high")
assert pq.pop() == (1, "high")
assert pq.pop() == (2, "medium")
assert pq.pop() == (3, "low")
if __name__ == "__main__":
test_ring_buffer_below_capacity()
test_ring_buffer_at_capacity()
test_sliding_window()
test_priority_queue()
print("All tests passed!")You now have a working priority queue in both languages. The same algorithm powers task schedulers, event systems, and network routing. Next up, you will wire all three data structures together into the sensor alert system.
Wire It All Together
You now have three working data structures in both languages. Each one handles a specific part of the sensor monitoring problem: fixed-memory storage, rolling statistics, and alert triage.
In this step, you will write a main program in each language that simulates a stream of temperature readings. The program feeds readings through the sliding window, checks for threshold violations, and pushes alerts into the priority queue for severity-ranked output.
In this step, get ready to:
- Build the Python main program that simulates sensor readings and surfaces priority-ranked alerts.
- Build the Rust main program with the same logic using your modules.
- Run both programs and verify they produce matching output.
Create the Python main program
The main program ties your three data structures into a pipeline. Readings flow into the sliding window, the window computes rolling stats, and any reading that exceeds the alert threshold gets pushed into the priority queue ranked by severity.
- Create a new file called main.py inside sensor-monitor/python/ (at the same level as the sensor_monitor/ package folder).
- Add the imports, simulated data, and severity helper by pasting this code:
from sensor_monitor.sliding_window import SlidingWindow
from sensor_monitor.priority_queue import PriorityQueue
# Simulated temperature readings (Celsius)
READINGS = [
22.1, 23.4, 25.0, 24.8, 26.3,
28.1, 30.5, 32.0, 29.7, 27.4,
35.2, 36.8, 33.1, 31.0, 28.5,
26.0, 24.5, 23.8, 22.9, 21.5,
]
WINDOW_SIZE = 5
ALERT_THRESHOLD = 30.0
def severity_from_temp(temp: float) -> float:
"""Higher temp = lower priority value = more urgent."""
return -temp # Negate so highest temps have lowest priority valueWhat does this code do?
- READINGS simulates 20 temperature sensor samples in Celsius. Some are normal, some spike above the threshold.
- WINDOW_SIZE = 5 means rolling stats are computed over the 5 most recent readings.
- severity_from_temp negates the temperature so that the highest temperature gets the lowest numeric priority, making it pop first from the min-heap.
- Add the main() function and entry point below the code you just pasted:
def main():
window = SlidingWindow(WINDOW_SIZE)
alerts = PriorityQueue()
print(f"Sensor Monitor - Window Size: {WINDOW_SIZE}, Alert Threshold: {ALERT_THRESHOLD}C")
print("=" * 60)
for i, reading in enumerate(READINGS):
window.push(reading)
if len(window) == WINDOW_SIZE:
avg = window.average()
wmax = window.maximum()
wmin = window.minimum()
print(
f"Reading {i+1:2d}: {reading:5.1f}C | "
f"Window avg={avg:.1f} max={wmax:.1f} min={wmin:.1f}"
)
if reading > ALERT_THRESHOLD:
alerts.push(severity_from_temp(reading), f"HIGH TEMP: {reading}C at reading {i+1}")
else:
print(f"Reading {i+1:2d}: {reading:5.1f}C | (filling window...)")
print("\n" + "=" * 60)
print("ALERTS (most severe first):")
print("-" * 40)
alert_num = 1
while len(alerts) > 0:
priority, message = alerts.pop()
print(f" {alert_num}. [{-priority:.1f}C] {message}")
alert_num += 1
if alert_num == 1:
print(" No alerts generated.")
if __name__ == "__main__":
main()What does this code do?
- The loop pushes each reading into the sliding window. Once the window is full (5 readings), it prints the rolling average, max, and min.
- Any reading above ALERT_THRESHOLD (30.0C) gets pushed into the priority queue with its negated temperature as the priority value.
- After all readings are processed, the program drains the priority queue. Because the min-heap pops the most-negative value first, alerts print from highest temperature to lowest.
- Save main.py.
You won't see a change yet. You will run this program in the final substep after completing the Rust version.
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py.
ⓧ I'd like to double check the full code
from sensor_monitor.sliding_window import SlidingWindow
from sensor_monitor.priority_queue import PriorityQueue
# Simulated temperature readings (Celsius)
READINGS = [
22.1, 23.4, 25.0, 24.8, 26.3,
28.1, 30.5, 32.0, 29.7, 27.4,
35.2, 36.8, 33.1, 31.0, 28.5,
26.0, 24.5, 23.8, 22.9, 21.5,
]
WINDOW_SIZE = 5
ALERT_THRESHOLD = 30.0
def severity_from_temp(temp: float) -> float:
"""Higher temp = lower priority value = more urgent."""
return -temp # Negate so highest temps have lowest priority value
def main():
window = SlidingWindow(WINDOW_SIZE)
alerts = PriorityQueue()
print(f"Sensor Monitor - Window Size: {WINDOW_SIZE}, Alert Threshold: {ALERT_THRESHOLD}C")
print("=" * 60)
for i, reading in enumerate(READINGS):
window.push(reading)
if len(window) == WINDOW_SIZE:
avg = window.average()
wmax = window.maximum()
wmin = window.minimum()
print(
f"Reading {i+1:2d}: {reading:5.1f}C | "
f"Window avg={avg:.1f} max={wmax:.1f} min={wmin:.1f}"
)
if reading > ALERT_THRESHOLD:
alerts.push(severity_from_temp(reading), f"HIGH TEMP: {reading}C at reading {i+1}")
else:
print(f"Reading {i+1:2d}: {reading:5.1f}C | (filling window...)")
print("\n" + "=" * 60)
print("ALERTS (most severe first):")
print("-" * 40)
alert_num = 1
while len(alerts) > 0:
priority, message = alerts.pop()
print(f" {alert_num}. [{-priority:.1f}C] {message}")
alert_num += 1
if alert_num == 1:
print(" No alerts generated.")
if __name__ == "__main__":
main()Create the Rust main program
Your main.rs currently has a placeholder println!("Hello, world!") body. You will replace the entire file contents with the wiring logic that mirrors the Python version.
- Open sensor-monitor/rust/src/main.rs in Cursor.
- Select all the existing code and replace it with the following:
mod priority_queue;
mod ring_buffer;
mod sliding_window;
use priority_queue::MinHeap;
use sliding_window::SlidingWindow;
const WINDOW_SIZE: usize = 5;
const ALERT_THRESHOLD: f64 = 30.0;
fn main() {
let readings: Vec<f64> = vec![
22.1, 23.4, 25.0, 24.8, 26.3, 28.1, 30.5, 32.0, 29.7, 27.4, 35.2, 36.8, 33.1, 31.0,
28.5, 26.0, 24.5, 23.8, 22.9, 21.5,
];
let mut window = SlidingWindow::new(WINDOW_SIZE);
// Store alerts as (negative_temp_as_i64_for_ordering, message)
// We use a wrapper to get min-heap on negative temp (most severe first)
let mut alerts: MinHeap<(i64, String)> = MinHeap::new();
println!(
"Sensor Monitor - Window Size: {}, Alert Threshold: {}C",
WINDOW_SIZE, ALERT_THRESHOLD
);
println!("{}", "=".repeat(60));
for (i, &reading) in readings.iter().enumerate() {
window.push(reading);
if window.len() == WINDOW_SIZE {
let avg = window.average();
let wmax = window.maximum();
let wmin = window.minimum();
println!(
"Reading {:2}: {:5.1}C | Window avg={:.1} max={:.1} min={:.1}",
i + 1,
reading,
avg,
wmax,
wmin
);
if reading > ALERT_THRESHOLD {
// Negate for min-heap: most severe (highest temp) gets lowest value
let severity = -(reading * 10.0) as i64;
let message = format!("HIGH TEMP: {}C at reading {}", reading, i + 1);
alerts.push((severity, message));
}
} else {
println!("Reading {:2}: {:5.1}C | (filling window...)", i + 1, reading);
}
}
println!("\n{}", "=".repeat(60));
println!("ALERTS (most severe first):");
println!("{}", "-".repeat(40));
let mut alert_num = 1;
while let Some((severity, message)) = alerts.pop() {
let temp = -(severity as f64) / 10.0;
println!(" {}. [{:.1}C] {}", alert_num, temp, message);
alert_num += 1;
}
if alert_num == 1 {
println!(" No alerts generated.");
}
}What's different from the Python version?
- Rust's f64 does not implement Ord (because of NaN), so you cannot use it directly in the MinHeap. Instead, temperature is multiplied by 10 and cast to i64 for ordering.
- The alert tuple is (i64, String). Tuples in Rust implement Ord by comparing elements left to right, so the severity field drives the heap ordering.
- The while let Some(...) = alerts.pop() pattern is Rust's idiomatic way to drain a collection until it returns None.
- Save main.rs.
✔️ Awesome, I've got everything!
Great. Double check you have saved main.rs.
ⓧ I'd like to double check the full code
mod priority_queue;
mod ring_buffer;
mod sliding_window;
use priority_queue::MinHeap;
use sliding_window::SlidingWindow;
const WINDOW_SIZE: usize = 5;
const ALERT_THRESHOLD: f64 = 30.0;
fn main() {
let readings: Vec<f64> = vec![
22.1, 23.4, 25.0, 24.8, 26.3, 28.1, 30.5, 32.0, 29.7, 27.4, 35.2, 36.8, 33.1, 31.0,
28.5, 26.0, 24.5, 23.8, 22.9, 21.5,
];
let mut window = SlidingWindow::new(WINDOW_SIZE);
// Store alerts as (negative_temp_as_i64_for_ordering, message)
// We use a wrapper to get min-heap on negative temp (most severe first)
let mut alerts: MinHeap<(i64, String)> = MinHeap::new();
println!(
"Sensor Monitor - Window Size: {}, Alert Threshold: {}C",
WINDOW_SIZE, ALERT_THRESHOLD
);
println!("{}", "=".repeat(60));
for (i, &reading) in readings.iter().enumerate() {
window.push(reading);
if window.len() == WINDOW_SIZE {
let avg = window.average();
let wmax = window.maximum();
let wmin = window.minimum();
println!(
"Reading {:2}: {:5.1}C | Window avg={:.1} max={:.1} min={:.1}",
i + 1,
reading,
avg,
wmax,
wmin
);
if reading > ALERT_THRESHOLD {
// Negate for min-heap: most severe (highest temp) gets lowest value
let severity = -(reading * 10.0) as i64;
let message = format!("HIGH TEMP: {}C at reading {}", reading, i + 1);
alerts.push((severity, message));
}
} else {
println!("Reading {:2}: {:5.1}C | (filling window...)", i + 1, reading);
}
}
println!("\n{}", "=".repeat(60));
println!("ALERTS (most severe first):");
println!("{}", "-".repeat(40));
let mut alert_num = 1;
while let Some((severity, message)) = alerts.pop() {
let temp = -(severity as f64) / 10.0;
println!(" {}. [{:.1}C] {}", alert_num, temp, message);
alert_num += 1;
}
if alert_num == 1 {
println!(" No alerts generated.");
}
}Run and compare output
Both programs process the same 20 readings with the same window size and threshold. The alert ordering and rolling statistics should match.
- Run the Python program by running these commands in your terminal:
cd sensor-monitor/python && python3 main.py- You should see a table of readings with rolling window stats, followed by a ranked list of alerts. The highest temperature (36.8C) should appear first in the alerts section.
- Run the Rust program by running these commands:
cd ../rust && cargo run- You should see the same alert ordering and statistics as the Python output. Both programs print alerts from most severe (36.8C) to least severe.
What should I see?
Both programs output a header line, then 20 readings with rolling stats (the first 4 show "filling window..."), then a separator, then the alerts ranked by temperature. You should see 6 alerts total (readings 7, 8, 11, 12, 13, 14 all exceed 30.0C).
Seeing a compilation error in Rust?
- If you see an error about use of undeclared crate or module, make sure the mod priority_queue;, mod ring_buffer;, and mod sliding_window; lines are at the top of main.rs.
- If you see a type mismatch error on alerts.push, check that the type annotation is MinHeap<(i64, String)> and that you are passing a tuple, not two separate arguments.
- If the Python version gives a ModuleNotFoundError, confirm you are running python3 main.py from inside the sensor-monitor/python/ directory (where the sensor_monitor/ package folder lives).
Help me debug my sensor monitor program
Secret mission
Monotonic Deque for O(n) Sliding Window Maximum
Your sliding window's maximum() method iterates the entire window every time, making it O(k) per call. There's a technique that finds the window maximum in amortized O(1) per element using a special deque. Can you implement it in both languages and prove the speedup with a benchmark?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your project files for reference, or delete them to free up disk space.
Resources you used:
- sensor-monitor/ project folder on your Desktop (containing all Rust and Python source code, compiled binaries, and test files).
Keep everything running
No action needed. Choose this if you want to revisit the code, experiment with the data structures, or extend the project further.
- Your project files remain in the sensor-monitor/ folder on your Desktop.
- Rust and Python remain installed on your machine.
Pause - I'll come back to this later
Close your editor to free up memory. Your source files stay on your Desktop for when you return.
- Close Cursor.
- Your project files remain untouched in sensor-monitor/ on your Desktop.
Delete - I don't want to use this again
Remove all project files from your machine. This deletes your source code, compiled binaries, and test files permanently.
macOS
- Navigate to your Desktop and remove the project folder by running these commands:
cd ~/Desktop
rm -rf sensor-monitorWindows
- Navigate to your Desktop and remove the project folder by running these commands:
cd ~\Desktop
rmdir /s /q sensor-monitor- Confirm the folder is gone by running:
lsYou should no longer see sensor-monitor in the output.
Nice Work!
Nice Work!
You just built three fundamental data structures from scratch in two languages. You now understand how ring buffers, sliding windows, and priority queues work internally, not just how to call their library equivalents.
You've learned how to:
- Implement a ring buffer and sliding window from scratch in both Python and Rust for O(1) fixed-memory sensor storage and rolling statistics.
- Construct a priority queue (min-heap) with O(log n) push/pop, then compare your hand-built version against stdlib equivalents like BinaryHeap and heapq.
- Wire all three data structures into a complete sensor alert system that processes temperature readings and surfaces priority-ranked alerts in both languages.
- Implement the monotonic deque sliding window maximum algorithm in O(n) time and benchmark it against the naive O(n*k) approach.
Ready to quiz yourself?