我们是通讯行业动力维护专业,因为外市电(三相交流电)的波动会对我们的设备造成一定的影响。于是想做一个交流电监测的应用,监测交流电的过压或欠压,所以我做了这么一个实时监测的工具。它可以在过压和欠压一定程度的时候告警。或者进行长期监测并记录电压数据,分析可能发生过压或欠压的原因。树莓派作为一个简单易用功能强大的计算平台,很好地支持Python。而且刚好有MCC118数据采集模块(HAT)可以直接扩展出数据采集功能,MCC118提供8通道模拟电压输入,基于树莓派的数据采集/数据记录系统。每张MCC118最大采样率为100kS/s,可以进行单电压点和电压波形采集。项目的代码就是在MCCDAQHATs提供的SDK示例代码中,continuous_scan.py的基础上修改来的。实现一个高速的采集及告警设备,来监测外市电的波动情况。长期监测主要通过长期数据记录,分析是否有未超出告警范围的波动,以分析供电质量。这只是一个工程样机,可以进行实时(延时3秒+)告警和储存发生告警时的电压波形;可以进行长期监测(不实时告警)、也可以长期监测并实时告警。内部构造如下图,其中MCC118在图中正上方。警告分为4种:本机的光警告、本机的声警告、继电器输出的外部警告、物联网平台(OneNet)警告。目前支持两路三相交流电监测,每路三相电的A、B、C相分别用:黄、绿、红,发光二极管警告。物联网平台的警告页面。分别是:一路三相电的一相警告,以及一路三相电的三相警告。采集到的正常交流电波形如下图。数据分析部分是用NumPy进行交流电每个周期的最大值、最小值判断,然后确定是否报警,是否写入硬盘。还有一个程序每一做任何判断,把采集到的所有数据都直接记录到硬盘,大概一小时会生成1.2G的数据。也是用Python写的脚本,通过NumPy把数据做一个转换,然后用NumPy的amax和amin对数值进行分析判断。如果数据超过阈值,就写硬盘文件,同时用GPIO输出控制LED来发出预警信号。以下是完整代码,其中关键代码已经添加了注释。#!/usr/bin/envpython#-*-coding:utf-8-*-"""MCC118FunctionsDemonstrated:mcc118.a_in_scan_startmcc118.a_in_scan_readmcc118.a_in_scan_stopmcc118.a_in_scan_cleanupPurpose:Performacontinuousacquisitionon1ormorechannels.Description:Continuouslyacquiresblocksofanaloginputdataforauser-specifiedgroupofchannelsuntiltheacquisitionisstoppedbytheuser.Thelastsampleofdataforeachchannelisdisplayedforeachblockofdatareceivedfromthedevice.MCC118共8个通道,可以监测两个三相变压器的6路市电,分别为变压器1的A、B、C相和变压器2的A、B、C相"""from__future__importprint_functionfromdaqhatsimportmcc118,OptionFlags,HatIDs,HatErrorfromdaqhats_utilsimportselect_hat_device,enum_mask_to_string,\chan_list_to_maskimportosimportthreadingimportnumpyasnpimportdatetimeimporttimefromsocketimport*importRPi.GPIOasGPIO#===GPIOsetup=====================================================================GPIO.setmode(GPIO.BCM)led_list=[5,6,19,16,20,21]#LED使用的GPIOGPIO.setup(led_list,GPIO.OUT,initial=GPIO.LOW)delta=datetime.timedelta(minutes=5)#===================================================================================READ_ALL_AVAILABLE=-1CURSOR_BACK_2='\x1b[2D'sERASE_TO_END_OF_LINE='\x1b[0K'#===================================================================================arraydata=[[0foriinrange(72000)]forhinrange(10)]#定义一个采样缓存数组,72000为read_request_size*信道数,10为#连续采样的次数c_time#=======OneNetparam===================================================================device_id="Your设备ID"#Your设备IDapi_key="YourAPI_KEY"#YourAPI_KEYHOST='api.heclouds.com'PORT=80BUFSIZ=1024ADDR=(HOST,PORT)#=======Channelset==================================================================#Storethechannelsinalistandconvertthelisttoachannelmaskthat#canbepassedasaparametertotheMCC118functions.channels=[0,1,2,3,4,5]channel_mask=chan_list_to_mask(channels)num_channels=len(channels)samples_per_channel=0options=OptionFlags.CONTINUOUSscan_rate=10000.0#采样速率最大值为100k/信道数read_request_size=12000#每通道此次采样的点数c_time=10#连续采样的次数timeout=5.0#=======上下限=================================================================volt=220#根据需要设定标准电压thread=0.2#根据需要设定upline=volt*(1+thread)*1.414*0.013558#0.013558是根据电阻分压确定的downline=volt*(1-thread)*1.414*0.013558#0.013558是根据电阻分压确定的#=======LED监测=================================================================base_time=datetime.datetime.now()#定义时间临时数组led_stime=[base_time,base_time,base_time,base_time,base_time,base_time]#定义状态临时数组led_status=[-1,-1,-1,-1,-1,-1]#=====================================================================================defled_detect():globalled_stime,led_status#变压器的A、B、C相transno=["Trans1A","Trans1B","Trans1C","Trans2A","Trans2B","Trans2C"]try:#判断每个LED的状态whilePORT>0:time.sleep(1)ifGPIO.input(5)==GPIO.HIGHandled_status[0]led_stime[0]=datetime.datetime.now()led_status[0]=1ifGPIO.input(6)==GPIO.HIGHandled_status[1]led_stime[1]=datetime.datetime.now()led_status[1]=1ifGPIO.input(19)==GPIO.HIGHandled_status[2]led_stime[2]=datetime.datetime.now()led_status[2]=1ifGPIO.input(16)==GPIO.HIGHandled_status[3]led_stime[3]=datetime.datetime.now()led_status[3]=1ifGPIO.input(20)==GPIO.HIGHandled_status[4]led_stime[4]=datetime.datetime.now()led_status[4]=1ifGPIO.input(21)==GPIO.HIGHandled_status[5]led_stime[5]=datetime.datetime.now()led_status[5]=1#相应引脚延时5分钟熄灭,并向平台发送数据清除告警foriinrange(6):now=datetime.datetime.now()ifnow>(led_stime[i]+delta)andled_status[i]>=0:GPIO.output(led_list[i],GPIO.LOW)led_status[i]=-1t5=threading.Thread(target=senddata,args=(transno[i],"0"))t5.start()exceptKeyboardInterrupt:#Clearthe'^C'fromthedisplay.print(CURSOR_BACK_2,ERASE_TO_END_OF_LINE,'\n')print('Stopping')#========senddatatoOneNet==============================================================defsenddata(T_dev,alm):Content=""Content+="{\"datastreams\":[{\"id\":\""+T_dev+"\",\"datapoints\":[{\"value\":"+alm+"}]}"Content+="]}\r\n"value=len(Content)data=""data+="POST/devices/"+device_id+"/datapointsHTTP/1.1\r\n"data+="Host:api.heclouds.com\r\n"data+="Connection:close\r\n"data+="api-key:"+api_key+"\r\n"data+="Content-Length:"+str(value)+"\r\n"data+="\r\n"data+=ContenttcpCliSock=socket(AF_INET,SOCK_STREAM)tcpCliSock.connect(ADDR)tcpCliSock.send(data.encode())#data1=tcpCliSock.recv(BUFSIZ).decode()#print(data1)#=====writefile==========================================================================defwritefile(arrayatt,fnat):print("writefile",datetime.datetime.now())np.savetxt(fnat,arrayatt.T,fmt="%1.2f",delimiter="")print("\033[0;31m","finishwrite",datetime.datetime.now(),"\033[0m")#====AnalyseandwritefileandAlarm=======================================================defanalyse(array,fnat):break_1_1=-1break_2_1=-1break_1_2=-1break_2_2=-1break_1_3=-1break_2_3=-1break_1=-1break_2=-1print("analysefile",datetime.datetime.now())#等于read_request_size=12000lll=read_request_sizefile1="1-"+fnatfile2="2-"+fnata=np.array(array)b=np.empty([3,c_time*read_request_size],dtype=float)#变压器1定义一个数组c=np.empty([3,c_time*read_request_size],dtype=float)#变压器2定义一个数组#板子的数据记录格式是:CH0、CH1、CH2、CH3、CH4、CH5、CH0、CH1、CH2、CH3、CH4、CH5、CH0、CH1、CH2、CH3、CH4、CH5、CH0、CH1、CH2、CH3、CH4、CH5、CH0、CH1、CH2、CH3、CH4、CH5、#通过下面拆分成6个CH每个CH一个list以便进行分CH判断b[0]=np.concatenate((a[0,0::6],a[1,0::6],a[2,0::6],a[3,0::6],a[4,0::6],a[5,0::6],a[6,0::6],a[7,0::6],a[8,0::6],a[9,0::6]))#提取变压器1,A相数据b[1]=np.concatenate((a[0,1::6],a[1,1::6],a[2,1::6],a[3,1::6],a[4,1::6],a[5,1::6],a[6,1::6],a[7,1::6],a[8,1::6],a[9,1::6]))#提取变压器1,B相数据b[2]=np.concatenate((a[0,2::6],a[1,2::6],a[2,2::6],a[3,2::6],a[4,2::6],a[5,2::6],a[6,2::6],a[7,2::6],a[8,2::6],a[9,2::6]))#提取变压器1,C相数据c[0]=np.concatenate((a[0,3::6],a[1,3::6],a[2,3::6],a[3,3::6],a[4,3::6],a[5,3::6],a[6,3::6],a[7,3::6],a[8,3::6],a[9,3::6]))#提取变压器2,A相数据c[1]=np.concatenate((a[0,4::6],a[1,4::6],a[2,4::6],a[3,4::6],a[4,4::6],a[5,4::6],a[6,4::6],a[7,4::6],a[8,4::6],a[9,4::6]))#提取变压器2,B相数据c[2]=np.concatenate((a[0,5::6],a[1,5::6],a[2,5::6],a[3,5::6],a[4,5::6],a[5,5::6],a[6,5::6],a[7,5::6],a[8,5::6],a[9,5::6]))#提取变压器2,C相数据#=====判断越线=================================================================================#200是scan_rate=10000.0除以50(交流电的频率)得出每个周期采样200个点#600是120000/200=600read_request_size*c_time本次总的采样点数,除以每周期200点,总共600个周期#如果scan_rate=5000.0就是100,alens是120000/100=1200foriinrange(600):#每个CH拆分成每个正弦波周期进行峰值判断a1=b[0][(i*200):((i+1)*200-1)]b1=b[1][(i*200):((i+1)*200-1)]c1=b[2][(i*200):((i+1)*200-1)]a2=c[0][(i*200):((i+1)*200-1)]b2=c[1][(i*200):((i+1)*200-1)]c2=c[2][(i*200):((i+1)*200-1)]#每一项分别判断上下限并分别告警ifnp.amax(a1)>uplineornp.amax(a1)ifbreak_1_1GPIO.output(5,GPIO.HIGH)#相应引脚置高电平,点亮LEDbreak_1_1=1t3=threading.Thread(target=senddata,args=("Trans1A","100"))#调用上传进程t3.start()ifnp.amax(b1)>uplineornp.amax(b1)ifbreak_1_2GPIO.output(6,GPIO.HIGH)#相应引脚置高电平,点亮LEDbreak_1_2=1t3=threading.Thread(target=senddata,args=("Trans1B","100"))#调用上传进程t3.start()ifnp.amax(c1)>uplineornp.amax(c1)ifbreak_1_3GPIO.output(19,GPIO.HIGH)#相应引脚置高电平,点亮LEDbreak_1_3=1t3=threading.Thread(target=senddata,args=("Trans1C","100"))#调用上传进程t3.start()ifnp.amax(a2)>uplineornp.amax(a2)ifbreak_2_1GPIO.output(16,GPIO.HIGH)#相应引脚置高电平,点亮LEDbreak_2_1=1t3=threading.Thread(target=senddata,args=("Trans2A","100"))#调用上传进程t3.start()ifnp.amax(b2)>uplineornp.amax(b2)ifbreak_2_2GPIO.output(20,GPIO.HIGH)#相应引脚置高电平,点亮LEDbreak_2_1=1t3=threading.Thread(target=senddata,args=("Trans2B","100"))#调用上传进程t3.start()ifnp.amax(c2)>uplineornp.amax(c2)ifbreak_2_3GPIO.output(21,GPIO.HIGH)#相应引脚置高电平,点亮LEDbreak_2_2=1t3=threading.Thread(target=senddata,args=("Trans2C","100"))#调用上传进程t3.start()#每个变压器任何一项有告警就调用写文件进程写文件ifnp.amax(a1)>uplineornp.amax(a1)uplineornp.amax(b1)uplineornp.amax(c1)ifbreak_1t1=threading.Thread(target=writefile,args=(b,file1,))#调用写文件进程t1.start()break_1=2ifnp.amax(a2)>uplineornp.amax(a2)uplineornp.amax(b2)uplineornp.amax(c2)ifbreak_2t1=threading.Thread(target=writefile,args=(c,file2,))#调用写文件进程t1.start()break_2=2print("\033[0;32m","analysefinish",datetime.datetime.now(),"\033[0m")#fontcolor#============================================================================================defmain():"""Thisfunctionisexecutedautomaticallywhenthemoduleisrundirectly."""#============================================================================================try:#SelectanMCC118HATdevicetouse.address=select_hat_device(HatIDs.MCC_118)hat=mcc118(address)'''print('\nSelectedMCC118HATdeviceataddress',address)actual_scan_rate=hat.a_in_scan_actual_rate(num_channels,scan_rate)print('\nMCC118continuousscanexample')print('Functionsdemonstrated:')print('mcc118.a_in_scan_start')print('mcc118.a_in_scan_read')print('mcc118.a_in_scan_stop')print('Channels:',end='')print(','.join([str(chan)forchaninchannels]))print('Requestedscanrate:',scan_rate)print('Actualscanrate:',actual_scan_rate)print('Options:',enum_mask_to_string(OptionFlags,options))#============================================================================================try:input('\nPressENTERtocontinue...')except(NameError,SyntaxError):pass'''#Configureandstartthescan.#Sincethecontinuousoptionisbeingused,thesamples_per_channel#parameterisignoredifthevalueislessthanthedefaultinternal#buffersize(10000*num_channelsinthiscase).Ifalargerinternal#buffersizeisdesired,setthevalueofthisparameteraccordingly.print('Startingscan...PressCtrl-Ctostop\n')hat.a_in_scan_start(channel_mask,samples_per_channel,scan_rate,options)#============================================================================================try:whileTrue:fna=str(datetime.datetime.now())+".txt"#连续采样10次foriinrange(c_time):read_result=hat.a_in_scan_read(read_request_size,timeout)#readchannelDataarraydata[i]=read_result.dataifread_result.hardware_overrun:print('\n\nHardwareoverrun\n')breakelifread_result.buffer_overrun:print('\n\nBufferoverrun\n')breakhat.a_in_scan_stop()hat.a_in_scan_cleanup()hat.a_in_scan_start(channel_mask,samples_per_channel,scan_rate,options)#调用分析进程arraydatat=arraydatat2=threading.Thread(target=analyse,args=(arraydatat,fna,))t2.start()exceptKeyboardInterrupt:#Clearthe'^C'fromthedisplay.print(CURSOR_BACK_2,ERASE_TO_END_OF_LINE,'\n')print('Stopping')hat.a_in_scan_stop()hat.a_in_scan_cleanup()except(HatError,ValueError)aserr:print('\n',err)if__name__=='__main__':#表用进程实时监测告警状态,判断是否到时间消除告警t4=threading.Thread(target=led_detect)t4.start()main()程序部署参考这篇教程安装好MCCDAQHATs的SDK和代码示例。https://shumeipai.nxez.com/2018/10/18/get-started-with-the-evaluation-for-mcc-118.html将上面的程序保存为main.py然后在程序所在目录运行下面的命令即可启动程序。sudopythonmain.py*以上内容翻译自网络,原作者:mc_six,如涉及侵权,可联系删除。