using Newtonsoft.Json;
using SensorHub.Servers;
using SensorHub.Servers.Commands.CASICCommands;
using SensorHub.Servers.JsonFormat;
using SuperSocket.SocketBase.Command;
using SuperSocket.SocketBase.Protocol;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Caching;
using System.Text;
namespace SensorHub.Liquid
{
public class Liquid : CommandBase<CasicSession, StringRequestInfo>
{
private static MemoryCache memoryCache = new MemoryCache("Liquid");
private readonly object SyncObj = new object();
public override void ExecuteCommand(CasicSession session, StringRequestInfo requestInfo)
{
//TODO: construct the receving casic data
String preamble = requestInfo.Parameters[0];
String version = requestInfo.Parameters[1];
String leng = requestInfo.Parameters[2];
String devCode = requestInfo.Parameters[3];
String routeFlag = requestInfo.Parameters[4];
String dstNodeAddr = requestInfo.Parameters[5];
String pduType = requestInfo.Parameters[6];
String seq = requestInfo.Parameters[7];
String settings = requestInfo.Parameters[8];
String source = requestInfo.Parameters[9];//数据来源
String devName = "Liquid";
if (source.Contains("-"))
{
session.Send("HTTP/1.1 200 OK\r\n\r\n\r\n");
session.Close();
}
if (routeFlag == "01")//433有分包，其他方式没分包传输
{
//分包传输，对各个数据分包进行缓存
var item = memoryCache.Get(devCode);
if (item != null && (((int)item >> (Convert.ToInt32(seq, 16) - 1)) & 0x01) == 1)
{
session.Logger.Info("已缓存数据序列号：" + Convert.ToString((int)item, 2));
session.Logger.Info("该数据已缓存！");
return;
}
else
{
lock (this.SyncObj)//锁定
{
item = memoryCache.Get(devCode);
if (item != null)
{
session.Logger.Info("已缓存数据序列号：" + Convert.ToString((int)item, 2));
if ((((int)item >> (Convert.ToInt32(seq, 16) - 1)) & 0x01) == 1)
{
session.Logger.Info("该数据已缓存！");
return;
}
else
{
var cachePolicy = new CacheItemPolicy();
cachePolicy.AbsoluteExpiration = DateTimeOffset.Now.AddMinutes(1);
int Seq = (int)item;
Seq |= (1 << (Convert.ToInt32(seq, 16) - 1));
memoryCache.Set(devCode, Seq, cachePolicy);
session.Logger.Info("设置已经接收数据序列号：" + Convert.ToString(Seq, 2));
}
}
else
{
var cachePolicy = new CacheItemPolicy();
cachePolicy.AbsoluteExpiration = DateTimeOffset.Now.AddMinutes(1);
int Seq = 0;
Seq |= (1 << (Convert.ToInt32(seq, 16) - 1));
memoryCache.Set(devCode, Seq, cachePolicy);
session.Logger.Info("设置已经接收数据序列号：" + Convert.ToString(Seq, 2));
}
}
}
}
//print the receving data
String devType = "液位监测仪";
String operType = Common.getOpeTypeByPdu(pduType);
session.Logger.Info("AD接收数据:" + requestInfo.Body);
session.Logger.Info("设备类型:" + devType);
session.Logger.Info("操作类型:" + operType);
session.Logger.Info("会话:" + session.HubAddr + "," + session.SessionID);
//判断是返回的设置确认数据帧, 回复第三方
if (operType == "SetResponse")
{
Common.sendSetResponse(session, devCode, devName);
return;
}
List<Tag> tags = Common.getTags(settings, session);
//具体业务处理
String collectDate = "";
int cell = -1;
int? pci = null;
int? rsrp = null;
int? snr = null;
String softwareVersion = "";
uint offset = 0;
uint size = 0;
List<String> eventList = new List<String>();
List<DatasJson> datasList = new List<DatasJson>();
List<String> startupList = new List<String>();
foreach (Tag tag in tags)
{
if (!(tag is UploadTag))
{
if (tag != null && tag is CellTag)
{
CellTag cellTag = tag as CellTag;
cell = cellTag.Cell;
continue;
}
if (tag != null && tag is PCITag)
{
PCITag pciTag = (PCITag)tag;
pci = pciTag.PCI;
continue;
}
if (tag != null && tag is RSRPTag)
{
RSRPTag rsrpTag = (RSRPTag)tag;
rsrp = rsrpTag.RSRP;
continue;
}
if (tag != null && tag is SNRTag)
{
SNRTag snrTag = (SNRTag)tag;
snr = snrTag.SNR;
continue;
}
//非业务处理
if (tag != null && tag is SystemDateTag)
{
SystemDateTag systemDateTag = tag as SystemDateTag;
collectDate = systemDateTag.CollectDate;
continue;
}
if (tag != null && tag is SensorException0Tag)
{
SensorException0Tag sensorException0 = tag as SensorException0Tag;
int state = sensorException0.state;
if (state == 0) continue;
eventList.Add(getLiquidAlarm(state));
session.Logger.Info("通道一发送容错信息：oid：" + tag.Oid + ";value:" + state);
continue;
}
if (tag != null && tag is SensorException1Tag)
{
SensorException1Tag sensorException1 = tag as SensorException1Tag;
int state = sensorException1.state;
if (state == 0) continue;
eventList.Add(getLiquidAlarm(state + 3));
session.Logger.Info("通道二发送容错信息：oid：" + tag.Oid + ";value:" + state);
continue;
}
if (tag != null && tag is SensorStartupTag)
{
SensorStartupTag sensorStartup = tag as SensorStartupTag;
String imei = sensorStartup.IMEI;
String iccid = sensorStartup.ICCID;
startupList.Add(imei);
startupList.Add(iccid);
session.Logger.Info("设备开机上报，设备编号DEVCODE：" + devCode + " IMEI:" + imei + " ICCID:" + iccid);
continue;
}
//非业务处理
if (tag != null && tag is SoftwareVersionTag)
{
SoftwareVersionTag versionTag = (SoftwareVersionTag)tag;
softwareVersion = versionTag.Version;
continue;
}
if (tag != null && tag is OffsetTag)
{
OffsetTag offsetTag = (OffsetTag)tag;
offset = offsetTag.Offset;
continue;
}
if (tag != null && tag is SizeTag)
{
SizeTag sizeTag = (SizeTag)tag;
size = sizeTag.Size;
continue;
}
}
else
{
//业务处理
UploadTag uploadTag = tag as UploadTag;
switch (uploadTag.BizType)
{
case 3:
//液位
TagHandler liquidHandler = new LiquidTagHandler();
liquidHandler.resolve(tag, session);
DateTime baseTime = Convert.ToDateTime(collectDate + " " + liquidHandler.CollecTime);
for (int i = 0; i < liquidHandler.DataList.Count; i++)
{
DateTime upTime = baseTime.AddMinutes(i * liquidHandler.Interval);
String uptime = upTime.ToString("yyyy") + upTime.ToString("MM") + upTime.ToString("dd")
+ upTime.ToString("HH") + upTime.ToString("mm") + upTime.ToString("ss");
datasList.Add(new LiquidDatasJson(uptime, (float)liquidHandler.DataList[i]));
}
break;
default:
session.Logger.Info("未知业务类型！");
break;
}
}
}
byte[] btPdu = new byte[2]; //2个字节
btPdu[0] = Common.getRespOperType(operType, source == "433" ? true : false);
if (btPdu[0] == 0xFF)//未知操作类型
{
return;
}
btPdu[1] = 0x87;
if (softwareVersion != "" || size != 0 || offset != 0)//进入远程升级流程
{
Common.remoteUpgrade(session, operType, devName, devCode, btPdu, softwareVersion, size, offset, source);
Common.sendVersinData(session, devName, devCode, softwareVersion);
return;
}
//Common.sendMessage(session, devName, devCode, cell, pci, rsrp, snr, eventList, datasList, startupList);
Common.kafkaProduce(session, devName, devCode, cell, pci, rsrp, snr, eventList, datasList, startupList);
//Common.sendConfig(session, devCode, routeFlag, source, btPdu);
if (isfinishe(pduType))
{
try
{
if (routeFlag == "03") //GPRS，3G网络，电信平台
{
Common.senderGPRSConfig(session, devCode, btPdu, source);
}
else //433
{
sender433Config(session, devCode, btPdu, 0x01);
}
}
catch (Exception e)
{
session.Logger.Error(e.ToString());
}
}
}
private void sender433Config(CasicSession session, string devCode, byte[] btPdu, byte routeflag)
{
CasicSender sender = new CasicSender(null);
byte[] config = sender.build433RspConfigFrame(devCode);
config[10] = routeflag;//通信方式
btPdu.CopyTo(config, 13);
var item = memoryCache.Get(devCode);//业务类型
if (item != null)
{
byte[] value = BitConverter.GetBytes((int)item);
config[34] = value[1];
config[35] = value[0];
}
SystemTimeConfig sysTimeConfig = new SystemTimeConfig(null);
byte[] btConfig = sysTimeConfig.getConfig(new byte[0]);
btConfig.CopyTo(config, 16);
session.Logger.Info("下发回复信息：" + BitConverter.ToString(config).Replace("-", ""));
session.Send(Common.CRC(config), 0, config.Length + 2);
memoryCache.Remove(devCode);//移除上一次缓存的分包状态
}
private bool isfinishe(String pduType)
{
Int16 btpduType = Int16.Parse(pduType, System.Globalization.NumberStyles.HexNumber);
int flag = btpduType & 0x80;
return flag == 128 ? true : false;
}
private String getLiquidAlarm(int state)
{
switch (state)
{
case 0:
return "LiquidPressureNormal";//液位静压探头正常
case 1:
return "LiquidPressureFail"; //液位静压探头采集失败
case 2:
return "LiquidPressureError"; //液位静压探头数据异常
case 3:
return "LiquidUltrasonicNormal"; //液位超声波探头正常
case 4:
return "LiquidUltrasonicFail"; //液位超声波探头采集失败
case 5:
return "LiquidUltrasonicError"; //液位超声波探头数据异常
default:
return "LiquidUnknown"; //液位未知异常
}
}
}
}
