Merge branch 'develop' into vectorbus_recv_refactoring

Author: zariiii9003

can/interfaces/ics_neovi/neovi_bus.py

@@ -11,7 +11,9 @@
 import logging
 import os
 import tempfile
-from collections import deque
+from collections import deque, defaultdict
+from itertools import cycle
+from threading import Event
 
 from can import Message, CanError, BusABC
 
@@ -55,6 +57,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 # Use inter-process mutex to prevent concurrent device open.
 # When neoVI server is enabled, there is an issue with concurrent device open.
 open_lock = FileLock(os.path.join(tempfile.gettempdir(), "neovi.lock"))
+description_id = cycle(range(1, 0x8000))
 
 
 class ICSApiError(CanError):
@@ -176,6 +179,7 @@ def __init__(self, channel, can_filters=None, **kwargs):
         logger.info("Using device: {}".format(self.channel_info))
 
         self.rx_buffer = deque()
+        self.message_receipts = defaultdict(Event)
 
     @staticmethod
     def channel_to_netid(channel_name_or_id):
@@ -267,6 +271,9 @@ def _process_msg_queue(self, timeout=0.1):
             if is_tx:
                 if bool(ics_msg.StatusBitField & ics.SPY_STATUS_GLOBAL_ERR):
                     continue
+                if ics_msg.DescriptionID:
+                    receipt_key = (ics_msg.ArbIDOrHeader, ics_msg.DescriptionID)
+                    self.message_receipts[receipt_key].set()
                 if not self._receive_own_messages:
                     continue
 
@@ -349,7 +356,19 @@ def _recv_internal(self, timeout=0.1):
             return None, False
         return msg, False
 
-    def send(self, msg, timeout=None):
+    def send(self, msg, timeout=0):
+        """Transmit a message to the CAN bus.
+
+        :param Message msg: A message object.
+
+        :param float timeout:
+            If > 0, wait up to this many seconds for message to be ACK'ed.
+            If timeout is exceeded, an exception will be raised.
+            None blocks indefinitely.
+
+        :raises can.CanError:
+            if the message could not be sent
+        """
         if not ics.validate_hobject(self.dev):
             raise CanError("bus not open")
         message = ics.SpyMessage()
@@ -385,7 +404,20 @@ def send(self, msg, timeout=None):
         else:
             raise ValueError("msg.channel must be set when using multiple channels.")
 
+        msg_desc_id = next(description_id)
+        message.DescriptionID = msg_desc_id
+        receipt_key = (msg.arbitration_id, msg_desc_id)
+
+        if timeout != 0:
+            self.message_receipts[receipt_key].clear()
+
         try:
             ics.transmit_messages(self.dev, message)
         except ics.RuntimeError:
             raise ICSApiError(*ics.get_last_api_error(self.dev))
+
+        # If timeout is set, wait for ACK
+        # This requires a notifier for the bus or
+        # some other thread calling recv periodically
+        if timeout != 0 and not self.message_receipts[receipt_key].wait(timeout):
+            raise CanError("Transmit timeout")