Get decoding diagnostic request from decoding thread of can_bus_t

Decoding divided in 2 subfunctions dedicated to decode either
pure CAN messages or diagnostic (obd2) message. About now,
a diagnostic request has a name then it will be pushed
on the event_queue as a SimpleMessage. Without name full
details of diagnostic response will be pushed as diagnostic
response. This behavior follows the one from OpenXC.

Change-Id: I255f3f6487fa9511ea47c74606014995a7b0f720
Signed-off-by: Romain Forlot <romain.forlot@iot.bzh>

8 files changed, 224 insertions, 51 deletions

diff --git a/src/can/can-bus-dev.cpp b/src/can/can-bus-dev.cpp
index 5c8673c8..a574b42d 100644
--- a/src/can/can-bus-dev.cpp
+++ b/src/can/can-bus-dev.cpp
@@ -42,6 +42,11 @@ std::string can_bus_dev_t::get_device_name() const
 	return device_name_;
 }
 
+uint32_t can_bus_dev_t::get_address() const
+{
+	return address_;
+}
+
 /// @brief Open the can socket and returning it
 /// @return -1 if something wrong.
 int can_bus_dev_t::open()
diff --git a/src/can/can-bus-dev.hpp b/src/can/can-bus-dev.hpp
index c0ef8daa..63e61631 100644
--- a/src/can/can-bus-dev.hpp
+++ b/src/can/can-bus-dev.hpp
@@ -49,6 +49,7 @@ public:
 	can_bus_dev_t(const std::string& dev_name, int32_t address);
 
 	std::string get_device_name() const;
+	uint32_t get_address() const;
 
 	int open();
 	int close();
diff --git a/src/can/can-bus.cpp b/src/can/can-bus.cpp
index 79fcd3fc..dfecea96 100644
--- a/src/can/can-bus.cpp
+++ b/src/can/can-bus.cpp
@@ -51,6 +51,87 @@ can_bus_t::can_bus_t(int conf_file)
 }
 
 
+int can_bus_t::process_can_signals(can_message_t& can_message)
+{
+	int processed_signals = 0;
+	std::vector <can_signal_t*> 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());
+	signals.clear();
+	configuration_t::instance().find_can_signals(search_key, signals);
+
+	/* Decoding the message ! Don't kill the messenger ! */
+	for(auto& sig : signals)
+	{
+		std::lock_guard<std::mutex> subscribed_signals_lock(get_subscribed_signals_mutex());
+		std::map<std::string, struct afb_event>& 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_generic_name(), decoded_message);
+			vehicle_message = build_VehicleMessage(s_message);
+
+			std::lock_guard<std::mutex> 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;
+}
+
+int can_bus_t::process_diagnostic_signals(active_diagnostic_request_t* entry, const can_message_t& can_message)
+{
+	int processed_signals = 0;
+	openxc_VehicleMessage vehicle_message;
+
+	diagnostic_manager_t& manager = configuration_t::instance().get_diagnostic_manager();
+	
+	std::lock_guard<std::mutex> subscribed_signals_lock(get_subscribed_signals_mutex());
+	std::map<std::string, struct afb_event>& s = get_subscribed_signals();
+
+	if( s.find(entry->get_name()) != s.end() && afb_event_is_valid(s[entry->get_name()]))
+	{
+		if(manager.get_can_bus_dev() == entry->get_can_bus_dev() && entry->get_in_flight())
+		{
+			DiagnosticResponse response = diagnostic_receive_can_frame(
+					// TODO: openXC todo task: eek, is bus address and array index this tightly coupled?
+					&manager.get_shims(),
+					entry->get_handle(), can_message.get_id(), can_message.get_data(), can_message.get_length());
+			if(response.completed && entry->get_handle()->completed)
+			{
+				if(entry->get_handle()->success)
+				{
+					vehicle_message = manager.relay_diagnostic_response(entry, response);
+					std::lock_guard<std::mutex> decoded_can_message_lock(decoded_can_message_mutex_);
+					push_new_vehicle_message(vehicle_message);
+					new_decoded_can_message_.notify_one();
+					processed_signals++;
+				}
+				else
+					DEBUG(binder_interface, "Fatal error sending or receiving diagnostic request");
+			}
+			else if(!response.completed && response.multi_frame)
+				// Reset the timeout clock while completing the multi-frame receive
+				entry->get_timeout_clock().tick();
+		}
+	}
+
+	return processed_signals;
+}
+
 /**
 * @brief thread to decoding raw CAN messages.
 *
@@ -59,13 +140,15 @@ can_bus_t::can_bus_t(int conf_file)
 *  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.
+*
+*  Depending on the nature of message, if id match a diagnostic request corresponding id for a response 
+*  then decoding a diagnostic message else use classic CAN signals decoding functions.
+*
+*  TODO: make diagnostic messages parsing optionnal.
 */
 void can_bus_t::can_decode_message()
 {
 	can_message_t can_message;
-	std::vector <can_signal_t*> signals;
-	openxc_VehicleMessage vehicle_message;
-	openxc_DynamicField search_key, decoded_message;
 
 	while(is_decoding_)
 	{
@@ -73,34 +156,11 @@ void can_bus_t::can_decode_message()
 		new_can_message_cv_.wait(can_message_lock);
 		can_message = next_can_message();
 
-		/* First we have to found which can_signal_t it is */
-		search_key = build_DynamicField((double)can_message.get_id());
-		signals.clear();
-		configuration_t::instance().find_can_signals(search_key, signals);
-
-		/* Decoding the message ! Don't kill the messenger ! */
-		for(auto& sig : signals)
-		{
-			std::lock_guard<std::mutex> subscribed_signals_lock(get_subscribed_signals_mutex());
-			std::map<std::string, struct afb_event>& 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_generic_name(), decoded_message);
-				vehicle_message = build_VehicleMessage_with_SimpleMessage(openxc_DynamicField_Type::openxc_DynamicField_Type_NUM, s_message);
-
-				std::lock_guard<std::mutex> decoded_can_message_lock(decoded_can_message_mutex_);
-				push_new_vehicle_message(vehicle_message);
-				new_decoded_can_message_.notify_one();
-			}
-		}
+		active_diagnostic_request_t* adr = configuration_t::instance().get_diagnostic_manager().is_diagnostic_response(can_message);
+		if(adr != nullptr)
+			process_diagnostic_signals(adr, can_message);
+		else
+			process_can_signals(can_message);
 	}
 }
 
