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                                         process_diagnostic_signals(application_t::instance().get_diagnostic_manager(), message, s);
 185                                 else
 186                                         process_signals(message, s);
 187                         }
 188                         can_message_lock.lock();
 189                 }
 190                 new_decoded_can_message_.notify_one();
 191                 can_message_lock.unlock();
 192         }
 193 }
 194
 195 /// @brief thread to push events to suscribers. It will read subscribed_signals map to look
 196 /// which are events that has to be pushed.
 197 void can_bus_t::can_event_push()
 198 {
 199         json_object* jo;
 200         utils::signals_manager_t& sm = utils::signals_manager_t::instance();
 201
 202         while(is_pushing_)
 203         {
 204                 std::unique_lock<std::mutex> decoded_can_message_lock(decoded_can_message_mutex_);
 205                 new_decoded_can_message_.wait(decoded_can_message_lock);
 206                 while(!vehicle_message_q_.empty())
 207                 {
 208                         std::pair<int, openxc_VehicleMessage> v_message = next_vehicle_message();
 209                         decoded_can_message_lock.unlock();
 210                         {
 211                                 std::lock_guard<std::mutex> subscribed_signals_lock(sm.get_subscribed_signals_mutex());
 212                                 map_subscription& s = sm.get_subscribed_signals();
 213                                 if(s.find(v_message.first) != s.end() && afb_event_is_valid(s[v_message.first]->get_event()))
 214                                 {
 215                                         jo = json_object_new_object();
 216                                         jsonify_vehicle(v_message.second, jo);
 217                                         if(afb_event_push(s[v_message.first]->get_event(), jo) == 0)
 218                                         {
 219                                                 if(v_message.second.has_diagnostic_response)
 220                                                         on_no_clients(s[v_message.first], v_message.second.diagnostic_response.pid, s);
 221                                                 else
 222                                                         on_no_clients(s[v_message.first], s);
 223                                         }
 224                                 }
 225                         }
 226                         decoded_can_message_lock.lock();
 227                 }
 228                 decoded_can_message_lock.unlock();
 229         }
 230 }
 231
 232 /// @brief Will initialize threads that will decode
 233 ///  and push subscribed events.
 234 void can_bus_t::start_threads()
 235 {
 236         is_decoding_ = true;
 237         th_decoding_ = std::thread(&can_bus_t::can_decode_message, this);
 238         th_decoding_.detach();
 239
 240         is_pushing_ = true;
 241         th_pushing_ = std::thread(&can_bus_t::can_event_push, this);
 242         th_pushing_.detach();
 243 }
 244
 245 /// @brief Will stop all threads holded by can_bus_t object
 246 ///  which are decoding and pushing then will wait that's
 247 /// they'll finish their job.
 248 void can_bus_t::stop_threads()
 249 {
 250         is_decoding_ = false;
 251         is_pushing_ = false;
 252 }
 253
 254 /// @brief return new_can_message_cv_ member
 255 ///
 256 /// @return  return new_can_message_cv_ member
 257 std::condition_variable& can_bus_t::get_new_can_message_cv()
 258 {
 259         return new_can_message_cv_;
 260 }
 261
 262 /// @brief return can_message_mutex_ member
 263 ///
 264 /// @return  return can_message_mutex_ member
 265 std::mutex& can_bus_t::get_can_message_mutex()
 266 {
 267         return can_message_mutex_;
 268 }
 269
 270 /// @brief Return first can_message_t on the queue
 271 ///
 272 /// @return a can_message_t
 273 std::shared_ptr<message_t> can_bus_t::next_can_message()
 274 {
 275         std::shared_ptr<message_t> msg;
 276
 277         if(!can_message_q_.empty())
 278         {
 279                 msg = can_message_q_.front();
 280                 can_message_q_.pop();
 281                 std::string debug = msg->get_debug_message();
 282                 AFB_DEBUG(debug.c_str());
 283                 return msg;
 284         }
 285
 286         return msg;
 287 }
 288
 289 /// @brief Push a message_t into the queue
 290 ///
 291 /// @param[in] msg - the const reference message_t object to push into the queue
 292 void can_bus_t::push_new_can_message(std::shared_ptr<message_t> msg)
 293 {
 294         can_message_q_.push(msg);
 295 }
 296
 297 /// @brief Return first openxc_VehicleMessage on the queue
 298 ///
 299 /// @return a openxc_VehicleMessage containing a decoded can message
 300 std::pair<int, openxc_VehicleMessage> can_bus_t::next_vehicle_message()
 301 {
 302         std::pair<int, openxc_VehicleMessage> v_msg;
 303
 304         if(! vehicle_message_q_.empty())
 305         {
 306                 v_msg = vehicle_message_q_.front();
 307                 vehicle_message_q_.pop();
 308                 AFB_DEBUG("next vehicle message poped");
 309                 return v_msg;
 310         }
 311
 312         return v_msg;
 313 }
 314
 315 /// @brief Push a openxc_VehicleMessage into the queue
 316 ///
 317 /// @param[in] v_msg - const reference openxc_VehicleMessage object to push into the queue
 318 void can_bus_t::push_new_vehicle_message(int subscription_id, const openxc_VehicleMessage& v_msg)
 319 {
 320         vehicle_message_q_.push(std::make_pair(subscription_id, v_msg));
 321 }
 322
 323 /// @brief Return the CAN device index from the map
 324 /// map are sorted so index depend upon alphabetical sorting.
 325 int can_bus_t::get_can_device_index(const std::string& bus_name) const
 326 {
 327         int i = 0;
 328         for(const auto& d: can_devices_mapping_)
 329         {
 330                 if(d.first == bus_name)
 331                         break;
 332                 i++;
 333         }
 334         return i;
 335 }
 336
 337 /// @brief Return CAN device name from a logical CAN device name gotten from
 338 /// the signals.json description file which comes from a CAN databases file in
 339 /// general.
 340 const std::string can_bus_t::get_can_device_name(const std::string& id_name) const
 341 {
 342         std::string ret = "";
 343         for(const auto& d: can_devices_mapping_)
 344         {
 345                 if(d.first == id_name)
 346                 {
 347                         ret = d.second;
 348                         break;
 349                 }
 350         }
 351         return ret;
 352 }