produce_window.go 73 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114
  1. package service
  2. import (
  3. commonConfig "cicv-data-closedloop/aarch64/pjisuv/common/config"
  4. "cicv-data-closedloop/aarch64/pjisuv/common/service"
  5. masterConfig "cicv-data-closedloop/aarch64/pjisuv/master/config"
  6. "cicv-data-closedloop/common/config/c_log"
  7. commonEntity "cicv-data-closedloop/common/entity"
  8. "cicv-data-closedloop/common/util"
  9. "cicv-data-closedloop/pjisuv_msgs"
  10. "cicv-data-closedloop/pjisuv_ticker"
  11. "github.com/bluenviron/goroslib/v2"
  12. "github.com/bluenviron/goroslib/v2/pkg/msgs/geometry_msgs"
  13. "github.com/bluenviron/goroslib/v2/pkg/msgs/nav_msgs"
  14. "github.com/bluenviron/goroslib/v2/pkg/msgs/sensor_msgs"
  15. "github.com/bluenviron/goroslib/v2/pkg/msgs/tf2_msgs"
  16. "github.com/bluenviron/goroslib/v2/pkg/msgs/visualization_msgs"
  17. "math"
  18. "sync"
  19. "time"
  20. )
  21. type Point struct {
  22. x, y float64
  23. }
  24. var (
  25. //定义了车间四个边界点,将车间抽象成一个四边形
  26. vertices = []Point{
  27. {x: 456128.413, y: 4397847.78},
  28. {x: 456288.257, y: 4397953.51},
  29. {x: 456359.022, y: 4397822.84},
  30. {x: 456191.065, y: 4397733.3},
  31. }
  32. cicvLocationTime = time.Now()
  33. // -----------------------------共享变量
  34. //cicv_location
  35. //NumOfCicvLocation = 0
  36. AccelXSlice = []float64{}
  37. //Yowslice = make([]float64, 0)
  38. //AngleSlice = make([][]float64, 0)
  39. // /tpperception
  40. ObjDicOfTpperception = make(map[uint32][][]float32)
  41. objTypeDicOfTpperception = make(map[uint32]uint8)
  42. objSpeedDicOfTpperception = make(map[uint32]float64)
  43. // /pji_control_pub
  44. numCountPjiControlCommandOfPjControlPub int
  45. egoSteeringCmdOfPjControlPub []float64
  46. egoThrottleCmdOfPjControlPub []float64
  47. // /data_read
  48. numCountDataReadOfDataRead int
  49. egoSteeringRealOfDataRead []float64
  50. egoThrottleRealOfDataRead []float64
  51. GearPosSlice = []int16{}
  52. // --------------------------------------------------
  53. shareVars = new(sync.Map)
  54. saveTimeWindowMutex sync.Mutex // 保存时间窗口需要锁,防止数据竟态
  55. latestTimeWindowEnd = util.GetTimeCustom(time.Now())
  56. triggerInterval = 3.0 // 每个触发器3秒触发一次
  57. )
  58. // 负责监听所有主题并修改时间窗口
  59. func ProduceWindow() {
  60. c_log.GlobalLogger.Info("订阅者 goroutine,启动。")
  61. var err error
  62. subscribers := make([]*goroslib.Subscriber, masterConfig.AllTopicsNumber)
  63. subscribersTimes := make([]time.Time, masterConfig.AllTopicsNumber)
  64. subscribersTimeMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
  65. subscribersMutexes := make([]sync.Mutex, masterConfig.AllTopicsNumber)
  66. for i, topic := range masterConfig.AllTopics {
  67. for {
  68. // 定时器,区别于订阅者
  69. if topic == pjisuv_ticker.TickerTopic {
  70. // 1 把所有触发器函数执行一遍,触发器内部创建额外的定时任务goroutine
  71. for _, f := range masterConfig.RuleOfCicvTicker {
  72. f(shareVars)
  73. }
  74. // 2 创建goroutine接收定时任务触发器返回的Label和Time,执行触发逻辑
  75. go func() {
  76. for {
  77. time.Sleep(time.Duration(1)) // 降低循环频率
  78. select {
  79. case tickInfo := <-pjisuv_ticker.TickerChan:
  80. faultLabel := tickInfo.FaultLabel
  81. faultHappenTime := tickInfo.FaultHappenTime
  82. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  83. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  84. default:
  85. }
  86. }
  87. }()
  88. }
  89. create := false // 判断是否创建成功,用于打印日志
  90. // 1
  91. if topic == masterConfig.TopicOfAmrPose && (len(masterConfig.RuleOfAmrPose1) > 0 || len(masterConfig.RuleOfAmrPose3) > 0) {
  92. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  93. Node: commonConfig.RosNode,
  94. Topic: topic,
  95. Callback: func(data *visualization_msgs.MarkerArray) {
  96. subscribersTimeMutexes[i].Lock()
  97. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  98. subscribersMutexes[i].Lock()
  99. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  100. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  101. faultLabel := ""
  102. if len(masterConfig.RuleOfAmrPose1) > 0 {
  103. for _, f := range masterConfig.RuleOfAmrPose1 {
  104. faultLabel = f(data)
  105. if faultLabel != "" {
  106. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  107. subscribersTimes[i] = time.Now()
  108. goto TriggerSuccess
  109. }
  110. }
  111. }
  112. if len(masterConfig.RuleOfAmrPose3) > 0 {
  113. for _, f := range masterConfig.RuleOfAmrPose3 {
  114. faultLabel = f(shareVars, data)
  115. if faultLabel != "" {
  116. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  117. subscribersTimes[i] = time.Now()
  118. goto TriggerSuccess
  119. }
  120. }
  121. }
  122. TriggerSuccess:
  123. subscribersMutexes[i].Unlock()
  124. }
  125. subscribersTimeMutexes[i].Unlock()
  126. },
  127. })
  128. if err == nil {
  129. create = true
  130. }
  131. }
  132. // 2
  133. if topic == masterConfig.TopicOfBoundingBoxesFast &&
  134. (len(masterConfig.RuleOfBoundingBoxesFast1) > 0 ||
  135. len(masterConfig.RuleOfBoundingBoxesFast3) > 0) {
  136. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  137. Node: commonConfig.RosNode,
  138. Topic: topic,
  139. Callback: func(data *pjisuv_msgs.BoundingBoxArray) {
  140. subscribersTimeMutexes[i].Lock()
  141. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  142. subscribersMutexes[i].Lock()
  143. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  144. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  145. faultLabel := ""
  146. if len(masterConfig.RuleOfBoundingBoxesFast1) > 0 {
  147. for _, f := range masterConfig.RuleOfBoundingBoxesFast1 {
  148. faultLabel = f(data)
  149. if faultLabel != "" {
  150. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  151. subscribersTimes[i] = time.Now()
  152. goto TriggerSuccess
  153. }
  154. }
  155. }
  156. if len(masterConfig.RuleOfBoundingBoxesFast3) > 0 {
  157. for _, f := range masterConfig.RuleOfBoundingBoxesFast3 {
  158. faultLabel = f(shareVars, data)
  159. if faultLabel != "" {
  160. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  161. subscribersTimes[i] = time.Now()
  162. goto TriggerSuccess
  163. }
  164. }
  165. }
  166. TriggerSuccess:
  167. subscribersMutexes[i].Unlock()
  168. }
  169. subscribersTimeMutexes[i].Unlock()
  170. },
  171. })
  172. if err == nil {
  173. create = true
  174. }
  175. }
  176. // 3
  177. if topic == masterConfig.TopicOfCameraFault &&
  178. (len(masterConfig.RuleOfCameraFault1) > 0 ||
  179. len(masterConfig.RuleOfCameraFault3) > 0) {
  180. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  181. Node: commonConfig.RosNode,
  182. Topic: topic,
  183. Callback: func(data *pjisuv_msgs.FaultVec) {
  184. subscribersTimeMutexes[i].Lock()
  185. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  186. subscribersMutexes[i].Lock()
  187. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  188. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  189. faultLabel := ""
  190. if len(masterConfig.RuleOfCameraFault1) > 0 {
  191. for _, f := range masterConfig.RuleOfCameraFault1 {
  192. faultLabel = f(data)
  193. if faultLabel != "" {
  194. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  195. subscribersTimes[i] = time.Now()
  196. goto TriggerSuccess
  197. }
  198. }
  199. }
  200. if len(masterConfig.RuleOfCameraFault3) > 0 {
  201. for _, f := range masterConfig.RuleOfCameraFault3 {
  202. faultLabel = f(shareVars, data)
  203. if faultLabel != "" {
  204. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  205. subscribersTimes[i] = time.Now()
  206. goto TriggerSuccess
  207. }
  208. }
  209. }
  210. TriggerSuccess:
  211. subscribersMutexes[i].Unlock()
  212. }
  213. subscribersTimeMutexes[i].Unlock()
  214. },
  215. })
  216. if err == nil {
  217. create = true
  218. }
  219. }
  220. // 4
  221. if topic == masterConfig.TopicOfCanData &&
  222. (len(masterConfig.RuleOfCanData1) > 0 ||
  223. len(masterConfig.RuleOfCanData3) > 0) {
  224. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  225. Node: commonConfig.RosNode,
  226. Topic: topic,
  227. Callback: func(data *pjisuv_msgs.Frame) {
  228. subscribersTimeMutexes[i].Lock()
  229. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  230. subscribersMutexes[i].Lock()
  231. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  232. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  233. faultLabel := ""
  234. if len(masterConfig.RuleOfCanData1) > 0 {
  235. for _, f := range masterConfig.RuleOfCanData1 {
  236. faultLabel = f(data)
  237. if faultLabel != "" {
  238. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  239. subscribersTimes[i] = time.Now()
  240. goto TriggerSuccess
  241. }
  242. }
  243. }
  244. if len(masterConfig.RuleOfCanData3) > 0 {
  245. for _, f := range masterConfig.RuleOfCanData3 {
  246. faultLabel = f(shareVars, data)
  247. if faultLabel != "" {
  248. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  249. subscribersTimes[i] = time.Now()
  250. goto TriggerSuccess
  251. }
  252. }
  253. }
  254. TriggerSuccess:
  255. subscribersMutexes[i].Unlock()
  256. }
  257. subscribersTimeMutexes[i].Unlock()
  258. },
  259. })
  260. if err == nil {
  261. create = true
  262. }
  263. }
  264. // 5
  265. if topic == masterConfig.TopicOfCh128x1LslidarPointCloud &&
  266. (len(masterConfig.RuleOfCh128x1LslidarPointCloud1) > 0 ||
  267. len(masterConfig.RuleOfCh128x1LslidarPointCloud3) > 0) {
  268. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  269. Node: commonConfig.RosNode,
  270. Topic: topic,
  271. Callback: func(data *sensor_msgs.PointCloud2) {
  272. subscribersTimeMutexes[i].Lock()
  273. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  274. subscribersMutexes[i].Lock()
  275. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  276. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  277. faultLabel := ""
  278. if len(masterConfig.RuleOfCh128x1LslidarPointCloud1) > 0 {
  279. for _, f := range masterConfig.RuleOfCh128x1LslidarPointCloud1 {
  280. faultLabel = f(data)
  281. if faultLabel != "" {
  282. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  283. subscribersTimes[i] = time.Now()
  284. goto TriggerSuccess
  285. }
  286. }
  287. }
  288. if len(masterConfig.RuleOfCh128x1LslidarPointCloud3) > 0 {
  289. for _, f := range masterConfig.RuleOfCh128x1LslidarPointCloud3 {
  290. faultLabel = f(shareVars, data)
  291. if faultLabel != "" {
  292. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  293. subscribersTimes[i] = time.Now()
  294. goto TriggerSuccess
  295. }
  296. }
  297. }
  298. TriggerSuccess:
  299. subscribersMutexes[i].Unlock()
  300. }
  301. subscribersTimeMutexes[i].Unlock()
  302. },
  303. })
  304. if err == nil {
  305. create = true
  306. }
  307. }
  308. // 6
  309. if topic == masterConfig.TopicOfCh64wLLslidarPointCloud &&
  310. (len(masterConfig.RuleOfCh64wLLslidarPointCloud1) > 0 ||
  311. len(masterConfig.RuleOfCh64wLLslidarPointCloud3) > 0) {
  312. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  313. Node: commonConfig.RosNode,
  314. Topic: topic,
  315. Callback: func(data *sensor_msgs.PointCloud2) {
  316. subscribersTimeMutexes[i].Lock()
  317. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  318. subscribersMutexes[i].Lock()
  319. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  320. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  321. faultLabel := ""
  322. if len(masterConfig.RuleOfCh64wLLslidarPointCloud1) > 0 {
  323. for _, f := range masterConfig.RuleOfCh64wLLslidarPointCloud1 {
  324. faultLabel = f(data)
  325. if faultLabel != "" {
  326. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  327. subscribersTimes[i] = time.Now()
  328. goto TriggerSuccess
  329. }
  330. }
  331. }
  332. if len(masterConfig.RuleOfCh64wLLslidarPointCloud3) > 0 {
  333. for _, f := range masterConfig.RuleOfCh64wLLslidarPointCloud3 {
  334. faultLabel = f(shareVars, data)
  335. if faultLabel != "" {
  336. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  337. subscribersTimes[i] = time.Now()
  338. goto TriggerSuccess
  339. }
  340. }
  341. }
  342. TriggerSuccess:
  343. subscribersMutexes[i].Unlock()
  344. }
  345. subscribersTimeMutexes[i].Unlock()
  346. },
  347. })
  348. if err == nil {
  349. create = true
  350. }
  351. }
  352. // 7
  353. if topic == masterConfig.TopicOfCh64wLScan &&
  354. (len(masterConfig.RuleOfCh64wLScan1) > 0 ||
  355. len(masterConfig.RuleOfCh64wLScan3) > 0) {
  356. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  357. Node: commonConfig.RosNode,
  358. Topic: topic,
  359. Callback: func(data *sensor_msgs.LaserScan) {
  360. subscribersTimeMutexes[i].Lock()
  361. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  362. subscribersMutexes[i].Lock()
  363. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  364. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  365. faultLabel := ""
  366. if len(masterConfig.RuleOfCh64wLScan1) > 0 {
  367. for _, f := range masterConfig.RuleOfCh64wLScan1 {
  368. faultLabel = f(data)
  369. if faultLabel != "" {
  370. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  371. subscribersTimes[i] = time.Now()
  372. goto TriggerSuccess
  373. }
  374. }
  375. }
  376. if len(masterConfig.RuleOfCh64wLScan3) > 0 {
  377. for _, f := range masterConfig.RuleOfCh64wLScan3 {
  378. faultLabel = f(shareVars, data)
  379. if faultLabel != "" {
  380. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  381. subscribersTimes[i] = time.Now()
  382. goto TriggerSuccess
  383. }
  384. }
  385. }
  386. TriggerSuccess:
  387. subscribersMutexes[i].Unlock()
  388. }
  389. subscribersTimeMutexes[i].Unlock()
  390. },
  391. })
  392. if err == nil {
  393. create = true
  394. }
  395. }
  396. // 8
  397. if topic == masterConfig.TopicOfCh64wRLslidarPointCloud &&
  398. (len(masterConfig.RuleOfCh64wRLslidarPointCloud1) > 0 ||
  399. len(masterConfig.RuleOfCh64wRLslidarPointCloud3) > 0) {
  400. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  401. Node: commonConfig.RosNode,
  402. Topic: topic,
  403. Callback: func(data *sensor_msgs.PointCloud2) {
  404. subscribersTimeMutexes[i].Lock()
  405. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  406. subscribersMutexes[i].Lock()
  407. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  408. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  409. faultLabel := ""
  410. if len(masterConfig.RuleOfCh64wRLslidarPointCloud1) > 0 {
  411. for _, f := range masterConfig.RuleOfCh64wRLslidarPointCloud1 {
  412. faultLabel = f(data)
  413. if faultLabel != "" {
  414. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  415. subscribersTimes[i] = time.Now()
  416. goto TriggerSuccess
  417. }
  418. }
  419. }
  420. if len(masterConfig.RuleOfCh64wRLslidarPointCloud3) > 0 {
  421. for _, f := range masterConfig.RuleOfCh64wRLslidarPointCloud3 {
  422. faultLabel = f(shareVars, data)
  423. if faultLabel != "" {
  424. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  425. subscribersTimes[i] = time.Now()
  426. goto TriggerSuccess
  427. }
  428. }
  429. }
  430. TriggerSuccess:
  431. subscribersMutexes[i].Unlock()
  432. }
  433. subscribersTimeMutexes[i].Unlock()
  434. },
  435. })
  436. if err == nil {
  437. create = true
  438. }
  439. }
  440. // 9
  441. if topic == masterConfig.TopicOfCh64wRScan &&
  442. (len(masterConfig.RuleOfCh64wRScan1) > 0 ||
  443. len(masterConfig.RuleOfCh64wRScan3) > 0) {
  444. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  445. Node: commonConfig.RosNode,
  446. Topic: topic,
  447. Callback: func(data *sensor_msgs.LaserScan) {
  448. subscribersTimeMutexes[i].Lock()
  449. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  450. subscribersMutexes[i].Lock()
  451. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  452. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  453. faultLabel := ""
  454. if len(masterConfig.RuleOfCh64wRScan1) > 0 {
  455. for _, f := range masterConfig.RuleOfCh64wRScan1 {
  456. faultLabel = f(data)
  457. if faultLabel != "" {
  458. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  459. subscribersTimes[i] = time.Now()
  460. goto TriggerSuccess
  461. }
  462. }
  463. }
  464. if len(masterConfig.RuleOfCh64wRScan3) > 0 {
  465. for _, f := range masterConfig.RuleOfCh64wRScan3 {
  466. faultLabel = f(shareVars, data)
  467. if faultLabel != "" {
  468. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  469. subscribersTimes[i] = time.Now()
  470. goto TriggerSuccess
  471. }
  472. }
  473. }
  474. TriggerSuccess:
  475. subscribersMutexes[i].Unlock()
  476. }
  477. subscribersTimeMutexes[i].Unlock()
  478. },
  479. })
  480. if err == nil {
  481. create = true
  482. }
  483. }
  484. // 10
  485. if topic == masterConfig.TopicOfCicvLidarclusterMovingObjects &&
  486. (len(masterConfig.RuleOfCicvLidarclusterMovingObjects1) > 0 ||
  487. len(masterConfig.RuleOfCicvLidarclusterMovingObjects3) > 0) {
  488. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  489. Node: commonConfig.RosNode,
  490. Topic: topic,
  491. Callback: func(data *pjisuv_msgs.PerceptionCicvMovingObjects) {
  492. subscribersTimeMutexes[i].Lock()
  493. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  494. subscribersMutexes[i].Lock()
  495. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  496. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  497. faultLabel := ""
  498. if len(masterConfig.RuleOfCicvLidarclusterMovingObjects1) > 0 {
  499. for _, f := range masterConfig.RuleOfCicvLidarclusterMovingObjects1 {
  500. faultLabel = f(data)
  501. if faultLabel != "" {
  502. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  503. subscribersTimes[i] = time.Now()
  504. goto TriggerSuccess
  505. }
  506. }
  507. }
  508. if len(masterConfig.RuleOfCicvLidarclusterMovingObjects3) > 0 {
  509. for _, f := range masterConfig.RuleOfCicvLidarclusterMovingObjects3 {
  510. faultLabel = f(shareVars, data)
  511. if faultLabel != "" {
  512. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  513. subscribersTimes[i] = time.Now()
  514. goto TriggerSuccess
  515. }
  516. }
  517. }
  518. TriggerSuccess:
  519. subscribersMutexes[i].Unlock()
  520. }
  521. subscribersTimeMutexes[i].Unlock()
  522. },
  523. })
  524. if err == nil {
  525. create = true
  526. }
  527. }
  528. // 11 有共享变量的订阅者必须被创建
  529. if topic == masterConfig.TopicOfCicvAmrTrajectory {
  530. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  531. Node: commonConfig.RosNode,
  532. Topic: topic,
  533. Callback: func(data *pjisuv_msgs.Trajectory) {
  534. subscribersTimeMutexes[i].Lock()
  535. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  536. subscribersMutexes[i].Lock()
  537. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  538. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  539. faultLabel := ""
  540. if len(masterConfig.RuleOfCicvAmrTrajectory1) > 0 {
  541. for _, f := range masterConfig.RuleOfCicvAmrTrajectory1 {
  542. faultLabel = f(data)
  543. if faultLabel != "" {
  544. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  545. subscribersTimes[i] = time.Now()
  546. goto TriggerSuccess
  547. }
  548. }
  549. }
  550. if len(masterConfig.RuleOfCicvAmrTrajectory3) > 0 {
  551. for _, f := range masterConfig.RuleOfCicvAmrTrajectory3 {
  552. faultLabel = f(shareVars, data)
  553. if faultLabel != "" {
  554. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  555. subscribersTimes[i] = time.Now()
  556. goto TriggerSuccess
  557. }
  558. }
  559. }
  560. TriggerSuccess:
  561. subscribersMutexes[i].Unlock()
  562. }
  563. subscribersTimeMutexes[i].Unlock()
  564. // 更新共享变量
  565. currentCurvateres := make([]float64, 0)
  566. for _, point := range data.Trajectoryinfo.Trajectorypoints {
  567. currentCurvateres = append(currentCurvateres, math.Abs(float64(point.Curvature)))
  568. }
  569. shareVars.Store("LastCurvaturesOfCicvAmrTrajectory", currentCurvateres)
  570. shareVars.Store("DecisionType", data.Trajectoryinfo.DecisionType)
  571. },
  572. })
  573. if err == nil {
  574. create = true
  575. }
  576. }
  577. // 12 有共享变量的订阅者必须被创建
  578. if topic == masterConfig.TopicOfCicvLocation {
  579. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  580. Node: commonConfig.RosNode,
  581. Topic: topic,
  582. Callback: func(data *pjisuv_msgs.PerceptionLocalization) {
  583. subscribersTimeMutexes[i].Lock()
  584. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  585. subscribersMutexes[i].Lock()
  586. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  587. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  588. faultLabel := ""
  589. if len(masterConfig.RuleOfCicvLocation1) > 0 {
  590. for _, f := range masterConfig.RuleOfCicvLocation1 {
  591. faultLabel = f(data)
  592. if faultLabel != "" {
  593. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  594. subscribersTimes[i] = time.Now()
  595. goto TriggerSuccess
  596. }
  597. }
  598. }
  599. if len(masterConfig.RuleOfCicvLocation3) > 0 {
  600. for _, f := range masterConfig.RuleOfCicvLocation3 {
  601. faultLabel = f(shareVars, data)
  602. if faultLabel != "" {
  603. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  604. subscribersTimes[i] = time.Now()
  605. goto TriggerSuccess
  606. }
  607. }
  608. }
  609. TriggerSuccess:
  610. subscribersMutexes[i].Unlock()
  611. }
  612. subscribersTimeMutexes[i].Unlock()
  613. // 更新共享变量
  614. AbsSpeed := math.Sqrt(math.Pow(data.VelocityX, 2) + math.Pow(data.VelocityY, 2))
  615. AccelXSlice = append(AccelXSlice, data.AccelX)
  616. shareVars.Store("AbsSpeed", AbsSpeed)
  617. shareVars.Store("AccelXSlice", AccelXSlice)
  618. shareVars.Store("VelocityXOfCicvLocation", data.VelocityX)
  619. shareVars.Store("VelocityYOfCicvLocation", data.VelocityY)
  620. shareVars.Store("VelocityZOfCicvLocation", data.VelocityZ)
  621. shareVars.Store("YawOfCicvLocation", data.Yaw)
  622. shareVars.Store("AngularVelocityZOfCicvLocation", data.AngularVelocityZ)
  623. shareVars.Store("PositionXOfCicvLocation", data.PositionX)
  624. shareVars.Store("PositionYOfCicvLocation", data.PositionY)
  625. shareVars.Store("Latitude", data.Latitude)
  626. shareVars.Store("Longitude", data.Longitude)
  627. /*用于 陡坡-DescendingSteepHill/ClimbingSteepHill触发器
  628. if NumOfCicvLocation%10==0{
  629. AngleSlice[0] = append(AngleSlice[0], data.Qx)
  630. AngleSlice[1] = append(AngleSlice[1], data.Qy)
  631. AngleSlice[2] = append(AngleSlice[2], data.Qz)
  632. AngleSlice[3] = append(AngleSlice[3], data.Qw)
  633. shareVars.Store("AngleSlice", AngleSlice)
  634. NumOfCicvLocation=1
  635. }
  636. NumOfCicvLocation++
  637. */
  638. /*用于 掉头TurnAround触发器
  639. Yowslice = append(Yowslice, data.Yaw)
  640. if len(Yowslice) >= 600 {
  641. Yowslice = Yowslice[1:]
  642. }
  643. shareVars.Store("Yowslice", Yowslice)
  644. */
  645. // 用于判断是否在车间内
  646. if time.Since(cicvLocationTime).Seconds() > 1 {
  647. p := Point{x: data.PositionX, y: data.PositionY}
  648. OutsideWorkshopFlag := isPointInPolygon(p, vertices) //在车间返回0,不在车间返回1
  649. shareVars.Store("OutsideWorkshopFlag", OutsideWorkshopFlag)
  650. cicvLocationTime = time.Now()
  651. }
  652. },
  653. })
  654. if err == nil {
  655. create = true
  656. }
  657. }
  658. // 13
  659. if topic == masterConfig.TopicOfCloudClusters &&
  660. (len(masterConfig.RuleOfCloudClusters1) > 0 ||
  661. len(masterConfig.RuleOfCloudClusters3) > 0) {
  662. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  663. Node: commonConfig.RosNode,
  664. Topic: topic,
  665. Callback: func(data *pjisuv_msgs.AutowareCloudClusterArray) {
  666. subscribersTimeMutexes[i].Lock()
  667. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  668. subscribersMutexes[i].Lock()
  669. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  670. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  671. faultLabel := ""
  672. if len(masterConfig.RuleOfCloudClusters1) > 0 {
  673. for _, f := range masterConfig.RuleOfCloudClusters1 {
  674. faultLabel = f(data)
  675. if faultLabel != "" {
  676. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  677. subscribersTimes[i] = time.Now()
  678. goto TriggerSuccess
  679. }
  680. }
  681. }
  682. if len(masterConfig.RuleOfCloudClusters3) > 0 {
  683. for _, f := range masterConfig.RuleOfCloudClusters3 {
  684. faultLabel = f(shareVars, data)
  685. if faultLabel != "" {
  686. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  687. subscribersTimes[i] = time.Now()
  688. goto TriggerSuccess
  689. }
  690. }
  691. }
  692. TriggerSuccess:
  693. subscribersMutexes[i].Unlock()
  694. }
  695. subscribersTimeMutexes[i].Unlock()
  696. },
  697. })
  698. if err == nil {
  699. create = true
  700. }
  701. }
  702. // 14
  703. if topic == masterConfig.TopicOfHeartbeatInfo &&
  704. (len(masterConfig.RuleOfHeartbeatInfo1) > 0 ||
  705. len(masterConfig.RuleOfHeartbeatInfo3) > 0) {
  706. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  707. Node: commonConfig.RosNode,
  708. Topic: topic,
  709. Callback: func(data *pjisuv_msgs.HeartBeatInfo) {
  710. subscribersTimeMutexes[i].Lock()
  711. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  712. subscribersMutexes[i].Lock()
  713. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  714. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  715. faultLabel := ""
  716. if len(masterConfig.RuleOfHeartbeatInfo1) > 0 {
  717. for _, f := range masterConfig.RuleOfHeartbeatInfo1 {
  718. faultLabel = f(data)
  719. if faultLabel != "" {
  720. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  721. subscribersTimes[i] = time.Now()
  722. goto TriggerSuccess
  723. }
  724. }
  725. }
  726. if len(masterConfig.RuleOfHeartbeatInfo3) > 0 {
  727. for _, f := range masterConfig.RuleOfHeartbeatInfo3 {
  728. faultLabel = f(shareVars, data)
  729. if faultLabel != "" {
  730. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  731. subscribersTimes[i] = time.Now()
  732. goto TriggerSuccess
  733. }
  734. }
  735. }
  736. TriggerSuccess:
  737. subscribersMutexes[i].Unlock()
  738. }
  739. subscribersTimeMutexes[i].Unlock()
  740. },
  741. })
  742. if err == nil {
  743. create = true
  744. }
  745. }
  746. // 15
  747. if topic == masterConfig.TopicOfLidarPretreatmentCost &&
  748. (len(masterConfig.RuleOfLidarPretreatmentCost1) > 0 ||
  749. len(masterConfig.RuleOfLidarPretreatmentCost3) > 0) {
  750. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  751. Node: commonConfig.RosNode,
  752. Topic: topic,
  753. Callback: func(data *geometry_msgs.Vector3Stamped) {
  754. subscribersTimeMutexes[i].Lock()
  755. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  756. subscribersMutexes[i].Lock()
  757. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  758. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  759. faultLabel := ""
  760. if len(masterConfig.RuleOfLidarPretreatmentCost1) > 0 {
  761. for _, f := range masterConfig.RuleOfLidarPretreatmentCost1 {
  762. faultLabel = f(data)
  763. if faultLabel != "" {
  764. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  765. subscribersTimes[i] = time.Now()
  766. goto TriggerSuccess
  767. }
  768. }
  769. }
  770. if len(masterConfig.RuleOfLidarPretreatmentCost3) > 0 {
  771. for _, f := range masterConfig.RuleOfLidarPretreatmentCost3 {
  772. faultLabel = f(shareVars, data)
  773. if faultLabel != "" {
  774. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  775. subscribersTimes[i] = time.Now()
  776. goto TriggerSuccess
  777. }
  778. }
  779. }
  780. TriggerSuccess:
  781. subscribersMutexes[i].Unlock()
  782. }
  783. subscribersTimeMutexes[i].Unlock()
  784. },
  785. })
  786. if err == nil {
  787. create = true
  788. }
  789. }
  790. // 16
  791. if topic == masterConfig.TopicOfLidarPretreatmentOdometry &&
  792. (len(masterConfig.RuleOfLidarPretreatmentOdometry1) > 0 ||
  793. len(masterConfig.RuleOfLidarPretreatmentOdometry3) > 0) {
  794. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  795. Node: commonConfig.RosNode,
  796. Topic: topic,
  797. Callback: func(data *nav_msgs.Odometry) {
  798. subscribersTimeMutexes[i].Lock()
  799. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  800. subscribersMutexes[i].Lock()
  801. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  802. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  803. faultLabel := ""
  804. if len(masterConfig.RuleOfLidarPretreatmentOdometry1) > 0 {
  805. for _, f := range masterConfig.RuleOfLidarPretreatmentOdometry1 {
  806. faultLabel = f(data)
  807. if faultLabel != "" {
  808. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  809. subscribersTimes[i] = time.Now()
  810. goto TriggerSuccess
  811. }
  812. }
  813. }
  814. if len(masterConfig.RuleOfLidarPretreatmentOdometry3) > 0 {
  815. for _, f := range masterConfig.RuleOfLidarPretreatmentOdometry3 {
  816. faultLabel = f(shareVars, data)
  817. if faultLabel != "" {
  818. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  819. subscribersTimes[i] = time.Now()
  820. goto TriggerSuccess
  821. }
  822. }
  823. }
  824. TriggerSuccess:
  825. subscribersMutexes[i].Unlock()
  826. }
  827. subscribersTimeMutexes[i].Unlock()
  828. },
  829. })
  830. if err == nil {
  831. create = true
  832. }
  833. }
  834. // 17
  835. if topic == masterConfig.TopicOfLidarRoi &&
  836. (len(masterConfig.RuleOfLidarRoi1) > 0 ||
  837. len(masterConfig.RuleOfLidarRoi3) > 0) {
  838. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  839. Node: commonConfig.RosNode,
  840. Topic: topic,
  841. Callback: func(data *geometry_msgs.PolygonStamped) {
  842. subscribersTimeMutexes[i].Lock()
  843. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  844. subscribersMutexes[i].Lock()
  845. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  846. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  847. faultLabel := ""
  848. if len(masterConfig.RuleOfLidarRoi1) > 0 {
  849. for _, f := range masterConfig.RuleOfLidarRoi1 {
  850. faultLabel = f(data)
  851. if faultLabel != "" {
  852. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  853. subscribersTimes[i] = time.Now()
  854. goto TriggerSuccess
  855. }
  856. }
  857. }
  858. if len(masterConfig.RuleOfLidarRoi3) > 0 {
  859. for _, f := range masterConfig.RuleOfLidarRoi3 {
  860. faultLabel = f(shareVars, data)
  861. if faultLabel != "" {
  862. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  863. subscribersTimes[i] = time.Now()
  864. goto TriggerSuccess
  865. }
  866. }
  867. }
  868. TriggerSuccess:
  869. subscribersMutexes[i].Unlock()
  870. }
  871. subscribersTimeMutexes[i].Unlock()
  872. },
  873. })
  874. if err == nil {
  875. create = true
  876. }
  877. }
  878. // 18
  879. if topic == masterConfig.TopicOfLine1 &&
  880. (len(masterConfig.RuleOfLine11) > 0 ||
  881. len(masterConfig.RuleOfLine13) > 0) {
  882. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  883. Node: commonConfig.RosNode,
  884. Topic: topic,
  885. Callback: func(data *nav_msgs.Path) {
  886. subscribersTimeMutexes[i].Lock()
  887. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  888. subscribersMutexes[i].Lock()
  889. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  890. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  891. faultLabel := ""
  892. if len(masterConfig.RuleOfLine11) > 0 {
  893. for _, f := range masterConfig.RuleOfLine11 {
  894. faultLabel = f(data)
  895. if faultLabel != "" {
  896. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  897. subscribersTimes[i] = time.Now()
  898. goto TriggerSuccess
  899. }
  900. }
  901. }
  902. if len(masterConfig.RuleOfLine13) > 0 {
  903. for _, f := range masterConfig.RuleOfLine13 {
  904. faultLabel = f(shareVars, data)
  905. if faultLabel != "" {
  906. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  907. subscribersTimes[i] = time.Now()
  908. goto TriggerSuccess
  909. }
  910. }
  911. }
  912. TriggerSuccess:
  913. subscribersMutexes[i].Unlock()
  914. }
  915. subscribersTimeMutexes[i].Unlock()
  916. },
  917. })
  918. if err == nil {
  919. create = true
  920. }
  921. }
  922. // 19
  923. if topic == masterConfig.TopicOfLine2 &&
  924. (len(masterConfig.RuleOfLine21) > 0 ||
  925. len(masterConfig.RuleOfLine23) > 0) {
  926. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  927. Node: commonConfig.RosNode,
  928. Topic: topic,
  929. Callback: func(data *nav_msgs.Path) {
  930. subscribersTimeMutexes[i].Lock()
  931. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  932. subscribersMutexes[i].Lock()
  933. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  934. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  935. faultLabel := ""
  936. if len(masterConfig.RuleOfLine21) > 0 {
  937. for _, f := range masterConfig.RuleOfLine21 {
  938. faultLabel = f(data)
  939. if faultLabel != "" {
  940. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  941. subscribersTimes[i] = time.Now()
  942. goto TriggerSuccess
  943. }
  944. }
  945. }
  946. if len(masterConfig.RuleOfLine23) > 0 {
  947. for _, f := range masterConfig.RuleOfLine23 {
  948. faultLabel = f(shareVars, data)
  949. if faultLabel != "" {
  950. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  951. subscribersTimes[i] = time.Now()
  952. goto TriggerSuccess
  953. }
  954. }
  955. }
  956. TriggerSuccess:
  957. subscribersMutexes[i].Unlock()
  958. }
  959. subscribersTimeMutexes[i].Unlock()
  960. },
  961. })
  962. if err == nil {
  963. create = true
  964. }
  965. }
  966. // 20
  967. if topic == masterConfig.TopicOfMapPolygon &&
  968. (len(masterConfig.RuleOfMapPolygon1) > 0 ||
  969. len(masterConfig.RuleOfMapPolygon3) > 0) {
  970. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  971. Node: commonConfig.RosNode,
  972. Topic: topic,
  973. Callback: func(data *pjisuv_msgs.PolygonStamped) {
  974. subscribersTimeMutexes[i].Lock()
  975. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  976. subscribersMutexes[i].Lock()
  977. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  978. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  979. faultLabel := ""
  980. if len(masterConfig.RuleOfMapPolygon1) > 0 {
  981. for _, f := range masterConfig.RuleOfMapPolygon1 {
  982. faultLabel = f(data)
  983. if faultLabel != "" {
  984. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  985. subscribersTimes[i] = time.Now()
  986. goto TriggerSuccess
  987. }
  988. }
  989. }
  990. if len(masterConfig.RuleOfMapPolygon3) > 0 {
  991. for _, f := range masterConfig.RuleOfMapPolygon3 {
  992. faultLabel = f(shareVars, data)
  993. if faultLabel != "" {
  994. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  995. subscribersTimes[i] = time.Now()
  996. goto TriggerSuccess
  997. }
  998. }
  999. }
  1000. TriggerSuccess:
  1001. subscribersMutexes[i].Unlock()
  1002. }
  1003. subscribersTimeMutexes[i].Unlock()
  1004. },
  1005. })
  1006. if err == nil {
  1007. create = true
  1008. }
  1009. }
  1010. // 21
  1011. if topic == masterConfig.TopicOfObstacleDisplay &&
  1012. (len(masterConfig.RuleOfObstacleDisplay1) > 0 ||
  1013. len(masterConfig.RuleOfObstacleDisplay3) > 0) {
  1014. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1015. Node: commonConfig.RosNode,
  1016. Topic: topic,
  1017. Callback: func(data *visualization_msgs.MarkerArray) {
  1018. subscribersTimeMutexes[i].Lock()
  1019. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1020. subscribersMutexes[i].Lock()
  1021. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1022. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1023. faultLabel := ""
  1024. if len(masterConfig.RuleOfObstacleDisplay1) > 0 {
  1025. for _, f := range masterConfig.RuleOfObstacleDisplay1 {
  1026. faultLabel = f(data)
  1027. if faultLabel != "" {
  1028. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1029. subscribersTimes[i] = time.Now()
  1030. goto TriggerSuccess
  1031. }
  1032. }
  1033. }
  1034. if len(masterConfig.RuleOfObstacleDisplay3) > 0 {
  1035. for _, f := range masterConfig.RuleOfObstacleDisplay3 {
  1036. faultLabel = f(shareVars, data)
  1037. if faultLabel != "" {
  1038. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1039. subscribersTimes[i] = time.Now()
  1040. goto TriggerSuccess
  1041. }
  1042. }
  1043. }
  1044. TriggerSuccess:
  1045. subscribersMutexes[i].Unlock()
  1046. }
  1047. subscribersTimeMutexes[i].Unlock()
  1048. },
  1049. })
  1050. if err == nil {
  1051. create = true
  1052. }
  1053. }
  1054. // 22 有共享变量的订阅者必须被创建
  1055. if topic == masterConfig.TopicOfPjControlPub {
  1056. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1057. Node: commonConfig.RosNode,
  1058. Topic: topic,
  1059. Callback: func(data *pjisuv_msgs.CommonVehicleCmd) {
  1060. subscribersTimeMutexes[i].Lock()
  1061. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1062. subscribersMutexes[i].Lock()
  1063. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1064. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1065. faultLabel := ""
  1066. if len(masterConfig.RuleOfPjControlPub1) > 0 {
  1067. for _, f := range masterConfig.RuleOfPjControlPub1 {
  1068. faultLabel = f(data)
  1069. if faultLabel != "" {
  1070. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1071. subscribersTimes[i] = time.Now()
  1072. goto TriggerSuccess
  1073. }
  1074. }
  1075. }
  1076. if len(masterConfig.RuleOfPjControlPub3) > 0 {
  1077. for _, f := range masterConfig.RuleOfPjControlPub3 {
  1078. faultLabel = f(shareVars, data)
  1079. if faultLabel != "" {
  1080. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1081. subscribersTimes[i] = time.Now()
  1082. goto TriggerSuccess
  1083. }
  1084. }
  1085. }
  1086. TriggerSuccess:
  1087. subscribersMutexes[i].Unlock()
  1088. }
  1089. subscribersTimeMutexes[i].Unlock()
  1090. // 更新共享变量
  1091. numCountPjiControlCommandOfPjControlPub++
  1092. if numCountPjiControlCommandOfPjControlPub == 10 {
  1093. egoSteeringCmdOfPjControlPub = append(egoSteeringCmdOfPjControlPub, data.ICPVCmdStrAngle)
  1094. egoThrottleCmdOfPjControlPub = append(egoThrottleCmdOfPjControlPub, data.ICPVCmdAccPelPosAct)
  1095. numCountPjiControlCommandOfPjControlPub = 0
  1096. }
  1097. shareVars.Store("NumCountPjiControlCommandOfPjControlPub", numCountPjiControlCommandOfPjControlPub)
  1098. shareVars.Store("EgoSteeringCmdOfPjControlPub", egoSteeringCmdOfPjControlPub)
  1099. shareVars.Store("EgoThrottleCmdOfPjControlPub", egoThrottleCmdOfPjControlPub)
  1100. },
  1101. })
  1102. if err == nil {
  1103. create = true
  1104. }
  1105. }
  1106. // 23
  1107. if topic == masterConfig.TopicOfPointsCluster &&
  1108. (len(masterConfig.RuleOfPointsCluster1) > 0 ||
  1109. len(masterConfig.RuleOfPointsCluster3) > 0) {
  1110. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1111. Node: commonConfig.RosNode,
  1112. Topic: topic,
  1113. Callback: func(data *sensor_msgs.PointCloud2) {
  1114. subscribersTimeMutexes[i].Lock()
  1115. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1116. subscribersMutexes[i].Lock()
  1117. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1118. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1119. faultLabel := ""
  1120. if len(masterConfig.RuleOfPointsCluster1) > 0 {
  1121. for _, f := range masterConfig.RuleOfPointsCluster1 {
  1122. faultLabel = f(data)
  1123. if faultLabel != "" {
  1124. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1125. subscribersTimes[i] = time.Now()
  1126. goto TriggerSuccess
  1127. }
  1128. }
  1129. }
  1130. if len(masterConfig.RuleOfPointsCluster3) > 0 {
  1131. for _, f := range masterConfig.RuleOfPointsCluster3 {
  1132. faultLabel = f(shareVars, data)
  1133. if faultLabel != "" {
  1134. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1135. subscribersTimes[i] = time.Now()
  1136. goto TriggerSuccess
  1137. }
  1138. }
  1139. }
  1140. TriggerSuccess:
  1141. subscribersMutexes[i].Unlock()
  1142. }
  1143. subscribersTimeMutexes[i].Unlock()
  1144. },
  1145. })
  1146. if err == nil {
  1147. create = true
  1148. }
  1149. }
  1150. // 24
  1151. if topic == masterConfig.TopicOfPointsConcat &&
  1152. (len(masterConfig.RuleOfPointsConcat1) > 0 ||
  1153. len(masterConfig.RuleOfPointsConcat3) > 0) {
  1154. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1155. Node: commonConfig.RosNode,
  1156. Topic: topic,
  1157. Callback: func(data *sensor_msgs.PointCloud2) {
  1158. subscribersTimeMutexes[i].Lock()
  1159. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1160. subscribersMutexes[i].Lock()
  1161. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1162. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1163. faultLabel := ""
  1164. if len(masterConfig.RuleOfPointsConcat1) > 0 {
  1165. for _, f := range masterConfig.RuleOfPointsConcat1 {
  1166. faultLabel = f(data)
  1167. if faultLabel != "" {
  1168. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1169. subscribersTimes[i] = time.Now()
  1170. goto TriggerSuccess
  1171. }
  1172. }
  1173. }
  1174. if len(masterConfig.RuleOfPointsConcat3) > 0 {
  1175. for _, f := range masterConfig.RuleOfPointsConcat3 {
  1176. faultLabel = f(shareVars, data)
  1177. if faultLabel != "" {
  1178. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1179. subscribersTimes[i] = time.Now()
  1180. goto TriggerSuccess
  1181. }
  1182. }
  1183. }
  1184. TriggerSuccess:
  1185. subscribersMutexes[i].Unlock()
  1186. }
  1187. subscribersTimeMutexes[i].Unlock()
  1188. },
  1189. })
  1190. if err == nil {
  1191. create = true
  1192. }
  1193. }
  1194. // 25
  1195. if topic == masterConfig.TopicOfReferenceDisplay &&
  1196. (len(masterConfig.RuleOfReferenceDisplay1) > 0 ||
  1197. len(masterConfig.RuleOfReferenceDisplay3) > 0) {
  1198. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1199. Node: commonConfig.RosNode,
  1200. Topic: topic,
  1201. Callback: func(data *nav_msgs.Path) {
  1202. subscribersTimeMutexes[i].Lock()
  1203. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1204. subscribersMutexes[i].Lock()
  1205. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1206. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1207. faultLabel := ""
  1208. if len(masterConfig.RuleOfReferenceDisplay1) > 0 {
  1209. for _, f := range masterConfig.RuleOfReferenceDisplay1 {
  1210. faultLabel = f(data)
  1211. if faultLabel != "" {
  1212. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1213. subscribersTimes[i] = time.Now()
  1214. goto TriggerSuccess
  1215. }
  1216. }
  1217. }
  1218. if len(masterConfig.RuleOfReferenceDisplay3) > 0 {
  1219. for _, f := range masterConfig.RuleOfReferenceDisplay3 {
  1220. faultLabel = f(shareVars, data)
  1221. if faultLabel != "" {
  1222. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1223. subscribersTimes[i] = time.Now()
  1224. goto TriggerSuccess
  1225. }
  1226. }
  1227. }
  1228. TriggerSuccess:
  1229. subscribersMutexes[i].Unlock()
  1230. }
  1231. subscribersTimeMutexes[i].Unlock()
  1232. },
  1233. })
  1234. if err == nil {
  1235. create = true
  1236. }
  1237. }
  1238. // 26
  1239. if topic == masterConfig.TopicOfReferenceTrajectory &&
  1240. (len(masterConfig.RuleOfReferenceTrajectory1) > 0 ||
  1241. len(masterConfig.RuleOfReferenceTrajectory3) > 0) {
  1242. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1243. Node: commonConfig.RosNode,
  1244. Topic: topic,
  1245. Callback: func(data *pjisuv_msgs.Trajectory) {
  1246. subscribersTimeMutexes[i].Lock()
  1247. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1248. subscribersMutexes[i].Lock()
  1249. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1250. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1251. faultLabel := ""
  1252. if len(masterConfig.RuleOfReferenceTrajectory1) > 0 {
  1253. for _, f := range masterConfig.RuleOfReferenceTrajectory1 {
  1254. faultLabel = f(data)
  1255. if faultLabel != "" {
  1256. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1257. subscribersTimes[i] = time.Now()
  1258. goto TriggerSuccess
  1259. }
  1260. }
  1261. }
  1262. if len(masterConfig.RuleOfReferenceTrajectory3) > 0 {
  1263. for _, f := range masterConfig.RuleOfReferenceTrajectory3 {
  1264. faultLabel = f(shareVars, data)
  1265. if faultLabel != "" {
  1266. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1267. subscribersTimes[i] = time.Now()
  1268. goto TriggerSuccess
  1269. }
  1270. }
  1271. }
  1272. TriggerSuccess:
  1273. subscribersMutexes[i].Unlock()
  1274. }
  1275. subscribersTimeMutexes[i].Unlock()
  1276. },
  1277. })
  1278. if err == nil {
  1279. create = true
  1280. }
  1281. }
  1282. // 27
  1283. if topic == masterConfig.TopicOfRoiPoints &&
  1284. (len(masterConfig.RuleOfRoiPoints1) > 0 ||
  1285. len(masterConfig.RuleOfRoiPoints3) > 0) {
  1286. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1287. Node: commonConfig.RosNode,
  1288. Topic: topic,
  1289. Callback: func(data *sensor_msgs.PointCloud2) {
  1290. subscribersTimeMutexes[i].Lock()
  1291. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1292. subscribersMutexes[i].Lock()
  1293. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1294. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1295. faultLabel := ""
  1296. if len(masterConfig.RuleOfRoiPoints1) > 0 {
  1297. for _, f := range masterConfig.RuleOfRoiPoints1 {
  1298. faultLabel = f(data)
  1299. if faultLabel != "" {
  1300. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1301. subscribersTimes[i] = time.Now()
  1302. goto TriggerSuccess
  1303. }
  1304. }
  1305. }
  1306. if len(masterConfig.RuleOfRoiPoints3) > 0 {
  1307. for _, f := range masterConfig.RuleOfRoiPoints3 {
  1308. faultLabel = f(shareVars, data)
  1309. if faultLabel != "" {
  1310. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1311. subscribersTimes[i] = time.Now()
  1312. goto TriggerSuccess
  1313. }
  1314. }
  1315. }
  1316. TriggerSuccess:
  1317. subscribersMutexes[i].Unlock()
  1318. }
  1319. subscribersTimeMutexes[i].Unlock()
  1320. },
  1321. })
  1322. if err == nil {
  1323. create = true
  1324. }
  1325. }
  1326. // 28
  1327. if topic == masterConfig.TopicOfRoiPolygon &&
  1328. (len(masterConfig.RuleOfRoiPolygon1) > 0 ||
  1329. len(masterConfig.RuleOfRoiPolygon3) > 0) {
  1330. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1331. Node: commonConfig.RosNode,
  1332. Topic: topic,
  1333. Callback: func(data *nav_msgs.Path) {
  1334. subscribersTimeMutexes[i].Lock()
  1335. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1336. subscribersMutexes[i].Lock()
  1337. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1338. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1339. faultLabel := ""
  1340. if len(masterConfig.RuleOfRoiPolygon1) > 0 {
  1341. for _, f := range masterConfig.RuleOfRoiPolygon1 {
  1342. faultLabel = f(data)
  1343. if faultLabel != "" {
  1344. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1345. subscribersTimes[i] = time.Now()
  1346. goto TriggerSuccess
  1347. }
  1348. }
  1349. }
  1350. if len(masterConfig.RuleOfRoiPolygon3) > 0 {
  1351. for _, f := range masterConfig.RuleOfRoiPolygon3 {
  1352. faultLabel = f(shareVars, data)
  1353. if faultLabel != "" {
  1354. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1355. subscribersTimes[i] = time.Now()
  1356. goto TriggerSuccess
  1357. }
  1358. }
  1359. }
  1360. TriggerSuccess:
  1361. subscribersMutexes[i].Unlock()
  1362. }
  1363. subscribersTimeMutexes[i].Unlock()
  1364. },
  1365. })
  1366. if err == nil {
  1367. create = true
  1368. }
  1369. }
  1370. // 29
  1371. if topic == masterConfig.TopicOfTf &&
  1372. (len(masterConfig.RuleOfTf1) > 0 ||
  1373. len(masterConfig.RuleOfTf3) > 0) {
  1374. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1375. Node: commonConfig.RosNode,
  1376. Topic: topic,
  1377. Callback: func(data *tf2_msgs.TFMessage) {
  1378. subscribersTimeMutexes[i].Lock()
  1379. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1380. subscribersMutexes[i].Lock()
  1381. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1382. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1383. faultLabel := ""
  1384. if len(masterConfig.RuleOfTf1) > 0 {
  1385. for _, f := range masterConfig.RuleOfTf1 {
  1386. faultLabel = f(data)
  1387. if faultLabel != "" {
  1388. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1389. subscribersTimes[i] = time.Now()
  1390. goto TriggerSuccess
  1391. }
  1392. }
  1393. }
  1394. if len(masterConfig.RuleOfTf3) > 0 {
  1395. for _, f := range masterConfig.RuleOfTf3 {
  1396. faultLabel = f(shareVars, data)
  1397. if faultLabel != "" {
  1398. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1399. subscribersTimes[i] = time.Now()
  1400. goto TriggerSuccess
  1401. }
  1402. }
  1403. }
  1404. TriggerSuccess:
  1405. subscribersMutexes[i].Unlock()
  1406. }
  1407. subscribersTimeMutexes[i].Unlock()
  1408. },
  1409. })
  1410. if err == nil {
  1411. create = true
  1412. }
  1413. }
  1414. // 30 有共享变量的订阅者必须被创建
  1415. if topic == masterConfig.TopicOfTpperception {
  1416. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1417. Node: commonConfig.RosNode,
  1418. Topic: topic,
  1419. Callback: func(data *pjisuv_msgs.PerceptionObjects) {
  1420. subscribersTimeMutexes[i].Lock()
  1421. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1422. subscribersMutexes[i].Lock()
  1423. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1424. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1425. faultLabel := ""
  1426. if len(masterConfig.RuleOfTpperception1) > 0 {
  1427. for _, f := range masterConfig.RuleOfTpperception1 {
  1428. faultLabel = f(data)
  1429. if faultLabel != "" {
  1430. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1431. subscribersTimes[i] = time.Now()
  1432. goto TriggerSuccess
  1433. }
  1434. }
  1435. }
  1436. if len(masterConfig.RuleOfTpperception3) > 0 {
  1437. for _, f := range masterConfig.RuleOfTpperception3 {
  1438. faultLabel = f(shareVars, data)
  1439. if faultLabel != "" {
  1440. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1441. subscribersTimes[i] = time.Now()
  1442. goto TriggerSuccess
  1443. }
  1444. }
  1445. }
  1446. TriggerSuccess:
  1447. subscribersMutexes[i].Unlock()
  1448. }
  1449. subscribersTimeMutexes[i].Unlock()
  1450. // 更新共享变量
  1451. for _, obj := range data.Objs {
  1452. if math.Abs(float64(obj.X)) >= 30 || math.Abs(float64(obj.Y)) >= 30 {
  1453. continue
  1454. }
  1455. if _, ok := ObjDicOfTpperception[obj.Id]; !ok {
  1456. ObjDicOfTpperception[obj.Id] = [][]float32{{}, {}, {}, {}, {}}
  1457. }
  1458. ObjDicOfTpperception[obj.Id][0] = append(ObjDicOfTpperception[obj.Id][0], obj.X)
  1459. ObjDicOfTpperception[obj.Id][1] = append(ObjDicOfTpperception[obj.Id][1], obj.Y)
  1460. ObjDicOfTpperception[obj.Id][2] = append(ObjDicOfTpperception[obj.Id][2], obj.Vxrel)
  1461. absspeed := math.Sqrt(math.Pow(float64(obj.Vxabs), 2) + math.Pow(float64(obj.Vyabs), 2))
  1462. ObjDicOfTpperception[obj.Id][3] = append(ObjDicOfTpperception[obj.Id][3], float32(absspeed))
  1463. ObjDicOfTpperception[obj.Id][4] = append(ObjDicOfTpperception[obj.Id][4], obj.Heading)
  1464. objTypeDicOfTpperception[obj.Id] = obj.Type
  1465. objSpeedDicOfTpperception[obj.Id] = math.Pow(math.Pow(float64(obj.Vxabs), 2)+math.Pow(float64(obj.Vyabs), 2), 0.5)
  1466. }
  1467. shareVars.Store("ObjDicOfTpperception", ObjDicOfTpperception)
  1468. shareVars.Store("ObjTypeDicOfTpperception", objTypeDicOfTpperception)
  1469. shareVars.Store("ObjSpeedDicOfTpperception", objSpeedDicOfTpperception)
  1470. },
  1471. })
  1472. if err == nil {
  1473. create = true
  1474. }
  1475. }
  1476. // 31
  1477. if topic == masterConfig.TopicOfTpperceptionVis &&
  1478. (len(masterConfig.RuleOfTpperceptionVis1) > 0 ||
  1479. len(masterConfig.RuleOfTpperceptionVis3) > 0) {
  1480. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1481. Node: commonConfig.RosNode,
  1482. Topic: topic,
  1483. Callback: func(data *visualization_msgs.MarkerArray) {
  1484. subscribersTimeMutexes[i].Lock()
  1485. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1486. subscribersMutexes[i].Lock()
  1487. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1488. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1489. faultLabel := ""
  1490. if len(masterConfig.RuleOfTpperceptionVis1) > 0 {
  1491. for _, f := range masterConfig.RuleOfTpperceptionVis1 {
  1492. faultLabel = f(data)
  1493. if faultLabel != "" {
  1494. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1495. subscribersTimes[i] = time.Now()
  1496. goto TriggerSuccess
  1497. }
  1498. }
  1499. }
  1500. if len(masterConfig.RuleOfTpperceptionVis3) > 0 {
  1501. for _, f := range masterConfig.RuleOfTpperceptionVis3 {
  1502. faultLabel = f(shareVars, data)
  1503. if faultLabel != "" {
  1504. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1505. subscribersTimes[i] = time.Now()
  1506. goto TriggerSuccess
  1507. }
  1508. }
  1509. }
  1510. TriggerSuccess:
  1511. subscribersMutexes[i].Unlock()
  1512. }
  1513. subscribersTimeMutexes[i].Unlock()
  1514. },
  1515. })
  1516. if err == nil {
  1517. create = true
  1518. }
  1519. }
  1520. // 32
  1521. if topic == masterConfig.TopicOfTprouteplan &&
  1522. (len(masterConfig.RuleOfTprouteplan1) > 0 ||
  1523. len(masterConfig.RuleOfTprouteplan3) > 0) {
  1524. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1525. Node: commonConfig.RosNode,
  1526. Topic: topic,
  1527. Callback: func(data *pjisuv_msgs.RoutePlan) {
  1528. subscribersTimeMutexes[i].Lock()
  1529. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1530. subscribersMutexes[i].Lock()
  1531. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1532. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1533. faultLabel := ""
  1534. if len(masterConfig.RuleOfTprouteplan1) > 0 {
  1535. for _, f := range masterConfig.RuleOfTprouteplan1 {
  1536. faultLabel = f(data)
  1537. if faultLabel != "" {
  1538. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1539. subscribersTimes[i] = time.Now()
  1540. goto TriggerSuccess
  1541. }
  1542. }
  1543. }
  1544. if len(masterConfig.RuleOfTprouteplan3) > 0 {
  1545. for _, f := range masterConfig.RuleOfTprouteplan3 {
  1546. faultLabel = f(shareVars, data)
  1547. if faultLabel != "" {
  1548. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1549. subscribersTimes[i] = time.Now()
  1550. goto TriggerSuccess
  1551. }
  1552. }
  1553. }
  1554. TriggerSuccess:
  1555. subscribersMutexes[i].Unlock()
  1556. }
  1557. subscribersTimeMutexes[i].Unlock()
  1558. },
  1559. })
  1560. if err == nil {
  1561. create = true
  1562. }
  1563. }
  1564. // 33
  1565. if topic == masterConfig.TopicOfTrajectoryDisplay &&
  1566. (len(masterConfig.RuleOfTrajectoryDisplay1) > 0 ||
  1567. len(masterConfig.RuleOfTrajectoryDisplay3) > 0) {
  1568. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1569. Node: commonConfig.RosNode,
  1570. Topic: topic,
  1571. Callback: func(data *nav_msgs.Path) {
  1572. subscribersTimeMutexes[i].Lock()
  1573. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1574. subscribersMutexes[i].Lock()
  1575. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1576. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1577. faultLabel := ""
  1578. if len(masterConfig.RuleOfTrajectoryDisplay1) > 0 {
  1579. for _, f := range masterConfig.RuleOfTrajectoryDisplay1 {
  1580. faultLabel = f(data)
  1581. if faultLabel != "" {
  1582. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1583. subscribersTimes[i] = time.Now()
  1584. goto TriggerSuccess
  1585. }
  1586. }
  1587. }
  1588. if len(masterConfig.RuleOfTrajectoryDisplay3) > 0 {
  1589. for _, f := range masterConfig.RuleOfTrajectoryDisplay3 {
  1590. faultLabel = f(shareVars, data)
  1591. if faultLabel != "" {
  1592. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1593. subscribersTimes[i] = time.Now()
  1594. goto TriggerSuccess
  1595. }
  1596. }
  1597. }
  1598. TriggerSuccess:
  1599. subscribersMutexes[i].Unlock()
  1600. }
  1601. subscribersTimeMutexes[i].Unlock()
  1602. },
  1603. })
  1604. if err == nil {
  1605. create = true
  1606. }
  1607. }
  1608. // 34
  1609. if topic == masterConfig.TopicOfUngroundCloudpoints &&
  1610. (len(masterConfig.RuleOfUngroundCloudpoints1) > 0 ||
  1611. len(masterConfig.RuleOfUngroundCloudpoints3) > 0) {
  1612. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1613. Node: commonConfig.RosNode,
  1614. Topic: topic,
  1615. Callback: func(data *sensor_msgs.PointCloud2) {
  1616. subscribersTimeMutexes[i].Lock()
  1617. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1618. subscribersMutexes[i].Lock()
  1619. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1620. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1621. faultLabel := ""
  1622. if len(masterConfig.RuleOfUngroundCloudpoints1) > 0 {
  1623. for _, f := range masterConfig.RuleOfUngroundCloudpoints1 {
  1624. faultLabel = f(data)
  1625. if faultLabel != "" {
  1626. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1627. subscribersTimes[i] = time.Now()
  1628. goto TriggerSuccess
  1629. }
  1630. }
  1631. }
  1632. if len(masterConfig.RuleOfUngroundCloudpoints3) > 0 {
  1633. for _, f := range masterConfig.RuleOfUngroundCloudpoints3 {
  1634. faultLabel = f(shareVars, data)
  1635. if faultLabel != "" {
  1636. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1637. subscribersTimes[i] = time.Now()
  1638. goto TriggerSuccess
  1639. }
  1640. }
  1641. }
  1642. TriggerSuccess:
  1643. subscribersMutexes[i].Unlock()
  1644. }
  1645. subscribersTimeMutexes[i].Unlock()
  1646. },
  1647. })
  1648. if err == nil {
  1649. create = true
  1650. }
  1651. }
  1652. // 35
  1653. if topic == masterConfig.TopicOfCameraImage &&
  1654. (len(masterConfig.RuleOfCameraImage1) > 0 ||
  1655. len(masterConfig.RuleOfCameraImage3) > 0) {
  1656. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1657. Node: commonConfig.RosNode,
  1658. Topic: topic,
  1659. Callback: func(data *sensor_msgs.Image) {
  1660. subscribersTimeMutexes[i].Lock()
  1661. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1662. subscribersMutexes[i].Lock()
  1663. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1664. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1665. faultLabel := ""
  1666. if len(masterConfig.RuleOfCameraImage1) > 0 {
  1667. for _, f := range masterConfig.RuleOfCameraImage1 {
  1668. faultLabel = f(data)
  1669. if faultLabel != "" {
  1670. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1671. subscribersTimes[i] = time.Now()
  1672. goto TriggerSuccess
  1673. }
  1674. }
  1675. }
  1676. if len(masterConfig.RuleOfCameraImage3) > 0 {
  1677. for _, f := range masterConfig.RuleOfCameraImage3 {
  1678. faultLabel = f(shareVars, data)
  1679. if faultLabel != "" {
  1680. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1681. subscribersTimes[i] = time.Now()
  1682. goto TriggerSuccess
  1683. }
  1684. }
  1685. }
  1686. TriggerSuccess:
  1687. subscribersMutexes[i].Unlock()
  1688. }
  1689. subscribersTimeMutexes[i].Unlock()
  1690. },
  1691. })
  1692. if err == nil {
  1693. create = true
  1694. }
  1695. }
  1696. // 36 有共享变量的订阅者必须被创建
  1697. if topic == masterConfig.TopicOfDataRead {
  1698. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1699. Node: commonConfig.RosNode,
  1700. Topic: topic,
  1701. Callback: func(data *pjisuv_msgs.Retrieval) {
  1702. subscribersTimeMutexes[i].Lock()
  1703. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1704. subscribersMutexes[i].Lock()
  1705. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1706. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1707. faultLabel := ""
  1708. if len(masterConfig.RuleOfDataRead1) > 0 {
  1709. for _, f := range masterConfig.RuleOfDataRead1 {
  1710. faultLabel = f(data)
  1711. if faultLabel != "" {
  1712. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1713. subscribersTimes[i] = time.Now()
  1714. goto TriggerSuccess
  1715. }
  1716. }
  1717. }
  1718. if len(masterConfig.RuleOfDataRead3) > 0 {
  1719. for _, f := range masterConfig.RuleOfDataRead3 {
  1720. faultLabel = f(shareVars, data)
  1721. if faultLabel != "" {
  1722. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1723. subscribersTimes[i] = time.Now()
  1724. goto TriggerSuccess
  1725. }
  1726. }
  1727. }
  1728. TriggerSuccess:
  1729. subscribersMutexes[i].Unlock()
  1730. }
  1731. subscribersTimeMutexes[i].Unlock()
  1732. // 更新共享变量
  1733. numCountDataReadOfDataRead++
  1734. if numCountDataReadOfDataRead == 10 {
  1735. egoSteeringRealOfDataRead = append(egoSteeringRealOfDataRead, data.ActStrWhAng)
  1736. egoThrottleRealOfDataRead = append(egoThrottleRealOfDataRead, data.AccPed2)
  1737. numCountDataReadOfDataRead = 0
  1738. }
  1739. GearPosSlice = append(GearPosSlice, data.GearPos)
  1740. shareVars.Store("NumCountDataReadOfDataRead", numCountDataReadOfDataRead)
  1741. shareVars.Store("EgoSteeringRealOfDataRead", egoSteeringRealOfDataRead)
  1742. shareVars.Store("EgoThrottleRealOfDataRead", egoThrottleRealOfDataRead)
  1743. shareVars.Store("ActStrWhAngOfDataRead", data.ActStrWhAng)
  1744. shareVars.Store("GearPosSlice", GearPosSlice)
  1745. },
  1746. })
  1747. if err == nil {
  1748. create = true
  1749. }
  1750. }
  1751. // 37
  1752. if topic == masterConfig.TopicOfPjiGps &&
  1753. (len(masterConfig.RuleOfPjiGps1) > 0 ||
  1754. len(masterConfig.RuleOfPjiGps3) > 0) {
  1755. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1756. Node: commonConfig.RosNode,
  1757. Topic: topic,
  1758. Callback: func(data *pjisuv_msgs.PerceptionLocalization) {
  1759. subscribersTimeMutexes[i].Lock()
  1760. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1761. subscribersMutexes[i].Lock()
  1762. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1763. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1764. faultLabel := ""
  1765. if len(masterConfig.RuleOfPjiGps1) > 0 {
  1766. for _, f := range masterConfig.RuleOfPjiGps1 {
  1767. faultLabel = f(data)
  1768. if faultLabel != "" {
  1769. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1770. subscribersTimes[i] = time.Now()
  1771. goto TriggerSuccess
  1772. }
  1773. }
  1774. }
  1775. if len(masterConfig.RuleOfPjiGps3) > 0 {
  1776. for _, f := range masterConfig.RuleOfPjiGps3 {
  1777. faultLabel = f(shareVars, data)
  1778. if faultLabel != "" {
  1779. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1780. subscribersTimes[i] = time.Now()
  1781. goto TriggerSuccess
  1782. }
  1783. }
  1784. }
  1785. TriggerSuccess:
  1786. subscribersMutexes[i].Unlock()
  1787. }
  1788. subscribersTimeMutexes[i].Unlock()
  1789. },
  1790. })
  1791. if err == nil {
  1792. create = true
  1793. }
  1794. }
  1795. // 39 有共享变量的订阅者必须被创建
  1796. if topic == masterConfig.TopicOfPjVehicleFdbPub {
  1797. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1798. Node: commonConfig.RosNode,
  1799. Topic: topic,
  1800. Callback: func(data *pjisuv_msgs.VehicleFdb) {
  1801. subscribersTimeMutexes[i].Lock()
  1802. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1803. subscribersMutexes[i].Lock()
  1804. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1805. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1806. faultLabel := ""
  1807. if len(masterConfig.RuleOfPjVehicleFdbPub1) > 0 {
  1808. for _, f := range masterConfig.RuleOfPjVehicleFdbPub1 {
  1809. faultLabel = f(data)
  1810. if faultLabel != "" {
  1811. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1812. subscribersTimes[i] = time.Now()
  1813. goto TriggerSuccess
  1814. }
  1815. }
  1816. }
  1817. if len(masterConfig.RuleOfPjVehicleFdbPub3) > 0 {
  1818. for _, f := range masterConfig.RuleOfPjVehicleFdbPub3 {
  1819. faultLabel = f(shareVars, data)
  1820. if faultLabel != "" {
  1821. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1822. subscribersTimes[i] = time.Now()
  1823. goto TriggerSuccess
  1824. }
  1825. }
  1826. }
  1827. TriggerSuccess:
  1828. subscribersMutexes[i].Unlock()
  1829. }
  1830. subscribersTimeMutexes[i].Unlock()
  1831. // 更新共享变量
  1832. shareVars.Store("AutomodeOfPjVehicleFdbPub", data.Automode)
  1833. },
  1834. })
  1835. if err == nil {
  1836. create = true
  1837. }
  1838. }
  1839. // 40 有共享变量的订阅者必须被创建
  1840. if topic == masterConfig.TopicOfEndPointMessage {
  1841. subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
  1842. Node: commonConfig.RosNode,
  1843. Topic: topic,
  1844. Callback: func(data *geometry_msgs.Point) {
  1845. subscribersTimeMutexes[i].Lock()
  1846. if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
  1847. subscribersMutexes[i].Lock()
  1848. faultHappenTime := util.GetNowTimeCustom() // 获取当前故障发生时间
  1849. lastTimeWindow := commonEntity.GetLastTimeWindow() // 获取最后一个时间窗口
  1850. faultLabel := ""
  1851. if len(masterConfig.RuleOfEndPointMessage1) > 0 {
  1852. for _, f := range masterConfig.RuleOfEndPointMessage1 {
  1853. faultLabel = f(data)
  1854. if faultLabel != "" {
  1855. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1856. subscribersTimes[i] = time.Now()
  1857. goto TriggerSuccess
  1858. }
  1859. }
  1860. }
  1861. if len(masterConfig.RuleOfEndPointMessage3) > 0 {
  1862. for _, f := range masterConfig.RuleOfEndPointMessage3 {
  1863. faultLabel = f(shareVars, data)
  1864. if faultLabel != "" {
  1865. saveTimeWindow(faultLabel, faultHappenTime, lastTimeWindow)
  1866. subscribersTimes[i] = time.Now()
  1867. goto TriggerSuccess
  1868. }
  1869. }
  1870. }
  1871. TriggerSuccess:
  1872. subscribersMutexes[i].Unlock()
  1873. }
  1874. subscribersTimeMutexes[i].Unlock()
  1875. // 更新共享变量
  1876. shareVars.Store("EndPointX", data.X)
  1877. shareVars.Store("EndPointY", data.Y)
  1878. },
  1879. })
  1880. if err == nil {
  1881. create = true
  1882. }
  1883. }
  1884. if err != nil {
  1885. c_log.GlobalLogger.Infof("创建订阅者报错,可能由于节点未启动,再次尝试【%v】", err)
  1886. time.Sleep(time.Duration(2) * time.Second)
  1887. continue
  1888. } else {
  1889. if create {
  1890. c_log.GlobalLogger.Infof("创建订阅者订阅话题【%v】", topic)
  1891. }
  1892. break
  1893. }
  1894. }
  1895. }
  1896. c_log.GlobalLogger.Infof("全部订阅者创建完成。")
  1897. select {
  1898. case signal := <-service.ChannelKillWindowProducer:
  1899. if signal == 1 {
  1900. commonConfig.RosNode.Close()
  1901. service.AddKillTimes("3")
  1902. return
  1903. }
  1904. }
  1905. }
  1906. func saveTimeWindow(faultLabel string, faultHappenTime string, lastTimeWindow *commonEntity.TimeWindow) {
  1907. saveTimeWindowMutex.Lock()
  1908. defer saveTimeWindowMutex.Unlock()
  1909. masterTopics, slaveTopics := getTopicsOfNode(faultLabel)
  1910. if lastTimeWindow == nil || util.TimeCustom1GreaterTimeCustom2(faultHappenTime, lastTimeWindow.TimeWindowEnd) { // 如果是不在旧故障窗口内,添加一个新窗口
  1911. exceptBegin := util.TimeCustomChange(faultHappenTime, -commonConfig.PlatformConfig.TaskBeforeTime)
  1912. finalTimeWindowBegin := ""
  1913. if util.TimeCustom1LessEqualThanTimeCustom2(exceptBegin, latestTimeWindowEnd) { // 窗口最早时间不能早于上一个窗口结束时间
  1914. finalTimeWindowBegin = latestTimeWindowEnd
  1915. } else {
  1916. finalTimeWindowBegin = exceptBegin
  1917. }
  1918. latestTimeWindowEnd = util.TimeCustomChange(faultHappenTime, commonConfig.PlatformConfig.TaskAfterTime)
  1919. newTimeWindow := commonEntity.TimeWindow{
  1920. FaultTime: faultHappenTime,
  1921. TimeWindowBegin: finalTimeWindowBegin,
  1922. TimeWindowEnd: latestTimeWindowEnd,
  1923. Length: util.CalculateDifferenceOfTimeCustom(finalTimeWindowBegin, latestTimeWindowEnd),
  1924. Labels: []string{faultLabel},
  1925. MasterTopics: masterTopics,
  1926. SlaveTopics: slaveTopics,
  1927. }
  1928. c_log.GlobalLogger.Infof("不在旧故障窗口内,向生产者队列添加一个新窗口,【Lable】=%v,【FaultTime】=%v,【Length】=%v", newTimeWindow.Labels, newTimeWindow.FaultTime, newTimeWindow.Length)
  1929. commonEntity.AddTimeWindowToTimeWindowProducerQueue(newTimeWindow)
  1930. } else { // 如果在旧故障窗口内
  1931. commonEntity.TimeWindowProducerQueueMutex.RLock()
  1932. defer commonEntity.TimeWindowProducerQueueMutex.RUnlock()
  1933. // 更新故障窗口end时间
  1934. expectEnd := util.TimeCustomChange(faultHappenTime, commonConfig.PlatformConfig.TaskAfterTime) // 窗口期望关闭时间是触发时间加上后置时间
  1935. expectLength := util.CalculateDifferenceOfTimeCustom(lastTimeWindow.TimeWindowBegin, expectEnd)
  1936. if expectLength < commonConfig.PlatformConfig.TaskMaxTime {
  1937. latestTimeWindowEnd = expectEnd
  1938. lastTimeWindow.TimeWindowEnd = latestTimeWindowEnd
  1939. lastTimeWindow.Length = util.CalculateDifferenceOfTimeCustom(lastTimeWindow.TimeWindowBegin, lastTimeWindow.TimeWindowEnd)
  1940. }
  1941. // 更新label
  1942. labels := lastTimeWindow.Labels
  1943. lastTimeWindow.Labels = util.AppendIfNotExists(labels, faultLabel)
  1944. // 更新 topic
  1945. sourceMasterTopics := lastTimeWindow.MasterTopics
  1946. lastTimeWindow.MasterTopics = util.MergeSlice(sourceMasterTopics, masterTopics)
  1947. sourceSlaveTopics := lastTimeWindow.SlaveTopics
  1948. lastTimeWindow.SlaveTopics = util.MergeSlice(sourceSlaveTopics, slaveTopics)
  1949. c_log.GlobalLogger.Infof("在旧故障窗口内,更新生产者队列最新的窗口,【Lable】=%v,【FaultTime】=%v,【Length】=%v", lastTimeWindow.Labels, lastTimeWindow.FaultTime, lastTimeWindow.Length)
  1950. }
  1951. }
  1952. func getTopicsOfNode(faultLabel string) (masterTopics []string, slaveTopics []string) {
  1953. // 获取所有需要采集的topic
  1954. var faultCodeTopics []string
  1955. for _, code := range commonConfig.CloudConfig.Triggers {
  1956. if code.Label == faultLabel {
  1957. faultCodeTopics = code.Topics
  1958. }
  1959. }
  1960. // 根据不同节点采集的topic进行分配采集
  1961. for _, acceptTopic := range faultCodeTopics {
  1962. for _, host := range commonConfig.CloudConfig.Hosts {
  1963. for _, topic := range host.Topics {
  1964. if host.Name == commonConfig.CloudConfig.Hosts[0].Name && acceptTopic == topic {
  1965. masterTopics = append(masterTopics, acceptTopic)
  1966. }
  1967. if host.Name == commonConfig.CloudConfig.Hosts[1].Name && acceptTopic == topic {
  1968. slaveTopics = append(slaveTopics, acceptTopic)
  1969. }
  1970. }
  1971. }
  1972. }
  1973. return masterTopics, slaveTopics
  1974. }
  1975. func isPointInPolygon(p Point, vertices []Point) bool {
  1976. intersections := 0
  1977. for i := 0; i < len(vertices); i++ {
  1978. j := (i + 1) % len(vertices)
  1979. if rayIntersectsSegment(p, vertices[i], vertices[j]) {
  1980. intersections++
  1981. }
  1982. }
  1983. return !(intersections%2 == 1)
  1984. }
  1985. func rayIntersectsSegment(p, p1, p2 Point) bool {
  1986. if p1.y > p2.y {
  1987. p1, p2 = p2, p1
  1988. }
  1989. if p.y == p1.y || p.y == p2.y {
  1990. p.y += 0.0001
  1991. }
  1992. if p.y < p1.y || p.y > p2.y {
  1993. return false
  1994. }
  1995. if p.x > max(p1.x, p2.x) {
  1996. return false
  1997. }
  1998. if p.x < min(p1.x, p2.x) {
  1999. return true
  2000. }
  2001. blueSlope := (p.y-p1.y)*(p2.x-p1.x)/(p2.y-p1.y) + p1.x
  2002. return p.x < blueSlope
  2003. }
  2004. func min(a, b float64) float64 {
  2005. if a < b {
  2006. return a
  2007. }
  2008. return b
  2009. }
  2010. func max(a, b float64) float64 {
  2011. if a > b {
  2012. return a
  2013. }
  2014. return b
  2015. }