X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=src%2Fcan%2Fcan-bus.cpp;h=cfb9d86d63b5c292531596f2dfb5c81012deff7e;hb=c0fd6e1982353d056c26a292186ef0df315adde3;hp=e92e2cbc9c23a8811b13a13f8fb978594ca4d6a8;hpb=9519e9d5b875b45fbf6f32267b728b1d11377276;p=apps%2Flow-level-can-service.git diff --git a/src/can/can-bus.cpp b/src/can/can-bus.cpp index e92e2cb..cfb9d86 100644 --- a/src/can/can-bus.cpp +++ b/src/can/can-bus.cpp @@ -27,98 +27,144 @@ #include #include -#include "can/can-decoder.hpp" -#include "utils/openxc-utils.hpp" +#include "can-bus.hpp" + +#include "can-signals.hpp" +#include "can-decoder.hpp" +#include "../configuration.hpp" +#include "../utils/signals.hpp" +#include "../utils/openxc-utils.hpp" extern "C" { #include } -#include "can/can-bus.hpp" - -/******************************************************************************** -* -* can_bus_t method implementation -* -*********************************************************************************/ -/** -* @brief Class constructor -* -* @param struct afb_binding_interface *interface between daemon and binding -* @param int file handle to the json configuration file. -*/ +/// @brief Class constructor +/// +/// @param[in] conf_file - handle to the json configuration file. can_bus_t::can_bus_t(int conf_file) : conf_file_{conf_file} { } +std::map> can_bus_t::can_devices_; + +/// @brief Will make the decoding operation on a classic CAN message. It will not +/// handle CAN commands nor diagnostic messages that have their own method to get +/// this happens. +/// +/// It will add to the vehicle_message queue the decoded message and tell the event push +/// thread to process it. +/// +/// @param[in] can_message - a single CAN message from the CAN socket read, to be decode. +/// +/// @return How many signals has been decoded. +int can_bus_t::process_can_signals(can_message_t& can_message) +{ + int processed_signals = 0; + std::vector signals; + openxc_DynamicField search_key, decoded_message; + openxc_VehicleMessage vehicle_message; + + // First we have to found which can_signal_t it is + search_key = build_DynamicField((double)can_message.get_id()); + configuration_t::instance().find_can_signals(search_key, signals); + + // Decoding the message ! Don't kill the messenger ! + for(auto& sig : signals) + { + std::lock_guard subscribed_signals_lock(get_subscribed_signals_mutex()); + std::map& s = get_subscribed_signals(); + + // DEBUG message to make easier debugger STL containers... + //DEBUG(binder_interface, "Operator[] key char: %s, event valid? %d", sig.generic_name, afb_event_is_valid(s[sig.generic_name])); + //DEBUG(binder_interface, "Operator[] key string: %s, event valid? %d", sig.generic_name, afb_event_is_valid(s[std::string(sig.generic_name)])); + //DEBUG(binder_interface, "Nb elt matched char: %d", (int)s.count(sig.generic_name)); + //DEBUG(binder_interface, "Nb elt matched string: %d", (int)s.count(std::string(sig.generic_name)); + if( s.find(sig->get_name()) != s.end() && afb_event_is_valid(s[sig->get_name()])) + { + decoded_message = decoder_t::translateSignal(*sig, can_message, configuration_t::instance().get_can_signals()); + + openxc_SimpleMessage s_message = build_SimpleMessage(sig->get_name(), decoded_message); + vehicle_message = build_VehicleMessage(s_message); + + std::lock_guard decoded_can_message_lock(decoded_can_message_mutex_); + push_new_vehicle_message(vehicle_message); + new_decoded_can_message_.notify_one(); + processed_signals++; + } + } + + DEBUG(binder_interface, "process_can_signals: %d/%d CAN signals processed.", processed_signals, (int)signals.size()); + return processed_signals; +} + +/// @brief Will make the decoding operation on a diagnostic CAN message.Then it find the subscribed signal +/// corresponding and will add the vehicle_message to the queue of event to pushed before notifying +/// the event push thread to process it. +/// +/// @param[in] manager - the diagnostic manager object that handle diagnostic communication +/// @param[in] can_message - a single CAN message from the CAN socket read, to be decode. +/// +/// @return How many signals has been decoded. +int can_bus_t::process_diagnostic_signals(diagnostic_manager_t& manager, const can_message_t& can_message) +{ + int processed_signals = 0; + + std::lock_guard subscribed_signals_lock(get_subscribed_signals_mutex()); + std::map& s = get_subscribed_signals(); + + openxc_VehicleMessage vehicle_message = manager.find_and_decode_adr(can_message); + if( (vehicle_message.has_simple_message && vehicle_message.simple_message.has_name) && + (s.find(vehicle_message.simple_message.name) != s.end() && afb_event_is_valid(s[vehicle_message.simple_message.name]))) + { + std::lock_guard decoded_can_message_lock(decoded_can_message_mutex_); + push_new_vehicle_message(vehicle_message); + new_decoded_can_message_.notify_one(); + processed_signals++; + } + + return processed_signals; +} -/** -* @brief thread to decoding raw CAN messages. -* -* @desc It will take from the can_message_q_ queue the next can message to process then it will search -* about signal subscribed if there is a valid afb_event for it. We only decode signal for which a -* subscription has been made. Can message will be decoded using translateSignal that will pass it to the -* corresponding decoding function if there is one assigned for that signal. If not, it will be the default -* noopDecoder function that will operate on it. -*/ +/// @brief thread to decoding raw CAN messages. +/// +/// Depending on the nature of message, if arbitration ID matches ID for a diagnostic response +/// then decoding a diagnostic message else use classic CAN signals decoding functions. +/// +/// It will take from the can_message_q_ queue the next can message to process then it search +/// about signal subscribed if there is a valid afb_event for it. We only decode signal for which a +/// subscription has been made. Can message will be decoded using translateSignal that will pass it to the +/// corresponding decoding function if there is one assigned for that signal. If not, it will be the default +/// noopDecoder function that will operate on it. +/// +/// TODO: make diagnostic messages parsing optionnal. void can_bus_t::can_decode_message() { can_message_t can_message; - std::vector signals; - openxc_VehicleMessage vehicle_message; - openxc_DynamicField search_key, decoded_message; - - decoder_t decoder; while(is_decoding_) { std::unique_lock can_message_lock(can_message_mutex_); new_can_message_cv_.wait(can_message_lock); can_message = next_can_message(); - - /* First we have to found which CanSignal it is */ - search_key = build_DynamicField((double)can_message.get_id()); - signals.clear(); - find_can_signals(search_key, signals); - - /* Decoding the message ! Don't kill the messenger ! */ - for(auto& sig : signals) - { - std::lock_guard subscribed_signals_lock(get_subscribed_signals_mutex()); - std::map& s = get_subscribed_signals(); - - /* DEBUG message to make easier debugger STL containers... - DEBUG(binder_interface, "Operator[] key char: %s, event valid? %d", sig.generic_name, afb_event_is_valid(s[sig.generic_name])); - DEBUG(binder_interface, "Operator[] key string: %s, event valid? %d", sig.generic_name, afb_event_is_valid(s[std::string(sig.generic_name)])); - DEBUG(binder_interface, "Nb elt matched char: %d", (int)s.count(sig.generic_name)); - DEBUG(binder_interface, "Nb elt matched string: %d", (int)s.count(std::string(sig.generic_name)));*/ - if( s.find(sig->generic_name) != s.end() && afb_event_is_valid(s[sig->generic_name])) - { - decoded_message = decoder.translateSignal(*sig, can_message, get_can_signals()); - - openxc_SimpleMessage s_message = build_SimpleMessage(sig->generic_name, decoded_message); - vehicle_message = build_VehicleMessage_with_SimpleMessage(openxc_DynamicField_Type::openxc_DynamicField_Type_NUM, s_message); - std::lock_guard decoded_can_message_lock(decoded_can_message_mutex_); - push_new_vehicle_message(vehicle_message); - new_decoded_can_message_.notify_one(); - } - } + if(configuration_t::instance().get_diagnostic_manager().is_diagnostic_response(can_message)) + process_diagnostic_signals(configuration_t::instance().get_diagnostic_manager(), can_message); + else + process_can_signals(can_message); } } -/** -* @brief thread to push events to suscribers. It will read subscribed_signals map to look -* which are events that has to be pushed. -*/ +/// @brief thread to push events to suscribers. It will read subscribed_signals map to look +/// which are events that has to be pushed. void can_bus_t::can_event_push() { openxc_VehicleMessage v_message; openxc_SimpleMessage s_message; json_object* jo; - + while(is_pushing_) { std::unique_lock decoded_can_message_lock(decoded_can_message_mutex_); @@ -139,42 +185,42 @@ void can_bus_t::can_event_push() } } -/** - * @brief Will initialize threads that will decode - * and push subscribed events. - */ +/// @brief Will initialize threads that will decode +/// and push subscribed events. void can_bus_t::start_threads() { is_decoding_ = true; th_decoding_ = std::thread(&can_bus_t::can_decode_message, this); if(!th_decoding_.joinable()) is_decoding_ = false; - + is_pushing_ = true; th_pushing_ = std::thread(&can_bus_t::can_event_push, this); if(!th_pushing_.joinable()) is_pushing_ = false; } -/** -* @brief Will stop all threads holded by can_bus_t object -* which are decoding and pushing then will wait that's -* they'll finish their job. -*/ +/// @brief Will stop all threads holded by can_bus_t object +/// which are decoding and pushing then will wait that's +/// they'll finish their job. void can_bus_t::stop_threads() { is_decoding_ = false; is_pushing_ = false; } -/** -* @brief Will initialize can_bus_dev_t objects after reading -* the configuration file passed in the constructor. -*/ +/// @brief Will initialize can_bus_dev_t objects after reading +/// the configuration file passed in the constructor. All CAN buses +/// Initialized here will be added to a vector holding them for +/// inventory and later access. +/// +/// That will initialize CAN socket reading too using a new thread. +/// +/// @return 0 if ok, other if not. int can_bus_t::init_can_dev() { std::vector devices_name; - int i; + int i = 0; size_t t; devices_name = read_conf(); @@ -182,20 +228,22 @@ int can_bus_t::init_can_dev() if (! devices_name.empty()) { t = devices_name.size(); - i=0; for(const auto& device : devices_name) { - can_devices_m_[device] = std::make_shared(device); - if (can_devices_m_[device]->open() == 0) + can_bus_t::can_devices_[device] = std::make_shared(device, i); + if (can_bus_t::can_devices_[device]->open() == 0) { - i++; DEBUG(binder_interface, "Start reading thread"); NOTICE(binder_interface, "%s device opened and reading", device.c_str()); - can_devices_m_[device]->start_reading(*this); + can_bus_t::can_devices_[device]->start_reading(*this); + i++; } else + { ERROR(binder_interface, "Can't open device %s", device.c_str()); + return 1; + } } NOTICE(binder_interface, "Initialized %d/%d can bus device(s)", i, t); @@ -205,12 +253,10 @@ int can_bus_t::init_can_dev() return 1; } -/** -* @brief read the conf_file_ and will parse json objects -* in it searching for canbus objects devices name. -* -* @return Vector of can bus device name string. -*/ +/// @brief read the conf_file_ and will parse json objects +/// in it searching for canbus objects devices name. +/// +/// @return Vector of can bus device name string. std::vector can_bus_t::read_conf() { std::vector ret; @@ -232,15 +278,7 @@ std::vector can_bus_t::read_conf() jo = json_tokener_parse(fd_conf_content.c_str()); if (jo == NULL || !json_object_object_get_ex(jo, "canbus", &canbus)) - {/** -* @brief Telling if the pushing thread is running -* This is the boolean value on which the while loop -* take its condition. Set it to false will stop the -* according thread. -* -* @return true if pushing thread is running, false if not. -*/ - + { ERROR(binder_interface, "Can't find canbus node in the configuration file. Please review it."); ret.clear(); } @@ -263,31 +301,25 @@ std::vector can_bus_t::read_conf() return ret; } -/** -* @brief return new_can_message_cv_ member -* -* @return return new_can_message_cv_ member -*/ +/// @brief return new_can_message_cv_ member +/// +/// @return return new_can_message_cv_ member std::condition_variable& can_bus_t::get_new_can_message_cv() { return new_can_message_cv_; } -/** -* @brief return can_message_mutex_ member -* -* @return return can_message_mutex_ member -*/ +/// @brief return can_message_mutex_ member +/// +/// @return return can_message_mutex_ member std::mutex& can_bus_t::get_can_message_mutex() { return can_message_mutex_; } -/** -* @brief Return first can_message_t on the queue -* -* @return a can_message_t -*/ +/// @brief Return first can_message_t on the queue +/// +/// @return a can_message_t can_message_t can_bus_t::next_can_message() { can_message_t can_msg; @@ -300,25 +332,21 @@ can_message_t can_bus_t::next_can_message() can_msg.get_data()[0], can_msg.get_data()[1], can_msg.get_data()[2], can_msg.get_data()[3], can_msg.get_data()[4], can_msg.get_data()[5], can_msg.get_data()[6], can_msg.get_data()[7]); return can_msg; } - + return can_msg; } -/** -* @brief Push a can_message_t into the queue -* -* @param the const reference can_message_t object to push into the queue -*/ +/// @brief Push a can_message_t into the queue +/// +/// @param[in] can_msg - the const reference can_message_t object to push into the queue void can_bus_t::push_new_can_message(const can_message_t& can_msg) { can_message_q_.push(can_msg); } -/** -* @brief Return first openxc_VehicleMessage on the queue -* -* @return a openxc_VehicleMessage containing a decoded can message -*/ +/// @brief Return first openxc_VehicleMessage on the queue +/// +/// @return a openxc_VehicleMessage containing a decoded can message openxc_VehicleMessage can_bus_t::next_vehicle_message() { openxc_VehicleMessage v_msg; @@ -330,159 +358,33 @@ openxc_VehicleMessage can_bus_t::next_vehicle_message() DEBUG(binder_interface, "next_vehicle_message: next vehicle message poped"); return v_msg; } - + return v_msg; } -/** -* @brief Push a openxc_VehicleMessage into the queue -* -* @param the const reference openxc_VehicleMessage object to push into the queue -*/ +/// @brief Push a openxc_VehicleMessage into the queue +/// +/// @param[in] v_msg - const reference openxc_VehicleMessage object to push into the queue void can_bus_t::push_new_vehicle_message(const openxc_VehicleMessage& v_msg) { vehicle_message_q_.push(v_msg); } -/** -* @brief Return a map with the can_bus_dev_t initialized -* -* @return map can_bus_dev_m_ map -*/ -std::map> can_bus_t::get_can_devices() -{ - return can_devices_m_; -} - -/******************************************************************************** -* -* can_bus_dev_t method implementation -* -*********************************************************************************/ - -/** - * @brief Open the can socket and returning it - * - * @return - */ -int can_bus_dev_t::close() -{ - ::close(can_socket_); - can_socket_ = -1; - return can_socket_; -} - -/** -* @brief Read the can socket and retrieve canfd_frame -* -* @param const struct afb_binding_interface* interface pointer. Used to be able to log -* using application framework logger. -*/ -std::pair can_bus_dev_t::read() -{ - ssize_t nbytes; - //int maxdlen; - struct canfd_frame cfd; - - /* Test that socket is really opened */ - if (can_socket_ < 0) - { - ERROR(binder_interface, "read_can: Socket unavailable. Closing thread."); - is_running_ = false; - } - - nbytes = ::read(can_socket_, &cfd, CANFD_MTU); - - /* if we did not fit into CAN sized messages then stop_reading. */ - if (nbytes != CANFD_MTU && nbytes != CAN_MTU) - { - if (errno == ENETDOWN) - ERROR(binder_interface, "read: %s CAN device down", device_name_); - ERROR(binder_interface, "read: Incomplete CAN(FD) frame"); - ::memset(&cfd, 0, sizeof(cfd)); - } - - DEBUG(binder_interface, "read: Found id: %X, length: %X, data %02X%02X%02X%02X%02X%02X%02X%02X", cfd.can_id, cfd.len, - cfd.data[0], cfd.data[1], cfd.data[2], cfd.data[3], cfd.data[4], cfd.data[5], cfd.data[6], cfd.data[7]); - return std::pair(cfd, nbytes); -} - -/** -* @brief start reading threads and set flag is_running_ -* -* @param can_bus_t reference can_bus_t. it will be passed to the thread -* to allow using can_bus_t queue. -*/ -void can_bus_dev_t::start_reading(can_bus_t& can_bus) -{ - DEBUG(binder_interface, "Launching reading thread"); - is_running_ = true; - th_reading_ = std::thread(&can_bus_dev_t::can_reader, this, std::ref(can_bus)); - if(!th_reading_.joinable()) - is_running_ = false; -} - -/** -* @brief stop the reading thread setting flag is_running_ to false and -* and wait that the thread finish its job. -*/ -void can_bus_dev_t::stop_reading() -{ - is_running_ = false; -} - -/** -* -* @brief Thread function used to read the can socket. -* -* @param[in] can_bus_dev_t object to be used to read the can socket -* @param[in] can_bus_t object used to fill can_message_q_ queue -*/ -void can_bus_dev_t::can_reader(can_bus_t& can_bus) +/// @brief Return a map with the can_bus_dev_t initialized +/// +/// @return map can_bus_dev_m_ map +const std::map>& can_bus_t::get_can_devices() const { - can_message_t can_message; - - while(is_running_) - { - can_message.convert_from_canfd_frame(read()); - - { - std::lock_guard can_message_lock(can_bus.get_can_message_mutex()); - can_bus.push_new_can_message(can_message); - } - can_bus.get_new_can_message_cv().notify_one(); - } + return can_bus_t::can_devices_; } -/** -* @brief Send a can message from a can_message_t object. -* -* @param const can_message_t& can_msg: the can message object to send -* @param const struct afb_binding_interface* interface pointer. Used to be able to log -* using application framework logger. -*/ -int can_bus_dev_t::send_can_message(can_message_t& can_msg) +/// @brief Return the shared pointer on the can_bus_dev_t initialized +/// with device_name "bus" +/// +/// @param[in] bus - CAN bus device name to retrieve. +/// +/// @return A shared pointer on an object can_bus_dev_t +std::shared_ptr can_bus_t::get_can_device(std::string bus) { - ssize_t nbytes; - canfd_frame f; - - f = can_msg.convert_to_canfd_frame(); - - if(can_socket_ >= 0) - { - nbytes = ::sendto(can_socket_, &f, sizeof(struct canfd_frame), 0, - (struct sockaddr*)&txAddress_, sizeof(txAddress_)); - if (nbytes == -1) - { - ERROR(binder_interface, "send_can_message: Sending CAN frame failed."); - return -1; - } - return (int)nbytes; - } - else - { - ERROR(binder_interface, "send_can_message: socket not initialized. Attempt to reopen can device socket."); - open(); - } - return 0; + return can_bus_t::can_devices_[bus]; }