diff --git a/src/can/can-bus.hpp b/src/can/can-bus.hpp
index a001d1ef..b4f1a3ee 100644
--- a/src/can/can-bus.hpp
+++ b/src/can/can-bus.hpp
@@ -29,6 +29,7 @@
 #include "can-signals.hpp"
 #include "can-message.hpp"
 #include "can-bus-dev.hpp"
+#include "../obd2/active-diagnostic-request.hpp"
 
 #include "../low-can-binding.hpp"
 
@@ -79,6 +80,9 @@ public:
 	void start_threads();
 	void stop_threads();
 
+	int process_can_signals(can_message_t& can_message);
+	int process_diagnostic_signals(active_diagnostic_request_t* adr, const can_message_t& can_message);
+
 	can_message_t next_can_message();
 	void push_new_can_message(const can_message_t& can_msg);
 	std::mutex& get_can_message_mutex();
diff --git a/src/diagnostic/diagnostic-manager.cpp b/src/diagnostic/diagnostic-manager.cpp
index 6867ec39..c27c2b1f 100644
--- a/src/diagnostic/diagnostic-manager.cpp
+++ b/src/diagnostic/diagnostic-manager.cpp
@@ -20,6 +20,7 @@
 #include "diagnostic-manager.hpp"
 
 #include "uds/uds.h"
+#include "../utils/openxc-utils.hpp"
 #include "../configuration.hpp"
 
 #define MAX_RECURRING_DIAGNOSTIC_FREQUENCY_HZ 10
@@ -259,6 +260,62 @@ bool diagnostic_manager_t::add_recurring_request(DiagnosticRequest* request, con
 	return added;
 }
 
