Cristallina/plot_channels
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cleanup, relative time fix

Browse Source
This commit is contained in:
steppke_a
2024-06-07 00:24:30 +02:00
parent 653eeafd2c
commit 2a3085ec14
3 changed files with 72 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+7 -4
View File
@@ -3,15 +3,18 @@
Basic plotter for EPICS or BS-based channels with output to either the terminal console or a web interface based on NiceGUI and Plotly.
````
Usage: plot_channels.py [OPTIONS] [CHANNEL_NAMES]...
Usage: plot_channels [OPTIONS] CHANNEL_NAMES...
Plot channels from BS (beam synchronous) or PV (EPICS) sources.
Plot channel data from BS (beam synchronous) or PV (EPICS) sources. Use
TEST_CHANNEL for test data.
Options:
--acquire-interval FLOAT Interval in seconds for data collection.
--plot-interval INTEGER Interval in seconds for plot update.
--save-prefix TEXT Prefix for the save filenames.
--in-subplots BOOLEAN Whether to plot in subplots.
--in-subplots Enable plotting each channel in a subplot.
--relative-time Use relative time since start of acquisition.
--web Whether to use the local web interface as a GUI.
--help Show this message and exit.
````
````
plot_channels.py
+41 -27
View File
@@ -4,24 +4,25 @@ from typing import Tuple, Optional
from itertools import cycle, islice
import re
import time
import asyncio
import random
from collections import deque
import click
import asyncio
import click
import numpy as np
from bsread import dispatcher, source
import epics
try:
from nicegui import ui, app, context, run
from nicegui import ui, app
from plots import PlotlyPlot
except ImportError as e:
print("nicegui not available, falling back to console plot.")
print("Nicegui not available, falling back to console plot.")
print(e)
from plots import ConsolePlot, PlotlyPlot
from plots import ConsolePlot
def clean_name(fname):
@@ -79,7 +80,7 @@ class ChannelData:
Append a new data point to the channel.
Args:
x: The x value.
x: The x value (timestamp).
y: The y value.
"""
@@ -137,7 +138,9 @@ def yield_bs_data(channel_names: list[str]):
# message.data.data['SINSB0-'].timestamp # timestamp of the IOC
# message.data.data['SINSB0-'].timestamp_offset # timestamp offset (ns) of the IOC
timestamp = message.data.global_timestamp + message.data.global_timestamp_offset / 1000_000_000
timestamp = (
message.data.global_timestamp + message.data.global_timestamp_offset / 1000_000_000
)
channel_data = {"pulse_id": {"value": message.data.pulse_id, "timestamp": timestamp}}
for key, value in message.data.data.items():
@@ -165,18 +168,17 @@ def yield_pv_data(channel_names: list[str]):
def yield_test_data(channel_names: list[str]):
"""Yields test data from the given channels with timestamp."""
import random
import time
while True:
channel_data = {
channel: {"value": random.random(), "timestamp": time.time()} for channel in channel_names
channel: {"value": random.random(), "timestamp": time.time()}
for channel in channel_names
}
yield channel_data
def roundrobin(*iterables):
"Visit input iterables in a cycle until each is exhausted."
"""Visit input iterables in a cycle until each is exhausted."""
# roundrobin('ABC', 'D', 'EF') → A D E B F C
# Algorithm credited to George Sakkis, posted as recipe in itertools
iterators = map(iter, iterables)
@@ -212,7 +214,9 @@ def yield_data(channel_names: list, only_pv=False):
def collect_data(channels: dict):
"""Collects data from the given channels and yields the updated data dictionary."""
"""Collects data from the given channels and respective sources
and yields the updated data dictionary.
"""
# typical channel_data:
# {'pulse_id': {'value': 21289615168, 'timestamp': 1717427820.5366151},
@@ -227,11 +231,9 @@ def collect_data(channels: dict):
yield channels
async def runner(
channels: dict,
plot_interface,
plot_interface: ConsolePlot | PlotlyPlot,
acquire_interval=0.1,
plot_interval=1,
in_subplots=False,
@@ -279,13 +281,20 @@ async def runner(
@click.command()
@click.argument("channel_names", nargs=-1)
@click.argument("channel_names", nargs=-1, type=str, required=True)
@click.option("--acquire-interval", default=0.1, help="Interval in seconds for data collection.")
@click.option("--plot-interval", default=1, help="Interval in seconds for plot update.")
@click.option("--save-prefix", default=None, help="Prefix for the save filenames.")
@click.option("--in-subplots", default=False, type=bool, help="Whether to plot in subplots.")
@click.option("--in-subplots", is_flag=True, default=False, type=bool, help="Enable plotting each channel in a subplot.")
@click.option(
"--web", is_flag=True, default=False, type=bool, help="Whether to use the local web interface as a GUI."
"--relative-time", default=True, is_flag=True, type=bool, help="Use relative time since start of acquisition."
)
@click.option(
"--web",
is_flag=True,
default=False,
type=bool,
help="Whether to use the local web interface as a GUI.",
)
def main(
channel_names: Tuple[str, ...],
@@ -293,15 +302,18 @@ def main(
plot_interval=1,
save_prefix: Optional[str] = None,
in_subplots=False,
relative_time=True,
web=False,
):
"""Plot channels from BS (beam synchronous) or PV (EPICS) sources."""
channels = {channel: ChannelData(channel, save_file_prefix=save_prefix) for channel in channel_names}
"""Plot channel data from BS (beam synchronous) or PV (EPICS) sources.
Use TEST_CHANNEL for test data.
"""
channels = {
channel: ChannelData(channel, save_file_prefix=save_prefix) for channel in channel_names
}
start_time = time.time()
if not web: # console plot version
plot = ConsolePlot(channels, start_time=start_time, in_subplots=in_subplots)
if not web: # console plot interface
plot = ConsolePlot(channels, in_subplots=in_subplots)
asyncio.run(
runner(
channels,
@@ -309,11 +321,12 @@ def main(
acquire_interval=float(acquire_interval),
plot_interval=plot_interval,
in_subplots=in_subplots,
relative_time=relative_time,
)
)
else: # nicegui version
plot = PlotlyPlot(channels, start_time=start_time, in_subplots=in_subplots)
else: # nicegui plotly interface
plot = PlotlyPlot(channels, in_subplots=in_subplots)
async def run_async():
await runner(
@@ -322,6 +335,7 @@ def main(
acquire_interval=float(acquire_interval),
plot_interval=plot_interval,
in_subplots=in_subplots,
relative_time=relative_time,
)
app.on_startup(run_async)
plots.py
+24 -13
View File
@@ -4,27 +4,38 @@ import plotext as plt
import plotly.express as px
from plotly.subplots import make_subplots
import plotly.graph_objects as go
from nicegui import ui, app, context
from nicegui import ui, context
class AbstractPlot(ABC):
@abstractmethod
def setup_plot(self):
def __init__(
self,
channels,
in_subplots=False,
) -> None:
pass
@abstractmethod
def update_plot(self, channels, start_time=0, in_subplots=False, marker="hd"):
def setup_plot(self):
"""Set up the plot."""
pass
@abstractmethod
def update_plot(self, channels, start_time=0, **kwargs):
"""Update the plot with new data from the given channels."""
pass
class PlotlyPlot(AbstractPlot):
def __init__(self, channels, start_time=0, in_subplots=False, marker="hd"):
def __init__(
self,
channels,
start_time=0,
in_subplots=False,
) -> None:
self.channels = channels
self.start_time = start_time
self.in_subplots = in_subplots
self.marker = marker
self.plot, self.fig = self.setup_plot()
@@ -33,10 +44,10 @@ class PlotlyPlot(AbstractPlot):
num_traces = len(self.channels)
labels = [channel.name for channel in self.channels.values()]
context.client.content.classes("h-[100vh]") # 1
context.client.content.classes("h-[100vh]") # full height and width of the container
colors = px.colors.qualitative.D3
fig = make_subplots(rows=num_traces, cols=1, shared_xaxes=False, vertical_spacing=0.05)
fig = make_subplots(rows=num_traces, cols=1, shared_xaxes=True, vertical_spacing=0.05)
subfigures = [
px.line(
@@ -63,15 +74,14 @@ class PlotlyPlot(AbstractPlot):
# set xaxis title for last subplot at the bottom
fig["layout"][f"xaxis{num_traces}"]["title"] = "time (s)"
# print(fig)
plot = ui.plotly(fig).classes("w-full h-full")
plot = ui.plotly(fig).classes("w-full h-full") # full height and width inside the container
return plot, fig
def update_plot(self, channels, start_time=0, **kwargs) -> None:
for i, channel in enumerate(channels.values()):
self.fig["data"][i]["x"] = channel.x
self.fig["data"][i]["x"] = channel.x - start_time
self.fig["data"][i]["y"] = channel.y
self.plot.update_figure(self.fig)
@@ -102,6 +112,7 @@ class ConsolePlot(AbstractPlot):
if in_subplots:
plt.subplots(num_channels, 1)
plt.main().plot_size(plt.tw(), plt.th() - 1)
plt.clear_data()