实验三 物联网中间件中实时数据预警及数据清洗

实验三物联网中间件的创建及数据清洗3.1实验背景近年来，随着物联网概念提出以及RFID射频识别技术在沃尔玛供应链管理上的成功运用，使得以往仅用于军事领域的RFID技术，凭借其成本低、体积小、无须接触等特性越来越多的运用在生产自动化、门禁、公路收费、货物跟踪等民用领域和矿山的人员定位系统中。然而，由于受到各种环境和射频技术本身的一些特点，RFID阅读器识别标签时会发生各种错误识别现象：积极读、消极读、标签冗余、阅读器冗余，这一些因素都会制约着RFID技术应用进一步发展。在传感器中同样会产生很多脏数据，例如错读，漏读现象。对上述几种脏数据，分别采用相应的算法加以清洗。由于原始数据流具有无限性，因此在清洗算法中使用了滑动窗口技术。其中，采用平均值策略消除数据冗余：即将一分钟的数据进行平均，将平均值来取代这一分钟内产生的多个数据，这样使得存入数据库的数据量大大减少，而当用户做历史查询时，也能体现出数据变化的大致趋势。移动平均值是对传感器数据错读现象的平滑处理，将该数据和该数据的前后几次数据求平均值，在很多情况下这个平均值就能更好的体现出这个时间点在时间段中的所要表达的数据。滑动窗口策略则可以消除积极读的现象，消除积极读实质上是去除原始数据流中的噪音，算法通过统计滑动窗口中的标签个数来判断标签是否为噪音。实时数据预警是判断采集的数据值是否在规定的范围内，如果不在，就设置预警标识，对实时数据进行预警。3.2实验目的1、了解中间件数据清洗的意义；2、了解中间件的开发；3、掌握中间件对数据处理的流程与控制;4、掌握中间件数据清洗的几种基本策略5、掌握中间件实时数据预警的方法3.3实验坏境VisualStudio20103.4实验内容一、具体内容1、中间件的开发2、编写中间件数据清洗策略3、为中间件添加实时数据预警功能二、内容详解1、中间件包括以下5个功能（1）初始化工作，包括初始化一级缓存和8小时缓存；（2）数据采集工作，采集由OPCserver产生的数据（3）数据处理工作，采用适当的清洗策略对数据进行清洗处理。（4）数据转存工作，将处理过的数据存入到数据库和8小时缓存中。（5）对外接口，供客户端调用。2、在本试验中使用的数据清洗策略包括3个分别是：（1）平均值策略：用一分钟产生数据的平均值取代这一分钟产生的所有数据，这样可以在不影响数据的大致走势下减少数据冗余减轻数据库的压力。（2）滑动窗口策略：定义一个固定大小的时间窗口，窗口每次滑动一个读取周期。如果窗口内的标签Tag.次数大于预先设定的最小阈值Min，即认为标签Tag.为非噪声标签，则该窗口就不会输出标签Tag.。滑动窗口策略可以给数据去噪。（3）移动平均值策略：用该数据的前几次数据和后几次数据加上该数据本身的平均值去取代该数据的值，这样就消除数据的随机性。3.5实验步骤一、新建项目打开VisualStudio2010,选择C#语言，新建一个类库程序，修改工程名。如下图所示。