+bool diagnostic_manager_t::is_diagnostic_response(const active_diagnostic_request_t& adr, const can_message_t& cm) const
+{
+	if(cm.get_id() == adr.get_id() + DIAGNOSTIC_RESPONSE_ARBITRATION_ID_OFFSET)
+		return true;
+	DEBUG(binder_interface, "Doesn't find an active diagnostic request that matches.");
+	return false;
+}
+
+active_diagnostic_request_t* diagnostic_manager_t::is_diagnostic_response(const can_message_t& can_message)
+{
+	for (auto& entry : non_recurring_requests_)
+	{
+		if(is_diagnostic_response(*entry, can_message))
+			return entry;
+	}
+
+	for (auto& entry : recurring_requests_)
+	{
+		if(is_diagnostic_response(*entry, can_message))
+			return entry;
+	}
+	return nullptr;
+}
+
+openxc_VehicleMessage diagnostic_manager_t::relay_diagnostic_response(active_diagnostic_request_t* adr, const DiagnosticResponse& response) const
+{
+	openxc_VehicleMessage message;
+	float value = (float)diagnostic_payload_to_integer(&response);
+	if(adr->get_decoder() != nullptr)
+	{
+		value = adr->get_decoder()(&response, value);
+	}
+
+	if((response.success && strnlen(adr->get_name().c_str(), adr->get_name().size())) > 0)
+	{
+		// If name, include 'value' instead of payload, and leave of response
+		// details.
+		message = build_VehicleMessage(build_SimpleMessage(adr->get_name(), build_DynamicField(value)));
+	}
+	else
+	{
+		// If no name, send full details of response but still include 'value'
+		// instead of 'payload' if they provided a decoder. The one case you
+		// can't get is the full detailed response with 'value'. We could add
+		// another parameter for that but it's onerous to carry that around.
+		message = build_VehicleMessage(adr, response, value);
+	}
+
+	if(adr->get_callback() != nullptr)
+	{
+		adr->get_callback()(adr, &response, value);
+	}
+
+	return message;
+}
+
 bool diagnostic_manager_t::conflicting(active_diagnostic_request_t* request, active_diagnostic_request_t* candidate) const
 {
 	return (candidate->get_in_flight() && candidate != request &&
diff --git a/src/diagnostic/diagnostic-manager.hpp b/src/diagnostic/diagnostic-manager.hpp
index 29223fa0..061f5f9c 100644
--- a/src/diagnostic/diagnostic-manager.hpp
+++ b/src/diagnostic/diagnostic-manager.hpp
@@ -22,6 +22,7 @@
 #include <vector>
 
 #include "uds/uds.h"
+#include "openxc.pb.h"
 #include "../can/can-bus-dev.hpp"
 #include "../can/can-message.hpp"
 #include "active-diagnostic-request.hpp"
@@ -32,6 +33,7 @@
  * match the maximum CAN controller count.
  */
 #define MAX_SHIM_COUNT can_bus_t.get_can_devices().size()
+#define DIAGNOSTIC_RESPONSE_ARBITRATION_ID_OFFSET 0x8
 
 class active_diagnostic_request_t;
 
@@ -95,6 +97,11 @@ public:
 		bool waitForMultipleResponses, const DiagnosticResponseDecoder decoder,
 		const DiagnosticResponseCallback callback, float frequencyHz);
 	
+	bool is_diagnostic_response(const active_diagnostic_request_t& adr, const can_message_t& cm) const;
+	active_diagnostic_request_t* is_diagnostic_response(const can_message_t& can_message);
+
+	openxc_VehicleMessage relay_diagnostic_response(active_diagnostic_request_t* adr, const DiagnosticResponse& response) const;
+
 	bool conflicting(active_diagnostic_request_t* request, active_diagnostic_request_t* candidate) const;
 	bool clear_to_send(active_diagnostic_request_t* request) const;
 	static int send_request(sd_event_source *s, uint64_t usec, void *userdata);
diff --git a/src/utils/openxc-utils.cpp b/src/utils/openxc-utils.cpp
index 40745a06..34fe7f01 100644
--- a/src/utils/openxc-utils.cpp
+++ b/src/utils/openxc-utils.cpp
@@ -18,32 +18,72 @@
 
 #include "openxc-utils.hpp"
 
-openxc_VehicleMessage build_VehicleMessage_with_SimpleMessage(openxc_DynamicField_Type type, const openxc_SimpleMessage& message)
+#include "../configuration.hpp"
+
+openxc_VehicleMessage build_VehicleMessage(active_diagnostic_request_t* request, const DiagnosticResponse& response, float parsed_value)
 {
-	struct timeb t_msec;
-	long long int timestamp_msec;
-	
-	openxc_VehicleMessage v;
-	
-	if(!::ftime(&t_msec))
+	openxc_VehicleMessage message;
+
+	message.has_type = true;
+	message.type = openxc_VehicleMessage_Type::openxc_VehicleMessage_Type_DIAGNOSTIC;
+	message.has_diagnostic_response = true;
+	message.diagnostic_response.has_bus = true;
+	message.diagnostic_response.bus = configuration_t::instance().get_diagnostic_manager().get_can_bus_dev()->get_address();
+	message.diagnostic_response.has_message_id = true;
+
+	if(request->get_id() != OBD2_FUNCTIONAL_BROADCAST_ID)
+	{
+		message.diagnostic_response.message_id = response.arbitration_id
+			- DIAGNOSTIC_RESPONSE_ARBITRATION_ID_OFFSET;
+	}
+	else
 	{
-		timestamp_msec = ((long long int) t_msec.time) * 1000ll + 
-							(long long int) t_msec.millitm;
+		// must preserve responding arb ID for responses to functional broadcast
+		// requests, as they are the actual module address and not just arb ID +
+		// 8.
+		message.diagnostic_response.message_id = response.arbitration_id;
+	}
 
-		v.has_type = true;
-		v.type = openxc_VehicleMessage_Type::openxc_VehicleMessage_Type_SIMPLE;
-		v.has_simple_message = true;
-		v.simple_message =  message;
-		v.has_timestamp = true;
-		v.timestamp = timestamp_msec;
+	message.diagnostic_response.has_mode = true;
+	message.diagnostic_response.mode = response.mode;
+	message.diagnostic_response.has_pid = response.has_pid;
+	if(message.diagnostic_response.has_pid)
+		message.diagnostic_response.pid = response.pid;
+	message.diagnostic_response.has_success = true;
+	message.diagnostic_response.success = response.success;
+	message.diagnostic_response.has_negative_response_code = !response.success;
+	message.diagnostic_response.negative_response_code =
+			response.negative_response_code;
 
-		return v;
+	if(response.payload_length > 0)
+	{
+		if(request->get_decoder() != nullptr)
+		{
+			message.diagnostic_response.has_value = true;
+			message.diagnostic_response.value = parsed_value;
+		}
+		else
+		{
+			message.diagnostic_response.has_payload = true;
+			::memcpy(message.diagnostic_response.payload.bytes, response.payload,
+					response.payload_length);
+			message.diagnostic_response.payload.size = response.payload_length;
+		}
 	}
 
+	return message;
+}
+
+openxc_VehicleMessage build_VehicleMessage(const openxc_SimpleMessage& message)
+{
+	openxc_VehicleMessage v;
+	
 	v.has_type = true,
 	v.type = openxc_VehicleMessage_Type::openxc_VehicleMessage_Type_SIMPLE;
 	v.has_simple_message = true;
 	v.simple_message =  message;
+	v.has_timestamp = true;
+	v.timestamp = system_time_ms();
 
 	return v;
 }
diff --git a/src/utils/openxc-utils.hpp b/src/utils/openxc-utils.hpp
index 3a4883e0..f837ab1f 100644
--- a/src/utils/openxc-utils.hpp
+++ b/src/utils/openxc-utils.hpp
@@ -23,19 +23,18 @@
 #include <sys/timeb.h>
 
 #include "openxc.pb.h"
+#include "../obd2/active-diagnostic-request.hpp"
 
 /**
- * @fn openxc_VehicleMessage build_VehicleMessage_with_SimpleMessage(openxc_DynamicField_Type type, const openxc_SimpleMessage& message);
- *
  * @brief Build a specific VehicleMessage containing a SimpleMessage.
  *
- * @param[in] type - The type of message to build
  * @param[in] message - simple message to include into openxc_VehicleMessage
  *
  * @return a vehicle message including simple message that will be convert into 
  * a JSON object before being pushed to the subscribers
  */
-openxc_VehicleMessage build_VehicleMessage_with_SimpleMessage(openxc_DynamicField_Type type, const openxc_SimpleMessage& message);
+openxc_VehicleMessage build_VehicleMessage(active_diagnostic_request_t* request, const DiagnosticResponse& response, float parsed_value);
+openxc_VehicleMessage build_VehicleMessage(const openxc_SimpleMessage& message);
 
 /**
  * @fn openxc_SimpleMessage build_SimpleMessage(const std::string& name, const openxc_DynamicField& value);