Implement minimum and maximum subscription filters
[apps/agl-service-can-low-level.git] / CAN-binder / low-can-binding / can / can-bus.cpp
1 /*
2  * Copyright (C) 2015, 2016 "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 "can-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 extern "C"
37 {
38         #include <afb/afb-binding.h>
39 }
40
41 /// @brief Class constructor
42 ///
43 /// @param[in] conf_file - handle to the json configuration file.
44 can_bus_t::can_bus_t(utils::config_parser_t conf_file)
45         : conf_file_{conf_file}
46 {}
47
48 bool can_bus_t::apply_filter(const openxc_VehicleMessage& vehicle_message, std::shared_ptr<low_can_subscription_t> can_subscription)
49 {
50         bool send = false;
51         if(is_valid(vehicle_message))
52         {
53                 float min = std::isnan(can_subscription->get_min()) ? -INFINITY : can_subscription->get_min();
54                 float max = std::isnan(can_subscription->get_max()) ? INFINITY : can_subscription->get_max();
55                 double value = get_numerical_from_DynamicField(vehicle_message);
56                 send = (value < min && value > max) ? false : true;
57         }
58         return send;
59 }
60
61 /// @brief Will make the decoding operation on a classic CAN message. It will not
62 /// handle CAN commands nor diagnostic messages that have their own method to get
63 /// this happens.
64 ///
65 /// It will add to the vehicle_message queue the decoded message and tell the event push
66 /// thread to process it.
67 ///
68 /// @param[in] can_message - a single CAN message from the CAN socket read, to be decode.
69 ///
70 /// @return How many signals has been decoded.
71 void can_bus_t::process_can_signals(const can_message_t& can_message)
72 {
73         int subscription_id = can_message.get_sub_id();
74         openxc_DynamicField decoded_message;
75         openxc_VehicleMessage vehicle_message;
76         utils::signals_manager_t& sm = utils::signals_manager_t::instance();
77
78         {
79                 std::lock_guard<std::mutex> subscribed_signals_lock(sm.get_subscribed_signals_mutex());
80                 std::map<int, std::shared_ptr<low_can_subscription_t> >& s = sm.get_subscribed_signals();
81
82                 // First we have to found which can_signal_t it is
83                 std::shared_ptr<low_can_subscription_t> sig = s[subscription_id];
84
85                 if( s.find(subscription_id) != s.end() && afb_event_is_valid(s[subscription_id]->get_event()))
86                 {
87                         bool send = true;
88                         decoded_message = decoder_t::translateSignal(*sig->get_can_signal(), can_message, application_t::instance().get_all_can_signals(), &send);
89                         openxc_SimpleMessage s_message = build_SimpleMessage(sig->get_name(), decoded_message);
90                         vehicle_message = build_VehicleMessage(s_message, can_message.get_timestamp());
91
92                         if(send && apply_filter(vehicle_message, sig))
93                         {
94                                 std::lock_guard<std::mutex> decoded_can_message_lock(decoded_can_message_mutex_);
95                                 push_new_vehicle_message(subscription_id, vehicle_message);
96                                 DEBUG(binder_interface, "%s: %s CAN signals processed.", __FUNCTION__,  sig->get_name().c_str());
97                         }
98                 }
99         }
100 }
101
102 /// @brief Will make the decoding operation on a diagnostic CAN message.Then it find the subscribed signal
103 /// corresponding and will add the vehicle_message to the queue of event to pushed before notifying
104 /// the event push thread to process it.
105 ///
106 /// @param[in] manager - the diagnostic manager object that handle diagnostic communication
107 /// @param[in] can_message - a single CAN message from the CAN socket read, to be decode.
108 ///
109 /// @return How many signals has been decoded.
110 void can_bus_t::process_diagnostic_signals(diagnostic_manager_t& manager, const can_message_t& can_message)
111 {
112         int subscription_id = can_message.get_sub_id();
113
114         utils::signals_manager_t& sm = utils::signals_manager_t::instance();
115
116         {
117                 std::lock_guard<std::mutex> subscribed_signals_lock(sm.get_subscribed_signals_mutex());
118                 std::map<int, std::shared_ptr<low_can_subscription_t> >& s = sm.get_subscribed_signals();
119
120                 openxc_VehicleMessage vehicle_message = manager.find_and_decode_adr(can_message);
121                 if( (vehicle_message.has_simple_message && vehicle_message.simple_message.has_name) &&
122                         s.find(subscription_id) != s.end() && afb_event_is_valid(s[subscription_id]->get_event()))
123                 {
124                         if (apply_filter(vehicle_message, s[subscription_id]))
125                         {
126                                 std::lock_guard<std::mutex> decoded_can_message_lock(decoded_can_message_mutex_);
127                                 push_new_vehicle_message(subscription_id, vehicle_message);
128                                 DEBUG(binder_interface, "%s: %s CAN signals processed.", __FUNCTION__,  s[subscription_id]->get_name().c_str());
129                         }
130                 }
131         }
132 }
133
134 /// @brief thread to decoding raw CAN messages.
135 ///
136 ///  Depending on the nature of message, if arbitration ID matches ID for a diagnostic response
137 ///  then decoding a diagnostic message else use classic CAN signals decoding functions.
138 ///
139 /// It will take from the can_message_q_ queue the next can message to process then it search
140 ///  about signal subscribed if there is a valid afb_event for it. We only decode signal for which a
141 ///  subscription has been made. Can message will be decoded using translateSignal that will pass it to the
142 ///  corresponding decoding function if there is one assigned for that signal. If not, it will be the default
143 ///  noopDecoder function that will operate on it.
144 ///
145 ///  TODO: make diagnostic messages parsing optionnal.
146 void can_bus_t::can_decode_message()
147 {
148         while(is_decoding_)
149         {
150                 {
151                         std::unique_lock<std::mutex> can_message_lock(can_message_mutex_);
152                         new_can_message_cv_.wait(can_message_lock);
153                         while(!can_message_q_.empty())
154                         {
155                                 const can_message_t can_message = next_can_message();
156
157                                 if(application_t::instance().get_diagnostic_manager().is_diagnostic_response(can_message))
158                                         process_diagnostic_signals(application_t::instance().get_diagnostic_manager(), can_message);
159                                 else
160                                         process_can_signals(can_message);
161                         }
162                 }
163                 new_decoded_can_message_.notify_one();
164         }
165 }
166
167 /// @brief thread to push events to suscribers. It will read subscribed_signals map to look
168 /// which are events that has to be pushed.
169 void can_bus_t::can_event_push()
170 {
171         std::pair<int, openxc_VehicleMessage> v_message;
172         openxc_SimpleMessage s_message;
173         json_object* jo;
174         utils::signals_manager_t& sm = utils::signals_manager_t::instance();
175
176         while(is_pushing_)
177         {
178                 std::unique_lock<std::mutex> decoded_can_message_lock(decoded_can_message_mutex_);
179                 new_decoded_can_message_.wait(decoded_can_message_lock);
180                 while(!vehicle_message_q_.empty())
181                 {
182                         v_message = next_vehicle_message();
183                         s_message = get_simple_message(v_message.second);
184                         {
185                                 std::lock_guard<std::mutex> subscribed_signals_lock(sm.get_subscribed_signals_mutex());
186                                 std::map<int, std::shared_ptr<low_can_subscription_t> >& s = sm.get_subscribed_signals();
187                                 if(s.find(v_message.first) != s.end() && afb_event_is_valid(s[v_message.first]->get_event()))
188                                 {
189                                         jo = json_object_new_object();
190                                         jsonify_simple(s_message, jo);
191                                         if(afb_event_push(s[v_message.first]->get_event(), jo) == 0)
192                                                 on_no_clients(std::string(s_message.name));
193                                 }
194                         }
195                 }
196         }
197 }
198
199 /// @brief Will initialize threads that will decode
200 ///  and push subscribed events.
201 void can_bus_t::start_threads()
202 {
203         is_decoding_ = true;
204         th_decoding_ = std::thread(&can_bus_t::can_decode_message, this);
205         if(!th_decoding_.joinable())
206                 is_decoding_ = false;
207
208         is_pushing_ = true;
209         th_pushing_ = std::thread(&can_bus_t::can_event_push, this);
210         if(!th_pushing_.joinable())
211                 is_pushing_ = false;
212 }
213
214 /// @brief Will stop all threads holded by can_bus_t object
215 ///  which are decoding and pushing then will wait that's
216 /// they'll finish their job.
217 void can_bus_t::stop_threads()
218 {
219         is_decoding_ = false;
220         is_pushing_ = false;
221 }
222
223 /// @brief return new_can_message_cv_ member
224 ///
225 /// @return  return new_can_message_cv_ member
226 std::condition_variable& can_bus_t::get_new_can_message_cv()
227 {
228         return new_can_message_cv_;
229 }
230
231 /// @brief return can_message_mutex_ member
232 ///
233 /// @return  return can_message_mutex_ member
234 std::mutex& can_bus_t::get_can_message_mutex()
235 {
236         return can_message_mutex_;
237 }
238
239 /// @brief Return first can_message_t on the queue
240 ///
241 /// @return a can_message_t
242 const can_message_t can_bus_t::next_can_message()
243 {
244         can_message_t can_msg;
245
246         if(!can_message_q_.empty())
247         {
248                 can_msg = can_message_q_.front();
249                 can_message_q_.pop();
250                 DEBUG(binder_interface, "%s: Here is the next can message : id %X, length %X, data %02X%02X%02X%02X%02X%02X%02X%02X", __FUNCTION__, can_msg.get_id(), can_msg.get_length(),
251                         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]);
252                 return can_msg;
253         }
254
255         return can_msg;
256 }
257
258 /// @brief Push a can_message_t into the queue
259 ///
260 /// @param[in] can_msg - the const reference can_message_t object to push into the queue
261 void can_bus_t::push_new_can_message(const can_message_t& can_msg)
262 {
263         can_message_q_.push(can_msg);
264 }
265
266 /// @brief Return first openxc_VehicleMessage on the queue
267 ///
268 /// @return a openxc_VehicleMessage containing a decoded can message
269 std::pair<int, openxc_VehicleMessage> can_bus_t::next_vehicle_message()
270 {
271         std::pair<int, openxc_VehicleMessage> v_msg;
272
273         if(! vehicle_message_q_.empty())
274         {
275                 v_msg = vehicle_message_q_.front();
276                 vehicle_message_q_.pop();
277                 DEBUG(binder_interface, "%s: next vehicle message poped", __FUNCTION__);
278                 return v_msg;
279         }
280
281         return v_msg;
282 }
283
284 /// @brief Push a openxc_VehicleMessage into the queue
285 ///
286 /// @param[in] v_msg - const reference openxc_VehicleMessage object to push into the queue
287 void can_bus_t::push_new_vehicle_message(int subscription_id, const openxc_VehicleMessage& v_msg)
288 {
289         vehicle_message_q_.push(std::make_pair(subscription_id, v_msg));
290 }
291
292 /// @brief Return the shared pointer on the can_bus_dev_t initialized 
293 /// with device_name "bus"
294 ///
295 /// @param[in] bus - CAN bus device name to retrieve.
296 ///
297 /// @return A shared pointer on an object can_bus_dev_t
298 void can_bus_t::set_can_devices()
299 {
300         can_devices_ = conf_file_.get_devices_name();
301
302         if(can_devices_.empty())
303         {
304                 ERROR(binder_interface, "%s: No mapping found in config file: '%s'. Check it that it have a CANbus-mapping section.",
305                         __FUNCTION__, conf_file_.filepath().c_str());
306         }
307 }
308
309 int can_bus_t::get_can_device_index(const std::string& bus_name) const
310 {
311         int i = 0;
312         for(const auto& d: can_devices_)
313         {
314                 if(d.first == bus_name)
315                         break;
316                 i++;
317         }
318         return i;
319 }
320
321 const std::string can_bus_t::get_can_device_name(const std::string& id_name) const
322 {
323         std::string ret;
324         for(const auto& d: can_devices_)
325         {
326                 if(d.first == id_name)
327                 {
328                         ret = d.second;
329                         break;
330                 }
331         }
332         return ret;
333 }