system_executor.cpp 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470
  1. //
  2. // Created by huli on 2020/7/2.
  3. //
  4. #include "system_executor.h"
  5. #include "../message/measure_message.pb.h"
  6. #include "../system/system_communication.h"
  7. #include "../wanji_lidar/wanji_manager.h"
  8. #include "../tool/common_data.h"
  9. #include "../wanji_lidar/measure_filter.h"
  10. std::ofstream g_debug_file;
  11. System_executor::System_executor()
  12. {
  13. }
  14. System_executor::~System_executor()
  15. {
  16. system_executor_uninit();
  17. }
  18. //初始化
  19. Error_manager System_executor::system_executor_init(int threads_size, int terminal_id)
  20. {
  21. m_thread_pool.thread_pool_init(threads_size);
  22. m_system_executor_status = SYSTEM_EXECUTOR_READY;
  23. m_terminal_id = terminal_id;
  24. return Error_code::SUCCESS;
  25. }
  26. //反初始化
  27. Error_manager System_executor::system_executor_uninit()
  28. {
  29. if(g_debug_file.is_open())
  30. g_debug_file.close();
  31. m_thread_pool.thread_pool_uninit();
  32. m_system_executor_status = SYSTEM_EXECUTOR_UNKNOW;
  33. return Error_code::SUCCESS;
  34. }
  35. //检查消息是否有效, 主要检查消息类型和接受者, 判断这条消息是不是给我的.
  36. Error_manager System_executor::check_msg(Communication_message* p_msg)
  37. {
  38. if ( p_msg == NULL )
  39. {
  40. return Error_manager(Error_code::POINTER_IS_NULL, Error_level::MINOR_ERROR,
  41. " POINTER IS NULL ");
  42. }
  43. //检查消息类型
  44. switch ( p_msg->get_message_type() )
  45. {
  46. case Communication_message::Message_type::eGround_detect_request_msg:
  47. {
  48. //检查接受人
  49. if ( p_msg->get_receiver() == Communication_message::Communicator::eGround_measurer)
  50. {
  51. message::Ground_detect_request_msg t_ground_detect_request_msg;
  52. //针对消息类型, 对消息进行二次解析
  53. if (t_ground_detect_request_msg.ParseFromString(p_msg->get_message_buf()))
  54. {
  55. // //检查终端id
  56. // 普爱不检查终端id
  57. // if ( t_ground_detect_request_msg.terminal_id() == m_terminal_id )
  58. // {
  59. return Error_code::SUCCESS;
  60. // }
  61. }
  62. }
  63. break;
  64. }
  65. default :
  66. ;
  67. break;
  68. }
  69. //无效的消息,
  70. return Error_manager(Error_code::INVALID_MESSAGE, Error_level::NEGLIGIBLE_ERROR,
  71. " INVALID_MESSAGE error ");
  72. }
  73. //检查执行者的状态, 判断能否处理这条消息,
  74. Error_manager System_executor::check_executer(Communication_message* p_msg)
  75. {
  76. //huli DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD 记得删除
  77. return Error_code::SUCCESS;
  78. if ( p_msg == NULL )
  79. {
  80. return Error_manager(Error_code::POINTER_IS_NULL, Error_level::MINOR_ERROR,
  81. " POINTER IS NULL ");
  82. }
  83. Error_manager t_error;
  84. //通过 p_msg->get_message_type() 和 p_msg->get_receiver() 找到处理模块的实例对象, 查询执行人是否可以处理这条消息
  85. switch ( p_msg->get_message_type() )
  86. {
  87. case Communication_message::Message_type::eGround_detect_request_msg:
  88. {
  89. Error_manager t_executor_result = System_executor::get_instance_references().check_status();
  90. if (t_executor_result == SUCCESS)
  91. {
  92. return Error_code::SUCCESS;
  93. }
  94. else
  95. {
  96. //整合所有的错误码
  97. t_error.compare_and_cover_error(t_executor_result);
  98. if (t_error.get_error_level() == NEGLIGIBLE_ERROR)//一级故障,轻微故障,
  99. {
  100. LOG(INFO) << "executer_is_busy , ";
  101. //返回繁忙之后, 通信模块1秒后再次调用check
  102. return Error_code::COMMUNICATION_EXCUTER_IS_BUSY;
  103. }
  104. else//返回二级故障,可以封装一条答复信息, 返回错误码
  105. {
  106. message::Ground_detect_request_msg t_ground_detect_request_msg;
  107. //针对消息类型, 对消息进行二次解析
  108. if (t_ground_detect_request_msg.ParseFromString(p_msg->get_message_buf()))
  109. {
  110. //创建一条答复消息
  111. message::Ground_detect_response_msg t_ground_detect_response_msg;
  112. t_ground_detect_response_msg.mutable_base_info()->set_msg_type(message::Message_type::eGround_detect_response_msg);
  113. t_ground_detect_response_msg.mutable_base_info()->set_timeout_ms(5000);
  114. t_ground_detect_response_msg.mutable_base_info()->set_sender(message::Communicator::eGround_measurer);
  115. t_ground_detect_response_msg.mutable_base_info()->set_receiver(message::Communicator::eMain);
  116. t_ground_detect_response_msg.set_command_key(t_ground_detect_request_msg.command_key());
  117. t_ground_detect_response_msg.set_terminal_id(t_ground_detect_request_msg.terminal_id());
  118. t_ground_detect_response_msg.mutable_error_manager()->set_error_code(t_error.get_error_code());
  119. t_ground_detect_response_msg.mutable_error_manager()->set_error_level((message::Error_level)t_error.get_error_level());
  120. t_ground_detect_response_msg.mutable_error_manager()->set_error_description(t_error.get_error_description());
  121. std::string t_msg = t_ground_detect_response_msg.SerializeAsString();
  122. System_communication::get_instance_references().encapsulate_msg(t_msg);
  123. LOG(INFO) << " System_executor::check_executer executer status error "<< this;
  124. return t_error;
  125. }
  126. else
  127. {
  128. LOG(INFO) << " System_executor::check_executer second PARSE_ERROR "<< this;
  129. return Error_manager(Error_code::SYSTEM_EXECUTOR_PARSE_ERROR, Error_level::MINOR_ERROR,
  130. " message::Measure_request_msg ParseFromString error ");
  131. }
  132. }
  133. }
  134. break;
  135. }
  136. default :
  137. ;
  138. break;
  139. }
  140. return t_error;
  141. }
  142. //处理消息的执行函数
  143. Error_manager System_executor::execute_msg(Communication_message* p_msg)
  144. {
  145. if ( p_msg == NULL )
  146. {
  147. return Error_manager(Error_code::POINTER_IS_NULL, Error_level::MINOR_ERROR,
  148. " POINTER IS NULL ");
  149. }
  150. // std::cout << " huli test ::333333333333333333333333333333:: " << " p_msg->get_message_type() = " << p_msg->get_message_type() << std::endl;
  151. switch ( p_msg->get_message_type() )
  152. {
  153. case Communication_message::eGround_detect_request_msg:
  154. {
  155. message::Ground_detect_request_msg t_ground_detect_request_msg;
  156. //针对消息类型, 对消息进行二次解析
  157. if (t_ground_detect_request_msg.ParseFromString(p_msg->get_message_buf()))
  158. {
  159. //往线程池添加执行任务, 之后会唤醒一个线程去执行他.
  160. m_thread_pool.enqueue(&System_executor::execute_for_measure, this,
  161. t_ground_detect_request_msg.command_key(), t_ground_detect_request_msg.terminal_id(),
  162. p_msg->get_receive_time());
  163. }
  164. else
  165. {
  166. return Error_manager(Error_code::SYSTEM_EXECUTOR_PARSE_ERROR, Error_level::MINOR_ERROR,
  167. " message::Measure_request_msg ParseFromString error ");
  168. }
  169. break;
  170. }
  171. default:
  172. {
  173. }
  174. }
  175. return Error_code::SUCCESS;
  176. }
  177. //检查状态
  178. Error_manager System_executor::check_status()
  179. {
  180. if ( m_system_executor_status == SYSTEM_EXECUTOR_READY )
  181. {
  182. if ( m_thread_pool.thread_is_full_load() == false )
  183. {
  184. return Error_code::SUCCESS;
  185. }
  186. else
  187. {
  188. return Error_manager(Error_code::SYSTEM_EXECUTOR_STATUS_BUSY, Error_level::NEGLIGIBLE_ERROR,
  189. " System_executor::check_status error ");
  190. }
  191. }
  192. else
  193. {
  194. return Error_manager(Error_code::SYSTEM_EXECUTOR_STATUS_ERROR, Error_level::MINOR_ERROR,
  195. " System_executor::check_status error ");
  196. }
  197. }
  198. //定时发送状态信息
  199. Error_manager System_executor::encapsulate_send_status()
  200. {
  201. if(!g_debug_file.is_open())
  202. {
  203. g_debug_file.open("./filter_debug_result.txt", std::ios::app);
  204. }
  205. Error_manager t_error;
  206. //创建一条状态消息
  207. message::Ground_status_msg t_ground_status_msg;
  208. t_ground_status_msg.mutable_base_info()->set_msg_type(message::Message_type::eLocate_status_msg);
  209. t_ground_status_msg.mutable_base_info()->set_timeout_ms(5000);
  210. t_ground_status_msg.mutable_base_info()->set_sender(message::Communicator::eGround_measurer);
  211. t_ground_status_msg.mutable_base_info()->set_receiver(message::Communicator::eEmpty);
  212. // 普爱统一一个万集节点, 各终端消息分别发送
  213. t_ground_status_msg.set_terminal_id(m_terminal_id);
  214. //万集716
  215. Wanji_manager::Wanji_manager_status t_wanji_manager_status = Wanji_manager::get_instance_references().get_status();
  216. t_ground_status_msg.set_wanji_manager_status((message::Wanji_manager_status)t_wanji_manager_status);
  217. std::map<int, Wanji_lidar_device*> & t_wanji_lidar_device_map = Wanji_manager::get_instance_references().get_wanji_lidar_device_map();
  218. for (auto iter = t_wanji_lidar_device_map.begin(); iter != t_wanji_lidar_device_map.end(); ++iter)
  219. {
  220. Wanji_lidar_device::Wanji_lidar_device_status t_wanji_lidar_device_status = (*iter).second->get_status();
  221. t_ground_status_msg.add_wanji_lidar_device_status((message::Wanji_lidar_device_status)t_wanji_lidar_device_status);
  222. }
  223. std::map<int, Region_worker*> & t_region_worker_map = Wanji_manager::get_instance_references().get_region_worker_map();
  224. int region_index = 0;
  225. for (auto iter = t_region_worker_map.begin(); iter != t_region_worker_map.end(); ++iter)
  226. {
  227. message::Ground_status_msg t_multi_status_msg;
  228. t_multi_status_msg.CopyFrom(t_ground_status_msg);
  229. t_multi_status_msg.set_terminal_id(region_index++);
  230. Region_worker::Region_worker_status t_region_worker_status = (*iter).second->get_status();
  231. // t_multi_status_msg.add_region_worker_status((message::Region_worker_status)t_region_worker_status);
  232. t_multi_status_msg.set_region_worker_status((message::Region_worker_status)t_region_worker_status);
  233. //万集雷达的自动定位信息
  234. Common_data::Car_wheel_information t_car_wheel_information;
  235. t_error = (*iter).second->get_last_wheel_information(&t_car_wheel_information, std::chrono::system_clock::now());
  236. //无论定位结果如何, 都要填充数据, 即使全部写0, 必须保证通信格式正确.
  237. message::Locate_information t_locate_information;
  238. //= t_multi_status_msg.add_locate_information_realtime();
  239. t_locate_information.set_locate_x(t_car_wheel_information.center_x);
  240. t_locate_information.set_locate_y(t_car_wheel_information.center_y);
  241. t_locate_information.set_locate_angle(t_car_wheel_information.car_angle);
  242. t_locate_information.set_locate_length(0);
  243. t_locate_information.set_locate_width(0);
  244. t_locate_information.set_locate_height(0);
  245. t_locate_information.set_locate_wheel_base(t_car_wheel_information.wheel_base);
  246. t_locate_information.set_locate_wheel_width(t_car_wheel_information.wheel_width);
  247. t_locate_information.set_locate_front_theta(t_car_wheel_information.front_theta);
  248. t_locate_information.set_locate_correct(t_car_wheel_information.correctness);
  249. t_multi_status_msg.mutable_locate_information_realtime()->CopyFrom(t_locate_information);
  250. t_multi_status_msg.mutable_error_manager()->set_error_code(t_error.get_error_code());
  251. t_multi_status_msg.mutable_error_manager()->set_error_level((message::Error_level)t_error.get_error_level());
  252. t_multi_status_msg.mutable_error_manager()->set_error_description(t_error.get_error_description(), t_error.get_description_length());
  253. std::string t_msg = t_multi_status_msg.SerializeAsString();
  254. System_communication::get_instance_references().encapsulate_msg(t_msg);
  255. // // 记录滤波前后测量数据,用于判断测量精度变化
  256. // static int t_line_count = 0;
  257. // if(region_index == 6 && ++t_line_count < 120) {
  258. // if (g_debug_file.is_open() && t_car_wheel_information.correctness) {
  259. // g_debug_file << "basic:"<<t_line_count<<"\n" <<std::setprecision(6)<<std::setiosflags(std::ios::fixed)
  260. // << t_car_wheel_information.center_x << ", "
  261. // << t_car_wheel_information.center_y << ", "
  262. // << t_car_wheel_information.car_angle << ", "
  263. // << t_car_wheel_information.wheel_base << ", "
  264. // << t_car_wheel_information.wheel_width << ", "
  265. // << t_car_wheel_information.front_theta << std::endl;
  266. // }
  267. // Common_data::Car_wheel_information t_filtered_result;
  268. // Error_manager ec = Measure_filter::get_instance_references().get_filtered_wheel_information(region_index - 1, t_filtered_result);
  269. // if(ec==SUCCESS){
  270. // if (g_debug_file.is_open()) {
  271. // g_debug_file << "filtered:"<<t_line_count<<"\n" <<std::setprecision(6)<<std::setiosflags(std::ios::fixed)
  272. // << t_filtered_result.center_x << ", "
  273. // << t_filtered_result.center_y << ", "
  274. // << t_filtered_result.car_angle << ", "
  275. // << t_filtered_result.wheel_base << ", "
  276. // << t_filtered_result.wheel_width << ", "
  277. // << t_filtered_result.front_theta << std::endl;
  278. // }
  279. // }else{
  280. // LOG(WARNING)<<ec.to_string();
  281. // }
  282. //// LOG(WARNING) << t_multi_status_msg.DebugString();
  283. // }
  284. }
  285. // std::cout << " huli test :::: " << " t_ground_status_msg.DebugString() = " << std::endl;
  286. // std::cout << t_ground_status_msg.DebugString() << std::endl;
  287. // std::cout << " huli test :::: " << " t_ground_status_msg.DebugString() = " << t_ground_status_msg.DebugString() << std::endl;
  288. return Error_code::SUCCESS;
  289. }
  290. //判断是否为待机,如果已经准备好,则可以执行任务。
  291. bool System_executor::is_ready()
  292. {
  293. if ( m_system_executor_status == SYSTEM_EXECUTOR_READY && m_thread_pool.thread_is_full_load() == false )
  294. {
  295. return true;
  296. }
  297. else
  298. {
  299. return false;
  300. }
  301. }
  302. System_executor::System_executor_status System_executor::get_system_executor_status()
  303. {
  304. return m_system_executor_status;
  305. }
  306. int System_executor::get_terminal_id()
  307. {
  308. return m_terminal_id;
  309. }
  310. //雷达感测定位 的处理函数
  311. //input::command_id, 消息指令id, 由主控制系统生成的唯一码
  312. //input::command_id, 终端id, 对应具体的某个车位
  313. //return::void, 没有返回, 执行结果直接生成一条答复消息, 然后通过通信返回
  314. void System_executor::execute_for_measure(std::string command_info, int terminal_id,std::chrono::system_clock::time_point receive_time)
  315. {
  316. Error_manager t_error;
  317. Error_manager t_result;
  318. LOG(INFO) << " System_executor::execute_for_measure run "<< this;
  319. //第1步, 按照时间生成中间文件的保存路径
  320. time_t nowTime;
  321. nowTime = time(NULL);
  322. struct tm* sysTime = localtime(&nowTime);
  323. char t_save_path[256] = { 0 };
  324. sprintf(t_save_path, "../data/%04d%02d%02d_%02d%02d%02d",
  325. sysTime->tm_year + 1900, sysTime->tm_mon + 1, sysTime->tm_mday, sysTime->tm_hour, sysTime->tm_min, sysTime->tm_sec);
  326. //第2步, 创建万集管理模块的任务单, 并发送到 wanji_manager
  327. Wanji_manager_task t_wanji_manager_task ;
  328. Common_data::Car_wheel_information t_car_information_by_wanji;
  329. t_wanji_manager_task.task_init(TASK_CREATED,"",NULL,std::chrono::milliseconds(3000),
  330. terminal_id, receive_time);
  331. t_error = Task_command_manager::get_instance_references().execute_task(&t_wanji_manager_task);
  332. //第3步, 等待任务单完成
  333. if ( t_error != Error_code::SUCCESS )
  334. {
  335. LOG(INFO) << " Wanji_manager execute_task error "<< this;
  336. }
  337. else
  338. {
  339. //判断任务是否结束, 阻塞 等待, 直到任务完成或者超时
  340. while ( t_wanji_manager_task.is_task_end() == false)
  341. {
  342. if ( t_wanji_manager_task.is_over_time() )
  343. {
  344. //超时处理。取消任务。
  345. Wanji_manager::get_instance_pointer()->cancel_task(&t_wanji_manager_task);
  346. t_error.error_manager_reset(Error_code::WJ_MANAGER_TASK_OVER_TIME, Error_level::MINOR_ERROR,
  347. " t_wanji_manager_task is_over_time ");
  348. t_wanji_manager_task.set_task_error_manager(t_error);
  349. }
  350. else
  351. {
  352. //继续等待
  353. std::this_thread::sleep_for(std::chrono::microseconds(1));
  354. std::this_thread::yield();
  355. }
  356. }
  357. //提取任务单 的错误码
  358. t_error = t_wanji_manager_task.get_task_error_manager();
  359. if ( t_error == Error_code::SUCCESS )
  360. {
  361. t_car_information_by_wanji = t_wanji_manager_task.get_car_wheel_information();
  362. }
  363. else
  364. {
  365. LOG(INFO) << " wanji_manager_task error :::::::" << t_wanji_manager_task.get_task_error_manager().to_string() << this;
  366. }
  367. }
  368. t_result.compare_and_cover_error(t_error);
  369. //measure 测量模块的最终结果
  370. Common_data::Car_measure_information t_car_information_result;
  371. //第4步, 生成反馈消息
  372. if(t_car_information_by_wanji.correctness == true )
  373. {
  374. t_car_information_result.center_x = t_car_information_by_wanji.center_x;
  375. t_car_information_result.center_y = t_car_information_by_wanji.center_y;
  376. t_car_information_result.car_angle = t_car_information_by_wanji.car_angle;
  377. t_car_information_result.wheel_base = t_car_information_by_wanji.wheel_base;
  378. t_car_information_result.wheel_base = t_car_information_result.wheel_base > 3000 ? 3000 : t_car_information_result.wheel_base;
  379. t_car_information_result.wheel_width = t_car_information_by_wanji.wheel_width;
  380. t_car_information_result.correctness = true;
  381. }else{
  382. t_result.set_error_level_down(NEGLIGIBLE_ERROR);
  383. }
  384. //第七步, 创建一条答复消息
  385. message::Ground_detect_response_msg t_ground_detect_response_msg;
  386. t_ground_detect_response_msg.mutable_base_info()->set_msg_type(message::Message_type::eGround_detect_response_msg);
  387. t_ground_detect_response_msg.mutable_base_info()->set_timeout_ms(5000);
  388. t_ground_detect_response_msg.mutable_base_info()->set_sender(message::Communicator::eGround_measurer);
  389. t_ground_detect_response_msg.mutable_base_info()->set_receiver(message::Communicator::eMain);
  390. t_ground_detect_response_msg.set_command_key(command_info);
  391. t_ground_detect_response_msg.set_terminal_id(terminal_id);
  392. t_ground_detect_response_msg.mutable_error_manager()->set_error_code(t_result.get_error_code());
  393. t_ground_detect_response_msg.mutable_error_manager()->set_error_level((message::Error_level)t_result.get_error_level());
  394. t_ground_detect_response_msg.mutable_error_manager()->set_error_description(t_result.get_error_description(), t_result.get_description_length());
  395. t_ground_detect_response_msg.mutable_locate_information()->set_locate_x(t_car_information_result.center_x);
  396. t_ground_detect_response_msg.mutable_locate_information()->set_locate_y(t_car_information_result.center_y);
  397. t_ground_detect_response_msg.mutable_locate_information()->set_locate_angle(t_car_information_result.car_angle);
  398. t_ground_detect_response_msg.mutable_locate_information()->set_locate_length(t_car_information_result.car_length);
  399. t_ground_detect_response_msg.mutable_locate_information()->set_locate_width(t_car_information_result.car_width);
  400. t_ground_detect_response_msg.mutable_locate_information()->set_locate_height(t_car_information_result.car_height);
  401. t_ground_detect_response_msg.mutable_locate_information()->set_locate_wheel_base(t_car_information_result.wheel_base);
  402. t_ground_detect_response_msg.mutable_locate_information()->set_locate_wheel_width(t_car_information_result.wheel_width);
  403. t_ground_detect_response_msg.mutable_locate_information()->set_locate_front_theta(t_car_information_result.front_theta);
  404. t_ground_detect_response_msg.mutable_locate_information()->set_locate_correct(t_car_information_result.correctness);
  405. std::string t_msg = t_ground_detect_response_msg.SerializeAsString();
  406. System_communication::get_instance_references().encapsulate_msg(t_msg);
  407. std::cout << "huli t_measure_response_msg = " << t_ground_detect_response_msg.DebugString() << std::endl;
  408. return ;
  409. }