import os import os.path import time import logging import matplotlib import matplotlib.gridspec as gs from matplotlib.backends.backend_wxagg import FigureCanvasWxAgg from matplotlib.figure import Figure matplotlib.use("WxAgg") matplotlib.interactive(False) import wx import numpy as np from matplotlib.axes import Axes from matplotlib.lines import Line2D from matplotlib.text import Annotation, Text import matplotlib.font_manager as font_manager from monitor.buffers import RollingBuffer import sys import pygame class ScopePanel(wx.Panel): def __init__(self, parent, id=wx.ID_ANY): super(ScopePanel, self).__init__(parent, id) self.figure = Figure() self.canvas = FigureCanvasWxAgg(self, id=wx.ID_ANY, figure=self.figure) sz = wx.BoxSizer(wx.HORIZONTAL) sz.Add(self.canvas, 1, wx.EXPAND) self.SetSizerAndFit(sz) self.scopeList = [] # event manager for clicking on the trigger Text Artist self.pickManager = self.figure.canvas.mpl_connect('pick_event', self.on_pick) def append(self, inData): # logging.debug('in ScopePanel.append(). received data %s', inData.shape) nbAxes = len(self.scopeList) inData = np.array(inData) # make sure we have a ndarray if inData.size == 0: return # empty array, nothing to do if len(inData.shape) == 1: nbLines = 1 else: nbLines = inData.shape[0] if nbAxes != nbLines: raise ValueError("ERROR in append(inData=%s): shape of inData incompatible with number of axes (%d)" % (inData.shape, nbAxes)) artists = [] for line, ax in zip(inData, self.scopeList): artists.extend(ax.append(line)) return artists def reset(self): # logging.debug('in ScopePanel.reset()') artists = [] for ax in self.scopeList: artists.extend(ax.reset()) return artists def addScopes(self, scopeList): self.scopeList.extend(scopeList) N = len(scopeList) PLOT_MARGIN = 0.05 PLOT_TOTAL_WIDTH = 1.0 PLOT_TOTAL_HEIGHT = 1.0 xLow = PLOT_MARGIN * PLOT_TOTAL_WIDTH width = PLOT_TOTAL_WIDTH - 2 * PLOT_TOTAL_WIDTH * PLOT_MARGIN height = (PLOT_TOTAL_HEIGHT - (N + 1) * PLOT_MARGIN * PLOT_TOTAL_HEIGHT) / N yLow = (1.0 - PLOT_MARGIN) * PLOT_TOTAL_HEIGHT - height for scope in scopeList: scope.set_position((xLow, yLow, width, height)) self.figure.add_axes(scope) yLow -= height + 1 * PLOT_MARGIN * PLOT_TOTAL_HEIGHT def on_pick(self, event): logging.debug('in ScopePanel.on_pick: %s | artist: %s', event, event.artist) # this is the handler for changing trigger types by double clicking on triggerMarkers if event.mouseevent.dblclick and isinstance(event.artist, TriggerMarker): event.artist.axes.triggerMode = triggerTypes.next(event.artist.axes.triggerMode) # this is the handler for switch alarm on and off if event.mouseevent.dblclick and isinstance(event.artist, AlarmLabel): if event.artist.isEnabled: logging.debug('Alarm is currently enabled, disabling...') event.artist.disableAlarm() else: logging.debug('Alarm is currently disabled, enabling...') event.artist.enableAlarm() # this is the handler for the title to mute the alarm if event.mouseevent and isinstance(event.artist, TrendPlotTitle): logging.debug('toggling mute status on %s' % event.artist) event.artist.toggleMute() class DumbScope(Axes): def __init__(self, fig, **kwargs): super(DumbScope, self).__init__(fig, (0.1, 0.1, 0.8, 0.8)) self.figure = fig self.lines = [] def reset(self): self.clear() self.lines = self.plot([], []) self.set_xlim(0, 1000) self.set_ylim(-0.01, 0.01) return self.lines def append(self, inData): n = inData.size x = np.linspace(0, n - 1, n) self.lines[0].set_data(x, inData) return self.lines class triggerTypes: AUTO = 0 RISINGTHRESHOLD = 1 FALLINGTHRESHOLD = 2 __nbTypes = 3 def __init__(self): pass @staticmethod def next(value): return (value + 1) % triggerTypes.__nbTypes @staticmethod def isValid(value): if value == triggerTypes.AUTO or \ value == triggerTypes.RISINGTHRESHOLD or \ value == triggerTypes.FALLINGTHRESHOLD: return True else: return False class TriggerScope(Axes): def __init__(self, fig, windowSize=30., dt=1., remanence=5, linecolor='b', linewidth=2., scaling=1.0, offset=0.0, title='plot', units=u'V', triggerMode=triggerTypes.AUTO, autoDefineThreshold=True, thresholdWindow=30., relThresholdLevel=0.75, thresholdLevel=0, autoscale=True, ymin=0., ymax=1.): self.figure = fig self.__dt = dt self.__windowSize = windowSize self.__maxNbPointsInLine = int(self.__windowSize / self.__dt) self._linecolor = linecolor self._linewidth = linewidth self._remanence = remanence self._thresholdWindow = thresholdWindow self._buffer = None self.scaling = scaling self.offset = offset self._scopeTitle = title self._scopeUnits = units self._triggerMode = triggerMode self.autoDefineThreshold = autoDefineThreshold self.relThresholdLevel = relThresholdLevel self._thresholdLevel = thresholdLevel self.autoscale = autoscale super(TriggerScope, self).__init__(fig, (0.1, 0.1, 0.8, 0.8)) self.set_ylim(ymin, ymax) self.lines = [] # noinspection PyCallingNonCallable self.__triggerLabel = TriggerMarker(self) self.tick_params(labelsize='small') def _rescaleData(self, inData): # noinspection PyTypeChecker return np.array(self.offset + self.scaling * inData) def _updateLineColor(self): if len(self.lines) > 0: for i, line in enumerate(self.lines): line.set_lw(self._linewidth / 2) line.set_color(self._linecolor) line.set_alpha((i + 1) * (1.0 / len(self.lines))) self.lines[-1].set_linewidth(self._linewidth) def _doAutoDefineThreshold(self): # logging.debug("trying to determine threshold automatically") minValue = self._buffer.min() maxValue = self._buffer.max() # logging.debug("peeking into data [%f-%f]", minValue, maxValue) overallRange = (maxValue - minValue) if self._triggerMode == triggerTypes.RISINGTHRESHOLD: self.thresholdLevel = minValue + self.relThresholdLevel * overallRange elif self._triggerMode == triggerTypes.FALLINGTHRESHOLD: self.thresholdLevel = maxValue - self.relThresholdLevel * overallRange else: pass # AUTO mode, no need for threshold def _waitForTrigger(self, inData): if self._triggerMode == triggerTypes.RISINGTHRESHOLD: a, = np.where(inData > self._thresholdLevel) if len(a) > 0 and a[0] > 0 and inData[a[0] - 1] <= self._thresholdLevel: # logging.debug("Threshold crossed at index %d. returning %d points ", a[0], len(inData[a[0]:])) return inData[a[0]:] else: return [] elif self._triggerMode == triggerTypes.FALLINGTHRESHOLD: a, = np.where(inData < self._thresholdLevel) if len(a) > 0 and a[0] > 0 and inData[a[0] - 1] >= self._thresholdLevel: # logging.debug("Threshold crossed at index %d. returning %d points ", a[0], len(inData[a[0]:])) return inData[a[0]:] else: return [] else: return inData # AUTO mode def reset(self): artistsToUpdate = [] self._buffer = RollingBuffer(size=(self._thresholdWindow / self.__dt)) self.clear() self.set_xlim(0, self.__windowSize) self.add_line(Line2D([], [])) self.__triggerLabel.reset() self.add_artist(self.__triggerLabel) t = self.set_title(self._scopeTitle) l = self.text(0.0, 1.05, "(%s)" % self._scopeUnits, transform=self.transAxes, size='x-small') artistsToUpdate.extend(self.lines) artistsToUpdate.extend([t]) artistsToUpdate.extend([l]) artistsToUpdate.extend([self.__triggerLabel]) return artistsToUpdate def append(self, inData): artistsToUpdate = [] inData = self._reshapeData(inData) # make sure we have a vector (shape (n,) ) # rescale inData = self._rescaleData(inData) # store for future reference self._buffer.append(inData) if self.autoDefineThreshold and not self.triggerMode == triggerTypes.AUTO: self._doAutoDefineThreshold() self.__triggerLabel.set_y(self._thresholdLevel) artistsToUpdate.extend([self.__triggerLabel]) nbPointsToAdd = inData.size currPos = len(self.lines[-1].get_xdata()) if currPos == 0: # we don't have any data yet, wait for trigger inData = np.array(self._waitForTrigger(inData)) if inData.size == 0: return [] else: nbPointsToAdd = inData.size if currPos + nbPointsToAdd <= self.__maxNbPointsInLine: # there is enough room to add out points to the line x = np.linspace(0, (currPos + nbPointsToAdd) * self.__dt, currPos + nbPointsToAdd, endpoint=False) y = np.append(self.lines[-1].get_ydata(), inData) self.lines[-1].set_data(x, y) else: # not enough room, add what we can and create a new line with the rest nbPointsPossible = self.__maxNbPointsInLine - currPos x = np.linspace(0, self.__windowSize, self.__maxNbPointsInLine, endpoint=False) y = np.append(self.lines[-1].get_ydata(), inData[:nbPointsPossible]) self.lines[-1].set_data(x, y) # new line l = Line2D([], []) self.add_line(l) if len(self.lines) > self._remanence: self.lines.pop(0) # since we added a new line we need to: self._updateLineColor() # update line colors self.append(inData[nbPointsPossible:]) if self.autoscale: self.relim() self.autoscale_view(tight=False, scalex=False, scaley=True) # TODO: add all y-axis artists to the list of artists to update artistsToUpdate.extend(self.lines) return artistsToUpdate @staticmethod def _reshapeData(inData): # let's check on the dimensions of inData first inData = np.array(inData) # make sure we have a ndarray if inData.size == 0: return [] # empty array, nothing to do if len(inData.shape) == 0: # we got a single value inData = np.reshape(inData, (1,)) elif len(inData.shape) == 1 or (len(inData.shape) == 2 and (inData.shape[0] == 1 or inData.shape[1] == 1)): # we got a vector, or a 1d matrix inData = np.reshape(inData, (inData.size,)) else: raise ValueError("ERROR in TriggerScope.update(inData=%s): " "wrong shape for inData (vector or 1d matrix only)" % inData.shape) # at this point inData has the dimensions of a vector (n,) return inData # self.__dt @property def dt(self): return self.__dt @dt.setter def dt(self, value): if value <= 0: raise ValueError("dt value must be > 0") self.__dt = value self.__maxNbPointsInLine = int(self.__windowSize / self.__dt) self.reset() # self.__windowSize @property def windowSize(self): return self.__windowSize @windowSize.setter def windowSize(self, value): if value <= 0: raise ValueError("windowSize must be > 0") self.__windowSize = value self.__maxNbPointsInLine = int(self.__windowSize / self.__dt) self.reset() # self._linecolor @property def linecolor(self): return self._linecolor @linecolor.setter def linecolor(self, value): self._linecolor = value self._updateLineColor() # self._linewidth @property def linewidth(self): return self._linewidth @linewidth.setter def linewidth(self, value): self._linewidth = value self._updateLineColor() # self._remanence @property def remanence(self): return self._remanence @remanence.setter def remanence(self, value): if value <= 1: logging.warning("remanence value must be >= 1. Remanence set to 1") value = 1 self._remanence = value self.reset() # self._bufferSize @property def bufferSize(self): return self._thresholdWindow @bufferSize.setter def bufferSize(self, value): self._thresholdWindow = value self.reset() # self._triggerMode @property def triggerMode(self): return self._triggerMode @triggerMode.setter def triggerMode(self, value): if not triggerTypes.isValid(self._triggerMode): raise ValueError("triggerMode must be either AUTO, RISINGTHRESHOLD or FALLINGTHRESHOLD") self._triggerMode = value self.reset() # self._thresholdLevel @property def thresholdLevel(self): return self._thresholdLevel @thresholdLevel.setter def thresholdLevel(self, value): # logging.debug('in TriggerPlot.setThresholdLevel(%s) [%s]', value, self) self._thresholdLevel = value class TriggerMarker(Annotation): def __init__(self, ax): """ TriggerMarker is a variation of Annotation displaying a [T] symbol or a [@] at the position of the trigger. Double clicking on the trigger marker alternates between the various modes of triggering """ # TODO: moving the marker should set the trigger level (and provide a way to revert to automatic trigger) super(TriggerMarker, self).__init__("", (0., 0.)) self.__FONTSIZE = 'x-small' self.__PICKER_RANGE = 10 self.__CHAR_AUTO = '@' self.__FGCOLOR_AUTO = 'blue' self.__BGCOLOR_AUTO = 'grey' self.__HA_AUTO = 'right' self.__VA_AUTO = 'bottom' self.__ROTATION_AUTO = 0. self.__CHAR_RISING = 'T' self.__FGCOLOR_RISING = 'white' self.__BGCOLOR_RISING = 'black' self.__HA_RISING = 'right' self.__VA_RISING = 'center' self.__ROTATION_RISING = 0. self.__CHAR_FALLING = 'T' self.__FGCOLOR_FALLING = 'white' self.__BGCOLOR_FALLING = 'black' self.__HA_FALLING = 'right' self.__VA_FALLING = 'center' self.__ROTATION_FALLING = 180. self.__CHAR_OUTRANGE = '#' self.__FGCOLOR_OUTRANGE = 'red' self.__BGCOLOR_OUTRANGE = 'grey' self.__HA_OUTRANGE = 'right' self.__VA_OUTRANGE = 'center' self.__ROTATION_OUTRANGE = 0. self.axes = ax self.__wasOutRange = False # adjust default parameters self.set_text(self.__CHAR_AUTO) self.set_fontsize(self.__FONTSIZE) self.set_clip_on(False) self.set_picker(self.__PICKER_RANGE) self.anncoords = 'data' self.xyann = (0, 0) def reset(self): xPos, _ = self.axes.get_xlim() if self.axes.triggerMode == triggerTypes.AUTO: self.anncoords = 'axes fraction' self.xyann = (0., 0.5) self.set_text(self.__CHAR_AUTO) self.set_backgroundcolor(self.__BGCOLOR_AUTO) self.set_color(self.__FGCOLOR_AUTO) self.set_rotation(self.__ROTATION_AUTO) self.set_verticalalignment(self.__VA_AUTO) self.set_horizontalalignment(self.__HA_AUTO) elif self.axes.triggerMode == triggerTypes.FALLINGTHRESHOLD: self.anncoords = 'data' self.xyann = (xPos, self.axes.thresholdLevel) self.set_text(self.__CHAR_FALLING) self.set_backgroundcolor(self.__BGCOLOR_FALLING) self.set_color(self.__FGCOLOR_FALLING) self.set_rotation(self.__ROTATION_FALLING) self.set_verticalalignment(self.__VA_FALLING) self.set_horizontalalignment(self.__HA_FALLING) elif self.axes.triggerMode == triggerTypes.RISINGTHRESHOLD: self.anncoords = 'data' self.xyann = (xPos, self.axes.thresholdLevel) self.set_text(self.__CHAR_RISING) self.set_backgroundcolor(self.__BGCOLOR_RISING) self.set_color(self.__FGCOLOR_RISING) self.set_rotation(self.__ROTATION_RISING) self.set_verticalalignment(self.__VA_RISING) self.set_horizontalalignment(self.__HA_RISING) def set_y(self, value): yMin, yMax = self.axes.get_ylim() x, _ = self.xyann if value > yMax or value < yMin: self.__wasOutRange = True self.anncoords = 'axes fraction' super(TriggerMarker, self).set_y(1. if value > yMax else 0.) self.set_text(self.__CHAR_OUTRANGE) self.set_backgroundcolor(self.__BGCOLOR_OUTRANGE) self.set_color(self.__FGCOLOR_OUTRANGE) self.set_rotation(self.__ROTATION_OUTRANGE) self.set_verticalalignment(self.__VA_OUTRANGE) self.set_horizontalalignment(self.__HA_OUTRANGE) else: if self.__wasOutRange: self.__wasOutRange = False self.reset() else: super(TriggerMarker, self).set_y(value) def set_position(self, pos): self.set_x(pos[0]) self.set_y(pos[1]) class AlarmLabel(Annotation): def __init__(self, trendAxes): self.__ALARMTOGGLELABEL_ON = "alarm ENABLED" self.__ALARMTOGGLELABEL_OFF = "alarm DISABLED" self.__ALARMTOGGLELABEL_COLOR_ON = "red" self.__ALARMTOGGLELABEL_COLOR_OFF = "black" self.__ALARMTOGGLELABEL_PICKER_RANGE = 10 super(AlarmLabel, self).__init__(self.__ALARMTOGGLELABEL_OFF, xy=(1., 1.), xycoords='axes fraction', xytext=(-3., -3.), textcoords='offset points', ha='right', va='top', color=self.__ALARMTOGGLELABEL_COLOR_OFF, picker=self.__ALARMTOGGLELABEL_PICKER_RANGE) self._enabled = False self.trendPlot = trendAxes @property def isEnabled(self): return self._enabled def enableAlarm(self): if not self._enabled: self._enabled = True self.trendPlot.alarmEnabled = True self.set_text(self.__ALARMTOGGLELABEL_ON) self.set_color(self.__ALARMTOGGLELABEL_COLOR_ON) def disableAlarm(self): if self._enabled: self._enabled = False self.trendPlot.alarmEnabled = False self.trendPlot.resetAlarm() self.set_text(self.__ALARMTOGGLELABEL_OFF) self.set_color(self.__ALARMTOGGLELABEL_COLOR_OFF) def reset(self): return [self] class TrendPlotTitle(Text): def __init__(self, trendAxes, text='', **kwargs): self.__ALARM_TITLECOLOR_OFF = 'black' self.__ALARM_TITLECOLOR_ON = 'red' self.__ALARM_TITLE_SUFFIX = ' (muted)' self.__PICKER_RANGE = 10 props = font_manager.FontProperties( size=matplotlib.rcParams['axes.titlesize'], weight=matplotlib.rcParams['axes.titleweight'] ) super(TrendPlotTitle, self).__init__(x=0.5, y=1.0, text=text, color=self.__ALARM_TITLECOLOR_OFF, verticalalignment=u'baseline', horizontalalignment=u'center', fontproperties=props, picker=self.__PICKER_RANGE, **kwargs) self._alarmTripped = False self._alarmMuted = False self._baseTitle = text self._trendPlot = trendAxes def tripAlarm(self): self._alarmTripped = True self.set_color(self.__ALARM_TITLECOLOR_ON) def resetAlarm(self): self._alarmTripped = False self.set_color(self.__ALARM_TITLECOLOR_OFF) self.set_text(self._baseTitle) def muteAlarm(self): if self._alarmTripped: self.set_text(self._baseTitle + self.__ALARM_TITLE_SUFFIX) self._alarmMuted = True self._trendPlot.muteAlarm() def unmuteAlarm(self): if self._alarmTripped: self.set_text(self._baseTitle) self._alarmMuted = False self._trendPlot.unmuteAlarm() def toggleMute(self): if not self._alarmMuted: self.muteAlarm() else: self.unmuteAlarm() class TrendScope(TriggerScope): # noinspection PyDefaultArgument def __init__(self, fig, windowSize=30., dt=1., remanence=5, linecolor='b', linewidth=2., scaling=1.0, offset=0.0, title='plot', units=u'V', triggerMode=triggerTypes.AUTO, autoDefineThreshold=True, thresholdWindow=30., relThresholdLevel=0.75, thresholdLevel=0, autoscale=True, ymin=0., ymax=1., trendMin=0., trendMax=1., trendWidth=1800., trendPeriod=30., trendFunction=None, trendFuncArgs={}, trendFormat='%.2f', trendUnits='', alarmEnabled=False, alarmLow=0., alarmHigh=1., alarmSoundFile=None): self.__ALARM_BGCOLOR_OFF = 'white' self.__ALARM_BGCOLOR_ON = (1., 1., 0.75) super(TrendScope, self).__init__(fig, windowSize, dt, remanence, linecolor, linewidth, scaling, offset, title, units, triggerMode, autoDefineThreshold, thresholdWindow, relThresholdLevel, thresholdLevel, autoscale, ymin, ymax) self.__trendLabel = self.text(1., 1.05, '', horizontalalignment='right', verticalalignment='baseline', size='small', color='red', transform=self.transAxes) self.__trendValueLabel = self.text(0.99, 0.05, trendFormat % 0.0, horizontalalignment='right', verticalalignment='baseline', size='large', weight='bold', color='red', transform=self.transAxes) self.title = TrendPlotTitle(self, self._scopeTitle) self.title.set_transform(self.transAxes + self.titleOffsetTrans) self.title.set_clip_box(None) self._set_artist_props(self.title) self.__trendData = RollingBuffer(size=trendWidth / trendPeriod, nLines=1, fill=np.nan) self.__trendBuffer = RollingBuffer(size=trendPeriod / self.dt, nLines=1) self._trendPeriod = trendPeriod self._trendMin = trendMin self._trendMax = trendMax self._trendWidth = trendWidth self._trendPeriod = trendPeriod self._trendFunction = trendFunction self._trendFuncArgs = trendFuncArgs self._trendOutputFormat = trendFormat self._trendUnits = trendUnits self._alarmEnabled = alarmEnabled self._alarmLow = alarmLow self._alarmHigh = alarmHigh self._alarmTripped = False self._alarmMuted = False self.yaxis.tick_left() # remove the right yaxis so that it does not clash with the trendPlot axis self.xaxis.tick_bottom() # remove the top xaxis # creates a separate axes, with the same dimensions as the original one self.__trendPlot = self.figure.add_axes(self.get_position()) self.__trendPlot.set_zorder(self.get_zorder() + 1) self.__trendLine = None # init in reset() self.__alarmLine_High = None # init in reset() self.__alarmLine_Low = None # init in reset() self.__lastCall = time.clock() self.__alarmToggleLabel = AlarmLabel(self) if self._alarmEnabled: self.__alarmToggleLabel.enableAlarm() self.reset() # call reset to adjust the properties of the axes if alarmSoundFile is not None and os.path.isfile(alarmSoundFile): self._alarmSound = pygame.mixer.Sound(alarmSoundFile) else: self._alarmSound = None def set_position(self, pos, which='both'): # we need to override this method so the new position is applied to both axes super(TrendScope, self).set_position(pos, which) self.__trendPlot.set_position(pos, which) def append(self, inData): l = super(TrendScope, self).append(inData) inData = self._reshapeData(inData) # make sure we have the right shape inData = self._rescaleData(inData) # rescale self.__trendBuffer.append(inData) # store for trend measurement now = time.clock() if now > self.__lastCall + self._trendPeriod: # logging.debug("timer in trendScope.append elapsed (last call was %f, now is %f)", # self.__lastCall, now) # logging.debug("taking measurement over buffer") if self._trendFunction is not None: out, units = self._trendFunction(self.__trendBuffer.values().flatten(), **self._trendFuncArgs) else: out = np.nan # logging.debug("measurement value returned: %.2f", out) self.__trendData.append([out]) self.__trendLine.set_ydata(self.__trendData.values()) self.__lastCall = now # reset timer for future call l.extend([self.__trendLine]) self.__trendValueLabel.set_text(self._trendOutputFormat % out) l.extend([self.__trendValueLabel]) # deal with alarm conditions if self.alarmEnabled: if out > self._alarmHigh and not np.isnan(out): if not self._alarmTripped: self.tripAlarm() elif out < self._alarmLow and not np.isnan(out): if not self._alarmTripped: self.tripAlarm() else: # value is between alarmLow and alarmHigh if self._alarmTripped: self.resetAlarm() return l @property def latestTrendData(self): return self.__trendData[:, -1] @property def trendUnits(self): return self._trendUnits @property def trendOutputFormat(self): return self._trendOutputFormat def reset(self): l = super(TrendScope, self).reset() self.title = TrendPlotTitle(self, self._scopeTitle) self.title.set_transform(self.transAxes + self.titleOffsetTrans) self.title.set_clip_box(None) self._set_artist_props(self.title) l.extend([self.title]) m = self._reset_trendPlot() l.extend(m) return l def _reset_trendPlot(self): l = [] self.__trendPlot.clear() self.__trendPlot.patch.set_visible(False) self.__trendPlot.yaxis.tick_right() self.__trendPlot.yaxis.set_label_position('right') self.__trendPlot.yaxis.set_offset_position('right') self.__trendPlot.xaxis.tick_top() self.__trendPlot.xaxis.set_label_position('top') self.__trendPlot.xaxis.set_visible(False) self.__trendPlot.set_frame_on(True) self.__trendPlot.set_ylim(self._trendMin, self._trendMax) self.__trendPlot.set_xlim(0., self._trendWidth) self.__trendPlot.tick_params(labelcolor='red', labelsize='small') x = np.linspace(0., self._trendWidth, self.__trendData.size(), endpoint=True) y = self.__trendData.values()[0, :] self.__trendLine, = self.__trendPlot.plot(x, y, '-', color='red', linewidth=2.) self.__alarmToggleLabel.reset() self.__trendPlot.add_artist(self.__alarmToggleLabel) # self.add_artist(self.__alarmToggleLabel) self.__alarmLine_High = self.__trendPlot.axhline(y=self._alarmLow, ls='--', color='red', lw=1., ) self.__alarmLine_Low = self.__trendPlot.axhline(y=self._alarmHigh, ls='--', color='red', lw=1.) l.extend([self.__alarmToggleLabel]) if self._alarmEnabled: self.__alarmLine_High.set_visible(True) self.__alarmLine_Low.set_visible(True) else: self.__alarmLine_High.set_visible(False) self.__alarmLine_Low.set_visible(False) if self._alarmTripped: self.set_axis_bgcolor(self.__ALARM_BGCOLOR_ON) self.title.tripAlarm() else: self.set_axis_bgcolor(self.__ALARM_BGCOLOR_OFF) self.title.resetAlarm() l.extend([self.__alarmLine_High]) l.extend([self.__alarmLine_Low]) if self._trendWidth > 60.: width = round(self._trendWidth / 60.) units = 'min' else: width = self._trendWidth units = 's' self.__trendLabel.set_text( '%s %s / %s %s' % (width, units, self._trendPeriod, 's')) self.add_artist(self.__trendLabel) self.add_artist(self.__trendValueLabel) l.extend([self.__trendLabel, self.__trendValueLabel]) return l @property def alarmEnabled(self): return self._alarmEnabled @alarmEnabled.setter def alarmEnabled(self, value): logging.debug("in TrendPlot.alarmEnabled: setting value to: %s" % value) self._alarmEnabled = value if self._alarmTripped and not value: self.resetAlarm() self._reset_trendPlot() @property def alarmHigh(self): return self._alarmHigh @alarmHigh.setter def alarmHigh(self, value): self._alarmHigh = value self.__alarmLine_High.set_y(value) @property def alarmLow(self): return self._alarmLow @alarmLow.setter def alarmLow(self, value): self._alarmLow = value self.__alarmLine_Low.set_y(value) def tripAlarm(self): if self.alarmEnabled: logging.debug("in TrendPlot.tripAlarm(): ALARM ALARM ALARM ALARM ALARM ALARM ALARM ALARM ALARM ALARM") self._alarmTripped = True self.set_axis_bgcolor(self.__ALARM_BGCOLOR_ON) self.title.tripAlarm() if self._alarmSound is not None: self._alarmSound.play(-1) self._alarmMuted = False def resetAlarm(self): logging.debug("in TrendPlot.resetAlarm()") self._alarmTripped = False self.set_axis_bgcolor(self.__ALARM_BGCOLOR_OFF) self.title.resetAlarm() if self._alarmSound is not None: self._alarmSound.pause() self._alarmMuted = False def muteAlarm(self): logging.debug("in TrendPlot.muteAlarm()") if self._alarmEnabled and self._alarmTripped: if self._alarmSound is not None: self._alarmSound.pause() self._alarmMuted = True def unmuteAlarm(self): logging.debug("in TrendPlot.muteAlarm()") if self._alarmEnabled and self._alarmTripped and self._alarmMuted: if self._alarmSound is not None: self._alarmSound.play(-1) self._alarmMuted = False def peakdet(v, delta, x=None): """ Converted from MATLAB script at http://billauer.co.il/peakdet.html Returns two arrays function [maxtab, mintab]=peakdet(v, delta, x) %PEAKDET Detect peaks in a vector % [MAXTAB, MINTAB] = PEAKDET(V, DELTA) finds the local % maxima and minima ("peaks") in the vector V. % MAXTAB and MINTAB consists of two columns. Column 1 % contains indices in V, and column 2 the found values. % % With [MAXTAB, MINTAB] = PEAKDET(V, DELTA, X) the indices % in MAXTAB and MINTAB are replaced with the corresponding % X-values. % % A point is considered a maximum peak if it has the maximal % value, and was preceded (to the left) by a value lower by % DELTA. % Eli Billauer, 3.4.05 (Explicitly not copyrighted). % This function is released to the public domain; Any use is allowed. :param x: :param delta: :param v: """ maxtab = [] mintab = [] if x is None: x = np.arange(len(v)) v = np.asarray(v) if len(v) != len(x): sys.exit('Input vectors v and x must have same length') if not np.isscalar(delta): sys.exit('Input argument delta must be a scalar') if delta < 0: sys.exit('Input argument delta must be positive') mn, mx = np.Inf, -np.Inf mnpos, mxpos = np.NaN, np.NaN lookformax = True for i in np.arange(len(v)): this = v[i] if this > mx: mx = this mxpos = x[i] if this < mn: mn = this mnpos = x[i] if lookformax: if this < (mx - delta): maxtab.append((mxpos, mx)) mn = this mnpos = x[i] lookformax = False else: if this > (mn + delta): mintab.append((mnpos, mn)) mx = this mxpos = x[i] lookformax = True return np.array(maxtab), np.array(mintab) # noinspection PyUnusedLocal def trend_random(inData, **kwargs): # logging.debug("in Scope.trend_random(). Received inData, kwargs=%s", kwargs) units = '' if 'units' in kwargs: units = kwargs['units'] minVal = kwargs.pop("minVal", 0.) maxVal = kwargs.pop("maxVal", 1.) return (maxVal - minVal) * np.random.random_sample() + minVal, units # noinspection PyUnusedLocal def trend_get_HR(inData, **kwargs): units = 'bpm' b = np.diff(inData) # Differentiate c = np.square(b) # square d = np.convolve(c, np.ones(10), 'same') # smooth # get RMS value to use in the peak detection algorithm rms = np.sqrt(np.mean(np.square(d))) # print 'RMS value:', rms.EKG e_max, e_min = peakdet(d, rms) # noinspection PyTypeChecker freq = 1. / np.diff(e_max[:, 0] * 1e-3) e = np.mean(freq) return e * 60., units # noinspection PyUnusedLocal def trend_get_max(inData, **kwargs): units = '' if 'units' in kwargs: units = kwargs['units'] return np.max(inData), units # noinspection PyUnusedLocal def trend_get_lastPeakValue(inData, **kwargs): units = '' if 'units' in kwargs: units = kwargs['units'] # get RMS value to use in the peak detection algorithm rms = np.sqrt(np.mean(np.square(inData))) # print 'RMS value:', rms.EKG e_max, e_min = peakdet(inData, rms) if len(e_max) > 0: ret = e_max[-1, 1] else: ret = np.nan return ret, units # noinspection PyUnusedLocal def trend_get_avgPeakValue(inData, **kwargs): units = '' if 'units' in kwargs: units = kwargs['units'] # get RMS value to use in the peak detection algorithm rms = np.sqrt(np.mean(np.square(inData))) # print 'RMS value:', rms.EKG e_max, e_min = peakdet(inData, rms) if len(e_max) > 0: ret = np.mean(e_max[:, 1]) else: ret = np.nan return ret, units def trend_get_avg(inData, **kwargs): units = '' if 'units' in kwargs: units = kwargs['units'] return np.mean(inData), units