Add NI-SCOPE EX Binary Acquisition Example

examples/niscope/README.md → examples/niscope/nicope_ex_fetch_forever/README.md

@@ -19,5 +19,5 @@ Script demonstrates NIScope waveform data getting continuously acquired.
 
 ```pwsh
 poetry install --with examples
-poetry run examples/niscope/niscope_ex_fetch_forever.py
+poetry run examples\niscope\niscope_ex_fetch_forever\niscope_ex_fetch_forever.py
 ```

examples/niscope/niscope_binary_acquisition/README.md

@@ -0,0 +1,23 @@
+Prerequisites
+===============
+Requires a Physical or Simulated Device. Refer to the [Getting Started Section](https://github.com/ni/nidaqmx-python/blob/master/README.rst) to learn how to create a simulated device. This example uses NI oscilloscopes/digitizers, which have the module numbering pattern _51xx_. One example is NI PXIe-5114.
+
+## Sample
+
+This is a nipanel example that displays an interactive Streamlit app and updates and fetches data from device.
+
+### Feature
+
+Script demonstrates NIScope waveform data getting continuously acquired and being converted to binary data.
+- Supports various data types
+
+### Required Software
+
+- Python 3.9 or later
+
+### Usage
+
+```pwsh
+poetry install --with examples
+poetry run examples\niscope\niscope_binary_acquisition\niscope_binary_acquisition.py
+```

examples/niscope/niscope_binary_acquisition/niscope_binary_acquisition.py

@@ -0,0 +1,110 @@
+"""Continuously acquires waveforms from NI-SCOPE, processes/scales them."""
+
+import time
+from pathlib import Path
+from typing import Any
+
+import hightime
+import niscope
+import numpy as np
+from niscope.enums import ArrayMeasurement
+from niscope.errors import Error
+
+import nipanel
+
+panel_script_path = Path(__file__).with_name("niscope_binary_acquisition_panel.py")
+panel = nipanel.create_streamlit_panel(panel_script_path)
+
+
+"""Example fetch data from device (Dev1)."""
+panel.set_value("is_running", False)
+panel.set_value("run_button", False)
+
+try:
+    panel.set_value("daq_error", "")
+    print(f"Panel URL: {panel.panel_url}")
+    print(f"Waiting for the 'Run' button to be pressed...")
+    print(f"(Press Ctrl + C to quit)")
+    while True:
+        panel.set_value("run_button", False)
+        while not panel.get_value("run_button", False):
+            time.sleep(0.1)
+        with niscope.Session(resource_name=panel.get_value("resource_name", "Dev1")) as session:
+            session.configure_vertical(
+                range=panel.get_value("vertical_range", 5.0),
+                coupling=niscope.VerticalCoupling.DC,
+                offset=panel.get_value("vertical_offset", 0.0),
+            )
+            session.configure_horizontal_timing(
+                min_sample_rate=panel.get_value("min_sample_rate", 200000000.0),
+                min_num_pts=1000,
+                ref_position=50.0,
+                num_records=1000,
+                enforce_realtime=True,
+            )
+
+            session.configure_trigger_immediate()
+
+            with session.initiate():
+                data_size = panel.get_value("data_size", 8)
+                wfm: np.ndarray[Any, Any]
+
+                if data_size == 8:
+                    wfm = np.ndarray(1000 * 1000, dtype=np.int8)
+
+                elif data_size == 16:
+                    wfm = np.ndarray(1000 * 1000, dtype=np.int16)
+                else:
+                    wfm = np.ndarray(1000 * 1000, dtype=np.int32)
+
+                waveforms = session.channels[panel.get_value("channel", 0)].fetch_into(
+                    relative_to=niscope.FetchRelativeTo.READ_POINTER,
+                    offset=0,
+                    timeout=hightime.timedelta(seconds=5.0),
+                    waveform=wfm,
+                )
+
+                try:
+                    panel.set_value("is_running", True)
+
+                    panel.set_value("stop_button", False)
+                    while not panel.get_value("stop_button", False):
+                        wfm_info = session.channels[0].fetch_array_measurement(
+                            array_meas_function=ArrayMeasurement.ADD_CHANNELS,
+                            num_records=1000,
+                            meas_num_samples=1000,
+                        )
+                        gain = 0
+                        offset = 0
+                        for waveform in waveforms:
+                            gain = waveform.gain
+                            offset = waveform.offset
+                            panel.set_value("gain_factor", gain)
+                            panel.set_value("offset", offset)
+
+                        for i in range(len(waveforms)):
+                            if panel.get_value("stop_button", True):
+                                break
+                            else:
+                                time.sleep(0.1)
+                                binary_data = waveforms[i].samples.tolist()
+                                panel.set_value("binary_data", waveforms[i].samples.tolist())
+
+                                samples_array = np.array(binary_data)
+                                voltage_values = samples_array * gain + offset
+                                panel.set_value("scaled_voltage_data", voltage_values.tolist())
+
+                    actual_binary_data_size = session.binary_sample_width
+                    panel.set_value("actual_binary_data_size", actual_binary_data_size)
+                except KeyboardInterrupt:
+                    pass
+                finally:
+                    panel.set_value("is_running", False)
+
+except Error as e:
+    daq_error = str(e)
+    print(daq_error)
+    panel.set_value("daq_error", daq_error)
+
+except KeyboardInterrupt:
+    pass

examples/niscope/niscope_binary_acquisition/niscope_binary_acquisition_panel.py

@@ -0,0 +1,192 @@
+"""Streamlit visualization script to display data acquired by nidaqmx_analog_input_filtering.py."""
+
+import streamlit as st
+from streamlit_echarts import st_echarts
+
+import nipanel
+
+st.set_page_config(page_title="NiSCOPE Binary", page_icon="📈", layout="wide")
+st.title("NiScope Binary Acquisition")
+panel = nipanel.get_streamlit_panel_accessor()
+
+left_col, right_col = st.columns(2)
+
+
+st.markdown(
+    """
+    <style>
+     div.stNumberInput {
+        max-width: 190px !important;
+    }
+    div.stTextInput {
+        max-width: 190px !important;
+    }
+
+    div[data-baseweb="select"] {
+        width: 190px !important; /* Adjust the width as needed */
+    }
+    </style>
+    <style>
+    iframe[title="streamlit_echarts.st_echarts"]{ height: 400px; width:100%;} 
+   </style>
+    """,
+    unsafe_allow_html=True,
+)
+
+
+with left_col:
+    with st.container(border=True):
+        if panel.get_value("is_running", False):
+            st.button(r"⏹️ Stop", key="stop_button")
+        elif not panel.get_value("is_running", False) and panel.get_value("daq_error", "") == "":
+            run_button = st.button(r"▶️ Run", key="run_button")
+        else:
+            st.error(
+                f"There was an error running the script. Fix the issue and re-run niscope_binary_acquisition.py \n\n {panel.get_value('daq_error', '')}"
+            )
+
+        st.text_input(
+            label="Resource Name",
+            value="Dev1",
+            key="resource_name",
+            disabled=panel.get_value("is_running", False),
+        )
+        st.number_input(
+            "Channel",
+            value=0,
+            step=1,
+            disabled=panel.get_value("is_running", False),
+            key="channel",
+        )
+
+        st.title("Vertical")
+        st.number_input(
+            "Vertical Range(V)",
+            value=5.0,
+            step=1.0,
+            disabled=panel.get_value("is_running", False),
+            key="vertical_range",
+        )
+        st.number_input(
+            "Vertical Offset",
+            value=0.0,
+            step=1.0,
+            disabled=panel.get_value("is_running", False),
+            key="vertical_offset",
+        )
+        st.title("Horizontal")
+        st.number_input(
+            "Min Sample Rate",
+            value=20000000.0,
+            step=1.0,
+            disabled=panel.get_value("is_running", False),
+            key="min_sample_rate",
+        )
+        st.number_input(
+            "Min Record Length",
+            value=1000,
+            step=1,
+            disabled=panel.get_value("is_running", False),
+            key="min_record_length",
+        )
+        data_size = st.selectbox(
+            "Binary Data Size",
+            options=[8, 16, 32],
+            disabled=panel.get_value("is_running", False),
+            key="data_size",
+        )
+        st.number_input(
+            "Actual Binary Data Size",
+            value=panel.get_value("actual_binary_data_size", 16),
+            step=1,
+            disabled=True,
+            key="actual_binary_data_size",
+        )
+        st.title("Scaling Calculation")
+        st.number_input(
+            "Gain Factor",
+            value=panel.get_value("gain_factor", 0.0000),
+            step=0.0001,
+            disabled=True,
+            format="%.10f",
+            key="gain_factor",
+        )
+        st.number_input(
+            "Offset",
+            value=panel.get_value("offset", 0.0000),
+            step=0.0001,
+            disabled=True,
+            key="offset",
+        )
+
+
+with right_col:
+    with st.container(border=True):
+        st.title("Binary Waveform Graph")
+        binary_data = panel.get_value("binary_data", [0])
+
+        binary_graph = {
+            "animation": False,
+            "tooltip": {"trigger": "axis"},
+            "legend": {"data": ["Amplitude (ADC Codes)"]},
+            "xAxis": {
+                "type": "category",
+                "data": list(range(len(binary_data))),
+                "name": "Samples",
+                "nameLocation": "center",
+                "nameGap": 40,
+            },
+            "yAxis": {
+                "type": "value",
+                "name": "Amplitude(ADC Codes)",
+                "nameRotate": 90,
+                "nameLocation": "center",
+                "nameGap": 40,
+            },
+            "series": [
+                {
+                    "name": "niscope data",
+                    "type": "line",
+                    "data": binary_data,
+                    "emphasis": {"focus": "series"},
+                    "smooth": True,
+                    "seriesLayoutBy": "row",
+                },
+            ],
+        }
+        st_echarts(options=binary_graph, height="400px", width="75%", key="binary_graph")
+    with st.container(border=True):
+        st.title("Scaled Voltage Graph")
+        scaled_voltage_data = panel.get_value("scaled_voltage_data", [0])
+        scaled_voltage_graph = {
+            "animation": False,
+            "tooltip": {"trigger": "axis"},
+            "legend": {"data": ["Amplitude (V)"]},
+            "xAxis": {
+                "type": "category",
+                "data": list(range(len(scaled_voltage_data))),
+                "name": "Samples",
+                "nameLocation": "center",
+                "nameGap": 40,
+            },
+            "yAxis": {
+                "type": "value",
+                "name": "Amplitude(V)",
+                "nameRotate": 90,
+                "nameLocation": "center",
+                "nameGap": 40,
+            },
+            "series": [
+                {
+                    "name": "niscope data",
+                    "type": "line",
+                    "data": scaled_voltage_data,
+                    "emphasis": {"focus": "series"},
+                    "smooth": True,
+                    "seriesLayoutBy": "row",
+                },
+            ],
+        }
+        st_echarts(
+            options=scaled_voltage_graph, height="400px", width="75%", key="scaled_voltage_graph"
+        )