Code: Select all

from ctypes import *
import matplotlib.pyplot as plt
import numpy as np
from picosdk.ps2000a import ps2000a
from picosdk.functions import assert_pico_ok
import time


# ------------------------------------------------------------------------------------------------------ PICOSCOPE 2407B
class PicoScope2407B:
    name = "PicoScope 2407B"

    def __init__(self):
        super().__init__()
        self.__default_picoscope_settings()
        self.status = {}

    def __default_picoscope_settings(self):
        channel = ps2000a.PS2000A_CHANNEL
        coupling = ps2000a.PS2000A_COUPLING
        voltage_range = ps2000a.PS2000A_RANGE
        direction = ps2000a.PS2000A_THRESHOLD_DIRECTION
        self.handle = c_int16()
        self.chA = {"channel": channel["PS2000A_CHANNEL_A"],
                    "enabled": True,
                    "coupling": coupling["PS2000A_DC"],
                    "voltage_range": voltage_range["PS2000A_5V"],
                    "analog_offset": float(0)}
        self.chB = {"channel": channel["PS2000A_CHANNEL_B"],
                    "enabled": True,
                    "coupling": coupling["PS2000A_DC"],
                    "voltage_range": voltage_range["PS2000A_5V"],
                    "analog_offset": float(0)}
        self.chC = {"channel": channel["PS2000A_CHANNEL_C"],
                    "enabled": True,
                    "coupling": coupling["PS2000A_DC"],
                    "voltage_range": voltage_range["PS2000A_5V"],
                    "analog_offset": float(0)}
        self.chD = {"channel": channel["PS2000A_CHANNEL_D"],
                    "enabled": False,
                    "coupling": coupling["PS2000A_DC"],
                    "voltage_range": voltage_range["PS2000A_5V"],
                    "analog_offset": float(0)}
        self.trigger = {"enabled": True,
                        "source": channel["PS2000A_CHANNEL_C"],
                        "threshold_adc": 0,
                        "direction": direction["PS2000A_RISING"],
                        "delay": 0,
                        "autoTrigger_ms": 0}

    def open_unit(self):
        self.status["OpenUnit"] = ps2000a.ps2000aOpenUnit(
            byref(self.handle),
            None
        )
        assert_pico_ok(self.status["OpenUnit"])

    def set_channels(self):
        channels = [self.chA, self.chB, self.chC, self.chD]
        names = ["A", "B", "C", "D"]
        for ch, name in zip(channels, names):
            self.status[f"SetCh{name}"] = ps2000a.ps2000aSetChannel(
                self.handle,
                ch["channel"],
                ch["enabled"],
                ch["coupling"],
                ch["voltage_range"],
                ch["analog_offset"]
            )
            assert_pico_ok(self.status[f"SetCh{name}"])

    def set_simple_trigger(self):
        self.status[f"SetSimpleTrigger"] = ps2000a.ps2000aSetSimpleTrigger(
            self.handle,
            self.trigger["enabled"],
            self.trigger["source"],
            self.trigger["threshold_adc"],
            self.trigger["direction"],
            self.trigger["delay"],
            self.trigger["autoTrigger_ms"]
        )
        assert_pico_ok(self.status[f"SetSimpleTrigger"])

    def get_timebase(self, samples=1000, wanted_time_interval_ns=10):
        timebase = 0
        time_interval_ns = c_float()
        oversample = c_int16(0)
        max_samples = c_int32()
        segment_index = 0
        while ps2000a.ps2000aGetTimebase2(
                self.handle,
                timebase,
                samples,
                byref(time_interval_ns),
                oversample,
                byref(max_samples),
                segment_index
            ) != 0 or time_interval_ns.value <= wanted_time_interval_ns:
            timebase += 1
        timebase -= 1
        self.status[f"GetTimebase2"] = ps2000a.ps2000aGetTimebase2(
            self.handle,
            timebase,
            samples,
            byref(time_interval_ns),
            oversample,
            byref(max_samples),
            segment_index
        )
        assert_pico_ok(self.status[f"GetTimebase2"])
        return timebase, time_interval_ns.value, max_samples.value

    def memory_segments(self, segments=10, samples=1000):
        max_samples = c_int32(samples)
        self.status[f"MemorySegments"] = ps2000a.ps2000aMemorySegments(
            self.handle,
            segments,
            byref(max_samples)
        )
        assert_pico_ok(self.status[f"MemorySegments"])
        return max_samples.value

    def set_captures(self, captures=10):
        self.status[f"SetNoOfCaptures"] = ps2000a.ps2000aSetNoOfCaptures(
            self.handle,
            captures
        )
        assert_pico_ok(self.status[f"SetNoOfCaptures"])

    def run_block(self, pre_trigger_samples=0, post_trigger_samples=1000, timebase=0, segment_index=0):
        oversample = 1
        time_indisposed_ms = c_int32()
        lpReady = None
        pParameter = None
        self.status[f"RunBlock"] = ps2000a.ps2000aRunBlock(
            self.handle,
            pre_trigger_samples,
            post_trigger_samples,
            timebase,
            oversample,
            byref(time_indisposed_ms),
            segment_index,
            lpReady,
            pParameter
        )
        assert_pico_ok(self.status[f"RunBlock"])
        return time_indisposed_ms.value

    def set_data_buffer(self, segments=10, samples=1000):
        channels = [self.chA, self.chB, self.chC, self.chD]
        mode = ps2000a.PS2000A_RATIO_MODE
        for ch in channels:
            if ch["enabled"]:
                ch["buffer"] = ((c_int16 * samples) * segments)()
                for segment_index in range(segments):
                    self.status[f"SetDataBuffer"] = ps2000a.ps2000aSetDataBuffer(
                        self.handle,
                        ch["channel"],
                        byref(ch["buffer"][:][segment_index]),
                        samples,
                        segment_index,
                        mode["PS2000A_RATIO_MODE_NONE"]
                    )
                assert_pico_ok(self.status[f"SetDataBuffer"])

    def is_ready(self):
        ready = c_int16(0)
        check = c_int16(0)
        while ready.value == check.value:
            self.status["IsReady"] = ps2000a.ps2000aIsReady(
                self.handle,
                byref(ready)
            )
        assert_pico_ok(self.status[f"IsReady"])

    def get_maxadc(self):
        value = c_int32()
        self.status[f"MaximumValue"] = ps2000a.ps2000aMaximumValue(
            self.handle,
            byref(value)
        )
        assert_pico_ok(self.status[f"MaximumValue"])
        return value.value

    def stop(self):
        self.status[f"Stop"] = ps2000a.ps2000aStop(
            self.handle
        )
        assert_pico_ok(self.status[f"Stop"])

    def close_unit(self):
        self.status[f"CloseUnit"] = ps2000a.ps2000aCloseUnit(
            self.handle
        )
        assert_pico_ok(self.status[f"CloseUnit"])

    def convert_values_adc2mV(self, maxADC=32767):
        channels = [self.chA, self.chB, self.chC, self.chD]
        channel_input_ranges = [10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 50000, 100000, 200000]
        for ch in channels:
            if ch["enabled"]:
                ch["buffer"] = (np.asarray(ch["buffer"]) / maxADC) * channel_input_ranges[ch["voltage_range"]]

    def display_data(self):
        channels = [self.chA, self.chB, self.chC, self.chD]
        for ch in channels:
            if ch["enabled"]:
                plt.figure()
                #plt.plot(np.mean(ch["buffer"], axis=0))
                plt.imshow(ch["buffer"], cmap='gray', aspect='auto')
                plt.show()

    def get_values_bulk(self, segments=10, samples=1000):
        max_samples = c_uint32(samples)
        down_sample_ratio = 0
        from_segment_index = 0
        to_segment_index = segments - 1
        down_sample_ratio_mode = ps2000a.PS2000A_RATIO_MODE["PS2000A_RATIO_MODE_NONE"]
        overflow = (c_int16 * segments)()
        self.status[f"GetValuesBulk"] = ps2000a.ps2000aGetValuesBulk(
            self.handle,
            byref(max_samples),
            from_segment_index,
            to_segment_index,
            down_sample_ratio,
            down_sample_ratio_mode,
            byref(overflow)
        )
        assert_pico_ok(self.status[f"GetValuesBulk"])
        return max_samples.value, overflow

    def get_values_trigger_time_offset_bulk(self, segments=10):
        times = (c_int64 * segments)()
        time_units = c_char()
        from_segment_index = 0
        to_segment_index = segments - 1
        self.status[f"GetValuesTriggerTimeOffsetBulk"] = ps2000a.ps2000aGetValuesTriggerTimeOffsetBulk64(
            self.handle,
            byref(times),
            byref(time_units),
            from_segment_index,
            to_segment_index
        )
        assert_pico_ok(self.status[f"GetValuesTriggerTimeOffsetBulk"])

    def rapid_block_mode(self):
        #  Block mode example like in the programming guide.
        n_segments = 3000
        max_samples = 3000

        #  Open the oscilloscope.
        self.open_unit()
        #  Select channel ranges and AC/DC coupling.
        self.set_channels()
        #  Set up single trigger
        self.set_simple_trigger()
        #  Get timebase information
        timebase, time_interval_ns, samples = self.get_timebase(samples=max_samples, wanted_time_interval_ns=10)
        #  Set the number of memory segments equal to or greater than the number of captures required.
        #  Specify the number of waveforms to capture before each run.
        self.memory_segments(segments=n_segments, samples=max_samples)
        self.set_captures(captures=n_segments)
        #  Start the oscilloscope running.
        self.run_block(pre_trigger_samples=0, post_trigger_samples=max_samples, timebase=timebase)
        #  Tell the driver where your memory buffers are. Call the function once for each channel/segment combination
        #  for which you require data.
        self.set_data_buffer(segments=n_segments, samples=max_samples)
        #  Start collecting data.
        self.is_ready()
        #  Transfer the blocks of data from the oscilloscope
        self.get_values_bulk(segments=n_segments, samples=max_samples)
        #  Get times.
        #self.get_values_trigger_time_offset_bulk(segments=n_segments)
        #  Find the max ADC count and convert values.
        maxADC = self.get_maxadc()
        self.convert_values_adc2mV(maxADC)
        #  Display data.
        self.display_data()
        #  Stop the oscilloscope.
        self.stop()
        #  Close Unit.
        self.close_unit()


if __name__ == '__main__':
    pico = PicoScope2407B()
    if pico.name == "PicoScope 2407B":
        pico.rapid_block_mode()
        print(pico.status)