Skip to content

Commit

Permalink
test and live plotting
Browse files Browse the repository at this point in the history
  • Loading branch information
actions-user committed Oct 30, 2024
1 parent 8b8bd8e commit cacf8af
Show file tree
Hide file tree
Showing 2 changed files with 48 additions and 1 deletion.
45 changes: 45 additions & 0 deletions src/qililab/result/live_plotting.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
import h5py

Check failure on line 1 in src/qililab/result/live_plotting.py

View workflow job for this annotation

GitHub Actions / code-quality

Ruff (CPY001)

src/qililab/result/live_plotting.py:1:1: CPY001 Missing copyright notice at top of file
import numpy as np
import matplotlib.pyplot as plt
import time

class LivePlot:

Check failure on line 6 in src/qililab/result/live_plotting.py

View workflow job for this annotation

GitHub Actions / code-quality

Ruff (I001)

src/qililab/result/live_plotting.py:1:1: I001 Import block is un-sorted or un-formatted

Check failure on line 6 in src/qililab/result/live_plotting.py

View workflow job for this annotation

GitHub Actions / code-quality

Ruff (E302)

src/qililab/result/live_plotting.py:6:1: E302 Expected 2 blank lines, found 1
"""
Live plotter for results saved in an HDF5 file with SWMR mode enabled.
Args:
path (str): Path to the HDF5 file.
refresh_interval (float): Interval in seconds to refresh the plot.
"""

def __init__(self, path: str, refresh_interval: float = 1.0):
self.path = path
self.refresh_interval = refresh_interval
self.fig, self.ax = plt.subplots()
self.line, = self.ax.plot([], [], 'b-') # Initialize an empty line

def update_plot(self, data):
"""Update the plot with new data."""
x = np.arange(data.shape[0])
self.line.set_data(x, data[:, 0]) # Assuming we're plotting the first column only
self.ax.relim()
self.ax.autoscale_view()

def run(self):
"""Continuously reads from the HDF5 file and updates the plot."""
plt.ion() # Interactive mode on
with h5py.File(self.path, "r", swmr=True) as f:
dataset = f["results"]
last_size = dataset.shape[0]

while plt.fignum_exists(self.fig.number): # Run until the plot window is closed
current_size = dataset.shape[0]
if current_size > last_size: # Check if there's new data
data = dataset[:]
self.update_plot(data)
plt.draw()
plt.pause(0.01) # Small pause to refresh the plot
last_size = current_size
time.sleep(self.refresh_interval)

Check failure on line 44 in src/qililab/result/live_plotting.py

View workflow job for this annotation

GitHub Actions / code-quality

Ruff (W293)

src/qililab/result/live_plotting.py:44:1: W293 Blank line contains whitespace
plt.ioff() # Turn interactive mode off
4 changes: 3 additions & 1 deletion tests/result/test_stream_results.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,9 @@ def create_dataset(self, _: str, data: np.ndarray):

def __exit__(self, *_):
"""mocks exit"""


def flush():
"""mocks flush"""

class TestStreamArray:
"""Test `StreamArray` functionalities."""
Expand Down

0 comments on commit cacf8af

Please sign in to comment.