pushing up initial virtual CAN testing

default.nix

@@ -1,5 +1,12 @@
-{ lib, python311Packages, mcap_support_pkg, py_mcap_pkg
-, py_foxglove_websocket_pkg, asyncudp_pkg, hytech_np_proto_py, proto_gen_pkg}:
+{ lib
+, python311Packages
+, mcap_support_pkg
+, py_mcap_pkg
+, py_foxglove_websocket_pkg
+, asyncudp_pkg
+, hytech_np_proto_py
+, proto_gen_pkg
+}:
 
 python311Packages.buildPythonApplication {
   pname = "py_data_acq";
@@ -9,6 +16,8 @@ python311Packages.buildPythonApplication {
     python311Packages.cantools
     python311Packages.systemd
     python311Packages.websockets
+    python311Packages.pprintpp
+    python311Packages.can
     asyncudp_pkg
     python311Packages.lz4
     python311Packages.zstandard

flake.nix

@@ -70,8 +70,8 @@
         shellHook = let icon = "f121";
         in ''
           path=${pkgs.proto_gen_pkg}
-          bin_path=path+"/bin"
-          dbc_path=path+"/dbc"
+          bin_path=$path"/bin"
+          dbc_path=$path"/dbc"
           
           export BIN_PATH=$bin_path
           export DBC_PATH=$dbc_path
@@ -96,6 +96,7 @@
         name = "nix-devshell";
         packages = with pkgs; [
           # Development Tools
+
           py_dbc_proto_gen_pkg
           protobuf
         ];

py_data_acq/broadcast-test.py

@@ -1,65 +1,35 @@
 #!/usr/bin/env python
 import socket
 import time
+import can
+import cantools
+from pprint import pprint
+import os
 
 from hytech_np_proto_py import hytech_pb2
 
 # Define the IP and port for the UDP socket
-UDP_IP = "127.0.0.1"
-UDP_PORT = 12345
-
+bus1 = can.interface.Bus('can0', bustype='virtual')
 
 def main():
-    # Create an instance of your message
-    my_message = hytech_pb2.id_dashboard_status()
-    my_message.start_button = True
-    my_message.buzzer_active = False
-    my_message.ssok_above_threshold = True
-    my_message.shutdown_h_above_threshold = True
-    my_message.mark_button = True
-    my_message.mode_button = True
-    my_message.motor_controller_cycle_button = True
-    my_message.launch_ctrl_button_ = True
-    my_message.torque_mode_button = True
-    my_message.led_dimmer_button = True
-
-    my_message.dial_state = "yo"
-    my_message.ams_led = "yo"
-    my_message.imd_led = "yo"
-    my_message.mode_led = "yo"
-    my_message.motor_controller_error_led = "yo"
-    my_message.start_status_led = "yo"
-    my_message.inertia_status_led = "yo"
-    my_message.mechanical_brake_led = "yo"
-    my_message.gen_purp_led = "yo"
-    my_message.bots_led = "yo"
-    my_message.cockpit_brb_led = "yo"
-    my_message.crit_charge_led = "yo"
-    my_message.glv_led = "yo"
-    my_message.launch_control_led = "yo"
-
+    path_to_dbc = os.environ.get('DBC_PATH')
+    full_path = os.path.join(path_to_dbc, "hytech.dbc") 
     # Serialize the message to bytes
-    serialized_message = my_message.SerializeToString()
+    db = cantools.database.load_file(full_path)
+    msg = db.get_message_by_name("ID_MC1_TORQUE_COMMAND")
+    print(msg.signals)
+    data = msg.encode({'torque_command': 100})
 
-    # Create a UDP socket
-    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    msg = can.Message(arbitration_id=msg.frame_id, is_extended_id=False, data=data)
+    print(msg)
+    while(1):
 
-    while 1:
-        time.sleep(0.2)
         try:
-            # Send the serialized message over the UDP socket
-            serialized_message = my_message.SerializeToString()
-            sock.sendto(serialized_message, (UDP_IP, UDP_PORT))
-            print(f"Message sent to {UDP_IP}:{UDP_PORT}")
-        except KeyboardInterrupt:
-            # Handle Ctrl+C to exit the loop gracefully
-            sock.close()
-            break
-        except Exception as e:
-            print(f"Error sending message: {e}")
-        # finally:
-        # sock.close()
-
+            bus1.send(msg)
+            print("Message sent on {}".format(bus1.channel_info))
+        except can.CanError:
+            print("Message NOT sent!  Please verify can0 is working first")
+        time.sleep(0.2)
 
 if __name__ == "__main__":
     main()

py_data_acq/test.py

@@ -10,6 +10,8 @@
 import threading
 import os
 import asyncudp
+import can
+import cantools
 
 # TODO we may want to have a config file handling to set params such as:
 #      - file save interval for MCAP file
@@ -18,12 +20,22 @@
 #      - protobuf binary schema file location and file name
 #      - config to inform io handler (say for different CAN baudrates)
 
-async def continuous_udp_receiver(queue, q2):
-    sock = await asyncudp.create_socket(local_addr=('127.0.0.1', 12345))
-    while True:
-        data, addr = await sock.recvfrom()
-        await queue.put(data)
-        await q2.put(data)
+async def continuous_can_receiver(queue, q2):
+    with can.Bus(
+        interface="virtual", channel="can0", receive_own_messages=True
+    ) as bus:
+        reader = can.AsyncBufferedReader()
+        listeners: List[MessageRecipient] = [
+            reader  # AsyncBufferedReader() listener
+        ]
+        loop = asyncio.get_running_loop()
+        notifier = can.Notifier(bus, listeners, loop=loop)
+        while True:
+            # data, addr = await sock.recvfrom()
+            msg = await reader.get_message()
+            print(msg)
+            # await queue.put(data)
+            # await q2.put(data)
 
 async def write_data_to_mcap(queue, mcap_writer):
     async with mcap_writer as mcw:
@@ -44,21 +56,21 @@ async def main():
     path_to_bin = os.environ.get('BIN_PATH')
     full_path = os.path.join(path_to_bin, "hytech.bin")
     print(full_path)
-    fx_s = HTProtobufFoxgloveServer("0.0.0.0", 8765, "asdf", full_path)
+    # fx_s = HTProtobufFoxgloveServer("0.0.0.0", 8765, "asdf", full_path)
 
     mcap_writer = HTPBMcapWriter(".")
 
-    receiver_task = asyncio.create_task(continuous_udp_receiver(queue, queue2))
+    receiver_task = asyncio.create_task(continuous_can_receiver(queue, queue2))
 
-    fx_task = asyncio.create_task(fxglv_websocket_consume_data(queue, fx_s))
+    # fx_task = asyncio.create_task(fxglv_websocket_consume_data(queue, fx_s))
 
     # in the mcap task I actually have to deserialize the any protobuf msg into the message ID and
     # the encoded message for the message id. I will need to handle the same association of message id
     # and schema in the foxglove websocket server. 
     # mcap_task = asyncio.create_task(write_data_to_mcap(queue, mcap_writer)) 
 
     # TODO the data consuming MCAP file task for writing MCAP files to specific directory
-    await asyncio.gather(receiver_task, fx_task)
+    await asyncio.gather(receiver_task)
     # await asyncio.gather(receiver_task, fx_task, mcap_task)
     # await asyncio.gather(receiver_task, mcap_task)
 if __name__ == "__main__":