HistoryTaskService.cs 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404
  1. using DevExpress.Utils.Extensions;
  2. using System;
  3. using System.Collections.Generic;
  4. using System.Configuration;
  5. using System.Globalization;
  6. using System.IO;
  7. using System.Linq;
  8. using System.Text;
  9. using System.Threading;
  10. using System.Threading.Tasks;
  11. using XdCxRhDW.Dto;
  12. namespace X2D1NoRefTaskServer.Service
  13. {
  14. public class HistoryTaskService
  15. {
  16. private readonly string baseUrl;
  17. CancellationTokenSource cts;
  18. public HistoryTaskService()
  19. {
  20. var posPlatformAddr = ConfigurationManager.AppSettings["PosPlatformAddr"].Trim();//like http://127.0.0.1:8091 or http://127.0.0.1:8091/
  21. if (posPlatformAddr.EndsWith("/"))
  22. this.baseUrl = posPlatformAddr + "api/";
  23. else
  24. this.baseUrl = posPlatformAddr + "/api/";
  25. }
  26. public void StartAsync(X2D1HistoryTaskHandleDto dto)
  27. {
  28. cts = new CancellationTokenSource();
  29. if (!Directory.Exists(dto.CapDir))
  30. {
  31. StopTask(dto.ID, $"文件路径[{dto.CapDir}]不存在,任务结束");
  32. return;
  33. }
  34. if (dto.DateDirFormat.Contains("\\"))
  35. {
  36. StopTask(dto.ID, $"子目录日期格式不能包含多级目录格式");
  37. return;
  38. }
  39. LogHelper.Info($"【任务{dto.ID}】开始执行...");
  40. int.TryParse(ConfigurationManager.AppSettings["PosDtoFactor"], out int posDtoFactor);
  41. if (posDtoFactor == 0) posDtoFactor = 1;//定位时时差系数
  42. LogHelper.Info($"【任务{dto.ID}】定位时差系数={posDtoFactor}");
  43. Task.Run(async () =>
  44. {
  45. bool canConnected = CanGetSatIdFromMySql();
  46. DateTime preTime = dto.StartTime;
  47. int formatFlag;
  48. if (string.IsNullOrWhiteSpace(dto.DateDirFormat))
  49. {
  50. //没有日期目录,处理完目录中的数据后停止任务
  51. formatFlag = 0;
  52. }
  53. else if (dto.DateDirFormat.ToLower().EndsWith("dd"))
  54. {
  55. //处理完一个目录后跳转到第二天的目录
  56. formatFlag = 1;
  57. }
  58. else if (dto.DateDirFormat.ToUpper().EndsWith("HH"))
  59. {
  60. //处理完一个目录后跳转到下一个小时的目录
  61. formatFlag = 2;
  62. }
  63. else
  64. {
  65. StopTask(dto.ID, $"执行异常,不支持的日期目录格式");
  66. return;
  67. }
  68. while (!cts.IsCancellationRequested && preTime <= dto.EndTime)
  69. {
  70. string filesDir = dto.CapDir;
  71. try
  72. {
  73. if (formatFlag != 0)
  74. {
  75. filesDir = Path.Combine(dto.CapDir, $"{preTime.ToString(dto.DateDirFormat)}");//yyyyMMdd
  76. }
  77. IEnumerable<string> files;
  78. if (!Directory.Exists(filesDir))
  79. {
  80. LogHelper.Info($"【任务{dto.ID}】目录[{filesDir}]不存在,跳过此目录");
  81. ResetTime(formatFlag, ref preTime);
  82. continue;
  83. }
  84. LogHelper.Info($"【任务{dto.ID}】正在处理[{filesDir}]目录中的数据...");
  85. files = Directory.EnumerateFiles(filesDir, "*.dat");
  86. if (!files.Any())
  87. {
  88. LogHelper.Info($"【任务{dto.ID}】目录[{filesDir}]中没有文件,跳过此目录");
  89. ResetTime(formatFlag, ref preTime);
  90. continue;
  91. }
  92. IOrderedEnumerable<IGrouping<DateTime, HistoryFile>> groups = null;
  93. groups = files.Select(f => FileToHistoryFile(dto, f, canConnected)).GroupBy(m => m.CapTime).OrderBy(m => m.Key);
  94. foreach (var item in groups)
  95. {
  96. if (cts.IsCancellationRequested) break;
  97. var finfos = item.ToList();
  98. var capTime = finfos.First().CapTime;
  99. if (capTime < dto.StartTime) continue;
  100. if (capTime > dto.EndTime) break;
  101. if (finfos.Count < 3)
  102. {
  103. LogHelper.Warning($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻文件数量只有{finfos.Count}个,跳过此组数据");
  104. continue;
  105. }
  106. //超短波信号
  107. var dinfo = finfos.FirstOrDefault(m => m.Ch == 1);
  108. if (dinfo == null)
  109. {
  110. LogHelper.Warning($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻未找到超短波信号ch1文件,跳过此组数据");
  111. continue;
  112. }
  113. var taskSig = dto.Sigs.FirstOrDefault(p => p.FreqUp == dinfo.FreqHz);
  114. if (taskSig == null)
  115. {
  116. continue;//跳过不是任务处理的频点
  117. }
  118. //主星
  119. var minfo = finfos.FirstOrDefault(m => m.Ch == 2);
  120. if (minfo == null)
  121. {
  122. LogHelper.Warning($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻未找到主星信号ch2文件,跳过此组数据");
  123. continue;
  124. }
  125. //邻1
  126. var ninfo = finfos.FirstOrDefault(m => m.Ch == 3);
  127. if (ninfo == null)
  128. {
  129. LogHelper.Info($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻未找到邻星信号ch3文件,跳过此组数据");
  130. continue;
  131. }
  132. try
  133. {
  134. var delay1 = taskSig.SigDelay.FirstOrDefault(p => p.SatInfoSatCode == minfo.SatId)?.Delay;
  135. var delay2 = taskSig.SigDelay.FirstOrDefault(p => p.SatInfoSatCode == ninfo.SatId)?.Delay;
  136. if (delay1 == null) delay1 = 0;
  137. if (delay2 == null) delay2 = 0;
  138. LogHelper.Info($"【任务{dto.ID}】信号[{taskSig.FreqUp / 1e6}],主星[{minfo.SatId}],转发时延{delay1}us");
  139. LogHelper.Info($"【任务{dto.ID}】信号[{taskSig.FreqUp / 1e6}],邻星[{ninfo.SatId}],转发时延{delay1}us");
  140. string mainFile = await HttpHelper.UploadFileAsync(minfo.FilePath, baseUrl + "File/UploadFileAsync", token: cts.Token);//主星文件
  141. string adjaFile = await HttpHelper.UploadFileAsync(ninfo.FilePath, baseUrl + "File/UploadFileAsync", token: cts.Token);//邻星文件
  142. string cdbFile = await HttpHelper.UploadFileAsync(dinfo.FilePath, baseUrl + "File/UploadFileAsync", token: cts.Token);//超短文件
  143. DetectDto detectDto = new DetectDto()
  144. {
  145. file1 = cdbFile,//11局使用上行泄露信号进行检测
  146. dmcType = taskSig.SigType,//上行信号检测目前的算法只能使用基于能量的KY或IBS检测
  147. fsHz = minfo.FsHz,
  148. };
  149. detectDto.dmcType = detectDto.dmcType & ~EnumSigCheckTypeDto.DAMA;//DAMA不支持上行信号检测,在这里移除掉
  150. var deteResp = await HttpHelper.PostRequestAsync<List<DetectResDto>>(baseUrl + "DetectCg/DetectCalc", detectDto, token: cts.Token);
  151. if (deteResp.code != 200)
  152. {
  153. LogHelper.Error($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻信号检测出错.{deteResp.msg}");
  154. continue;
  155. }
  156. LogHelper.Info($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻信号检测完成,共{deteResp.data.Count}个时隙");
  157. var smps = deteResp.data.Select(m => new SmpPosition(m.Start, m.Length)).ToList();//怎么补0?
  158. var cgDto = new CpuCgMultiDto()
  159. {
  160. dtCenter = 260000,
  161. dtRange = 50000,
  162. file1 = cdbFile,
  163. file2 = mainFile,
  164. samplingRate = minfo.FsHz,
  165. smpPositions = smps,
  166. snrThreshold = 15,
  167. };
  168. var result1 = await HttpHelper.PostRequestAsync<List<CpuCgResDto>>(baseUrl + "DetectCg/CpuCgMultiCalc", cgDto, token: cts.Token);
  169. if (result1.code != 200)
  170. {
  171. LogHelper.Error($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻主星超短CPU参估出错.{result1.msg}");
  172. continue;
  173. }
  174. LogHelper.Info($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻主星超短CPU参估完成.");
  175. cgDto = new CpuCgMultiDto()
  176. {
  177. dtCenter = 260000,
  178. dtRange = 50000,
  179. file1 = cdbFile,
  180. file2 = adjaFile,
  181. samplingRate = minfo.FsHz,
  182. smpPositions = smps,
  183. snrThreshold = 15,
  184. };
  185. var result2 = await HttpHelper.PostRequestAsync<List<CpuCgResDto>>(baseUrl + "DetectCg/CpuCgMultiCalc", cgDto, token: cts.Token);
  186. if (result2.code != 200)
  187. {
  188. LogHelper.Error($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻邻星超短CPU参估出错.{result2.msg}");
  189. continue;
  190. }
  191. LogHelper.Info($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻邻星超短CPU参估完成");
  192. var data1 = result1.data;
  193. var data2 = result2.data;
  194. if (data1.Count != data2.Count || data1.Count != deteResp.data.Count)
  195. {
  196. LogHelper.Error($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻参估结果个数和检测结果个数不匹配");
  197. continue;
  198. }
  199. for (int i = 0; i < data1.Count; i++)
  200. {
  201. try
  202. {
  203. if (cts.IsCancellationRequested) break;
  204. X2D1NoXlNoParlPosDto x2D1 = new X2D1NoXlNoParlPosDto()
  205. {
  206. TaskID = dto.ID,
  207. SigTime = minfo.CapTime.AddSeconds(data1[i].Smpstart / minfo.FsHz),
  208. MainCode = minfo.SatId,
  209. AdjaCode = ninfo.SatId,
  210. //卫星转发某些频点可能有时延,无参定位由于不能抵消需要减去这个时延
  211. SxDto = data1[i].Dt * posDtoFactor - data2[i].Dt * posDtoFactor - delay1.Value + delay2.Value,
  212. SxDfo = data1[i].Df,
  213. SxSnr = data1[i].Snr,
  214. XdDto = data1[i].Dt * posDtoFactor - delay1.Value,
  215. XdDfo = data2[i].Df,
  216. XdSnr = data2[i].Snr,
  217. SatTxLon = dto.CapLon,
  218. SatTxLat = dto.CapLat,
  219. CdbTxLon = dto.CapLon,
  220. CdbTxLat = dto.CapLat,
  221. FreqDown = minfo.FreqHz,
  222. FreqUp = dinfo.FreqHz,
  223. CheckRes = new CheckResDto()
  224. {
  225. FileName = Path.GetFileName(dinfo.FilePath),
  226. ModRate = deteResp.data[i].ModRate,
  227. ModType = deteResp.data[i].ModType,
  228. SmpCount = deteResp.data[i].Length,
  229. SmpStart = deteResp.data[i].Start,
  230. UserName = deteResp.data[i].UserName,
  231. PosCheckType = deteResp.data[i].DmcType.GetEnumByDisplayName<EnumPosCheckTypeDto>(),
  232. }
  233. };
  234. var result = await HttpHelper.PostRequestAsync<PosResDto>(baseUrl + "Pos/PosX2D1NoXlNoParAsync", x2D1);
  235. if (result.code != 200)
  236. {
  237. LogHelper.Error($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻时隙位置{data1[i].Smpstart}定位异常.{result.msg}");
  238. }
  239. }
  240. catch (Exception ex)
  241. {
  242. LogHelper.Error($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻时隙位置{data1[i].Smpstart}定位异常", ex);
  243. }
  244. }
  245. LogHelper.Info($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻定位完成");
  246. }
  247. catch (Exception ex)
  248. {
  249. LogHelper.Error($"【任务{dto.ID}】{capTime:yyyyMMddHHmmss}时刻文件处理异常", ex);
  250. continue;
  251. }
  252. }
  253. ResetTime(formatFlag, ref preTime);
  254. LogHelper.Info($"【任务{dto.ID}】目录[{filesDir}]中的数据处理完成");
  255. }
  256. catch (Exception ex)
  257. {
  258. LogHelper.Error($"【任务{dto.ID}】目录[{filesDir}]中的数据处理出错,跳过此目录", ex);
  259. ResetTime(formatFlag, ref preTime);
  260. }
  261. }
  262. StopTask(dto.ID, $"数据处理完成,任务结束");
  263. }, cts.Token);
  264. }
  265. private void ResetTime(int formatFlag, ref DateTime time)
  266. {
  267. if (formatFlag == 0)
  268. {
  269. time = DateTime.MaxValue;
  270. }
  271. else if (formatFlag == 1)
  272. {
  273. time = time.AddHours(24);
  274. }
  275. else
  276. {
  277. time = time.AddHours(1);
  278. }
  279. }
  280. private HistoryFile FileToHistoryFile(X2D1HistoryTaskHandleDto dto, string filePath, bool canConnected)
  281. { //读取采集文件
  282. //2024_01_31_10_01_51_000000000_ch11_-1__Nxx.xxxxxx_Exx.xxxxxx_xxxxx.xxxHz_xxx.xxxMHz_ch1_xd1.dat
  283. HistoryFile historyFile = new HistoryFile();
  284. historyFile.FilePath = filePath;
  285. var fileName = Path.GetFileNameWithoutExtension(filePath);
  286. var strs = fileName.Split(new string[] { "_", "MHz", "Hz", "ch" }, StringSplitOptions.RemoveEmptyEntries);
  287. //采集时间
  288. var datestr = string.Join("_", strs.Take(6));
  289. DateTime dateTime;
  290. bool istime = DateTime.TryParseExact(datestr, "yyyy_MM_dd_HH_mm_ss", null, DateTimeStyles.None, out dateTime);
  291. if (!istime) return historyFile;
  292. historyFile.CapTime = dateTime;
  293. //采集通道
  294. var chstr = strs.Skip(strs.Length - 2).Take(1).First();
  295. int ch;
  296. int.TryParse(chstr, out ch);
  297. //采集频点
  298. var freqstr = strs.Skip(strs.Length - 3).Take(1).First();
  299. double freqMHz;
  300. double.TryParse(freqstr, out freqMHz);
  301. //采样率
  302. var fsstr = strs.Skip(strs.Length - 4).Take(1).First();
  303. double fsHz;
  304. double.TryParse(fsstr, out fsHz);
  305. historyFile.FreqHz = (long)((decimal)freqMHz * 1000000);
  306. historyFile.Ch = ch;
  307. historyFile.FsHz = fsHz;
  308. if (canConnected)
  309. historyFile.SatId = GetSatId(historyFile.FreqHz);
  310. if (ch == 2)
  311. {
  312. if (historyFile.SatId == 0)
  313. {
  314. LogHelper.Warning($"【任务{dto.ID}】主星无法通过MySql查询到卫星编号,使用任务中的卫星[{dto.MainSatCode}]");
  315. historyFile.SatId = dto.MainSatCode;
  316. }
  317. }
  318. else if (ch == 3)
  319. {
  320. if (historyFile.SatId == 0)
  321. {
  322. LogHelper.Warning($"【任务{dto.ID}】邻星无法通过MySql查询到卫星编号,邻星使用任务中的卫星[{dto.AdjaSatCode}]");
  323. historyFile.SatId = dto.AdjaSatCode;
  324. }
  325. }
  326. return historyFile;
  327. }
  328. int GetSatId(double freqdown)
  329. {
  330. int satId = 0;
  331. string sql = $"select 卫星ID from freguencysatid where 下行 = '{freqdown}'and 洋区 = 'I' LIMIT 1";
  332. try
  333. {
  334. var res = MySqlTools.ExecuteScalar(System.Data.CommandType.Text, sql);
  335. if (res == null || res == DBNull.Value)
  336. {
  337. LogHelper.Error($"下行频点{freqdown * 1e-6}未能从MySql中找到对应的卫星编号.SQL={sql}");
  338. }
  339. bool isInt = int.TryParse($"{res}", out satId);
  340. if (!isInt)
  341. {
  342. //System.Configuration.ConfigurationManager.ConnectionStrings["MySql"].ConnectionString
  343. LogHelper.Error($"卫星编号转换失败,obj={res}");
  344. }
  345. }
  346. catch (Exception ex)
  347. {
  348. LogHelper.Error($"下行频点{freqdown * 1e-6}MySql查询卫星编号异常.ConnectionString={System.Configuration.ConfigurationManager.ConnectionStrings["MySql"].ConnectionString}", ex);
  349. }
  350. return satId;
  351. }
  352. bool CanGetSatIdFromMySql()
  353. {
  354. string sql = $"select 卫星ID from freguencysatid where 下行 = '{0}'and 洋区 = 'I' LIMIT 1";
  355. try
  356. {
  357. var res = MySqlTools.ExecuteScalar(System.Data.CommandType.Text, sql);
  358. return true;
  359. }
  360. catch
  361. {
  362. return false;
  363. }
  364. }
  365. void StopTask(int taskID, string stopReason)
  366. {
  367. TaskStopHandleDto stopDto = new TaskStopHandleDto() { ID = taskID, StopReason = stopReason };
  368. if (string.IsNullOrWhiteSpace(stopReason))
  369. LogHelper.Info($"【任务{taskID}】正常结束");
  370. else
  371. LogHelper.Error($"【任务{taskID}】{stopReason}");
  372. var stopResp = HttpHelper.PostRequestAsync(baseUrl + "Task/StopTask", stopDto).Result;
  373. if (stopResp.code != 200)
  374. {
  375. LogHelper.Error($"【任务{taskID}】停止异常.{stopResp.msg}");
  376. return;
  377. }
  378. }
  379. public void Stop()
  380. {
  381. cts?.Cancel();
  382. }
  383. }
  384. }