low-can-binding/can/can-bus.cpp

   1 /*
   2  * Copyright (C) 2015, 2018 "IoT.bzh"
   3  * Author "Romain Forlot" <romain.forlot@iot.bzh>
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *       http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  */
  17
  18 #include <net/if.h>
  19 #include <sys/socket.h>
  20 #include <json-c/json.h>
  21 #include <linux/can/raw.h>
  22 #include <map>
  23 #include <cerrno>
  24 #include <vector>
  25 #include <string>
  26 #include <algorithm>
  27
  28 #include "can-bus.hpp"
  29
  30 #include "signals.hpp"
  31 #include "can-decoder.hpp"
  32 #include "../binding/application.hpp"
  33 #include "../utils/signals.hpp"
  34 #include "../utils/openxc-utils.hpp"
  35
  36 /// @brief Class destructor
  37 ///
  38 /// @param[in] conf_file - Stop threads and unlock them to correctly finish them
  39 /// even without any activity on the CAN bus.
  40 can_bus_t::~can_bus_t()
  41 {
  42         stop_threads();
  43         new_can_message_cv_.notify_one();
  44 }
  45
  46 /// @brief Class constructor
  47 can_bus_t::can_bus_t()
  48 {}
  49
  50 /// @brief Fills the CAN device map member with value from device
  51 ///
  52 /// @param[in] mapping configuration section.
  53 void can_bus_t::set_can_devices(json_object *mapping)
  54 {
  55         if (! mapping)
  56         {
  57                 AFB_ERROR("Can't initialize CAN buses with this empty mapping.");
  58                 return;
  59         }
  60
  61         struct json_object_iterator it = json_object_iter_begin(mapping);
  62         struct json_object_iterator end = json_object_iter_end(mapping);
  63         while (! json_object_iter_equal(&it, &end)) {
  64                 can_devices_mapping_.push_back(std::make_pair(
  65                         json_object_iter_peek_name(&it),
  66                         json_object_get_string(json_object_iter_peek_value(&it))
  67                         ));
  68                 json_object_iter_next(&it);
  69         }
  70 }
  71
  72 /// @brief Take a decoded message to determine if its value complies with the desired
  73 /// filters.
  74 ///
  75 /// @param[in] vehicle_message - The decoded message to be analyzed.
  76 /// @param[in] can_subscription - the subscription which will be notified depending
  77 ///  on its filtering values. Filtering values are stored in the event_filtermember.
  78 ///
  79 /// @return True if the value is compliant with event filter values, false if not...
  80 bool can_bus_t::apply_filter(const openxc_VehicleMessage& vehicle_message, std::shared_ptr<low_can_subscription_t> can_subscription)
  81 {
  82         bool send = false;
  83         if(is_valid(vehicle_message))
  84         {
  85                 float min = can_subscription->get_min();
  86                 float max = can_subscription->get_max();
  87                 double value = get_numerical_from_DynamicField(vehicle_message);
  88                 send = (value < min || value > max) ? false : true;
  89         }
  90         return send;
  91 }
  92
  93 /// @brief Will make the decoding operation on a classic CAN message. It will not
  94 /// handle CAN commands nor diagnostic messages that have their own method to get
  95 /// this happens.
  96 ///
  97 /// It will add to the vehicle_message queue the decoded message and tell the event push
  98 /// thread to process it.
  99 ///
 100 /// @param[in] can_message - a single CAN message from the CAN socket read, to be decode.
 101 ///
 102 /// @return How many signals has been decoded.
 103 void can_bus_t::process_signals(std::shared_ptr<message_t> message, map_subscription& s)
 104 {
 105         int subscription_id = message->get_sub_id();
 106         openxc_DynamicField decoded_message;
 107         openxc_VehicleMessage vehicle_message;
 108
 109         if( s.find(subscription_id) != s.end() && afb_event_is_valid(s[subscription_id]->get_event()))
 110         {
 111                 bool send = true;
 112                 // First we have to found which signal_t it is
 113                 std::shared_ptr<low_can_subscription_t> sig = s[subscription_id];
 114
 115                 decoded_message = decoder_t::translate_signal(*sig->get_signal(), message, &send);
 116                 openxc_SimpleMessage s_message = build_SimpleMessage(sig->get_name(), decoded_message);
 117                 vehicle_message = build_VehicleMessage(s_message, message->get_timestamp());
 118
 119                 if(send && apply_filter(vehicle_message, sig))
 120                 {
 121                         std::lock_guard<std::mutex> decoded_can_message_lock(decoded_can_message_mutex_);
 122                         push_new_vehicle_message(subscription_id, vehicle_message);
 123                         AFB_DEBUG("%s CAN signals processed.", sig->get_name().c_str());
 124                 }
 125         }
 126 }
 127
 128 /// @brief Will make the decoding operation on a diagnostic CAN message.Then it find the subscribed signal
 129 /// corresponding and will add the vehicle_message to the queue of event to pushed before notifying
 130 /// the event push thread to process it.
 131 ///
 132 /// @param[in] manager - the diagnostic manager object that handle diagnostic communication
 133 /// @param[in] can_message - a single CAN message from the CAN socket read, to be decode.
 134 ///
 135 /// @return How many signals has been decoded.
 136 void can_bus_t::process_diagnostic_signals(diagnostic_manager_t& manager, std::shared_ptr<message_t> message, map_subscription& s)
 137 {
 138         int subscription_id = message->get_sub_id();
 139
 140         openxc_VehicleMessage vehicle_message = manager.find_and_decode_adr(message);
 141         if (message->get_timestamp())
 142                 vehicle_message.timestamp = message->get_timestamp();
 143         if( (vehicle_message.has_simple_message && vehicle_message.simple_message.has_name) &&
 144                 s.find(subscription_id) != s.end() && afb_event_is_valid(s[subscription_id]->get_event()))
 145         {
 146                 if (apply_filter(vehicle_message, s[subscription_id]))
 147                 {
 148                         std::lock_guard<std::mutex> decoded_can_message_lock(decoded_can_message_mutex_);
 149                         push_new_vehicle_message(subscription_id, vehicle_message);
 150                         AFB_DEBUG("%s CAN signals processed.",  s[subscription_id]->get_name().c_str());
 151                 }
 152         }
 153 }
 154
 155 /// @brief thread to decoding raw CAN messages.
 156 ///
 157 ///  Depending on the nature of message, if arbitration ID matches ID for a diagnostic response
 158 ///  then decoding a diagnostic message else use classic CAN signals decoding functions.
 159 ///
 160 /// It will take from the can_message_q_ queue the next can message to process then it search
 161 ///  about signal subscribed if there is a valid afb_event for it. We only decode signal for which a
 162 ///  subscription has been made. Can message will be decoded using translate_signal that will pass it to the
 163 ///  corresponding decoding function if there is one assigned for that signal. If not, it will be the default
 164 ///  noopDecoder function that will operate on it.
 165 ///
 166 ///  TODO: make diagnostic messages parsing optionnal.
 167 void can_bus_t::can_decode_message()
 168 {
 169         utils::signals_manager_t& sm = utils::signals_manager_t::instance();
 170
 171         while(is_decoding_)
 172         {
 173                 std::unique_lock<std::mutex> can_message_lock(can_message_mutex_);
 174                 new_can_message_cv_.wait(can_message_lock);
 175                 while(!can_message_q_.empty())
 176                 {
 177                         std::shared_ptr<message_t>  message = next_can_message();
 178                         can_message_lock.unlock();
 179
 180                         {
 181                                 std::lock_guard<std::mutex> subscribed_signals_lock(sm.get_subscribed_signals_mutex());
 182                                 map_subscription& s = sm.get_subscribed_signals();
 183                                 if(application_t::instance().get_diagnostic_manager().is_diagnostic_response(message))
 184                                 {
 185                                         process_diagnostic_signals(application_t::instance().get_diagnostic_manager(), message, s);
 186                                 }
 187                                 else
 188                                         process_signals(message, s);
 189                         }
 190                         can_message_lock.lock();
 191                 }
 192                 new_decoded_can_message_.notify_one();
 193                 can_message_lock.unlock();
 194         }
 195 }
 196
 197 /// @brief thread to push events to suscribers. It will read subscribed_signals map to look
 198 /// which are events that has to be pushed.
 199 void can_bus_t::can_event_push()
 200 {
 201         json_object* jo;
 202         utils::signals_manager_t& sm = utils::signals_manager_t::instance();
 203
 204         while(is_pushing_)
 205         {
 206                 std::unique_lock<std::mutex> decoded_can_message_lock(decoded_can_message_mutex_);
 207                 new_decoded_can_message_.wait(decoded_can_message_lock);
 208                 while(!vehicle_message_q_.empty())
 209                 {
 210                         std::pair<int, openxc_VehicleMessage> v_message = next_vehicle_message();
 211                         decoded_can_message_lock.unlock();
 212                         {
 213                                 std::lock_guard<std::mutex> subscribed_signals_lock(sm.get_subscribed_signals_mutex());
 214                                 map_subscription& s = sm.get_subscribed_signals();
 215                                 if(s.find(v_message.first) != s.end() && afb_event_is_valid(s[v_message.first]->get_event()))
 216                                 {
 217                                         jo = json_object_new_object();
 218                                         jsonify_vehicle(v_message.second, jo);
 219                                         if(afb_event_push(s[v_message.first]->get_event(), jo) == 0)
 220                                         {
 221                                                 if(v_message.second.has_diagnostic_response)
 222                                                         on_no_clients(s[v_message.first], v_message.second.diagnostic_response.pid, s);
 223                                                 else
 224                                                         on_no_clients(s[v_message.first], s);
 225                                         }
 226                                 }
 227                         }
 228                         decoded_can_message_lock.lock();
 229                 }
 230                 decoded_can_message_lock.unlock();
 231         }
 232 }
 233
 234 /// @brief Will initialize threads that will decode
 235 ///  and push subscribed events.
 236 void can_bus_t::start_threads()
 237 {
 238         is_decoding_ = true;
 239         th_decoding_ = std::thread(&can_bus_t::can_decode_message, this);
 240         th_decoding_.detach();
 241
 242         is_pushing_ = true;
 243         th_pushing_ = std::thread(&can_bus_t::can_event_push, this);
 244         th_pushing_.detach();
 245 }
 246
 247 /// @brief Will stop all threads holded by can_bus_t object
 248 ///  which are decoding and pushing then will wait that's
 249 /// they'll finish their job.
 250 void can_bus_t::stop_threads()
 251 {
 252         is_decoding_ = false;
 253         is_pushing_ = false;
 254 }
 255
 256 /// @brief return new_can_message_cv_ member
 257 ///
 258 /// @return  return new_can_message_cv_ member
 259 std::condition_variable& can_bus_t::get_new_can_message_cv()
 260 {
 261         return new_can_message_cv_;
 262 }
 263
 264 /// @brief return can_message_mutex_ member
 265 ///
 266 /// @return  return can_message_mutex_ member
 267 std::mutex& can_bus_t::get_can_message_mutex()
 268 {
 269         return can_message_mutex_;
 270 }
 271
 272 /// @brief Return first can_message_t on the queue
 273 ///
 274 /// @return a can_message_t
 275 std::shared_ptr<message_t> can_bus_t::next_can_message()
 276 {
 277         std::shared_ptr<message_t> msg;
 278
 279         if(!can_message_q_.empty())
 280         {
 281                 msg = can_message_q_.front();
 282                 can_message_q_.pop();
 283                 std::string debug = msg->get_debug_message();
 284                 AFB_DEBUG(debug.c_str());
 285                 return msg;
 286         }
 287
 288         return msg;
 289 }
 290
 291 /// @brief Push a message_t into the queue
 292 ///
 293 /// @param[in] msg - the const reference message_t object to push into the queue
 294 void can_bus_t::push_new_can_message(std::shared_ptr<message_t> msg)
 295 {
 296         can_message_q_.push(msg);
 297 }
 298
 299 /// @brief Return first openxc_VehicleMessage on the queue
 300 ///
 301 /// @return a openxc_VehicleMessage containing a decoded can message
 302 std::pair<int, openxc_VehicleMessage> can_bus_t::next_vehicle_message()
 303 {
 304         std::pair<int, openxc_VehicleMessage> v_msg;
 305
 306         if(! vehicle_message_q_.empty())
 307         {
 308                 v_msg = vehicle_message_q_.front();
 309                 vehicle_message_q_.pop();
 310                 AFB_DEBUG("next vehicle message poped");
 311                 return v_msg;
 312         }
 313
 314         return v_msg;
 315 }
 316
 317 /// @brief Push a openxc_VehicleMessage into the queue
 318 ///
 319 /// @param[in] v_msg - const reference openxc_VehicleMessage object to push into the queue
 320 void can_bus_t::push_new_vehicle_message(int subscription_id, const openxc_VehicleMessage& v_msg)
 321 {
 322         vehicle_message_q_.push(std::make_pair(subscription_id, v_msg));
 323 }
 324
 325 /// @brief Return the CAN device index from the map
 326 /// map are sorted so index depend upon alphabetical sorting.
 327 int can_bus_t::get_can_device_index(const std::string& bus_name) const
 328 {
 329         int i = 0;
 330         for(const auto& d: can_devices_mapping_)
 331         {
 332                 if(d.first == bus_name)
 333                         break;
 334                 i++;
 335         }
 336         return i;
 337 }
 338
 339 /// @brief Return CAN device name from a logical CAN device name gotten from
 340 /// the signals.json description file which comes from a CAN databases file in
 341 /// general.
 342 const std::string can_bus_t::get_can_device_name(const std::string& id_name) const
 343 {
 344         std::string ret = "";
 345         for(const auto& d: can_devices_mapping_)
 346         {
 347                 if(d.first == id_name)
 348                 {
 349                         ret = d.second;
 350                         break;
 351                 }
 352         }
 353         return ret;
 354 }