#!/usr/bin/python3 import math import sys import os import re import datetime as dt import matplotlib.pyplot as plt import matplotlib.dates as mdates from collections import deque from matplotlib.animation import FuncAnimation import matplotlib.ticker as mtick from matplotlib.widgets import TextBox from matplotlib.widgets import CheckButtons from matplotlib.widgets import Button import numpy as np import simpleaudio as sa def getv(ind,val): if len(sys.argv) > ind and len(sys.argv[ind])>0: return sys.argv[ind] else: return val timeframe = int(getv(1,180)) # minutes last_time = dt.datetime.now()-dt.timedelta(minutes=timeframe+10) last_day = last_time.strftime('%d.%m.%Y') prefix = 'E:/DCS_2' if os.name == 'nt' else '/mnt/DCS/DCS_2' filename = getv(2,prefix+'/DCS2_2019_19_v3_34 '+last_day+' archiv_DCS2/'+last_day+'___magnetometr VES.txt') if filename.find(last_day) == -1: try: last_day = open(filename,'r').readline().split()[0] last_time = dt.datetime.strptime(last_day,'%d.%m.%Y') except Exception as e: print(e) lower_alarm_limit = float(getv(3,1028)) upper_alarm_limit = float(getv(4,1028.5)) title = getv(5,'M13 '+last_day) print('%s %s "%s" %s %s "%s"'%(sys.argv[0],timeframe,filename,lower_alarm_limit,upper_alarm_limit,title)) #prepare alarm sound A_freq = 440 Csh_freq = A_freq * 2 ** (4 / 12) E_freq = A_freq * 2 ** (7 / 12) # get timesteps for each sample, T is note duration in seconds sample_rate = 44100 T = 1 t = np.linspace(0, T, T * sample_rate, False) # generate sine wave notes A_note = np.sin(A_freq * t * 2 * np.pi) Csh_note = np.sin(Csh_freq * t * 2 * np.pi) E_note = np.sin(E_freq * t * 2 * np.pi) silence = np.zeros(sample_rate*T) # concatenate notes fragment = np.hstack((A_note, Csh_note, E_note, silence)) audio = np.hstack([fragment for _ in range(8)]) # normalize to 16-bit range audio *= 32767 / np.max(np.abs(audio)) # convert to 16-bit data audio = audio.astype(np.int16) play_obj = None # assuming 2 measurements per minute data_time = deque([],timeframe*2) data_field = deque([],timeframe*2) def read_from_file(): global last_time,filename if len(data_time)>0: last_time = data_time[-1] with open(filename,'r') as file: for line in file: try: a,b,c,d,e,f,g,h = line.split() dt_point = dt.datetime.strptime(a+' '+b+re.sub('.*,',',',h)[0:7],'%d.%m.%Y %H:%M:%S,%f') if dt_point > last_time: F_point = float(re.sub(',','.',f)) data_time.append(dt_point) data_field.append(F_point) except Exception as e: print(e) fig, ax = plt.subplots() fig.autofmt_xdate() plt.minorticks_on() plt.grid(True, which='both') #plt.title(title) ax.text(0.5, 1.025, title, transform=ax.transAxes, ha='center', size='large') alarm = False beep_on_alarm = True def alarm_check(): global alarm global play_obj if not alarm and (lower_alarm_limit > data_field[-1] or data_field[-1] > upper_alarm_limit): fig.set_facecolor('#FF0000') alarm = True if play_obj is not None and play_obj.is_playing(): play_obj.stop() if beep_on_alarm: play_obj = sa.play_buffer(audio, 1, 2, sample_rate) #print('enable alarm') fig.canvas.draw_idle() fig.canvas.flush_events() return if alarm and (not beep_on_alarm or (lower_alarm_limit < data_field[-1] and data_field[-1] < upper_alarm_limit)): fig.set_facecolor('#FFFFFF') alarm = False if play_obj is not None and play_obj.is_playing(): play_obj.stop() #print('disable alarm') fig.canvas.draw_idle() fig.canvas.flush_events() return if alarm and beep_on_alarm and play_obj is not None and not play_obj.is_playing(): play_obj = sa.play_buffer(audio, 1, 2, sample_rate) def plot(ax): alarm_check() ln, = ax.plot(data_time,data_field,'b.-') return ln def update_date(): global last_time,last_day,filename,title,ax today = dt.datetime.now().strftime('%d.%m.%Y') if today != last_day: new_time = dt.datetime.strptime(last_day,'%d.%m.%Y')+dt.timedelta(days=1) new_day = new_time.strftime('%d.%m.%Y') filename = filename.replace(last_day,new_day) if last_day in title: ax.text(0.5, 1.025, title, color='white', bbox={'facecolor': 'white', 'edgecolor': 'white', 'pad': 2}, transform=ax.transAxes, ha='center', size='large') title = title.replace(last_day,new_day) ax.text(0.5, 1.025, title, transform=ax.transAxes, ha='center', size='large') last_day = new_day print('%s %s "%s" %s %s "%s"'%(sys.argv[0],timeframe,filename,lower_alarm_limit,upper_alarm_limit,title)) return True return False read_from_file() while update_date(): read_from_file() ln = plot(ax) def update_mean(): if len(data_field)>0: tb_last.set_val("{:7.2f}".format(data_field[-1])) val = [data_field[-x-1] for x in range(min(10,len(data_field)))] aver = sum(val)/float(len(val)) tb_mean.set_val("{:7.2f}".format(aver)) ax_last = fig.add_axes([0.06, 0.05, 0.1, 0.05]) tb_last = TextBox(ax_last, "Last") tb_last.set_active(0) ax_mean = fig.add_axes([0.3, 0.05, 0.1, 0.05]) tb_mean = TextBox(ax_mean, "Mean(10)") tb_mean.set_active(0) update_mean() def init(): print('init') global timeframe if timeframe>24*60: ax.xaxis.set_major_formatter(mdates.DateFormatter('%d %H:%M')) else: ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M')) ax.yaxis.set_major_formatter(mtick.FormatStrFormatter('%.2f')) ax.tick_params(which='both',direction='in') ax.tick_params(which='major',width=2) ax.xaxis.set_minor_locator(mtick.AutoMinorLocator(5)) return ln,#fig.get_children()[0] def update(c): update_date() read_from_file() alarm_check() ln.set_data(data_time,data_field) update_mean() ax.relim() ax.autoscale_view() return ln,#fig.get_children()[0] ani = FuncAnimation(fig, update, frames=None, interval=30*1000, #milliseconds init_func=init, blit=False, repeat=False, cache_frame_data=False ) def submit_lower(val): global lower_alarm_limit try: lower_alarm_limit = float(val) except: tb_lower.set_val(str(lower_alarm_limit)) alarm_check() ax_lower = fig.add_axes([0.55, 0.05, 0.1, 0.05]) tb_lower = TextBox(ax_lower, "Lower lim") tb_lower.on_submit(submit_lower) tb_lower.set_val(str(lower_alarm_limit)) def submit_upper(val): global upper_alarm_limit try: upper_alarm_limit = float(val) except: tb_upper.set_val(str(upper_alarm_limit)) alarm_check() ax_upper = fig.add_axes([0.8, 0.05, 0.1, 0.05]) tb_upper = TextBox(ax_upper, "Upper lim") tb_upper.on_submit(submit_upper) tb_upper.set_val(str(upper_alarm_limit)) ax_check = fig.add_axes([0.74, 0.89, 0.25, 0.1]) check = CheckButtons( ax=ax_check, labels=["Beep on alarm"], actives=[beep_on_alarm] ) def callback(label): global beep_on_alarm beep_on_alarm = not beep_on_alarm alarm_check() check.on_clicked(callback) def callback_autozoom(e): ax.set_autoscale_on(True) ax.relim() ax.autoscale_view() ax_autozoom = fig.add_axes([0.1, 0.89, 0.12, 0.05]) btn_autozoom = Button(ax_autozoom, 'Autozoom') btn_autozoom.on_clicked(callback_autozoom) plt.show()