k8s-configuration: updated plot of ADR-8 transitions.

import collections

import matplotlib.pyplot as plt

import matplotlib.animation as animation

from matplotlib.widgets import Button

#import struct

from tango import EventType

        self.plotTimer = 0

        self.previousTimer = 0

        self.proxy = 0

        self.thread = None

        self.isRun = True

        self.device = device

        self.anim = None

        try:

            self.proxy = tango.DeviceProxy("mid_csp/elt/subarray_01")

            print('Connected to device {}'.format(device))

            self.proxy.subscribe_event("obsState", tango.EventType.CHANGE_EVENT,

                                      self.attributes_change_evt_cb,

                                      stateless=False)

        if self.thread == None:

            self.thread = Thread(target=self.backgroundThread)

            self.thread.start()

    def getAttributeData(self, frame, lines, lineValueText, lineLabel, timeText):

        currentTimer = time.perf_counter()

        self.plotTimer = int((currentTimer - self.previousTimer) * 1000)     # the first reading will be erroneous

        self.previousTimer = currentTimer

        timeText.set_text('Plot Interval = ' + str(self.plotTimer) + 'ms')

        #timeText.set_text('Plot Interval = ' + str(self.plotTimer) + 'ms')

        self.data.append(self.obs_state)    # we get the latest data point and append it to our array

        lines.set_data(range(self.plotMaxLength), self.data)

        lineValueText.set_text('[' + lineLabel + '] = ' + str(self.obs_state))

    def backgroundThread(self):    # retrieve data

        time.sleep(0.2)  # give some buffer time for retrieving data

        while (self.isRun):

            time.sleep(0.05)  

    def attributes_change_evt_cb(self, evt):

        dev_name = evt.device.dev_name()

        if not evt.err:

                    self.obs_state = evt.attr_value.value

                    self.data.append(self.obs_state)    # we get the latest data point and append it to our array

                    self.data.append(self.obs_state)    # we get the latest data point and append it to our array

                    print("Received event on {}/{}: {}".format(dev_name, 

                                                          str(evt.attr_value.name),

                                                          str(evt.attr_value.value)))

                    #print("Received event on {}/{}: {}".format(dev_name, 

                    #                                      str(evt.attr_value.name),

                    #                                      str(evt.attr_value.value)))

            except tango.DevFailed as df:

                self.logger.error(str(df.args[0].desc))

            except Exception as except_occurred:

            for item in evt.errors:

                if item.reason == "API_EventTimeout":

                    print("API_EventTimeout")

    def close(self):

        self.isRun = False

        self.thread.join()

        print('Disconnected...')

    def on_press(self, event): 

        if event.key == 'x':

            self.anim.event_source.stop()

        if event.key == 'z':

            self.anim.event_source.start()

        if event.key == 'escape':

            exit()

def main():

    maxPlotLength = 100

    s.readAttributeStart()                                               # starts background thread

    # plotting starts below

    pltInterval = 100    # Period at which the plot animation updates [ms]

    pltInterval = 200    # Period at which the plot animation updates [ms]

    xmin = 0

    xmax = maxPlotLength

    ymin = -1 

    timeText = ax.text(0.50, 0.95, '', transform=ax.transAxes)

    lines = ax.plot([], [], label=lineLabel)[0]

    lineValueText = ax.text(0.50, 0.90, '', transform=ax.transAxes)

    anim = animation.FuncAnimation(fig, s.getAttributeData, fargs=(lines, lineValueText, lineLabel, timeText), interval=pltInterval)    # fargs has to be a tuple

    fig.canvas.mpl_connect('key_press_event', s.on_press)

    #s.anim = animation.FuncAnimation(fig, s.getAttributeData, fargs=(lines, lineValueText, lineLabel, timeText), interval=pltInterval)    # fargs has to be a tuple

    s.anim = animation.FuncAnimation(fig, s.getAttributeData, fargs=(lines, lineValueText, lineLabel, ''), interval=pltInterval)    # fargs has to be a tuple

    plt.legend(loc="upper left")

    plt.show()

    s.close()

    exit()