二、数据库准备工作因为中间件要对采集的数据进行初始化工作而且要把处理之后的数据转存到数据库中，所以必须对数据库进行操作。这里我们专门新建一个项目来对数据库进行操作。第一步:新建项目，右击解决方案，添加新建项目，命名为：GetDataAccess。第二步：添加SQLHelper的项，这个类的作用是通过存储过程来对数据库进行操作，这里就不对这个类进行详解。首先将老师给的源码里面的SQLHelper.cs复制到我们刚刚新4|启睥决宅2仅个项目）~[etDataAccess|圈生成®LProperty重新生成【日由引用=|DBOperation.csliddleWsre计算代码建星值©项目依赖项⑶小项目生成顺序m，建的GetDataAccess文件夹下面，然后在VS解决方案中添加现有项，如下图所示，找到SQLHeleper并添加。新部迎...Ctrl+Shift+A西]新部迎...Ctrl+Shift+A西]现有项回...Shift-FAlt-FA囤Windows窗体心…用户控件二3添加引用旧...添加S务引用⑶...设为启动项目叫Access项目屈性第三步：新建一个名为DBOperation的类，加入下列代码，其中连接字符串中的Server一项的IP为数据库所在IP，password一项为数据库的密码，这两项应该依现实做相应修改。publicclassDBOperation(publicstringconnect="Server=;InitialCatalog=IOTDataBase;uid=sa;password=310;ConnectTimeout=500”;#region连接数据库///<summary>///获取数据库的连接///</summary>///<returns></returns>privateSqlConnectiongetConnection()(try(stringconnectionString=GetConfigFromXml.DataBaseConnectionString;SqlConnectionconn=newSqlConnection(connectionString);if(conn.State!=ConnectionState.Open)(conn.Open();}returnconn;}catch(Exceptionee)(throwee;}}#endregion///<summary>///获取数据库中表T_TAG_INDEX中的所有字段///</summary>///<returns></returns>publicDataSetgetTAG_INDEXAllValue()(stringconn=GetConfigFromXml.DataBaseConnectionString;DataSetds=SqlHelper.ExecuteDataset(conn,"zc_SelectTagIndexValue");returnds;}第四步：在数据库中添加相应的存储过程，脚本语言如下。SETANSI_NULLSONGOSETQUOTED_IDENTIFIERONGOCREATEPROCEDURE[dbo].[zc_SelectTagIndexValue]ASBEGIN--SETNOCOUNTONaddedtopreventextraresultsetsfrom--interferingwithSELECTstatements.SETNOCOUNTON;--InsertstatementsforprocedurehereSELECTTAG_ID,TAG_NAME,TYPE_ID,UNIT,MIN,MAX,REMARK,IS_ALARM,SAVE_RATE,DEFAULT_VALfromT_TAG_INDEXEND三、具体编码（1）新建TagObject类来封装中间件获取的数据信息。首先右击项目，找到添加中的新建项，将类名修改为TagObjecto(2)编写TagObject首先写入字段属性，对应传感器或者读卡器产生数据。privatestringtag_id;〃读卡器ID，传感器IDprivatestringtag_name;privateinttype_id;//类型编号privatestringunit;〃计量单位privatestringdefault_value;//默认值privatestringmax;//最大值privatestringmin;//最小值privateintrate;//采集频率privatestringremarks;//说明privateboolis_alarm;//是否报警privateDateTimegetValueDateTime;//获取值得时间privateintquality;//质量码privateobjecttagValue;//传感器所监控到的值或者读卡器读到的ID给字段属性添加Set和Get方法，下表只做示例，其余属性相同处理。publicstringTAG_ID(get(returntag_id;}set(this.tag_id=value;添加必要的构造数和方法。publicTagObject(stringtag_id,stringtag_name)(this.tag_id=tag_id;this.tag_name=tag_name;}publicTagObject(stringtag_id,objecttag_value,DateTimeget_time,intquality)(this.tag_id=tag_id;this.tagValue=tag_value;this.getValueDateTime=get_time;this.quality=quality;}publicoverridestringToString()(DateTimetime=this.GETVALUEDATETIME;if(tagValue!=null)(return(this.tag_name+"-"+time.ToLongTimeString()+":"+this.tagValue.ToString());}else(return(this.tag_name+"-"+time.ToLongTimeString());}}（3）按上述方法依次新建RTDataRemoteObject类（中间件类，负责数据采集，清洗，及转存），TagCacheL1类（一级缓存，保存所有传感器读卡器的实时值），CacheTagList类（8小时缓存，保存服务器运行8小时的值），DateObject类（保存上次采集时间和采集频率的类，用途是控制采集频率）。下面一一详细介绍这些类的具体编码。DateObject类publicclassDateObject（publicintcol_rate;//最采集频率publicDateTimeget_time;//采集时的时间publicDateObject（intcol_rate,DateTimeget_time）this.col_rate=col_rate;this.get_time=get_time;TagCacheLl类，是中间件的一级缓存。publicclassTagCacheLl(privatestaticTagCacheL1myInstance;//获取自己的一个实例publicHashtabletagValueHashtable=Hashtable.Synchronized(newHashtable());//保存实时读到的键值对///<summary>///返回一个CacheL1对象///</summary>///<returns></returns>publicstaticTagCacheL1GetInstance()(if(myInstance==null)(myInstance=newTagCacheL1();}returnmyInstance;}}CacheTagList类，是8小时缓存publicclassCacheTagList(///<summary>///存放的是tag_name以及对应的包含平均值的一个TagObject对象///</summary>publicHashtablecurrentTagMinuteAverageHt=newHashtable();///<summary>///存放的是tag_name以及对应的包含多组平均值的Array[]数组///</summary>publicHashtabletagNameToValueArraysHT=null;publicintip_position=0;///<summary>///初始化tagNameToValueArraysHT///</summary>///<paramname="tagName">包含tag_name的数组</param>///<returns></returns>publicboolIntilize8HDB(string[]tagName)(boolflag=false;this.tagNameToValueArraysHT=newHashtable();for(inti=0;i<tagName.Length;i++)(object[]objArray=newobject[480];this.tagNameToValueArraysHT.Add(tagName[i],objArray);flag=true;}returnflag;}///<summary>///将currentTagMinuteAverageHt中存储的平均值///存入到tagNameToValueArraysHT中的array[]中///</summary>publicvoidAddMinuteAverageValueToHt()(try(object[]objArray;if(this.ip_position>=0x1df)//0x1df=479(this.ip_position=0;}else(this.ip_position++;}if(this.currentTagMinuteAverageHt!=null)(IDictionaryEnumeratorenumerator=this.currentTagMinuteAverageHt.GetEnumerator();while(enumerator.MoveNext())(stringkey=(string)enumerator.Key;〃当currentTagMinuteAverageHt和tagNameToValueArraysHT包含相同key时if(this.tagNameToValueArraysHT.ContainsKey(key)){//currentTagMinuteAverageHt中的值是TagObject对象TagObjectobj2=(TagObject)enumerator.Value;//tagNameToValueArraysHT中的值是一个Object[]数组〃数组中存放的是一个TagObject的tagValue值objArray=(object[])this.tagNameToValueArraysHT[key];if(obj2!=null)(objArray[this.ip_position]=obj2;this.tagNameToValueArraysHT[key]=objArray;}}}}}catch(Exceptione)(LogWriter.LogE("实时数据采集-分钟平均值缓存",e.Message);}}RTDataRemoteObject类，是最重要的类，其负责数据的采集、清洗与转存，还提供对外的数据接口。我们分功能对其进行展示。&导入包及定义变量首先引用包usingSystem.Collections;usingSystem.Windows.Forms;usingSystem.Threading;usingSystem.Diagnostics;usingGetDataAcess;其中System.Windows.Forms的使用需要添加.NET引用，在解决方案资源管理器中，右击引用，如下图所示。雾突方段器▼毕xhIS.®.囹MiddlewareWProperties画'J添加引用回…硝匚瀚]澳务弓!用㈤…碧GetConfigFromXmI.cs盟LogWriter.es费OPC.cs费RTDataRemoteOLJect.cs*TagCacheLl.cs购TagObjsct.csOPCdotNETLib然后，在添加引用窗口中找到.NET选择卡，添加其中的System.Windows.Form.如下所示。8添加弓1用]NET1项目|瞄［知筛选为：，METFramework4沮件名称版本运行时路径，System.Web.Routing4.0.0.Dv4.0.30319C:\ProgramFiles(x&6)\System.Web.Servicesv4.0.30319C:\ProgramFiles(x&6)\Syst&tn.Windows.Fortns.D...v4.0.30319C:\ProgramFiles(x&6]\System.Windciws.Forms.D...V4.0.30319CAProgramFiles〔乂86八]Sy戒em,Wind□ws.Form54.ao.oV4.O.3O319C:\PrcigramSystem.Windciws.TnputM...v4.0.30319C:\PrograiTiFiles(k&6]\System.Windows.Present...4.D.D.0v4.0.30319C:\ProgramFiles(x860\System.Workflow.Activitiesv4.0.30319C:\ProgramFiles(x86)\System.Workflow.Compo...4.D.0.Dv4.0.30319C:\ProgramFiles(x86)\System.Workflow.Runtimev4.O.3O319C:\PrograrnFilas(x36]\SvstmtTL'WcirkRciwSiervicES叮4.0.D.Orrrv4.0.30319C:\PrciciramFil*fx&61\TL诵定前有类似的我们还需要引用GetDataAcess.dll,如下图所示。最后我们定义必要的变量；#region定义变量DBOperationdb=newDBOperation();〃一级缓存publicstaticTagCacheLltagCacheLl=newTagCacheL1();//八小时缓存publicstaticCacheTagListCache8HTagList=newCacheTagList();//8小时缓存是否启动publicboolis8HDBInitilized=false;//用于保存从OPC中取出tagid与tagobject的键值对publicstaticHashtableOPCtagCacheLHashtable=Hashtable.Synchronized(newHashtable());//保存tag的tag_name与tag_id键值对publicstaticHashtabletagHashtable=Hashtable.Synchronized(newHashtable());//OPCtag_id字符串，用于构造Hashtable的键值privatestring[]strOPCTagId=null;//保存DateObject与Tag_id的键值对，用于采集频率的计算publicstaticHashtableDateHashtable=Hashtable.Synchronized(newHashtable());//保存tag每分钟的值的hashtable(用于分钟平均值计算)publicstaticHashtableOPCTagMinuteValuesHashtable=Hashtable.Synchronized(newHashtable());//保存tag标签的最大值最小值publicstaticHashtableMINorMAXHashtable=Hashtable.Synchronized(newHashtable());//保存计算后的移动平均值的实时值publicstaticHashtablemovingAverageHt=Hashtable.Synchronized(newHashtable());//保存计算前移动平均值需要的数publicstaticHashtablemovingAverageListHt=Hashtable.Synchronized(newHashtable());//移动平均值窗口大小staticintWSIZE=5;//保存滑动窗口去噪策略最小阈值(不同的标签会有不同的阈值)publicstaticHashtableMINHashtable=Hashtable.Synchronized(newHashtable());〃是否启动移动平均值策略清洗数据publicboolisStartMovingAverage=false;//平均值计算线程privateThreaddataComputeThread=null;privateSystem.Threading.TimerdataComputeTimer=null;//OPC数据采集线程privateThreadOPCcurrentDataHarvestThread=null;privateSystem.Threading.TimerOPCDataHarvestTimer=null;publicHashtableOPCHashTable=newHashtable();//privateOPCOpc=newOPC();〃线程等待时间初始化函数，因为我们不能要求建立中间件对象的时候，就要让它工作；#region启动函数publicvoidInitRemoteObject()(//初始化一级缓存this.InitCacheLFromDB();//初始化8小时缓存this.InitCacheTagList();//采集。PCServer产生的实时数据this.OPCDataHarvestThread();//平均值计算线程包括转存数据this.DtaComputeThread();//this.DataHarvestThread();}#endregionC:准备工作，初始化一级缓存，作用是为采集，处理数据做初始化工作;#region初始化一级缓存publicvoidInitCacheLFromDB()(try(DataSetds=db.getTAG_INDEXAllValue();DataTabledt=ds.Tables[0];intcount=dt.Rows.Count;this.strOPCTagId=newstring[count];for(inti=0;i<count;i++)(string[]MAXMIN=newstring[2];//保存最大最小值stringtag_id=dt.Rows[i]["TAG_ID"].ToString().Trim();stringtag_name=dt.Rows[i]["TAG_NAME"].ToString();intrate=int.Parse(dt.Rows[i]["SAVE_RATE"].ToString());DateObjectdateobj=newDateObject(rate,DateTime.Now);DateHashtable.Add(tag_id,dateobj);this.strOPCTagId[i]=tag_id;TagObjectobj2=newTagObject(tag_id,tag_name);obj2.RATE=rate;if(dt.Rows[i]["TYPE_ID"]!=null)(obj2.TYPE_ID=int.Parse(dt.Rows[i]["TYPE_ID"].ToString());}if(dt.Rows[i]["UNIT"]!=null)(obj2.UNIT=dt.Rows[i]["UNIT"].ToString();}if(dt.Rows[i]["DEFAULT_VAL"]!=null)(obj2.TAGVALUE=dt.Rows[i]["DEFAULT_VAL"].ToString();}if(dt.Rows[i]["MIN"]!=null)(obj2.MIN=dt.Rows[i]["MIN"].ToString();MAXMIN[1]=dt.Rows[i]["MIN"].ToString();}if(dt.Rows[i]["MAX"]!=null)(obj2.MAX=dt.Rows[i]["MAX"].ToString();MAXMIN[0]=dt.Rows[i]["MAX"].ToString();}obj2.RATE=rate;if(dt.Rows[i]["REMARK"]!=null)(obj2.REMARKS=dt.Rows[i]["REMARK"].ToString();}if(dt.Rows[i]["IS_ALARM"]!=null)(obj2.IS_ALARM=bool.Parse(dt.Rows[i]["IS_ALARM"].ToString());}obj2.GETVALUEDATETIME=DateTime.Now;obj2.QUALITY=1;//tagCacheLHashtable.Add(tag_id,obj2);OPCtagCacheLHashtable.Add(tag_id,obj2);MINorMAXHashtable.Add(tag_id,MAXMIN);tagHashtable.Add(tag_id,tag_name);movingAverageListHt.Add(tag_id,newArrayList());tagCacheL1.tagValueHashtable=OPCtagCacheLHashtable;inttimes=60000/rate;MINHashtable.Add(tag_id,times);}}catch(Exceptione)MessageBox.Show("Error:初始化一级缓存出错”+e.Message);throw(e);JD：准备工作，初始化8小时缓存；#region初始化八小时缓存///<summary>///根据传感器TAG初始化八小时缓存///</summary>///<returns></returns>publicvoidInitCacheTagList()(try(Cache8HTagList.Intilize8HDB(this.strOPCTagId);this.is8HDBInitilized=true;}catch(Exceptione)(MessageBox.Show(”实时数据采集-初始化八小时缓存:"+e.Message);}}#endregionE：数据采集工作，采用线程每隔一秒中执行一次采集函数，采集一组数据到缓存中。大家注意代码标红的那一段，这一段的作用就是设置实时预警，当数据大于最大值，小于最小值时，报警标识标记为True；#regionOPCServer实时数据采集publicvoidOPCDataHarvestThread()(ThreadStartstart=newThreadStart(this.OPCDataHarvestService);this.OPCcurrentDataHarvestThread=newThread(start);this.OPCcurrentDataHarvestThread.Start();}protectedvoidOPCDataHarvestService()(intdeTime=this.wait;〃延后时间TimerCallbackCallback=newTimerCallback(this.OPCCurrentDataHarvest);this.OPCDataHarvestTimer=newSystem.Threading.Timer(Callback,null,deTime,1000);publicvoidOPCCurrentDataHarvest(objectstate)(ArrayListtagIds=newArrayList();try(Hashtableht=newHashtable(DateHashtable);IDictionaryEnumeratorenumerator=ht.GetEnumerator();if(enumerator!=null)(while(enumerator.MoveNext())(stringkey=enumerator.Key.ToString();DateObjectdateObject=(DateObject)enumerator.Value;if(DateTime.Now>=dateObject.get_time.AddMilliseconds(dateObject.col_rate))(tagIds.Add(key);dateObject.get_time=DateTime.Now;DateHashtable[key]=dateObject;}}}}catch(Exceptione)(MessageBox.Show("Error:OPC采集实时数据出错，出错段为给tag刷新数据:"+e.Message);throw(e);}try(if(tagIds!=null)(for(inti=0;i<tagIds.Count;i++)(stringtagId=tagIds[i].ToString();if(OPCtagCacheLHashtable.ContainsKey(tagId))(TagObjecttagObject=(TagObject)OPCtagCacheLHashtable[tagId];if(tagObject!=null)(try(if(OPCHashTable[tagId]!=null)(tagObject.TAGVALUE=((TagObject)OPCHashTable[tagId]).TAGVALUE;tagObject.GETVALUEDATETIME=((TagObject)OPCHashTable[tagId]).GETVALUEDATETIME;tagObject.QUALITY=((TagObject)OPCHashTable[tagId]).QUALITY;if(MINorMAXHashtable[tagObject.TAG_ID]!=null)(tagObject.MAX=((string[])MINorMAXHashtable[tagObject.TAG_ID])[0];tagObject.MIN=((string[])MINorMAXHashtable[tagObject.TAG_ID])[1];}doublevalue=Convert.ToDouble(tagObject.TAGVALUE);doublemin=Convert.ToDouble(tagObject.MIN);doublemax=Convert.ToDouble(tagObject.MAX);if(value<max&&value>=min)(tagObject.IS_ALARM=false;〃当标签是温度时，即使是最小值也需要报警if(tagObject.TAG_ID.Contains("TP")&&value==min)(tagObject.IS_ALARM=true;}}else(tagObject.IS_ALARM=true;}if(tagCacheL1.tagValueHashtable.ContainsKey(tagObject.TAG_ID))(tagCacheL1.tagValueHashtable[tagObject.TAG_ID]=tagObject;//将数据转存到OPCTagMinuteValuesHashtable，此哈希表用于求每分钟的平均值elsetagObject.TAGVALUE=-1;tagObject.GETVALUEDATETIME=DateTime.Now;「…=0;if(isStartMovingAverage)”"catch(Exceptione)MessageBox.Show("Error:OPC采集实时数据出错，出错段为给tag刷新数据:"+e.Message);throw(e);}}}}}catch(Exceptione)(MessageBox.Show(”实时数据采集-OPC采集实时数据出错:"+e.Message);throw(e);}}publicboolAddTagValueToTagList(TagObjecttagObject)(boolflag=true;try(//这些TagObject对像必须是在Cache中的if(tagObject!=null)(ArrayListlist;if(OPCTagMinuteValuesHashtable.ContainsKey(tagObject.TAG_ID))(list=(ArrayList)OPCTagMinuteValuesHashtable[tagObject.TAG_ID];list.Add(tagObject.TAGVALUE);OPCTagMinuteValuesHashtable[tagObject.TAG_ID]=list;}else(list=newArrayList();list.Add(tagObject.TAGVALUE);OPCTagMinuteValuesHashtable.Add(tagObject.TAG_ID,list);}}}catch(Exceptione)(flag=false;MessageBox.Show(”实时数据采集-存入一分钟缓存:"+e.Message);}returnflag;}#endregionF：移动平均值策略#region计算移动平均值publicvoidComputeMovingAverage(TagObjecttag)(doublesecond=-(WSIZE-1)/2;doublesum=0;doubleaverageValue=0;TagObjectaverageTag;if(movingAverageListHt.ContainsKey(tag.TAG_ID))(ArrayListmovingAverageList=(ArrayList)movingAverageListHt[tag.TAG_ID];if(movingAverageList.Count<WSIZE-1)(movingAverageList.Add(tag.TAGVALUE);DateTimedtime=tag.GETVALUEDATETIME.AddSeconds(second);averageTag=newTagObject(tag.TAG_ID,-1,dtime,0);}else(movingAverageList.Add(tag.TAGVALUE);for(inti=0;i<movingAverageList.Count;i++)(doubled=Convert.ToDouble(movingAverageList[i]);sum+=d;}//decimal.Round(Convert.ToDecimal(floatPart),3);averageValue=(double)decimal.Round(Convert.ToDecimal(sum/WSIZE),3);DateTimedtime=tag.GETVALUEDATETIME.AddSeconds(second);averageTag=newTagObject(tag.TAG_ID,averageValue,dtime,1);movingAverageList.RemoveAt(0);}if(movingAverageHt.ContainsKey(tag.TAG_ID))(movingAverageHt[tag.TAG_ID]=averageTag;}else(movingAverageHt.Add(tag.TAG_ID,averageTag);}}}#endregionG:平均值策略和滑动窗口策略，平均值策略即求1分钟的平均值，减少了存入数据库数据的冗余，又不影响数据的走势。而滑动窗口策略是为了给数据去噪，去掉那些极可能是干扰的数据。#region一分钟数据平均值计算线程publicvoidDtaComputeThread()(ThreadStartstart=newThreadStart(this.dataComputeService);this.dataComputeThread=newThread(start);this.dataComputeThread.Start();protectedvoiddataComputeService()(intdelay=0x7530;//30秒intInterval=0xea60;//60秒TimerCallbackCallback=newTimerCallback(this.MinuteDataAverageCompute);this.dataComputeTimer=newSystem.Threading.Timer(Callback,null,delay,Interval);}publicvoidMinuteDataAverageCompute(objectstate)(HashtabletagMinuteValuesHashtable=newHashtable(OPCTagMinuteValuesHashtable);OPCTagMinuteValuesHashtable.Clear();this.ComputingAllTagValuesMinuteAverage(tagMinuteValuesHashtable);Cache8HTagList.AddMinuteAverageValueToHt();}#region计算每分钟数据的平均值///<summary>///tagMinuteValuesHashtable中的键为tag_name值为list里面存放的是tag_Value///</summary>///<returns></returns>publicboolComputingAllTagValuesMinuteAverage(HashtabletagMinuteValuesHashtable)(boolflag=false;IDictionaryEnumeratorenumerator=null;if(tagMinuteValuesHashtable!=null)(enumerator=tagMinuteValuesHashtable.GetEnumerator();}try(if(enumerator==null)(returnflag;}while(enumerator.MoveNext())(〃针对每个tag_namestringkey=(string)enumerator.Key;ArrayListlist=(ArrayList)enumerator.Value;〃针对每个Tagname求取if(list!=null)(lEnumeratorenumerator2=list.GetEnumerator();intcount=list.Count;//滑动窗口去噪,如果采集的数量小于设定的最小阈值说明是噪声，应该去掉inttimes=(int)MINHashtable[key];if(times!=null)(if(count<times/3)(continue;}}doublenum1=0f;if(count>0)(doublenum2=0f;//if(sensorTagHashtable.ContainsKey(key))//tag_id=sensorTagHashtable[key].ToString();while(enumerator2.MoveNext())(if(enumerator2.Current!=null)(//求和num2+=Convert.ToDouble(enumerator2.Current);}}num1=num2/((float)count);num1=(double)decimal.Round(Convert.ToDecimal(num1),3);}TagObjectcurrent=newTagObject(key,num1,DateTime.Now,1);db.InsertMinuteAverage(key,DateTime.Now,num1,1);if(Cache8HTagList.currentTagMinuteAverageHt.ContainsKey(key))(Cache8HTagList.currentTagMinuteAverageHt[key]=current;}else(Cache8HTagList.currentTagMinuteAverageHt.Add(key,current);}flag=true;}else(flag=false;}}}catch(Exceptione)(flag=false;MessageBox.Show("实时数据采集-平均值计算”+e.Message);}returnflag;}#endregion#region清空保存传感器类一分钟数据的HashTable///<summary>///清空保存传感器类一分钟数据的HashTable///</summary>///<returns></returns>publicboolRemoveAllMinuteTagValues()(boolflag=true;if(OPCTagMinuteValuesHashtable!=null)(IDictionaryEnumeratorenumerator=OPCTagMinuteValuesHashtable.GetEnumerator();try(while(enumerator.MoveNext())(stringkey=(string)enumerator.Key;((ArrayList)OPCTagMinuteValuesHashtable[key]).Clear();}}catch(InvalidOperationExceptione)(flag=false;MessageBox.Show("实时数据采集-清空传感器一分钟数据"+e.Message);}}returnflag;}#endregion#endregionH:对外接口，供应用软件调用。#region实时数据访问接口///<summary>///获取TAG最新读取到的值///</summary>///<returns>TagObject</returns>publicTagObjectgetRealtimeTagValue(stringtagId)(TagObjectobj=null;if(tagCacheL1.tagValueHashtable.ContainsKey(tagId))(TagObjectobj2=(TagObject)tagCacheL1.tagValueHashtable[tagId];if(obj2!=null)(obj=obj2.Clone();}}returnobj;}///<summary>///启动移动平均值策略///</summary>publicvoidStartMovingAverage()(this.isStartMovingAverage=true;return;}///<summary>///停止移动平均值策略///</summary>publicvoidStopMovingAverage()(this.isStartMovingAverage=false;return;}///<summary>///获取移动平均值的实时值///</summary>///<paramname="tagId"></param>///<returns></returns>publicTagObjectgetMovingAverageTagValue(stringtagId)(if(isStartMovingAverage==true)(TagObjectobj=null;if(movingAverageHt.ContainsKey(tagId))(TagObjectobj2=(TagObject)movingAverageHt[tagId];if(obj2!=null)(obj=obj2.Clone();}}returnobj;}else(returnnull;}}///<summary>///根据输入的【。数组获取对应的实时值链表///</summary>///<paramname="tagId"></param>///<returns></returns>publicArrayListgetTagValueList(string[]tagId)(ArrayListtagList=newArrayList();TagObjectobj=null;inti=tagId.Length;for(intj=0;j<i;j++)(if(tagCacheL1.tagValueHashtable.ContainsKey(tagId[j]))(TagObjectobj2=(TagObject)tagCacheL1.tagValueHashtable[tagId[j]];if(obj2!=null)(obj=obj2.Clone();}}tagList.Add(obj);}returntagList;}///<summary>Ill获取OPCServer的最新值///<lsummary>Ill<paramname="ht"><lparam>Ill<paramname="secret"><lparam>publicvoidSetOPCHashtable(Hashtableht,stringsecret)(if(ht!=null&&secret=="310”)(OPCHashTable=ht;}return;}lll<summary>lll获取传感器最近八个小时的平均值，值的个数由arrayLength决定lll<lsummary>lll<returns>object<lreturns>publicobject[]getTagObjectRange(stringtagId,intarrayLength)(object[]objArray=null;intindex=Cache8HTagList.ip_position;if(Cache8HTagList.tagNameToValueArraysHT.ContainsKey(tagId))(objArray=newobject[arrayLength];object[]objArray2=(object[])Cache8HTagList.tagNameToValueArraysHT[tagId];for(inti=0;i<objArray.Length;i++)(if(index<0)(index=0x1df;}objArray[i]=null;if(objArray2[index]!=null)(objArray[i]=((TagObject)objArray2[index]).TAGVALUE;}index--;}}returnobjArray;}publicArrayListgetTagObjectRangeBytime(stringtagId,DateTimetimel,DateTimetime2)(ArrayListlist=newArrayList();intindex=Cache8HTagList.ipposition+1;if(Cache8HTagList.tagNameToValueArraysHT.ContainsKey(tagId))(object[]objArray=(object[])Cache8HTagList.tagNameToValueArraysHT[tagId];for(inti=0;i<480;i++)(if(index>0x1df)(index=0;}if(objArray[index]!=null)(TagObjecttag=(TagObject)objArray[index];if(tag.GETVALUEDATETIME>=time1&&tag.GETVALUEDATETIME<=time2)(list.Add(tag);}}index++;}}returnlist;}publicHashtablegetHash()(returntagCacheL1.tagValueHashtable;}#endregion至此，中间件的编写就初步结束了。但是此时的中间件是不能独立运行的，因为其还需要与OPCClient连接起来去OPCServer中取数据。我们打开前一个实验完成的IOTServer项目，将我们刚刚完成的中间件MiddleWare.dll添加到该项目的引用中，展开Mai